diff --git a/DEPLOY.md b/DEPLOY.md new file mode 100644 index 0000000..7aa432a --- /dev/null +++ b/DEPLOY.md @@ -0,0 +1,224 @@ +# Hosting and hardening + +This covers two things people usually ask for together: keeping the MCP endpoint up +even when your analysis machine reboots, and making sure nobody in the middle can read +what goes over MCP. + +## Read this first: what hosting can and can't do + +The MCP server is almost stateless. It is a registry of which IDA instances are alive +plus a proxy that forwards tool calls to them. **All the real capability lives in the +IDA instances**, which run on your analysis machine next to IDA. + +So: + +- Moving the server onto an always-on host (a small VPS) keeps the **endpoint** alive. + Clients stay connected and never have to change the URL. When your Mac reboots and IDA + comes back, the plugin re-registers on its own (the heartbeat already does this) and + tools start working again with no manual step. +- It does **not** give you functional high availability. If the machine running IDA dies, + there is nothing to serve until it comes back — `ida_list_instances` just returns empty. + Real HA would mean running IDA on more than one machine, which is a different project. + +Second thing to be clear-eyed about: now that the plugin exposes `exec_python` and +`exec_idc`, the MCP channel is effectively **remote code execution** on the analysis +machine. Transport encryption and strong auth are not optional niceties anymore — anyone +who can read or reach the channel can run code on that box. Treat the token and the +private key like SSH keys. + +## Why a free HTTP tunnel is not private + +A plain-HTTP `ngrok` (or any HTTP tunnel) terminates TLS at the provider's edge and +forwards your traffic to the agent in the clear. The tunnel from your machine to the +provider is encrypted, but **the provider's servers see your plaintext** — the +decompiler output, the addresses, everything. If the goal is "nobody learns what goes +over MCP," an HTTP tunnel does not meet it. You need one of: + +- a VPN, so the transport is encrypted and the endpoint is not public at all, or +- your own TLS endpoint, where the only machine that sees plaintext is one you control, or +- end-to-end TLS through a pure TCP/TLS passthrough tunnel (the tunnel moves bytes it + cannot read). + +## Pick your shape based on who the clients are + +**If your MCP client is local** (Claude Code / desktop app on your own machine): put +everything on a private network (Tailscale or WireGuard) and expose nothing publicly. +This is the strongest option — there is no public endpoint to find, and every leg is +encrypted by the VPN. + +**If your MCP client is a hosted connector** (e.g. a claude.ai connector that connects +from someone else's servers): that client cannot join your VPN and usually cannot present +a client certificate, so it needs a public HTTPS endpoint with a strong bearer token. Keep +the plugin-to-server leg on the VPN so it never touches the public internet. The only +machine that sees plaintext is your own VPS. + +Both shapes share the same back-channel: **the IDA plugins talk to the server over the +VPN**, never over a public plaintext port. + +--- + +## Build A — everything on a private network (local client, strongest) + +Tailscale is the low-footgun choice; raw WireGuard is the DIY equivalent. + +1. Install Tailscale on the always-on host and on the analysis machine, and `tailscale up` + on both. Each gets a stable `100.x.y.z` address. +2. On the always-on host, run the server bound to its tailnet address, with a token: + + ```bash + MCP_BIND_HOST=100.x.y.z \ + IDA_MCP_TOKEN=$(openssl rand -hex 32) \ + python ida-mcp-server.py + ``` + +3. Point the IDA plugin at the server's tailnet address, and have it listen only on the + analysis machine's own tailnet address: + + - `IDA_MCP_HOST` = the analysis machine's `100.x.y.z` (plugin listens only there) + - `IDA_MCP_ADVERTISE_HOST` = the same address (what it reports to the server) + - `IDA_MCP_REGISTRY_URL` = `http://100.x.y.z:8888` (the server's tailnet address) + - `IDA_MCP_TOKEN` = the same token as the server + +4. Point your local MCP client at `http://100.x.y.z:8888/sse` (the server's tailnet + address). + +Nothing is published to the internet. Every hop is inside the tailnet, which is +end-to-end encrypted. Plain HTTP is fine here because the network itself is the +encryption layer — but keep the token set so a compromised tailnet device still can't +drive it. + +--- + +## Build B — public HTTPS front door + private back-channel (hosted client) + +Use this only when the client must reach you from the public internet. + +Layout: + +``` +hosted client ──HTTPS(443)──▶ Caddy (VPS) ──▶ uvicorn :8888 (VPS, localhost) + ▲ +analysis machine ──WireGuard/Tailscale──────────────┘ (plugin register + proxied calls) +``` + +1. On the VPS, run the server with a token (bound to all interfaces; the firewall and + Caddy decide who actually reaches it): + + ```bash + MCP_BIND_HOST=0.0.0.0 IDA_MCP_TOKEN=$(openssl rand -hex 32) python ida-mcp-server.py + ``` + +2. Put Caddy in front for automatic Let's Encrypt TLS, and **only expose `/sse` and the + OAuth paths** — block the instance registry from the public side (the plugin reaches + it over the VPN, not through Caddy): + + ``` + mcp.example.com { + @registry path /instances /instances/* + respond @registry "forbidden" 403 + reverse_proxy 127.0.0.1:8888 + } + ``` + +3. Firewall the VPS so the app port is never reachable directly from the internet — only + 443 (Caddy) is public, and the app port is reachable from localhost and the VPN subnet: + + ```bash + ufw default deny incoming + ufw allow 443/tcp + ufw allow in on wg0 # or the tailscale interface + ufw allow 51820/udp # WireGuard handshake, if using raw WireGuard + ufw enable + ``` + +4. Join the analysis machine to the VPN (Tailscale as in Build A, or WireGuard) and set + the plugin's `IDA_MCP_REGISTRY_URL` / `IDA_MCP_ADVERTISE_HOST` to the VPS's VPN + address, exactly as in Build A step 3. The plugin never uses the public hostname. + +5. Point the hosted client at `https://mcp.example.com/sse` with the bearer token. + +--- + +## Native TLS and mutual TLS (no reverse proxy) + +The server can serve HTTPS itself, and can require client certificates (mutual TLS) so +that even someone who knows the URL cannot open a connection without a cert you issued. +This is ideal when your client is your own script or Claude Code (which you can give a +client cert); most hosted connectors cannot present one, so use token-over-TLS for those. + +Generate a CA, a server cert, and one client cert: + +```bash +# CA +openssl genrsa -out ca.key 4096 +openssl req -x509 -new -nodes -key ca.key -sha256 -days 3650 -subj "/CN=ida-mcp-ca" -out ca.crt + +# server cert (CN must match how the client addresses it) +openssl genrsa -out server.key 4096 +openssl req -new -key server.key -subj "/CN=mcp.example.com" -out server.csr +openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -days 825 -sha256 -out server.crt + +# client cert +openssl genrsa -out client.key 4096 +openssl req -new -key client.key -subj "/CN=analyst-1" -out client.csr +openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial -days 825 -sha256 -out client.crt +``` + +Run with TLS (add the CA line for mutual TLS): + +```bash +MCP_TLS_CERT=server.crt \ +MCP_TLS_KEY=server.key \ +MCP_TLS_CLIENT_CA=ca.crt \ +IDA_MCP_TOKEN=$(openssl rand -hex 32) \ +python ida-mcp-server.py +``` + +With `MCP_TLS_CLIENT_CA` set, the server refuses any connection that doesn't present a +cert signed by your CA. Bundle `client.crt` + `client.key` into the client (for `curl`: +`--cert client.crt --key client.key --cacert ca.crt`). + +--- + +## Keep it running (systemd on the VPS) + +```ini +# /etc/systemd/system/ida-mcp.service +[Unit] +Description=IDA MCP server +After=network-online.target +Wants=network-online.target + +[Service] +WorkingDirectory=/opt/ida-mcp-server +Environment=MCP_BIND_HOST=0.0.0.0 +Environment=IDA_MCP_TOKEN=REPLACE_WITH_openssl_rand_hex_32 +ExecStart=/usr/bin/python3 /opt/ida-mcp-server/ida-mcp-server.py +Restart=always +RestartSec=2 +# hardening +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=true +ReadWritePaths=/opt/ida-mcp-server +PrivateTmp=true + +[Install] +WantedBy=multi-user.target +``` + +```bash +systemctl daemon-reload && systemctl enable --now ida-mcp +``` + +`Restart=always` covers a server crash; the plugin's heartbeat covers an IDA restart on +your side. Together the endpoint stays up across both, without anyone reconfiguring a URL. + +## Checklist + +- [ ] Token set (`openssl rand -hex 32`), same value on server and every plugin. +- [ ] Plugin-to-server traffic goes over a VPN, never a public plaintext port. +- [ ] No free HTTP tunnel in the path (it can read everything). VPN, own TLS, or TLS passthrough only. +- [ ] Public exposure (if any) is 443 behind a proxy or native TLS; the app port is firewalled off the internet. +- [ ] `/instances*` is not reachable from the public side. +- [ ] Remember `exec_python`/`exec_idc` are RCE on the analysis box — the token and any private key are as sensitive as an SSH key. diff --git a/README.md b/README.md index b8d555f..062bffe 100644 --- a/README.md +++ b/README.md @@ -90,8 +90,18 @@ server's environment. That gates: typed in during authorization; it isn't handed out automatically) Without a token, anyone who has the URL has full read/write access to whatever -IDA session is open -- decompiling, renaming, patching, all of it. Don't skip -the token if the server is reachable from outside your own machine. +IDA session is open -- decompiling, renaming, patching, all of it. And because +`exec_python`/`exec_idc` run arbitrary code inside IDA, an open channel is +effectively remote code execution on the analysis machine -- so the token (and +any TLS key) is as sensitive as an SSH key. + +The server can serve HTTPS itself (`MCP_TLS_CERT` / `MCP_TLS_KEY`) and can require +client certificates for mutual TLS (`MCP_TLS_CLIENT_CA`), and it binds to +`MCP_BIND_HOST` (default `0.0.0.0`). A plain HTTP tunnel like free ngrok is **not** +private -- it terminates TLS at the provider, which then sees your traffic in the +clear. For hosting the endpoint so clients survive a reboot, and for keeping the +traffic actually confidential (VPN back-channel, public TLS front door, mutual TLS), +see [DEPLOY.md](DEPLOY.md). ## What the tools can do diff --git a/ida-mcp-server.py b/ida-mcp-server.py index 286a66d..d8725af 100644 --- a/ida-mcp-server.py +++ b/ida-mcp-server.py @@ -1,3 +1,4 @@ +import asyncio import json import os import secrets @@ -10,10 +11,18 @@ from starlette.responses import JSONResponse, Response, HTMLResponse from starlette.routing import Route SSE_PORT = int(os.environ.get("MCP_PORT", 8888)) +BIND_HOST = os.environ.get("MCP_BIND_HOST", "0.0.0.0") NGROK_URL = os.environ.get("NGROK_URL", "") AUTH_TOKEN = os.environ.get("IDA_MCP_TOKEN") INSTANCE_TTL = int(os.environ.get("IDA_MCP_INSTANCE_TTL", 30)) +# TLS: set both to serve HTTPS directly (no reverse proxy needed). +TLS_CERT = os.environ.get("MCP_TLS_CERT") +TLS_KEY = os.environ.get("MCP_TLS_KEY") +# mTLS: set to a CA bundle to REQUIRE clients to present a cert signed by it. +# Nobody without a matching client cert can even open the connection. +TLS_CLIENT_CA = os.environ.get("MCP_TLS_CLIENT_CA") + http = httpx.AsyncClient(timeout=120.0) @@ -124,6 +133,41 @@ TOOLS = [ ), "inputSchema": {"type": "object", "properties": {}}, }, + { + "name": "ida_overview", + "description": ( + "One-shot triage of an instance: file, Hex-Rays availability, analysis-complete flag, function " + "count, segments, entry points, and the top functions by xref count. Call this FIRST on an " + "unfamiliar binary instead of several separate calls." + ), + "inputSchema": _schema(), + }, + { + "name": "ida_batch", + "description": ( + "Run several IDA tool calls in ONE request and get all results back together. Pass calls as " + '[{"name": "ida_decompile", "arguments": {...}}, ...]. Cuts round-trips when you already know the ' + "list of things to fetch (e.g. decompile + xrefs + lvars for one function, or the same call across " + "many targets). Sub-calls run concurrently; each returns its own result or error." + ), + "inputSchema": { + "type": "object", + "properties": { + "calls": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "arguments": {"type": "object"}, + }, + "required": ["name"], + }, + } + }, + "required": ["calls"], + }, + }, { "name": "ida_ping", "description": "Check connectivity to an IDA instance; reports whether Hex-Rays decompiler is available.", @@ -370,6 +414,50 @@ TOOLS = [ ), "inputSchema": _schema({"filter": {"type": "string"}, "limit": {"type": "integer", "default": 200}}), }, + { + "name": "ida_search_names", + "description": ( + "Search ALL named locations by substring, not just functions -- global variables, guard/checksum " + "keys, data labels, anything with a symbol. Use this when ida_search_functions comes up empty for " + "something you found referenced in disassembly (e.g. a data symbol loaded by LDR)." + ), + "inputSchema": _schema({"q": {"type": "string"}, "limit": {"type": "integer", "default": 200}}, ["q"]), + }, + { + "name": "ida_search_text", + "description": "Substring search across both named locations AND strings in one call -- convenience wrapper over search_names + search_strings.", + "inputSchema": _schema({"q": {"type": "string"}, "limit": {"type": "integer", "default": 100}}, ["q"]), + }, + { + "name": "ida_exec_python", + "description": ( + "Run arbitrary Python inside IDA on its main thread, with idaapi/idc/idautils and the common ida_* " + "modules preloaded. Assign to a variable named `result` to return a value; stdout/stderr are captured. " + "This is the escape hatch for anything the other tools don't cover -- and the most powerful endpoint, " + "so use it deliberately. Example: result = [hex(e) for e in idautils.Functions()][:5]" + ), + "inputSchema": _schema({"code": {"type": "string"}}, ["code"]), + }, + { + "name": "ida_exec_idc", + "description": 'Evaluate an IDC expression, e.g. "get_screen_ea()" or "sizeof(0x1000)". Returns the IDC value.', + "inputSchema": _schema({"code": {"type": "string"}}, ["code"]), + }, + { + "name": "ida_get_function_info", + "description": "Rich metadata for one function: prototype, size, frame size, flags, and caller/callee lists -- one call instead of several.", + "inputSchema": _schema({"target": {"type": "string"}}, ["target"]), + }, + { + "name": "ida_get_lvars", + "description": "List the local variables of a decompiled function with their recovered types -- read-only counterpart to ida_set_lvar_type.", + "inputSchema": _schema({"target": {"type": "string"}}, ["target"]), + }, + { + "name": "ida_segment_at", + "description": "Report which segment contains an address (name, range, permissions). Useful to know where an ea lives before reading/patching.", + "inputSchema": _schema({"target": {"type": "string"}}, ["target"]), + }, { "name": "ida_find_immediate", "description": ( @@ -459,6 +547,46 @@ async def execute_tool(name: str, arguments: dict): b = await ida_get(hb, pb, f"/decompile/{arguments['target_b']}") return {"instance_a": arguments["instance_a"], "instance_b": arguments["instance_b"], "a": a, "b": b} + if name == "ida_batch": + calls = arguments.get("calls", []) + if len(calls) > 64: + return {"error": "batch too large (max 64 calls)", "code": "bad_request"} + + async def _one(c): + sub = c.get("name") + if sub in ("ida_batch",): + return {"error": "nested ida_batch is not allowed", "code": "bad_request"} + return await execute_tool(sub, c.get("arguments", {}) or {}) + + results = await asyncio.gather(*[_one(c) for c in calls], return_exceptions=True) + out = [] + for c, r in zip(calls, results): + if isinstance(r, Exception): + out.append({"name": c.get("name"), "error": str(r), "code": "internal_error"}) + else: + out.append({"name": c.get("name"), "result": r}) + return {"count": len(out), "results": out} + + if name == "ida_overview": + host, port, inst_id, auto_picked = resolve_instance(arguments.get("instance")) + ping, segs, eps, top, funcs = await asyncio.gather( + ida_get(host, port, "/ping"), + ida_get(host, port, "/segments"), + ida_get(host, port, "/entry_points"), + ida_get(host, port, "/top_functions?by=xrefs&limit=10"), + ida_get(host, port, "/functions?filter=&limit=0"), + ) + return { + "instance": inst_id, + "file": ping.get("file"), + "hexrays_available": ping.get("hexrays_available"), + "analysis_complete": ping.get("analysis_complete"), + "function_count": funcs.get("total"), + "entry_points": eps, + "segments": segs, + "top_functions": top, + } + host, port, inst_id, auto_picked = resolve_instance(arguments.get("instance")) result = await _dispatch(name, arguments, host, port) if isinstance(result, dict) and auto_picked: @@ -579,6 +707,20 @@ async def _dispatch(name: str, arguments: dict, host, port): elif name == "ida_get_objc_classes": f = arguments.get("filter", "") return await ida_get(host, port, f"/objc_classes?filter={f}&limit={arguments.get('limit', 200)}") + elif name == "ida_search_names": + return await ida_get(host, port, f"/names/search?q={arguments['q']}&limit={arguments.get('limit', 200)}") + elif name == "ida_search_text": + return await ida_get(host, port, f"/search_text?q={arguments['q']}&limit={arguments.get('limit', 100)}") + elif name == "ida_exec_python": + return await ida_post(host, port, "/exec_python", {"code": arguments["code"]}) + elif name == "ida_exec_idc": + return await ida_post(host, port, "/exec_idc", {"code": arguments["code"]}) + elif name == "ida_get_function_info": + return await ida_get(host, port, f"/function_info/{arguments['target']}") + elif name == "ida_get_lvars": + return await ida_get(host, port, f"/lvars/{arguments['target']}") + elif name == "ida_segment_at": + return await ida_get(host, port, f"/segment_at/{arguments['target']}") elif name == "ida_find_immediate": return await ida_get(host, port, f"/find_immediate?value={arguments['value']}&limit={arguments.get('limit', 100)}") elif name == "ida_find_vtables": @@ -863,13 +1005,30 @@ def create_app(): if __name__ == "__main__": + import ssl import uvicorn - print(f"[IDA MCP] Dynamic multi-instance mode. MCP endpoint: http://0.0.0.0:{SSE_PORT}/sse") - print(f"[IDA MCP] Waiting for IDA plugin(s) to self-register on /instances/register ...") + ssl_kwargs = {} + if TLS_CERT and TLS_KEY: + ssl_kwargs["ssl_certfile"] = TLS_CERT + ssl_kwargs["ssl_keyfile"] = TLS_KEY + if TLS_CLIENT_CA: + ssl_kwargs["ssl_ca_certs"] = TLS_CLIENT_CA + ssl_kwargs["ssl_cert_reqs"] = ssl.CERT_REQUIRED + scheme = "https" if ssl_kwargs else "http" + + print(f"[IDA MCP] Dynamic multi-instance mode. MCP endpoint: {scheme}://{BIND_HOST}:{SSE_PORT}/sse") + print("[IDA MCP] Waiting for IDA plugin(s) to self-register on /instances/register ...") + if ssl_kwargs.get("ssl_cert_reqs"): + print("[IDA MCP] TLS + mutual TLS ON — clients must present a cert signed by MCP_TLS_CLIENT_CA.") + elif ssl_kwargs: + print("[IDA MCP] TLS ON — serving HTTPS directly.") + else: + print("[IDA MCP] TLS OFF — plaintext HTTP. Only safe behind a VPN/localhost or a TLS-terminating proxy.") if not AUTH_TOKEN: - print("[IDA MCP] WARNING: IDA_MCP_TOKEN is not set — server is open to anyone who can reach it.") - print("[IDA MCP] Set IDA_MCP_TOKEN before exposing this via ngrok/the internet.") - if NGROK_URL: - print(f"[IDA MCP] NGROK_URL: {NGROK_URL}") - uvicorn.run(create_app(), host="0.0.0.0", port=SSE_PORT) + print("[IDA MCP] WARNING: IDA_MCP_TOKEN is not set — anyone who can reach the port has full control.") + if NGROK_URL and "ngrok" in NGROK_URL and not ssl_kwargs: + print("[IDA MCP] WARNING: a plain-HTTP ngrok tunnel terminates TLS at ngrok's edge — ngrok can read") + print("[IDA MCP] everything sent over MCP. For real confidentiality use a VPN (WireGuard/") + print("[IDA MCP] Tailscale) or your own TLS endpoint, not a free HTTP tunnel. See DEPLOY.md.") + uvicorn.run(create_app(), host=BIND_HOST, port=SSE_PORT, **ssl_kwargs) diff --git a/ida_mcp_plugin.py b/ida_mcp_plugin.py index d06fa48..1f143a3 100644 --- a/ida_mcp_plugin.py +++ b/ida_mcp_plugin.py @@ -52,6 +52,8 @@ HEARTBEAT_INTERVAL = 10 PORT = None # set in _start() INSTANCE_ID = None # set in _start() +_PLUGIN_INSTANCE = None # set in IdaMcpPlugin.init(), used by _reload_plugin_code +_FORCED_PORT = None # set by the settings dialog to pin a specific port def _find_free_port(start: int, host: str, tries: int = 50) -> int: @@ -678,15 +680,12 @@ def _rebuild_name_cache(): def _reload_plugin_code(): - """Re-reads this .py file from disk and swaps in the new handler functions - without restarting IDA. This works because do_GET/do_POST look up - _decompile/_get_imports/etc. by name in the module's global namespace on - every request -- after reload, that lookup finds the new code. - - Limitation: PORT/INSTANCE_ID are restored manually (reload would otherwise - reset them), and the heartbeat thread/HTTPServer keep running with their - old code until an actual IDA restart, since they were started with a - direct function reference rather than a by-name lookup.""" + """Re-reads this .py file from disk, swaps in the new handler functions, + and rebinds a fresh HTTP server on the same port using the newly defined + IDAHandler class -- so new routes work too, not just edits to existing + function bodies. The heartbeat thread keeps running with its old code + until an actual IDA restart, since it holds a direct reference to the + old target function rather than looking it up by name.""" def _(): import importlib @@ -716,11 +715,34 @@ def _reload_plugin_code(): mod._name_cache_built = False threading.Thread(target=lambda: run_in_main_thread(mod._build_name_cache), daemon=True).start() + server_swapped = False + inst = mod._PLUGIN_INSTANCE + if inst is not None and getattr(inst, "server", None) is not None: + try: + old_server = inst.server + new_server = mod.ThreadingHTTPServer((mod.BIND_HOST, mod.PORT), mod.IDAHandler) + new_thread = threading.Thread(target=new_server.serve_forever, daemon=True) + new_thread.start() + inst.server = new_server + inst.thread = new_thread + old_server.shutdown() + old_server.server_close() + server_swapped = True + except Exception as e: + print(f"[IDA MCP:{mod.PORT}] Server swap on reload failed (routes may be stale): {e}") + return { "ok": True, "port": mod.PORT, "id": mod.INSTANCE_ID, - "note": "Endpoint handlers reloaded. Heartbeat thread keeps its old code until a full IDA restart.", + "server_swapped": server_swapped, + "note": ( + "Endpoint handlers and routes reloaded (new HTTP server bound). " + "Heartbeat thread keeps its old code until a full IDA restart." + if server_swapped + else "Endpoint handler bodies reloaded, but the HTTP server could not be swapped " + "-- new routes/methods on IDAHandler may still 404 until IDA restarts." + ), } return run_in_main_thread(_) @@ -1153,6 +1175,211 @@ def _find_vtables(segment_filter: str = None, min_entries: int = 2, limit: int = return run_in_main_thread(_) +def _search_names(q: str, limit: int = 200): + def _(): + _ensure_cache() + with _name_cache_lock: + items = list(_name_cache.items()) + q_lower = q.lower() + results = [] + for name, ea in items: + if q_lower in name.lower(): + results.append({"ea": hex(ea), "name": name}) + if len(results) >= limit: + break + return {"count": len(results), "names": results} + + return run_in_main_thread(_) + + +def _exec_python(code: str): + """Runs arbitrary Python on IDA's main thread. Full IDA API access. Convention: + assign to a variable named `result` to return a value; stdout is captured too. + This is the most powerful (and most dangerous) endpoint -- gated by the token.""" + + def _(): + import io + import contextlib + import traceback + + buf = io.StringIO() + ns = {"idaapi": idaapi, "idc": idc, "idautils": idautils} + for mod in ( + "ida_funcs", "ida_bytes", "ida_name", "ida_segment", "ida_kernwin", + "ida_hexrays", "ida_typeinf", "ida_nalt", "ida_xref", "ida_ua", + ): + try: + ns[mod] = __import__(mod) + except Exception: + pass + try: + with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf): + exec(code, ns) + result = ns.get("result", None) + try: + json.dumps(result) + except Exception: + result = repr(result) + return {"ok": True, "stdout": buf.getvalue(), "result": result} + except Exception as e: + return { + "ok": False, + "error": str(e), + "traceback": traceback.format_exc(), + "stdout": buf.getvalue(), + } + + return run_in_main_thread(_) + + +def _exec_idc(code: str): + """Evaluates an IDC expression/snippet via idc.eval_idc (IDA 9.x), falling back + to ida_expr.eval_idc_expr if the convenience wrapper is unavailable.""" + + def _(): + try: + rv = idc.eval_idc(code) + return {"ok": True, "result": rv} + except AttributeError: + try: + import ida_expr + + val = ida_expr.idc_value_t() + err = ida_expr.eval_idc_expr(val, idaapi.BADADDR, code) + if err: + return {"ok": False, "error": err} + return {"ok": True, "result": val.c_str() if hasattr(val, "c_str") else str(val.num)} + except Exception as e: + return {"ok": False, "error": f"idc eval failed: {e}"} + except Exception as e: + return {"ok": False, "error": str(e)} + + return run_in_main_thread(_) + + +def _get_function_info(target: str): + def _(): + import ida_funcs + + ea = _resolve_inner(target) + f = ida_funcs.get_func(ea) + if not f: + raise ApiError(f"No function at {hex(ea)}", code="not_found") + + callers = set() + for x in idautils.XrefsTo(f.start_ea, 0): + cf = idaapi.get_func(x.frm) + if cf: + callers.add(cf.start_ea) + callees = set() + for head in idautils.FuncItems(f.start_ea): + for r in idautils.CodeRefsFrom(head, 0): + cf = idaapi.get_func(r) + if cf and cf.start_ea != f.start_ea: + callees.add(cf.start_ea) + + frame_size = None + try: + frame_size = idc.get_frame_size(f.start_ea) + except Exception: + pass + + return { + "ea": hex(f.start_ea), + "end": hex(f.end_ea), + "name": idc.get_func_name(f.start_ea), + "size": f.size(), + "prototype": idc.get_type(f.start_ea) or "", + "flags": f.flags, + "frame_size": frame_size, + "callers_count": len(callers), + "callees_count": len(callees), + "callers": [hex(a) for a in sorted(callers)[:50]], + "callees": [hex(a) for a in sorted(callees)[:50]], + } + + return run_in_main_thread(_) + + +def _get_lvars(target: str): + def _(): + import ida_hexrays + + ea = _resolve_inner(target) + cfunc = ida_hexrays.decompile(ea) + if not cfunc: + raise ApiError("Cannot decompile function", code="hexrays_unavailable") + out = [] + for lv in cfunc.get_lvars(): + try: + tname = str(lv.type()) + except Exception: + tname = "" + out.append( + { + "name": lv.name, + "type": tname, + "is_arg": bool(lv.is_arg_var), + "width": getattr(lv, "width", None), + } + ) + return {"ea": hex(ea), "name": idc.get_func_name(ea), "lvars": out} + + return run_in_main_thread(_) + + +def _segment_at(target: str): + def _(): + import ida_segment + + ea = _resolve_inner(target) + seg = ida_segment.getseg(ea) + if not seg: + raise ApiError(f"No segment at {hex(ea)}", code="not_found") + return { + "ea": hex(ea), + "name": ida_segment.get_segm_name(seg), + "start": hex(seg.start_ea), + "end": hex(seg.end_ea), + "perm": { + "r": bool(seg.perm & ida_segment.SEGPERM_READ), + "w": bool(seg.perm & ida_segment.SEGPERM_WRITE), + "x": bool(seg.perm & ida_segment.SEGPERM_EXEC), + }, + } + + return run_in_main_thread(_) + + +def _search_text(q: str, limit: int = 100): + """Convenience: substring search across both named locations and strings in one call.""" + + def _(): + ql = q.lower() + _ensure_cache() + with _name_cache_lock: + items = list(_name_cache.items()) + names = [] + for name, ea in items: + if ql in name.lower(): + names.append({"ea": hex(ea), "name": name}) + if len(names) >= limit: + break + strings = [] + for s in idautils.Strings(): + try: + v = str(s) + if ql in v.lower(): + strings.append({"ea": hex(s.ea), "value": v}) + if len(strings) >= limit: + break + except Exception: + pass + return {"query": q, "names": names, "strings": strings} + + return run_in_main_thread(_) + + def _get_objc_classes(filter_str: str = None, limit: int = 200): def _(): import re @@ -1239,10 +1466,17 @@ HELP_ROUTES = [ {"method": "GET", "path": "/local_types?limit=", "desc": "List structs/enums from Local Types"}, {"method": "POST", "path": "/create_struct", "desc": '{"name","fields":[{"name","type"},...]}'}, {"method": "GET", "path": "/objc_classes?filter=&limit=", "desc": "Objective-C classes/methods by naming convention"}, + {"method": "GET", "path": "/names/search?q=&limit=", "desc": "Search all named locations (not just functions) by substring"}, {"method": "GET", "path": "/find_immediate?value=&limit=", "desc": "Search for an immediate operand with a given value"}, {"method": "GET", "path": "/find_vtables?segment=&min_entries=&limit=", "desc": "Heuristic vtable-like pointer array search -- experimental"}, {"method": "POST", "path": "/create_type", "desc": '{"decl"} -- raw C declaration for struct/enum/union/typedef in Local Types'}, {"method": "POST", "path": "/xrefs_to_batch", "desc": '{"targets": [...]} -- xrefs_to in one call'}, + {"method": "POST", "path": "/exec_python", "desc": '{"code"} -- run arbitrary Python in IDA; set `result` to return a value'}, + {"method": "POST", "path": "/exec_idc", "desc": '{"code"} -- evaluate an IDC expression'}, + {"method": "GET", "path": "/function_info/", "desc": "Rich function metadata (prototype, frame, callers/callees)"}, + {"method": "GET", "path": "/lvars/", "desc": "Decompiled local variables with their types"}, + {"method": "GET", "path": "/segment_at/", "desc": "Which segment contains an address"}, + {"method": "GET", "path": "/search_text?q=&limit=", "desc": "Substring search across names and strings in one call"}, {"method": "GET", "path": "/reload", "desc": "Reload the plugin's code from disk without restarting IDA"}, ] @@ -1340,6 +1574,8 @@ class IDAHandler(BaseHTTPRequestHandler): self.send_json(_list_local_types(limit=int(qs.get("limit", 200)))) elif path == "/objc_classes": self.send_json(_get_objc_classes(filter_str=qs.get("filter"), limit=int(qs.get("limit", 200)))) + elif path == "/names/search": + self.send_json(_search_names(qs.get("q", ""), limit=int(qs.get("limit", 200)))) elif path == "/find_immediate": self.send_json(_find_immediate(qs.get("value", "0"), limit=int(qs.get("limit", 100)))) elif path == "/find_vtables": @@ -1350,6 +1586,14 @@ class IDAHandler(BaseHTTPRequestHandler): limit=int(qs.get("limit", 50)), ) ) + elif path.startswith("/function_info/"): + self.send_json(_get_function_info(path[len("/function_info/") :])) + elif path.startswith("/lvars/"): + self.send_json(_get_lvars(path[len("/lvars/") :])) + elif path.startswith("/segment_at/"): + self.send_json(_segment_at(path[len("/segment_at/") :])) + elif path == "/search_text": + self.send_json(_search_text(qs.get("q", ""), limit=int(qs.get("limit", 100)))) elif path == "/reload": self.send_json(_reload_plugin_code()) else: @@ -1416,6 +1660,12 @@ class IDAHandler(BaseHTTPRequestHandler): elif path == "/xrefs_to_batch": _require(body, ["targets"]) self.send_json(_xrefs_to_many(body["targets"])) + elif path == "/exec_python": + _require(body, ["code"]) + self.send_json(_exec_python(body["code"])) + elif path == "/exec_idc": + _require(body, ["code"]) + self.send_json(_exec_idc(body["code"])) else: self.send_json({"error": "Unknown endpoint", "code": "not_found"}, 404) @@ -1436,6 +1686,40 @@ class ThreadingHTTPServer(socketserver.ThreadingMixIn, HTTPServer): # ───────────────────────────────────────────── +def _make_settings_form(ida_kernwin, status, bind, port, adv, reg, token): + """Builds the Plugins -> IDA MCP settings dialog. Uses ida_kernwin.Form, which is + binding-agnostic (works whether IDA ships PyQt5 or PySide6).""" + + class _SettingsForm(ida_kernwin.Form): + def __init__(self): + ida_kernwin.Form.__init__( + self, + r"""STARTITEM 0 +BUTTON YES* Apply & Restart +BUTTON CANCEL Cancel +IDA MCP settings + +Status: %s + +<#Address the plugin listens on (0.0.0.0 for remote/SSH)#Bind host:{iBind}> +<#Port to listen on -- leave blank to auto-pick a free one#Port:{iPort}> +<#Address reported back to the MCP server#Advertise host:{iAdv}> +<#MCP server URL for self-registration -- blank disables it#Registry URL:{iReg}> +<#Shared secret -- blank means no authentication#Token:{iToken}> +""" + % status, + { + "iBind": ida_kernwin.Form.StringInput(value=bind), + "iPort": ida_kernwin.Form.StringInput(value=port), + "iAdv": ida_kernwin.Form.StringInput(value=adv), + "iReg": ida_kernwin.Form.StringInput(value=reg), + "iToken": ida_kernwin.Form.StringInput(value=token), + }, + ) + + return _SettingsForm() + + class IdaMcpPlugin(idaapi.plugin_t): flags = idaapi.PLUGIN_KEEP comment = "IDA MCP REST server (auto-port + self-registration)" @@ -1444,10 +1728,12 @@ class IdaMcpPlugin(idaapi.plugin_t): wanted_hotkey = "Ctrl-Shift-M" def init(self): + global _PLUGIN_INSTANCE self.server = None self.thread = None self.heartbeat_stop = None self.heartbeat_thread = None + _PLUGIN_INSTANCE = self self._start() threading.Thread(target=self._build_cache_bg, daemon=True).start() return idaapi.PLUGIN_KEEP @@ -1460,15 +1746,63 @@ class IdaMcpPlugin(idaapi.plugin_t): print(f"[IDA MCP:{PORT}] Cache build error: {e}") def run(self, arg): - if self.server: - self._stop() - else: - self._start() + # Plugins -> IDA MCP opens the settings window. + self._show_settings() + + def _show_settings(self): + import ida_kernwin + + cur_bind = BIND_HOST + cur_port = str(PORT) if PORT else "" + cur_adv = ADVERTISE_HOST + cur_reg = REGISTRY_URL or "" + cur_token = TOKEN or "" + status = f"running on {ADVERTISE_HOST}:{PORT}" if self.server else "stopped" + + got = None + try: + form = _make_settings_form(ida_kernwin, status, cur_bind, cur_port, cur_adv, cur_reg, cur_token) + form.Compile() + if form.Execute() == 1: + got = ( + form.iBind.value.strip(), + form.iPort.value.strip(), + form.iAdv.value.strip(), + form.iReg.value.strip(), + form.iToken.value.strip(), + ) + form.Free() + except Exception as e: + # Form DSL can be finicky across IDA builds -- fall back to simple prompts. + print(f"[IDA MCP] settings Form failed ({e}); using text prompts") + b = ida_kernwin.ask_str(cur_bind, 0, "IDA MCP -- bind host") + if b is not None: + p = ida_kernwin.ask_str(cur_port, 0, "IDA MCP -- port (blank = auto)") + a = ida_kernwin.ask_str(cur_adv, 0, "IDA MCP -- advertise host") + r = ida_kernwin.ask_str(cur_reg, 0, "IDA MCP -- registry URL (blank = off)") + t = ida_kernwin.ask_str(cur_token, 0, "IDA MCP -- token (blank = none)") + got = (b.strip(), (p or "").strip(), (a or "").strip(), (r or "").strip(), (t or "").strip()) + + if got: + self._apply_settings(*got) + ida_kernwin.info( + f"IDA MCP restarted:\n http://{ADVERTISE_HOST}:{PORT}\n registry: {REGISTRY_URL or 'off'}\n auth: {'token' if TOKEN else 'none'}" + ) + + def _apply_settings(self, bind, port_str, adv, reg, token): + global BIND_HOST, ADVERTISE_HOST, REGISTRY_URL, TOKEN, _FORCED_PORT + BIND_HOST = bind or "127.0.0.1" + ADVERTISE_HOST = adv or (BIND_HOST if BIND_HOST != "0.0.0.0" else "127.0.0.1") + REGISTRY_URL = reg or "" + TOKEN = token or None + _FORCED_PORT = int(port_str) if port_str.isdigit() else None + self._stop() + self._start() def _start(self): global PORT, INSTANCE_ID try: - forced = os.environ.get("IDA_MCP_PORT") + forced = _FORCED_PORT if _FORCED_PORT is not None else os.environ.get("IDA_MCP_PORT") PORT = int(forced) if forced else _find_free_port(BASE_PORT, BIND_HOST) INSTANCE_ID = _make_instance_id()