""" 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/", "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/", "desc": "Decompile a function"}, {"method": "POST", "path": "/decompile_batch", "desc": '{"targets": [...]} -- decompile in one call'}, {"method": "GET", "path": "/disasm/?count=", "desc": "Disassemble a function"}, {"method": "GET", "path": "/xrefs_to/", "desc": "Who references this address"}, {"method": "GET", "path": "/xrefs_from/", "desc": "What this function references"}, {"method": "GET", "path": "/prototype/", "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/", "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/?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/?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/?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()