Frida MCP

Instructions

This tool allows for dynamic scripting against a process previously attached to
via `create_interactive_session`.

Implementation Reference

• src/frida_mcp/cli.py:501-723 (handler)

  Core handler that executes the provided JavaScript code in the specified Frida session. Supports both one-off execution and persistent scripts with message queuing for async events like hooks.

  def execute_in_session(
      session_id: str = Field(
          description="The unique identifier of the active Frida session. This ID is obtained when the session is first created."
      ),
      javascript_code: str = Field(
          description="A string containing the JavaScript code to be executed in the target process's context. The script can use Frida's JavaScript API (e.g., Interceptor, Memory, Module, rpc)."
      ),
      keep_alive: bool = Field(
          default=False,
          description="A boolean flag indicating whether the script should remain loaded in the target process after its initial execution. If False (default), the script is unloaded after initial run. If True, it persists for hooks/RPC and messages are retrieved via get_session_messages. Note: With keep_alive=True, JavaScript code should manage log volume (limits, deduplication) to prevent too many messages.",
      ),
  ) -> Dict[str, Any]:
      """Execute JavaScript code within an existing interactive Frida session.
  
      This tool allows for dynamic scripting against a process previously attached to
      via `create_interactive_session`.
      """
      if session_id not in _scripts:
          raise ValueError(f"Session with ID {session_id} not found")
  
      session = _scripts[session_id]
      lock = _message_locks[session_id]
  
      try:
          # For interactive use, we need to handle console.log output
          # and properly format the result
  
          # Wrap the code to capture console.log output and return values
          # This basic wrapper sends back immediate script result/errors and console.log output
          # For keep_alive=True, subsequent messages from the script (e.g., from Interceptor)
          # will be handled by the persistent on_message handler.
          wrapped_code = f"""
          (function() {{
              function formatValue(value) {{
                  if (value === null) {{
                      return 'null';
                  }}
                  if (value === undefined) {{
                      return 'undefined';
                  }}
  
                  var valueType = typeof value;
                  if (valueType === 'string') {{
                      return value;
                  }}
                  if (valueType === 'number' || valueType === 'boolean' || valueType === 'bigint') {{
                      return String(value);
                  }}
                  if (valueType === 'symbol' || valueType === 'function') {{
                      try {{
                          return value.toString();
                      }} catch (e) {{
                          return '[unrepresentable value]';
                      }}
                  }}
                  if (valueType === 'object') {{
                      try {{
                          var cache = new Set();
                          var json = JSON.stringify(value, function(key, val) {{
                              if (typeof val === 'object' && val !== null) {{
                                  if (cache.has(val)) {{
                                      return '[Circular]';
                                  }}
                                  cache.add(val);
                              }}
                              return val;
                          }});
                          if (json !== undefined) {{
                              return json;
                          }}
                      }} catch (jsonErr) {{
                          // Fall through to more generic formatting below
                      }}
                      try {{
                          return Object.prototype.toString.call(value);
                      }} catch (descriptorErr) {{
                          // Fall through to final String conversion
                      }}
                  }}
  
                  try {{
                      return String(value);
                  }} catch (stringErr) {{
                      return '[unrepresentable value]';
                  }}
              }}
  
              var initialLogs = [];
              var originalLog = console.log;
  
              console.log = function() {{
                  var args = Array.prototype.slice.call(arguments);
                  var logMsg = args.map(formatValue).join(' ');
                  initialLogs.push(logMsg);
                  originalLog.apply(console, arguments); // Also keep original console behavior
              }};
  
              var scriptResult;
              var scriptError;
              try {{
                  scriptResult = eval({javascript_code!r});
              }} catch (e) {{
                  scriptError = {{ message: e.toString(), stack: e.stack }};
              }}
  
              console.log = originalLog; // Restore
  
              send({{ // This send is for the initial execution result
                  type: 'execution_receipt',
                  result: scriptError ? undefined : formatValue(scriptResult),
                  error: scriptError,
                  initial_logs: initialLogs
              }});
          }})();
          """
  
          script = session.create_script(wrapped_code)
  
          # This list captures messages from the initial execution of the script (the wrapper)
          initial_execution_results = []
  
          def on_initial_message(message, data):
              # This handler is for the initial execution wrapper's send()
              if (
                  message["type"] == "send"
                  and message["payload"]["type"] == "execution_receipt"
              ):
                  initial_execution_results.append(message["payload"])
              elif message["type"] == "error":  # Script compilation/syntax errors
                  initial_execution_results.append(
                      {"script_error": message["description"], "details": message}
                  )
  
          # This handler is for persistent messages if keep_alive is true
          def on_persistent_message(message, data):
              with lock:
                  _script_messages[session_id].append(
                      {
                          "type": message["type"],
                          "payload": message.get("payload"),
                          "data": data,
                      }
                  )
  
          if keep_alive:
              # For keep_alive, messages go to the global queue _script_messages
              # The script object itself will handle these.
              script.on("message", on_persistent_message)
              # Store the script object if we need to interact with it later (e.g., specific unload)
              # For now, it's attached to the session and will be cleaned up when session is detached or process ends.
              if (
                  session_id not in global_persistent_scripts
              ):  # Requires global_persistent_scripts dict
                  global_persistent_scripts[session_id] = []
              global_persistent_scripts[session_id].append(script)
  
          else:
              # For non-persistent scripts, use the local handler for immediate results
              script.on("message", on_initial_message)
  
          script.load()
  
          # For non-persistent scripts, give a short time for the initial_execution_results
          # For persistent scripts, this sleep is less critical as it's about setting up.
          if not keep_alive:
              time.sleep(0.2)  # Slightly increased for safety
  
          # Process initial results (for both modes, but primarily for non-keep_alive)
          final_result = {}
          if initial_execution_results:
              # Use the first message from the execution_receipt
              receipt = initial_execution_results[0]
              if "script_error" in receipt:
                  final_result = {
                      "status": "error",
                      "error": "Script execution error",
                      "details": receipt["script_error"],
                  }
              elif receipt.get("error"):
                  final_result = {
                      "status": "error",
                      "error": receipt["error"]["message"],
                      "stack": receipt["error"]["stack"],
                      "initial_logs": receipt.get("initial_logs", []),
                  }
              else:
                  final_result = {
                      "status": "success",
                      "result": receipt["result"],
                      "initial_logs": receipt.get("initial_logs", []),
                  }
          elif keep_alive:
              final_result = {
                  "status": "success",
                  "message": "Script loaded persistently. Use get_session_messages to retrieve asynchronous messages.",
                  "initial_logs": [],
              }
          else:  # No messages received, could be an issue or just a silent script
              final_result = {
                  "status": "nodata",  # Or "success" if empty result is fine
                  "message": "Script loaded but sent no initial messages.",
                  "initial_logs": [],
              }
  
          if not keep_alive:
              script.unload()
              final_result["script_unloaded"] = True
          else:
              final_result["script_unloaded"] = False
              final_result["info"] = (
                  "Script is persistent. Remember to manage its lifecycle if necessary."
              )
  
          return final_result
  
      except frida.InvalidOperationError as e:  # E.g. session detached
          return {
              "status": "error",
              "error": f"Frida operation error: {str(e)} (Session may be detached)",
          }
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/frida_mcp/cli.py:502-512 (schema)

  Input schema defined using Pydantic Field objects, specifying session_id, javascript_code, and keep_alive parameters with descriptions and defaults.

      session_id: str = Field(
          description="The unique identifier of the active Frida session. This ID is obtained when the session is first created."
      ),
      javascript_code: str = Field(
          description="A string containing the JavaScript code to be executed in the target process's context. The script can use Frida's JavaScript API (e.g., Interceptor, Memory, Module, rpc)."
      ),
      keep_alive: bool = Field(
          default=False,
          description="A boolean flag indicating whether the script should remain loaded in the target process after its initial execution. If False (default), the script is unloaded after initial run. If True, it persists for hooks/RPC and messages are retrieved via get_session_messages. Note: With keep_alive=True, JavaScript code should manage log volume (limits, deduplication) to prevent too many messages.",
      ),
  ) -> Dict[str, Any]:

• src/frida_mcp/cli.py:500-500 (registration)

  Registers the execute_in_session function as an MCP tool using the FastMCP @mcp.tool() decorator.

  @mcp.tool()

• src/frida_mcp/cli.py:77-82 (helper)

  Global state variables used by execute_in_session to store sessions, messages, locks, and persistent scripts.

  # Global dictionary to store scripts and their messages
  # This allows us to retrieve messages from scripts after they've been created
  _scripts = {}
  _script_messages = {}
  _message_locks = {}
  global_persistent_scripts = {}  # Added for managing persistent scripts