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 65ed039896 Add dynamic multi-instance IDA MCP server and plugin 
Self-registering IDA plugin (auto port allocation, heartbeat-based
instance registry) plus an MCP server exposing ~40 tools for
decompilation, xrefs, type/struct editing, byte-level search and
patching, Objective-C class recovery, and cross-instance comparison.
2026-07-31 13:23:46 +03:00

1506 lines
56 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()
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 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."""
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()
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.",
}
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 _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": "/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": "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 == "/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 == "/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"]))
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
# ─────────────────────────────────────────────
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):
self.server = None
self.thread = None
self.heartbeat_stop = None
self.heartbeat_thread = None
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):
if self.server:
self._stop()
else:
self._start()
def _start(self):
global PORT, INSTANCE_ID
try:
forced = 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