漏洞检测代理：使用Claude Agent SDK构建智能安全分析工具

本教程面向有编程基础的开发者，指导如何使用Claude Agent SDK构建一个自动化漏洞发现代理。通过多轮ClaudeSDKClient会话进行威胁建模，利用内置的Read、Grep、Glob工具驱动代理查找内存安全漏洞，并将查找、分类和报告作为独立的query()调用链接起来，最终输出符合JSON模式的结构化报告。教程涵盖环境设置、威胁建模、查找循环、分类过滤和报告生成，并提供了可执行的实践步骤、常见错误排查及生产环境优化建议。

技术背景

安全团队面临内存安全漏洞检测的挑战：静态分析器误报率高，导致审查者忽略报告；模糊测试需要为每个入口点手动编写测试工具，效率低下。本教程旨在利用Claude Agent SDK构建一个智能代理，自动读取源代码、推理潜在漏洞，并生成可操作的发现结果，从而在攻击者之前发现漏洞。该技术适用于代码审计、安全评估和漏洞研究，主要限制包括需要授权评估、依赖模型能力，以及可能触发API层的网络安全防护。

核心原理

核心原理是利用Claude Agent SDK封装Claude Code的内置工具（Read、Grep、Glob），通过自然语言提示驱动代理自主探索代码库。代理遵循引导-访谈的威胁建模流程：先由模型阅读代码生成初步威胁模型，再通过访谈补充上下文信息，最终形成完整的威胁模型。查找循环中，代理根据威胁模型聚焦高风险区域，使用文件工具进行探索-阅读-推理循环，并依据质量等级标准筛选高价值漏洞。分类步骤独立验证发现，合并重复项并重新推导严重性。最后，无工具query()将分类结果转换为符合JSON模式的结构化报告。

架构分析

系统架构分为四个阶段：威胁建模、查找、分类和报告。威胁建模使用ClaudeSDKClient会话，包含引导和访谈两轮，共享上下文。查找阶段使用一次性query()调用，配置allowed_tools为Read、Grep、Glob，禁用Bash以保持只读。分类阶段使用新的query()，仅允许Read和Grep，独立验证发现。报告阶段使用无工具query()，强制输出JSON。关键设计决策包括：使用cwd限定代理工作目录，通过system_prompt附加参与上下文，利用质量等级标准控制报告质量，以及将分类独立于查找以捕捉过度自信。

完整实战教程

安全团队希望在攻击者之前发现内存安全漏洞，但现有工具难以实现：静态分析器产生大量误报，导致审查者不再阅读；模糊测试器在发现任何问题之前，需要为每个入口点手动编写测试工具。本教程展示如何使用Claude Agent SDK构建一个漏洞发现代理，该代理利用Claude Code内置的Read、Grep和Glob工具读取源代码，推理哪些输入可能破坏内存，并编写审查者可以操作的发现结果。

完成本教程后，您将能够：

先决条件

所需知识：

所需工具：

任何真实目标所需： 授权评估您指向的代码。本笔记本附带一个微型的自包含canary.c，其中植入了漏洞，以便您无需接触生产代码即可端到端运行。

步骤 1：设置环境和参与背景

我们定义一个 ENGAGEMENT_CONTEXT 块，并将其作为 system_prompt 传递给每个代理。它记录了本次评估的范围（由代码所有者授权、隔离的只读沙箱、发现结果将用于负责任披露），以便流程中的每一步都遵循相同的文档化基本规则。对于任何真实目标，请确保这三项声明均为真。

关于网络安全的说明： Claude 在 API 层应用实时网络安全防护。如果您在真实代码库上的工作触发了这些防护，请通过该页面申请网络安全验证计划（CVP）：这是一个免费的基于申请的计划，允许专业人员以最小干扰继续合法的双重用途安全工作。
%%capture
%pip install -U claude-agent-sdk python-dotenv

import json
from collections.abc import AsyncIterator
from pathlib import Path

from dotenv import load_dotenv

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    Message,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    query,
)

load_dotenv()

MODEL_NAME = "claude-opus-4-7"
# This notebook expects to be run from the claude_agent_sdk/ directory
# (Jupyter's default when you open the file from there). The assert makes
# the failure explicit if the kernel was started elsewhere.
TARGET_DIR = Path("vulnerability_detection_agent/canary").resolve()
assert TARGET_DIR.is_dir(), f"run this notebook from claude_agent_sdk/ (got cwd={Path.cwd()})"

ENGAGEMENT_CONTEXT = """\
## Engagement context

This is authorized security research conducted as a defensive security
assessment on a self-contained canary target vendored in this notebook. The
target is read-only source (no execution). Findings are collected for
demonstration and responsible-disclosure workflow testing.
"""


async def collect(stream: AsyncIterator[Message]) -> str:
    """Consume an Agent SDK message stream; print tool calls; return final text.

    Both ``query()`` and ``ClaudeSDKClient.receive_response()`` return an
    ``AsyncIterator[Message]`` that terminates after a ``ResultMessage``.
    This is the same ``async for msg in ...`` loop the other notebooks in this
    series write inline; it is factored out here because this notebook runs
    the loop four times (TM bootstrap, TM interview, find, triage) and the
    ``isinstance`` ladder would otherwise repeat verbatim.
    """
    final = ""
    async for msg in stream:
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, ToolUseBlock):
                    args = str(block.input)
                    args = args if len(args) <= 120 else args[:120] + "...}"
                    print(f"  [tool] {block.name} {args}")
                elif isinstance(block, TextBlock) and block.text.strip():
                    final += block.text
        elif isinstance(msg, ResultMessage) and msg.is_error:
            raise RuntimeError(msg.result)
    return final


print(f"Model: {MODEL_NAME}")

步骤 2：加载金丝雀目标

vulnerability_detection_agent/canary/canary.c 是一个约 45 行的 C 程序，其中故意植入了三个内存安全漏洞（堆缓冲区溢出、栈缓冲区溢出和释放后使用），每个漏洞都可通过输入开头的不同“魔法字节”触发。这些漏洞没有标记；步骤 4 中的查找代理必须像在真实代码中一样通过阅读逻辑来定位它们。当你准备好尝试自己的代码时，将 TARGET_DIR 指向你的代码库。

print((TARGET_DIR / "canary.c").read_text())

第 3 步：对目标进行威胁建模（先引导，再访谈）

威胁模型回答“这个系统可能出什么问题，谁会这么做，哪些结果重要？”——独立于任何特定漏洞。威胁（“攻击者通过不可信的文件解析实现内存破坏”）在补丁后依然存在；漏洞（“第 31 行没有对 len 进行边界检查”）则不会。查找循环寻找漏洞；威胁模型告诉它去哪里寻找，并告诉分类如何评分。

我们在一个 ClaudeSDKClient 会话中用两轮来构建它：

  1. 引导。 Claude 使用内置的 Read 工具阅读代码，并起草模型（上下文、资产、入口点和信任边界、威胁，以及代码无法回答的开放问题）。
  2. 访谈。 应用程序所有者回答开放问题，Claude 完善可能性和影响，然后在目标旁边写入 THREAT_MODEL.md。

将两轮保持在同一个客户端会话中意味着访谈轮可以看到引导轮的工具结果，而无需我们重新发送源代码。在 45 行的金丝雀上，两轮都很薄；这里的重点是输出形状：入口点表（在真实仓库中，你会将其分区到并行的查找代理中）和开放问题列表（引导到访谈的交接）。

(TARGET_DIR / "THREAT_MODEL.md").unlink(missing_ok=True)

TM_SCHEMA = """\
# Threat Model: <system name>
## 1. System context
## 2. Assets
| asset | description | sensitivity |
## 3. Entry points & trust boundaries
| entry_point | description | trust_boundary | reachable_assets |
## 4. Threats
| id | threat | surface | asset | impact | likelihood |
## 5. Open questions
- (things the code alone cannot answer: deployment context, which inputs are
  attacker-controlled in practice, blast radius)
"""

BOOTSTRAP_PROMPT = f"""\
You are bootstrapping a threat model from source code alone; no application
owner is available yet. Read `canary.c` in this directory and emit a draft
threat model in the schema below. Be explicit in section 5 about what you could
NOT determine from the code: those open questions are the agenda for the owner
interview. Do not write any files yet.

## Schema

{TM_SCHEMA}
"""

OWNER_ANSWERS = """\
- Deployment: `canary` is a local CLI that reads a file path from argv; it is
  not network-facing.
- Attacker control: the input file is fully attacker-controlled (think email
  attachment or downloaded file opened by the user).
- Blast radius: the process runs as the invoking user with no sandboxing;
  memory corruption is code execution as that user.
"""

INTERVIEW_PROMPT = f"""\
The application owner has now answered your open questions:

{OWNER_ANSWERS}

Refine the threat model: update likelihood and impact in section 4 using the
owner's answers, resolve every item in section 5 that the answers cover, and
add any new threats the deployment context implies. Keep the same schema, then
write the refined model to `THREAT_MODEL.md` in this directory.
"""

tm_options = ClaudeAgentOptions(
    model=MODEL_NAME,
    cwd=str(TARGET_DIR),
    system_prompt={"type": "preset", "preset": "claude_code", "append": ENGAGEMENT_CONTEXT},
    allowed_tools=["Read", "Write", "Edit"],
    disallowed_tools=["Bash"],
    permission_mode="acceptEdits",
)

async with ClaudeSDKClient(options=tm_options) as tm_agent:
    # collect() fully drains receive_response() through its terminating
    # ResultMessage, so the second query() sees a clean stream.
    await tm_agent.query(BOOTSTRAP_PROMPT)
    draft_tm = await collect(tm_agent.receive_response())
    print("--- bootstrap draft ---\n" + draft_tm + "\n")

    await tm_agent.query(INTERVIEW_PROMPT)
    await collect(tm_agent.receive_response())

tm_path = TARGET_DIR / "THREAT_MODEL.md"
if not tm_path.exists():
    raise RuntimeError("interview agent did not write THREAT_MODEL.md; check the trace above")
threat_model = tm_path.read_text()
print("--- refined THREAT_MODEL.md ---\n" + threat_model)

步骤 4：运行代理查找循环

使用原始 Messages API，这一步需要手写一个 while stop_reason == "tool_use" 循环，并自定义文件工具。Agent SDK 处理了所有这些：我们只需调用一次 query()，并设置 allowed_tools=["Read", "Grep", "Glob"] 和 disallowed_tools=["Bash"]，Claude Code 就会自行运行探索-阅读-推理循环。cwd=str(TARGET_DIR) 将代理指向金丝雀文件，system_prompt={"type": "preset", "preset": "claude_code", "append": ...} 保留 Claude Code 的默认系统提示（其中已告知代理其工作目录），同时附加我们的参与上下文，因此代理无需猜测自身位置。通过禁用 Bash/Write/Edit，代理保持只读状态。

提示中最重要的部分仍然是质量等级标准。 没有它，LLM 漏洞猎人会报告他们能找到的每一个空指针解引用和失败的断言，这些确实是真实的崩溃，但几乎都不可利用。该标准告诉代理应提交哪些崩溃类别（堆/栈溢出、释放后使用、受控地址写入），以及哪些是继续阅读的指示牌。这一块内容在很大程度上决定了报告是安全工程师会采取行动，还是会被忽略。

生产版本会将 "Bash" 添加到 allowed_tools 中，以便代理可以使用 -fsanitize=address 进行编译并确认每次崩溃；这应放在锁定的容器内，因此此笔记本保持只读。

FIND_PROMPT = f"""\
Find memory-safety bugs in the target source tree using the file tools
available to you.

## Threat model

Focus on the entry points and threats identified here; you do not need to
re-derive them.

{threat_model}

## Quality tiers: what to report

**HIGH VALUE (report these):**
- heap-buffer-overflow (especially WRITE)
- heap-use-after-free / double-free
- stack-buffer-overflow
- global-buffer-overflow

**LOW VALUE (note but keep looking):**
- assertion failures (clean abort, no corruption)
- stack exhaustion from recursion (DoS only)
- null-pointer deref at fixed small offsets

## Output

For each HIGH VALUE finding emit a block:

<finding>
<id>F-NN</id>
<file>path:line</file>
<category>heap-buffer-overflow | stack-buffer-overflow | use-after-free | ...</category>
<description>one paragraph: root cause, attacker control, trigger condition</description>
</finding>
"""

find_options = ClaudeAgentOptions(
    model=MODEL_NAME,
    cwd=str(TARGET_DIR),
    system_prompt={"type": "preset", "preset": "claude_code", "append": ENGAGEMENT_CONTEXT},
    allowed_tools=["Read", "Grep", "Glob"],
    disallowed_tools=["Bash"],
)

findings_text = await collect(query(prompt=FIND_PROMPT, options=find_options))
print("\n" + findings_text)

第5步：对原始发现进行分类

find 代理针对召回率进行了调优，因此其输出通常包含重复项（从两条路径到达同一根因），偶尔也会出现不成立的发现。分类就是过滤器：新的 query() 会重新读取代码，验证每个发现是否与实际代码行相符，按根因合并重复项，并根据威胁模型中跨信任边界的可达性重新推导严重性。我们特意不让分类继承 find 代理的严重性评分；独立重新推导是捕捉过度自信的廉价方法。

TRIAGE_PROMPT = f"""\
Triage these findings against the source in this directory and the threat model
below. For each: verify it is real (cite the line), derive severity from
reachability across the trust boundaries in the threat model, and collapse
duplicates by root cause.

## Threat model

{threat_model}

## Raw findings

{findings_text}
"""

triage_options = ClaudeAgentOptions(
    model=MODEL_NAME,
    cwd=str(TARGET_DIR),
    system_prompt={"type": "preset", "preset": "claude_code", "append": ENGAGEMENT_CONTEXT},
    allowed_tools=["Read", "Grep"],
    disallowed_tools=["Bash"],
)

triaged_text = await collect(query(prompt=TRIAGE_PROMPT, options=triage_options))
print("\n" + triaged_text)

步骤 6：生成结构化报告

下游系统（问题跟踪器、仪表盘、SIEM）需要结构化数据。最后，我们使用无工具的 query() 将分类后的发现转换为符合明确模式的 JSON。我们将每个字段标记为必填，并使用 null 表示“不适用”，这样模型就不必猜测可选性；在生产环境中，您可以使用 jsonschema 进行验证，并在失败时重试。

# Every key is in `required` so the model never silently drops a field; values
# may be null when a field is not applicable (e.g., no recommendation yet).
REPORT_SCHEMA = {
    "type": "object",
    "properties": {
        "findings": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "id": {"type": ["string", "null"]},
                    "category": {"type": ["string", "null"]},
                    "severity": {"type": "string", "enum": ["critical", "high", "medium", "low"]},
                    "file": {"type": ["string", "null"]},
                    "description": {"type": ["string", "null"]},
                    "recommendation": {"type": ["string", "null"]},
                },
                "required": ["id", "category", "severity", "file", "description", "recommendation"],
            },
        }
    },
    "required": ["findings"],
}

REPORT_PROMPT = f"""\
Convert the triaged findings below into strict JSON conforming to this schema.
Every field is required; use null for not-applicable. Respond with JSON only,
no surrounding prose or code fences.

## Schema

{json.dumps(REPORT_SCHEMA, indent=2)}

## Triaged findings

{triaged_text}
"""

report_options = ClaudeAgentOptions(
    model=MODEL_NAME,
    system_prompt={"type": "preset", "preset": "claude_code", "append": ENGAGEMENT_CONTEXT},
    allowed_tools=[],
)

report_json = await collect(query(prompt=REPORT_PROMPT, options=report_options))
raw = report_json.strip()
if raw.startswith("```"):
    raw = raw.split("\n", 1)[1].rsplit("```", 1)[0]
try:
    report = json.loads(raw)
except json.JSONDecodeError as e:
    print(f"[report agent did not return clean JSON: {e}]\n")
    print(raw)
else:
    print(json.dumps(report, indent=2))

总结与后续步骤

你已经使用 Agent SDK 构建了完整的威胁建模、发现、分类、报告流水线：一个多轮 ClaudeSDKClient 会话用于威胁建模，三个一次性 query() 调用用于其余部分，Claude Code 的内置文件工具负责探索。需要记住的关键模式：

进一步探索

常见问题

常见错误包括：未正确设置TARGET_DIR导致断言失败；未安装依赖或未加载环境变量；代理未写入THREAT_MODEL.md（检查权限和提示）；报告代理返回非JSON（需清理代码围栏）；触发API网络安全防护（需申请CVP）。排查方法：检查cwd路径、确认API密钥有效、查看工具调用日志、验证输出格式。生产环境优化：在沙箱容器中启用Bash以支持编译和动态分析；使用asyncio.gather并行处理多个入口点；使用jsonschema验证报告并重试失败。

企业应用扩展

进一步探索包括：使用Claude Code Security托管版本；扩展到真实仓库时，在沙箱中启用Bash并添加-fsanitize=address编译确认；将报告接入问题跟踪器，映射到SARIF或自定义工单模式。

总结

本教程面向有编程基础的开发者，指导如何使用Claude Agent SDK构建一个自动化漏洞发现代理。通过多轮ClaudeSDKClient会话进行威胁建模，利用内置的Read、Grep、Glob工具驱动代理查找内存安全漏洞，并将查找、分类和报告作为独立的query()调用链接起来，最终输出符合JSON模式的结构化报告。教程涵盖环境设置、威胁建模、查找循环、分类过滤和报告生成，并提供了可执行的实践步骤、常见错误排查及生产环境优化建议。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程