模块 1 · 双栈鉴权
JWT 双栈:
deps 与 gateway 各管一线
模块 API 走 app/api/deps.py(get_auth_context / get_platform_auth_context);
宿主 Wave 0 网关走 app/gateway/auth_deps.py。
模块禁止 import gateway——指挥 AI 改鉴权时,先确认改的是哪条栈。
两套鉴权表(背下来少踩坑)
| 通道 | 入口函数 | X-Agent-Type | 典型路由 |
|---|---|---|---|
| Agent / 对话 | get_auth_context | JWT 通道必填 + 准入矩阵 | /api/chat, /api/risk/*, simulate |
| 平台只读 | get_platform_auth_context | 不要 | /api/customers/*, /api/analyst/* |
| 宿主网关 | gateway/auth_deps.get_auth_context | 宿主口径 | Wave 0 四件套(模块不 import) |
APP_ENV=development 且无 RS256 公钥时,可用 X-Debug-Role + X-Debug-Actor 冒充身份——pytest 和部分演示 SOP 依赖此通道;生产一律 Bearer JWT。
agent_type = request.headers.get("X-Agent-Type")
if not agent_type:
raise ApiError(401, "AUTH_401_MISSING_AGENT_TYPE")
assert_agent_access(auth, agent_type, ...)
# 平台线 — 无上述检查
def get_platform_auth_context(request):
"""不要求 X-Agent-Type"""
对话线:验完 JWT 还要读 Agent 头,并对照 AGENT_ACCESS_MATRIX。
例如客户 token 不能带 X-Agent-Type: risk,否则 403。
平台线:验 JWT 就放行到归属断言,不问你走哪条 Agent。
gateway 栈是宿主合并用,和 deps 双栈并存,别在模块里混 import。
准入矩阵速览
模块 2 · 会话记忆
session_repository + memory_service:
Redis 窗口,MySQL 权威
对话每轮消息同步写 MySQL(权威),同时用
Redis
缓存最近 N≤20 条,TTL 2 小时。读历史时 Redis 命中快,miss 则回源
session_repository。
数据流:用户发第二条消息
Key 格式:sess:{agent}:{session_id}:msgs
def window_key(agent_type, session_id):
return f"sess:{agent_type}:{session_id}:msgs"
def get_recent(agent_type, session_id, limit=20):
raw = get_gateway().lrange(window_key(...), -limit, -1)
if raw: return [json.loads(item) for item in raw]
return SessionRepository().list_messages(session_id, limit=limit)
def append_window(...):
client.rpush(key, ...); client.ltrim(key, -20, -1); client.expire(key, 7200)
每个会话在 Redis 里是一条 List,key 带 agent 类型防串线。
读:先捞 List 尾部 N 条;空或报错就去 MySQL 查 agent_message 表。
写:新消息 RPUSH 进去,LTRIM 砍掉太老的,EXPIRE 两小时自动过期。
session_repository 管表结构、分页列表、关闭会话——和合规审计表分开。
模块 3 · 防护与审计
input_guard 限流注入、
audit_middleware 留痕
用户消息进 LLM 前要经过 input_guard; 每个 HTTP 请求经过 audit_middleware 写访问审计。鉴权 403 还会双写 audit_log + input_guard_log(平台线部分跳过 ENUM 限制)。
input_guard 三类检查
prompt_injection
黑名单短语(「忽略上文」「system prompt」等)→ 拒答 + 落库 guard_type=injection
oversize
业务上限 4000 字(guard 层判定才能留痕;Pydantic 硬顶只防 DoS)
rate_limit
check_rate_limit(agent_type, actor_id) 超限 → AUTH 类拒绝 + 限流日志
群聊:恶意输入被拦下
if not input_guard.check_rate_limit(agent_type, auth.actor_id):
insert_input_guard_log(..., guard_type=GUARD_RATE_LIMIT)
raise ApiError(...)
verdict = input_guard.inspect_message(message)
if verdict.reject:
insert_input_guard_log(..., guard_type=verdict.guard_type)
# main.py 挂载
async def audit_middleware(request, call_next):
# INSERT audit_log 单请求访问记录
先发消息前查频率:同 actor 刷太快就拒,并记限流日志。
再扫内容:注入/超长命中就不调 LLM,直接拒答+留痕。
audit_middleware 包在更外层:每个 HTTP 进来都记一条访问审计。
deps.deny 鉴权失败也会双写——平台 agent_type 时 input_guard_log 可能跳过 ENUM 限制。