# Ghidra插件开发实战：用Python解锁逆向工程新维度 如果你已经习惯了在Ghidra里点点菜单、看看反编译代码，可能会觉得这个工具虽然强大，但有些重复性工作还是太繁琐。我自己在分析大型固件时，经常需要批量重命名函数、提取特定字符串模式，或者自动标注某些API调用——每次手动操作不仅耗时，还容易出错。直到我开始研究Ghidra的插件开发，才发现原来这个开源逆向工具的真正潜力藏在它的扩展API里。 Ghidra的Python插件开发能力，本质上是一种“工具定制化”的思维方式转变。它让你从被动的工具使用者，变成主动的流程设计者。想象一下，当你面对一个包含上千个函数的二进制文件，能够用几十行Python代码自动完成分类标记；或者当你需要定期分析同一家族的恶意软件时，能够一键运行预设的分析流水线——这种效率提升不是简单的快捷键操作能比拟的。 这篇文章面向的是已经熟悉Ghidra基本操作，但希望进一步提升自动化能力的逆向工程师、安全研究员和软件开发者。无论你是想简化日常分析流程，还是构建团队内部的标准分析工具，Python插件开发都能提供一条相对平缓的学习曲线。与Java插件开发相比，Python版本更轻量、更快速原型，特别适合那些不想深入JVM生态但需要实用自动化脚本的从业者。 我会从最实际的环境搭建讲起，逐步深入到API的核心用法，最后通过几个我实际用过的案例，展示如何将想法转化为可运行的插件。你会发现，给Ghidra“编程”并不需要多高深的计算机科学知识，更多是一种对逆向工作流程的重新思考。 ## 1. 开发环境搭建与基础配置 开始编写Ghidra插件前，你需要一个能正常工作的开发环境。很多人卡在这一步，不是因为技术复杂，而是因为Ghidra的文档比较分散。我建议完全避开Eclipse或IntelliJ那些重型IDE的配置教程——对于Python插件开发，轻量级编辑器配合正确的环境变量就足够了。 ### 1.1 Python环境与Ghidra版本匹配 首先确认你的Ghidra版本。从Ghidra 9.2开始，官方正式支持Python 3，但不同版本对Python的具体要求有差异。我目前使用的是Ghidra 10.3，它内置了Jython 2.7和Python 3.9的桥接支持。你可以通过以下方式检查： ```bash # 进入Ghidra安装目录 cd /path/to/ghidra ./support/analyzeHeadless -help | grep -i python ``` 如果输出显示Python相关选项，说明你的版本支持Python插件开发。一个常见的误区是认为需要单独安装Python——实际上Ghidra通过JPype或Jython与Python解释器通信，你系统上的Python环境只是被“桥接”进来。 > 注意：如果你在Windows上开发，确保Python已添加到系统PATH环境变量。在Linux/macOS上，Ghidra通常会使用`/usr/bin/python3`，但你可以通过环境变量指定自定义路径。 我推荐创建一个专门的Python虚拟环境用于Ghidra开发，避免与系统Python包冲突： ```bash # 创建虚拟环境 python3 -m venv ~/ghidra_dev # 激活环境 source ~/ghidra_dev/bin/activate # 安装基础工具包 pip install ipython jupyter ``` 虚拟环境不是必须的，但当你需要安装第三方Python库（如`requests`用于网络操作，`pandas`用于数据分析）时，它能保持环境干净。 ### 1.2 插件目录结构与项目初始化 Ghidra的插件可以放在几个特定位置。对于个人开发，最方便的是用户目录下的`.ghidra/.ghidra_版本/Extensions`文件夹。例如： ``` ~/.ghidra/.ghidra_10.3_PUBLIC/Extensions/ └── MyFirstPlugin/ ├── Module.manifest ├── data/ ├── lib/ ├── src/ │ └── main.py └── support/ ``` `Module.manifest`是插件的配置文件，采用简单的属性格式： ```properties # Module.manifest MODULE_NAME=MyFirstPlugin MODULE_VERSION=1.0 MODULE_AUTHOR=YourName MODULE_DESCRIPTION=A simple Python plugin for Ghidra MODULE_SOURCE=src ``` 更简单的方法是使用Ghidra自带的脚本管理器作为起点。打开Ghidra，进入`Window -> Script Manager`，点击`New`按钮选择Python，Ghidra会自动生成一个模板脚本。将这个脚本保存到上述Extensions目录的src文件夹中，就完成了最基本的插件结构。 不过，对于更复杂的插件，我建议采用以下目录组织方式： ``` CustomAnalyzerPlugin/ ├── Module.manifest ├── src/ │ ├── __init__.py │ ├── main_plugin.py # 插件主入口 │ ├── analyzers/ # 自定义分析器 │ │ ├── __init__.py │ │ └── string_analyzer.py │ ├── ui/ # 用户界面组件 │ │ ├── __init__.py │ │ └── custom_dialog.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── logging_utils.py ├── data/ # 资源文件 │ └── config.json ├── lib/ # 第三方JAR或Python包 └── docs/ # 文档 ``` 这种结构虽然看起来复杂，但当你需要添加新功能时，模块化的设计能让代码保持清晰。`__init__.py`文件可以是空文件，它们只是标记Python包目录。 ### 1.3 开发工作流与调试技巧 开发Ghidra插件最痛苦的部分是调试。由于插件运行在Ghidra的JVM中，你不能简单地用`print()`语句输出到控制台（除非你从命令行启动Ghidra）。我常用的几种调试方法： **方法一：使用Ghidra的日志系统** ```python import logging from ghidra.util import Msg # 创建插件专用日志器 logger = logging.getLogger("MyPlugin") logger.setLevel(logging.DEBUG) # 输出到Ghidra的消息窗口 def log_debug(message): Msg.info(logger, f"[DEBUG] {message}") def log_error(message): Msg.error(logger, f"[ERROR] {message}") ``` **方法二：文件日志输出** ```python import logging from logging.handlers import RotatingFileHandler def setup_file_logging(): logger = logging.getLogger("MyPluginFile") handler = RotatingFileHandler( "/tmp/ghidra_plugin.log", maxBytes=1024*1024, # 1MB backupCount=5 ) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) handler.setFormatter(formatter) logger.addHandler(handler) return logger ``` **方法三：交互式调试（高级）** 对于复杂问题，我有时会启动一个简单的TCP服务器，在插件中发送调试信息： ```python import socket import json class DebugSocket: def __init__(self, host='127.0.0.1', port=9999): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.sock.connect((host, port)) except ConnectionRefusedError: print("Debug server not running") def send(self, data): if isinstance(data, dict): data = json.dumps(data) self.sock.sendall(data.encode() + b'\n') def close(self): self.sock.close() # 在另一终端运行：nc -l 9999 ``` 开发工作流上，我推荐这样的步骤： 1. 在外部编辑器（VS Code、Sublime等）编写代码 2. 将插件目录符号链接到Ghidra的Extensions文件夹 3. 在Ghidra中通过`File -> Configure -> Experimental`启用插件 4. 使用`Window -> Script Manager`重新加载脚本（快捷键`F5`） 5. 测试功能，查看日志输出 > 提示：Ghidra 10.1+支持热重载Python脚本，但修改`Module.manifest`或添加新文件后，需要重启Ghidra。 ## 2. Ghidra Python API核心概念解析 Ghidra的Python API本质上是Java API的包装器，通过JPype实现Python到Java的调用。这意味着你实际上是在用Python语法操作Java对象。理解这一点很重要，因为某些Java特有的模式（如接口、抽象类）在Python中表现方式不同。 ### 2.1 程序（Program）与域（Domain）对象模型 所有Ghidra分析都围绕`Program`对象展开。这个对象代表了加载到Ghidra中的整个二进制文件，包含代码、数据、符号、引用等所有信息。获取当前活动程序的方法： ```python from ghidra.app.decompiler import DecompInterface from ghidra.program.model.listing import Program # 获取当前程序 currentProgram = state.getProject().getCurrentProgram() # 或者通过工具对象获取 tool = state.getTool() if tool: program = tool.getService(ghidra.app.plugin.core.analysis.AnalysisService).getCurrentProgram() ``` `Program`对象有几个关键子组件： - **Listing（列表）**：程序的线性视图，按地址顺序显示所有代码和数据单元 - **SymbolTable（符号表）**：存储函数名、变量名、标签等符号信息 - **ReferenceManager（引用管理器）**：跟踪代码和数据之间的交叉引用 - **FunctionManager（函数管理器）**：管理程序中所有函数定义 - **DataTypeManager（数据类型管理器）**：处理结构体、联合体、枚举等自定义类型 一个实用的例子是遍历程序中的所有函数： ```python from ghidra.program.model.listing import FunctionManager def iterate_functions(program): """遍历程序中的所有函数并统计信息""" func_manager = program.getFunctionManager() functions = func_manager.getFunctions(True) # True表示向前迭代 stats = { 'total': 0, 'named': 0, 'library': 0, 'user': 0 } for func in functions: stats['total'] += 1 # 检查函数是否有自定义名称 if not func.isThunk() and not func.isExternal(): if func.getName().startswith('FUN_') or func.getName().startswith('sub_'): stats['user'] += 1 else: stats['named'] += 1 else: stats['library'] += 1 # 获取函数的反编译代码 decompiler = DecompInterface() decompiler.openProgram(program) results = decompiler.decompileFunction(func, 30, None) if results.decompileCompleted(): decompiled_code = results.getDecompiledFunction().getC() # 分析反编译代码... return stats ``` ### 2.2 地址空间与内存访问 Ghidra使用`Address`对象表示内存地址，而不是简单的整数。这是因为二进制文件可能有多个地址空间（如RAM、ROM、IO空间）。理解地址空间对编写正确的插件至关重要： ```python from ghidra.program.model.address import AddressSpace def address_operations_example(program): """地址操作示例""" addr_factory = program.getAddressFactory() # 获取默认地址空间（通常是RAM） default_space = addr_factory.getDefaultAddressSpace() # 创建地址对象 addr1 = addr_factory.getAddress("0x00401000") addr2 = default_space.getAddress(0x401000) # 相同地址，不同创建方式 # 地址算术运算 next_addr = addr1.add(4) # 加4字节 prev_addr = addr1.subtract(1) # 减1字节 # 比较地址 if addr1.compareTo(addr2) == 0: print("地址相同") # 遍历地址范围 start_addr = addr_factory.getAddress("0x00400000") end_addr = addr_factory.getAddress("0x00401000") addr_range = start_addr.getAddressSpace().getAddressSet(start_addr, end_addr) # 获取地址所在的内存块 memory = program.getMemory() block = memory.getBlock(start_addr) if block: print(f"块名称: {block.getName()}") print(f"块权限: 读={block.isRead()}, 写={block.isWrite()}, 执行={block.isExecute()}") return next_addr, prev_addr ``` 内存访问时需要注意权限检查。尝试读取不可读或写入不可写的地址会导致异常： ```python def safe_memory_read(program, address, size): """安全的内存读取函数""" memory = program.getMemory() # 检查地址是否有效且可读 if not memory.contains(address): raise ValueError(f"地址 {address} 不在内存映射中") block = memory.getBlock(address) if not block or not block.isRead(): raise PermissionError(f"地址 {address} 不可读") # 读取数据 data = bytearray(size) bytes_read = memory.getBytes(address, data) if bytes_read != size: print(f"警告: 请求 {size} 字节，但只读取到 {bytes_read} 字节") return bytes(data[:bytes_read]) if bytes_read > 0 else b'' ``` ### 2.3 数据类型与结构体处理 Ghidra的数据类型系统非常强大，但API有些复杂。以下表格对比了常见数据类型的创建和使用方法： | 数据类型 | 创建方法 | 示例代码 | 主要用途 | |---------|---------|---------|---------| | 基本类型 | `program.getDataTypeManager().getBuiltInDataTypes()` | `ByteDataType(), WordDataType()` | 表示字节、字、双字等 | | 指针类型 | `PointerDataType(base_type)` | `PointerDataType(ByteDataType())` | 内存地址引用 | | 数组类型 | `ArrayDataType(element_type, count)` | `ArrayDataType(CharDataType(), 256)` | 缓冲区、字符串数组 | | 结构体 | `StructureDataType(name, size)` | `struct = StructureDataType("MY_STRUCT", 0)` | C结构体表示 | | 联合体 | `UnionDataType(name)` | `union = UnionDataType("MY_UNION")` | 重叠内存布局 | | 枚举 | `EnumDataType(name, size)` | `enum = EnumDataType("STATUS", 4)` | 命名常量集合 | | 类型定义 | `TypedefDataType(name, base_type)` | `TypedefDataType("DWORD", DWordDataType())` | 类型别名 | 处理结构体的一个完整示例： ```python from ghidra.program.model.data import StructureDataType, DataTypeComponent from ghidra.program.model.symbol import SourceType def create_and_apply_structure(program, address, struct_name="MyStruct"): """创建并应用结构体到指定地址""" dtm = program.getDataTypeManager() listing = program.getListing() # 创建新结构体 struct = StructureDataType(struct_name, 0) # 添加字段 struct.add(ByteDataType(), "field1", "第一个字节字段") struct.add(WordDataType(), "field2", "两字节字段") struct.add(DWordDataType(), "field3", "四字节字段") struct.add(PointerDataType(ByteDataType()), "ptr_field", "指向字节的指针") # 计算结构体大小 struct_size = struct.getLength() print(f"结构体 {struct_name} 大小: {struct_size} 字节") # 将结构体应用到当前地址 data = listing.getDataAt(address) if data: # 如果该地址已有数据，尝试替换 try: listing.clearCodeUnits(address, address.add(struct_size-1), False) except Exception as e: print(f"清除原有数据失败: {e}") # 创建结构体实例 created = listing.createData(address, struct) if created: # 为字段添加注释 for i in range(struct.getNumComponents()): component = struct.getComponent(i) field_addr = address.add(component.getOffset()) listing.setComment(field_addr, ghidra.program.model.listing.CodeUnit.EOL_COMMENT, f"{component.getFieldName()}: {component.getDataType().getName()}") # 标记为已分析 program.getBookmarkManager().setBookmark( address, "Structure", f"Applied {struct_name}", f"Structure of size {struct_size}" ) return True return False ``` ### 2.4 反编译器接口深度使用 Ghidra的反编译器是它的核心功能之一，通过Python API可以直接访问反编译结果： ```python from ghidra.app.decompiler import DecompInterface, DecompileOptions from ghidra.util.task import ConsoleTaskMonitor class DecompilerAnalyzer: """反编译器封装类，提供高级分析功能""" def __init__(self, program): self.program = program self.decompiler = DecompInterface() # 配置反编译器选项 options = DecompileOptions() options.setMaxPayloadMBytes(2048) # 内存限制 options.setTimeout(60) # 超时时间（秒） self.decompiler.setOptions(options) self.decompiler.openProgram(program) def decompile_function(self, function, simplify=True): """反编译单个函数""" monitor = ConsoleTaskMonitor() # 执行反编译 results = self.decompiler.decompileFunction( function, 60, # 超时时间 monitor ) if not results.decompileCompleted(): print(f"反编译失败: {results.getErrorMessage()}") return None decompiled_func = results.getDecompiledFunction() if simplify: # 获取简化后的C代码 return decompiled_func.getC() else: # 获取完整的反编译中间表示 return { 'c_code': decompiled_func.getC(), 'high_function': decompiled_func.getHighFunction(), 'symbol_map': decompiled_func.getSymbolMap() } def analyze_function_complexity(self, function): """分析函数复杂度""" results = self.decompiler.decompileFunction(function, 30, None) if not results.decompileCompleted(): return None high_func = results.getDecompiledFunction().getHighFunction() if not high_func: return None # 获取P-code（低级中间表示） pcode_ops = [] iterator = high_func.getPcodeOps() while iterator.hasNext(): pcode_ops.append(iterator.next()) # 计算基本指标 metrics = { 'function_name': function.getName(), 'entry_point': str(function.getEntryPoint()), 'body_size': function.getBody().getNumAddresses(), 'pcode_operations': len(pcode_ops), 'local_vars': high_func.getLocalSymbolMap().getNumSymbols(), 'c_code_length': len(results.getDecompiledFunction().getC()) } # 分析控制流复杂度 basic_blocks = 0 edges = 0 # 这里简化处理，实际需要遍历基本块 # ... metrics['estimated_cyclomatic'] = edges - basic_blocks + 2 return metrics def find_pattern_in_decompiled(self, pattern, case_sensitive=False): """在反编译代码中搜索模式""" matches = [] func_manager = self.program.getFunctionManager() for func in func_manager.getFunctions(True): if func.isThunk() or func.isExternal(): continue code = self.decompiler.decompileFunction(func, 30, None) if not code.decompileCompleted(): continue c_code = code.getDecompiledFunction().getC() search_text = c_code if case_sensitive else c_code.lower() search_pattern = pattern if case_sensitive else pattern.lower() if search_pattern in search_text: matches.append({ 'function': func.getName(), 'address': str(func.getEntryPoint()), 'snippet': self._extract_context(c_code, pattern, 3) }) return matches def _extract_context(self, text, pattern, context_lines=3): """提取匹配模式的上下文""" lines = text.split('\n') for i, line in enumerate(lines): if pattern in line: start = max(0, i - context_lines) end = min(len(lines), i + context_lines + 1) return '\n'.join(lines[start:end]) return "" ``` ## 3. 实用插件开发模式与架构设计 开发Ghidra插件不仅仅是写几行脚本，更需要考虑代码的可维护性、可扩展性和用户体验。经过多个项目的实践，我总结了几种有效的插件架构模式。 ### 3.1 分析器（Analyzer）插件模式 分析器是Ghidra中最常见的插件类型，在二进制文件导入或用户请求时自动运行。创建自定义分析器可以自动化许多重复性分析任务。 **基础分析器模板：** ```python from ghidra.program.util import ProgramLocation from ghidra.program.model.listing import Program from ghidra.program.model.address import AddressSetView from ghidra.util.task import TaskMonitor from ghidra.app.services import AnalyzerType from ghidra.app.plugin.core.analysis import AbstractAnalyzer class StringReferenceAnalyzer(AbstractAnalyzer): """字符串引用分析器：识别并标记代码中的字符串引用""" def __init__(self): super(StringReferenceAnalyzer, self).__init__( "字符串引用分析器", "识别代码中对字符串的引用并添加交叉引用", AnalyzerType.BYTE_ANALYZER ) # 分析器属性设置 self.setPriority(AnalyzerType.BYTE_ANALYZER.priorityAfter()) self.setSupportsOneTimeAnalysis(True) self.setDefaultEnablement(True) def canAnalyze(self, program): """检查是否支持分析此程序""" # 这里可以添加架构或文件格式检查 return True def added(self, program, setView, monitor, messageLog): """分析器被添加到程序时调用""" messageLog.setStatus("字符串引用分析器已加载") return True def removed(self, program): """分析器被移除时调用""" pass def analyze(self, program, setView, monitor, messageLog): """执行分析的主要方法""" try: # 设置进度监视器 monitor.setMaximum(setView.getNumAddresses()) monitor.setProgress(0) monitor.setMessage("正在分析字符串引用...") # 获取字符串管理器 string_manager = program.getListing().getDefinedStrings() if not string_manager: messageLog.appendMsg("未找到字符串数据") return True # 遍历所有字符串 strings = list(string_manager) monitor.setMaximum(len(strings)) for i, string_data in enumerate(strings): if monitor.isCancelled(): break monitor.setProgress(i) monitor.setMessage(f"处理字符串 {i+1}/{len(strings)}") self._process_string(program, string_data, monitor) messageLog.appendMsg(f"完成分析，处理了 {len(strings)} 个字符串") return True except Exception as e: messageLog.appendException(e) return False def _process_string(self, program, string_data, monitor): """处理单个字符串""" string_addr = string_data.getAddress() string_value = string_data.getData().getValue() # 查找对此字符串的引用 ref_manager = program.getReferenceManager() refs_to = ref_manager.getReferencesTo(string_addr) for ref in refs_to: from_addr = ref.getFromAddress() # 添加书签标记 program.getBookmarkManager().setBookmark( from_addr, "StringRef", f"引用字符串: {string_value[:50]}...", f"在 {string_addr} 的字符串引用" ) # 添加注释 listing = program.getListing() code_unit = listing.getCodeUnitAt(from_addr) if code_unit: existing_comment = code_unit.getComment( ghidra.program.model.listing.CodeUnit.EOL_COMMENT ) new_comment = f"引用字符串: {string_value[:30]}..." if existing_comment: new_comment = existing_comment + "; " + new_comment listing.setComment( from_addr, ghidra.program.model.listing.CodeUnit.EOL_COMMENT, new_comment ) ``` **分析器配置与注册：** 为了让分析器出现在Ghidra的分析器列表中，需要创建扩展点描述文件： ```xml <!-- 在插件目录的data/analyzer目录下创建 --> <!-- StringReferenceAnalyzer.analyzer --> <ANALYZER> <NAME>字符串引用分析器</NAME> <DESCRIPTION>自动识别并标记代码中的字符串引用</DESCRIPTION> <CLASS>string_analyzer.StringReferenceAnalyzer</CLASS> <TYPE>BYTE_ANALYZER</TYPE> <PRIORITY>100</PRIORITY> <DEFAULT_ENABLED>true</DEFAULT_ENABLED> </ANALYZER> ``` ### 3.2 提供器（Provider）与用户界面集成 对于需要交互的插件，创建自定义提供器（Provider）是更好的选择。提供器可以集成到Ghidra的主界面，提供专门的视图。 **表格数据提供器示例：** ```python from ghidra.app.plugin.core.table import TableComponentProvider from ghidra.program.model.listing import Program from ghidra.util.table import GhidraTableModel from docking.widgets.table import GTable import java.awt.BorderLayout as BorderLayout import javax.swing.JPanel as JPanel import javax.swing.JScrollPane as JScrollPane from java.awt.event import MouseAdapter class FunctionTableProvider(TableComponentProvider): """函数信息表格提供器""" def __init__(self, plugin, program): super(FunctionTableProvider, self).__init__( "函数分析表", plugin.getName(), None ) self.plugin = plugin self.program = program self.table_model = FunctionTableModel(program) # 创建表格 self.table = GTable(self.table_model) self.table.setAutoCreateRowSorter(True) # 设置列宽 self.table.getColumnModel().getColumn(0).setPreferredWidth(150) # 函数名 self.table.getColumnModel().getColumn(1).setPreferredWidth(100) # 地址 self.table.getColumnModel().getColumn(2).setPreferredWidth(80) # 大小 self.table.getColumnModel().getColumn(3).setPreferredWidth(120) # 调用次数 # 添加鼠标双击事件 self.table.addMouseListener(self._create_mouse_listener()) # 创建面板 panel = JPanel(BorderLayout()) scroll_pane = JScrollPane(self.table) panel.add(scroll_pane, BorderLayout.CENTER) # 设置提供器组件 self.setComponent(panel) # 添加工具栏操作 self._setup_toolbar() def _create_mouse_listener(self): """创建表格鼠标监听器""" class TableMouseListener(MouseAdapter): def __init__(self, outer): self.outer = outer def mouseClicked(self, event): if event.getClickCount() == 2: # 双击跳转到函数 row = self.outer.table.getSelectedRow() if row >= 0: model_row = self.outer.table.convertRowIndexToModel(row) address = self.outer.table_model.getAddressAt(model_row) if address: self.outer.plugin.goTo(address) return TableMouseListener(self) def _setup_toolbar(self): """设置工具栏按钮""" from docking.ActionContext import ActionContext from docking.action import DockingAction # 刷新按钮 refresh_action = DockingAction("刷新", self.plugin.getName()): def actionPerformed(self, context): self.outer.table_model.reload() refresh_action.setToolBarData( DockingAction.ToolBarData( ghidra.util.bean.GIcon("images/refresh.png"), "工具栏" ) ) self.addLocalAction(refresh_action) # 导出按钮 export_action = DockingAction("导出CSV", self.plugin.getName()): def actionPerformed(self, context): self.outer._export_to_csv() export_action.setToolBarData( DockingAction.ToolBarData( ghidra.util.bean.GIcon("images/export.png"), "工具栏" ) ) self.addLocalAction(export_action) def _export_to_csv(self): """导出表格数据到CSV""" import csv import os from ghidra.util import Msg try: # 选择保存路径 chooser = ghidra.util.bean.GFileChooser("导出CSV") chooser.setSelectedFile( java.io.File(f"{self.program.getName()}_functions.csv") ) if chooser.showSaveDialog(None) != javax.swing.JFileChooser.APPROVE_OPTION: return output_file = chooser.getSelectedFile() with open(str(output_file), 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) # 写入表头 headers = [] for col in range(self.table_model.getColumnCount()): headers.append(self.table_model.getColumnName(col)) writer.writerow(headers) # 写入数据 for row in range(self.table_model.getRowCount()): row_data = [] for col in range(self.table_model.getColumnCount()): value = self.table_model.getValueAt(row, col) row_data.append(str(value) if value else "") writer.writerow(row_data) Msg.info(self, f"数据已导出到: {output_file}") except Exception as e: Msg.error(self, f"导出失败: {e}") def reload(self): """重新加载数据""" self.table_model.reload() self.table.repaint() class FunctionTableModel(GhidraTableModel): """函数表格数据模型""" def __init__(self, program): super(FunctionTableModel, self).__init__("函数分析") self.program = program self.functions = [] self._reload_data() def _reload_data(self): """重新加载函数数据""" self.functions = [] func_manager = self.program.getFunctionManager() for func in func_manager.getFunctions(True): if func.isExternal(): continue # 收集函数信息 func_info = { 'name': func.getName(), 'address': func.getEntryPoint(), 'size': func.getBody().getNumAddresses(), 'call_count': len(list(func.getCallingFunctions())), 'called_count': len(list(func.getCalledFunctions(None))), 'has_comment': func.getComment() != "", 'is_library': func.isLibraryFunction() } self.functions.append(func_info) def reload(self): """公开的重新加载方法""" old_size = len(self.functions) self._reload_data() self.fireTableDataChanged() return len(self.functions) - old_size def getRowCount(self): return len(self.functions) def getColumnCount(self): return 7 # 根据实际列数调整 def getColumnName(self, column): columns = [ "函数名", "地址", "大小", "被调用", "调用", "有注释", "库函数" ] return columns[column] if column < len(columns) else "" def getValueAt(self, row, column): if row >= len(self.functions): return None func = self.functions[row] if column == 0: return func['name'] elif column == 1: return str(func['address']) elif column == 2: return func['size'] elif column == 3: return func['call_count'] elif column == 4: return func['called_count'] elif column == 5: return "是" if func['has_comment'] else "否" elif column == 6: return "是" if func['is_library'] else "否" return None def getAddressAt(self, row): if row < len(self.functions): return self.functions[row]['address'] return None ``` ### 3.3 服务（Service）插件架构 对于需要在多个插件间共享功能的情况，服务模式是最佳选择。服务是单例对象，可以在整个Ghidra实例中访问。 **自定义分析服务示例：** ```python from ghidra.framework.plugintool import PluginTool from ghidra.framework.plugintool.util import PluginException from ghidra.util import Msg import threading import time class AnalysisService: """自定义分析服务，提供复杂的分析功能""" _instance = None _lock = threading.Lock() @classmethod def get_instance(cls, tool=None): """获取服务实例（单例模式）""" with cls._lock: if cls._instance is None: if tool is None: raise ValueError("首次调用需要提供PluginTool实例") cls._instance = cls(tool) return cls._instance def __init__(self, tool): self.tool = tool self.program = None self.analyzers = {} self._init_analyzers() def _init_analyzers(self): """初始化分析器集合""" self.analyzers = { 'string_pattern': StringPatternAnalyzer(), 'api_tracker': APITrackerAnalyzer(), 'vulnerability': VulnerabilityDetector(), 'code_similarity': CodeSimilarityAnalyzer() } def set_current_program(self, program): """设置当前分析的程序""" self.program = program for analyzer in self.analyzers.values(): analyzer.set_program(program) def run_analysis(self, analyzer_name, options=None): """运行指定的分析器""" if analyzer_name not in self.analyzers: raise ValueError(f"未知的分析器: {analyzer_name}") if self.program is None: raise RuntimeError("未设置当前程序") analyzer = self.analyzers[analyzer_name] return analyzer.analyze(options or {}) def run_batch_analysis(self, analyzer_names, progress_callback=None): """批量运行多个分析器""" results = {} for i, name in enumerate(analyzer_names): if progress_callback: progress_callback(i, len(analyzer_names), f"运行 {name}") try: results[name] = self.run_analysis(name) except Exception as e: Msg.error(self, f"分析器 {name} 执行失败: {e}") results[name] = {'error': str(e)} # 短暂暂停，避免UI卡顿 time.sleep(0.1) return results def register_custom_analyzer(self, name, analyzer): """注册自定义分析器""" if not hasattr(analyzer, 'analyze'): raise TypeError("分析器必须实现analyze方法") self.analyzers[name] = analyzer if self.program: analyzer.set_program(self.program) Msg.info(self, f"已注册分析器: {name}") def get_available_analyzers(self): """获取可用的分析器列表""" return list(self.analyzers.keys()) class StringPatternAnalyzer: """字符串模式分析器""" def __init__(self): self.program = None self.patterns = [ (r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', 'URL'), (r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}', 'Email'), (r'\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}', 'IP地址'), (r'[0-9a-fA-F]{32}', 'MD5哈希'), (r'[0-9a-fA-F]{40}', 'SHA-1哈希'), (r'[0-9a-fA-F]{64}', 'SHA-256哈希'), ] def set_program(self, program): self.program = program def analyze(self, options): import re results = { 'patterns_found': [], 'statistics': {} } string_manager = self.program.getListing().getDefinedStrings() if not string_manager: return results for string_data in string_manager: string_value = str(string_data.getData().getValue()) for pattern, pattern_type in self.patterns: matches = re.findall(pattern, string_value) if matches: for match in matches: results['patterns_found'].append({ 'address': str(string_data.getAddress()), 'type': pattern_type, 'value': match, 'context': string_value[:100] + ('...' if len(string_value) > 100 else '') }) # 统计 type_count = {} for item in results['patterns_found']: type_count[item['type']] = type_count.get(item['type'], 0) + 1 results['statistics'] = { 'total_patterns': len(results['patterns_found']), 'by_type': type_count } return results # 在插件中注册服务 class MyPlugin(ghidra.app.plugin.ProgramPlugin): def __init__(self, tool): super(MyPlugin, self).__init__(tool, True, True) # 注册服务 self.analysis_service = AnalysisService.get_instance(tool) # 注册到工具的服务管理器 tool.addService(AnalysisService, self.analysis_service) # 创建UI组件 self._create_actions() def _create_actions(self): """创建插件动作""" from docking.action import DockingAction from docking.ActionContext import ActionContext # 分析动作 analyze_action = DockingAction("运行自定义分析", self.getName()): def actionPerformed(self, context): self._run_custom_analysis() analyze_action.setMenuBarData( DockingAction.MenuBarData( ["分析", "自定义分析"], "CustomAnalysis", None ) ) self.addAction(analyze_action) def _run_custom_analysis(self): """运行自定义分析""" # 获取当前程序 program = self.getCurrentProgram() if not program: Msg.showWarn(self, None, "错误", "请先打开一个程序") return # 设置当前程序 self.analysis_service.set_current_program(program) # 运行分析 results = self.analysis_service.run_batch_analysis( ['string_pattern', 'api_tracker'], self._update_progress ) # 显示结果 self._show_results(results) def _update_progress(self, current, total, message): """更新进度回调""" # 这里可以更新进度条或状态栏 pass def _show_results(self, results): """显示分析结果""" # 创建结果显示对话框 pass ``` ## 4. 实战案例：从需求到完整插件的开发过程 理论讲得再多，不如实际动手做一个插件。我以开发一个"函数调用图生成器"插件为例，展示完整的开发流程。这个插件需要从二进制文件中提取函数调用关系，生成可视化的调用图，并支持导出为多种格式。 ### 4.1 需求分析与设计 首先明确插件的核心需求： 1. **功能需求**： - 自动分析程序中的函数调用关系 - 生成可视化的调用图 - 支持过滤和搜索功能 - 导出为图片、DOT格式或JSON格式 - 支持交互式探索 2. **技术需求**： - 使用Ghidra的API获取函数和调用信息 - 集成图形库进行可视化 - 提供响应式用户界面 - 支持大型程序的高效处理 3. **用户体验需求**： - 进度反馈，避免界面卡死 - 直观的图形操作界面 - 可定制的显示选项 - 结果可重复生成 基于这些需求，我设计了以下架构： ``` CallGraphPlugin/ ├── core/ │ ├── graph_builder.py # 调用图构建器 │ ├── filters.py # 过滤逻辑 │ └── exporters.py # 导出器 ├── ui/ │ ├── graph_panel.py # 图形显示面板 │ ├── control_panel.py # 控制面板 │ └── progress_dialog.py # 进度对话框 ├── utils/ │ ├── algorithms.py # 图算法 │ └── performance.py # 性能优化 └── main_plugin.py # 插件主入口 ``` ### 4.2 核心图构建器实现 调用图构建器是插件的核心，需要高效处理可能包含数千个函数的程序： ```python import networkx as nx from collections import defaultdict from ghidra.program.model.listing import FunctionManager from ghidra.program.model.symbol import ReferenceManager from ghidra.util.task import TaskMonitor import time class CallGraphBuilder: """函数调用图构建器""" def __init__(self, program): self.program = program self.graph = nx.DiGraph() # 有向图 self.function_info = {} # 函数详细信息 self.metrics = {} # 构建指标 def build(self, monitor=None, options=None): """构建调用图""" options = options or {} start_time = time.time() # 初始化监视器 if monitor: monitor.setMessage("正在构建调用图...") monitor.setIndeterminate(False) # 获取所有函数 func_manager = self.program.getFunctionManager() all_functions = list(func_manager.getFunctions(True)) if monitor: monitor.setMaximum(len(all_functions)) # 第一遍：添加所有节点 for i, func in enumerate(all_functions): if monitor and monitor.isCancelled(): return False if monitor: monitor.setProgress(i) monitor.setMessage(f"添加函数节点: {func.getName()}") self._add_function_node(func) # 第二遍：添加边（调用关系） edge_count = 0 for i, func in enumerate(all_functions): if monitor and monitor.isCancelled(): return False if monitor: monitor.setProgress(i) monitor.setMessage(f"分析调用关系: {func.getName()}") edges_added = self._add_function_edges(func) edge_count += edges_added # 计算图指标 self._calculate_metrics(start_time, len(all_functions), edge_count) return True def _add_function_node(self, function): """添加函数节点到图中""" func_addr = str(function.getEntryPoint()) func_name = function.getName() # 收集函数信息 info = { 'address': func_addr, 'name': func_name, 'size': function.getBody().getNumAddresses(), 'is_library': function.isLibraryFunction(), 'is_external': function.isExternal(), 'is_thunk': function.isThunk(), 'comment': function.getComment() or "", 'calling_functions': [], 'called_functions': [] } # 添加到图 self.graph.add_node(func_addr, **info) self.function_info[func_addr] = info return info def _add_function_edges(self, function): """添加函数调用边""" func_addr = str(function.getEntryPoint()) edges_added = 0 # 获取函数体中的所有引用 ref_manager = self.program.getReferenceManager() body = function.getBody() for addr in body.getAddresses(True): # 获取从该地址出发的引用 refs_from = ref_manager.getReferencesFrom(addr) for ref in refs_from: to_addr = ref.getToAddress() # 检查引用目标是否是函数 to_func = self.program.getFunctionManager().getFunctionAt(to_addr) if to_func: to_func_addr = str(to_func.getEntryPoint()) # 添加边 if not self.graph.has_edge(func_addr, to_func_addr): self.graph.add_edge(func_addr, to_func_addr) # 更新节点信息 if to_func_addr not in self.function_info[func_addr]['called_functions']: self.function_info[func_addr]['called_functions'].append(to_func_addr) if func_addr not in self.function_info[to_func_addr]['calling_functions']: self.function_info[to_func_addr]['calling_functions'].append(func_addr) edges_added += 1 return edges_added def _calculate_metrics(self, start_time, node_count, edge_count): """计算构建指标""" end_time = time.time() self.metrics = { 'build_time': end_time - start_time, 'nodes': node_count, 'edges': edge_count, 'density': edge_count / (node_count * (node_count - 1)) if node_count > 1 else 0, 'connected_components': nx.number_weakly_connected_components(self.graph), 'is_dag': nx.is_directed_acyclic_graph(self.graph), 'max_in_degree': max(dict(self.graph.in_degree()).values()) if node_count > 0 else 0, 'max_out_degree': max(dict(self.graph.out_degree()).values()) if node_count > 0 else 0 } def get_subgraph(self, center_func_addr, depth=2): """获取以指定函数为中心的指定深度的子图""" if center_func_addr not in self.graph: return None # 使用BFS获取指定深度内的节点 visited = {center_func_addr} queue = [(center_func_addr, 0)] while queue: current, current_depth = queue.pop(0) if current_depth >= depth: continue # 前向（调用的函数） for neighbor in self.graph.successors(current): if neighbor not in visited: visited.add(neighbor) queue.append((neighbor, current_depth + 1)) # 后向（调用此函数的函数） for neighbor in self.graph.predecessors(current): if neighbor not in visited: visited.add(neighbor) queue.append((neighbor, current_depth + 1)) # 创建子图 subgraph = self.graph.subgraph(visited).copy() return subgraph def find_cycles(self): """查找图中的循环（递归调用）""" try: cycles = list(nx.simple_cycles(self.graph)) return cycles except nx.NetworkXNoCycle: return [] def get_metrics_report(self): """获取指标报告""" report = [ "=== 调用图分析报告 ===", f"构建时间: {self.metrics['build_time']:.2f}秒", f"函数数量: {self.metrics['nodes']}", f"调用关系数量: {self.metrics['edges']}", f"图密度: {self.metrics['density']:.6f}", f"连通分量: {self.metrics['connected_components']}", f"是否无环图: {'是' if self.metrics['is_dag'] else '否'}", f"最大入度: {self.metrics['max_in_degree']}", f"最大出度: {self.metrics['max_out_degree']}", "" ] # 添加循环信息 cycles = self.find_cycles() if cycles: report.append(f"发现循环数量: {len(cycles)}") for i, cycle in enumerate(cycles[:5]): # 只显示前5个循环 cycle_names = [self.function_info[addr]['name'] for addr in cycle] report.append(f" 循环 {i+1}: {' -> '.join(cycle_names)}") if len(cycles) > 5: report.append(f" ... 还有 {len(cycles)-5} 个循环未显示") else: report.append("未发现函数调用循环") return "\n".join(report) def export_to_dot(self, filepath, subgraph_nodes=None): """导出为DOT格式（Graphviz）""" import networkx as nx graph_to_export = self.graph if subgraph_nodes: graph_to_export = self.graph.subgraph(subgraph_nodes) # 自定义节点属性 def node_attributes(node): info = self.function_info.get(node, {}) attrs = { 'label': info.get('name', node), 'shape': 'box', 'style': 'filled', 'fillcolor': '#e1f5fe' if info.get('is_library') else '#f3e5f5' } if info.get('is_external'): attrs['shape'] = 'ellipse' attrs['fillcolor'] = '#fff3e0' return attrs # 生成DOT dot_lines = ["digraph CallGraph {"] dot_lines.append(" rankdir=LR;") dot_lines.append(" node [fontname=\"Arial\"];") # 添加节点 for node in graph_to_export.nodes(): attrs = node_attributes(node) attr_str = " ".join([f'{k}="{v}"' for k, v in attrs.items()]) dot_lines.append(f' "{node}" [{attr_str}];') # 添加边 for src, dst in graph_to_export.edges(): dot_lines.append(f' "{src}" -> "{dst}";') dot_lines.append("}") # 写入文件 with open(filepath, 'w', encoding='utf-8') as f: f.write("\n".join(dot_lines)) return len(graph_to_export.nodes()), len(graph_to_export.edges()) ``` ### 4.3 图形用户界面实现 调用图需要可视化展示，我选择使用Jython兼容的JGraphX库（mxGraph的Java版本）： ```python import com.mxgraph.swing as mxgraph_swing import com.mxgraph.view as mxgraph_view import com.mxgraph.layout as mxgraph_layout import com.mxgraph.util as mxgraph_util import javax.swing as swing import java.awt as awt from java.awt.event import MouseAdapter class GraphVisualizationPanel(swing.JPanel): """图形可视化面板""" def __init__(self, graph_builder): super(GraphVisualizationPanel, self).__init__(awt.BorderLayout()) self.graph_builder = graph_builder self.current_subgraph = None # 创建图形组件 self.graph_component = mxgraph_swing.mxGraphComponent(self._create_graph()) self.graph = self.graph_component.getGraph() # 设置图形样式 self._setup_graph_style() # 添加组件 self.add(self.graph_component, awt.BorderLayout.CENTER) # 添加鼠标监听 self.graph_component.getGraphControl().addMouseListener( self._create_graph_mouse_listener() ) # 创建工具栏 self._create_toolbar() def _create_graph(self): """创建空图形""" graph = mxgraph_view.mxGraph() # 设置图形属性 graph.setCellsEditable(False) graph.setCellsMovable(True) graph.setCellsResizable(False) graph.setCellsSelectable(True) graph.setCellsDisconnectable(False) return graph def _setup_graph_style(self): """设置图形样式""" stylesheet = self.graph.getStylesheet() # 默认顶点样式 default_vertex_style = stylesheet.getDefaultVertexStyle() default_vertex_style[mxgraph_util.mxConstants.STYLE_SHAPE] = mxgraph_util.mxConstants.SHAPE_RECTANGLE default_vertex_style[mxgraph_util.mxConstants.STYLE_ROUNDED] = True default_vertex_style[mxgraph_util.mxConstants.STYLE_FONTSIZE] = 11 default_vertex_style[mxgraph_util.mxConstants.STYLE_FONTCOLOR] = "#333333" # 库函数样式 library_style = default_vertex_style.clone() library_style[mxgraph_util.mxConstants.STYLE_FILLCOLOR] = "#e1f5fe" library_style[mxgraph_util.mxConstants.STYLE_STROKECOLOR] = "#0288d1" self.graph.getStylesheet().putCellStyle("library", library_style) # 用户函数样式 user_style = default_vertex_style.clone() user_style[mxgraph_util.mxConstants.STYLE_FILLCOLOR] = "#f3e5f5" user_style[mxgraph_util.mxConstants.STYLE_STROKECOLOR] = "#7b1fa2" self.graph.getStylesheet().putCellStyle("user", user_style) # 外部函数样式 external_style = default_vertex_style.clone() external_style[mxgraph_util.mxConstants.STYLE_SHAPE] = mxgraph_util.mxConstants.SHAPE_ELLIPSE external_style[mxgraph_util.mxConstants.STYLE_FILLCOLOR] = "#fff3e0" external_style[mxgraph_util.mxConstants.STYLE_STROKECOLOR] = "#f57c00" self.graph.getStylesheet().putCellStyle("external", external_style) # 边样式 default_edge_style = stylesheet.getDefaultEdgeStyle() default_edge_style[mxgraph_util.mxConstants.STYLE_EDGE] = mxgraph_util.mxConstants.EDGESTYLE_ORTHOGONAL default_edge_style[mxgraph_util.mxConstants.STYLE_STROKECOLOR] = "#666666" default_edge_style[mxgraph_util.mxConstants.STYLE_FONTCOLOR] = "#666666" def _create_toolbar(self): """创建工具栏""" toolbar = swing.JPanel(awt.FlowLayout(awt.FlowLayout.LEFT)) # 布局算法按钮 layout_buttons = [ ("层次布局", self._apply_hierarchical_layout), ("有机布局", self._apply_organic_layout), ("树状布局", self._apply_tree_layout), ("环形布局", self._apply_circle_layout), ] for text, action in layout_buttons: btn = swing.JButton(text) btn.addActionListener(lambda e, a=action: a()) toolbar.add(btn) # 缩放控制 zoom_label = swing.JLabel("缩放:") toolbar.add(zoom_label) zoom_in_btn = swing.JButton("+") zoom_in_btn.addActionListener(lambda e: self.graph_component.zoomIn()) toolbar.add(zoom_in_btn) zoom_out_btn = swing.JButton("-") zoom_out_btn.addActionListener(lambda e: self.graph_component.zoomOut()) toolbar.add(zoom_out_btn) reset_zoom_btn = swing.JButton("重置") reset_zoom_btn.addActionListener(lambda e: self.graph_component.zoomTo(1.0)) toolbar.add(reset_zoom_btn) self.add(toolbar, awt.BorderLayout.NORTH) def _create_graph_mouse_listener(self): """创建图形鼠标监听器""" class GraphMouseListener(MouseAdapter): def __init__(self, outer): self.outer = outer def mouseClicked(self, event): if event.getClickCount() == 2: # 双击节点跳转到函数 cell = self.outer.graph_component.getCellAt(event.getX(), event.getY()) if cell and self.outer.graph.isVertex(cell): func_addr = self.outer.graph.getLabel(cell) self.outer._navigate_to_function(func_addr) return GraphMouseListener(self) def _navigate_to_function(self, func_addr): """导航到指定函数""" # 这里需要实现跳转到Ghidra代码浏览器的功能 # 实际实现会调用Ghidra的导航API print(f"跳转到函数: {func_addr}") def display_graph(self, subgraph_nodes=None): """显示图形""" # 清除现有图形 model = self.graph.getModel() model.beginUpdate() try: # 移除所有现有单元格 cells = self.graph.getChildCells(self.graph.getDefaultParent(), True, True) self.graph.removeCells(cells) # 获取要显示的图 if subgraph_nodes: display_graph = self.graph_builder.graph.subgraph(subgraph_nodes) else: display_graph = self.graph_builder.graph # 创建节点映射 vertex_map = {} # 添加节点 for node in display_graph.nodes(): info = self.graph_builder.function_info.get(node, {}) # 确定样式 if info.get('is_external'): style = "external" elif info.get('is_library'): style = "library" else: style = "user" # 创建顶点 vertex = self.graph.insertVertex( self.graph.getDefaultParent(), None, info.get('name', node), 0, 0, # 位置由布局算法决定 120, 40, # 宽度，高度 style ) vertex_map[node] = vertex # 添加边 for src, dst in display_graph.edges(): if src in vertex_map and dst in vertex_map: self.graph.insertEdge( self.graph.getDefaultParent(), None, "", # 边标签 vertex_map[src], vertex_map[dst] ) self.current_subgraph = subgraph_nodes finally: model.endUpdate() # 应用默认布局 self._apply_hierarchical_layout() # 居中显示 self.graph_component.zoomToFit() def _apply_hierarchical_layout(self): """应用层次布局""" layout = mxgraph_layout.mxHierarchicalLayout(self.graph) layout.setIntraCellSpacing(50) layout.setInterRankCellSpacing(100) layout.setOrientation(mxgraph_util.mxConstants.DIRECTION_NORTH) layout.execute(self.graph.getDefaultParent()) def _apply_organic_layout(self): """应用有机布局""" layout = mxgraph_layout.mxOrganicLayout(self.graph) layout.setForceConstant(100) layout.execute(self.graph.getDefaultParent()) def _apply_tree_layout(self): """应用树状布局""" layout = mxgraph_layout.mxCompactTreeLayout(self.graph) layout.setLevelDistance(50) layout.setNodeDistance(30) layout.execute(self.graph.getDefaultParent()) def _apply_circle_layout(self): """应用环形布局""" layout = mxgraph_layout.mxCircleLayout(self.graph) layout.setRadius(300) layout.execute(self.graph.getDefaultParent()) def export_to_image(self, filepath, format="PNG"): """导出为图片""" from javax.imageio import ImageIO try: # 创建图像 img = self.graph_component.createImage( self.graph_component.getBackground(), self.graph_component.getWidth(), self.graph_component.getHeight() ) # 保存图像 ImageIO.write(img, format, java.io.File(filepath)) return True except Exception as e: print(f"导出图像失败: {e}") return False ``` ### 4.4 完整插件集成与测试 最后，将各个组件集成到主插件中： ```python from ghidra.app.plugin import ProgramPlugin from ghidra.util import Msg from ghidra.util.task import TaskMonitor import java.awt.BorderLayout as BorderLayout import javax.swing.JSplitPane as JSplitPane import javax.swing.JTabbedPane as JTabbedPane class CallGraphPlugin(ProgramPlugin): """函数调用图插件主类""" def __init__(self, tool): super(CallGraphPlugin, self).__init__(tool, True, True) self.graph_builder = None self.current_program = None # 创建UI组件 self._create_ui() # 注册服务 self._register_services() # 添加事件监听 self._add_listeners() def _create_ui(self): """创建用户界面""" # 主面板 self.main_panel = swing.JPanel(BorderLayout()) # 创建分割面板 split_pane = JSplitPane(JSplitPane.HORIZONTAL_SPLIT) # 左侧控制面板 self.control_panel = ControlPanel(self) # 右侧图形面板 self.graph_panel = GraphVisualizationPanel(None) # 标签面板（用于多个视图） self.tabbed_pane = JTabbedPane() self.tabbed_pane.addTab("调用图", self.graph_panel) # 添加组件 split_pane.setLeftComponent(self.control_panel) split_pane.setRightComponent(self.tabbed_pane) split_pane.setDividerLocation(300) self.main_panel.add(split_pane, BorderLayout.CENTER) # 创建提供器 self.provider = CallGraphProvider(self, "函数调用图分析") self.provider.setComponent(self.main_panel) # 添加到工具窗口 tool.addComponentProvider(self.provider, False) def _register_services(self): """注册插件服务""" # 这里可以注册分析服务、导出服务等 pass def _add_listeners(self): """添加事件监听器""" # 监听程序变化 tool.addProgramLocationListener(self) tool.addProgramSelectionListener(self) def programActivated(self, program): """程序激活时调用""" self.current_program = program self.control_panel.set_program(program) Msg.info(self, f"已激活程序: {program.getName()}") def programDeactivated(self, program): """程序停用时调用""" self.current_program = None self.control_panel.set_program(None) self.graph_panel.display_graph(None) Msg.info(self, "程序已停用") def build_call_graph(self, options=None): """构建调用图""" if not self.current_program: Msg.showWarn(self, None, "错误", "请先打开一个程序") return # 显示进度对话框 monitor = TaskMonitor() progress_dialog = ProgressDialog( self.tool.getActiveWindow(), "构建调用图", monitor ) # 在新线程中构建 import threading def build_task(): try: self.graph_builder = CallGraphBuilder(self.current_program) success = self.graph_builder.build(monitor, options) if success: # 更新UI（需要在事件分发线程中执行） def update_ui(): self.graph_panel.graph_builder = self.graph_builder self.graph_panel.display_graph() self.control_panel.on_graph_built(self.graph_builder) progress_dialog.close() Msg.info(self, "调用图构建完成") swing.SwingUtilities.invokeLater(update_ui) else: swing.SwingUtilities.invokeLater(lambda: Msg.showError( s