安全Agent从0到1
·
前言
自ChatGPT问世以来，大语言模型发展以不可想象的速度发展，随时间流失，工业界也终于有了它的反应，智能体Agent应运而生，ClaudeCode、Cursor、Copilot、MiniMax、Trae、通义灵码等国内外各个产品被推出。
目前，各个垂直领域也闻风而动，安全领域也开始了定制化开发
-
Shannon: 完全自主的AI工具，集成了NMAP和漏洞利用功能。
-
WPeChatGPT: 基于与 ChatGPT 相同模型的 IDA 插件。
-
IDA Pro MCP: 针对IDA开发的MCP服务。
-
mcp-kali-server: Kali定制化MCP服务。
-
Evasion SubAgents: 基于 Claude Code 的免杀技术研究与 Shellcode Loader 生成框架。
-
red-run: Claude Code 的攻击性安全工具包。
-
NexusAI: AI驱动的IDA Pro插件
-
IDA Auto MCP: 无界面 IDA Pro MCP 服务器，让 AI 智能体自动打开、分析和查询多个二进制文件，无需手动操作 IDA GUI
-
AI-Assisted Reverse Engineering with Ghidra: 逆向工具Ghidra的MCP服务
-
METATRON: 基于Linux（Parrot OS）的本地LLM驱动的AI渗透测试助手
-
DeepZero: DeepZero 是一个自动化、由人工智能驱动的框架，用于大规模发现 Windows 内核驱动中的零日漏洞。
-
MCP Clients: 宝藏网站，里面可以搜集到各种不同的 mcp server
-
MCP Server for WinDbg Crash Analysis: 一款模型上下文协议服务器，连接 AI 模型与 WinDbg，用于崩溃转储分析和远程调试。
-
x64dbg MCP: 一台MCP服务器，可以桥接各种LLMS（Claude and Cursor testes）与x64dbg调试器，通过提示直接访问调试应用。
-
ghidraMCP: ghidraMCP 是一个模型上下文协议服务器，允许大型语言模型自主逆向工程应用。它向MCP客户端展示了大量Ghidra核心功能的工具。
同样的，还有像腾讯安全众测智能渗透挑战赛，有很多项目可以参考
IDA Pro MCP
安装（这里权限问题，将插件安装到了C盘，不建议像我这样操作
D:\xx xx\IDA Professional 9.3\python-3.13.12>python -m pip install --no-build-isolation https://github.com/mrexodia/ida-pro-mcp/archive/refs/heads/main.zip
Collecting https://github.com/mrexodia/ida-pro-mcp/archive/refs/heads/main.zip
Using cached https://github.com/mrexodia/ida-pro-mcp/archive/refs/heads/main.zip (1.5 MB)
Preparing metadata (pyproject.toml) ... done
………………………………………………………………………………………………………………………………
Successfully installed ida-pro-mcp-2.0.0 idapro-0.0.7 tomli-w-1.2.0
D:\xx xx\IDA Professional 9.3\python-3.13.12>cd Scripts
D:\xx xx\IDA Professional 9.3\python-3.13.12\Scripts>ida-pro-mcp --install
Installed IDA Pro plugin (IDA restart required)
loader: xx\AppData\Roaming\Hex-Rays\IDA Pro\plugins\ida_mcp.py
package: xx\AppData\Roaming\Hex-Rays\IDA Pro\plugins\ida_mcp
[?25l[1mSelect transport mode:[0m
(up/down: move, enter: confirm, esc: cancel)
[36m>[0m Streamable HTTP (recommended)
stdio
SSE
同时ida需要手动启动MCP
查看配置，并将配置拷贝到客户端中（cursor、vscode插件cline、trae等）
D:\xx xx\IDA Professional 9.3\python-3.13.12\Scripts>ida-pro-mcp --config
[STDIO MCP CONFIGURATION]
{
"mcpServers": {
"ida-pro-mcp": {
"command": "D:\\xx xx\\IDA Professional 9.3\\python-3.13.12\\python.exe",
"args": [
"D:\\xx xx\\IDA Professional 9.3\\python-3.13.12\\Lib\\site-packages\\ida_pro_mcp\\server.py",
"--ida-rpc",
"http://127.0.0.1:13337"
]
}
}
}
…………………………………………………………………………………………………………………………
这里使用vscode插件trae进行配置
开始分析，成功给出分析结果。体验可以节省大量时间
NexusAI
安装极其方便，默认即可
在窗口中，选择编辑 → NexusAI → 设置进行配置
然后快捷键Ctrl+Shift+K打开交互窗口（注：这里发现该交互窗口无AI回复，但实际上AI已完成操作，可从历史中查看日志，发现交互。但经过多轮测试，大概率是交互逻辑存在一些问题，未深入排查
实际体验有待提升（其实一般般，感觉不如IDA Pro MCP，不能说没用，但达不到好用
OpenClaw（题外话
龙虾，说实话，我感觉目前对我没有什么帮助（未来不知道，个人观点
但龙虾热的都快熟了，怎么也得亲自上手体验一番，才有自己的理解。
- 安装nodejs，默认选择最新版即可
- 安装git，默认即可
前提依赖安装完成
C:\Users\test>node -v
v24.14.1
C:\Users\test>git -v
git version 2.53.0.windows.2
- 安装openclaw
- 配置openclaw，配置时model、key、skill、bot全部建议跳过，后续慢慢配置即可
修改配置文件C:\Users\test\.openclaw\openclaw.json，补充以下信息（替换为自己的API key
"models": {
"providers": {
"volcengine": {
"baseUrl": "https://ark.cn-beijing.volces.com/api/v3",
"apiKey": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
"api": "openai-completions",
"models": [
{
"id": "doubao-seed-2-0-pro-xxxxxx",
"name": "Doubao-Seed-2.0-pro",
"reasoning": true,
"input": ["text"],
"cost": {
"input": 0.0032,
"output": 0.016,
"cacheRead": 0,
"cacheWrite": 0
},
"contextWindow": 32768,
"maxTokens": 4096
}
]
}
}
},
"agents": {
"defaults": {
"model": {
"primary": "volcengine/doubao-seed-2-0-pro-xxxxxx"
},
"workspace": "C:\\Users\\test\\.openclaw\\workspace"
}
},
- 重启，执行
openclaw gateway restart命令
CTF Web Agent
我们先来个简单的Agent练练手！
平时针对CTF竞赛中关于web题目，在之前，是通过拥有大量经验的专家来进行的。而大模型出来之后，所谓专家，则可以由LLM进行替代，假如我们替LLM装上一个简单的命令行终端，那么它就可以替代人类专家
数据集XBOW - 网络攻击工具熟练度基准测试，这个可以测试智能体的性能，或者使用BUUCTF等平台题目进行测试
配置沙箱
使用docker拉取镜像，给Agent提供沙箱用以执行命令
xxx@xxx:~$ docker pull ubuntu:24.04
24.04: Pulling from library/ubuntu
817807f3c64e: Pull complete
Digest: sha256:186072bba1b2f436cbb91ef2567abca677337cfc786c86e107d25b7072feef0c
Status: Downloaded newer image for ubuntu:24.04
docker.io/library/ubuntu:24.04
xxx@xxx:~$ docker pull kalilinux/kali-rolling
Using default tag: latest
latest: Pulling from kalilinux/kali-rolling
194d70789344: Pull complete
Digest: sha256:287cd5cfa409e258e9ec3661db4dff0bfbb45fc95734e82d3b270a0f749629ca
Status: Downloaded newer image for kalilinux/kali-rolling:latest
docker.io/kalilinux/kali-rolling:latest
xxx@xxx:~$ docker pull python:3.12-slim
3.12-slim: Pulling from library/python
ec781dee3f47: Pull complete
b1a20e2fae4c: Pull complete
a6d1911b36ac: Pull complete
e3c59d77c03e: Pull complete
Digest: sha256:3d5ed973e45820f5ba5e46bd065bd88b3a504ff0724d85980dcd05eab361fcf4
Status: Downloaded newer image for python:3.12-slim
docker.io/library/python:3.12-slim
欧克，首先我们配置kali的沙箱环境
xxx@xxx:~$ docker run -it --name my-kali -p 8888:8888 kalilinux/kali-rolling
┌──(root㉿6be6a794a792)-[/]
└─# apt update -y
Get:1 http://mirror.nyist.edu.cn/kali kali-rolling InRelease [34.0 kB]
Get:2 http://mirror.nyist.edu.cn/kali kali-rolling/non-free-firmware amd64 Packages [14.3 kB]
Get:3 http://mirror.nyist.edu.cn/kali kali-rolling/main amd64 Packages [21.0 MB]
Get:4 http://mirror.nyist.edu.cn/kali kali-rolling/contrib amd64 Packages [119 kB]
Get:5 http://mirror.nyist.edu.cn/kali kali-rolling/non-free amd64 Packages [186 kB]
Fetched 21.4 MB in 7s (3041 kB/s)
All packages are up to date.
……………………………………………………………………………………………………………………
自行安装工具
……………………………………………………………………………………………………………………
配置python沙盒环境
xxx@xxx:~$ docker run -it --name my-python -p 8889:8889 python:3.12-slim /bin/sh
root@e9f63784db96:/# pip install requests scapy beautifulsoup4 lxml pwntools pycryptodome flask
Requirement already satisfied: requests in /usr/local/lib/python3.12/site-packages (2.33.0)
Collecting scapy
Downloading scapy-2.7.0-py3-none-any.whl.metadata (5.8 kB)
Collecting beautifulsoup4
Downloading beautifulsoup4-4.14.3-py3-none-any.whl.metadata (3.8 kB)
……………………………………………………………………………………………………………………
编写程序
首先我们可以定义基类Agent
from core.config import Config
from abc import ABC, abstractmethod
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, AgentState, before_model, after_model
from langchain.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.runtime import Runtime
from typing import Any
from langchain.chat_models import init_chat_model
from langchain.agents.structured_output import ProviderStrategy
import os
LOG_FD = 1
# 日志记录(模型调用前)
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
name = "recv <- " + str(state["messages"][-1].name)
first_message = name + ': ' + str(state["messages"][-1].content) + '\n'
os.write(LOG_FD, first_message.encode('gb2312'))
return None
# 日志记录(模型调用后)
@after_model()
def log_after_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
name = "send -> " + str(state["messages"][-2].name)
last_message = name + ': ' + str(state["messages"][-1].content) + '\n'
os.write(LOG_FD, last_message.encode('gb2312'))
return None
class Agent(ABC):
def __init__(self):
self.config = Config()
self.model_name = self.config.get('llm.default.model_name')
self.api_key = self.config.get('llm.default.api_key')
self.base_url = self.config.get('llm.default.base_url')
self.middleware = [log_before_model, log_after_model]
self.message = []
self.tools = []
self.max_step = 30
def init_model(self):
self.llm = init_chat_model(
model_provider = "openai",
model = self.model_name,
api_key = self.api_key,
base_url = self.base_url,
)
def init_agent(self):
if not hasattr(self, 'response_format'):
self.response_format = None
self.agent = create_agent(
model = self.llm,
tools = self.tools,
middleware = self.middleware,
system_prompt = SystemMessage(content=self.init_system_prompt()),
response_format = self.response_format
)
def send(self) -> str:
if not hasattr(self, 'agent'):
self.init_agent()
response = self.agent.invoke({"messages": self.message})
return response['messages'][-1].content
def run(self, task: str):
self.message = []
self.message.append(HumanMessage(content=task, name=self.name))
content = self.send()
return content
@abstractmethod
def init_system_prompt(self) -> str:
"""初始化系统提示"""
pass
配置类Config
import yaml, os
class Config:
def __init__(self, config_path=None):
self.config_path = config_path or os.path.join(os.path.dirname(__file__), '../config.yml')
self.config = self._load_config()
def _load_config(self):
with open(self.config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def get(self, key, default=None):
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
实际上一个demo是很容易实现的
from langchain_core.messages import HumanMessage
from core.tools.python_tool import run_python_command, run_system_command
from core.agent.main_agent import Agent
class CTFWebAgent(Agent):
def __init__(self):
super().__init__()
self.name = 'CTFWebAgent'
self.model_name = self.config.get('llm.ctf_web_agent.model_name') or self.model_name
self.api_key = self.config.get('llm.ctf_web_agent.api_key') or self.api_key
self.base_url = self.config.get('llm.ctf_web_agent.base_url') or self.base_url
self.tools = self.tools + [run_python_command, run_system_command]
self.init_model()
self.init_agent()
def init_system_prompt(self) -> str:
system_prompt = """
你是一名专业的CTF Web安全自动化解题Agent，目标是高效完成CTF中的Web题目。请严格遵循以下步骤：
请按照以下逻辑执行解题操作：
1. 分析目标URL，识别可能的漏洞类型（如SQL注入、XSS、CSRF、文件包含、命令执行等）
2. 基于当前进度，生成下一步具体操作指令（包括Payload构造、工具调用、参数测试、代码审计目标等，直接使用提供的工具执行）
3. 若当前进度存在错误操作，给出修正后的操作方案
注意：
- 输出仅包含必要的操作指令，避免冗余
- 复杂漏洞分步骤给出可执行指令
- 每次操作前都需要先进行思考
- 工具执行后需要记录观察结果
- 基于观察结果调整后续策略
"""
return system_prompt
工具实现
工具的实现，可谓是核心中的核心，它是智能体的双手。这里我们关于tool的实现，首先是沙箱kali中的server端
import socket
import subprocess
import json
import threading
class KaliMCPServer:
def __init__(self, host='0.0.0.0', port=8888):
self.host = host
self.port = port
self.server_socket = None
self.running = False
def start(self): # 启动 MCP 服务器
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_socket.bind((self.host, self.port))
self.server_socket.listen(5)
self.running = True
print(f"Kali MCP Server start success, listen {self.host}:{self.port}")
while self.running:
try:
client_socket, client_address = self.server_socket.accept()
print(f"accept client {client_address}")
# 为每个客户端创建一个线程处理
client_thread = threading.Thread(
target=self.handle_client,
args=(client_socket,)
)
client_thread.daemon = True
client_thread.start()
except socket.error as e:
if not self.running:
break
print(f"server error: {e}")
def handle_client(self, client_socket): # 处理客户端连接
try:
# 接收客户端发送的命令
data = client_socket.recv(1024)
if not data:
return
try: # 解析命令
request = json.loads(data.decode('utf-8'))
command = request.get('command')
if not command:
response = {
'success': False,
'message': 'missing command parameter'
}
else:
result = self.execute_command(command)
response = {
'success': True,
'result': result
}
except json.JSONDecodeError:
response = {
'success': False,
'message': 'invalid JSON format'
}
except Exception as e:
response = {
'success': False,
'message': f'handle command error: {str(e)}'
}
client_socket.sendall(json.dumps(response).encode('utf-8'))
except socket.error as e:
print(f"client error handle: {e}")
finally:
client_socket.close()
def execute_command(self, command):
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30
)
return {
'stdout': result.stdout,
'stderr': result.stderr,
'returncode': result.returncode
}
except subprocess.TimeoutExpired:
return {
'stdout': '',
'stderr': 'command execute timeout',
'returncode': -1
}
except Exception as e:
return {
'stdout': '',
'stderr': f'execute command error: {str(e)}',
'returncode': -1
}
def stop(self): # 停止 MCP 服务器
self.running = False
if self.server_socket:
self.server_socket.close()
print("Kali MCP Server stop success")
if __name__ == "__main__":
server = KaliMCPServer()
try:
server.start()
except KeyboardInterrupt:
server.stop()
对接到Agent的客户端
from langchain.tools import tool
import socket
import json
class KaliToolClient:
def __init__(self, host='localhost', port=8888, timeout=30):
self.host = host
self.port = port
self.timeout = timeout
def execute_command(self, command):
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(self.timeout)
s.connect((self.host, self.port))
request = json.dumps({'command': command})
s.sendall(request.encode('utf-8'))
response = b''
while True:
data = s.recv(1024)
if not data:
break
response += data
if not response:
return {'success': False, 'message': 'mcp server execute error!'}
result = json.loads(response.decode('utf-8'))
print(f"response: {result}")
return result
except Exception as e:
return {'success': False, 'message': f'mcp server execute error: {str(e)}'}
kali_client = KaliToolClient()
@tool
def run_kali_command(command: str) -> str:
"""于 Kali Linux 中执行命令
Args:
command: 于 Kali Linux 中执行的命令
Returns:
str: 命令执行结果
"""
result = kali_client.execute_command(command)
if result.get('success'):
command_result = result.get('result', {})
stdout = command_result.get('stdout', '')
stderr = command_result.get('stderr', '')
output = stdout
if stderr:
output += stderr
return output.strip()
else:
return result.get('message', 'mcp server execute error!')
@tool
def list_kali_command() -> None:
"""
环境 Kali Linux 中额外可用的工具列表：
1. fscan 内网扫描、端口扫描
2. nuclei 自动化漏洞扫描
3. sqlmap SQL注入检测
4. python3 Python执行环境
5. nc 反弹shell、端口监听
6. ffuf 目录爆破、FUZZ测试
"""
pass
上面这种方法其实并不推荐，但也算是一种方案
同样的，我们也可以采用Kali官方MCP服务 mcp-kali-server对接到我们的智能体中，不过经过测试，好像这个并不完善
接下来，我们采用官方mcp库来编写python mcp server，这样就简单多了
#!/usr/bin/env python3
import subprocess
import asyncio
import argparse
from typing import Any, Literal
async def execute_command(command: str, is_system_command: bool = False) -> dict[str, Any]:
"""执行命令的内部函数"""
try:
if is_system_command:
result = await asyncio.create_subprocess_shell(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(
result.communicate(),
timeout=30
)
return {
'stdout': stdout.decode('utf-8', errors='replace'),
'stderr': stderr.decode('utf-8', errors='replace'),
'returncode': result.returncode
}
except asyncio.TimeoutError:
result.kill()
await result.wait()
return {
'stdout': '',
'stderr': 'command execute timeout',
'returncode': -1
}
else:
result = await asyncio.create_subprocess_exec(
'python', '-c', command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(
result.communicate(),
timeout=30
)
return {
'stdout': stdout.decode('utf-8', errors='replace'),
'stderr': stderr.decode('utf-8', errors='replace'),
'returncode': result.returncode
}
except asyncio.TimeoutError:
result.kill()
await result.wait()
return {
'stdout': '',
'stderr': 'command execute timeout',
'returncode': -1
}
except Exception as e:
return {
'stdout': '',
'stderr': f'execute command error: {str(e)}',
'returncode': -1
}
# ========== 启动 MCP 服务 ==========
def run_mcp_server(transport: Literal["stdio", "sse", "streamable-http"] = "stdio", host: str = "0.0.0.0", port: int = 8889):
try:
from mcp.server.fastmcp import FastMCP
app = FastMCP("python-mcp-server", host=host, port=port)
@app.tool()
async def run_python_command(code: str) -> str:
"""于 Python 环境中执行 Python 代码
Args:
code: 要执行的 Python 代码
Returns:
str: 代码执行结果
"""
result = await execute_command(code, is_system_command=False)
return result['stdout'] + result['stderr']
@app.tool()
async def run_system_command(command: str) -> str:
"""于 Python 环境中执行系统命令
Args:
command: 要执行的系统命令
Returns:
str: 命令执行结果
"""
result = await execute_command(command, is_system_command=True)
return result['stdout'] + result['stderr']
app.run(transport=transport)
except ImportError:
print("Error: FastMCP create failed for MCP server")
exit(1)
def main():
"""主函数"""
parser = argparse.ArgumentParser(description="Python MCP Server")
parser.add_argument(
"--stdio",
action="store_true",
help="Run in stdio mode (default: HTTP/SSE mode)"
)
parser.add_argument(
"--host",
default="0.0.0.0",
help="Host to listen on (HTTP/SSE mode only)"
)
parser.add_argument(
"--port",
type=int,
default=8889,
help="Port to listen on (HTTP/SSE mode only)"
)
args = parser.parse_args()
if args.stdio:
transport = "stdio"
else:
print(f"Starting Python MCP Server on http://{args.host}:{args.port}")
transport = "streamable-http"
try:
run_mcp_server(transport=transport, host=args.host, port=args.port)
except KeyboardInterrupt:
print(f"Error: MCP server stopped by user")
exit(0)
if __name__ == "__main__":
main()
客户端也极其简单，这里给出了同步方案与异步方案
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.tools import BaseTool
import asyncio
from typing import List, Tuple
async def get_python_tools_async(
host: str = '172.28.154.126',
port: int = 8889
) -> Tuple[List[BaseTool], MultiServerMCPClient]:
client = MultiServerMCPClient({
"python": {
"transport": "http",
"url": f"http://{host}:{port}/mcp"
}
})
tools = await client.get_tools()
return tools, client
def get_python_tools_sync(
host: str = '172.28.154.126',
port: int = 8889
) -> Tuple[List[BaseTool], MultiServerMCPClient]:
return asyncio.run(get_python_tools_async(host, port))
sync_tools, sync_client = get_python_tools_sync()
if __name__ == "__main__":
tools, client = asyncio.run(get_python_tools_async(port=9000))
print(tools)
最终汇总，编写一个ctf web demo
from langchain_core.messages import HumanMessage
from core.tools.python_tool import sync_tools
from core.tools.kali_tool import run_kali_command, list_kali_command
from core.agent.main_agent import Agent
class CTFWebAgent(Agent):
def __init__(self):
super().__init__()
self.name = 'CTFWebAgent'
self.model_name = self.config.get('llm.ctf_web_agent.model_name') or self.model_name
self.api_key = self.config.get('llm.ctf_web_agent.api_key') or self.api_key
self.base_url = self.config.get('llm.ctf_web_agent.base_url') or self.base_url
self.tools = self.tools + sync_tools + [run_kali_command, list_kali_command]
self.init_model()
self.init_agent()
async def arun(self, task: str):
self.message = []
self.message.append(HumanMessage(content=task, name=self.name))
response = await self.agent.ainvoke({"messages": self.message})
return response['messages'][-1].content
def init_system_prompt(self) -> str:
system_prompt = """
你是一名专业的CTF Web安全自动化解题Agent，目标是高效完成CTF中的Web题目。请严格遵循以下步骤：
请按照以下逻辑执行解题操作：
1. 分析目标URL，识别可能的漏洞类型（如SQL注入、XSS、CSRF、文件包含、命令执行等）
2. 基于当前进度，生成下一步具体操作指令（包括Payload构造、工具调用、参数测试、代码审计目标等，直接使用提供的工具执行）
3. 若当前进度存在错误操作，给出修正后的操作方案
注意：
- 输出仅包含必要的操作指令，避免冗余
- 复杂漏洞分步骤给出可执行指令
- 每次操作前都需要先进行思考
- 工具执行后需要记录观察结果
- 基于观察结果调整后续策略
"""
return system_prompt
这个demo可以通过一些单线条漏洞的测试，但是经过深入测试，发现在一些复杂组合场景中（比如渗透，像渗透则需要Multi-Agent来进行工作，且需要对细节进行适配），这种简单的架构仍然不行
不过这里就不再多编写了，可以在此基础上引入多智能体编排，重试机制等一些措施来提升能力
CTF Pwn Agent
同样的，这里给出关于针对Pwn环境的沙箱，通过MCP协议操控，服务端代码
#!/usr/bin/env python3
import subprocess
import asyncio
import argparse
from typing import Any, Literal
async def execute_command(command: str) -> dict[str, Any]:
"""执行命令的内部函数"""
try:
result = await asyncio.create_subprocess_shell(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(
result.communicate(),
timeout=30
)
return {
'stdout': stdout.decode('utf-8', errors='replace'),
'stderr': stderr.decode('utf-8', errors='replace'),
'returncode': result.returncode
}
except asyncio.TimeoutError:
result.kill()
await result.wait()
return {
'stdout': '',
'stderr': 'command execute timeout',
'returncode': -1
}
except Exception as e:
return {
'stdout': '',
'stderr': f'execute command error: {str(e)}',
'returncode': -1
}
def interact_with_gdb():
gdb_process = subprocess.Popen(
['gdb', '--interpreter=mi'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True
)
while True:
output = gdb_process.stdout.readline()
if not output:
break
if output.strip() == "(gdb)" or output.strip().startswith("$"):
break
return gdb_process
gdb_process = None
def execute_gdb_shell(user_input : str) -> str:
global gdb_process
if gdb_process is None:
gdb_process = interact_with_gdb()
try:
if user_input.lower().strip() in ["quit", "exit", "q"]:
gdb_process.stdin.write("quit\n")
gdb_process.stdin.flush()
buf_size = ""
while True:
output = gdb_process.stdout.readline()
if not output:
break
buf_size += output
gdb_process.wait()
gdb_process = None
return buf_size.rstrip()
gdb_process.stdin.write(user_input + "\n")
gdb_process.stdin.flush()
buf_size = ""
while True:
output = gdb_process.stdout.readline()
if not output:
break
buf_size += output
if output.strip() == "(gdb)" or output.strip().startswith("$"):
break
return buf_size.rstrip()
except Exception as e:
if gdb_process and gdb_process.poll() is None:
gdb_process.terminate()
gdb_process.wait()
gdb_process = None
return f"gdb error: {str(e)}"
# ========== 启动 MCP 服务 ==========
def run_mcp_server(transport: Literal["stdio", "sse", "streamable-http"] = "stdio", host: str = "0.0.0.0", port: int = 8890):
try:
from mcp.server.fastmcp import FastMCP
app = FastMCP("ubuntu-mcp-server", host=host, port=port)
@app.tool()
async def run_ubuntu_command(command: str) -> str:
"""于 Ubuntu Linux 中执行命令
Args:
command: 于 Ubuntu Linux 中执行的命令
Returns:
str: 命令执行结果
"""
result = await execute_command(command)
return result['stdout'] + result['stderr']
@app.tool()
async def list_commands() -> None:
"""
1. readelf
2. checksec
3. file
4. objdump
5. nc
6. strings
7. radare2
"""
pass
@app.tool()
async def execute_gdb(command: str) -> str:
"""于交互式会话中执行 GDB 命令
Args:
command: GDB 命令
Returns:
str: GDB 命令执行结果
"""
try:
user_input = command# input("gdb> ")
output = execute_gdb_shell(user_input)
return output
except Exception as e:
return f"execute_gdb tool error: {str(e)}"
app.run(transport=transport)
except ImportError:
print("Error: FastMCP create failed for MCP server")
exit(1)
def main():
"""主函数"""
parser = argparse.ArgumentParser(description="Ubuntu MCP Server")
parser.add_argument(
"--stdio",
action="store_true",
help="Run in stdio mode (default: HTTP/SSE mode)"
)
parser.add_argument(
"--host",
default="0.0.0.0",
help="Host to listen on (HTTP/SSE mode only)"
)
parser.add_argument(
"--port",
type=int,
default=8890,
help="Port to listen on (HTTP/SSE mode only)"
)
args = parser.parse_args()
if args.stdio:
transport = "stdio"
else:
print(f"Starting Ubuntu MCP Server on http://{args.host}:{args.port}")
transport = "streamable-http"
try:
run_mcp_server(transport=transport, host=args.host, port=args.port)
except KeyboardInterrupt:
print(f"Error: MCP server stopped by user")
exit(0)
if __name__ == "__main__":
main()
接下来就是客户端代码
from langchain.tools import tool
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.tools import BaseTool
import asyncio
from typing import List, Tuple
async def get_ubuntu_tools_async(
host: str = 'localhost',
port: int = 8890
) -> Tuple[List[BaseTool], MultiServerMCPClient]:
client = MultiServerMCPClient({
"ubuntu": {
"transport": "http",
"url": f"http://{host}:{port}/mcp"
}
})
tools = await client.get_tools()
return tools, client
def get_ubuntu_tools_sync(
host: str = 'localhost',
port: int = 8890
) -> Tuple[List[BaseTool], MultiServerMCPClient]:
return asyncio.run(get_ubuntu_tools_async(host, port))
# 全局工具实例
sync_ubuntu_tools, sync_ubuntu_client = get_ubuntu_tools_sync()
if __name__ == "__main__":
tools, client = asyncio.run(get_ubuntu_tools_async())
print(tools)
以及我们定义的prompt
from langchain_core.messages import HumanMessage
from core.tools.python_tool import sync_tools
from core.tools.ubuntu_tool import sync_ubuntu_tools
from core.agent.main_agent import Agent
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
async def log_tool_call(request, handler):
tool_name, command = "", ""
if hasattr(request, 'tool_call'):
tool_name = request.tool_call["name"]
if hasattr(request.tool_call, "args") and hasattr(request.tool_call["args"], "command"):
command = request.tool_call["args"]["command"]
print(f"tool name: {tool_name} , args: {command}", end=" ")
res = await handler(request)
print(f"return: {res.content[0]}")
return res
class CTFPwnAgent(Agent):
def __init__(self):
super().__init__()
self.name = 'CTFPwnAgent'
self.model_name = self.config.get('llm.ctf_pwn_agent.model_name') or self.model_name
self.api_key = self.config.get('llm.ctf_pwn_agent.api_key') or self.api_key
self.base_url = self.config.get('llm.ctf_pwn_agent.base_url') or self.base_url
self.tools = self.tools + sync_tools + sync_ubuntu_tools
self.middleware = self.middleware + [log_tool_call]
self.init_model()
self.init_agent()
async def arun(self, task: str):
self.message = []
self.message.append(HumanMessage(content=task, name=self.name))
response = await self.agent.ainvoke({"messages": self.message})
return response['messages'][-1].content
def init_system_prompt(self) -> str:
system_prompt = """
你是一名专业的CTF Pwn题目安全自动化解题Agent，目标是高效完成CTF中的Pwn题目。请严格遵循以下步骤：
请按照以下逻辑执行Pwn题目解题操作：
1. 分析目标程序（如ELF、PE等二进制文件）或远程服务，识别可能的Pwn漏洞类型（如栈溢出、堆溢出、格式化字符串漏洞、UAF、ROP、ret2libc、ret2syscall、整数溢出、use-after-free、double free等）
2. 基于当前进度，生成下一步具体操作指令（包括Payload构造、工具调用（如checksec、gdb、pwntools、IDA Pro、Ghidra等）、参数测试、代码审计目标等，直接使用提供的工具执行）
3. 若当前进度存在错误操作，给出修正后的操作方案
注意：
- 输出仅包含必要的操作指令，避免冗余
- 复杂Pwn漏洞分步骤给出可执行指令
- 每次操作前都需要先进行思考
- 工具执行后需要记录相关观察结果
- 基于观察结果调整后续解题策略
"""
return system_prompt
Driver Analyse Agent
安装ida auto mcp服务，作为我们的工具提供给智能体使用
C:\Users\xxx\Downloads\ida-auto-mcp>pip install -e .
Obtaining file:///C:/Users/xxx/Downloads/ida-auto-mcp
Installing build dependencies ... done
Checking if build backend supports build_editable ... done
Getting requirements to build editable ... done
Preparing editable metadata (pyproject.toml) ... done
Building wheels for collected packages: ida-auto-mcp
Building editable for ida-auto-mcp (pyproject.toml) ... done
Created wheel for ida-auto-mcp: filename=ida_auto_mcp-1.0.0-0.editable-py3-none-any.whl size=7640 sha256=a59aacbaac2a1bd09bcf056577102c28d807205a06a34d747a6e205daf1b6402
Stored in directory: C:\Users\xxx\AppData\Local\Temp\pip-ephem-wheel-cache-qw4pgf3z\wheels\dc\72\ed\47b21cd68bffbd9ed9a1e64716d63aa4346e163a7a33f1d6eb
Successfully built ida-auto-mcp
Installing collected packages: ida-auto-mcp
Successfully installed ida-auto-mcp-1.0.0
安装ida python库
C:\Users\xxx\Downloads\ida-auto-mcp>pip install "D:\Program Files\IDA Professional 9.3\idalib\python"
Processing d:\program files\ida professional 9.3\idalib\python
Installing build dependencies ... done
Getting requirements to build wheel ... done
Preparing metadata (pyproject.toml) ... done
Building wheels for collected packages: idapro
Building wheel for idapro (pyproject.toml) ... done
Created wheel for idapro: filename=idapro-0.0.2-py3-none-any.whl size=4290 sha256=0b967ddf3547ae16286bffd68933bf866114ec029e97b6fb4a3c84c177d9a6e3
Stored in directory: C:\Users\xxx\AppData\Local\Temp\pip-ephem-wheel-cache-nyd8vj91\wheels\f5\ad\e4\6202892ab27edd68456d3b2765f625d3d08773be959ffa0d19
Successfully built idapro
Installing collected packages: idapro
Attempting uninstall: idapro
Found existing installation: idapro 0.0.2
Uninstalling idapro-0.0.2:
Successfully uninstalled idapro-0.0.2
Successfully installed idapro-0.0.2
启动mcp服务，正常启动
(py312) C:\Users\xxx\Downloads\ida-auto-mcp>python -m ida_auto_mcp --ida-dir "D:\Program Files\IDA Professional 9.3"
场景：
在已经搭建好的driver analyse自动化驱动分析平台（原理是通过符号执行判断漏洞位置）所生成的漏洞报告，通常需要人为验证判断，当大批量的驱动报告需要分析，我们采用LLM来辅助验证并给出poc，是一个正确的决定
后续我们可以将智能化拼接到自动化分析平台尾部，全流程无人为干预，这将节省人力成本，一次开发，终生受益
规划：
我们需要处理大批量的驱动文件和对应漏洞报告，最简单的方式是给执行体添加ida工具即可
这里的规划是，编排 -> 执行体 -> 汇总。简单的工作流，三个智能体足以（甚至一个智能体就够了，引入多智能体将不可避免的面对一个问题，重复性，最令人担心的是，这很可能是负向优化）
开发：
- 这里将ida auto mcp作为智能体的工具，这里需要注意，必须将
transport设置为"http"，设置为"stdio"因为其他插件冲突导致运行报错
# ========== IDA Auto Pro 客户端 ==========
class IdaAutoProClient:
def __init__(self):
self.shell = None
self.client = None
async def initialize(self):
self.mcp_server_start()
await asyncio.sleep(1)
self.client = MultiServerMCPClient({
"ida": {
"transport": "http",
"url": "http://127.0.0.1:8765/mcp"
}
})
def mcp_server_start(self):
python_path = r"python.exe path"
self.shell = subprocess.Popen(
[
python_path,
"-m", "ida_auto_mcp",
"--transport", "http",
"--port", "8765",
"--ida-dir", r"IDA Professional 9.3 path"
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
def mcp_server_stop(self):
if self.shell:
self.shell.terminate()
self.shell.wait()
async def get_tools(self):
if not self.client:
raise Exception("please await initialize() first")
return await self.client.get_tools()
async def close(self):
if self.client:
await self.client.aclose()
self.mcp_server_stop()
- 接下来，编写
ExecutionAgent智能体，用于分析单个任务。这里发现在极少数情况下，调用工具的参数会被污染导致整个程序崩溃，可以在tool_after_handle做具体限制
# ========== ExecutionAgent ==========
@before_model
def tool_before_handle(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
#print("tool_before_handle args:\n", state["messages"][-1].content)
pass
@after_model
def tool_after_handle(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
# print("tool_after_handle args:\n", state["messages"][-1].tool_calls)
for tool_call in state["messages"][-1].tool_calls: # 处理函数调用
pass
class ExecutionAgent(Agent):
def __init__(self, tools: List[BaseTool] = None):
super().__init__()
self.name = "ExecutionAgent"
self.tools = tools or []
self.middleware.append(tool_before_handle)
self.middleware.append(tool_after_handle)
self.init_model()
self.init_agent()
async def run(self, task: str) -> str:
self.message = []
self.message.append(AIMessage(content=task, name=self.name))
response = await self.agent.ainvoke({"messages": self.message})
content = response['messages'][-1].content
return content
def init_system_prompt(self) -> str:
"""初始化系统提示"""
return """
你是“驱动漏洞验证与POC开发专家”，核心目标是基于用户提供的驱动文件和漏洞报告，完成漏洞存在性验证、可利用性分析，并为可利用漏洞输出函数调用链条、函数伪代码摘要及最小化POC伪代码。
# 任务拆解：
1. **漏洞存在性验证**：使用IDA工具分析驱动文件，对照漏洞报告中的漏洞描述（如漏洞类型、涉及函数、触发条件等），判断漏洞是否真实存在。若不存在，直接排除该漏洞。
2. **可利用性分析**：针对存在的漏洞，评估是否具备实际利用条件（如是否存在权限限制绕过、是否能控制关键内存地址、是否无有效防护机制等）。若不可利用，排除该漏洞。
3. **可利用漏洞输出**：对通过前两步的漏洞，输出以下内容：
- 函数调用链条：清晰列出从触发点到漏洞核心函数的调用路径（如`用户输入→IOCTL调用→驱动处理函数X→漏洞函数Y`）
- 函数伪代码摘要：提炼漏洞核心函数的关键代码
- 最小化POC伪代码：仅保留关键步骤（包括payload构造、API调用逻辑，重复性基础代码无需编写），示例框架如下
```c
// POC伪代码示例
HANDLE hDriver = CreateFileA("\\\\.\\TargetDriver", GENERIC_READ|GENERIC_WRITE, 0, NULL, OPEN_EXISTING, 0, NULL);
char payload[0x1000]; // 构造溢出/越界等恶意payload
memset(payload, 'A', sizeof(payload));
DWORD bytesReturned;
DeviceIoControl(hDriver, 0x12345678, payload, sizeof(payload), NULL, 0, &bytesReturned, NULL);
CloseHandle(hDriver);
```
# 约束规则
- **必做**：所有分析需基于IDA对驱动文件的逆向结果，不可主观臆断；POC伪代码需聚焦核心利用逻辑，避免冗余
- **禁止**：不可虚构漏洞存在性或可利用性；不可输出与漏洞无关的函数或代码
- **注意**：如果不存在漏洞报告，则自行分析驱动文件，判断是否存在可利用漏洞
# 输入输出规范
- **输入**：需明确驱动文件路径、漏洞报告JSON文件路径
- **输出**：按“漏洞ID→存在性验证结果→可利用性分析→函数调用链条→伪代码摘要→POC伪代码”的结构组织，若无可利用漏洞，需明确说明“未发现可利用的驱动漏洞”。
"""
- 接下来，编写
ReportAgent智能体，用于汇总所有任务结果，并给出报告
# ========== ReportAgent ==========
@tool
def write_mdown_report(path: str, report: str) -> str:
"""将Markdown格式的漏洞分析报告写入文件
Args:
path: 目标文件路径
report: 漏洞分析报告内容
Returns:
str: 操作结果
"""
with open(path, "w") as f:
result = f.write(report)
if result == len(report):
return "success"
else:
return "failed"
return "error"
class ReportAgent(Agent):
def __init__(self):
super().__init__()
self.name = "ReportAgent"
self.tools = [write_mdown_report]
self.init_model()
self.init_agent()
def init_system_prompt(self) -> str:
return """
# 角色设定
你是漏洞分析报告整理专家，核心职责是将所有漏洞分析内容整合为结构清晰、内容完整的Markdown格式报告。
# 核心规则
## 必做事项
1. 完整性检查：必须覆盖所有输入的Agent输出内容，不得遗漏任何漏洞ID及对应的分析信息
2. 结构化整合：严格按照“漏洞ID→存在性验证结果→可利用性分析→函数调用链条→伪代码摘要→POC伪代码”的固定顺序组织每个漏洞的信息
3. 异常处理：若某漏洞无可利用性，需在对应位置明确标注“未发现可利用的驱动漏洞”
4. 格式规范：输出必须为Markdown格式，使用二级标题（##）标注每个漏洞模块，子项使用三级标题（###）或列表呈现。
## 约束条件
1. 禁止篡改原始Agent输出的核心结论（如存在性验证结果、可利用性判断）
2. 禁止添加未经Agent输出确认的主观分析内容
3. 若多个Agent输出存在重复漏洞ID，需合并相同ID的信息，优先保留详细程度更高的分析内容
4. 若Agent输出格式混乱，需先梳理关键信息再按要求整合，不得直接复制杂乱内容。
# 输入处理
1. 读取顺序：按输入中Agent输出的先后顺序逐一处理每个漏洞条目
2. 异常处理：若某Agent输出缺少某字段（如无POC伪代码），使用pass占位，不中断整合流程。
# 执行流程
1. 信息提取：遍历所有Agent输出内容，按Agent分析驱动文件分组，提取每个对应的存在性验证结果、可利用性分析、函数调用链条、伪代码摘要、POC伪代码
2. 内容整合：对每个漏洞ID，按照“漏洞ID→存在性验证结果→可利用性分析→函数调用链条→伪代码摘要→POC伪代码”的结构拼接信息，若无可利用性则替换为指定说明
3. 格式优化：使用Markdown二级标题标注每个漏洞模块，子项用三级标题或有序列表呈现，确保层次清晰
4. 最终校验：检查是否所有Agent输出均已整合，格式是否符合要求，无遗漏或错误后输出报告
# 输出规范
1. 结构框架：
- 报告标题：# 驱动名称：[驱动文件名]
- 漏洞分析模块：## 漏洞ID: [具体ID]
- ### 存在性验证结果：[具体结果]
- ### 可利用性分析：[具体分析/未发现可利用的驱动漏洞]
- ### 函数调用链条：[具体链条]
- ### 伪代码摘要：[具体摘要]
- ### POC伪代码：[具体代码/pass]
2. 标签使用：仅使用Markdown原生标签（标题、列表、代码块）
3. 语言风格：正式、客观，避免口语化表达
4. 字数限制：每个漏洞模块字数控制字数，POC伪代码可单独占块不受字数限制。
"""
- 接下来，编写
OrchestrationAgent智能体，用于编排任务，下发给具体执行智能体
# ========== OrchestrationAgent ==========
@tool
def dir(path: str) -> str:
"""执行dir命令
Args:
path: 目标文件夹路径
Returns:
str: 命令执行结果
"""
import subprocess
result = subprocess.run(f"dir {path}", shell=True, capture_output=True, text=True)
return result.stdout or result.stderr
class OrchestrationAgent(Agent):
"""编排Agent，统筹大局，管理多个任务"""
def __init__(self):
super().__init__()
self.name = "OrchestrationAgent"
self.tools = self.tools + [dir]
self.init_model()
self.init_agent()
def init_system_prompt(self) -> str:
"""初始化系统提示"""
return """
# 身份定位
你是**驱动-报告关联编排专员**，核心职责是读取目标文件夹内下的sys驱动文件与json报告文件，完成文件关联匹配后生成可执行的任务清单。
# 规则边界
## 必做动作
- 当检测到sys驱动与json报告文件名存在相同标识（如设备ID、版本号）时，必须建立一一对应关系
## 约束条件
- 禁止关联文件名无匹配标识的sys驱动与json报告
- 若文件夹内存在同一驱动的不同版本，仅保留最新修改时间的驱动文件，如example_1.sys与example_1.json，example_2.sys与example_2.json，仅保留较新版本的驱动文件与报告文件。
## 输出规范
需输出JSON格式，每项任务需包含以下2个键，且文件对统一按指定格式填写：
1. "sys"：字符类型，驱动绝对路径+驱动名
2. "report"：字符类型，报告绝对路径+报告名
示例：
[
{
"sys": "C/Windows/System32/drivers/example1.sys",
"report": "C/Windows/System32/drivers/example1.json"
},
{
"sys": "C/Windows/System32/drivers/etc/hiworld.sys",
"report": "C/Windows/System32/drivers/etc/hiworld.json"
}
]
"""
- 最终封装在
DriverAnalyseAgent智能体中，这里构建了一个简单的图结构
# 定义状态结构
class SecurityAnalysisState(TypedDict):
overall_task: str # 总任务
sub_tasks: List[str] # 子任务列表
execution_agent: ExecutionAgent # 执行Agent
task_results: List[str] # 任务结果
success: bool # 是否成功
report: str # 生成的报告内容
# 定义节点函数
def decompose_task_node(state, orchestration_agent: OrchestrationAgent):
"""分解任务节点"""
result =orchestration_agent.run(state["overall_task"])
import ast
state["sub_tasks"] = ast.literal_eval(result)
return state
async def execute_task_node(state):
"""执行任务"""
for task_info in state["sub_tasks"]:
with open(task_info["report"], "r") as f:
report = f.read()
tmp_task = {
"sys": task_info["sys"],
"report": report
}
result = await state["execution_agent"].run(str(tmp_task))
state["task_results"].append(result)
return state
def generate_report_node(state, report_agent: ReportAgent):
"""生成报告节点"""
state["report"] = report_agent.run(", ".join(state["task_results"]))
return state
class DriverAnalyseAgent:
"""驱动分析Agent，整合编排、执行和报告功能"""
def __init__(self):
self.orchestration_agent = OrchestrationAgent()
self.ida_tools = IdaAutoProClient()
self.execution_agent = None
self.report_agent = ReportAgent()
async def initialize(self):
await self.ida_tools.initialize()
tools = await self.ida_tools.get_tools()
self.execution_agent = ExecutionAgent(tools)
self.graph = self._build_graph()
def _build_graph(self):
"""构建LangGraph工作流"""
from langgraph.graph import StateGraph, END
workflow = StateGraph(SecurityAnalysisState) # 创建状态图
# 添加节点
workflow.add_node("decompose_task", lambda state: decompose_task_node(state, self.orchestration_agent))
workflow.add_node("execute_task", execute_task_node)
workflow.add_node("generate_report", lambda state: generate_report_node(state, self.report_agent))
# 添加边
workflow.set_entry_point("decompose_task") # workflow.add_edge(START, "decompose_task")
workflow.add_edge("decompose_task", "execute_task")
workflow.add_edge("execute_task", "generate_report")
workflow.add_edge("generate_report", END)
return workflow.compile() # 编译图
async def run(self, task: str) -> Dict[str, Any]:
"""运行分析任务"""
if not hasattr(self, "graph"):
await self.initialize()
try:
# 初始化状态
initial_state: SecurityAnalysisState = {
"overall_task": task,
"sub_tasks": [],
"execution_agent": self.execution_agent,
"task_results": [],
"success": True,
"report": ""
}
# 执行图
result = await self.graph.ainvoke(initial_state)
return result
except Exception as e:
return {
"overall_task": task,
"sub_tasks": [],
"task_results": [],
"success": False,
"report": str(e)
}
- 测试测试
async def main():
d = DriverAnalyseAgent()
await d.initialize()
task = "C:\\work" # 该目录下是sys驱动文件，以及漏洞报告文件
r = await d.run(task)
print(r)
if __name__ == "__main__":
asyncio.run(main())
- 补充
这里经过大量测试，发现有很大优化空间，举例说明
- 自动化测试平台原理是通过符号执行获取漏洞，漏洞报告相当于Agent的先验知识，会导致Agent陷入先验知识中，需要纠偏
- 提示词仍然有较大优化空间，如，可以加入分析驱动权限校验部分（这部分必须加入，否则我们将会得到很多无法利用的漏洞）
结语
上面还仅仅只是demo，还有很多需要我们去学习，尤其现在AI发展速度极快，所以更应该去追踪前沿的消息，实时跟踪更新自己的知识库。
虽然AI与人类的相遇，就像站台上火车越过的一瞬，但是在这一刻，仍然希望再多看两眼。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
6
0- 0
已为社区贡献1条内容
所有评论(0)