171 lines
7.5 KiB
Python
171 lines
7.5 KiB
Python
# -*- 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}"
|