feat: Phase 3 FBA UI 预构建产物 + Phase 4 按钮级权限（fba_jwt + perms）（Epic #159）

2026-08-06 10:18:15 +08:00
parent 067ba0abe1
commit 4ea81cea91
7 changed files with 342 additions and 4 deletions
@@ -0,0 +1,170 @@
+# -*- coding: utf-8 -*-
+"""FBA JWT 校验（Epic #159 · Phase 4）—— iAOP 业务接口接入 FBA 统一认证。
+
+背景：认证/RBAC 基座已迁移到 FBA（fastapi-best-architecture）。
+FBA access token 为 HS256 JWT，claims：{sub: 用户ID, session_uuid, exp}。
+iAOP 后端为零依赖标准库实现，本模块用 hmac/hashlib 手写 HS256 校验，
+不引入 PyJWT。
+
+用法（任意 iAOP HTTP 服务内）：
+
+    from core.auth.fba_jwt import FbaAuth, FbaAuthError
+
+    fba = FbaAuth.from_env()            # 读 FBA_TOKEN_SECRET_KEY / FBA_API_BASE
+    if fba is not None:                 # 未配置 → 调用方走旧 core/auth 会话
+        claims = fba.verify(authorization_header)   # 无效抛 FbaAuthError(401)
+        fba.require_perm(authorization_header, "iaop:admin")  # 无权抛 403
+
+权限码（按钮级）：
+    前端菜单播种时已写入 perms=iaop:<module>（deploy/fba/seed_iaop_menus.py）。
+    require_perm() 携用户 token 回源 FBA GET /auth/codes 校验（带 TTL 缓存，
+    避免每请求打满 FBA）。超管在 FBA 侧自动拥有全部权限码。
+
+双轨过渡：FBA_TOKEN_SECRET_KEY 未配置时 from_env() 返回 None，
+业务接口应继续接受旧 core/auth HMAC 会话（见 auth_api.require_auth），
+待 Phase 5 收尾时再移除旧轨。
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import time
+import urllib.request
+import urllib.error
+from http import HTTPStatus
+from typing import Optional
+
+
+class FbaAuthError(Exception):
+    """FBA 认证/授权失败。status: 401 未认证，403 无权限，503 FBA 不可达。"""
+
+    def __init__(self, status: int, message: str) -> None:
+        super().__init__(message)
+        self.status = status
+        self.message = message
+
+
+def _b64url_decode(seg: str) -> bytes:
+    pad = "=" * (-len(seg) % 4)
+    return base64.urlsafe_b64decode(seg + pad)
+
+
+def _b64url_encode(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+
+
+class FbaAuth:
+    """FBA JWT 校验器（HS256，与 FBA backend/common/security/jwt.py 对齐）。"""
+
+    def __init__(self, secret_key: str, api_base: str = "http://127.0.0.1:8001/api/v1",
+                 perm_cache_ttl: int = 300) -> None:
+        if not secret_key:
+            raise ValueError("secret_key 不能为空")
+        self._secret = secret_key.encode()
+        self.api_base = api_base.rstrip("/")
+        self.perm_cache_ttl = perm_cache_ttl
+        # token → (expires_at, codes set)；简单内存缓存，进程级
+        self._perm_cache: dict[str, tuple[float, set]] = {}
+
+    @classmethod
+    def from_env(cls) -> Optional["FbaAuth"]:
+        """读环境变量构造；未配置 FBA_TOKEN_SECRET_KEY 时返回 None（降级旧轨）。"""
+        secret = os.environ.get("FBA_TOKEN_SECRET_KEY", "").strip()
+        if not secret:
+            return None
+        return cls(secret, os.environ.get(
+            "FBA_API_BASE", "http://127.0.0.1:8001/api/v1"))
+
+    # ------------------------------------------------------------------ JWT
+    def verify(self, authorization: str) -> dict:
+        """校验 'Bearer <token>'，返回 claims；失败抛 FbaAuthError(401)。"""
+        token = self._extract_bearer(authorization)
+        parts = token.split(".")
+        if len(parts) != 3:
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "Token 格式无效")
+        signing_input = (parts[0] + "." + parts[1]).encode()
+        expect = _b64url_encode(hmac.new(self._secret, signing_input,
+                                         hashlib.sha256).digest())
+        if not hmac.compare_digest(expect, parts[2]):
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "Token 签名无效")
+        try:
+            header = json.loads(_b64url_decode(parts[0]))
+            payload = json.loads(_b64url_decode(parts[1]))
+        except (ValueError, UnicodeDecodeError):
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "Token 解析失败")
+        if header.get("alg") != "HS256":
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "Token 算法不受支持")
+        if not payload.get("sub") or not payload.get("session_uuid"):
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "Token 缺少必要字段")
+        exp = payload.get("exp")
+        if not isinstance(exp, (int, float)) or exp < time.time():
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "Token 已过期")
+        return payload
+
+    @staticmethod
+    def _extract_bearer(authorization: str) -> str:
+        if not authorization:
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "未携带认证信息")
+        parts = authorization.split(None, 1)
+        if len(parts) != 2 or parts[0].lower() != "bearer":
+            raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "认证头格式应为 Bearer <token>")
+        return parts[1].strip()
+
+    # ------------------------------------------------------------- 权限码
+    def get_codes(self, authorization: str) -> set:
+        """携 token 回源 FBA /auth/codes 取权限码集合（带 TTL 缓存）。"""
+        token = self._extract_bearer(authorization)
+        now = time.time()
+        hit = self._perm_cache.get(token)
+        if hit and hit[0] > now:
+            return hit[1]
+        req = urllib.request.Request(
+            self.api_base + "/auth/codes",
+            headers={"Authorization": f"Bearer {token}"})
+        try:
+            with urllib.request.urlopen(req, timeout=5) as r:
+                d = json.loads(r.read().decode())
+        except urllib.error.HTTPError as e:
+            if e.code in (401, 403):
+                raise FbaAuthError(HTTPStatus.UNAUTHORIZED, "Token 已失效")
+            raise FbaAuthError(HTTPStatus.SERVICE_UNAVAILABLE,
+                               f"FBA 权限码查询失败：HTTP {e.code}")
+        except OSError:
+            raise FbaAuthError(HTTPStatus.SERVICE_UNAVAILABLE, "FBA 服务不可达")
+        if d.get("code") != 200 or not isinstance(d.get("data"), list):
+            raise FbaAuthError(HTTPStatus.SERVICE_UNAVAILABLE, "FBA 权限码响应异常")
+        codes = set(d["data"])
+        self._perm_cache[token] = (now + self.perm_cache_ttl, codes)
+        # 简单清理过期缓存项
+        for k in [k for k, v in self._perm_cache.items() if v[0] <= now]:
+            self._perm_cache.pop(k, None)
+        return codes
+
+    def require_perm(self, authorization: str, perm: str) -> dict:
+        """校验 token + 权限码；返回 claims。401/403 抛 FbaAuthError。"""
+        claims = self.verify(authorization)
+        codes = self.get_codes(authorization)
+        if perm not in codes:
+            raise FbaAuthError(HTTPStatus.FORBIDDEN, f"缺少权限：{perm}")
+        return claims
+
+
+# ---------------------------------------------------------------------------
+# 自测试用：生成与 FBA 同构的 HS256 token（tests 使用）
+# ---------------------------------------------------------------------------
+def make_test_token(secret: str, user_id: int = 1,
+                    session_uuid: str = "test-session",
+                    exp_offset: int = 3600) -> str:
+    header = _b64url_encode(json.dumps(
+        {"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode())
+    payload = _b64url_encode(json.dumps({
+        "sub": str(user_id), "session_uuid": session_uuid,
+        "exp": time.time() + exp_offset,
+    }, separators=(",", ":")).encode())
+    sig = _b64url_encode(hmac.new(secret.encode(),
+                                  f"{header}.{payload}".encode(),
+                                  hashlib.sha256).digest())
+    return f"{header}.{payload}.{sig}"