03 - 统一网关：身份认证、路由与流量管控

一句话：网关是 SSO 身份换成 Runtime Token 的唯一节点，剥掉客户端伪造头、从 Vault 取钥匙、按 user_id 做一致性哈希寻址，再把 SSE 重编码成 chunked 流；全链路的可用性因此押在 ext_authz 与 Redis 缓存上，失败模式取 deny 就要接受认证服务抖动即全员 401，而实例暂存只能被翻译成 202 waking 这种等待语义。

一、网关定位

统一网关（Higress）是员工请求进入系统的唯一入口。所有用户流量——无论来自 Web 控制台、企业 IM 还是定时任务——一律经过网关，由网关完成身份认证、路由寻址、密钥注入、限流审计后，转发到对应的员工实例。

网关承担 10 项核心能力：

#	能力	实现机制
1	SSO 身份接入	OIDC/JWKS 校验 id_token，提取 sub → user_id
2	身份路由	user_id → Nacos 寻址 → 实例端点
3	RT 钥匙注入	ext_authz 服务从 Vault 缓存取 RT 注入 header，剥离客户端伪造
4	HTTP 反向代理	透传方法/头/体，支持页面、API、静态资源
5	SSE 流式适配	chunked 重编码 + 心跳（30s ping comment）
6	角色权限	employee/admin 路由白名单，管理端点员工一律 403
7	sticky 会话	一致性哈希 + 虚拟节点（每实例 160 个虚拟点）
8	分布式限流	Redis 滑动窗口，按虚 SK 配额分级
9	审计轨迹	JSONL → OTel → 审计平台，trace_id 贯通
10	实例状态感知	healthz 探活 + staged → 202 waking

二、网关在 K8s 中的部署位置

三、SSO 身份认证

3.1 认证流程

3.2 JWT 验签与 JWKS 缓存

ext_authz 服务启动时从企业 SSO 拉取 JWKS 公钥集，缓存 1 小时，过期自动刷新。刷新失败时降级使用缓存公钥（窗口期内有效）。

# ext_authz 服务核心逻辑

JWKS_URI = "https://sso.example.com/.well-known/jwks.json"
JWKS_AUDIENCE = "qwenpaw-agent-platform"
JWKS_ISSUER = "https://sso.example.com"
JWKS_CACHE_TTL = 3600  # 1 小时

class JWTAuthenticator:
    def __init__(self):
        self._jwks_keys: dict[str, dict] = {}
        self._cache_time: float = 0

    async def refresh_jwks(self):
        """从 SSO 拉取 JWKS 公钥集"""
        async with httpx.AsyncClient() as client:
            resp = await client.get(JWKS_URI, timeout=5.0)
            resp.raise_for_status()
            data = resp.json()
            self._jwks_keys = {k["kid"]: k for k in data["keys"]}
            self._cache_time = time.time()

    def verify(self, token: str) -> dict:
        """
        验签 JWT，返回 payload。
        失败抛出 ExpiredSignatureError / InvalidTokenError。
        """
        if time.time() - self._cache_time > JWKS_CACHE_TTL:
            try:
                await self.refresh_jwks()
            except Exception:
                pass  # 降级使用缓存

        kid = jwt.get_unverified_header(token).get("kid")
        if kid not in self._jwks_keys:
            raise jwt.InvalidTokenError(f"Unknown kid: {kid}")

        public_key = jwt.algorithms.RSAAlgorithm.from_jwk(
            json.dumps(self._jwks_keys[kid])
        )
        return jwt.decode(
            token, public_key,
            algorithms=["RS256"],
            audience=JWKS_AUDIENCE,
            issuer=JWKS_ISSUER,
        )

3.3 Header 安全

网关在转发请求前执行严格的 header 清洗：

操作	Header	原因
剥离	x-qwenpaw-runtime-token	防客户端伪造 RT
剥离	x-sso-user、x-sso-email	防客户端伪造身份
剥离	x-request-id	由网关统一生成
注入	x-sso-user: {user_id}	从 JWT sub 提取，已验签
注入	x-sso-email: {email}	可选，用于审计
注入	x-request-id: {trace_id}	UUID v4，贯穿全链路
注入	x-qwenpaw-runtime-token: {rt}	由 ext_authz 从 Vault 注入

四、RT 注入：ext_authz 实现

4.1 为什么需要 ext_authz

Higress 基于 Envoy，本身不内置"从 Vault 取密钥注入 header"的能力。Envoy 原生支持 ext_authz 过滤器——将每个请求的认证决策委托给外部服务，外部服务返回允许/拒绝及可注入的 header 列表。

这是 Higress 官方推荐的认证扩展方式，也是本方案 RT 注入的实现路径。

4.2 完整调用时序

4.3 ext_authz 服务完整实现

"""
ext_authz 认证服务
部署：与 Higress 同 Namespace（qwenpaw-gateway），Deployment ×2+
通信：HTTP，Higress 通过 Envoy ext_authz filter 调用
"""

from fastapi import FastAPI, Request, Response
from cryptography.fernet import Fernet
import redis.asyncio as redis
import asyncpg
import json, time, uuid, os

app = FastAPI(title="QwenPaw ext_authz")

# ── 配置 ────────────────────────────────────────────────
REDIS_URL = os.getenv("REDIS_URL", "redis://redis.qwenpaw-infra:6379")
DB_DSN = os.getenv("DB_DSN", "postgresql://ctrl:***@postgres.qwenpaw-control:5432/qwenpaw")
FERNET_KEY = os.environ["FERNET_KEY"].encode()  # 从 K8s Secret 注入
RT_CACHE_TTL = 300  # 5 分钟

# ── 全局连接 ────────────────────────────────────────────
redis_client: redis.Redis = None
db_pool: asyncpg.Pool = None
fernet: Fernet = None

@app.on_event("startup")
async def startup():
    global redis_client, db_pool, fernet
    redis_client = redis.from_url(REDIS_URL, decode_responses=True)
    db_pool = await asyncpg.create_pool(DB_DSN, min_size=2, max_size=10)
    fernet = Fernet(FERNET_KEY)

# ── 会话验证 ────────────────────────────────────────────
async def resolve_session(session_token: str) -> dict | None:
    """session_token → user_id + role（Redis 存储，网关登录时写入）"""
    data = await redis_client.get(f"session:{session_token}")
    if data:
        return json.loads(data)
    return None

# ── RT 获取 ─────────────────────────────────────────────
async def get_runtime_token(user_id: str) -> str | None:
    """从 Redis 缓存获取 RT，未命中则从 Vault（PostgreSQL）拉取"""
    cache_key = f"rt:{user_id}"
    rt = await redis_client.get(cache_key)
    if rt is not None:
        return rt

    # 缓存未命中 → 从 Vault 拉取
    row = await db_pool.fetchrow("""
        SELECT c.encrypted_value
        FROM credentials c
        JOIN instance_registry r ON r.instance_id = c.instance_id
        WHERE r.user_id = $1
          AND c.credential_type = 'runtime_token'
          AND r.observed_state = 'active'
    """, user_id)

    if not row:
        return None

    rt = fernet.decrypt(row["encrypted_value"]).decode()
    await redis_client.set(cache_key, rt, ex=RT_CACHE_TTL)
    return rt

# ── 认证端点 ────────────────────────────────────────────
@app.post("/authz")
async def authorize(request: Request):
    """
    Higress ext_authz 端点。
    返回 200 = 放行（附带 allowed_headers 注入到上游请求）。
    返回 4xx = 拒绝（响应体直接返回客户端）。
    """
    # 1. 提取 session_token
    session_token = request.cookies.get("session_token")
    auth_header = request.headers.get("authorization", "")
    if auth_header.startswith("Bearer "):
        session_token = auth_header[7:]  # 服务间调用（聚合器）也用此路径

    if not session_token:
        return Response(
            content=json.dumps({"error": "missing_credentials"}),
            status_code=401,
            headers={"Content-Type": "application/json"},
        )

    # 2. 验证会话
    session = await resolve_session(session_token)
    if not session:
        return Response(
            content=json.dumps({"error": "invalid_session"}),
            status_code=401,
            headers={"Content-Type": "application/json"},
        )

    user_id = session["user_id"]
    role = session.get("role", "employee")

    # 3. 查找 RT
    rt = await get_runtime_token(user_id)
    if rt is None:
        # 检查实例状态
        state = await db_pool.fetchval(
            "SELECT observed_state FROM instance_registry WHERE user_id = $1",
            user_id,
        )
        if state == "staged":
            return Response(
                content=json.dumps({
                    "status": "waking",
                    "eta_seconds": 30,
                    "poll": "/api/healthz",
                }),
                status_code=202,
                headers={"Content-Type": "application/json"},
            )
        if state == "retired" or state is None:
            return Response(
                content=json.dumps({"error": "instance_not_found"}),
                status_code=404,
                headers={"Content-Type": "application/json"},
            )
        return Response(
            content=json.dumps({"error": "instance_not_ready"}),
            status_code=503,
            headers={"Content-Type": "application/json"},
        )

    # 4. 角色权限检查
    path = request.headers.get("x-original-path", request.url.path)
    if role == "employee" and path.startswith(("/api/providers", "/api/settings", "/api/backup", "/api/admin")):
        return Response(
            content=json.dumps({"error": "forbidden"}),
            status_code=403,
            headers={"Content-Type": "application/json"},
        )

    # 5. 构造注入 header
    trace_id = str(uuid.uuid4())
    return Response(
        content=json.dumps({
            "allowed_headers": {
                "x-qwenpaw-runtime-token": rt,
                "x-sso-user": user_id,
                "x-sso-role": role,
                "x-request-id": trace_id,
            }
        }),
        status_code=200,
        headers={"Content-Type": "application/json"},
    )

# ── RT 缓存失效（控制面轮换时调用）──────────────────────
# P0 修复：内部端点增加 mTLS + 共享密钥双重认证
INTERNAL_API_SECRET = os.environ["INTERNAL_API_SECRET"]  # 控制面与 ext_authz 共享密钥

@app.post("/internal/invalidate-rt")
async def invalidate_rt(request: Request):
    """控制面 RT 轮换后调用，清除 Redis 缓存
    
    认证要求：
    1. mTLS：请求必须来自 qwenpaw-control namespace（NetworkPolicy 保证）
    2. 共享密钥：Header 中必须携带 x-internal-api-secret
    """
    # 验证共享密钥
    provided_secret = request.headers.get("x-internal-api-secret", "")
    if not secrets.compare_digest(provided_secret, INTERNAL_API_SECRET):
        return JSONResponse(
            status_code=403,
            content={"error": "forbidden", "message": "Invalid internal API secret"},
        )
    
    # 验证来源 namespace（通过 mTLS 客户端证书或 NetworkPolicy 保证）
    source_ns = request.headers.get("x-source-namespace", "")
    if source_ns != "qwenpaw-control":
        audit_log.warning(f"Unauthorized namespace access: {source_ns}")
        return JSONResponse(
            status_code=403,
            content={"error": "forbidden", "message": "Unauthorized namespace"},
        )
    
    body = await request.json()
    user_id = body.get("user_id")
    if user_id:
        await redis_client.delete(f"rt:{user_id}")
        audit_log.info(f"RT cache invalidated for user: {user_id}, source: {source_ns}")
    return {"status": "ok"}

4.4 Higress 路由配置

# Higress Ingress 配置

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: qwenpaw-agent
  namespace: qwenpaw-gateway
  annotations:
    # ext_authz 配置
    higress.io/authz-endpoint: "http://authz-svc.qwenpaw-gateway:8090/authz"
    higress.io/authz-failure-mode: "deny"           # 认证服务不可用 = 拒绝
    higress.io/authz-timeout: "3000"                 # 3s 超时
    # 注入 header 白名单（ext_authz 返回的 header 才允许透传）
    higress.io/authz-allowed-headers: >
      x-qwenpaw-runtime-token,x-sso-user,x-sso-role,x-request-id
    # 剥离客户端可能伪造的 header
    higress.io/request-header-control.remove: >
      x-qwenpaw-runtime-token,x-sso-user,x-sso-email,x-request-id
    # SSE 支持
    higress.io/proxy-buffering: "off"
    higress.io/proxy-read-timeout: "3600"
    # 动态路由（按 x-sso-user 寻址到 Nacos 注册的实例）
    higress.io/destination: "qwenpaw"
    higress.io/backend-protocol: "HTTP"
spec:
  ingressClassName: higress
  rules:
    - host: agent.example.com
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: higress-dynamic-backend
                port:
                  number: 8088

4.5 RT 轮换与紧急吊销

五、身份路由与 Nacos 寻址

5.1 路由查找流程

5.2 Nacos 注册信息

每个员工实例在 Nacos 中注册为一个 service instance：

{
  "serviceName": "qwenpaw",
  "ip": "10.0.1.23",
  "port": 8088,
  "metadata": {
    "user_id": "zhangsan",
    "instance_id": "inst_abc123",
    "state": "active",
    "namespace": "qwenpaw-instances",
    "pod_name": "qwenpaw-zhangsan-0",
    "image_version": "v1.2.0"
  }
}

5.3 一致性哈希

网关使用一致性哈希保证同一用户的请求始终路由到同一实例。虚拟节点策略保证实例增删时路由抖动最小：

hash_key = user_id
hash_ring = 每实例 160 个虚拟节点
hash_func = MD5(user_id + vnode_index) → ring 位置

• 实例扩容（新员工加入）：仅影响相邻虚拟节点的请求，抖动 < 2%
• 实例缩容（员工离职/暂存）：同上，会话数据在 PVC 中保留
• 网关多副本：每个副本维护相同的哈希环（从 Nacos 订阅），路由一致

六、SSE 流式转发

6.1 SSE 透传机制

实例返回 SSE 事件流（text/event-stream），网关必须 chunked 重编码转发以保证浏览器打字机效果：

# Higress Lua 插件处理 SSE 转发
def forward_sse(upstream_response, client_connection):
    """SSE 必须 chunked 重编码，否则浏览器等连接关闭才渲染"""
    client_connection.send_header("Transfer-Encoding", "chunked")
    client_connection.send_header("Connection", "keep-alive")
    client_connection.send_header("Content-Type", "text/event-stream")
    client_connection.send_header("Cache-Control", "no-cache")
    client_connection.end_headers()

    while True:
        chunk = upstream_response.read(65536)
        if not chunk:
            break
        # HTTP chunked 编码：hex_size\r\ndata\r\n
        client_connection.wfile.write(b"%x\r\n" % len(chunk))
        client_connection.wfile.write(chunk)
        client_connection.wfile.write(b"\r\n")
        client_connection.wfile.flush()

    # 终止块
    client_connection.wfile.write(b"0\r\n\r\n")
    client_connection.wfile.flush()

6.2 心跳保活

实例每 30s 发送 SSE comment（:ping），网关透传。客户端 EventSource 自动处理重连（Last-Event-ID）。

6.3 断线恢复

客户端 EventSource 断线
  → 自动重连（携带 Last-Event-ID）
  → 网关重新建立到实例的连接
  → 实例从 Last-Event-ID 续传（实例侧缓存最近 100 条事件）
  → 超过缓存范围 → 返回 session_expired，客户端刷新会话

七、角色权限

7.1 角色定义

角色	可达端点	说明
employee	/api/console/chat、/api/healthz、/api/agents/*/memory	对话、查看状态
admin	全部端点	管理操作（需额外审计）
system	仅控制面 RT 可达	内部服务调用

7.2 权限拦截实现

角色检查在 ext_authz 服务中执行（见 §4.3 代码），网关只做透传。拦截规则：

ADMIN_PREFIXES = [
    "/api/providers",
    "/api/settings",
    "/api/backup",
    "/api/admin",
    "/api/mcp",
    "/api/skills/hub",
]

def check_role(role: str, path: str) -> bool:
    if role == "admin":
        return True   # 管理员放行，实例侧再做细粒度权限
    if role == "employee":
        return not any(path.startswith(p) for p in ADMIN_PREFIXES)
    return False      # 未知角色拒绝

职责边界：网关只做角色级粗粒度拦截（员工不可达管理端点），实例内部的细粒度权限（如管理员能否修改模型配置）由控制面在实例侧执行。

八、分布式限流

8.1 限流策略

限流维度：per user（按虚 SK 归属）
算法：Redis 滑动窗口
参数（按订阅档位）：

┌──────────┬────────────┬────────────┬──────────────────┬──────────────────┐
│ 档位     │ QPS 限制   │ 突发容量   │ 日 Token 配额    │ 月成本配额       │
├──────────┼────────────┼────────────┼──────────────────┼──────────────────┤
│ 基础档   │ 10 req/s   │ 20         │ 100K tokens      │ ¥100            │
│ 标准档   │ 30 req/s   │ 60         │ 500K tokens      │ ¥500            │
│ 增强档   │ 100 req/s  │ 200        │ 2M tokens        │ ¥2000           │
└──────────┴────────────┴────────────┴──────────────────┴──────────────────┘

8.2 限流实现

class RateLimiter:
    """Redis 滑动窗口限流器"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    async def check(self, user_id: str, tier: str) -> tuple[bool, dict]:
        key = f"ratelimit:{user_id}"
        now = time.time()
        window = 1.0

        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(key, 0, now - window)
        pipe.zadd(key, {str(now): now})
        pipe.zcard(key)
        pipe.expire(key, int(window * 2))
        results = await pipe.execute()

        count = results[2]
        limit = TIER_LIMITS[tier]["rate"]

        if count > limit:
            return False, {
                "retry_after": 1.0,
                "limit": limit,
                "count": count,
            }
        return True, {
            "limit": limit,
            "count": count,
            "remaining": limit - count,
        }

8.3 超限响应

HTTP/1.1 429 Too Many Requests
Content-Type: application/json

{
  "error": "rate_limit_exceeded",
  "message": "请求频率超限，请稍后重试",
  "retry_after": 1.0,
  "tier": "basic",
  "upgrade_hint": "联系管理员升级订阅以获得更高配额"
}

九、审计轨迹

9.1 审计事件格式

{
  "trace_id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2026-10-25T10:30:00.123Z",
  "user_id": "zhangsan",
  "instance_id": "inst_abc123",
  "virtual_sk": "vsk_zhangsan_001",
  "method": "POST",
  "path": "/api/console/chat",
  "status": 200,
  "request_tokens": 150,
  "response_tokens": 2300,
  "latency_ms": 1234,
  "source_ip": "10.0.0.1",
  "user_agent": "FeishuBot/1.0"
}

9.2 审计链路

9.3 trace_id 贯通

十、实例状态感知

10.1 健康检查

网关通过 ext_authz 服务间接感知实例状态。当 ext_authz 发现 RT 不存在时，查询 registry 状态：

实例状态	ext_authz 响应	网关行为	前端表现
active	200 + 注入 RT	正常转发	正常对话
staged	202 + {"status":"waking","eta_seconds":30}	返回 202	"正在唤醒你的助手，约 30 秒" + 轮询
pending	503 + {"status":"starting"}	返回 503	"助手正在启动中"
retired / 无记录	404	返回 404	"未开通，点击申请"
failed	503	返回 503	"助手异常，正在恢复"

10.2 网关侧探活

网关独立维护实例健康状态（与 ext_authz 互补）：

网关后台探活任务（每 10s 执行）：
  对 Nacos 中所有 active 实例：
    GET /api/healthz（携带 RT）
    → 200：更新 last_healthy 时间戳
    → 503：标记 starting，继续等待
    → 超时 / 连接拒绝：连续失败计数 +1

  连续 3 次失败（5 分钟内）：
    → 从 Nacos 注销该实例
    → 通知控制面触发自动重建
    → 网关返回 503 "助手暂不可用，正在恢复"

十一、网关高可用

维度	方案
多副本	Higress ≥ 2 副本 + ext_authz ≥ 2 副本，Ingress/LB 负载均衡
无状态	网关和 ext_authz 均不存储会话状态，RT 缓存在 Redis
滚动发布	Argo Rollouts，新副本就绪后再切流（先 ready 后 traffic switch）
故障转移	副本 down → LB 自动摘除，一致性哈希重映射到存活副本
Nacos 容灾	ext_authz 本地缓存 registry 快照（PostgreSQL），Nacos 不可用时降级寻址
Redis 容灾	Redis Sentinel（1 主 2 从），主从切换 < 30s
限流降级	Redis 不可用时降级为本地令牌桶（精度下降但不拒绝服务）