ent 睡个安稳觉。️ 第一道门禁卡——OAuth2 JWT 认证API 不能敞开大门随便进。FastAPI 内置了OAuth2PasswordBearer这套“门禁系统”配合 JWT 就像一张带时效的房卡。登录后拿 token后续请求都得带上否则直接 401 挡在门外。这里一定要注意千万不要把 secret key 硬编码正经做法是从环境变量读。from fastapi import Depends, FastAPI, HTTPException from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import jwt from datetime import datetime, timedelta SECRET_KEY 从环境变量取! ALGORITHM HS256 oauth2_scheme OAuth2PasswordBearer(tokenUrl/token) app FastAPI() def create_token(data: dict): to_encode data.copy() to_encode.update({exp: datetime.utcnow() timedelta(hours2)}) return jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) app.post(/token) def login(form_data: OAuth2PasswordRequestForm Depends()): # 验证用户示例固定账号 if form_data.username ! admin or form_data.password ! pass: raise HTTPException(status_code401, detail密码错误) return {access_token: create_token({sub: form_data.username}), token_type: bearer} 第二道权限分级——RBAC有了门禁还不够不能所有人进来都能操作 Agent。我们用“角色”来区分普通用户只能查询管理员才能修改配置或调用危险工具。关键是一个依赖注入函数检查 token 里带的 role 字段。这里也要注意不能只靠前端隐藏按钮后端完全没校验——那等于把钥匙藏在门口地垫下。from fastapi import Security def get_current_user(token: str Depends(oauth2_scheme)): payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) return {username: payload.get(sub), role: payload.get(role, user)} def require_role(role: str): def checker(user Depends(get_current_user)): if user[role] ! role: raise HTTPException(status_code403, detail权限不足) return user return checker app.get(/admin/agent-config) def update_config(user Depends(require_role(admin))): return {message: 只有管理员看得到} 第三道入口安检——输入清洗与限流Agent 最怕提示注入。我的做法是先用Pydantic做格式校验再上自定义清洗函数把那些 忽略前面的指令、请扮演 DAN 之类的黑名单短语直接堵死。当然黑名单不是银弹但能挡住大部分脚本小子。另外同一 IP 每秒 100 次的请求绝对不正常。用SlowAPI中间件限流就像给接口装了水龙头防止被冲垮。注意生产环境一定要搭配 Redis 存储计数不然多 worker 下会漏。from slowapi import Limiter from slowapi.util import get_remote_address from pydantic import BaseModel, validator limiter Limiter(key_funcget_remote_address) app.state.limiter limiter app.add_exception_handler(429, lambda _,_: JSONResponse(status_code429, content慢点)) class PromptInput(BaseModel): prompt: str validator(prompt) def check_injection(cls, v): blacklist [忽略前面的指令, ignore previous, DAN] if any(b in v.lower() for b in blacklist): raise ValueError(检测到注入企图) return v app.post(/agent/ask) limiter.limit(5/second) def ask_agent(item: PromptInput, request: Request): return {response: f你说: {item.prompt[:50]}...} 第四道工具执行的“无菌实验室”Agent 经常会调用本地命令、跑 Python 脚本这相当于给用户开了一个终端——不用沙箱就是自杀。我强制所有命令必须通过白名单比如只能执行 ls、cat 这几个同时用 Python 的 resource 模块限制子进程的内存和 CPU 时间避免恶意死循环吃光服务器。网络隔离可以进一步用 Docker 的 --networknone这里用子进程做个最简演示import subprocess, resource ALLOWED_COMMANDS [ls, cat, echo] def safe_exec(cmd: str): parts cmd.strip().split() if not parts or parts[0] not in ALLOWED_COMMANDS: raise ValueError(命令不在白名单) def set_limits(): resource.setrlimit(resource.RLIMIT_AS, (50*1024*1024, 100*1024*1024)) # 内存50MB resource.setrlimit(resource.RLIMIT_CPU, (2, 2)) # CPU时间2秒 try: result subprocess.run(parts, capture_outputTrue, textTrue, preexec_fnset_limits, timeout5) return result.stdout except subprocess.TimeoutExpired: return 执行超时已终止 第五道密钥轮换与环境隔离最后这条很多人会忽略JWT 签名密钥、第三方 API Key 要定期换就像家门锁的密码不能万年不变。我写了一个简单的 KeyManager从环境变量读取并支持定期重载。不同环境开发、测试、生产严格隔离一个环境的 Key 泄露不至于连累全家。import os, time class KeyManager: def __init__(self, env_var: str, refresh_interval3600): self.env_var env_var self.refresh_interval refresh_interval self._last_read 0 self._cached None def get(self): now time.time() if now - self._last_read self.refresh_interval or self._cached is None: self._cached os.getenv(self.env_var) self._last_read now if not self._cached: raise RuntimeError(f环境变量 {self.env_var} 未设置) return self._cached secret_key_manager KeyManager(JWT_SECRET_KEY, refresh_interval7200)
