GhidraMCP 搭建及核心源碼閱讀

前言

手動逆向太費勁了 想找個 AI 助手幫我逆向一下 逛 github 的時候看到了一個 GhidraMCP 於是打算拿來用用 所以本文記錄一下 GhidraMCP 搭建以及源碼解析

什麼是 MCP

什麼是 GhidraMCP

環境要求

環境配置

git clone https://github.com/LaurieWired/GhidraMCP.git
pip install -r .\requirements.txt

當然也可以直接編輯 json 文件

    "XXX": {
      "name": "GhidraMCP",
      "type": "stdio",
      "description": "",
      "isActive": true,
      "command": "python",
      "args": [
        "/ABSOLUTE_PATH_TO/bridge_mcp_ghidra.py"
      ]
    },

配置完成後即可啓用 GhidraMCP

同時還需選擇一個支持函數調用的模型 (帶扳手的就行)

  1. 運行 Ghidra

  2. 選擇 File -> Install Extensions

  3. 點擊 + 按鈕

  4. 選中 GhidraMCP-1-3.zip

  5. 重啓 Ghidra

  6. 確保在File -> Configure -> Developer中啓用了 GhidraMCPPlugin

  7. 可選: 在Edit -> Tool Options -> GhidraMCP HTTP Server中配置端口

GhidraMCP 實戰

核心源碼閱讀

以下是該插件執行流程

GhidraMCPPlugin.java

    private void startServer() throws IOException {
        // 讀取配置端口
        Options options = tool.getOptions(OPTION_CATEGORY_NAME);
        int port = options.getInt(PORT_OPTION_NAME, DEFAULT_PORT);
        // 如有實例則關閉現有實例
        if (server != null) {
            Msg.info(this, "Stopping existing HTTP server before starting new one.");
            server.stop(0);
            server = null;
        }
        // 服務器實例初始化
        server = HttpServer.create(new InetSocketAddress(port), 0);
server.createContext("/decompile", exchange -> { /* ... */ });
server.createContext("/renameFunction", exchange -> { /* ... */ });
server.createContext("/renameData", exchange -> { /* ... */ });
server.createContext("/renameVariable", exchange -> { /* ... */ });
server.createContext("/segments", exchange -> { /* ... */ });
server.createContext("/imports", exchange -> { /* ... */ });
server.createContext("/exports", exchange -> { /* ... */ });
server.createContext("/namespaces", exchange -> { /* ... */ });
server.createContext("/data", exchange -> { /* ... */ });
server.createContext("/searchFunctions", exchange -> { /* ... */ });
server.createContext("/get_function_by_address", exchange -> { /* ... */ });
server.createContext("/decompile_function", exchange -> { /* ... */ });
server.createContext("/disassemble_function", exchange -> { /* ... */ });
server.createContext("/set_decompiler_comment", exchange -> { /* ... */ });
server.createContext("/rename_function_by_address", exchange -> { /* ... */ });
server.createContext("/set_function_prototype", exchange -> {/* ... */});
server.createContext("/xrefs_to", exchange -> { /* ... */ });
server.createContext("/xrefs_from", exchange -> { /* ... */ });
        // 服務器線程管理：異步啓動並處理端口衝突
        server.setExecutor(null);
        new Thread(() -> {
            try {
                server.start();
                Msg.info(this, "GhidraMCP HTTP server started on port " + port);
            } catch (Exception e) {
                Msg.error(this, "Failed to start HTTP server on port " + port + ". Port might be in use.", e);
                server = null;
            }
        }, "GhidraMCP-HTTP-Server").start();
    }

bridge_mcp_ghidra.py

該文件執行流程

//初始化FastMCP實例
mcp = FastMCP("ghidra-mcp")
def safe_get(endpoint: str, params: dict = None) -> list:
    """
    Perform a GET request with optional query parameters.
    """
    if params isNone:
        params = {}

    url = urljoin(ghidra_server_url, endpoint)

    try:
        response = requests.get(url, params=params, timeout=5)
        response.encoding = 'utf-8'
        if response.ok:
            return response.text.splitlines()
        else:
            return [f"Error {response.status_code}: {response.text.strip()}"]
    except Exception as e:
        return [f"Request failed: {str(e)}"]

def safe_post(endpoint: str, data: dict | str) -> str:
    try:
        url = urljoin(ghidra_server_url, endpoint)
        if isinstance(data, dict):
            response = requests.post(url, data=data, timeout=5)
        else:
            response = requests.post(url, data=data.encode("utf-8"), timeout=5)
        response.encoding = 'utf-8'
        if response.ok:
            return response.text.strip()
        else:
            returnf"Error {response.status_code}: {response.text.strip()}"
    except Exception as e:
        returnf"Request failed: {str(e)}"
list_methods: 列出程序中的所有函數名。
list_classes: 列出程序中的所有類名。
decompile_function: 反編譯特定的函數並返回反編譯的 C 代碼。
rename_function: 通過當前函數名將其重命名爲用戶定義的新名。
rename_data: 重命名指定地址處的數據標籤。
list_segments: 列出程序中的所有內存段。
list_imports: 列出程序中的所有導入符號。
list_exports: 列出程序中的所有導出函數/符號。
list_namespaces: 列出程序中的所有非全局命名空間。
list_data_items: 列出程序中定義的所有數據標籤及其值。
search_functions_by_name: 搜索名稱包含給定子字符串的函數。
rename_variable: 重命名函數內的局部變量。
get_function_by_address: 通過地址獲取函數。
get_current_address: 獲取用戶當前選擇的地址。
get_current_function: 獲取用戶當前選擇的函數。
list_functions: 列出數據庫中的所有函數。
decompile_function_by_address: 反編譯給定地址處的函數。
disassemble_function: 獲取函數的彙編代碼（地址：指令；註釋）。
set_decompiler_comment: 在函數僞代碼中爲給定地址設置註釋。
set_disassembly_comment: 在函數反彙編中爲給定地址設置註釋。
rename_function_by_address: 通過地址重命名函數。
set_function_prototype: 設置函數的原型。
set_local_variable_type: 設置局部變量的類型。
get_xrefs_to: 獲取所有對指定地址的引用（xref to）。
get_xrefs_from: 獲取所有從指定地址的引用（xref from）。
get_function_xrefs: 獲取所有對指定函數的引用。
list_strings: 列出程序中定義的所有字符串及其地址。

就這樣吧:)

參考鏈接

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/Jj4g2szNRvMCE_XRoSvsnA

猜你喜歡