# -*- 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:（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 '，返回 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 ") 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}"