# Directory Structure
```
├── .github
│ ├── FUNDING.yml
│ └── logo.png
├── .gitignore
├── .python-version
├── ida-plugin.json
├── LICENSE
├── pyproject.toml
├── README.md
├── src
│ └── ida_pro_mcp
│ ├── __init__.py
│ ├── __main__.py
│ ├── idalib_server.py
│ ├── mcp-plugin.py
│ └── server.py
├── uv-package.sh
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.11
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv*/
venv*/
# Logs
*.log
# Test files
*.idb
*.i64
*.nam
*.id0
*.id1
*.id2
*.til
*.elf
# Generated server files
server_generated.py
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# IDA Pro MCP
Simple [MCP Server](https://modelcontextprotocol.io/introduction) to allow vibe reversing in IDA Pro.
https://github.com/user-attachments/assets/6ebeaa92-a9db-43fa-b756-eececce2aca0
The binaries and prompt for the video are available in the [mcp-reversing-dataset](https://github.com/mrexodia/mcp-reversing-dataset) repository.
Available functionality:
- `check_connection()`: Check if the IDA plugin is running.
- `get_metadata()`: Get metadata about the current IDB.
- `get_function_by_name(name)`: Get a function by its name.
- `get_function_by_address(address)`: Get a function by its address.
- `get_current_address()`: Get the address currently selected by the user.
- `get_current_function()`: Get the function currently selected by the user.
- `convert_number(text, size)`: Convert a number (decimal, hexadecimal) to different representations.
- `list_functions(offset, count)`: List all functions in the database (paginated).
- `list_globals_filter(offset, count, filter)`: List matching globals in the database (paginated, filtered).
- `list_globals(offset, count)`: List all globals in the database (paginated).
- `list_imports(offset, count)`: List all imported symbols with their name and module (paginated).
- `list_strings_filter(offset, count, filter)`: List matching strings in the database (paginated, filtered).
- `list_strings(offset, count)`: List all strings in the database (paginated).
- `list_local_types()`: List all Local types in the database.
- `decompile_function(address)`: Decompile a function at the given address.
- `disassemble_function(start_address)`: Get assembly code for a function (API-compatible with older IDA builds).
- `get_xrefs_to(address)`: Get all cross references to the given address.
- `get_xrefs_to_field(struct_name, field_name)`: Get all cross references to a named struct field (member).
- `get_callees(function_address)`: Get all the functions called (callees) by the function at function_address.
- `get_callers(function_address)`: Get all callers of the given address.
- `get_entry_points()`: Get all entry points in the database.
- `set_comment(address, comment)`: Set a comment for a given address in the function disassembly and pseudocode.
- `rename_local_variable(function_address, old_name, new_name)`: Rename a local variable in a function.
- `rename_global_variable(old_name, new_name)`: Rename a global variable.
- `set_global_variable_type(variable_name, new_type)`: Set a global variable's type.
- `patch_address_assembles(address, instructions)`: <no description>.
- `get_global_variable_value_by_name(variable_name)`: Read a global variable's value (if known at compile-time).
- `get_global_variable_value_at_address(address)`: Read a global variable's value by its address (if known at compile-time).
- `rename_function(function_address, new_name)`: Rename a function.
- `set_function_prototype(function_address, prototype)`: Set a function's prototype.
- `declare_c_type(c_declaration)`: Create or update a local type from a C declaration.
- `set_local_variable_type(function_address, variable_name, new_type)`: Set a local variable's type.
- `get_stack_frame_variables(function_address)`: Retrieve the stack frame variables for a given function.
- `get_defined_structures()`: Returns a list of all defined structures.
- `analyze_struct_detailed(name)`: Detailed analysis of a structure with all fields.
- `get_struct_at_address(address, struct_name)`: Get structure field values at a specific address.
- `get_struct_info_simple(name)`: Simple function to get basic structure information.
- `search_structures(filter)`: Search for structures by name pattern.
- `rename_stack_frame_variable(function_address, old_name, new_name)`: Change the name of a stack variable for an IDA
function.
- `create_stack_frame_variable(function_address, offset, variable_name, type_name)`: For a given function, create a stack variable at an offset and with a specific type.
- `set_stack_frame_variable_type(function_address, variable_name, type_name)`: For a given disassembled function, set
the type of a stack variable.
- `delete_stack_frame_variable(function_address, variable_name)`: Delete the named stack variable for a given function.
- `read_memory_bytes(memory_address, size)`: Read bytes at a given address.
- `data_read_byte(address)`: Read the 1 byte value at the specified address.
- `data_read_word(address)`: Read the 2 byte value at the specified address as a WORD.
- `data_read_dword(address)`: Read the 4 byte value at the specified address as a DWORD.
- `data_read_qword(address)`: Read the 8 byte value at the specified address as a QWORD.
- `data_read_string(address)`: Read the string at the specified address.
Unsafe functions (`--unsafe` flag required):
- `dbg_get_registers()`: Get all registers and their values. This function is only available when debugging.
- `dbg_get_call_stack()`: Get the current call stack.
- `dbg_list_breakpoints()`: List all breakpoints in the program.
- `dbg_start_process()`: Start the debugger, returns the current instruction pointer.
- `dbg_exit_process()`: Exit the debugger.
- `dbg_continue_process()`: Continue the debugger, returns the current instruction pointer.
- `dbg_run_to(address)`: Run the debugger to the specified address.
- `dbg_set_breakpoint(address)`: Set a breakpoint at the specified address.
- `dbg_step_into()`: Step into the current instruction.
- `dbg_step_over()`: Step over the current instruction.
- `dbg_delete_breakpoint(address)`: Delete a breakpoint at the specified address.
- `dbg_enable_breakpoint(address, enable)`: Enable or disable a breakpoint at the specified address.
## Prerequisites
- [Python](https://www.python.org/downloads/) (**3.11 or higher**)
- Use `idapyswitch` to switch to the newest Python version
- [IDA Pro](https://hex-rays.com/ida-pro) (8.3 or higher, 9 recommended), **IDA Free is not supported**
- Supported MCP Client (pick one you like)
- [Cline](https://cline.bot)
- [Roo Code](https://roocode.com)
- [Claude](https://claude.ai/download)
- [Cursor](https://cursor.com)
- [VSCode Agent Mode](https://github.blog/news-insights/product-news/github-copilot-agent-mode-activated/)
- [Windsurf](https://windsurf.com)
- [Other MCP Clients](https://modelcontextprotocol.io/clients#example-clients): Run `ida-pro-mcp --config` to get the JSON config for your client.
## Installation
Install the latest version of the IDA Pro MCP package:
```sh
pip uninstall ida-pro-mcp
pip install https://github.com/mrexodia/ida-pro-mcp/archive/refs/heads/main.zip
```
Configure the MCP servers and install the IDA Plugin:
```
ida-pro-mcp --install
```
**Important**: Make sure you completely restart IDA/Visual Studio Code/Claude for the installation to take effect. Claude runs in the background and you need to quit it from the tray icon.
https://github.com/user-attachments/assets/65ed3373-a187-4dd5-a807-425dca1d8ee9
_Note_: You need to load a binary in IDA before the plugin menu will show up.
## Prompt Engineering
LLMs are prone to hallucinations and you need to be specific with your prompting. For reverse engineering the conversion between integers and bytes are especially problematic. Below is a minimal example prompt, feel free to start a discussion or open an issue if you have good results with a different prompt:
> Your task is to analyze a crackme in IDA Pro. You can use the MCP tools to retrieve information. In general use the following strategy:
> - Inspect the decompilation and add comments with your findings
> - Rename variables to more sensible names
> - Change the variable and argument types if necessary (especially pointer and array types)
> - Change function names to be more descriptive
> - If more details are necessary, disassemble the function and add comments with your findings
> - NEVER convert number bases yourself. Use the convert_number MCP tool if needed!
> - Do not attempt brute forcing, derive any solutions purely from the disassembly and simple python scripts
> - Create a report.md with your findings and steps taken at the end
> - When you find a solution, prompt to user for feedback with the password you found
This prompt was just the first experiment, please share if you found ways to improve the output!
Live stream discussing prompting and showing some real-world malware analysis:
[](https://www.youtube.com/watch?v=iFxNuk3kxhk)
## Tips for Enhancing LLM Accuracy
Large Language Models (LLMs) are powerful tools, but they can sometimes struggle with complex mathematical calculations or exhibit "hallucinations" (making up facts). Make sure to tell the LLM to use the `conver_number` MCP and you might also need [math-mcp](https://github.com/EthanHenrickson/math-mcp) for certain operations.
Another thing to keep in mind is that LLMs will not perform well on obfuscated code. Before trying to use an LLM to solve the problem, take a look around the binary and spend some time (automatically) removing the following things:
- String encryption
- Import hashing
- Control flow flattening
- Code encryption
- Anti-decompilation tricks
You should also use a tool like Lumina or FLIRT to try and resolve all the open source library code and the C++ STL, this will further improve the accuracy.
## SSE Transport & Headless MCP
You can run an SSE server to connect to the user interface like this:
```sh
uv run ida-pro-mcp --transport http://127.0.0.1:8744/sse
```
After installing [`idalib`](https://docs.hex-rays.com/user-guide/idalib) you can also run a headless SSE server:
```sh
uv run idalib-mcp --host 127.0.0.1 --port 8745 path/to/executable
```
_Note_: The `idalib` feature was contributed by [Willi Ballenthin](https://github.com/williballenthin).
## Manual Installation
_Note_: This section is for LLMs and power users who need detailed installation instructions.
<details>
## Manual MCP Server Installation (Cline/Roo Code)
To install the MCP server yourself, follow these steps:
1. Install [uv](https://github.com/astral-sh/uv) globally:
- Windows: `pip install uv`
- Linux/Mac: `curl -LsSf https://astral.sh/uv/install.sh | sh`
2. Clone this repository, for this example `C:\MCP\ida-pro-mcp`.
3. Navigate to the Cline/Roo Code _MCP Servers_ configuration (see screenshot).
4. Click on the _Installed_ tab.
5. Click on _Configure MCP Servers_, which will open `cline_mcp_settings.json`.
6. Add the `ida-pro-mcp` server:
```json
{
"mcpServers": {
"github.com/mrexodia/ida-pro-mcp": {
"command": "uv",
"args": [
"--directory",
"c:\\MCP\\ida-pro-mcp",
"run",
"server.py",
"--install-plugin"
],
"timeout": 1800,
"disabled": false
}
}
}
```
To check if the connection works you can perform the following tool call:
```
<use_mcp_tool>
<server_name>github.com/mrexodia/ida-pro-mcp</server_name>
<tool_name>check_connection</tool_name>
<arguments></arguments>
</use_mcp_tool>
```
## IDA Plugin installation
The IDA Pro plugin will be installed automatically when the MCP server starts. If you disabled the `--install-plugin` option, use the following steps:
1. Copy (**not move**) `src/ida_pro_mcp/mcp-plugin.py` in your plugins folder (`%appdata%\Hex-Rays\IDA Pro\plugins` on Windows).
2. Open an IDB and click `Edit -> Plugins -> MCP` to start the server.
</details>
## Comparison with other MCP servers
There are a few IDA Pro MCP servers floating around, but I created my own for a few reasons:
1. Installation should be fully automated.
2. The architecture of other plugins make it difficult to add new functionality quickly (too much boilerplate of unnecessary dependencies).
3. Learning new technologies is fun!
If you want to check them out, here is a list (in the order I discovered them):
- https://github.com/taida957789/ida-mcp-server-plugin (SSE protocol only, requires installing dependencies in IDAPython).
- https://github.com/fdrechsler/mcp-server-idapro (MCP Server in TypeScript, excessive boilerplate required to add new functionality).
- https://github.com/MxIris-Reverse-Engineering/ida-mcp-server (custom socket protocol, boilerplate).
Feel free to open a PR to add your IDA Pro MCP server here.
## Development
Adding new features is a super easy and streamlined process. All you have to do is add a new `@jsonrpc` function to [`mcp-plugin.py`](https://github.com/mrexodia/ida-pro-mcp/blob/164df8cf4ae251cc9cc0f464591fa6df8e0d9df4/src/ida_pro_mcp/mcp-plugin.py#L406-L419) and your function will be available in the MCP server without any additional boilerplate! Below is a video where I add the `get_metadata` function in less than 2 minutes (including testing):
https://github.com/user-attachments/assets/951de823-88ea-4235-adcb-9257e316ae64
To test the MCP server itself:
```sh
uv run mcp dev src/ida_pro_mcp/server.py
```
This will open a web interface at http://localhost:5173 and allow you to interact with the MCP tools for testing.
For testing I create a symbolic link to the IDA plugin and then POST a JSON-RPC request directly to `http://localhost:13337/mcp`. After [enabling symbolic links](https://learn.microsoft.com/en-us/windows/apps/get-started/enable-your-device-for-development) you can run the following command:
```sh
uv run ida-pro-mcp --install
```
Generate the changelog of direct commits to `main`:
```sh
git log --first-parent --no-merges 1.2.0..main "--pretty=- %s"
```
```
--------------------------------------------------------------------------------
/src/ida_pro_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/uv-package.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
uv run ida-pro-mcp --generate-docs
rm -rf dist
uv build
uv publish
```
--------------------------------------------------------------------------------
/src/ida_pro_mcp/__main__.py:
--------------------------------------------------------------------------------
```python
import sys
from ida_pro_mcp.server import main
if __name__ == "__main__":
sys.argv[0] = "ida_pro_mcp"
main()
```
--------------------------------------------------------------------------------
/ida-plugin.json:
--------------------------------------------------------------------------------
```json
{
"IDAMetadataDescriptorVersion": 1,
"plugin": {
"name": "IDA Pro MCP",
"entryPoint": "src/ida_pro_mcp/mcp-plugin.py",
"categories": ["api-scripting-and-automation", "debugging-and-tracing", "malware-analysis"],
"logoPath": ".github/logo.png",
"description": "AI-powered reverse engineering assistant that bridges IDA Pro with language models through MCP.",
"idaVersions": ">=8.3",
"version": "1.6.0"
}
}
```
--------------------------------------------------------------------------------
/.github/FUNDING.yml:
--------------------------------------------------------------------------------
```yaml
# These are supported funding model platforms
github: [mrexodia]
#patreon: # Replace with a single Patreon username
#open_collective: # Replace with a single Open Collective username
#ko_fi: # Replace with a single Ko-fi username
#tidelift: # Replace with a single Tidelift platform-name/package-name e.g., npm/babel
#community_bridge: # Replace with a single Community Bridge project-name e.g., cloud-foundry
#liberapay: # Replace with a single Liberapay username
#issuehunt: x64dbg/x64dbg # Replace with a single IssueHunt username
#otechie: # Replace with a single Otechie username
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "ida-pro-mcp"
version = "1.5.0"
description = "Vibe reversing with IDA Pro"
readme = "README.md"
requires-python = ">=3.11"
authors = [{ name = "mrexodia" }]
keywords = ["ida", "mcp", "llm", "plugin"]
classifiers = [
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.11",
"Operating System :: MacOS",
"Operating System :: Microsoft :: Windows",
]
dependencies = [
"mcp>=1.16.0",
]
[project.urls]
Repository = "https://github.com/mrexodia/ida-pro-mcp"
Issues = "https://github.com/mrexodia/ida-pro-mcp/issues"
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"
[dependency-groups]
dev = [
"mcp[cli]>=1.6.0",
"ty>=0.0.1a21",
]
[project.scripts]
ida-pro-mcp = "ida_pro_mcp.server:main"
idalib-mcp = "ida_pro_mcp.idalib_server:main"
[tool.pyright]
include = ["src/ida_pro_mcp"]
exclude = ["src/ida_pro_mcp/server_generated.py"]
```
--------------------------------------------------------------------------------
/src/ida_pro_mcp/idalib_server.py:
--------------------------------------------------------------------------------
```python
import sys
import inspect
import logging
import argparse
import importlib
from pathlib import Path
import typing_inspection.introspection as intro
from mcp.server.fastmcp import FastMCP
# idapro must go first to initialize idalib
import idapro
import ida_auto
import ida_hexrays
logger = logging.getLogger(__name__)
mcp = FastMCP("github.com/mrexodia/ida-pro-mcp#idalib")
def fixup_tool_argument_descriptions(mcp: FastMCP):
# In our tool definitions within `mcp-plugin.py`, we use `typing.Annotated` on function parameters
# to attach documentation. For example:
#
# def get_function_by_name(
# name: Annotated[str, "Name of the function to get"]
# ) -> Function:
# """Get a function by its name"""
# ...
#
# However, the interpretation of Annotated is left up to static analyzers and other tools.
# FastMCP doesn't have any special handling for these comments, so we splice them into the
# tool metadata ourselves here.
#
# Example, before:
#
# tool.parameter={
# properties: {
# name: {
# title: "Name",
# type: "string"
# }
# },
# required: ["name"],
# title: "get_function_by_nameArguments",
# type: "object"
# }
#
# Example, after:
#
# tool.parameter={
# properties: {
# name: {
# title: "Name",
# type: "string"
# description: "Name of the function to get"
# }
# },
# required: ["name"],
# title: "get_function_by_nameArguments",
# type: "object"
# }
#
# References:
# - https://docs.python.org/3/library/typing.html#typing.Annotated
# - https://fastapi.tiangolo.com/python-types/#type-hints-with-metadata-annotations
# unfortunately, FastMCP.list_tools() is async, so we break with best practices and reach into `._tool_manager`
# rather than spinning up an asyncio runtime just to fetch the (non-async) list of tools.
for tool in mcp._tool_manager.list_tools():
sig = inspect.signature(tool.fn)
for name, parameter in sig.parameters.items():
# this instance is a raw `typing._AnnotatedAlias` that we can't do anything with directly.
# it renders like:
#
# typing.Annotated[str, 'Name of the function to get']
if not parameter.annotation:
continue
# this instance will look something like:
#
# InspectedAnnotation(type=<class 'str'>, qualifiers=set(), metadata=['Name of the function to get'])
#
annotation = intro.inspect_annotation(
parameter.annotation,
annotation_source=intro.AnnotationSource.ANY
)
# for our use case, where we attach a single string annotation that is meant as documentation,
# we extract that string and assign it to "description" in the tool metadata.
if annotation.type is not str:
continue
if len(annotation.metadata) != 1:
continue
description = annotation.metadata[0]
if not isinstance(description, str):
continue
logger.debug("adding parameter documentation %s(%s='%s')", tool.name, name, description)
tool.parameters["properties"][name]["description"] = description
def main():
parser = argparse.ArgumentParser(description="MCP server for IDA Pro via idalib")
parser.add_argument("--verbose", "-v", action="store_true", help="Show debug messages")
parser.add_argument("--host", type=str, default="127.0.0.1", help="Host to listen on, default: 127.0.0.1")
parser.add_argument("--port", type=int, default=8745, help="Port to listen on, default: 8745")
parser.add_argument("--unsafe", action="store_true", help="Enable unsafe functions (DANGEROUS)")
parser.add_argument("input_path", type=Path, help="Path to the input file to analyze.")
args = parser.parse_args()
if args.verbose:
log_level = logging.DEBUG
idapro.enable_console_messages(True)
else:
log_level = logging.INFO
idapro.enable_console_messages(False)
mcp.settings.log_level = logging.getLevelName(log_level)
mcp.settings.host = args.host
mcp.settings.port = args.port
logging.basicConfig(level=log_level)
# reset logging levels that might be initialized in idapythonrc.py
# which is evaluated during import of idalib.
logging.getLogger().setLevel(log_level)
if not args.input_path.exists():
raise FileNotFoundError(f"Input file not found: {args.input_path}")
# TODO: add a tool for specifying the idb/input file (sandboxed)
logger.info("opening database: %s", args.input_path)
if idapro.open_database(str(args.input_path), run_auto_analysis=True):
raise RuntimeError("failed to analyze input file")
logger.debug("idalib: waiting for analysis...")
ida_auto.auto_wait()
if not ida_hexrays.init_hexrays_plugin():
raise RuntimeError("failed to initialize Hex-Rays decompiler")
plugin = importlib.import_module("ida_pro_mcp.mcp-plugin")
logger.debug("adding tools...")
for name, callable in plugin.rpc_registry.methods.items():
if args.unsafe or name not in plugin.rpc_registry.unsafe:
logger.debug("adding tool: %s: %s", name, callable)
mcp.add_tool(callable, name)
# NOTE: https://github.com/modelcontextprotocol/python-sdk/issues/466
fixup_tool_argument_descriptions(mcp)
# NOTE: npx @modelcontextprotocol/inspector for debugging
logger.info("MCP Server availabile at: http://%s:%d/sse", mcp.settings.host, mcp.settings.port)
try:
mcp.run(transport="sse")
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/ida_pro_mcp/server.py:
--------------------------------------------------------------------------------
```python
import os
import sys
import ast
import json
import shutil
import argparse
import http.client
from urllib.parse import urlparse
from glob import glob
from mcp.server.fastmcp import FastMCP
# The log_level is necessary for Cline to work: https://github.com/jlowin/fastmcp/issues/81
mcp = FastMCP("ida-pro-mcp", log_level="ERROR")
jsonrpc_request_id = 1
ida_host = "127.0.0.1"
ida_port = 13337
def make_jsonrpc_request(method: str, *params):
"""Make a JSON-RPC request to the IDA plugin"""
global jsonrpc_request_id, ida_host, ida_port
conn = http.client.HTTPConnection(ida_host, ida_port)
request = {
"jsonrpc": "2.0",
"method": method,
"params": list(params),
"id": jsonrpc_request_id,
}
jsonrpc_request_id += 1
try:
conn.request("POST", "/mcp", json.dumps(request), {
"Content-Type": "application/json"
})
response = conn.getresponse()
data = json.loads(response.read().decode())
if "error" in data:
error = data["error"]
code = error["code"]
message = error["message"]
pretty = f"JSON-RPC error {code}: {message}"
if "data" in error:
pretty += "\n" + error["data"]
raise Exception(pretty)
result = data["result"]
# NOTE: LLMs do not respond well to empty responses
if result is None:
result = "success"
return result
except Exception:
raise
finally:
conn.close()
@mcp.tool()
def check_connection() -> str:
"""Check if the IDA plugin is running"""
try:
metadata = make_jsonrpc_request("get_metadata")
return f"Successfully connected to IDA Pro (open file: {metadata['module']})"
except Exception as e:
if sys.platform == "darwin":
shortcut = "Ctrl+Option+M"
else:
shortcut = "Ctrl+Alt+M"
return f"Failed to connect to IDA Pro! Did you run Edit -> Plugins -> MCP ({shortcut}) to start the server?"
# Code taken from https://github.com/mrexodia/ida-pro-mcp (MIT License)
class MCPVisitor(ast.NodeVisitor):
def __init__(self):
self.types: dict[str, ast.ClassDef] = {}
self.functions: dict[str, ast.FunctionDef] = {}
self.descriptions: dict[str, str] = {}
self.unsafe: list[str] = []
def visit_FunctionDef(self, node):
for decorator in node.decorator_list:
if isinstance(decorator, ast.Name):
if decorator.id == "jsonrpc":
for i, arg in enumerate(node.args.args):
arg_name = arg.arg
arg_type = arg.annotation
if arg_type is None:
raise Exception(f"Missing argument type for {node.name}.{arg_name}")
if isinstance(arg_type, ast.Subscript):
assert isinstance(arg_type.value, ast.Name)
assert arg_type.value.id == "Annotated"
assert isinstance(arg_type.slice, ast.Tuple)
assert len(arg_type.slice.elts) == 2
annot_type = arg_type.slice.elts[0]
annot_description = arg_type.slice.elts[1]
assert isinstance(annot_description, ast.Constant)
node.args.args[i].annotation = ast.Subscript(
value=ast.Name(id="Annotated", ctx=ast.Load()),
slice=ast.Tuple(
elts=[
annot_type,
ast.Call(
func=ast.Name(id="Field", ctx=ast.Load()),
args=[],
keywords=[
ast.keyword(
arg="description",
value=annot_description)])],
ctx=ast.Load()),
ctx=ast.Load())
elif isinstance(arg_type, ast.Name):
pass
else:
raise Exception(f"Unexpected type annotation for {node.name}.{arg_name} -> {type(arg_type)}")
body_comment = node.body[0]
if isinstance(body_comment, ast.Expr) and isinstance(body_comment.value, ast.Constant):
new_body = [body_comment]
self.descriptions[node.name] = body_comment.value.value
else:
new_body = []
call_args = [ast.Constant(value=node.name)]
for arg in node.args.args:
call_args.append(ast.Name(id=arg.arg, ctx=ast.Load()))
new_body.append(ast.Return(
value=ast.Call(
func=ast.Name(id="make_jsonrpc_request", ctx=ast.Load()),
args=call_args,
keywords=[])))
decorator_list = [
ast.Call(
func=ast.Attribute(
value=ast.Name(id="mcp", ctx=ast.Load()),
attr="tool",
ctx=ast.Load()),
args=[],
keywords=[]
)
]
node_nobody = ast.FunctionDef(node.name, node.args, new_body, decorator_list, node.returns, node.type_comment, lineno=node.lineno, col_offset=node.col_offset)
assert node.name not in self.functions, f"Duplicate function: {node.name}"
self.functions[node.name] = node_nobody
elif decorator.id == "unsafe":
self.unsafe.append(node.name)
def visit_ClassDef(self, node):
for base in node.bases:
if isinstance(base, ast.Name):
if base.id == "TypedDict":
self.types[node.name] = node
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
IDA_PLUGIN_PY = os.path.join(SCRIPT_DIR, "mcp-plugin.py")
GENERATED_PY = os.path.join(SCRIPT_DIR, "server_generated.py")
# NOTE: This is in the global scope on purpose
if not os.path.exists(IDA_PLUGIN_PY):
raise RuntimeError(f"IDA plugin not found at {IDA_PLUGIN_PY} (did you move it?)")
with open(IDA_PLUGIN_PY, "r", encoding="utf-8") as f:
code = f.read()
module = ast.parse(code, IDA_PLUGIN_PY)
visitor = MCPVisitor()
visitor.visit(module)
code = """# NOTE: This file has been automatically generated, do not modify!
# Architecture based on https://github.com/mrexodia/ida-pro-mcp (MIT License)
import sys
if sys.version_info >= (3, 12):
from typing import Annotated, Optional, TypedDict, Generic, TypeVar, NotRequired
else:
from typing_extensions import Annotated, Optional, TypedDict, Generic, TypeVar, NotRequired
from pydantic import Field
T = TypeVar("T")
"""
for type in visitor.types.values():
code += ast.unparse(type)
code += "\n\n"
for function in visitor.functions.values():
code += ast.unparse(function)
code += "\n\n"
try:
if os.path.exists(GENERATED_PY):
with open(GENERATED_PY, "rb") as f:
existing_code_bytes = f.read()
else:
existing_code_bytes = b""
code_bytes = code.encode("utf-8").replace(b"\r", b"")
if code_bytes != existing_code_bytes:
with open(GENERATED_PY, "wb") as f:
f.write(code_bytes)
except:
print(f"Failed to generate code: {GENERATED_PY}", file=sys.stderr, flush=True)
exec(compile(code, GENERATED_PY, "exec"))
MCP_FUNCTIONS = ["check_connection"] + list(visitor.functions.keys())
UNSAFE_FUNCTIONS = visitor.unsafe
SAFE_FUNCTIONS = [f for f in MCP_FUNCTIONS if f not in UNSAFE_FUNCTIONS]
def generate_readme():
print("README:")
print(f"- `check_connection()`: Check if the IDA plugin is running.")
def get_description(name: str):
function = visitor.functions[name]
signature = function.name + "("
for i, arg in enumerate(function.args.args):
if i > 0:
signature += ", "
signature += arg.arg
signature += ")"
description = visitor.descriptions.get(function.name, "<no description>").strip().split("\n")[0]
if description[-1] != ".":
description += "."
return f"- `{signature}`: {description}"
for safe_function in SAFE_FUNCTIONS:
if safe_function != "check_connection":
print(get_description(safe_function))
print("\nUnsafe functions (`--unsafe` flag required):\n")
for unsafe_function in UNSAFE_FUNCTIONS:
print(get_description(unsafe_function))
print("\nMCP Config:")
mcp_config = {
"mcpServers": {
"github.com/mrexodia/ida-pro-mcp": {
"command": "uv",
"args": [
"--directory",
"c:\\MCP\\ida-pro-mcp",
"run",
"server.py",
"--install-plugin"
],
"timeout": 1800,
"disabled": False,
}
}
}
print(json.dumps(mcp_config, indent=2))
def get_python_executable():
"""Get the path to the Python executable"""
venv = os.environ.get("VIRTUAL_ENV")
if venv:
if sys.platform == "win32":
python = os.path.join(venv, "Scripts", "python.exe")
else:
python = os.path.join(venv, "bin", "python3")
if os.path.exists(python):
return python
for path in sys.path:
if sys.platform == "win32":
path = path.replace("/", "\\")
split = path.split(os.sep)
if split[-1].endswith(".zip"):
path = os.path.dirname(path)
if sys.platform == "win32":
python_executable = os.path.join(path, "python.exe")
else:
python_executable = os.path.join(path, "..", "bin", "python3")
python_executable = os.path.abspath(python_executable)
if os.path.exists(python_executable):
return python_executable
return sys.executable
def copy_python_env(env: dict[str, str]):
# Reference: https://docs.python.org/3/using/cmdline.html#environment-variables
python_vars = [
"PYTHONHOME",
"PYTHONPATH",
"PYTHONSAFEPATH",
"PYTHONPLATLIBDIR",
"PYTHONPYCACHEPREFIX",
"PYTHONNOUSERSITE",
"PYTHONUSERBASE",
]
# MCP servers are run without inheriting the environment, so we need to forward
# the environment variables that affect Python's dependency resolution by hand.
# Issue: https://github.com/mrexodia/ida-pro-mcp/issues/111
result = False
for var in python_vars:
value = os.environ.get(var)
if value:
result = True
env[var] = value
return result
def print_mcp_config():
mcp_config = {
"command": get_python_executable(),
"args": [
__file__,
],
"timeout": 1800,
"disabled": False,
}
env = {}
if copy_python_env(env):
print(f"[WARNING] Custom Python environment variables detected")
mcp_config["env"] = env
print(json.dumps({
"mcpServers": {
mcp.name: mcp_config
}
}, indent=2)
)
def install_mcp_servers(*, uninstall=False, quiet=False, env={}):
if sys.platform == "win32":
configs = {
"Cline": (os.path.join(os.getenv("APPDATA", ""), "Code", "User", "globalStorage", "saoudrizwan.claude-dev", "settings"), "cline_mcp_settings.json"),
"Roo Code": (os.path.join(os.getenv("APPDATA", ""), "Code", "User", "globalStorage", "rooveterinaryinc.roo-cline", "settings"), "mcp_settings.json"),
"Kilo Code": (os.path.join(os.getenv("APPDATA", ""), "Code", "User", "globalStorage", "kilocode.kilo-code", "settings"), "mcp_settings.json"),
"Claude": (os.path.join(os.getenv("APPDATA", ""), "Claude"), "claude_desktop_config.json"),
"Cursor": (os.path.join(os.path.expanduser("~"), ".cursor"), "mcp.json"),
"Windsurf": (os.path.join(os.path.expanduser("~"), ".codeium", "windsurf"), "mcp_config.json"),
"Claude Code": (os.path.join(os.path.expanduser("~")), ".claude.json"),
"LM Studio": (os.path.join(os.path.expanduser("~"), ".lmstudio"), "mcp.json"),
}
elif sys.platform == "darwin":
configs = {
"Cline": (os.path.join(os.path.expanduser("~"), "Library", "Application Support", "Code", "User", "globalStorage", "saoudrizwan.claude-dev", "settings"), "cline_mcp_settings.json"),
"Roo Code": (os.path.join(os.path.expanduser("~"), "Library", "Application Support", "Code", "User", "globalStorage", "rooveterinaryinc.roo-cline", "settings"), "mcp_settings.json"),
"Kilo Code": (os.path.join(os.path.expanduser("~"), "Library", "Application Support", "Code", "User", "globalStorage", "kilocode.kilo-code", "settings"), "mcp_settings.json"),
"Claude": (os.path.join(os.path.expanduser("~"), "Library", "Application Support", "Claude"), "claude_desktop_config.json"),
"Cursor": (os.path.join(os.path.expanduser("~"), ".cursor"), "mcp.json"),
"Windsurf": (os.path.join(os.path.expanduser("~"), ".codeium", "windsurf"), "mcp_config.json"),
"Claude Code": (os.path.join(os.path.expanduser("~")), ".claude.json"),
"LM Studio": (os.path.join(os.path.expanduser("~"), ".lmstudio"), "mcp.json"),
}
elif sys.platform == "linux":
configs = {
"Cline": (os.path.join(os.path.expanduser("~"), ".config", "Code", "User", "globalStorage", "saoudrizwan.claude-dev", "settings"), "cline_mcp_settings.json"),
"Roo Code": (os.path.join(os.path.expanduser("~"), ".config", "Code", "User", "globalStorage", "rooveterinaryinc.roo-cline", "settings"), "mcp_settings.json"),
"Kilo Code": (os.path.join(os.path.expanduser("~"), ".config", "Code", "User", "globalStorage", "kilocode.kilo-code", "settings"), "mcp_settings.json"),
# Claude not supported on Linux
"Cursor": (os.path.join(os.path.expanduser("~"), ".cursor"), "mcp.json"),
"Windsurf": (os.path.join(os.path.expanduser("~"), ".codeium", "windsurf"), "mcp_config.json"),
"Claude Code": (os.path.join(os.path.expanduser("~")), ".claude.json"),
"LM Studio": (os.path.join(os.path.expanduser("~"), ".lmstudio"), "mcp.json"),
}
else:
print(f"Unsupported platform: {sys.platform}")
return
installed = 0
for name, (config_dir, config_file) in configs.items():
config_path = os.path.join(config_dir, config_file)
if not os.path.exists(config_dir):
action = "uninstall" if uninstall else "installation"
if not quiet:
print(f"Skipping {name} {action}\n Config: {config_path} (not found)")
continue
if not os.path.exists(config_path):
config = {}
else:
with open(config_path, "r", encoding="utf-8") as f:
data = f.read().strip()
if len(data) == 0:
config = {}
else:
try:
config = json.loads(data)
except json.decoder.JSONDecodeError:
if not quiet:
print(f"Skipping {name} uninstall\n Config: {config_path} (invalid JSON)")
continue
if "mcpServers" not in config:
config["mcpServers"] = {}
mcp_servers = config["mcpServers"]
# Migrate old name
old_name = "github.com/mrexodia/ida-pro-mcp"
if old_name in mcp_servers:
mcp_servers[mcp.name] = mcp_servers[old_name]
del mcp_servers[old_name]
if uninstall:
if mcp.name not in mcp_servers:
if not quiet:
print(f"Skipping {name} uninstall\n Config: {config_path} (not installed)")
continue
del mcp_servers[mcp.name]
else:
# Copy environment variables from the existing server if present
if mcp.name in mcp_servers:
for key, value in mcp_servers[mcp.name].get("env", {}).items():
env[key] = value
if copy_python_env(env):
print(f"[WARNING] Custom Python environment variables detected")
mcp_servers[mcp.name] = {
"command": get_python_executable(),
"args": [
__file__,
],
"timeout": 1800,
"disabled": False,
"autoApprove": SAFE_FUNCTIONS,
"alwaysAllow": SAFE_FUNCTIONS,
}
if env:
mcp_servers[mcp.name]["env"] = env
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
if not quiet:
action = "Uninstalled" if uninstall else "Installed"
print(f"{action} {name} MCP server (restart required)\n Config: {config_path}")
installed += 1
if not uninstall and installed == 0:
print("No MCP servers installed. For unsupported MCP clients, use the following config:\n")
print_mcp_config()
def install_ida_plugin(*, uninstall: bool = False, quiet: bool = False):
if sys.platform == "win32":
ida_folder = os.path.join(os.getenv("APPDATA"), "Hex-Rays", "IDA Pro")
else:
ida_folder = os.path.join(os.path.expanduser("~"), ".idapro")
free_licenses = glob(os.path.join(ida_folder, "idafree_*.hexlic"))
if len(free_licenses) > 0:
print(f"IDA Free does not support plugins and cannot be used. Purchase and install IDA Pro instead.")
sys.exit(1)
ida_plugin_folder = os.path.join(ida_folder, "plugins")
plugin_destination = os.path.join(ida_plugin_folder, "mcp-plugin.py")
if uninstall:
if not os.path.exists(plugin_destination):
print(f"Skipping IDA plugin uninstall\n Path: {plugin_destination} (not found)")
return
os.remove(plugin_destination)
if not quiet:
print(f"Uninstalled IDA plugin\n Path: {plugin_destination}")
else:
# Create IDA plugins folder
if not os.path.exists(ida_plugin_folder):
os.makedirs(ida_plugin_folder)
# Skip if symlink already up to date
realpath = os.path.realpath(plugin_destination)
if realpath == IDA_PLUGIN_PY:
if not quiet:
print(f"Skipping IDA plugin installation (symlink up to date)\n Plugin: {realpath}")
else:
# Remove existing plugin
if os.path.lexists(plugin_destination):
os.remove(plugin_destination)
# Symlink or copy the plugin
try:
os.symlink(IDA_PLUGIN_PY, plugin_destination)
except OSError:
shutil.copy(IDA_PLUGIN_PY, plugin_destination)
if not quiet:
print(f"Installed IDA Pro plugin (IDA restart required)\n Plugin: {plugin_destination}")
def main():
global ida_host, ida_port
parser = argparse.ArgumentParser(description="IDA Pro MCP Server")
parser.add_argument("--install", action="store_true", help="Install the MCP Server and IDA plugin")
parser.add_argument("--uninstall", action="store_true", help="Uninstall the MCP Server and IDA plugin")
parser.add_argument("--generate-docs", action="store_true", help=argparse.SUPPRESS)
parser.add_argument("--install-plugin", action="store_true", help=argparse.SUPPRESS)
parser.add_argument("--transport", type=str, default="stdio", help="MCP transport protocol to use (stdio or http://127.0.0.1:8744)")
parser.add_argument("--ida-rpc", type=str, default=f"http://{ida_host}:{ida_port}", help=f"IDA RPC server to use (default: http://{ida_host}:{ida_port})")
parser.add_argument("--unsafe", action="store_true", help="Enable unsafe functions (DANGEROUS)")
parser.add_argument("--config", action="store_true", help="Generate MCP config JSON")
args = parser.parse_args()
if args.install and args.uninstall:
print("Cannot install and uninstall at the same time")
return
if args.install:
install_ida_plugin()
install_mcp_servers()
return
if args.uninstall:
install_ida_plugin(uninstall=True)
install_mcp_servers(uninstall=True)
return
# NOTE: Developers can use this to generate the README
if args.generate_docs:
generate_readme()
return
# NOTE: This is silent for automated Cline installations
if args.install_plugin:
install_ida_plugin(quiet=True)
if args.config:
print_mcp_config()
return
# Parse IDA RPC server argument
ida_rpc = urlparse(args.ida_rpc)
if ida_rpc.hostname is None or ida_rpc.port is None:
raise Exception(f"Invalid IDA RPC server: {args.ida_rpc}")
ida_host = ida_rpc.hostname
ida_port = ida_rpc.port
# Remove unsafe tools
if not args.unsafe:
mcp_tools = mcp._tool_manager._tools
for unsafe in UNSAFE_FUNCTIONS:
if unsafe in mcp_tools:
del mcp_tools[unsafe]
try:
if args.transport == "stdio":
mcp.run(transport="stdio")
else:
url = urlparse(args.transport)
if url.hostname is None or url.port is None:
raise Exception(f"Invalid transport URL: {args.transport}")
mcp.settings.host = url.hostname
mcp.settings.port = url.port
# NOTE: npx @modelcontextprotocol/inspector for debugging
print(f"MCP Server availabile at http://{mcp.settings.host}:{mcp.settings.port}/sse")
mcp.settings.log_level = "INFO"
mcp.run(transport="sse")
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/ida_pro_mcp/mcp-plugin.py:
--------------------------------------------------------------------------------
```python
import os
import sys
if sys.version_info < (3, 11):
raise RuntimeError("Python 3.11 or higher is required for the MCP plugin")
import json
import struct
import threading
import http.server
from urllib.parse import urlparse
from typing import (
Any,
Callable,
get_type_hints,
TypedDict,
Optional,
Annotated,
TypeVar,
Generic,
NotRequired,
overload,
Literal,
)
class JSONRPCError(Exception):
def __init__(self, code: int, message: str, data: Any = None):
self.code = code
self.message = message
self.data = data
class RPCRegistry:
def __init__(self):
self.methods: dict[str, Callable] = {}
self.unsafe: set[str] = set()
def register(self, func: Callable) -> Callable:
self.methods[func.__name__] = func
return func
def mark_unsafe(self, func: Callable) -> Callable:
self.unsafe.add(func.__name__)
return func
def dispatch(self, method: str, params: Any) -> Any:
if method not in self.methods:
raise JSONRPCError(-32601, f"Method '{method}' not found")
func = self.methods[method]
hints = get_type_hints(func)
# Remove return annotation if present
hints.pop("return", None)
if isinstance(params, list):
if len(params) != len(hints):
raise JSONRPCError(-32602, f"Invalid params: expected {len(hints)} arguments, got {len(params)}")
# Validate and convert parameters
converted_params = []
for value, (param_name, expected_type) in zip(params, hints.items()):
try:
if not isinstance(value, expected_type):
value = expected_type(value)
converted_params.append(value)
except (ValueError, TypeError):
raise JSONRPCError(-32602, f"Invalid type for parameter '{param_name}': expected {expected_type.__name__}")
return func(*converted_params)
elif isinstance(params, dict):
if set(params.keys()) != set(hints.keys()):
raise JSONRPCError(-32602, f"Invalid params: expected {list(hints.keys())}")
# Validate and convert parameters
converted_params = {}
for param_name, expected_type in hints.items():
value = params.get(param_name)
try:
if not isinstance(value, expected_type):
value = expected_type(value)
converted_params[param_name] = value
except (ValueError, TypeError):
raise JSONRPCError(-32602, f"Invalid type for parameter '{param_name}': expected {expected_type.__name__}")
return func(**converted_params)
else:
raise JSONRPCError(-32600, "Invalid Request: params must be array or object")
rpc_registry = RPCRegistry()
def jsonrpc(func: Callable) -> Callable:
"""Decorator to register a function as a JSON-RPC method"""
global rpc_registry
return rpc_registry.register(func)
def unsafe(func: Callable) -> Callable:
"""Decorator to register mark a function as unsafe"""
return rpc_registry.mark_unsafe(func)
class JSONRPCRequestHandler(http.server.BaseHTTPRequestHandler):
def send_jsonrpc_error(self, code: int, message: str, id: Any = None):
response = {
"jsonrpc": "2.0",
"error": {
"code": code,
"message": message
}
}
if id is not None:
response["id"] = id
response_body = json.dumps(response).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(response_body)))
self.end_headers()
self.wfile.write(response_body)
def do_POST(self):
global rpc_registry
parsed_path = urlparse(self.path)
if parsed_path.path != "/mcp":
self.send_jsonrpc_error(-32098, "Invalid endpoint", None)
return
content_length = int(self.headers.get("Content-Length", 0))
if content_length == 0:
self.send_jsonrpc_error(-32700, "Parse error: missing request body", None)
return
request_body = self.rfile.read(content_length)
try:
request = json.loads(request_body)
except json.JSONDecodeError:
self.send_jsonrpc_error(-32700, "Parse error: invalid JSON", None)
return
# Prepare the response
response: dict[str, Any] = {
"jsonrpc": "2.0"
}
if request.get("id") is not None:
response["id"] = request.get("id")
try:
# Basic JSON-RPC validation
if not isinstance(request, dict):
raise JSONRPCError(-32600, "Invalid Request")
if request.get("jsonrpc") != "2.0":
raise JSONRPCError(-32600, "Invalid JSON-RPC version")
if "method" not in request:
raise JSONRPCError(-32600, "Method not specified")
# Dispatch the method
result = rpc_registry.dispatch(request["method"], request.get("params", []))
response["result"] = result
except JSONRPCError as e:
response["error"] = {
"code": e.code,
"message": e.message
}
if e.data is not None:
response["error"]["data"] = e.data
except IDAError as e:
response["error"] = {
"code": -32000,
"message": e.message,
}
except Exception as e:
traceback.print_exc()
response["error"] = {
"code": -32603,
"message": "Internal error (please report a bug)",
"data": traceback.format_exc(),
}
try:
response_body = json.dumps(response).encode("utf-8")
except Exception as e:
traceback.print_exc()
response_body = json.dumps({
"error": {
"code": -32603,
"message": "Internal error (please report a bug)",
"data": traceback.format_exc(),
}
}).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(response_body)))
self.end_headers()
self.wfile.write(response_body)
def log_message(self, format, *args):
# Suppress logging
pass
class MCPHTTPServer(http.server.HTTPServer):
allow_reuse_address = False
class Server:
HOST = "localhost"
PORT = 13337
def __init__(self):
self.server = None
self.server_thread = None
self.running = False
def start(self):
if self.running:
print("[MCP] Server is already running")
return
self.server_thread = threading.Thread(target=self._run_server, daemon=True)
self.running = True
self.server_thread.start()
def stop(self):
if not self.running:
return
self.running = False
if self.server:
self.server.shutdown()
self.server.server_close()
if self.server_thread:
self.server_thread.join()
self.server = None
print("[MCP] Server stopped")
def _run_server(self):
try:
# Create server in the thread to handle binding
self.server = MCPHTTPServer((Server.HOST, Server.PORT), JSONRPCRequestHandler)
print(f"[MCP] Server started at http://{Server.HOST}:{Server.PORT}")
self.server.serve_forever()
except OSError as e:
if e.errno == 98 or e.errno == 10048: # Port already in use (Linux/Windows)
print("[MCP] Error: Port 13337 is already in use")
else:
print(f"[MCP] Server error: {e}")
self.running = False
except Exception as e:
print(f"[MCP] Server error: {e}")
finally:
self.running = False
# A module that helps with writing thread safe ida code.
# Based on:
# https://web.archive.org/web/20160305190440/http://www.williballenthin.com/blog/2015/09/04/idapython-synchronization-decorator/
import logging
import queue
import traceback
import functools
from enum import IntEnum, IntFlag
import ida_hexrays
import ida_kernwin
import ida_funcs
import ida_gdl
import ida_lines
import ida_idaapi
import idc
import idaapi
import idautils
import ida_nalt
import ida_bytes
import ida_typeinf
import ida_xref
import ida_entry
import idautils
import ida_idd
import ida_dbg
import ida_name
import ida_ida
import ida_frame
ida_major, ida_minor = map(int, idaapi.get_kernel_version().split("."))
class IDAError(Exception):
def __init__(self, message: str):
super().__init__(message)
@property
def message(self) -> str:
return self.args[0]
class IDASyncError(Exception):
pass
# Important note: Always make sure the return value from your function f is a
# copy of the data you have gotten from IDA, and not the original data.
#
# Example:
# --------
#
# Do this:
#
# @idaread
# def ts_Functions():
# return list(idautils.Functions())
#
# Don't do this:
#
# @idaread
# def ts_Functions():
# return idautils.Functions()
#
logger = logging.getLogger(__name__)
# Enum for safety modes. Higher means safer:
class IDASafety(IntEnum):
SAFE_NONE = ida_kernwin.MFF_FAST
SAFE_READ = ida_kernwin.MFF_READ
SAFE_WRITE = ida_kernwin.MFF_WRITE
call_stack = queue.LifoQueue()
def sync_wrapper(ff, safety_mode: IDASafety):
"""
Call a function ff with a specific IDA safety_mode.
"""
#logger.debug('sync_wrapper: {}, {}'.format(ff.__name__, safety_mode))
if safety_mode not in [IDASafety.SAFE_READ, IDASafety.SAFE_WRITE]:
error_str = 'Invalid safety mode {} over function {}'\
.format(safety_mode, ff.__name__)
logger.error(error_str)
raise IDASyncError(error_str)
# No safety level is set up:
res_container = queue.Queue()
def runned():
#logger.debug('Inside runned')
# Make sure that we are not already inside a sync_wrapper:
if not call_stack.empty():
last_func_name = call_stack.get()
error_str = ('Call stack is not empty while calling the '
'function {} from {}').format(ff.__name__, last_func_name)
#logger.error(error_str)
raise IDASyncError(error_str)
call_stack.put((ff.__name__))
try:
res_container.put(ff())
except Exception as x:
res_container.put(x)
finally:
call_stack.get()
#logger.debug('Finished runned')
ret_val = idaapi.execute_sync(runned, safety_mode)
res = res_container.get()
if isinstance(res, Exception):
raise res
return res
def idawrite(f):
"""
decorator for marking a function as modifying the IDB.
schedules a request to be made in the main IDA loop to avoid IDB corruption.
"""
@functools.wraps(f)
def wrapper(*args, **kwargs):
ff = functools.partial(f, *args, **kwargs)
ff.__name__ = f.__name__ # type: ignore
return sync_wrapper(ff, idaapi.MFF_WRITE)
return wrapper
def idaread(f):
"""
decorator for marking a function as reading from the IDB.
schedules a request to be made in the main IDA loop to avoid
inconsistent results.
MFF_READ constant via: http://www.openrce.org/forums/posts/1827
"""
@functools.wraps(f)
def wrapper(*args, **kwargs):
ff = functools.partial(f, *args, **kwargs)
ff.__name__ = f.__name__ # type: ignore
return sync_wrapper(ff, idaapi.MFF_READ)
return wrapper
def is_window_active():
"""Returns whether IDA is currently active"""
try:
from PyQt5.QtWidgets import QApplication
except ImportError:
return False
app = QApplication.instance()
if app is None:
return False
for widget in app.topLevelWidgets():
if widget.isActiveWindow():
return True
return False
class Metadata(TypedDict):
path: str
module: str
base: str
size: str
md5: str
sha256: str
crc32: str
filesize: str
def get_image_size() -> int:
try:
# https://www.hex-rays.com/products/ida/support/sdkdoc/structidainfo.html
info = idaapi.get_inf_structure() # type: ignore
omin_ea = info.omin_ea
omax_ea = info.omax_ea
except AttributeError:
import ida_ida
omin_ea = ida_ida.inf_get_omin_ea()
omax_ea = ida_ida.inf_get_omax_ea()
# Bad heuristic for image size (bad if the relocations are the last section)
image_size = omax_ea - omin_ea
# Try to extract it from the PE header
header = idautils.peutils_t().header()
if header and header[:4] == b"PE\0\0":
image_size = struct.unpack("<I", header[0x50:0x54])[0]
return image_size
@jsonrpc
@idaread
def get_metadata() -> Metadata:
"""Get metadata about the current IDB"""
# Fat Mach-O binaries can return a None hash:
# https://github.com/mrexodia/ida-pro-mcp/issues/26
def hash(f):
try:
return f().hex()
except:
return ""
return Metadata(path=idaapi.get_input_file_path(),
module=idaapi.get_root_filename(),
base=hex(idaapi.get_imagebase()),
size=hex(get_image_size()),
md5=hash(ida_nalt.retrieve_input_file_md5),
sha256=hash(ida_nalt.retrieve_input_file_sha256),
crc32=hex(ida_nalt.retrieve_input_file_crc32()),
filesize=hex(ida_nalt.retrieve_input_file_size()))
def get_prototype(fn: ida_funcs.func_t) -> Optional[str]:
try:
prototype: ida_typeinf.tinfo_t = fn.get_prototype()
if prototype is not None:
return str(prototype)
else:
return None
except AttributeError:
try:
return idc.get_type(fn.start_ea)
except:
tif = ida_typeinf.tinfo_t()
if ida_nalt.get_tinfo(tif, fn.start_ea):
return str(tif)
return None
except Exception as e:
print(f"Error getting function prototype: {e}")
return None
class Function(TypedDict):
address: str
name: str
size: str
def parse_address(address: str | int) -> int:
if isinstance(address, int):
return address
try:
return int(address, 0)
except ValueError:
for ch in address:
if ch not in "0123456789abcdefABCDEF":
raise IDAError(f"Failed to parse address: {address}")
raise IDAError(f"Failed to parse address (missing 0x prefix): {address}")
@overload
def get_function(address: int, *, raise_error: Literal[True]) -> Function: ...
@overload
def get_function(address: int) -> Function: ...
@overload
def get_function(address: int, *, raise_error: Literal[False]) -> Optional[Function]: ...
def get_function(address, *, raise_error=True):
fn = idaapi.get_func(address)
if fn is None:
if raise_error:
raise IDAError(f"No function found at address {hex(address)}")
return None
try:
name = fn.get_name()
except AttributeError:
name = ida_funcs.get_func_name(fn.start_ea)
return Function(address=hex(address), name=name, size=hex(fn.end_ea - fn.start_ea))
DEMANGLED_TO_EA = {}
def create_demangled_to_ea_map():
for ea in idautils.Functions():
# Get the function name and demangle it
# MNG_NODEFINIT inhibits everything except the main name
# where default demangling adds the function signature
# and decorators (if any)
demangled = idaapi.demangle_name(
idc.get_name(ea, 0), idaapi.MNG_NODEFINIT)
if demangled:
DEMANGLED_TO_EA[demangled] = ea
def get_type_by_name(type_name: str) -> ida_typeinf.tinfo_t:
# 8-bit integers
if type_name in ('int8', '__int8', 'int8_t', 'char', 'signed char'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_INT8)
elif type_name in ('uint8', '__uint8', 'uint8_t', 'unsigned char', 'byte', 'BYTE'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_UINT8)
# 16-bit integers
elif type_name in ('int16', '__int16', 'int16_t', 'short', 'short int', 'signed short', 'signed short int'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_INT16)
elif type_name in ('uint16', '__uint16', 'uint16_t', 'unsigned short', 'unsigned short int', 'word', 'WORD'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_UINT16)
# 32-bit integers
elif type_name in ('int32', '__int32', 'int32_t', 'int', 'signed int', 'long', 'long int', 'signed long', 'signed long int'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_INT32)
elif type_name in ('uint32', '__uint32', 'uint32_t', 'unsigned int', 'unsigned long', 'unsigned long int', 'dword', 'DWORD'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_UINT32)
# 64-bit integers
elif type_name in ('int64', '__int64', 'int64_t', 'long long', 'long long int', 'signed long long', 'signed long long int'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_INT64)
elif type_name in ('uint64', '__uint64', 'uint64_t', 'unsigned int64', 'unsigned long long', 'unsigned long long int', 'qword', 'QWORD'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_UINT64)
# 128-bit integers
elif type_name in ('int128', '__int128', 'int128_t', '__int128_t'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_INT128)
elif type_name in ('uint128', '__uint128', 'uint128_t', '__uint128_t', 'unsigned int128'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_UINT128)
# Floating point types
elif type_name in ('float', ):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_FLOAT)
elif type_name in ('double', ):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_DOUBLE)
elif type_name in ('long double', 'ldouble'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_LDOUBLE)
# Boolean type
elif type_name in ('bool', '_Bool', 'boolean'):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_BOOL)
# Void type
elif type_name in ('void', ):
return ida_typeinf.tinfo_t(ida_typeinf.BTF_VOID)
# If not a standard type, try to get a named type
tif = ida_typeinf.tinfo_t()
if tif.get_named_type(None, type_name, ida_typeinf.BTF_STRUCT):
return tif
if tif.get_named_type(None, type_name, ida_typeinf.BTF_TYPEDEF):
return tif
if tif.get_named_type(None, type_name, ida_typeinf.BTF_ENUM):
return tif
if tif.get_named_type(None, type_name, ida_typeinf.BTF_UNION):
return tif
if tif := ida_typeinf.tinfo_t(type_name):
return tif
raise IDAError(f"Unable to retrieve {type_name} type info object")
@jsonrpc
@idaread
def get_function_by_name(
name: Annotated[str, "Name of the function to get"]
) -> Function:
"""Get a function by its name"""
function_address = idaapi.get_name_ea(idaapi.BADADDR, name)
if function_address == idaapi.BADADDR:
# If map has not been created yet, create it
if len(DEMANGLED_TO_EA) == 0:
create_demangled_to_ea_map()
# Try to find the function in the map, else raise an error
if name in DEMANGLED_TO_EA:
function_address = DEMANGLED_TO_EA[name]
else:
raise IDAError(f"No function found with name {name}")
return get_function(function_address)
@jsonrpc
@idaread
def get_function_by_address(
address: Annotated[str, "Address of the function to get"],
) -> Function:
"""Get a function by its address"""
return get_function(parse_address(address))
@jsonrpc
@idaread
def get_current_address() -> str:
"""Get the address currently selected by the user"""
return hex(idaapi.get_screen_ea())
@jsonrpc
@idaread
def get_current_function() -> Optional[Function]:
"""Get the function currently selected by the user"""
return get_function(idaapi.get_screen_ea())
class ConvertedNumber(TypedDict):
decimal: str
hexadecimal: str
bytes: str
ascii: Optional[str]
binary: str
@jsonrpc
def convert_number(
text: Annotated[str, "Textual representation of the number to convert"],
size: Annotated[Optional[int], "Size of the variable in bytes"],
) -> ConvertedNumber:
"""Convert a number (decimal, hexadecimal) to different representations"""
try:
value = int(text, 0)
except ValueError:
raise IDAError(f"Invalid number: {text}")
# Estimate the size of the number
if not size:
size = 0
n = abs(value)
while n:
size += 1
n >>= 1
size += 7
size //= 8
# Convert the number to bytes
try:
bytes = value.to_bytes(size, "little", signed=True)
except OverflowError:
raise IDAError(f"Number {text} is too big for {size} bytes")
# Convert the bytes to ASCII
ascii = ""
for byte in bytes.rstrip(b"\x00"):
if byte >= 32 and byte <= 126:
ascii += chr(byte)
else:
ascii = None
break
return ConvertedNumber(
decimal=str(value),
hexadecimal=hex(value),
bytes=bytes.hex(" "),
ascii=ascii,
binary=bin(value),
)
T = TypeVar("T")
class Page(TypedDict, Generic[T]):
data: list[T]
next_offset: Optional[int]
def paginate(data: list[T], offset: int, count: int) -> Page[T]:
if count == 0:
count = len(data)
next_offset = offset + count
if next_offset >= len(data):
next_offset = None
return {
"data": data[offset:offset + count],
"next_offset": next_offset,
}
def pattern_filter(data: list[T], pattern: str, key: str) -> list[T]:
if not pattern:
return data
# TODO: implement /regex/ matching
def matches(item) -> bool:
return pattern.lower() in item[key].lower()
return list(filter(matches, data))
@jsonrpc
@idaread
def list_functions(
offset: Annotated[int, "Offset to start listing from (start at 0)"],
count: Annotated[int, "Number of functions to list (100 is a good default, 0 means remainder)"],
) -> Page[Function]:
"""List all functions in the database (paginated)"""
functions = [get_function(address) for address in idautils.Functions()]
return paginate(functions, offset, count)
class Global(TypedDict):
address: str
name: str
@jsonrpc
@idaread
def list_globals_filter(
offset: Annotated[int, "Offset to start listing from (start at 0)"],
count: Annotated[int, "Number of globals to list (100 is a good default, 0 means remainder)"],
filter: Annotated[str, "Filter to apply to the list (required parameter, empty string for no filter). Case-insensitive contains or /regex/ syntax"],
) -> Page[Global]:
"""List matching globals in the database (paginated, filtered)"""
globals: list[Global] = []
for addr, name in idautils.Names():
# Skip functions and none
if not idaapi.get_func(addr) or name is None:
globals += [Global(address=hex(addr), name=name)]
globals = pattern_filter(globals, filter, "name")
return paginate(globals, offset, count)
@jsonrpc
def list_globals(
offset: Annotated[int, "Offset to start listing from (start at 0)"],
count: Annotated[int, "Number of globals to list (100 is a good default, 0 means remainder)"],
) -> Page[Global]:
"""List all globals in the database (paginated)"""
return list_globals_filter(offset, count, "")
class Import(TypedDict):
address: str
imported_name: str
module: str
@jsonrpc
@idaread
def list_imports(
offset: Annotated[int, "Offset to start listing from (start at 0)"],
count: Annotated[int, "Number of imports to list (100 is a good default, 0 means remainder)"],
) -> Page[Import]:
""" List all imported symbols with their name and module (paginated) """
nimps = ida_nalt.get_import_module_qty()
rv = []
for i in range(nimps):
module_name = ida_nalt.get_import_module_name(i)
if not module_name:
module_name = "<unnamed>"
def imp_cb(ea, symbol_name, ordinal, acc):
if not symbol_name:
symbol_name = f"#{ordinal}"
acc += [Import(address=hex(ea), imported_name=symbol_name, module=module_name)]
return True
imp_cb_w_context = lambda ea, symbol_name, ordinal: imp_cb(ea, symbol_name, ordinal, rv)
ida_nalt.enum_import_names(i, imp_cb_w_context)
return paginate(rv, offset, count)
class String(TypedDict):
address: str
length: int
string: str
@jsonrpc
@idaread
def list_strings_filter(
offset: Annotated[int, "Offset to start listing from (start at 0)"],
count: Annotated[int, "Number of strings to list (100 is a good default, 0 means remainder)"],
filter: Annotated[str, "Filter to apply to the list (required parameter, empty string for no filter). Case-insensitive contains or /regex/ syntax"],
) -> Page[String]:
"""List matching strings in the database (paginated, filtered)"""
strings: list[String] = []
for item in idautils.Strings():
if item is None:
continue
try:
string = str(item)
if string:
strings += [
String(address=hex(item.ea), length=item.length, string=string),
]
except:
continue
strings = pattern_filter(strings, filter, "string")
return paginate(strings, offset, count)
@jsonrpc
def list_strings(
offset: Annotated[int, "Offset to start listing from (start at 0)"],
count: Annotated[int, "Number of strings to list (100 is a good default, 0 means remainder)"],
) -> Page[String]:
"""List all strings in the database (paginated)"""
return list_strings_filter(offset, count, "")
@jsonrpc
@idaread
def list_local_types():
"""List all Local types in the database"""
error = ida_hexrays.hexrays_failure_t()
locals = []
idati = ida_typeinf.get_idati()
type_count = ida_typeinf.get_ordinal_limit(idati)
for ordinal in range(1, type_count):
try:
tif = ida_typeinf.tinfo_t()
if tif.get_numbered_type(idati, ordinal):
type_name = tif.get_type_name()
if not type_name:
type_name = f"<Anonymous Type #{ordinal}>"
locals.append(f"\nType #{ordinal}: {type_name}")
if tif.is_udt():
c_decl_flags = (ida_typeinf.PRTYPE_MULTI | ida_typeinf.PRTYPE_TYPE | ida_typeinf.PRTYPE_SEMI | ida_typeinf.PRTYPE_DEF | ida_typeinf.PRTYPE_METHODS | ida_typeinf.PRTYPE_OFFSETS)
c_decl_output = tif._print(None, c_decl_flags)
if c_decl_output:
locals.append(f" C declaration:\n{c_decl_output}")
else:
simple_decl = tif._print(None, ida_typeinf.PRTYPE_1LINE | ida_typeinf.PRTYPE_TYPE | ida_typeinf.PRTYPE_SEMI)
if simple_decl:
locals.append(f" Simple declaration:\n{simple_decl}")
else:
message = f"\nType #{ordinal}: Failed to retrieve information."
if error.str:
message += f": {error.str}"
if error.errea != idaapi.BADADDR:
message += f"from (address: {hex(error.errea)})"
raise IDAError(message)
except:
continue
return locals
def decompile_checked(address: int) -> ida_hexrays.cfunc_t:
if not ida_hexrays.init_hexrays_plugin():
raise IDAError("Hex-Rays decompiler is not available")
error = ida_hexrays.hexrays_failure_t()
cfunc = ida_hexrays.decompile_func(address, error, ida_hexrays.DECOMP_WARNINGS)
if not cfunc:
if error.code == ida_hexrays.MERR_LICENSE:
raise IDAError("Decompiler license is not available. Use `disassemble_function` to get the assembly code instead.")
message = f"Decompilation failed at {hex(address)}"
if error.str:
message += f": {error.str}"
if error.errea != idaapi.BADADDR:
message += f" (address: {hex(error.errea)})"
raise IDAError(message)
return cfunc # type: ignore (this is a SWIG issue)
@jsonrpc
@idaread
def decompile_function(
address: Annotated[str, "Address of the function to decompile"],
) -> str:
"""Decompile a function at the given address"""
start = parse_address(address)
cfunc = decompile_checked(start)
if is_window_active():
ida_hexrays.open_pseudocode(start, ida_hexrays.OPF_REUSE)
sv = cfunc.get_pseudocode()
pseudocode = ""
for i, sl in enumerate(sv):
sl: ida_kernwin.simpleline_t
item = ida_hexrays.ctree_item_t()
addr = None if i > 0 else cfunc.entry_ea
if cfunc.get_line_item(sl.line, 0, False, None, item, None): # type: ignore (IDA SDK type hint wrong)
dstr: str | None = item.dstr()
if dstr:
ds = dstr.split(": ")
if len(ds) == 2:
try:
addr = int(ds[0], 16)
except ValueError:
pass
line = ida_lines.tag_remove(sl.line)
if len(pseudocode) > 0:
pseudocode += "\n"
if not addr:
pseudocode += f"/* line: {i} */ {line}"
else:
pseudocode += f"/* line: {i}, address: {hex(addr)} */ {line}"
return pseudocode
class DisassemblyLine(TypedDict):
segment: NotRequired[str]
address: str
label: NotRequired[str]
instruction: str
comments: NotRequired[list[str]]
class Argument(TypedDict):
name: str
type: str
class StackFrameVariable(TypedDict):
name: str
offset: str
size: str
type: str
class DisassemblyFunction(TypedDict):
name: str
start_ea: str
return_type: NotRequired[str]
arguments: NotRequired[list[Argument]]
stack_frame: list[StackFrameVariable]
lines: list[DisassemblyLine]
@jsonrpc
@idaread
def disassemble_function(
start_address: Annotated[str, "Address of the function to disassemble"],
) -> DisassemblyFunction:
"""Get assembly code for a function (API-compatible with older IDA builds)"""
start = parse_address(start_address)
func = idaapi.get_func(start)
if not func:
raise IDAError(f"No function found at address {hex(start)}")
if is_window_active():
ida_kernwin.jumpto(start)
func_name: str = ida_funcs.get_func_name(func.start_ea) or "<unnamed>"
lines: list[DisassemblyLine] = []
for ea in idautils.FuncItems(func.start_ea):
if ea == idaapi.BADADDR:
continue
seg = idaapi.getseg(ea)
segment: str | None = idaapi.get_segm_name(seg) if seg else None
label: str | None = idc.get_name(ea, 0)
if not label or (label == func_name and ea == func.start_ea):
label = None
comments: list[str] = []
c: str | None = idaapi.get_cmt(ea, False)
if c:
comments.append(c)
c = idaapi.get_cmt(ea, True)
if c:
comments.append(c)
mnem: str = idc.print_insn_mnem(ea) or ""
ops: list[str] = []
for n in range(8):
if idc.get_operand_type(ea, n) == idaapi.o_void:
break
ops.append(idc.print_operand(ea, n) or "")
instruction = f"{mnem} {', '.join(ops)}".rstrip()
line: DisassemblyLine = {
"address": hex(ea),
"instruction": instruction
}
if segment:
line["segment"] = segment
if label:
line["label"] = label
if comments:
line["comments"] = comments
lines.append(line)
# prototype and args via tinfo (safe across versions)
rettype = None
args: Optional[list[Argument]] = None
tif = ida_typeinf.tinfo_t()
if ida_nalt.get_tinfo(tif, func.start_ea) and tif.is_func():
ftd = ida_typeinf.func_type_data_t()
if tif.get_func_details(ftd):
rettype = str(ftd.rettype)
args = [Argument(name=(a.name or f"arg{i}"), type=str(a.type))
for i, a in enumerate(ftd)]
out: DisassemblyFunction = {
"name": func_name,
"start_ea": hex(func.start_ea),
"stack_frame": get_stack_frame_variables_internal(func.start_ea, False),
"lines": lines,
}
if rettype:
out["return_type"] = rettype
if args is not None:
out["arguments"] = args
return out
class Xref(TypedDict):
address: str
type: str
function: Optional[Function]
@jsonrpc
@idaread
def get_xrefs_to(
address: Annotated[str, "Address to get cross references to"],
) -> list[Xref]:
"""Get all cross references to the given address"""
xrefs = []
xref: ida_xref.xrefblk_t
for xref in idautils.XrefsTo(parse_address(address)): # type: ignore (IDA SDK type hints are incorrect)
xrefs += [
Xref(address=hex(xref.frm),
type="code" if xref.iscode else "data",
function=get_function(xref.frm, raise_error=False))
]
return xrefs
@jsonrpc
@idaread
def get_xrefs_to_field(
struct_name: Annotated[str, "Name of the struct (type) containing the field"],
field_name: Annotated[str, "Name of the field (member) to get xrefs to"],
) -> list[Xref]:
"""Get all cross references to a named struct field (member)"""
# Get the type library
til = ida_typeinf.get_idati()
if not til:
raise IDAError("Failed to retrieve type library.")
# Get the structure type info
tif = ida_typeinf.tinfo_t()
if not tif.get_named_type(til, struct_name, ida_typeinf.BTF_STRUCT, True, False):
print(f"Structure '{struct_name}' not found.")
return []
# Get The field index
idx = ida_typeinf.get_udm_by_fullname(None, struct_name + '.' + field_name) # type: ignore (IDA SDK type hints are incorrect)
if idx == -1:
print(f"Field '{field_name}' not found in structure '{struct_name}'.")
return []
# Get the type identifier
tid = tif.get_udm_tid(idx)
if tid == ida_idaapi.BADADDR:
raise IDAError(f"Unable to get tid for structure '{struct_name}' and field '{field_name}'.")
# Get xrefs to the tid
xrefs = []
xref: ida_xref.xrefblk_t
for xref in idautils.XrefsTo(tid): # type: ignore (IDA SDK type hints are incorrect)
xrefs += [
Xref(address=hex(xref.frm),
type="code" if xref.iscode else "data",
function=get_function(xref.frm, raise_error=False))
]
return xrefs
@jsonrpc
@idaread
def get_callees(
function_address: Annotated[str, "Address of the function to get callee functions"],
) -> list[dict[str, str]]:
"""Get all the functions called (callees) by the function at function_address"""
func_start = parse_address(function_address)
func = idaapi.get_func(func_start)
if not func:
raise IDAError(f"No function found containing address {function_address}")
func_end = idc.find_func_end(func_start)
callees: list[dict[str, str]] = []
current_ea = func_start
while current_ea < func_end:
insn = idaapi.insn_t()
idaapi.decode_insn(insn, current_ea)
if insn.itype in [idaapi.NN_call, idaapi.NN_callfi, idaapi.NN_callni]:
target = idc.get_operand_value(current_ea, 0)
target_type = idc.get_operand_type(current_ea, 0)
# check if it's a direct call - avoid getting the indirect call offset
if target_type in [idaapi.o_mem, idaapi.o_near, idaapi.o_far]:
# in here, we do not use get_function because the target can be external function.
# but, we should mark the target as internal/external function.
func_type = (
"internal" if idaapi.get_func(target) is not None else "external"
)
func_name = idc.get_name(target)
if func_name is not None:
callees.append(
{"address": hex(target), "name": func_name, "type": func_type}
)
current_ea = idc.next_head(current_ea, func_end)
# deduplicate callees
unique_callee_tuples = {tuple(callee.items()) for callee in callees}
unique_callees = [dict(callee) for callee in unique_callee_tuples]
return unique_callees # type: ignore
@jsonrpc
@idaread
def get_callers(
function_address: Annotated[str, "Address of the function to get callers"],
) -> list[Function]:
"""Get all callers of the given address"""
callers = {}
for caller_address in idautils.CodeRefsTo(parse_address(function_address), 0):
# validate the xref address is a function
func = get_function(caller_address, raise_error=False)
if not func:
continue
# load the instruction at the xref address
insn = idaapi.insn_t()
idaapi.decode_insn(insn, caller_address)
# check the instruction is a call
if insn.itype not in [idaapi.NN_call, idaapi.NN_callfi, idaapi.NN_callni]:
continue
# deduplicate callers by address
callers[func["address"]] = func
return list(callers.values())
@jsonrpc
@idaread
def get_entry_points() -> list[Function]:
"""Get all entry points in the database"""
result = []
for i in range(ida_entry.get_entry_qty()):
ordinal = ida_entry.get_entry_ordinal(i)
address = ida_entry.get_entry(ordinal)
func = get_function(address, raise_error=False)
if func is not None:
result.append(func)
return result
@jsonrpc
@idawrite
def set_comment(
address: Annotated[str, "Address in the function to set the comment for"],
comment: Annotated[str, "Comment text"],
):
"""Set a comment for a given address in the function disassembly and pseudocode"""
ea = parse_address(address)
if not idaapi.set_cmt(ea, comment, False):
raise IDAError(f"Failed to set disassembly comment at {hex(ea)}")
if not ida_hexrays.init_hexrays_plugin():
return
# Reference: https://cyber.wtf/2019/03/22/using-ida-python-to-analyze-trickbot/
# Check if the address corresponds to a line
try:
cfunc = decompile_checked(ea)
except IDAError:
# Skip decompiler comment if decompilation fails
return
# Special case for function entry comments
if ea == cfunc.entry_ea:
idc.set_func_cmt(ea, comment, True)
cfunc.refresh_func_ctext()
return
eamap = cfunc.get_eamap()
if ea not in eamap:
print(f"Failed to set decompiler comment at {hex(ea)}")
return
nearest_ea = eamap[ea][0].ea
# Remove existing orphan comments
if cfunc.has_orphan_cmts():
cfunc.del_orphan_cmts()
cfunc.save_user_cmts()
# Set the comment by trying all possible item types
tl = idaapi.treeloc_t()
tl.ea = nearest_ea
for itp in range(idaapi.ITP_SEMI, idaapi.ITP_COLON):
tl.itp = itp
cfunc.set_user_cmt(tl, comment)
cfunc.save_user_cmts()
cfunc.refresh_func_ctext()
if not cfunc.has_orphan_cmts():
return
cfunc.del_orphan_cmts()
cfunc.save_user_cmts()
print(f"Failed to set decompiler comment at {hex(ea)}")
def refresh_decompiler_widget():
widget = ida_kernwin.get_current_widget()
if widget is not None:
vu = ida_hexrays.get_widget_vdui(widget)
if vu is not None:
vu.refresh_ctext()
def refresh_decompiler_ctext(function_address: int):
error = ida_hexrays.hexrays_failure_t()
cfunc: ida_hexrays.cfunc_t = ida_hexrays.decompile_func(function_address, error, ida_hexrays.DECOMP_WARNINGS)
if cfunc:
cfunc.refresh_func_ctext()
@jsonrpc
@idawrite
def rename_local_variable(
function_address: Annotated[str, "Address of the function containing the variable"],
old_name: Annotated[str, "Current name of the variable"],
new_name: Annotated[str, "New name for the variable (empty for a default name)"],
):
"""Rename a local variable in a function"""
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
if not ida_hexrays.rename_lvar(func.start_ea, old_name, new_name):
raise IDAError(f"Failed to rename local variable {old_name} in function {hex(func.start_ea)}")
refresh_decompiler_ctext(func.start_ea)
@jsonrpc
@idawrite
def rename_global_variable(
old_name: Annotated[str, "Current name of the global variable"],
new_name: Annotated[str, "New name for the global variable (empty for a default name)"],
):
"""Rename a global variable"""
ea = idaapi.get_name_ea(idaapi.BADADDR, old_name)
if not idaapi.set_name(ea, new_name):
raise IDAError(f"Failed to rename global variable {old_name} to {new_name}")
refresh_decompiler_ctext(ea)
@jsonrpc
@idawrite
def set_global_variable_type(
variable_name: Annotated[str, "Name of the global variable"],
new_type: Annotated[str, "New type for the variable"],
):
"""Set a global variable's type"""
ea = idaapi.get_name_ea(idaapi.BADADDR, variable_name)
tif = get_type_by_name(new_type)
if not tif:
raise IDAError(f"Parsed declaration is not a variable type")
if not ida_typeinf.apply_tinfo(ea, tif, ida_typeinf.PT_SIL):
raise IDAError(f"Failed to apply type")
def patch_address_assemble(
ea: int,
assemble: str,
) -> int:
"""Patch Address Assemble"""
(check_assemble, bytes_to_patch) = idautils.Assemble(ea, assemble)
if check_assemble == False:
raise IDAError(f"Failed to assemble instruction: {assemble}")
try:
ida_bytes.patch_bytes(ea, bytes_to_patch)
except:
raise IDAError(f"Failed to patch bytes at address {hex(ea)}")
return len(bytes_to_patch)
@jsonrpc
@idawrite
def patch_address_assembles(
address: Annotated[str, "Starting Address to apply patch"],
instructions: Annotated[str, "Assembly instructions separated by ';'"],
) -> str:
ea = parse_address(address)
assembles = instructions.split(";")
for assemble in assembles:
assemble = assemble.strip()
try:
patch_bytes_len = patch_address_assemble(ea, assemble)
except IDAError as e:
raise IDAError(f"Failed to patch bytes at address {hex(ea)}: {e}")
ea += patch_bytes_len
return f"Patched {len(assembles)} instructions"
@jsonrpc
@idaread
def get_global_variable_value_by_name(variable_name: Annotated[str, "Name of the global variable"]) -> str:
"""
Read a global variable's value (if known at compile-time)
Prefer this function over the `data_read_*` functions.
"""
ea = idaapi.get_name_ea(idaapi.BADADDR, variable_name)
if ea == idaapi.BADADDR:
raise IDAError(f"Global variable {variable_name} not found")
return get_global_variable_value_internal(ea)
@jsonrpc
@idaread
def get_global_variable_value_at_address(address: Annotated[str, "Address of the global variable"]) -> str:
"""
Read a global variable's value by its address (if known at compile-time)
Prefer this function over the `data_read_*` functions.
"""
ea = parse_address(address)
return get_global_variable_value_internal(ea)
def get_global_variable_value_internal(ea: int) -> str:
# Get the type information for the variable
tif = ida_typeinf.tinfo_t()
if not ida_nalt.get_tinfo(tif, ea):
# No type info, maybe we can figure out its size by its name
if not ida_bytes.has_any_name(ea):
raise IDAError(f"Failed to get type information for variable at {ea:#x}")
size = ida_bytes.get_item_size(ea)
if size == 0:
raise IDAError(f"Failed to get type information for variable at {ea:#x}")
else:
# Determine the size of the variable
size = tif.get_size()
# Read the value based on the size
if size == 0 and tif.is_array() and tif.get_array_element().is_decl_char():
return_string = idaapi.get_strlit_contents(ea, -1, 0).decode("utf-8").strip()
return f"\"{return_string}\""
elif size == 1:
return hex(ida_bytes.get_byte(ea))
elif size == 2:
return hex(ida_bytes.get_word(ea))
elif size == 4:
return hex(ida_bytes.get_dword(ea))
elif size == 8:
return hex(ida_bytes.get_qword(ea))
else:
# For other sizes, return the raw bytes
return ' '.join(hex(x) for x in ida_bytes.get_bytes(ea, size))
@jsonrpc
@idawrite
def rename_function(
function_address: Annotated[str, "Address of the function to rename"],
new_name: Annotated[str, "New name for the function (empty for a default name)"],
):
"""Rename a function"""
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
if not idaapi.set_name(func.start_ea, new_name):
raise IDAError(f"Failed to rename function {hex(func.start_ea)} to {new_name}")
refresh_decompiler_ctext(func.start_ea)
@jsonrpc
@idawrite
def set_function_prototype(
function_address: Annotated[str, "Address of the function"],
prototype: Annotated[str, "New function prototype"],
):
"""Set a function's prototype"""
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
try:
tif = ida_typeinf.tinfo_t(prototype, None, ida_typeinf.PT_SIL)
if not tif.is_func():
raise IDAError(f"Parsed declaration is not a function type")
if not ida_typeinf.apply_tinfo(func.start_ea, tif, ida_typeinf.PT_SIL):
raise IDAError(f"Failed to apply type")
refresh_decompiler_ctext(func.start_ea)
except Exception as e:
raise IDAError(f"Failed to parse prototype string: {prototype}")
class my_modifier_t(ida_hexrays.user_lvar_modifier_t):
def __init__(self, var_name: str, new_type: ida_typeinf.tinfo_t):
ida_hexrays.user_lvar_modifier_t.__init__(self)
self.var_name = var_name
self.new_type = new_type
def modify_lvars(self, lvinf):
for lvar_saved in lvinf.lvvec:
lvar_saved: ida_hexrays.lvar_saved_info_t
if lvar_saved.name == self.var_name:
lvar_saved.type = self.new_type
return True
return False
# NOTE: This is extremely hacky, but necessary to get errors out of IDA
def parse_decls_ctypes(decls: str, hti_flags: int) -> tuple[int, list[str]]:
if sys.platform == "win32":
import ctypes
assert isinstance(decls, str), "decls must be a string"
assert isinstance(hti_flags, int), "hti_flags must be an int"
c_decls = decls.encode("utf-8")
c_til = None
ida_dll = ctypes.CDLL("ida")
ida_dll.parse_decls.argtypes = [
ctypes.c_void_p,
ctypes.c_char_p,
ctypes.c_void_p,
ctypes.c_int,
]
ida_dll.parse_decls.restype = ctypes.c_int
messages: list[str] = []
@ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_char_p, ctypes.c_char_p)
def magic_printer(fmt: bytes, arg1: bytes):
if fmt.count(b"%") == 1 and b"%s" in fmt:
formatted = fmt.replace(b"%s", arg1)
messages.append(formatted.decode("utf-8"))
return len(formatted) + 1
else:
messages.append(f"unsupported magic_printer fmt: {repr(fmt)}")
return 0
errors = ida_dll.parse_decls(c_til, c_decls, magic_printer, hti_flags)
else:
# NOTE: The approach above could also work on other platforms, but it's
# not been tested and there are differences in the vararg ABIs.
errors = ida_typeinf.parse_decls(None, decls, False, hti_flags)
messages = []
return errors, messages
@jsonrpc
@idawrite
def declare_c_type(
c_declaration: Annotated[str, "C declaration of the type. Examples include: typedef int foo_t; struct bar { int a; bool b; };"],
):
"""Create or update a local type from a C declaration"""
# PT_SIL: Suppress warning dialogs (although it seems unnecessary here)
# PT_EMPTY: Allow empty types (also unnecessary?)
# PT_TYP: Print back status messages with struct tags
flags = ida_typeinf.PT_SIL | ida_typeinf.PT_EMPTY | ida_typeinf.PT_TYP
errors, messages = parse_decls_ctypes(c_declaration, flags)
pretty_messages = "\n".join(messages)
if errors > 0:
raise IDAError(f"Failed to parse type:\n{c_declaration}\n\nErrors:\n{pretty_messages}")
return f"success\n\nInfo:\n{pretty_messages}"
@jsonrpc
@idawrite
def set_local_variable_type(
function_address: Annotated[str, "Address of the decompiled function containing the variable"],
variable_name: Annotated[str, "Name of the variable"],
new_type: Annotated[str, "New type for the variable"],
):
"""Set a local variable's type"""
try:
# Some versions of IDA don't support this constructor
new_tif = ida_typeinf.tinfo_t(new_type, None, ida_typeinf.PT_SIL)
except Exception:
try:
new_tif = ida_typeinf.tinfo_t()
# parse_decl requires semicolon for the type
ida_typeinf.parse_decl(new_tif, None, new_type + ";", ida_typeinf.PT_SIL) # type: ignore (IDA SDK type hints are incorrect)
except Exception:
raise IDAError(f"Failed to parse type: {new_type}")
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
if not ida_hexrays.rename_lvar(func.start_ea, variable_name, variable_name):
raise IDAError(f"Failed to find local variable: {variable_name}")
modifier = my_modifier_t(variable_name, new_tif)
if not ida_hexrays.modify_user_lvars(func.start_ea, modifier):
raise IDAError(f"Failed to modify local variable: {variable_name}")
refresh_decompiler_ctext(func.start_ea)
@jsonrpc
@idaread
def get_stack_frame_variables(
function_address: Annotated[str, "Address of the disassembled function to retrieve the stack frame variables"]
) -> list[StackFrameVariable]:
""" Retrieve the stack frame variables for a given function """
return get_stack_frame_variables_internal(parse_address(function_address), True)
def get_stack_frame_variables_internal(function_address: int, raise_error: bool) -> list[StackFrameVariable]:
# TODO: IDA 8.3 does not support tif.get_type_by_tid
if ida_major < 9:
return []
func = idaapi.get_func(function_address)
if not func:
if raise_error:
raise IDAError(f"No function found at address {function_address}")
return []
tif = ida_typeinf.tinfo_t()
if not tif.get_type_by_tid(func.frame) or not tif.is_udt():
return []
members: list[StackFrameVariable] = []
udt = ida_typeinf.udt_type_data_t()
tif.get_udt_details(udt)
for udm in udt:
if not udm.is_gap():
name = udm.name
offset = udm.offset // 8
size = udm.size // 8
type = str(udm.type)
members.append(StackFrameVariable(
name=name,
offset=hex(offset),
size=hex(size),
type=type
))
return members
class StructureMember(TypedDict):
name: str
offset: str
size: str
type: str
class StructureDefinition(TypedDict):
name: str
size: str
members: list[StructureMember]
@jsonrpc
@idaread
def get_defined_structures() -> list[StructureDefinition]:
""" Returns a list of all defined structures """
rv = []
limit = ida_typeinf.get_ordinal_limit()
for ordinal in range(1, limit):
tif = ida_typeinf.tinfo_t()
tif.get_numbered_type(None, ordinal)
if tif.is_udt():
udt = ida_typeinf.udt_type_data_t()
members = []
if tif.get_udt_details(udt):
members = [
StructureMember(name=x.name,
offset=hex(x.offset // 8),
size=hex(x.size // 8),
type=str(x.type))
for _, x in enumerate(udt)
]
rv += [StructureDefinition(name=tif.get_type_name(), # type: ignore (IDA SDK type hints are incorrect)
size=hex(tif.get_size()),
members=members)]
return rv
@jsonrpc
@idaread
def analyze_struct_detailed(name: Annotated[str, "Name of the structure to analyze"]) -> dict:
"""Detailed analysis of a structure with all fields"""
# Get tinfo object
tif = ida_typeinf.tinfo_t()
if not tif.get_named_type(None, name):
raise IDAError(f"Structure '{name}' not found!")
result = {
"name": name,
"type": str(tif._print()),
"size": tif.get_size(),
"is_udt": tif.is_udt()
}
if not tif.is_udt():
result["error"] = "This is not a user-defined type!"
return result
# Get UDT (User Defined Type) details
udt_data = ida_typeinf.udt_type_data_t()
if not tif.get_udt_details(udt_data):
result["error"] = "Failed to get structure details!"
return result
result["member_count"] = udt_data.size()
result["is_union"] = udt_data.is_union
result["udt_type"] = "Union" if udt_data.is_union else "Struct"
# Output information about each field
members = []
for i, member in enumerate(udt_data):
offset = member.begin() // 8 # Convert bits to bytes
size = member.size // 8 if member.size > 0 else member.type.get_size()
member_type = member.type._print()
member_name = member.name
member_info = {
"index": i,
"offset": f"0x{offset:08X}",
"size": size,
"type": member_type,
"name": member_name,
"is_nested_udt": member.type.is_udt()
}
# If this is a nested structure, show additional information
if member.type.is_udt():
member_info["nested_size"] = member.type.get_size()
members.append(member_info)
result["members"] = members
result["total_size"] = tif.get_size()
return result
@jsonrpc
@idaread
def get_struct_at_address(address: Annotated[str, "Address to analyze structure at"],
struct_name: Annotated[str, "Name of the structure"]) -> dict:
"""Get structure field values at a specific address"""
addr = parse_address(address)
# Get structure tinfo
tif = ida_typeinf.tinfo_t()
if not tif.get_named_type(None, struct_name):
raise IDAError(f"Structure '{struct_name}' not found!")
# Get structure details
udt_data = ida_typeinf.udt_type_data_t()
if not tif.get_udt_details(udt_data):
raise IDAError("Failed to get structure details!")
result = {
"struct_name": struct_name,
"address": f"0x{addr:X}",
"members": []
}
for member in udt_data:
offset = member.begin() // 8
member_addr = addr + offset
member_type = member.type._print()
member_name = member.name
member_size = member.type.get_size()
# Try to get value based on size
try:
if member.type.is_ptr():
# Pointer
is_64bit = ida_ida.inf_is_64bit() if ida_major >= 9 else idaapi.get_inf_structure().is_64bit()
if is_64bit:
value = idaapi.get_qword(member_addr)
value_str = f"0x{value:016X}"
else:
value = idaapi.get_dword(member_addr)
value_str = f"0x{value:08X}"
elif member_size == 1:
value = idaapi.get_byte(member_addr)
value_str = f"0x{value:02X} ({value})"
elif member_size == 2:
value = idaapi.get_word(member_addr)
value_str = f"0x{value:04X} ({value})"
elif member_size == 4:
value = idaapi.get_dword(member_addr)
value_str = f"0x{value:08X} ({value})"
elif member_size == 8:
value = idaapi.get_qword(member_addr)
value_str = f"0x{value:016X} ({value})"
else:
# For large structures, read first few bytes
bytes_data = []
for i in range(min(member_size, 16)):
try:
byte_val = idaapi.get_byte(member_addr + i)
bytes_data.append(f"{byte_val:02X}")
except:
break
value_str = f"[{' '.join(bytes_data)}{'...' if member_size > 16 else ''}]"
except:
value_str = "<failed to read>"
member_info = {
"offset": f"0x{offset:08X}",
"type": member_type,
"name": member_name,
"value": value_str
}
result["members"].append(member_info)
return result
@jsonrpc
@idaread
def get_struct_info_simple(name: Annotated[str, "Name of the structure"]) -> dict:
"""Simple function to get basic structure information"""
tif = ida_typeinf.tinfo_t()
if not tif.get_named_type(None, name):
raise IDAError(f"Structure '{name}' not found!")
info = {
'name': name,
'type': tif._print(),
'size': tif.get_size(),
'is_udt': tif.is_udt()
}
if tif.is_udt():
udt_data = ida_typeinf.udt_type_data_t()
if tif.get_udt_details(udt_data):
info['member_count'] = udt_data.size()
info['is_union'] = udt_data.is_union
members = []
for member in udt_data:
members.append({
'name': member.name,
'type': member.type._print(),
'offset': member.begin() // 8,
'size': member.type.get_size()
})
info['members'] = members
return info
@jsonrpc
@idaread
def search_structures(filter: Annotated[str, "Filter pattern to search for structures (case-insensitive)"]) -> list[dict]:
"""Search for structures by name pattern"""
results = []
limit = ida_typeinf.get_ordinal_limit()
for ordinal in range(1, limit):
tif = ida_typeinf.tinfo_t()
if tif.get_numbered_type(None, ordinal):
type_name: str = tif.get_type_name() # type: ignore (IDA SDK type hints are incorrect)
if type_name and filter.lower() in type_name.lower():
if tif.is_udt():
udt_data = ida_typeinf.udt_type_data_t()
member_count = 0
if tif.get_udt_details(udt_data):
member_count = udt_data.size()
results.append({
"name": type_name,
"size": tif.get_size(),
"member_count": member_count,
"is_union": udt_data.is_union if tif.get_udt_details(udt_data) else False,
"ordinal": ordinal
})
return results
@jsonrpc
@idawrite
def rename_stack_frame_variable(
function_address: Annotated[str, "Address of the disassembled function to set the stack frame variables"],
old_name: Annotated[str, "Current name of the variable"],
new_name: Annotated[str, "New name for the variable (empty for a default name)"]
):
""" Change the name of a stack variable for an IDA function """
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
frame_tif = ida_typeinf.tinfo_t()
if not ida_frame.get_func_frame(frame_tif, func):
raise IDAError("No frame returned.")
idx, udm = frame_tif.get_udm(old_name) # type: ignore (IDA SDK type hints are incorrect)
if not udm:
raise IDAError(f"{old_name} not found.")
tid = frame_tif.get_udm_tid(idx)
if ida_frame.is_special_frame_member(tid):
raise IDAError(f"{old_name} is a special frame member. Will not change the name.")
udm = ida_typeinf.udm_t()
frame_tif.get_udm_by_tid(udm, tid)
offset = udm.offset // 8
if ida_frame.is_funcarg_off(func, offset):
raise IDAError(f"{old_name} is an argument member. Will not change the name.")
sval = ida_frame.soff_to_fpoff(func, offset)
if not ida_frame.define_stkvar(func, new_name, sval, udm.type):
raise IDAError("failed to rename stack frame variable")
@jsonrpc
@idawrite
def create_stack_frame_variable(
function_address: Annotated[str, "Address of the disassembled function to set the stack frame variables"],
offset: Annotated[str, "Offset of the stack frame variable"],
variable_name: Annotated[str, "Name of the stack variable"],
type_name: Annotated[str, "Type of the stack variable"]
):
""" For a given function, create a stack variable at an offset and with a specific type """
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
ea = parse_address(offset)
frame_tif = ida_typeinf.tinfo_t()
if not ida_frame.get_func_frame(frame_tif, func):
raise IDAError("No frame returned.")
tif = get_type_by_name(type_name)
if not ida_frame.define_stkvar(func, variable_name, ea, tif):
raise IDAError("failed to define stack frame variable")
@jsonrpc
@idawrite
def set_stack_frame_variable_type(
function_address: Annotated[str, "Address of the disassembled function to set the stack frame variables"],
variable_name: Annotated[str, "Name of the stack variable"],
type_name: Annotated[str, "Type of the stack variable"]
):
""" For a given disassembled function, set the type of a stack variable """
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
frame_tif = ida_typeinf.tinfo_t()
if not ida_frame.get_func_frame(frame_tif, func):
raise IDAError("No frame returned.")
idx, udm = frame_tif.get_udm(variable_name) # type: ignore (IDA SDK type hints are incorrect)
if not udm:
raise IDAError(f"{variable_name} not found.")
tid = frame_tif.get_udm_tid(idx)
udm = ida_typeinf.udm_t()
frame_tif.get_udm_by_tid(udm, tid)
offset = udm.offset // 8
tif = get_type_by_name(type_name)
if not ida_frame.set_frame_member_type(func, offset, tif):
raise IDAError("failed to set stack frame variable type")
@jsonrpc
@idawrite
def delete_stack_frame_variable(
function_address: Annotated[str, "Address of the function to set the stack frame variables"],
variable_name: Annotated[str, "Name of the stack variable"]
):
""" Delete the named stack variable for a given function """
func = idaapi.get_func(parse_address(function_address))
if not func:
raise IDAError(f"No function found at address {function_address}")
frame_tif = ida_typeinf.tinfo_t()
if not ida_frame.get_func_frame(frame_tif, func):
raise IDAError("No frame returned.")
idx, udm = frame_tif.get_udm(variable_name) # type: ignore (IDA SDK type hints are incorrect)
if not udm:
raise IDAError(f"{variable_name} not found.")
tid = frame_tif.get_udm_tid(idx)
if ida_frame.is_special_frame_member(tid):
raise IDAError(f"{variable_name} is a special frame member. Will not delete.")
udm = ida_typeinf.udm_t()
frame_tif.get_udm_by_tid(udm, tid)
offset = udm.offset // 8
size = udm.size // 8
if ida_frame.is_funcarg_off(func, offset):
raise IDAError(f"{variable_name} is an argument member. Will not delete.")
if not ida_frame.delete_frame_members(func, offset, offset+size):
raise IDAError("failed to delete stack frame variable")
@jsonrpc
@idaread
def read_memory_bytes(
memory_address: Annotated[str, "Address of the memory value to be read"],
size: Annotated[int, "size of memory to read"]
) -> str:
"""
Read bytes at a given address.
Only use this function if `get_global_variable_at` and `get_global_variable_by_name`
both failed.
"""
return ' '.join(f'{x:#02x}' for x in ida_bytes.get_bytes(parse_address(memory_address), size))
@jsonrpc
@idaread
def data_read_byte(
address: Annotated[str, "Address to get 1 byte value from"],
) -> int:
"""
Read the 1 byte value at the specified address.
Only use this function if `get_global_variable_at` failed.
"""
ea = parse_address(address)
return ida_bytes.get_wide_byte(ea)
@jsonrpc
@idaread
def data_read_word(
address: Annotated[str, "Address to get 2 bytes value from"],
) -> int:
"""
Read the 2 byte value at the specified address as a WORD.
Only use this function if `get_global_variable_at` failed.
"""
ea = parse_address(address)
return ida_bytes.get_wide_word(ea)
@jsonrpc
@idaread
def data_read_dword(
address: Annotated[str, "Address to get 4 bytes value from"],
) -> int:
"""
Read the 4 byte value at the specified address as a DWORD.
Only use this function if `get_global_variable_at` failed.
"""
ea = parse_address(address)
return ida_bytes.get_wide_dword(ea)
@jsonrpc
@idaread
def data_read_qword(
address: Annotated[str, "Address to get 8 bytes value from"]
) -> int:
"""
Read the 8 byte value at the specified address as a QWORD.
Only use this function if `get_global_variable_at` failed.
"""
ea = parse_address(address)
return ida_bytes.get_qword(ea)
@jsonrpc
@idaread
def data_read_string(
address: Annotated[str, "Address to get string from"]
) -> str:
"""
Read the string at the specified address.
Only use this function if `get_global_variable_at` failed.
"""
try:
return idaapi.get_strlit_contents(parse_address(address),-1,0).decode("utf-8")
except Exception as e:
return "Error:" + str(e)
class RegisterValue(TypedDict):
name: str
value: str
class ThreadRegisters(TypedDict):
thread_id: int
registers: list[RegisterValue]
def dbg_ensure_running() -> "ida_idd.debugger_t":
dbg = ida_idd.get_dbg()
if not dbg:
raise IDAError("Debugger not running")
if ida_dbg.get_ip_val() is None:
raise IDAError("Debugger not running")
return dbg
@jsonrpc
@idaread
@unsafe
def dbg_get_registers() -> list[ThreadRegisters]:
"""Get all registers and their values. This function is only available when debugging."""
result: list[ThreadRegisters] = []
dbg = dbg_ensure_running()
for thread_index in range(ida_dbg.get_thread_qty()):
tid = ida_dbg.getn_thread(thread_index)
regs = []
regvals: ida_idd.regvals_t = ida_dbg.get_reg_vals(tid)
for reg_index, rv in enumerate(regvals):
rv: ida_idd.regval_t
reg_info = dbg.regs(reg_index)
# NOTE: Apparently this can fail under some circumstances
try:
reg_value = rv.pyval(reg_info.dtype)
except ValueError:
reg_value = ida_idaapi.BADADDR
if isinstance(reg_value, int):
reg_value = hex(reg_value)
if isinstance(reg_value, bytes):
reg_value = reg_value.hex(" ")
else:
reg_value = str(reg_value)
regs.append({
"name": reg_info.name,
"value": reg_value,
})
result.append({
"thread_id": tid,
"registers": regs,
})
return result
@jsonrpc
@idaread
@unsafe
def dbg_get_call_stack() -> list[dict[str, str]]:
"""Get the current call stack."""
callstack = []
try:
tid = ida_dbg.get_current_thread()
trace = ida_idd.call_stack_t()
if not ida_dbg.collect_stack_trace(tid, trace):
return []
for frame in trace:
frame_info = {
"address": hex(frame.callea),
}
try:
module_info = ida_idd.modinfo_t()
if ida_dbg.get_module_info(frame.callea, module_info):
frame_info["module"] = os.path.basename(module_info.name)
else:
frame_info["module"] = "<unknown>"
name = (
ida_name.get_nice_colored_name(
frame.callea,
ida_name.GNCN_NOCOLOR
| ida_name.GNCN_NOLABEL
| ida_name.GNCN_NOSEG
| ida_name.GNCN_PREFDBG,
)
or "<unnamed>"
)
frame_info["symbol"] = name
except Exception as e:
frame_info["module"] = "<error>"
frame_info["symbol"] = str(e)
callstack.append(frame_info)
except Exception as e:
pass
return callstack
class Breakpoint(TypedDict):
ea: str
enabled: bool
condition: Optional[str]
def list_breakpoints():
breakpoints: list[Breakpoint] = []
for i in range(ida_dbg.get_bpt_qty()):
bpt = ida_dbg.bpt_t()
if ida_dbg.getn_bpt(i, bpt):
breakpoints.append(Breakpoint(
ea=hex(bpt.ea),
enabled=bpt.flags & ida_dbg.BPT_ENABLED,
condition=str(bpt.condition) if bpt.condition else None,
))
return breakpoints
@jsonrpc
@idaread
@unsafe
def dbg_list_breakpoints():
"""List all breakpoints in the program."""
return list_breakpoints()
@jsonrpc
@idaread
@unsafe
def dbg_start_process():
"""Start the debugger, returns the current instruction pointer"""
if len(list_breakpoints()) == 0:
for i in range(ida_entry.get_entry_qty()):
ordinal = ida_entry.get_entry_ordinal(i)
address = ida_entry.get_entry(ordinal)
if address != ida_idaapi.BADADDR:
ida_dbg.add_bpt(address, 0, idaapi.BPT_SOFT)
if idaapi.start_process("", "", "") == 1:
ip = ida_dbg.get_ip_val()
if ip is not None:
return hex(ip)
raise IDAError("Failed to start debugger (did the user configure the debugger manually one time?)")
@jsonrpc
@idaread
@unsafe
def dbg_exit_process():
"""Exit the debugger"""
dbg_ensure_running()
if idaapi.exit_process():
return
raise IDAError("Failed to exit debugger")
@jsonrpc
@idaread
@unsafe
def dbg_continue_process() -> str:
"""Continue the debugger, returns the current instruction pointer"""
dbg_ensure_running()
if idaapi.continue_process():
ip = ida_dbg.get_ip_val()
if ip is not None:
return hex(ip)
raise IDAError("Failed to continue debugger")
@jsonrpc
@idaread
@unsafe
def dbg_run_to(
address: Annotated[str, "Run the debugger to the specified address"],
):
"""Run the debugger to the specified address"""
dbg_ensure_running()
ea = parse_address(address)
if idaapi.run_to(ea):
ip = ida_dbg.get_ip_val()
if ip is not None:
return hex(ip)
raise IDAError(f"Failed to run to address {hex(ea)}")
@jsonrpc
@idaread
@unsafe
def dbg_set_breakpoint(
address: Annotated[str, "Set a breakpoint at the specified address"],
):
"""Set a breakpoint at the specified address"""
ea = parse_address(address)
if idaapi.add_bpt(ea, 0, idaapi.BPT_SOFT):
return f"Breakpoint set at {hex(ea)}"
breakpoints = list_breakpoints()
for bpt in breakpoints:
if bpt["ea"] == hex(ea):
return
raise IDAError(f"Failed to set breakpoint at address {hex(ea)}")
@jsonrpc
@idaread
@unsafe
def dbg_step_into():
"""Step into the current instruction"""
dbg_ensure_running()
if idaapi.step_into():
ip = ida_dbg.get_ip_val()
if ip is not None:
return hex(ip)
raise IDAError("Failed to step into")
@jsonrpc
@idaread
@unsafe
def dbg_step_over():
"""Step over the current instruction"""
dbg_ensure_running()
if idaapi.step_over():
ip = ida_dbg.get_ip_val()
if ip is not None:
return hex(ip)
raise IDAError("Failed to step over")
@jsonrpc
@idaread
@unsafe
def dbg_delete_breakpoint(
address: Annotated[str, "del a breakpoint at the specified address"],
):
"""Delete a breakpoint at the specified address"""
ea = parse_address(address)
if idaapi.del_bpt(ea):
return
raise IDAError(f"Failed to delete breakpoint at address {hex(ea)}")
@jsonrpc
@idaread
@unsafe
def dbg_enable_breakpoint(
address: Annotated[str, "Enable or disable a breakpoint at the specified address"],
enable: Annotated[bool, "Enable or disable a breakpoint"],
):
"""Enable or disable a breakpoint at the specified address"""
ea = parse_address(address)
if idaapi.enable_bpt(ea, enable):
return
raise IDAError(f"Failed to {'' if enable else 'disable '}breakpoint at address {hex(ea)}")
class MCP(idaapi.plugin_t):
flags = idaapi.PLUGIN_KEEP
comment = "MCP Plugin"
help = "MCP"
wanted_name = "MCP"
wanted_hotkey = "Ctrl-Alt-M"
def init(self):
self.server = Server()
hotkey = MCP.wanted_hotkey.replace("-", "+")
if sys.platform == "darwin":
hotkey = hotkey.replace("Alt", "Option")
print(f"[MCP] Plugin loaded, use Edit -> Plugins -> MCP ({hotkey}) to start the server")
return idaapi.PLUGIN_KEEP
def run(self, arg):
self.server.start()
def term(self):
self.server.stop()
def PLUGIN_ENTRY():
return MCP()
```