JWT 防御验证流水线、CI 门禁与结果矩阵

单次人工排查只能确认当前时刻的某个接口，而认证中间件升级、网关规则调整或新微服务上线随时可能重新引入 alg=none、宽泛 algorithms 列表或差异化报错。防御工程化的核心是把 09-toolchain-operations.md 中的五阶段变体矩阵反转为 CI/CD 自动化拒绝门禁——任何一条攻击路径返回非 401 或暴露差异化错误哈希，即阻断发布。

五阶段防御门禁架构

Stage 1: 静态代码与依赖扫描
  └─ 阻断 verify_signature=False / parseClaimsJwt() / 硬编码密钥 / 过时 JWT 库
Stage 2: 密钥与配置基线校验
  └─ 确认 HMAC 密钥 >= 256-bit 高熵或已锁定 RS256/ES256 单一非对称算法与静态 jwks_uri
Stage 3: 负向 Token 变体生成（Negative Case Matrix）
  └─ 自动生成覆盖 01-08 全线攻击向量的标准失败用例集（JSONL）
Stage 4: 本地/预发验证器批量运行
  └─ 对所有负向用例验证返回同一拒绝错误，且不触发外连网络与文件/SQL 异常
Stage 5: 收敛比对与发布放行
  └─ 核对状态码全为 401、响应体摘要全等、无库指纹泄漏，全绿通过后放行部署

防御工作流

1. 构建标准负向矩阵：在测试夹具中固定生成 alg=none、大小写 NoNe、RS->HS 混淆、越权 kid、外部 jku、缺失/过期 exp、错误 aud/iss、typ 混用等标准对抗样本。
2. 统一异常收敛检查：不仅断言验证函数抛出拒绝异常，还验证对外 HTTP 响应状态码与错误消息字节级完全一致，杜绝侧信道库指纹识别。
3. 验证无副作用隔离：在沙箱内监听局部回环端口或 mock HTTP 客户端，确保包含 jku/x5u/kid=http://... 的用例绝不发出任何出站请求。
4. 集成发布前阻断门禁：将下方批量验证器接入单元测试与预发集成测试（每次变更认证中间件或依赖升级必跑）。

Python 自动化 JWT 防御回归门禁运行器

import base64
import hashlib
import hmac
import json
import time
from typing import Any, Callable


def _b64u(obj: dict[str, Any]) -> str:
    raw = json.dumps(obj, separators=(",", ":")).encode("utf-8")
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def build_negative_case_matrix(
    valid_payload: dict[str, Any],
    hmac_secret_for_test: bytes,
    known_kid: str = "key-v1",
) -> list[dict[str, str]]:
    """生成覆盖 01-08 全部防线的标准化负向回归用例集。"""
    def sign_hs(h: dict[str, Any], p: dict[str, Any], key: bytes = hmac_secret_for_test) -> str:
        msg = f"{_b64u(h)}.{_b64u(p)}"
        sig = base64.urlsafe_b64encode(hmac.new(key, msg.encode("ascii"), hashlib.sha256).digest()).rstrip(b"=").decode("ascii")
        return f"{msg}.{sig}"

    base_h = {"alg": "HS256", "typ": "at+jwt", "kid": known_kid}
    cases = [
        {"case": "01_alg_none_lower", "token": f"{_b64u({'alg': 'none', 'typ': 'at+jwt'})}.{_b64u(valid_payload)}."},
        {"case": "01_alg_none_mixed", "token": f"{_b64u({'alg': 'NoNe', 'typ': 'at+jwt'})}.{_b64u(valid_payload)}."},
        {"case": "01_alg_array", "token": f"{_b64u({'alg': ['none', 'HS256'], 'typ': 'at+jwt'})}.{_b64u(valid_payload)}."},
        {"case": "02_rs_to_hs_confusion", "token": sign_hs({"alg": "RS256", "typ": "at+jwt", "kid": known_kid}, valid_payload)},
        {"case": "03_wrong_secret", "token": sign_hs(base_h, valid_payload, b"weak-secret-12345678901234567890")},
        {"case": "04_kid_traversal", "token": sign_hs({**base_h, "kid": "../../dev/null"}, valid_payload, b"")},
        {"case": "04_kid_sqli", "token": sign_hs({**base_h, "kid": "x' UNION SELECT 'evil'--"}, valid_payload, b"evil")},
        {"case": "05_jku_hijack", "token": sign_hs({**base_h, "jku": "https://attacker.example/jwks.json"}, valid_payload)},
        {"case": "05_x5u_hijack", "token": sign_hs({**base_h, "x5u": "https://attacker.example/cert.pem"}, valid_payload)},
        {"case": "06_exp_expired", "token": sign_hs(base_h, {**valid_payload, "exp": 1000000000})},
        {"case": "06_exp_missing", "token": sign_hs(base_h, {k: v for k, v in valid_payload.items() if k != "exp"})},
        {"case": "06_aud_mismatch", "token": sign_hs(base_h, {**valid_payload, "aud": "other-service"})},
        {"case": "06_iss_mismatch", "token": sign_hs(base_h, {**valid_payload, "iss": "https://other-idp.example"})},
        {"case": "06_typ_id_confusion", "token": sign_hs({**base_h, "typ": "JWT"}, valid_payload)},
        {"case": "08_kid_object_type", "token": sign_hs({**base_h, "kid": {"toString": "x"}}, valid_payload)},
    ]
    return cases


def evaluate_defense_gate(
    verify_endpoint: Callable[[str], tuple[int, str]],
    positive_token: str,
    negative_cases: list[dict[str, str]],
) -> dict[str, Any]:
    """执行门禁验证：正例必须 200，负例必须全部 401 且错误响应摘要完全收敛。"""
    pos_status, _ = verify_endpoint(positive_token)
    if pos_status != 200:
        raise AssertionError(f"positive baseline failed with status {pos_status}")

    error_hashes: set[str] = set()
    failures: list[dict[str, Any]] = []
    for item in negative_cases:
        status, body = verify_endpoint(item["token"])
        body_hash = hashlib.sha256(body.encode("utf-8", errors="ignore")).hexdigest()[:16]
        error_hashes.add(f"{status}:{body_hash}")
        if status != 401:
            failures.append({"case": item["case"], "status": status, "body": body})

    if failures:
        raise AssertionError(f"defense gate bypassed: {failures}")
    if len(error_hashes) != 1:
        raise AssertionError(f"non-convergent error responses (side-channel leak risk): {error_hashes}")

    return {"positive": 200, "negative_total": len(negative_cases), "converged_error": next(iter(error_hashes))}

验证与判定标准

运行下方自检脚本以自包含验证器驱动门禁运行器，确认全量 15 个负向用例全部以同一 401 响应摘要收敛拒绝：

def run_toolchain_defense_gate_selfcheck() -> None:
    secret = b"S" * 32
    now = int(time.time())
    valid_payload = {
        "sub": "u-100",
        "iss": "https://auth.example.com",
        "aud": "api.example.com",
        "nbf": now - 10,
        "exp": now + 600,
        "jti": "jti-good",
    }
    header = {"alg": "HS256", "typ": "at+jwt", "kid": "key-v1"}
    msg = f"{_b64u(header)}.{_b64u(valid_payload)}"
    sig = base64.urlsafe_b64encode(hmac.new(secret, msg.encode(), hashlib.sha256).digest()).rstrip(b"=").decode()
    good_token = f"{msg}.{sig}"

    def hardened_endpoint(tok: str) -> tuple[int, str]:
        try:
            parts = tok.split(".")
            if len(parts) != 3 or not parts[2]:
                raise ValueError("bad format")
            h = json.loads(base64.urlsafe_b64decode(parts[0] + "=" * (-len(parts[0]) % 4)))
            p = json.loads(base64.urlsafe_b64decode(parts[1] + "=" * (-len(parts[1]) % 4)))
            s = base64.urlsafe_b64decode(parts[2] + "=" * (-len(parts[2]) % 4))
            if not isinstance(h, dict) or not isinstance(p, dict):
                raise ValueError("bad json")
            if h.get("alg") != "HS256" or h.get("typ") != "at+jwt" or h.get("kid") != "key-v1":
                raise ValueError("bad header")
            if any(k in h for k in ("jku", "x5u", "jwk", "x5c")):
                raise ValueError("forbidden key source")
            exp_sig = hmac.new(secret, f"{parts[0]}.{parts[1]}".encode(), hashlib.sha256).digest()
            if not hmac.compare_digest(exp_sig, s):
                raise ValueError("bad sig")
            if p.get("iss") != "https://auth.example.com" or p.get("aud") != "api.example.com":
                raise ValueError("bad iss/aud")
            if not isinstance(p.get("exp"), int) or now >= p["exp"]:
                raise ValueError("expired")
            return 200, '{"ok":true}'
        except Exception:
            # 统一收敛错误输出，禁止泄露具体失败阶段
            return 401, '{"error":"unauthorized"}'

    matrix = build_negative_case_matrix(valid_payload, secret)
    report = evaluate_defense_gate(hardened_endpoint, good_token, matrix)
    assert report["negative_total"] == 15
    print("PASS: JWT defense CI gate passed:", report)


if __name__ == "__main__":
    run_toolchain_defense_gate_selfcheck()

判定标准：

1. 100% 负向拦截：[01]–[08] 全线 15 类标准对抗 Token 在门禁测试中全部返回 401。
2. 错误摘要单一化：所有失败样本的 (status, sha256(body)) 集合大小严格等于 1，确保没有信息泄露分支。
3. 基线正例可用：合法签发的 at+jwt 在相同端点保持 200 正常通行，避免加固误伤生产流量。

MCP 工具映射

防御阶段	MCP 工具	说明
防御路线索引	kb_router	按 JWT门禁、回归矩阵、错误收敛 检索对应防御条目
规程读取	kb_read_file	读取 00-overview-defense 及 01–08 对应加固文档
目录编排核查	kb_catalog	列出 02-auth/jwt/ 下全部配对防御文件确认覆盖完整度