1. Claude Opus 4.6 发布后,Python Agent 安全审计流水线怎么跑通
Claude Opus 4.6 是 Anthropic 刚上线的新一代旗舰模型,API 模型名claude-opus-4-6,上下文窗口首次拉到 100 万 tokens,在 Terminal-Bench 2.0 上拿到 65.4% 的代码能力评分。它能做什么?简单说,你把一整个项目的代码库丢进去,它能在里面翻出跨文件的调用链问题。适合谁?适合手里有一堆遗留代码、又不想雇一队安全工程师逐行看的中小团队,或者想自己搭一条自动化审计链路的独立开发者。
这次发布里最抓眼球的一个细节,是 Anthropic 红队在沙盒环境里给 Claude 配了 Python、调试器、模糊测试器,没给任何特定指令,让它自己在开源代码里找漏洞,结果找出 500 多个此前没人发现的安全问题,每一个都经过人工验证。GhostScript 里那个 PDF 处理漏洞,传统模糊测试和手动分析都没抓到,Claude 是翻 Git 提交记录才定位到的。
这个场景对普通开发者最大的启发不是"模型多强",而是"这条链路我可以自己复现"。你不需要 Anthropic 的内部沙盒,用一套统一的 API 通道 + 一个 Python Agent 循环,就能在本地跑出一条从模型调用到结果落盘的完整审计流水线。我下面要交付的就是这条链路:可复制的 Agent 配置片段、漏洞扫描脚本、API 调用验证步骤,以及跑不通时怎么排查。
整条链路的核心思路是:Agent 拿到一个代码目录,按文件切块,把每个块连同"找安全漏洞"的指令发给 Claude Opus 4.6,模型返回结构化的问题列表,Agent 把结果写进 JSON 落盘,最后汇总去重。听起来简单,但真正跑起来会踩几个坑:上下文怎么切、返回的 JSON 怎么稳定解析、并发请求怎么控速、长上下文计费怎么不爆。这些我都会在对应章节里给具体参数。
先说清楚一件事:这条流水线是"辅助审计",不是"替代审计"。模型找出来的东西需要人复核,尤其是它给的修复建议,别直接往生产代码上套。但作为第一轮粗筛,把几千个文件缩到几十个可疑点,它的效率比人高太多。
2. TaoToken 统一 API 通道前置准备:Key、Base URL 与模型 ID
在写 Agent 之前,先把通道打通。TaoToken 在这里扮演的角色是统一 API 入口:你不需要为每个模型单独维护一套鉴权、计费、限流逻辑,一个 Key 走一个 Base URL,模型 ID 换一下就能切模型。这对审计场景特别有用,因为有时候你想用 Claude Opus 4.6 做深度分析,有时候想用更便宜的模型做初筛,通道不变,只换 model 字段。
前置准备分三件事:拿 Key、确认 Base URL、确认模型 ID。
拿 Key 的入口在控制台的 API Keys 页面,登录后新建一个 Key,复制出来存到环境变量里,别硬编码进脚本。Base URL 用https://taotoken.net/api,注意这个地址不带任何查询参数,直接作为 OpenAI 兼容客户端的base_url使用。模型 ID 就是claude-opus-4-6,这是 Anthropic 官方公布的 API 模型名,直接填。
这里有个容易搞混的点:Base URL 到底要不要带/v1。不同客户端处理方式不一样。用 OpenAI 官方 Python SDK 时,base_url填https://taotoken.net/api,SDK 会自动拼/chat/completions;如果你用的是裸requests发请求,那完整路径就是https://taotoken.net/api/v1/chat/completions。我建议统一用 SDK,少踩拼接的坑。
环境变量这样设,Linux/macOS 下:
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"Windows PowerShell 下:
$env:TAOTOKEN_API_KEY="sk-你的key" $env:TAOTOKEN_BASE_URL="https://taotoken.net/api"设完之后验证一下环境变量有没有生效:
echo $TAOTOKEN_API_KEY能打印出你的 Key 就对了。如果打印出来是空的,说明当前 shell 会话没加载到,检查你是写进了.bashrc还是只在当前终端 export。
关于模型选择,这里给一个对照,方便你按预算和任务复杂度切:
| 模型 ID | 适用场景 | 相对成本 |
|---|---|---|
| claude-opus-4-6 | 深度审计、跨文件调用链分析 | 高 |
| claude-sonnet-4-5 | 单文件初筛、批量扫描 | 中 |
| claude-haiku-4-5 | 格式化、去重、结果归类 | 低 |
审计流水线的典型用法是三层:Haiku 做文件分类和预处理,Sonnet 做单文件初筛,Opus 4.6 只对初筛出来的可疑文件做深度分析。这样既用上了 Opus 4.6 的推理能力,又不至于让成本失控。如果你只是想先跑通链路,全用 Opus 4.6 也行,跑通再优化。
还有一点,Opus 4.6 默认"想得比较深",在简单任务上会显得慢。Anthropic 官方提到可以用/effort把思考强度从 high 调到 medium。在 API 调用里,这个对应的是请求参数里的思考预算控制,具体字段名以你用的 SDK 版本为准,跑通基础链路后再去调这个。
3. 可复制的 Agent 配置片段与漏洞扫描脚本
这一节是整篇的核心,给的是能直接复制粘贴跑起来的代码。我按"配置 → 扫描器 → 主循环"三层拆开,每层都能单独测试。
先建项目结构:
mkdir -p audit-agent/{config,scanner,results} cd audit-agent python -m venv venv source venv/bin/activate pip install openai python-dotenv配置文件用 JSON,放在config/agent.json。这个文件把通道、模型、并发、切块参数都集中管理,改参数不用动代码:
{ "api": { "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "timeout": 120, "max_retries": 3 }, "models": { "deep_audit": "claude-opus-4-6", "triage": "claude-sonnet-4-5", "classify": "claude-haiku-4-5" }, "scan": { "chunk_size_lines": 400, "chunk_overlap_lines": 40, "max_concurrency": 4, "target_extensions": [".py", ".js", ".go", ".java"], "exclude_dirs": ["node_modules", ".git", "venv", "__pycache__"] }, "output": { "results_dir": "results", "format": "jsonl" } }chunk_size_lines设 400、overlap 设 40,是因为安全漏洞经常跨函数边界,纯按行切会把上下文切断,留 40 行重叠能让模型看到跨块的调用关系。max_concurrency设 4 是保守值,先跑通再往上加,加太快容易触发限流。
扫描器scanner/scan.py,负责遍历目录、切块、过滤:
import os import json from pathlib import Path def load_config(path="config/agent.json"): with open(path, "r", encoding="utf-8") as f: return json.load(f) def iter_target_files(root, cfg): exts = set(cfg["scan"]["target_extensions"]) excludes = set(cfg["scan"]["exclude_dirs"]) for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d not in excludes] for name in filenames: if Path(name).suffix in exts: yield Path(dirpath) / name def chunk_file(path, cfg): size = cfg["scan"]["chunk_size_lines"] overlap = cfg["scan"]["chunk_overlap_lines"] with open(path, "r", encoding="utf-8", errors="ignore") as f: lines = f.readlines() chunks = [] start = 0 while start < len(lines): end = min(start + size, len(lines)) chunks.append({ "file": str(path), "start_line": start + 1, "end_line": end, "content": "".join(lines[start:end]) }) if end == len(lines): break start = end - overlap return chunks主循环scanner/agent.py,负责调模型、解析返回、落盘:
import os import json import asyncio from openai import AsyncOpenAI from scanner.scan import load_config, iter_target_files, chunk_file SYSTEM_PROMPT = """你是一名安全审计专家。分析给定的代码片段,找出安全漏洞。 只返回 JSON 数组,每个元素包含字段: - type: 漏洞类型(如 sql_injection, xss, path_traversal, hardcoded_secret) - severity: high / medium / low - line: 行号(相对片段起始行) - evidence: 触发漏洞的代码片段 - fix: 修复建议 如果没有发现漏洞,返回空数组 []。不要输出任何解释文字。""" async def audit_chunk(client, model, chunk): user_msg = f"文件: {chunk['file']}\n起始行: {chunk['start_line']}\n\n```\n{chunk['content']}\n```" resp = await client.chat.completions.create( model=model, messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_msg} ], temperature=0 ) raw = resp.choices[0].message.content.strip() if raw.startswith("```"): raw = raw.split("```")[1] if raw.startswith("json"): raw = raw[4:] try: findings = json.loads(raw) except json.JSONDecodeError: findings = [{"type": "parse_error", "raw": raw[:500]}] for f in findings: f["file"] = chunk["file"] f["chunk_start"] = chunk["start_line"] return findings async def main(target_dir): cfg = load_config() client = AsyncOpenAI( base_url=cfg["api"]["base_url"], api_key=os.environ[cfg["api"]["api_key_env"]], timeout=cfg["api"]["timeout"], max_retries=cfg["api"]["max_retries"] ) model = cfg["models"]["deep_audit"] sem = asyncio.Semaphore(cfg["scan"]["max_concurrency"]) out_path = os.path.join(cfg["output"]["results_dir"], "findings.jsonl") async def worker(chunk): async with sem: return await audit_chunk(client, model, chunk) tasks = [] for fp in iter_target_files(target_dir, cfg): for chunk in chunk_file(fp, cfg): tasks.append(worker(chunk)) with open(out_path, "w", encoding="utf-8") as out: for coro in asyncio.as_completed(tasks): findings = await coro for f in findings: out.write(json.dumps(f, ensure_ascii=False) + "\n") print(f"完成,结果写入 {out_path}") if __name__ == "__main__": import sys asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "."))跑起来:
python -m scanner.agent /path/to/your/codebase这套代码的关键设计点:temperature=0保证结果稳定可复现;asyncio.Semaphore控并发;返回内容先剥 markdown 代码块再json.loads,因为模型经常把 JSON 包在 ```json 里;解析失败不抛异常,而是记一条parse_error落盘,方便你事后看是哪个片段让模型输出跑偏了。
4. 验证请求与成功结果:从单次调用到结果落盘
代码写完别急着扫整个代码库,先用一个最小请求验证通道是通的。这一步能帮你把"通道问题"和"代码问题"分开。
写个verify.py:
import os from openai import OpenAI client = OpenAI( base_url="https://taotoken.net/api", api_key=os.environ["TAOTOKEN_API_KEY"] ) resp = client.chat.completions.create( model="claude-opus-4-6", messages=[{"role": "user", "content": "回复两个字:通了"}], temperature=0 ) print(resp.choices[0].message.content) print("model:", resp.model) print("usage:", resp.usage)跑python verify.py,预期输出类似:
通了 model: claude-opus-4-6 usage: CompletionUsage(prompt_tokens=12, completion_tokens=4, total_tokens=16)看到model字段回显claude-opus-4-6,说明通道和模型 ID 都对。如果model回显的是别的名字,说明你的请求被路由到了别的模型,检查 model 字段拼写。
通道验证通过后,拿一个故意有漏洞的小文件测扫描器。建test_vuln.py:
import sqlite3 def get_user(username): conn = sqlite3.connect("app.db") cursor = conn.cursor() query = "SELECT * FROM users WHERE name = '" + username + "'" cursor.execute(query) return cursor.fetchall() API_KEY = "sk-hardcoded-1234567890"跑:
python -m scanner.agent ./test_dir预期在results/findings.jsonl里看到至少两条记录,一条sql_injection,一条hardcoded_secret。内容大概长这样:
{"type": "sql_injection", "severity": "high", "line": 6, "evidence": "query = \"SELECT * FROM users WHERE name = '\" + username + \"'\"", "fix": "使用参数化查询 cursor.execute('SELECT * FROM users WHERE name = ?', (username,))", "file": "./test_dir/test_vuln.py", "chunk_start": 1} {"type": "hardcoded_secret", "severity": "high", "line": 10, "evidence": "API_KEY = \"sk-hardcoded-1234567890\"", "fix": "将密钥移至环境变量或密钥管理服务", "file": "./test_dir/test_vuln.py", "chunk_start": 1}看到这两条,说明整条链路通了:文件遍历 → 切块 → 模型调用 → JSON 解析 → 落盘。这时候再拿真实代码库跑,把max_concurrency从 4 慢慢往上调,观察有没有限流报错。
结果落盘后做个简单汇总,按严重级别和类型统计:
python -c " import json from collections import Counter c = Counter() with open('results/findings.jsonl') as f: for line in f: d = json.loads(line) c[(d.get('severity'), d.get('type'))] += 1 for k, v in c.most_common(): print(k, v) "这个汇总能让你一眼看出代码库的主要风险集中在哪类问题上。如果parse_error占比超过 5%,说明你的 system prompt 需要再收紧,或者切块粒度要调。
5. 常见报错排查:401、local proxy failed、reading choices、OAuth
跑这条链路时,报错基本集中在四类。我按实际遇到的频率排。
401 Unauthorized。最常见的原因是 Key 没读到或者读错了。先确认环境变量:
echo $TAOTOKEN_API_KEY如果为空,说明 export 没生效。如果非空但还是 401,检查 Key 有没有多余空格,或者是不是复制的时候带上了引号。还有一种情况是你把 Key 写进了.env文件但代码里没调load_dotenv()。用python-dotenv的话,在脚本开头加:
from dotenv import load_dotenv load_dotenv()local proxy failed / connection error。这类报错通常是网络层的问题,不是鉴权问题。先确认 Base URL 拼对了:https://taotoken.net/api,不要多写/v1也不要少写。如果你在代码里用了requests并手动拼了完整 URL,确认是https://taotoken.net/api/v1/chat/completions。另外检查你的timeout设置,Opus 4.6 在深度思考模式下响应可能超过 60 秒,timeout 设 120 比较稳。如果报错信息里出现proxy字样,检查你的 shell 里有没有残留的HTTP_PROXY/HTTPS_PROXY环境变量,有的话 unset 掉:
unset HTTP_PROXY HTTPS_PROXYreading 'choices' / KeyError: 'choices'。这个报错说明你拿到的响应体里没有choices字段,通常是请求本身失败了但你没检查状态码。在audit_chunk里加一层防御:
if not resp.choices: return [{"type": "empty_response", "file": chunk["file"]}]更常见的原因是模型返回了错误对象,比如限流或者参数错误,但 SDK 没抛异常而是返回了一个空结构。打印完整响应体看:
print(resp.model_dump_json(indent=2))如果看到error字段,按里面的 message 对症处理。限流的话把max_concurrency降到 2,或者加个退避重试。
OAuth / authentication 相关报错。如果你用的是某些 CLI 工具(比如 Claude Code 或 Codex 的本地客户端),它们可能走的是 OAuth 流程而不是 API Key。这类工具接入 TaoToken 时,需要把鉴权方式从 OAuth 切到 API Key,配置三件套写全:Base URL 填https://taotoken.net/api,Key 填你的 TaoToken Key,Model ID 填claude-opus-4-6。三个字段缺一个都会报鉴权失败。如果你用的是 Cline 或 CC Switch 这类插件,在设置里找 "API Provider" 选 OpenAI Compatible,然后填这三个字段。
排查顺序建议固定成:先跑verify.py确认通道,再跑单文件扫描确认代码逻辑,最后才跑全量。这样报错范围能快速缩小到某一层。
6. 把审计流水线接进日常:从一次性脚本到可复用工具
链路跑通之后,下一步是让它变成你日常能用的东西,而不是每次手动敲命令。
第一个改造是把结果落盘格式从 JSONL 换成带时间戳的目录,方便对比不同版本的扫描结果:
import datetime run_id = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") out_dir = os.path.join(cfg["output"]["results_dir"], run_id) os.makedirs(out_dir, exist_ok=True)这样每次扫描结果独立存放,你可以用diff对比两次提交之间新增了哪些风险点。
第二个改造是加一个"只扫变更文件"的模式。全量扫描一个中型项目可能要几十分钟,但日常开发中你只关心这次改了哪些文件。用git diff --name-only HEAD~1拿到变更列表,只对这些文件跑扫描:
git diff --name-only HEAD~1 | grep -E '\.(py|js|go|java)$' > changed.txt python -m scanner.agent --files changed.txt在agent.py里加个--files参数,读文件列表而不是遍历目录,能省掉大量重复扫描。
第三个改造是接进 CI。在 GitHub Actions 或 GitLab CI 里加一个 job,每次 PR 触发时跑增量扫描,把findings.jsonl作为 artifact 上传。如果发现severity: high的记录超过阈值,就让 job 失败,强制人工复核。这样安全审计就从"想起来才做"变成了"每次提交都过一遍"。
关于成本控制,给个实测参考:一个 5 万行的 Python 项目,按 400 行切块大约 125 个块,全用 Opus 4.6 跑一轮,输入 token 大概在 60 万到 80 万之间(含重叠部分),输出 token 取决于漏洞数量。如果你先用 Haiku 做初筛、只把可疑块送给 Opus 4.6,成本能压到三分之一左右。具体数字随代码密度波动,建议先拿一个小模块跑一轮,看usage字段的实际消耗再决定策略。
最后一个实用技巧:把 system prompt 里的漏洞类型列表按你项目的技术栈定制。比如纯前端项目就去掉sql_injection,加上dom_xss、prototype_pollution;Go 项目加上goroutine_leak、unsafe_pointer。prompt 越贴合你的代码特征,误报越少。这个列表可以放在config/agent.json里,改配置不用动代码。
跑通这条链路之后,你会发现 Claude Opus 4.6 那 500 个漏洞的新闻不再只是新闻,而是一个你可以自己复现的工程流程。模型能力是上限,但真正决定你能挖出多少东西的,是切块策略、prompt 设计和结果复核这三件事。