mahmoudimus/ida-taskr

# ida-taskr  ## 概述 IDA Taskr 是一个纯 Python 库，用于 IDA Pro 并行计算。它允许你利用 Qt（内置于 IDA！）的强大功能和 Python 的 multiprocessing，将计算密集型任务卸载到 worker 进程中，而不会冻结 IDA Pro 的 UI。 **主要特性：** - 🚀 简单的装饰器 API - 只需添加 `@cpu_task` 即可在后台运行 - 🔄 基于进程的并行机制，实现真正的多核执行 - 📦 支持大型二进制数据的共享内存 - 🎯 集成 Qt 信号，用于进度回调 - ⚡ 兼容 IDA Pro 9.1 (PyQt5) 和 9.2+ (PySide6) ## 安装 **选项 1：单文件（无需安装）** 下载 [`ida_taskr_amalgamated.py`](https://github.com/mahmoudimus/ida-taskr/releases/latest) 并放入 IDA 的 plugins 文件夹中。就这样 - 一个文件，零依赖！ **选项 2：pip install** ``` pip install ida-taskr # 或者从支持 Qt 的 source 构建 pip install -e .[pyqt5] # For IDA Pro 9.1 pip install -e .[pyside6] # For IDA Pro 9.2+ ``` **选项 3：IDA Plugin Manager (HCLI)** 安装一次 [HCLI](https://hcli.docs.hex-rays.com/)，然后搜索并 安装 Taskr： ``` hcli plugin search ida-taskr hcli plugin install ida-taskr ``` HCLI 会将插件安装到 `$IDAUSR/plugins/ida-taskr` 下，并且首先 将匹配的 `ida-taskr` 包安装到 IDA 的 Python 环境中。IDA 会 提供 Qt 绑定，因此不会单独安装 PyQt 或 PySide 包。在 命令执行完成后重启 IDA。 对于嵌入或非 HCLI 安装，请使用 IDA 使用的 Python 解释器安装该包： ``` python -m pip install ida-taskr ``` ## 快速开始 ### 最简单的方式：`@cpu_task` 添加一个装饰器，你的函数就会在后台运行： ``` from ida_taskr import cpu_task @cpu_task def analyze_binary(data): """This runs in a background thread - UI stays responsive!""" result = [] for byte in data: result.append(process_byte(byte)) return result # Usage - 立即返回！ future = analyze_binary(binary_data) # 在它运行时执行其他工作... # 在需要时获取 result result = future.result() ``` 就这样。一行代码。你的函数现在运行时不会阻塞 IDA。 ### 带回调 在你的任务完成时收到通知： ``` from ida_taskr import cpu_task @cpu_task(on_complete=lambda r: print(f"Done! Found {len(r)} patterns")) def find_patterns(data): return scan_for_patterns(data) # Fire and forget - callback 处理 result find_patterns(binary_data) ``` ### 并行处理 在多个 worker 线程中处理多项： ``` from ida_taskr import parallel @parallel(max_workers=8) def analyze_function(func_ea): """Analyze a single function.""" return get_function_signature(func_ea) # 并行处理 100 个函数 function_addresses = list(idautils.Functions()) futures = [analyze_function(addr) for addr in function_addresses] results = [f.result() for f in futures] ``` ### 使用共享内存处理大型数据 对于大型二进制 blob（兆字节），使用共享内存以避免复制： ``` from ida_taskr import shared_memory_task @shared_memory_task(num_chunks=8) def find_signatures(chunk_data, chunk_id, total_chunks): """ Process one chunk of the binary. chunk_data: memoryview of this chunk (zero-copy!) chunk_id: which chunk this is (0-7) total_chunks: total number of chunks (8) """ signatures = [] for i in range(len(chunk_data) - 16): if is_interesting_pattern(chunk_data[i:i+16]): signatures.append(bytes(chunk_data[i:i+16])) return signatures # ida-taskr 处理所有 shared memory 的复杂性！ binary_data = ida_bytes.get_bytes(start_ea, size) # e.g., 8MB all_signatures = find_signatures(binary_data) # Returns list of 8 results ``` ## 装饰器参考 | 装饰器 | 用例 | 示例 | |-----------|----------|---------| | `@cpu_task` | CPU 密集型工作 | 模式扫描，签名生成 | | `@io_task` | I/O 密集型工作 | 网络请求，文件操作 | | `@parallel(n)` | 多个并行任务 | 批量函数分析 | | `@background_task` | 带回调的完全控制 | 进度报告 | | `@shared_memory_task` | 大型数据处理 | 多 MB 的二进制分析 | ### `@background_task` - 完全控制 ``` from ida_taskr import background_task @background_task( max_workers=4, on_complete=lambda r: print(f"Result: {r}"), on_error=lambda e: print(f"Error: {e}"), on_progress=lambda p, m: print(f"[{p}%] {m}"), executor_type='process' # 'thread' or 'process' ) def heavy_analysis(data, progress_callback=None): for i, chunk in enumerate(chunks(data, 100)): process(chunk) if progress_callback: progress_callback(i * 10, f"Processed chunk {i}") return "done" ``` ## 高级用法 ### 直接访问 Executor 要进行更多控制，请直接使用 executor： ``` from ida_taskr import ProcessPoolExecutor, ThreadExecutor # 基于 Process（真正的 parallelism，绕过 GIL） with ProcessPoolExecutor(max_workers=4) as executor: futures = [executor.submit(cpu_task, arg) for arg in args] results = [f.result() for f in futures] # 基于 Thread（适用于释放 GIL 的 IDA SDK 调用） with ThreadExecutor(max_workers=8) as executor: futures = [executor.submit(analyze_func, ea) for ea in function_list] results = [f.result() for f in futures] ``` ### Worker 脚本（双向 IPC） 适用于需要持久化 worker 和双向通信的复杂场景。 在以下情况使用： - 在任务之间保持活动状态的长期运行 worker 进程 - IDA 和 worker 之间的自定义消息协议 - 在工作进展时将结果流式传输回 IDA - 跨多个命令持久化的 worker 状态 ``` from ida_taskr import TaskRunner runner = TaskRunner( worker_script="path/to/worker.py", worker_args=["arg1", "arg2"] ) @runner.on('worker_results') def handle_results(results): print(f"Results: {results}") @runner.on('worker_message') def handle_progress(msg): print(f"Progress: {msg}") runner.start() runner.send_command({"command": "process", "data": [1, 2, 3]}) # Worker 保持存活，可以发送更多 commands... runner.send_command({"command": "analyze", "target": 0x401000}) runner.stop() ``` 请参阅 [examples/](examples/) 获取更详细的示例，包括： - [极简示例](examples/ultra_minimal.py) - 最小可行代码 - [共享内存模式](examples/shared_memory_parallel_example.py) - 大型数据处理 - [签名生成](examples/signature_generation_example.py) - 真实的 IDA 用例 - [QtAsyncio 集成](examples/qtasyncio_event_loop.py) - Async/await 支持 ## 测试 ``` # 运行所有 unit tests ./run_tests.sh # 运行 Qt integration tests pytest tests/integration/test_integration_qt_core.py -v # 运行以获取 coverage pytest tests/ --cov=src/ida_taskr --cov-report=html ``` **支持的配置：** - ✅ Python 3.11, 3.12, 3.13 - ✅ PyQt5 (IDA Pro 9.1) - ✅ PySide6 (IDA Pro 9.2+) ## 文档 - [QtAsyncio 集成](docs/QTASYNCIO.md) - Async/await 和事件循环细节 - [IDA 测试指南](docs/IDA_TESTING.md) - 在 IDA Pro 中运行测试 - [示例 README](examples/README.md) - 全面的示例指南 ## 许可证 📜 MIT 许可证 - 详见 [LICENSE](LICENSE)。 ## 联系方式 📧 有问题或疑问？在 GitHub 上提 issue，或者联系 [@mahmoudimus](https://github.com/mahmoudimus)。

标签：IDA Pro插件, PyQt, Python, 云资产清单, 并行计算, 异步编程, 无后门, 计算机取证, 逆向工具, 逆向工程