wisedev/ida-mcp-server
1
2
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ida-mcp-server/ida_mcp_plugin.py
WiseDev 35f9a8b1fb Add IDC/Python exec, richer decompile/segment/search endpoints, settings dialog, TLS, batch/overview 
Plugin:
- exec_python / exec_idc endpoints (token-gated arbitrary IDA scripting)
- function_info, lvars, segment_at, search_text (IDA 9.0 API)
- Plugins -> IDA MCP opens a settings dialog (bind/port/advertise/registry/token)
  via ida_kernwin.Form, with a text-prompt fallback; Apply restarts the server live

Server:
- ida_overview (one-shot triage) and ida_batch (many calls, one round-trip)
- native TLS and mutual-TLS support (MCP_TLS_CERT/KEY/CLIENT_CA), MCP_BIND_HOST
- clearer warnings when running plaintext behind an HTTP tunnel

Docs:
- DEPLOY.md: hosting for endpoint availability + confidentiality (VPN back-channel,
  public TLS front door, mutual TLS, firewall, systemd), and what hosting can't do

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-09-09 22:49:46 +03:00

1840 lines
68 KiB
Python
Raw Blame History

"""
IDA MCP Plugin.
Each instance finds a free TCP port on its own, starts an HTTP server on it,
and every 10s sends a heartbeat registration to the MCP server (IDA_MCP_REGISTRY_URL,
defaults to http://127.0.0.1:8888). Open as many IDA instances as you want -- each
registers itself under its own id (filename_pid); the MCP server sees all of them
through ida_list_instances. No hardcoded ports, no manual setup per instance.
Env vars:
IDA_MCP_PORT force a specific port (default: auto-pick)
IDA_MCP_BASE_PORT where auto-pick starts from (default 7777)
IDA_MCP_HOST bind address (default 127.0.0.1; 0.0.0.0 for remote/SSH)
IDA_MCP_ADVERTISE_HOST address to report back to the MCP server (defaults to
IDA_MCP_HOST unless that's 0.0.0.0, then 127.0.0.1 --
set explicitly for a remote machine)
IDA_MCP_REGISTRY_URL MCP server base URL for self-registration (default
http://127.0.0.1:8888; empty disables registration)
IDA_MCP_TOKEN shared secret, checked on incoming requests here and sent
as Authorization: Bearer when registering with the MCP
server -- set the same value on both sides
See README.md for the full endpoint list.
"""
import json
import os
import socket
import socketserver
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
from urllib.request import Request as URLRequest, urlopen
import idaapi
import idautils
import idc
# ─────────────────────────────────────────────
# Config / self-registration
# ─────────────────────────────────────────────
BASE_PORT = int(os.environ.get("IDA_MCP_BASE_PORT", 7777))
BIND_HOST = os.environ.get("IDA_MCP_HOST", "127.0.0.1")
ADVERTISE_HOST = os.environ.get(
"IDA_MCP_ADVERTISE_HOST", BIND_HOST if BIND_HOST != "0.0.0.0" else "127.0.0.1"
)
REGISTRY_URL = os.environ.get("IDA_MCP_REGISTRY_URL", "http://127.0.0.1:8888")
TOKEN = os.environ.get("IDA_MCP_TOKEN")
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:
for p in range(start, start + tries):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
s.bind((host if host != "0.0.0.0" else "", p))
return p
except OSError:
continue
finally:
s.close()
raise RuntimeError(f"No free port found in range {start}-{start + tries}")
def _make_instance_id() -> str:
try:
base = os.path.splitext(os.path.basename(idc.get_input_file_path() or "ida"))[0]
base = base or "ida"
except Exception:
base = "ida"
return f"{base}_{os.getpid()}"
def _registry_call(path: str, payload: dict):
if not REGISTRY_URL:
return
try:
data = json.dumps(payload).encode("utf-8")
req = URLRequest(
f"{REGISTRY_URL}{path}",
data=data,
method="POST",
headers={"Content-Type": "application/json"},
)
if TOKEN:
req.add_header("Authorization", f"Bearer {TOKEN}")
urlopen(req, timeout=5).read()
except Exception as e:
print(f"[IDA MCP:{PORT}] registry {path} failed: {e}")
def _heartbeat_loop(stop_event: threading.Event):
while not stop_event.is_set():
try:
path = idc.get_input_file_path()
except Exception:
path = "unknown"
_registry_call(
"/instances/register",
{
"id": INSTANCE_ID,
"host": ADVERTISE_HOST,
"port": PORT,
"file": path,
"pid": os.getpid(),
},
)
stop_event.wait(HEARTBEAT_INTERVAL)
# ─────────────────────────────────────────────
# Auth
# ─────────────────────────────────────────────
def _check_auth(handler) -> bool:
if not TOKEN:
return True
return handler.headers.get("Authorization", "") == f"Bearer {TOKEN}"
# ─────────────────────────────────────────────
# Name cache (str_* and other non-standard symbols)
# ─────────────────────────────────────────────
_name_cache = {} # { name_str -> ea }
_name_cache_built = False
_name_cache_lock = threading.Lock()
def _build_name_cache():
"""Builds the full name cache via idautils.Names(). Called once."""
global _name_cache, _name_cache_built
cache = {}
for ea, name in idautils.Names():
cache[name] = ea
with _name_cache_lock:
_name_cache = cache
_name_cache_built = True
print(f"[IDA MCP:{PORT}] Name cache built: {len(cache)} entries")
def _ensure_cache():
global _name_cache_built
if not _name_cache_built:
_build_name_cache()
def _lookup_name(name: str) -> int:
if name.startswith("0x") or name.startswith("0X"):
return int(name, 16)
ea = idc.get_name_ea_simple(name)
if ea != idc.BADADDR:
return ea
# The decompiler shows str_ instead of .str. -- try both forms
alt_name = None
if name.startswith("str_"):
alt_name = ".str." + name[4:]
elif name.startswith("str"):
alt_name = ".str" + name[3:]
if alt_name:
ea = idc.get_name_ea_simple(alt_name)
if ea != idc.BADADDR:
return ea
_ensure_cache()
with _name_cache_lock:
ea = _name_cache.get(name, idc.BADADDR)
if ea == idc.BADADDR and alt_name:
ea = _name_cache.get(alt_name, idc.BADADDR)
return ea
# ─────────────────────────────────────────────
# Thread safety
# ─────────────────────────────────────────────
def run_in_main_thread(fn):
"""Runs fn on the IDA main thread synchronously, returns its result."""
result = [None]
error = [None]
event = threading.Event()
def callback():
try:
result[0] = fn()
except Exception as e:
error[0] = e
finally:
event.set()
return 0
idaapi.execute_sync(callback, idaapi.MFF_FAST)
if not event.wait(timeout=90):
raise TimeoutError(
"IDA main thread did not respond in time (busy with another operation?)"
)
if error[0]:
raise error[0]
return result[0]
# ─────────────────────────────────────────────
# Resolve helper
# ─────────────────────────────────────────────
class ApiError(ValueError):
"""ValueError with a machine-readable code, so callers can branch on it
instead of parsing the error text."""
def __init__(self, message: str, code: str = "bad_request"):
super().__init__(message)
self.code = code
def _resolve_inner(name: str) -> int:
if name.startswith("0x") or name.startswith("0X"):
return int(name, 16)
ea = idc.get_name_ea_simple(name)
if ea == idc.BADADDR:
raise ApiError(f"Symbol not found: {name}", code="not_found")
return ea
def _require(body: dict, fields):
missing = [f for f in fields if f not in body or body[f] is None]
if missing:
raise ApiError(f"Missing required field(s): {', '.join(missing)}", code="bad_request")
# ─────────────────────────────────────────────
# IDA operations
# ─────────────────────────────────────────────
def _ping():
def _():
try:
path = idc.get_input_file_path()
except Exception:
path = "unknown"
try:
import ida_hexrays
hexrays_ok = ida_hexrays.init_hexrays_plugin()
except Exception:
hexrays_ok = False
try:
analysis_complete = idaapi.auto_is_ok()
except Exception:
analysis_complete = None
return {
"status": "ok",
"id": INSTANCE_ID,
"host": ADVERTISE_HOST,
"port": PORT,
"file": path,
"hexrays_available": bool(hexrays_ok),
"analysis_complete": analysis_complete,
"name_cache": len(_name_cache) if _name_cache_built else "not built",
"registry_url": REGISTRY_URL or None,
}
return run_in_main_thread(_)
def _list_functions(filter_str=None, limit=200, offset=0):
def _():
import ida_funcs
funcs = []
for ea in idautils.Functions():
name = idc.get_func_name(ea)
if filter_str and filter_str.lower() not in name.lower():
continue
f = ida_funcs.get_func(ea)
if f:
funcs.append({"ea": hex(f.start_ea), "name": name, "size": f.size()})
total = len(funcs)
return {
"total": total,
"offset": offset,
"limit": limit,
"functions": funcs[offset : offset + limit],
}
return run_in_main_thread(_)
def _search_functions(q: str, limit=100):
def _():
results = []
q_lower = q.lower()
for ea in idautils.Functions():
name = idc.get_func_name(ea)
if q_lower in name.lower():
results.append({"ea": hex(ea), "name": name})
if len(results) >= limit:
break
return {"count": len(results), "functions": results}
return run_in_main_thread(_)
def _search_strings(q: str, limit=50):
def _():
results = []
q_lower = q.lower()
for s in idautils.Strings():
try:
val = str(s)
if q_lower in val.lower():
results.append({"ea": hex(s.ea), "value": val})
if len(results) >= limit:
break
except Exception:
pass
return {"count": len(results), "strings": results}
return run_in_main_thread(_)
def _decompile_one(target: str):
"""Returns a dict with the code, or an explicit 'error' field (never embedded in the code text)."""
import ida_funcs
import ida_hexrays
ea = _resolve_inner(target)
name = idc.get_func_name(ea)
f = ida_funcs.get_func(ea)
try:
cfunc = ida_hexrays.decompile(ea)
except Exception as e:
return {"target": target, "ea": hex(ea), "name": name, "error": f"decompile exception: {e}"}
if not cfunc:
return {
"target": target,
"ea": hex(ea),
"name": name,
"error": "decompile failed (no Hex-Rays for this arch, or not a function?)",
}
return {
"target": target,
"ea": hex(ea),
"name": name,
"size": f.size() if f else 0,
"code": str(cfunc),
}
def _decompile(target: str):
def _():
return _decompile_one(target)
return run_in_main_thread(_)
def _decompile_many(targets):
def _():
results = []
for t in targets:
try:
results.append(_decompile_one(t))
except Exception as e:
results.append({"target": t, "error": str(e)})
return {"count": len(results), "results": results}
return run_in_main_thread(_)
def _get_disasm(target: str, count: int = 30, offset: int = 0):
def _():
import ida_lines
import ida_funcs
ea = _resolve_inner(target)
f = ida_funcs.get_func(ea)
end = f.end_ea if f else None
# Known instruction count up front, instead of guessing from what's returned.
total = len(list(idautils.FuncItems(ea))) if f else None
lines = []
cur = ea
n = 0
skipped = 0
while cur != idc.BADADDR:
if end is not None and cur >= end:
break
if skipped < offset:
skipped += 1
cur = idc.next_head(cur, end if end is not None else cur + 0x2000)
continue
if n >= count:
break
raw = idc.generate_disasm_line(cur, 0) or ""
lines.append({"ea": hex(cur), "text": ida_lines.tag_remove(raw)})
cur = idc.next_head(cur, end if end is not None else cur + 0x2000)
n += 1
truncated = cur != idc.BADADDR and (end is None or cur < end)
return {
"ea": hex(ea),
"func_end": hex(end) if end is not None else None,
"total_instructions": total,
"offset": offset,
"returned": len(lines),
"truncated": truncated,
"lines": lines,
}
return run_in_main_thread(_)
def _xrefs_to(target: str):
def _():
ea = _resolve_inner(target)
results = []
for xref in idautils.XrefsTo(ea, 0):
from_ea = xref.frm
func = idaapi.get_func(from_ea)
results.append(
{
"from_ea": hex(from_ea),
"from_func": idc.get_func_name(from_ea) or "",
"func_ea": hex(func.start_ea) if func else hex(from_ea),
}
)
return results
return run_in_main_thread(_)
def _xrefs_to_many(targets):
def _():
result = {}
for t in targets:
try:
ea = _resolve_inner(t)
entries = []
for xref in idautils.XrefsTo(ea, 0):
from_ea = xref.frm
func = idaapi.get_func(from_ea)
entries.append(
{
"from_ea": hex(from_ea),
"from_func": idc.get_func_name(from_ea) or "",
"func_ea": hex(func.start_ea) if func else hex(from_ea),
}
)
result[t] = entries
except Exception as e:
result[t] = {"error": str(e)}
return result
return run_in_main_thread(_)
def _xrefs_from(target: str):
def _():
ea = _resolve_inner(target)
return [
{"to_ea": hex(r), "to_func": idc.get_func_name(r)}
for r in idautils.CodeRefsFrom(ea, 0)
]
return run_in_main_thread(_)
def _get_imports(filter_str=None, limit=500):
def _():
# IDA 9.x moved these from idaapi to ida_nalt and renamed _count -> _qty;
# fall back to the old idaapi names for older IDA versions.
try:
import ida_nalt as _imp
qty = _imp.get_import_module_qty()
except (ImportError, AttributeError):
_imp = idaapi
qty = idaapi.get_import_module_count()
imports = []
for i in range(qty):
mod = _imp.get_import_module_name(i)
def cb(ea, name, ord, mod=mod):
if not filter_str or filter_str.lower() in (name or "").lower():
imports.append(
{"module": mod, "name": name or f"ord_{ord}", "ea": hex(ea)}
)
return True
_imp.enum_import_names(i, cb)
return imports[:limit]
return run_in_main_thread(_)
def _get_segments():
def _():
import ida_segment
segs = []
for ea in idautils.Segments():
seg = ida_segment.getseg(ea)
segs.append(
{
"name": idc.get_segm_name(ea),
"start": hex(seg.start_ea),
"end": hex(seg.end_ea),
"size": seg.end_ea - seg.start_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 segs
return run_in_main_thread(_)
def _get_entry_points():
def _():
import ida_entry
result = []
for i in range(ida_entry.get_entry_qty()):
ordinal = ida_entry.get_entry_ordinal(i)
ea = ida_entry.get_entry(ordinal)
name = ida_entry.get_entry_name(ordinal)
result.append({"ordinal": ordinal, "ea": hex(ea), "name": name})
return result
return run_in_main_thread(_)
def _get_prototype(target: str):
def _():
ea = _resolve_inner(target)
return {"ea": hex(ea), "name": idc.get_func_name(ea), "prototype": idc.get_type(ea) or ""}
return run_in_main_thread(_)
def _set_prototype(target: str, prototype: str):
def _():
ea = _resolve_inner(target)
decl = prototype if prototype.rstrip().endswith(";") else prototype + ";"
ok = idc.set_type(ea, decl)
if not ok:
raise ValueError(f"IDA rejected prototype: {prototype}")
return {"ok": True, "ea": hex(ea), "prototype": idc.get_type(ea)}
return run_in_main_thread(_)
def _set_lvar_type(function: str, lvar_name: str, type_str: str):
"""Experimental: the Hex-Rays API for changing a local variable's type can
differ slightly across IDA versions. Errors are surfaced as-is so it's
clear what to fix."""
def _():
import ida_hexrays
import ida_typeinf
ea = _resolve_inner(function)
cfunc = ida_hexrays.decompile(ea)
if not cfunc:
raise ValueError("Cannot decompile function")
tif = ida_typeinf.tinfo_t()
decl = type_str.strip()
parseable = decl if decl.endswith(";") else decl + " x;"
if not ida_typeinf.parse_decl(tif, None, parseable, ida_typeinf.PT_SIL) and not tif.get_named_type(
None, type_str
):
raise ValueError(f"Cannot parse type: {type_str}")
for lvar in cfunc.get_lvars():
if lvar.name == lvar_name:
if not lvar.set_lvar_type(tif):
raise ValueError(
"set_lvar_type() rejected by Hex-Rays (incompatible type or read-only var)"
)
cfunc.save_user_lvar_settings()
return {"ok": True, "function": idc.get_func_name(ea), "lvar": lvar_name, "type": type_str}
raise ValueError(f"Local variable '{lvar_name}' not found")
return run_in_main_thread(_)
def _rename_function(target: str, new_name: str):
def _():
ea = _resolve_inner(target)
old_name = idc.get_func_name(ea)
ok = idc.set_name(ea, new_name, idc.SN_NOWARN)
if not ok:
raise ValueError(f"IDA rejected name '{new_name}' (duplicate or invalid?)")
with _name_cache_lock:
if old_name in _name_cache:
del _name_cache[old_name]
_name_cache[new_name] = ea
return {"ok": True, "ea": hex(ea), "old_name": old_name, "new_name": new_name}
return run_in_main_thread(_)
def _set_comment(target: str, comment: str):
def _():
ea = _resolve_inner(target)
idc.set_func_cmt(ea, comment, 0)
return {"ok": True}
return run_in_main_thread(_)
def _rename_local(function: str, old_name: str, new_name: str):
def _():
import ida_hexrays
ea = _resolve_inner(function)
cfunc = ida_hexrays.decompile(ea)
if not cfunc:
raise ValueError("Cannot decompile function")
for lvar in cfunc.get_lvars():
if lvar.name == old_name:
lvar.name = new_name
cfunc.save_user_lvar_settings()
return {"ok": True}
raise ValueError(f"Local variable '{old_name}' not found")
return run_in_main_thread(_)
def _get_name_value(name: str):
def _():
ea = _lookup_name(name)
if ea == idc.BADADDR:
raise ValueError(f"Not found: {name}")
s = idc.get_strlit_contents(ea, -1, idc.STRTYPE_C)
if s is not None:
return {
"ea": hex(ea),
"type": "string",
"value": s.decode("utf-8", errors="replace"),
}
val = idc.get_wide_dword(ea)
return {"ea": hex(ea), "type": "dword", "value": hex(val)}
return run_in_main_thread(_)
def _rebuild_name_cache():
def _():
global _name_cache_built
_name_cache_built = False
_build_name_cache()
return {"ok": True, "entries": len(_name_cache)}
return run_in_main_thread(_)
def _reload_plugin_code():
"""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
import sys
modname = __name__
mod = sys.modules.get(modname)
if mod is None:
raise ValueError(f"Cannot find own module in sys.modules as '{modname}'")
saved_port, saved_id = PORT, INSTANCE_ID
try:
importlib.reload(mod)
except Exception:
# IDA-loaded plugin modules often have no usable importlib spec
# (loaded via IDA's own mechanism, not a normal import) — fall back
# to exec'ing the file straight into the module's own namespace.
try:
with open(mod.__file__, "r", encoding="utf-8") as f:
src = f.read()
exec(compile(src, mod.__file__, "exec"), mod.__dict__)
except Exception as e:
raise ValueError(f"Reload failed (syntax error in the file?): {e}")
mod.PORT = saved_port
mod.INSTANCE_ID = saved_id
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,
"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(_)
# ─────────────────────────────────────────────
# Notes (netnode-backed -- persist across sessions and tools)
# ─────────────────────────────────────────────
_NOTES_NODE_NAME = "$ ida_mcp_notes"
def _notes_node():
import ida_netnode
return ida_netnode.netnode(_NOTES_NODE_NAME, 0, True)
def _get_notes(target: str):
def _():
ea = _resolve_inner(target)
val = _notes_node().supval(ea)
text = val.decode("utf-8", errors="replace") if val else ""
return {"ea": hex(ea), "note": text}
return run_in_main_thread(_)
def _set_notes(target: str, note: str):
def _():
ea = _resolve_inner(target)
nn = _notes_node()
if note:
nn.supset(ea, note.encode("utf-8"))
else:
nn.supdel(ea)
return {"ok": True, "ea": hex(ea)}
return run_in_main_thread(_)
# ─────────────────────────────────────────────
# Bytes / binary search
# ─────────────────────────────────────────────
def _get_bytes(target: str, size: int = 64):
def _():
ea = _resolve_inner(target)
data = idc.get_bytes(ea, size)
if data is None:
raise ValueError(f"Cannot read {size} bytes at {hex(ea)} (unmapped?)")
return {"ea": hex(ea), "size": size, "hex": data.hex()}
return run_in_main_thread(_)
def _find_pattern(pattern: str, max_results: int = 50):
"""Searches for an IDA-style hex pattern ("48 8B ?? ??") across the whole image.
Experimental: the bin_search API has changed across IDA versions; falls
back to the legacy idc.find_binary."""
def _():
results = []
modern_ok = False
modern_err = None
try:
import ida_bytes
patterns = ida_bytes.compiled_binpat_vec_t()
if not ida_bytes.parse_binpat_str(patterns, 0, pattern, 16):
modern_err = "parse_binpat_str() returned False (bad pattern syntax?)"
else:
modern_ok = True
ea = 0
while len(results) < max_results:
found = ida_bytes.bin_search(ea, idaapi.BADADDR, patterns, ida_bytes.BIN_SEARCH_FORWARD)
found_ea = found[0] if isinstance(found, tuple) else found
if found_ea is None or found_ea == idaapi.BADADDR:
break
results.append(hex(found_ea))
ea = found_ea + 1
except Exception as e:
modern_ok = False
modern_err = f"{type(e).__name__}: {e}"
results = []
if not modern_ok:
try:
ea = idc.find_binary(0, idc.SEARCH_DOWN, pattern)
while ea != idc.BADADDR and len(results) < max_results:
results.append(hex(ea))
ea = idc.find_binary(ea + 1, idc.SEARCH_DOWN, pattern)
except Exception as e:
raise ValueError(
f"bin_search path failed ({modern_err}); legacy find_binary also failed: {type(e).__name__}: {e}"
)
return {"pattern": pattern, "count": len(results), "matches": results, "used_legacy_find_binary": not modern_ok}
return run_in_main_thread(_)
def _patch_bytes(target: str, hex_bytes: str):
def _():
import ida_bytes
ea = _resolve_inner(target)
data = bytes.fromhex(hex_bytes.replace(" ", ""))
for i, b in enumerate(data):
ida_bytes.patch_byte(ea + i, b)
return {"ok": True, "ea": hex(ea), "patched_bytes": len(data)}
return run_in_main_thread(_)
def _make_data(target: str, data_type: str, struct_name: str = None):
"""Experimental: some struct-related idc functions are legacy and may have moved in IDA 9.x."""
def _():
import ida_bytes
ea = _resolve_inner(target)
sizes = {"byte": (ida_bytes.FF_BYTE, 1), "word": (ida_bytes.FF_WORD, 2), "dword": (ida_bytes.FF_DWORD, 4), "qword": (ida_bytes.FF_QWORD, 8)}
if data_type in sizes:
flag, size = sizes[data_type]
ok = ida_bytes.create_data(ea, flag, size, idaapi.BADADDR)
elif data_type == "struct":
if not struct_name:
raise ValueError("struct_name is required for data_type='struct'")
tid = idc.get_struc_id(struct_name)
if tid == idc.BADADDR:
raise ValueError(f"Struct not found: {struct_name}")
size = idc.get_struc_size(tid)
ok = ida_bytes.create_struct(ea, size, tid)
else:
raise ValueError(f"Unknown data_type: {data_type} (use byte/word/dword/qword/struct)")
if not ok:
raise ValueError("Failed to create data (already defined at this address, or bad location?)")
return {"ok": True, "ea": hex(ea), "type": data_type}
return run_in_main_thread(_)
# ─────────────────────────────────────────────
# Comments at a specific address (set_comment is whole-function)
# ─────────────────────────────────────────────
def _get_line_comment(target: str, repeatable: bool = False):
def _():
ea = _resolve_inner(target)
return {"ea": hex(ea), "comment": idc.get_cmt(ea, repeatable) or ""}
return run_in_main_thread(_)
def _set_line_comment(target: str, comment: str, repeatable: bool = False):
def _():
ea = _resolve_inner(target)
idc.set_cmt(ea, comment, repeatable)
return {"ok": True, "ea": hex(ea)}
return run_in_main_thread(_)
# ─────────────────────────────────────────────
# Triage: top functions, call graph
# ─────────────────────────────────────────────
def _top_functions(by: str = "xrefs", limit: int = 20):
def _():
import ida_funcs
stats = []
for ea in idautils.Functions():
f = ida_funcs.get_func(ea)
if not f:
continue
if by == "size":
score = f.size()
else:
score = sum(1 for _ in idautils.XrefsTo(ea, 0))
stats.append({"ea": hex(ea), "name": idc.get_func_name(ea), by: score})
stats.sort(key=lambda x: x[by], reverse=True)
return stats[:limit]
return run_in_main_thread(_)
def _call_graph(target: str, depth: int = 2, direction: str = "callees", limit_per_node: int = 30):
def _():
import ida_funcs
start = _resolve_inner(target)
def neighbors(ea):
if direction == "callers":
found = set()
for x in idautils.XrefsTo(ea, 0):
func = idaapi.get_func(x.frm)
if func:
found.add(func.start_ea)
return list(found)[:limit_per_node]
f = ida_funcs.get_func(ea)
if not f:
return []
found = set()
for head in idautils.FuncItems(ea):
for r in idautils.CodeRefsFrom(head, 0):
func = idaapi.get_func(r)
if func and func.start_ea != ea:
found.add(func.start_ea)
return list(found)[:limit_per_node]
nodes = {start: 0}
edges = []
frontier = [start]
for d in range(depth):
next_frontier = []
for node in frontier:
for n in neighbors(node):
edges.append({"from": hex(node), "to": hex(n)})
if n not in nodes:
nodes[n] = d + 1
next_frontier.append(n)
frontier = next_frontier
return {
"direction": direction,
"nodes": [{"ea": hex(ea), "name": idc.get_func_name(ea), "depth": d} for ea, d in nodes.items()],
"edges": edges,
}
return run_in_main_thread(_)
# ─────────────────────────────────────────────
# Functions: create/undefine
# ─────────────────────────────────────────────
def _make_function(target: str):
def _():
import ida_funcs
ea = _resolve_inner(target)
if not ida_funcs.add_func(ea):
raise ValueError(f"add_func failed at {hex(ea)} (already a function, or not valid code?)")
return {"ok": True, "ea": hex(ea), "name": idc.get_func_name(ea)}
return run_in_main_thread(_)
def _undefine(target: str):
def _():
import ida_funcs
ea = _resolve_inner(target)
f = idaapi.get_func(ea)
if f:
ida_funcs.del_func(f.start_ea)
idc.del_items(ea, idc.DELIT_SIMPLE, 1)
return {"ok": True, "ea": hex(ea)}
return run_in_main_thread(_)
# ─────────────────────────────────────────────
# Local types (structs/enums)
# ─────────────────────────────────────────────
def _list_local_types(limit: int = 200):
def _():
import ida_typeinf
til = ida_typeinf.get_idati()
qty = None
for getter in (
lambda: ida_typeinf.get_ordinal_qty(til),
lambda: til.get_ordinal_qty(),
lambda: ida_typeinf.get_ordinal_count(til),
):
try:
qty = getter()
break
except Exception:
continue
if qty is None:
raise ValueError("Cannot determine local-types count on this IDA version (API changed)")
results = []
for ordinal in range(1, qty + 1):
tif = ida_typeinf.tinfo_t()
if tif.get_numbered_type(til, ordinal):
results.append(
{
"ordinal": ordinal,
"name": tif.get_type_name() or "",
"size": tif.get_size() if tif.get_size() != idaapi.BADSIZE else None,
"is_struct": tif.is_struct(),
"is_enum": tif.is_enum(),
}
)
if len(results) >= limit:
break
return results
return run_in_main_thread(_)
def _create_struct(name: str, fields):
"""fields: [{"name": "field1", "type": "int"}, ...] -- adds a struct to Local Types."""
def _():
decl = f"struct {name} {{\n"
for f in fields:
decl += f" {f['type']} {f['name']};\n"
decl += "};"
errors = idc.parse_decls(decl, False)
if errors != 0:
raise ValueError(f"parse_decls reported {errors} error(s) for:\n{decl}")
return {"ok": True, "name": name, "decl": decl}
return run_in_main_thread(_)
def _create_local_type(decl: str):
"""Generalized version of _create_struct -- takes a raw C declaration
(struct/enum/union/typedef), not just a struct built from a field list."""
def _():
errors = idc.parse_decls(decl, False)
if errors != 0:
raise ValueError(f"parse_decls reported {errors} error(s) for:\n{decl}")
return {"ok": True, "decl": decl}
return run_in_main_thread(_)
# ─────────────────────────────────────────────
# Objective-C: classes/methods by naming convention (+[Class sel:], -[Class sel:])
# ─────────────────────────────────────────────
_OBJC_METHOD_RE = None # compiled lazily
def _find_immediate(value: str, limit: int = 100):
"""Searches all instructions for an immediate operand matching the given value."""
def _():
try:
target = int(value, 0)
except ValueError:
raise ValueError(f"Cannot parse value: {value} (use decimal or 0x-hex)")
results = []
for func_ea in idautils.Functions():
for ea in idautils.FuncItems(func_ea):
for n in range(6):
try:
optype = idc.get_operand_type(ea, n)
except Exception:
break
if optype == idc.o_void:
break
if optype == idc.o_imm and idc.get_operand_value(ea, n) == target:
results.append({"ea": hex(ea), "func": idc.get_func_name(ea), "operand": n})
if len(results) >= limit:
return {"value": hex(target), "count": len(results), "matches": results}
return {"value": hex(target), "count": len(results), "matches": results}
return run_in_main_thread(_)
def _find_vtables(segment_filter: str = None, min_entries: int = 2, limit: int = 50):
"""Heuristic, experimental: looks for runs of consecutive pointers in
non-executable segments where each one points to the start of a real
function -- the classic vtable signature. Can be slow on large binaries
and gives false positives; pass segment_filter (e.g. "__const"/"__data")
to narrow the scan."""
def _():
import ida_segment
import ida_funcs
ptr_size = 8 if idaapi.get_inf_structure().is_64bit() else 4
get_ptr = idc.get_qword if ptr_size == 8 else idc.get_wide_dword
def is_func_start(val):
if not val or val == idc.BADADDR:
return False
f = ida_funcs.get_func(val)
return bool(f and f.start_ea == val)
candidates = []
for seg_ea in idautils.Segments():
seg = ida_segment.getseg(seg_ea)
if seg.perm & ida_segment.SEGPERM_EXEC:
continue
name = idc.get_segm_name(seg_ea)
if segment_filter and segment_filter.lower() not in name.lower():
continue
ea = seg.start_ea
while ea + ptr_size <= seg.end_ea and len(candidates) < limit:
if not is_func_start(get_ptr(ea)):
ea += ptr_size
continue
run_start = ea
entries = []
cur = ea
while cur + ptr_size <= seg.end_ea:
ptr = get_ptr(cur)
if not is_func_start(ptr):
break
entries.append({"ea": hex(cur), "func_ea": hex(ptr), "func_name": idc.get_func_name(ptr)})
cur += ptr_size
if len(entries) >= min_entries:
candidates.append({"ea": hex(run_start), "segment": name, "entries": entries})
ea = cur if cur > run_start else run_start + ptr_size
if len(candidates) >= limit:
break
return {"count": len(candidates), "vtables": candidates}
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
global _OBJC_METHOD_RE
if _OBJC_METHOD_RE is None:
_OBJC_METHOD_RE = re.compile(r"^([+-])\[([\w]+)(\([\w]+\))?\s+([^\]]+)\]$")
_ensure_cache()
classes = {}
# 1) class objects themselves, from runtime symbols
with _name_cache_lock:
names = list(_name_cache.items())
for name, ea in names:
if name.startswith("_OBJC_CLASS_$_"):
cls = name[len("_OBJC_CLASS_$_") :]
classes.setdefault(cls, {"name": cls, "ea": hex(ea), "methods": []})
# 2) methods, from functions named +[Class sel] / -[Class sel]
for ea in idautils.Functions():
fname = idc.get_func_name(ea)
m = _OBJC_METHOD_RE.match(fname)
if not m:
continue
kind, cls, category, selector = m.groups()
entry = classes.setdefault(cls, {"name": cls, "ea": None, "methods": []})
entry["methods"].append(
{
"kind": "class" if kind == "+" else "instance",
"selector": selector,
"category": category.strip("()") if category else None,
"ea": hex(ea),
}
)
result = list(classes.values())
if filter_str:
fl = filter_str.lower()
result = [c for c in result if fl in c["name"].lower()]
return {"count": len(result), "classes": result[:limit]}
return run_in_main_thread(_)
# ─────────────────────────────────────────────
# HTTP handler
# ─────────────────────────────────────────────
HELP_ROUTES = [
{"method": "GET", "path": "/ping", "desc": "Instance status, Hex-Rays availability"},
{"method": "GET", "path": "/help", "desc": "This list"},
{"method": "GET", "path": "/functions?filter=&limit=&offset=", "desc": "List functions"},
{"method": "GET", "path": "/functions/search?q=&limit=", "desc": "Search functions by substring"},
{"method": "GET", "path": "/strings/search?q=&limit=", "desc": "Search strings"},
{"method": "GET", "path": "/name/<name_or_0xADDR>", "desc": "Value at a named symbol"},
{"method": "GET", "path": "/name_cache/rebuild", "desc": "Rebuild the name cache"},
{"method": "GET", "path": "/imports?filter=", "desc": "Import table"},
{"method": "GET", "path": "/segments", "desc": "Segments with permissions"},
{"method": "GET", "path": "/entry_points", "desc": "Entry points"},
{"method": "GET", "path": "/decompile/<target>", "desc": "Decompile a function"},
{"method": "POST", "path": "/decompile_batch", "desc": '{"targets": [...]} -- decompile in one call'},
{"method": "GET", "path": "/disasm/<target>?count=", "desc": "Disassemble a function"},
{"method": "GET", "path": "/xrefs_to/<target>", "desc": "Who references this address"},
{"method": "GET", "path": "/xrefs_from/<target>", "desc": "What this function references"},
{"method": "GET", "path": "/prototype/<target>", "desc": "Function prototype"},
{"method": "POST", "path": "/set_prototype", "desc": '{"target","prototype"}'},
{"method": "POST", "path": "/set_lvar_type", "desc": '{"function","lvar_name","type"} -- experimental'},
{"method": "GET", "path": "/notes/<target>", "desc": "Note stored at an address"},
{"method": "POST", "path": "/notes", "desc": '{"target","note"} -- note="" deletes it'},
{"method": "POST", "path": "/rename_function", "desc": '{"target","new_name"}'},
{"method": "POST", "path": "/rename_local", "desc": '{"function","old_name","new_name"}'},
{"method": "POST", "path": "/set_comment", "desc": '{"target","comment"}'},
{"method": "GET", "path": "/bytes/<target>?size=", "desc": "Raw bytes at an address (hex)"},
{"method": "GET", "path": "/find?pattern=&limit=", "desc": 'Search a hex pattern, e.g. "48 8B ?? ??" -- experimental'},
{"method": "POST", "path": "/patch_bytes", "desc": '{"target","hex_bytes"} -- writes to the binary!'},
{"method": "POST", "path": "/make_data", "desc": '{"target","data_type":"byte|word|dword|qword|struct","struct_name"} -- experimental'},
{"method": "GET", "path": "/line_comment/<target>?repeatable=", "desc": "Comment at a specific address (not the whole function)"},
{"method": "POST", "path": "/line_comment", "desc": '{"target","comment","repeatable"}'},
{"method": "GET", "path": "/top_functions?by=xrefs|size&limit=", "desc": "Triage: most-referenced/largest functions"},
{"method": "GET", "path": "/call_graph/<target>?depth=&direction=callees|callers", "desc": "Call graph N levels deep"},
{"method": "POST", "path": "/make_function", "desc": '{"target"} -- create a function at an address'},
{"method": "POST", "path": "/undefine", "desc": '{"target"} -- remove function/data definition at an address'},
{"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/<target>", "desc": "Rich function metadata (prototype, frame, callers/callees)"},
{"method": "GET", "path": "/lvars/<target>", "desc": "Decompiled local variables with their types"},
{"method": "GET", "path": "/segment_at/<ea>", "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"},
]
class IDAHandler(BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
pass
def send_json(self, data, code=200):
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def read_body(self):
length = int(self.headers.get("Content-Length", 0))
return json.loads(self.rfile.read(length)) if length else {}
def parse_path(self):
parsed = urlparse(self.path)
qs = parse_qs(parsed.query)
return parsed.path, {k: v[0] for k, v in qs.items()}
def do_GET(self):
try:
if not _check_auth(self):
self.send_json({"error": "unauthorized", "code": "unauthorized"}, 401)
return
path, qs = self.parse_path()
if path == "/ping":
self.send_json(_ping())
elif path == "/help":
self.send_json({"instance": INSTANCE_ID, "port": PORT, "routes": HELP_ROUTES})
elif path == "/functions":
self.send_json(
_list_functions(
filter_str=qs.get("filter"),
limit=int(qs.get("limit", 200)),
offset=int(qs.get("offset", 0)),
)
)
elif path == "/functions/search":
self.send_json(_search_functions(q=qs.get("q", ""), limit=int(qs.get("limit", 100))))
elif path == "/strings/search":
self.send_json(_search_strings(q=qs.get("q", ""), limit=int(qs.get("limit", 50))))
elif path == "/imports":
self.send_json(_get_imports(filter_str=qs.get("filter")))
elif path == "/segments":
self.send_json(_get_segments())
elif path == "/entry_points":
self.send_json(_get_entry_points())
elif path.startswith("/name/"):
self.send_json(_get_name_value(path[len("/name/") :]))
elif path == "/name_cache/rebuild":
self.send_json(_rebuild_name_cache())
elif path.startswith("/decompile/"):
self.send_json(_decompile(path[len("/decompile/") :]))
elif path.startswith("/disasm/"):
self.send_json(
_get_disasm(
path[len("/disasm/") :], count=int(qs.get("count", 30)), offset=int(qs.get("offset", 0))
)
)
elif path.startswith("/xrefs_to/"):
self.send_json(_xrefs_to(path[len("/xrefs_to/") :]))
elif path.startswith("/xrefs_from/"):
self.send_json(_xrefs_from(path[len("/xrefs_from/") :]))
elif path.startswith("/prototype/"):
self.send_json(_get_prototype(path[len("/prototype/") :]))
elif path.startswith("/notes/"):
self.send_json(_get_notes(path[len("/notes/") :]))
elif path.startswith("/bytes/"):
self.send_json(_get_bytes(path[len("/bytes/") :], size=int(qs.get("size", 64))))
elif path == "/find":
self.send_json(_find_pattern(qs.get("pattern", ""), max_results=int(qs.get("limit", 50))))
elif path.startswith("/line_comment/"):
self.send_json(
_get_line_comment(path[len("/line_comment/") :], repeatable=qs.get("repeatable") == "1")
)
elif path == "/top_functions":
self.send_json(_top_functions(by=qs.get("by", "xrefs"), limit=int(qs.get("limit", 20))))
elif path.startswith("/call_graph/"):
self.send_json(
_call_graph(
path[len("/call_graph/") :],
depth=int(qs.get("depth", 2)),
direction=qs.get("direction", "callees"),
)
)
elif path == "/local_types":
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":
self.send_json(
_find_vtables(
segment_filter=qs.get("segment"),
min_entries=int(qs.get("min_entries", 2)),
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:
self.send_json({"error": "Unknown endpoint", "code": "not_found"}, 404)
except ApiError as e:
self.send_json({"error": str(e), "code": e.code}, 400)
except ValueError as e:
self.send_json({"error": str(e), "code": "bad_request"}, 400)
except Exception as e:
self.send_json({"error": str(e), "code": "internal_error"}, 500)
def do_POST(self):
try:
if not _check_auth(self):
self.send_json({"error": "unauthorized", "code": "unauthorized"}, 401)
return
body = self.read_body()
path, _ = self.parse_path()
if path == "/rename_function":
_require(body, ["target", "new_name"])
self.send_json(_rename_function(body["target"], body["new_name"]))
elif path == "/rename_local":
_require(body, ["function", "old_name", "new_name"])
self.send_json(_rename_local(body["function"], body["old_name"], body["new_name"]))
elif path == "/set_comment":
_require(body, ["target", "comment"])
self.send_json(_set_comment(body["target"], body["comment"]))
elif path == "/notes":
_require(body, ["target", "note"])
self.send_json(_set_notes(body["target"], body["note"]))
elif path == "/set_prototype":
_require(body, ["target", "prototype"])
self.send_json(_set_prototype(body["target"], body["prototype"]))
elif path == "/set_lvar_type":
_require(body, ["function", "lvar_name", "type"])
self.send_json(_set_lvar_type(body["function"], body["lvar_name"], body["type"]))
elif path == "/decompile_batch":
_require(body, ["targets"])
self.send_json(_decompile_many(body["targets"]))
elif path == "/patch_bytes":
_require(body, ["target", "hex_bytes"])
self.send_json(_patch_bytes(body["target"], body["hex_bytes"]))
elif path == "/make_data":
_require(body, ["target", "data_type"])
self.send_json(_make_data(body["target"], body["data_type"], body.get("struct_name")))
elif path == "/line_comment":
_require(body, ["target", "comment"])
self.send_json(_set_line_comment(body["target"], body["comment"], bool(body.get("repeatable", False))))
elif path == "/make_function":
_require(body, ["target"])
self.send_json(_make_function(body["target"]))
elif path == "/undefine":
_require(body, ["target"])
self.send_json(_undefine(body["target"]))
elif path == "/create_struct":
_require(body, ["name", "fields"])
self.send_json(_create_struct(body["name"], body["fields"]))
elif path == "/create_type":
_require(body, ["decl"])
self.send_json(_create_local_type(body["decl"]))
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)
except ApiError as e:
self.send_json({"error": str(e), "code": e.code}, 400)
except ValueError as e:
self.send_json({"error": str(e), "code": "bad_request"}, 400)
except Exception as e:
self.send_json({"error": str(e), "code": "internal_error"}, 500)
class ThreadingHTTPServer(socketserver.ThreadingMixIn, HTTPServer):
daemon_threads = True
# ─────────────────────────────────────────────
# IDA Plugin
# ─────────────────────────────────────────────
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)"
help = "Exposes IDA Pro via HTTP for Claude/any MCP client"
wanted_name = "IDA MCP"
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
def _build_cache_bg(self):
time.sleep(3) # give IDA time to finish loading the database
try:
run_in_main_thread(_build_name_cache)
except Exception as e:
print(f"[IDA MCP:{PORT}] Cache build error: {e}")
def run(self, arg):
# 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 = _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()
self.server = ThreadingHTTPServer((BIND_HOST, PORT), IDAHandler)
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
self.heartbeat_stop = threading.Event()
self.heartbeat_thread = threading.Thread(
target=_heartbeat_loop, args=(self.heartbeat_stop,), daemon=True
)
self.heartbeat_thread.start()
print(
f"[IDA MCP] Started: id={INSTANCE_ID} http://{ADVERTISE_HOST}:{PORT} "
f"(bind {BIND_HOST}:{PORT}), registry={REGISTRY_URL or 'off'}"
)
except Exception as e:
print(f"[IDA MCP] Failed to start: {e}")
def _stop(self):
if self.heartbeat_stop:
self.heartbeat_stop.set()
_registry_call("/instances/unregister", {"id": INSTANCE_ID})
if self.server:
self.server.shutdown()
self.server = None
print(f"[IDA MCP:{PORT}] Stopped.")
def term(self):
self._stop()
def PLUGIN_ENTRY():
return IdaMcpPlugin()
Reference in a new issue View git blame Copy permalink