fix(gateway): resolve high-risk issues #1-#4

- #1: JWT algorithm now injected via SUPABASE_JWT_ALGORITHM (default
  HS256, aligned with Supabase default signing); placeholder detection
  prevents 'configured' false impression causing whole-site 401
- #2: remove double rollback on streaming rate-limit overrun - the Lua
  script already atomically DECRs, manual rollback let users drain
  their own quotas into negative counts
- #3: error_pattern chain/router defensive validation - missing fields
  from LLM JSON output filtered instead of 500 (ValidationError)
- #4: GLM generate_vision returns actual clamped max_tokens; vision
  chain logs truncation warning when output approaches the limit

Author: Aparencia

server/ai-gateway/chains/error_pattern_chain.py

@@ -57,9 +57,25 @@ async def run(
             }
         """
         # 构建输入文本
+        # GW-2#3: 防御性取值——原实现直接键索引 e['correctAnswer']/e['userAnswer']，
+        # 调用方传入缺字段 dict 时 KeyError 导致 500；改为 get() + str() 归一化，
+        # 并过滤空值项（避免 prompt 中出现无意义的空错误记录）
+        def _safe_text(value: Any) -> str:
+            return str(value) if value is not None else ""
+
+        normalized_errors: list[tuple[str, str]] = []
+        for e in golden_errors[:20]:  # 最多分析前 20 条
+            if not isinstance(e, dict):
+                continue
+            correct = _safe_text(e.get("correctAnswer"))
+            user = _safe_text(e.get("userAnswer"))
+            if not correct and not user:
+                continue
+            normalized_errors.append((correct, user))
+
         errors_text = "\n".join([
-            f"【错误 #{i+1}】\n正确答案：{e['correctAnswer']}\n用户回答：{e['userAnswer']}\n---"
-            for i, e in enumerate(golden_errors[:20])  # 最多分析前 20 条
+            f"【错误 #{i+1}】\n正确答案：{correct}\n用户回答：{user}\n---"
+            for i, (correct, user) in enumerate(normalized_errors)
         ])
 
         template = self._load_prompt_template()

server/ai-gateway/chains/vision_extract_chain.py

@@ -278,8 +278,23 @@ async def run(
             _feature="vision_extract",
         )
 
+        # GW-2#4: 显式截断检测——GLM 等 provider 可能把 max_tokens clamp 到
+        # 低于请求值（full 模式请求 4096 被 clamp 到 1024），输出在接近上限时
+        # 大概率被截断。原实现静默返回残缺 JSON（结构化字段丢失）且照常缓存，
+        # 用户无感知；此处记录 truncated 告警供排查与降级重试决策
+        content = result["content"]
+        actual_max_tokens = result.get("max_tokens")
+        if actual_max_tokens and len(content) > actual_max_tokens * 3:
+            # 1 token ≈ 3 字符的保守估算（中文 1 token ≈ 1-2 字符，英文 ≈ 4）
+            logger.warning(
+                "视觉提取疑似截断: mode=%s, model=%s, content=%d 字符, "
+                "max_tokens=%d（provider clamp 后），结构化字段可能缺失",
+                effective_mode, result.get("model", self.model),
+                len(content), actual_max_tokens,
+            )
+
         # 解析结构化结果
-        structured = self._parse_response(result["content"])
+        structured = self._parse_response(content)
 
         return {
             "content": result["content"],

server/ai-gateway/config/app.py

@@ -4,6 +4,8 @@
 @ai-context: 汇聚运行环境、CORS、JWT 与中间件连接串等应用级配置。
 Supabase JWKS 端点从 SUPABASE_URL 推导（标准路径含 /auth/v1 前缀），
 亦可经 SUPABASE_JWKS_URL 显式覆盖。所有值均经环境变量注入，无硬编码密钥。
+@ai-context: GW-2#1——jwt_algorithm 由 SUPABASE_JWT_ALGORITHM 注入，默认 HS256
+（Supabase 默认对称密钥签发机制）；ES256 仅在开启自定义 JWT 时选用。
 """
 
 import os
@@ -25,7 +27,9 @@
         if origin.strip()
     ],
     "jwt_secret": os.getenv("SUPABASE_JWT_SECRET", ""),
-    "jwt_algorithm": "ES256",
+    # GW-2#1: 算法由环境变量注入（默认 HS256 与 Supabase 默认签发机制对齐）；
+    # 原硬编码 ES256 导致默认 HS256 项目（无 JWKS 端点）全站 401
+    "jwt_algorithm": os.getenv("SUPABASE_JWT_ALGORITHM", "HS256"),
     "supabase_url": _supabase_url,
     "supabase_jwks_url": os.getenv(
         "SUPABASE_JWKS_URL",

server/ai-gateway/middleware/auth.py

@@ -33,20 +33,40 @@
 # 不需要认证的路径白名单
 PUBLIC_PATHS = {"/health", "/health/quick", "/health/live"}
 
+# GW-2#1: 占位符识别——SUPABASE_URL 模板中的示例项目域名与常见示例密钥
+# 被视为"未配置"，避免"已配置"假象导致认证链路断裂后全站 401 且难排查
+_PLACEHOLDER_PROJECT_ID = "your-project-id"
+_PLACEHOLDER_SECRETS = {
+    "change-this-to-a-random-string-in-production",
+    "your-jwt-secret",
+    "placeholder",
+    "supabase_jwt_secret",
+}
+
+
+def _is_placeholder_text(value: str) -> bool:
+    """判断配置值是否为占位符/示例值"""
+    return value.strip().lower() in _PLACEHOLDER_SECRETS
+
+
 def _jwt_verification_configured() -> bool:
     """
     检查当前算法是否具备验证所需的密钥材料。
 
-    - HS256/RS256：需要 SUPABASE_JWT_SECRET（对称密钥或 PEM 公钥）
-    - ES256：需要 SUPABASE_JWKS_URL 或 SUPABASE_URL（用于获取 JWKS 公钥）
+    - HS256/RS256：需要 SUPABASE_JWT_SECRET（对称密钥或 PEM 公钥），
+      占位符/示例值视为未配置；
+    - ES256：需要 SUPABASE_JWKS_URL 或 SUPABASE_URL（用于获取 JWKS 公钥），
+      URL 含 your-project-id 占位符域名时视为未配置。
     """
     alg = APP_CONFIG.get("jwt_algorithm", "HS256")
     if alg == "ES256":
-        return bool(
+        jwks_url = (
             APP_CONFIG.get("supabase_jwks_url", "")
             or APP_CONFIG.get("supabase_url", "")
         )
-    return bool(APP_CONFIG.get("jwt_secret", ""))
+        return bool(jwks_url) and _PLACEHOLDER_PROJECT_ID not in jwks_url
+    secret = APP_CONFIG.get("jwt_secret", "")
+    return bool(secret) and not _is_placeholder_text(secret)
 
 
 # 启动时检查密钥配置
@@ -71,8 +91,12 @@ def _jwt_verification_configured() -> bool:
             stacklevel=2,
         )
 
-# 启动日志：输当前 JWT 验证算法
-logger.info("JWT 验证算法: %s", APP_CONFIG["jwt_algorithm"])
+# 启动日志：输当前 JWT 验证算法与配置状态（GW-2#1: 避免"已配置"假象）
+logger.info(
+    "JWT 验证算法: %s (验证材料已配置=%s)",
+    APP_CONFIG["jwt_algorithm"],
+    _jwt_verification_configured(),
+)
 
 
 def _normalize_pem_key(raw: str) -> str:
@@ -136,6 +160,13 @@ def _resolve_jwks_url() -> str:
     """解析 JWKS 端点 URL：优先 SUPABASE_JWKS_URL，其次从 SUPABASE_URL 推导"""
     jwks_url = APP_CONFIG.get("supabase_jwks_url", "")
     if jwks_url:
+        if _PLACEHOLDER_PROJECT_ID in jwks_url:
+            # GW-2#1: 占位符域名推导出的 JWKS 端点必然不可达，显式告警
+            #（此分支通常不会被走到——_jwt_verification_configured 已将其判为未配置）
+            logger.warning(
+                "ES256 验签 JWKS URL 含占位符域名 %s，请配置真实的 SUPABASE_URL/SUPABASE_JWKS_URL",
+                jwks_url,
+            )
         return jwks_url
     supabase_url = APP_CONFIG.get("supabase_url", "")
     if supabase_url:

server/ai-gateway/middleware/rate_limit.py

@@ -229,7 +229,13 @@ async def _lua_check_limit(key: str, limit: int, ttl: int) -> int:
 
 
 async def rollback_rate_limit(user_id: str, feature: str) -> None:
-    """回退频率限制计数器（原子 INCR 超出限额或请求失败时调用）"""
+    """回退频率限制计数器。
+
+    ⚠️ 仅允许在 check_rate_limit 返回 True（本次 INCR 已生效）之后、
+    请求未成功消费配额（非 2xx）时调用——此时功能级与全局计数均已 +1，
+    回滚两层是准确的。超限路径（check 返回 False）Lua 脚本已原子回滚，
+    禁止调用本函数，否则双重 DECR 会把配额刷穿（GW-2#2）。
+    """
     cache = get_cache()
     if not cache._client:
         return

server/ai-gateway/providers/glm_provider.py

@@ -249,6 +249,10 @@ async def generate_vision(
                 "tokens_used": tokens_used,
                 "model": model,
                 "latency_ms": latency_ms,
+                # GW-2#4: clamp 后实际生效的 max_tokens，供 chain 侧截断检测使用
+                #（与 generate_vision_multi 保持一致；原实现缺此字段，
+                # full 模式请求 4096 被 clamp 到 1024 后截断不可感知）
+                "max_tokens": max_tokens,
             }
 
         except Exception as e:

server/ai-gateway/routers/error_pattern.py

@@ -66,6 +66,39 @@ class ErrorPatternResponse(BaseModel):
 # ============================================================
 
 
+def _validate_pattern_item(item: Any) -> dict | None:
+    """校验 LLM 输出的错误模式项：缺失必填字段/类型非法时返回 None（过滤）。
+
+    GW-2#3: LLM JSON 输出截断或格式漂移时，PatternItem(**p) 严格构造会抛
+    ValidationError 导致 500——与 quiz_gen_chain._validate_question 相同的
+    逐项校验+过滤模式，非法项丢弃而非崩溃。
+    """
+    if not isinstance(item, dict):
+        return None
+    required = {"type", "keywords", "explanation", "suggestion"}
+    if not all(k in item and item[k] not in (None, "") for k in required):
+        return None
+    keywords = item.get("keywords")
+    if not isinstance(keywords, list) or not all(isinstance(k, str) for k in keywords):
+        return None
+    return item
+
+
+def _validate_top_offender(item: Any) -> dict | None:
+    """校验 LLM 输出的高频错误卡片项：缺 flashcardId 时返回 None（过滤）。"""
+    if not isinstance(item, dict):
+        return None
+    flashcard_id = item.get("flashcardId")
+    if not isinstance(flashcard_id, str) or not flashcard_id:
+        return None
+    try:
+        count = int(item.get("count", 0))
+    except (TypeError, ValueError):
+        # GW-2#3: count 非数字时容错为 0，而非让 pydantic 抛 500
+        count = 0
+    return {"flashcardId": flashcard_id, "count": count}
+
+
 @router.post("/error-pattern", response_model=ErrorPatternResponse, summary="分析黄金错误模式")
 async def error_pattern(request: Request, body: ErrorPatternRequest) -> ErrorPatternResponse:
     """
@@ -126,12 +159,27 @@ async def _run_chain(provider, model_name):
         await cache.set_ai_cache(cache_key, chain_result, expire=3600)
 
     # 构建响应对象
-    patterns = [
-        PatternItem(**p) for p in patterns_data
-    ]
-    top_offenders = [
-        TopOffender(**o) for o in top_offenders_data
-    ]
+    # GW-2#3: LLM 输出逐项校验+过滤，缺字段项不导致 500（截断/格式漂移降级）
+    patterns = []
+    dropped_patterns = 0
+    for p in patterns_data:
+        validated = _validate_pattern_item(p)
+        if validated is not None:
+            patterns.append(PatternItem(**validated))
+        else:
+            dropped_patterns += 1
+
+    top_offenders = []
+    for o in top_offenders_data:
+        validated = _validate_top_offender(o)
+        if validated is not None:
+            top_offenders.append(TopOffender(**validated))
+
+    if dropped_patterns > 0:
+        logger.warning(
+            "错误模式分析: %d 条模式项因缺字段被过滤（LLM 输出质量问题）",
+            dropped_patterns,
+        )
 
     return ErrorPatternResponse(
         patterns=patterns,

server/ai-gateway/routers/streaming.py

@@ -24,7 +24,7 @@
 from pydantic import BaseModel, Field
 
 from config import call_with_fallback_stream
-from middleware.rate_limit import check_rate_limit, rollback_rate_limit
+from middleware.rate_limit import check_rate_limit
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/v1/ai", tags=["流式输出"])
@@ -278,7 +278,9 @@ async def stream_ai(request: Request, feature: str, body: StreamRequest):
     # 此处复用与中间件一致的 Redis 限流逻辑（含功能级 + 全局每日总量）
     is_allowed, detail = await check_rate_limit(user_id, config_key)
     if not is_allowed:
-        await rollback_rate_limit(user_id, config_key)
+        # GW-2#2: 超限时禁止再次 rollback——Lua 脚本已在超限分支内原子 DECR，
+        # 路由层再 rollback 会双重回滚（feature 减到负数、global 被无条件 DECR），
+        # 恶意用户可反复刷超限请求把自己的配额刷低甚至清零
         raise HTTPException(status_code=429, detail=detail)
 
     try:

server/ai-gateway/tests/test_config.py

@@ -9,6 +9,7 @@
 - _resolve_model_name() 在 fallback 链中正确选择模型名
 """
 
+import os
 import sys
 from pathlib import Path
 
@@ -254,9 +255,12 @@ def test_evaluate_returns_glm_model(self):
 class TestConfigAssertions:
     """配置值精确断言（JWT 算法、API Key 校验、超时配置、Fallback 链尾）"""
 
-    def test_jwt_algorithm_is_es256(self):
-        """APP_CONFIG jwt_algorithm 必须为 ES256（与 Supabase JWKS 一致）"""
-        assert APP_CONFIG["jwt_algorithm"] == "ES256"
+    def test_jwt_algorithm_defaults_to_hs256(self):
+        """APP_CONFIG jwt_algorithm 默认 HS256（GW-2#1: 与 Supabase 默认对称密钥
+        签发机制对齐；原硬编码 ES256 导致无 JWKS 的默认项目全站 401）。
+        显式配置 SUPABASE_JWT_ALGORITHM 时使用配置值。"""
+        expected = os.getenv("SUPABASE_JWT_ALGORITHM", "HS256")
+        assert APP_CONFIG["jwt_algorithm"] == expected
 
     def test_is_valid_api_key_rejects_placeholder(self):
         """占位符 API Key 应被识别为无效"""