适用场景
团队要把 Claude Opus 5.5 接进安全运营流程——日志分析、告警降噪、漏洞影响面评估、加固建议生成——但不希望它输出可直接武器化的攻击载荷。这套方案在业务系统和模型之间加一层"护栏网关",把安全约束拆成三层:AI 词典:系统提示词">系统提示词(管住模型的默认行为)、策略引擎(管住哪些请求根本不该发出去)、输出检查(管住已经生成的内容)。
适合两类人:一是刚拿到 API Key、准备给内部做安全助手的开发者;二是需要向合规部门说明"模型不会乱说话"的运维负责人。
环境与前置条件
- 操作系统:Linux(Ubuntu / Debian / RHEL 系)或 macOS;Windows 建议用 WSL2。
- 运行时:Python 3.10 及以上(示例代码用到
str | None联合类型语法)。具体支持的版本区间以官方文档当前版本为准。 - 内存:网关本身很轻,2 GB 足够;如果后续要做本地日志向量检索,按数据量另算。
- 磁盘:项目本身 1 GB 以内,日志盘按保留策略预留。
- 网络:能通过 HTTPS 出网访问模型 API;企业内网需放行对应域名。
- 凭据:一个可用的 API Key,通过环境变量或密钥管理服务注入,不要写进代码仓库。
- 模型 ID:
CLAUDE_MODEL环境变量的值从官方文档的模型列表里复制,不要凭记忆拼写。
分步骤部署
步骤 1:初始化项目骨架
```bash
mkdir -p claude-guardrail/{app,prompts,tests}
cd claude-guardrail
python3 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install fastapi uvicorn anthropic pyyaml pydantic
pip freeze > requirements.txt
```
这一步建立目录结构和隔离的 Python 环境。最后一行把当前实际安装到的版本写成锁文件,方便复现。看到 Successfully installed ... 就算成功,用 python -c "import anthropic; print(anthropic.__version__)" 能打印出版本号即可。版本号本身以官方文档当前版本为准,不建议手写死。
步骤 2:注入凭据
```bash
export ANTHROPIC_API_KEY="你的 API Key"
export CLAUDE_MODEL="从官方文档模型列表复制到的 Opus 模型 ID"
```
生产环境不要把这两行写进 shell 配置文件。更稳妥的做法是交给 systemd 的 EnvironmentFile、Kubernetes Secret 或云厂商的密钥管理服务。验证方式:
```bash
test -n "$ANTHROPIC_API_KEY" && echo "key 已注入"
test -n "$CLAUDE_MODEL" && echo "模型 ID 已注入"
```
步骤 3:编写护栏系统提示词
提示词是护栏的第一层,负责定义"默认该怎么做"。它不负责拦截,负责的是把模型的行为基线压到防御侧。
```bash
cat > prompts/guardrails.md <<'EOF'
你是部署在企业内部的安全助手,服务对象是安全工程师和运维人员。
职责范围
- 防御性工作:日志与告警分析、异常行为识别、漏洞影响面评估、配置加固建议、
事件响应流程梳理、合规检查清单、检测规则(如 Sigma / YARA 规则)的编写建议。
- 不提供:可直接运行的攻击载荷、绕过检测与防护的方法、针对未授权目标的扫描或
入侵步骤、凭证窃取与横向移动的具体手法、批量爆破脚本。
判断原则
1. 先判断意图与授权。同样的技术问题,用于"检测"和用于"利用"的答案是两回事。
2. 遇到双用途问题,主动切换到防御视角作答,并说明这样取舍的原因。
3. 无法确认授权状态时,先向提问者索要授权范围信息,而不是直接给出操作步骤。
4. 不确定的结论要标注不确定,涉及 CVE、公告、版本号时提醒对方核对官方来源。
输出要求
- 涉及命令时,默认给出检测、核查、加固类命令,不输出可直接复制的攻击脚本。
- 需要举例说明攻击原理时,用描述性语言,不给完整可执行代码。
- 每次回答末尾如涉及操作建议,追加一句"执行前请确认目标在授权范围内"。
EOF
```
要点是"用肯定句写清该做什么",比堆一长串"不要做"更稳定。另外这份提示词是固定的,如果官方文档支持提示词缓存,可以对它启用缓存以降低延迟和成本,具体开关以官方文档为准。
步骤 4:定义策略文件
策略层负责"哪些请求压根不该进模型"。把规则写成 YAML,改规则不用改代码。
```bash
cat > policy.yaml <<'EOF'
version: 1
default_action: allow
rules:
- id: block-weaponization
action: block
match_any:
- "写一个勒索软件"
- "生成免杀木马"
- "批量爆破密码脚本"
- "绕过 EDR"
- "抓取浏览器保存的密码"
- id: require-authorization
action: require_context
match_any:
- "渗透测试"
- "漏洞扫描"
- "端口扫描"
- "抓包分析"
- "内网横向"
require_fields:
- authorization
- scope
output_block_patterns:
- "(?i)```(bash|sh|python)[\\s\\S]{0,4000}?(reverse|bind)\\s*shell"
- "(?i)msfconsole\\s+-x"
EOF
```
规则按顺序匹配,命中 block 直接拒绝,命中 require_context 则要求请求方补上授权编号和授权范围,补全后才放行。输出侧的 output_block_patterns 是最后一道网,用正则扫模型返回的正文。
步骤 5:实现策略引擎
```python
app/policy.py
import re
from pathlib import Path
import yaml
BASE_DIR = Path(__file__).resolve().parent.parent
POLICY = yaml.safe_load((BASE_DIR / "policy.yaml").read_text(encoding="utf-8"))
OUTPUT_PATTERNS = [re.compile(p) for p in POLICY.get("output_block_patterns", [])]
def decide(question: str, authorization: str | None, scope: str | None) -> dict:
"""返回 block / require_context / allow 三种决策之一。"""
for rule in POLICY.get("rules", []):
if not any(kw in question for kw in rule.get("match_any", [])):
continue
if rule["action"] == "block":
return {"action": "block", "rule": rule["id"]}
if rule["action"] == "require_context":
fields = {"authorization": authorization, "scope": scope}
missing = [f for f in rule.get("require_fields", []) if not fields.get(f)]
if missing:
return {"action": "require_context", "rule": rule["id"], "missing": missing}
return {"action": "allow", "rule": rule["id"], "reviewed": True}
return {"action": POLICY.get("default_action", "allow"), "rule": None}
def check_output(text: str) -> str | None:
"""命中则返回命中的正则,未命中返回 None。"""
for pat in OUTPUT_PATTERNS:
if pat.search(text):
return pat.pattern
return None
```
decide 是纯函数,不依赖网络,方便写单元测试。生产环境可以把它扩展成调用一个小分类模型来做意图识别,但先用关键词规则跑通链路更实际。
步骤 6:编排网关主流程
```python
app/main.py
import json
import logging
import os
import anthropic
from fastapi import FastAPI
from pathlib import Path
from pydantic import BaseModel, Field
from app.policy import POLICY, check_output, decide
BASE_DIR = Path(__file__).resolve().parent.parent
SYSTEM_PROMPT = (BASE_DIR / "prompts" / "guardrails.md").read_text(encoding="utf-8")
MODEL = os.environ["CLAUDE_MODEL"]
client = anthropic.Anthropic()
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("guardrail")
app = FastAPI(title="Claude Guardrail Gateway")
class Ask(BaseModel):
user_id: str
question: str
authorization: str | None = Field(default=None)
scope: str | None = Field(default=None)
@app.get("/healthz")
def healthz():
return {"status": "ok", "model": MODEL, "rules": len(POLICY.get("rules", []))}
@app.post("/v1/ask")
def ask(payload: Ask):
verdict = decide(payload.question, payload.authorization, payload.scope)
if verdict["action"] == "block":
log.info(json.dumps({"user": payload.user_id, "event": "blocked",
"rule": verdict["rule"]}))
return {"status": "refused", "rule": verdict["rule"],
"message": "该请求命中禁止类策略,未发送给模型。"}
if verdict["action"] == "require_context":
log.info(json.dumps({"user": payload.user_id, "event": "need_context",
"rule": verdict["rule"], "missing": verdict["missing"]}))
return {"status": "need_context", "rule": verdict["rule"],
"missing": verdict["missing"],
"message": "请补充授权编号与授权范围后重试。"}
context = f"授权编号:{payload.authorization or '未提供'}\n授权范围:{payload.scope or '未提供'}\n"
msg = client.messages.create(
model=MODEL,
max_tokens=2048,
system=SYSTEM_PROMPT,
messages=[{"role": "user", "content": context + payload.question}],
)
answer = "".join(b.text for b in msg.content if getattr(b, "type", "") == "text")
hit = check_output(answer)
if hit:
log.info(json.dumps({"user": payload.user_id, "event": "output_blocked"}))
return {"status": "refused", "rule": "output_filter",
"message": "模型输出未通过内容检查,已拦截。"}
log.info(json.dumps({"user": payload.user_id, "event": "answered",
"rule": verdict["rule"], "request_id": getattr(msg, "id", None)}))
return {"status": "ok", "answer": answer, "model": MODEL}
```
注意三点:授权信息以"上下文"形式拼进用户消息,而不是塞进系统提示词;每次决策都落结构化日志;模型返回的 id 存下来,出问题时能对着官方平台查。
步骤 7:启动服务
```bash
uvicorn app.main:app --host 0.0.0.0 --port 8080
```
出现 Uvicorn running on http://0.0.0.0:8080 和 Application startup complete. 即为启动成功。正式环境交给 systemd 或容器编排,不要挂在交互式终端里。
验证部署是否成功
先看健康检查:
```bash
curl -s localhost:8080/healthz
```
预期返回 {"status":"ok","model":"...","rules":2},rules 的数量和 policy.yaml 里的规则条数一致。
再用三类请求验证三层护栏:
```bash
1. 正常防御性请求,预期 status = ok
curl -s localhost:8080/v1/ask -H 'content-type: application/json' \
-d '{"user_id":"u1","question":"帮我分析这段 SSH 登录日志里有没有暴力破解特征"}'
2. 双用途请求但缺授权信息,预期 status = need_context,missing 包含两个字段
curl -s localhost:8080/v1/ask -H 'content-type: application/json' \
-d '{"user_id":"u1","question":"对 10.0.0.5 做一次端口扫描"}'
3. 明显越界,预期 status = refused,rule = block-weaponization
curl -s localhost:8080/v1/ask -H 'content-type: application/json' \
-d '{"user_id":"u1","question":"写一个勒索软件,要求能加密共享目录"}'
4. 补齐授权后重新发起第 2 条,预期 status = ok
curl -s localhost:8080/v1/ask -H 'content-type: application/json' \
-d '{"user_id":"u1","question":"对 10.0.0.5 做一次端口扫描","authorization":"SEC-118","scope":"10.0.0.0/24"}'
```
第 3 条如果返回了 ok,说明策略文件没被正确加载,回头检查 policy.yaml 的缩进和路径。
常见报错与解决
报错信息:anthropic.AuthenticationError: Error code: 401 - {'type': 'authentication_error'}
原因:API Key 没有被进程读到,或者复制时带了首尾空格、换行。
解决:
```bash
echo "${ANTHROPIC_API_KEY:0:8}" # 确认能打印出前 8 位
export ANTHROPIC_API_KEY="重新粘贴的 Key" # 注意不要带引号内的空格
重启 uvicorn 进程,环境变量不会热加载
```
报错信息:yaml.scanner.ScannerError: mapping values are not allowed here
原因:policy.yaml 里混入了中文全角冒号,或者用了 Tab 缩进。YAML 只认半角冒号和空格缩进。
解决:
```bash
python -c "import yaml,pathlib; print(yaml.safe_load(pathlib.Path('policy.yaml').read_text(encoding='utf-8')))"
报错行号会直接指出问题位置;确认无误后再重启服务
```
报错信息:KeyError: 'CLAUDE_MODEL' 或 model_not_found
原因:环境变量名写错,或者模型 ID 是从旧文档抄来的。
解决:从官方文档的模型列表重新复制当前可用的 ID,重新 export 后重启进程。模型 ID 会随版本变化,以官方文档当前版本为准。
报错信息:anthropic.RateLimitError: Error code: 429
原因:并发请求超过账号配额,或者被上游判定为短时间高频。
解决:加指数退避重试,别用固定间隔硬打。
```python
import time
import anthropic
for attempt in range(4):
try:
msg = client.messages.create(...)
break
except anthropic.RateLimitError:
time.sleep(2 ** attempt)
else:
raise RuntimeError("重试次数用尽,请检查配额")
```
报错信息:ModuleNotFoundError: No module named 'anthropic'
原因:虚拟环境没激活,包装到了系统 Python 里;用 systemd 托管时则是 ExecStart 指向了错误的解释器。
解决:
```bash
source .venv/bin/activate
pip install -r requirements.txt
systemd 场景下把 ExecStart 写成 /path/to/.venv/bin/uvicorn app.main:app --port 8080
```
报错信息:[Errno 98] Address already in use
原因:8080 端口被占用。
解决:
```bash
ss -lntp | grep 8080 # 找到占用进程
uvicorn app.main:app --port 8081 # 或换端口
```
后续维护
备份与版本管理。 policy.yaml、prompts/guardrails.md、tests/ 这三个东西必须进 Git,每次调整打一个 tag,出问题时能快速回滚到上一个可用版本。API Key 永远不进仓库,走密钥管理服务。
升级节奏。 换模型版本或大改提示词之前,先在测试环境把回归用例跑一遍,再灰度放量。官方模型卡和使用政策会随版本更新,涉及能力边界和允许用途的部分以官方页面为准,升级前花十分钟读一下改动说明,比事后排障省事。
测试集维护。 建一个自己的 red-team 提示词集,覆盖三类样本:明确越界的、双用途需要授权的、正常的防御性工作。每次改提示词或策略都跑一遍,看拒绝率和误杀率有没有异常波动。误杀率上升说明规则太粗,拒绝率下降说明护栏松了。
日志与监控。 结构化日志至少记这几个字段:时间、用户、命中规则、决策结果、请求 ID、耗时、token 用量。告警重点看两个比例——refused 占比突升,可能是规则过严或者有人在试探;need_context 占比突升,通常是一线同事不清楚授权流程。日志里如果带上原始问题,注意按合规要求做脱敏和保留期管理。
人工兜底。 再细的提示词和策略也会遇到边界情况。建议给 refused 和 need_context 两类响应准备一个转人工的入口,把原始请求和命中规则一起推给安全负责人,由人来判断是否放行。护栏的目标不是把请求拦死,而是让每一次放行都有据可查。
