浅念
f938118fee
LoginLimiter._fails 与 rate_limit._buckets 只增不删——懒清理只覆盖被 回访的键，扫号换 IP/设备轮换产生的键永不被回访 → 进程长跑缓慢累积。 两处各加 sweep()（全表清「全部时间戳均已出窗」的键，窗沿 <= 与既有 保留判据 t > now - W 同沿），main.lifespan 起周期任务（300s loop） 启动创建/关停取消（cancel 后 await 落地，不留 pending task）。 Co-Authored-By: Claude Code <noreply@anthropic.com>
319 lines
13 KiB
Python
319 lines
13 KiB
Python
"""平台认证内核(Phase 2 Task 2):哈希上移 + 内存 token + 登录限流 + 依赖。
|
|
|
|
本模块零 app 内依赖(不 import db/main)——hash_password/verify_password
|
|
从 db.py 上移到此为单一事实源(不留双份漂移),db.py 反向 re-export 兼容
|
|
既有导入;verify 签名解 ORM 绑定为纯参式 (password, password_hash, salt)
|
|
(Task 1 加固 M-2b)。
|
|
|
|
构成:
|
|
- hash_password / verify_password: 版本化 scrypt,格式与解析兼容不变
|
|
("scrypt$<n>$<r>$<p>$<salt_hex>$<hash_hex>";无前缀过渡行 = salt 列 +
|
|
默认参,兼容照旧)
|
|
- TokenStore: 内存 token 表,secrets.token_urlsafe(32),TTL 默认 8h(可注入)
|
|
- LoginLimiter: (username, ip) 滑动窗失败计数,5 次/15 分钟(窗长可注入),
|
|
风格照抄 engine/services/rate_limit.py
|
|
- UserContext + get_current_user / require_super / require_tenant(slug) /
|
|
enforce_password_changed / platform_user: FastAPI 依赖
|
|
|
|
单进程纪律: TokenStore/LoginLimiter 均进程内存态,uvicorn 必须 --workers 1
|
|
(>1 时各 worker 各持一份:token 在 A 签发 B 不识 → 401 误伤;限流配额随
|
|
worker 数翻倍)。与 rate_limit.py 同一取舍(单 worker 部署形态)。
|
|
|
|
安全语义:
|
|
- 防枚举: 用户不存在/密码错/停用同一 401 静态文案(auth_routes);用户不
|
|
存在时也跑一次等代价哑 scrypt,抹平时序侧信道差。
|
|
- 密码/token 永不进日志: 本层与路由层不 log 任何凭据字段,错误 detail
|
|
均静态串。
|
|
- token 走 Authorization: Bearer 头,无 Cookie 免 CSRF(spec §7.2)。
|
|
- 偏差注记: spec §7.2 写 bcrypt/argon2,实现用 hashlib.scrypt(stdlib 零
|
|
新依赖,抗 GPU 性能同级)——有意取舍。单次 ~50ms CPU,登录/改密路径
|
|
可接受(不在每请求热路径——依赖只查内存 token 表)。
|
|
"""
|
|
import functools
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import time
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, Request
|
|
|
|
# ---- 密码哈希(scrypt,从 db.py 上移;偏差注记见模块 docstring)------------------
|
|
|
|
_SCRYPT_N = 2 ** 14
|
|
_SCRYPT_R = 8
|
|
_SCRYPT_P = 1
|
|
|
|
|
|
def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]:
|
|
"""scrypt(password, salt) → (password_hash, salt_hex)。
|
|
|
|
password_hash 为版本化串 "scrypt$<n>$<r>$<p>$<salt_hex>$<hash_hex>":
|
|
参数与摘要同存,将来提参/换算法时旧行自解释,有迁移路径(M-1)。
|
|
salt 返回值(列存)与串内嵌的同一盐 hex——单一写方在此同步两处,
|
|
verify 以串内为准。salt 缺省每次随机 16 字节;显式传 salt 可复现
|
|
(测试/verify)。"""
|
|
if salt is None:
|
|
salt = secrets.token_bytes(16)
|
|
digest = hashlib.scrypt(
|
|
password.encode("utf-8"), salt=salt,
|
|
n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P)
|
|
versioned = (
|
|
f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}"
|
|
f"${salt.hex()}${digest.hex()}")
|
|
return versioned, salt.hex()
|
|
|
|
|
|
def verify_password(password: str, password_hash: str, salt: str) -> bool:
|
|
"""解析版本化串重算比对,常量时间比较(hmac.compare_digest)。
|
|
|
|
纯参式 (password, password_hash, salt)(M-2b 解 ORM 绑定):salt 仅在
|
|
无前缀过渡行时使用(列存 hex),版本化行以串内为准。
|
|
|
|
解析兼容(M-1): 带 "scrypt$" 前缀 → 参数/盐/摘要全取自串内;无前缀
|
|
(过渡/手工行)→ 盐取 salt 列、参数取当前默认。任何解析失败(坏 hex/
|
|
坏段数/坏整数)按验败返回 False 而非炸——存储异常不该变 500。"""
|
|
stored = password_hash
|
|
try:
|
|
if stored.startswith("scrypt$"):
|
|
parts = stored.split("$")
|
|
if len(parts) != 6:
|
|
return False
|
|
_, n_s, r_s, p_s, salt_hex, hash_hex = parts
|
|
n, r, p = int(n_s), int(r_s), int(p_s)
|
|
salt_bytes = bytes.fromhex(salt_hex)
|
|
expected = bytes.fromhex(hash_hex)
|
|
else:
|
|
salt_bytes = bytes.fromhex(salt)
|
|
expected = bytes.fromhex(stored)
|
|
n, r, p = _SCRYPT_N, _SCRYPT_R, _SCRYPT_P
|
|
except ValueError:
|
|
return False
|
|
actual = hashlib.scrypt(
|
|
password.encode("utf-8"), salt=salt_bytes, n=n, r=r, p=p)
|
|
return hmac.compare_digest(expected, actual)
|
|
|
|
|
|
@functools.lru_cache(maxsize=1)
|
|
def dummy_credentials() -> tuple[str, str]:
|
|
"""固定哑凭据 (password_hash, salt)——登录查无用户时跑一次同代价
|
|
verify,抹平"用户不存在秒回/密码错 50ms"的时序枚举面。公共面:
|
|
auth_routes 的防枚举时序等化依赖它。"""
|
|
return hash_password("", salt=b"\x00" * 16)
|
|
|
|
|
|
# ---- TokenStore -----------------------------------------------------------------
|
|
|
|
TOKEN_TTL_SECONDS = 8 * 3600 # 默认 8h(可注入,测试拨窗用)
|
|
|
|
|
|
@dataclass
|
|
class _TokenRecord:
|
|
user_id: int
|
|
role: str # "super" | "tenant"
|
|
tenant_slug: str | None
|
|
must_change_password: bool
|
|
expires_at: float # time.time() 纪元秒
|
|
|
|
|
|
class TokenStore:
|
|
"""内存 token 表: token → 用户快照。单进程纪律见模块 docstring。
|
|
|
|
must_change_password 是签发时快照;change-password 就地清零当前
|
|
token 的标志(auth_routes 语义注记),其他会话的旧 token 保持各自
|
|
快照(重登后消除)。"""
|
|
|
|
def __init__(self, ttl_seconds: float = TOKEN_TTL_SECONDS) -> None:
|
|
self.ttl_seconds = ttl_seconds
|
|
self._tokens: dict[str, _TokenRecord] = {}
|
|
|
|
def issue(
|
|
self, *, user_id: int, role: str, tenant_slug: str | None,
|
|
must_change_password: bool) -> str:
|
|
self._prune() # 懒清理(签发时)
|
|
token = secrets.token_urlsafe(32) # 256 bit,碰撞可忽略
|
|
self._tokens[token] = _TokenRecord(
|
|
user_id=user_id, role=role, tenant_slug=tenant_slug,
|
|
must_change_password=must_change_password,
|
|
expires_at=time.time() + self.ttl_seconds)
|
|
return token
|
|
|
|
def validate(self, token: str) -> _TokenRecord | None:
|
|
"""有效返回快照;缺失/过期返回 None(过期顺手删除——懒 prune)。"""
|
|
record = self._tokens.get(token)
|
|
if record is None:
|
|
return None
|
|
if record.expires_at <= time.time():
|
|
del self._tokens[token]
|
|
return None
|
|
return record
|
|
|
|
def revoke(self, token: str) -> None:
|
|
"""logout: 立即失效。不存在也静默(重复 logout 幂等)。"""
|
|
self._tokens.pop(token, None)
|
|
|
|
def clear_must_change(self, token: str) -> None:
|
|
"""改密成功后就地清当前 token 的引导标志(见类 docstring)。"""
|
|
record = self._tokens.get(token)
|
|
if record is not None:
|
|
record.must_change_password = False
|
|
|
|
def _prune(self) -> None:
|
|
now = time.time()
|
|
for token in [t for t, r in self._tokens.items()
|
|
if r.expires_at <= now]:
|
|
del self._tokens[token]
|
|
|
|
|
|
# ---- LoginLimiter ---------------------------------------------------------------
|
|
|
|
LOGIN_WINDOW_SECONDS = 15 * 60
|
|
LOGIN_MAX_FAILS = 5
|
|
|
|
|
|
class LoginLimiter:
|
|
"""(username, ip) 键滑动窗失败计数: 窗内第 5 次失败后拒绝后续尝试,
|
|
窗过恢复;登录成功即清该键计数(正确密码不受历史失败连累)。
|
|
|
|
单进程内存态(同 TokenStore 纪律);风格照抄 rate_limit.py。只计
|
|
失败不计成功——正常用户重复登录不锁。"""
|
|
|
|
def __init__(
|
|
self,
|
|
window_seconds: float = LOGIN_WINDOW_SECONDS,
|
|
max_fails: int = LOGIN_MAX_FAILS) -> None:
|
|
self.window_seconds = window_seconds
|
|
self.max_fails = max_fails
|
|
self._fails: dict[tuple[str, str], list[float]] = defaultdict(list)
|
|
|
|
def allow(self, username: str, ip: str) -> bool:
|
|
"""窗内失败数是否仍低于上限(不记录——记录走 record_fail)。"""
|
|
self._prune((username, ip), time.time())
|
|
return len(self._fails.get((username, ip), ())) < self.max_fails
|
|
|
|
def record_fail(self, username: str, ip: str) -> None:
|
|
key = (username, ip)
|
|
now = time.time()
|
|
self._prune(key, now)
|
|
self._fails[key].append(now)
|
|
|
|
def reset(self, username: str, ip: str) -> None:
|
|
"""登录成功清计数(窗口恢复的正路)。"""
|
|
self._fails.pop((username, ip), None)
|
|
|
|
def _prune(self, key: tuple[str, str], now: float) -> None:
|
|
fails = self._fails.get(key)
|
|
if fails is None:
|
|
return
|
|
kept = [t for t in fails if t > now - self.window_seconds]
|
|
if kept:
|
|
self._fails[key] = kept
|
|
else:
|
|
self._fails.pop(key, None)
|
|
|
|
def sweep(self, now: float | None = None) -> int:
|
|
"""清扫过期键(spec 2026-10-19 §7——内存上限铁律):删「全部时间戳
|
|
均已出窗」的键,返回清除数。与 _prune 的单键懒清理互补——allow/
|
|
record_fail 只顺手清被回访的键,失败一次即消失的键(扫号换 IP/
|
|
用户名)永不被回访,靠本方法全表兜底;由 main.lifespan 周期任务
|
|
调用。时间戳按记录序追加,ts[-1] 即最新——最新出窗则全键出窗。"""
|
|
now = time.time() if now is None else now
|
|
dead = [k for k, ts in self._fails.items()
|
|
if not ts or ts[-1] <= now - self.window_seconds]
|
|
for k in dead:
|
|
del self._fails[k]
|
|
return len(dead)
|
|
|
|
|
|
# 模块级单例: 路由与依赖共用同一份进程内存态(同 rate_limit._buckets 模式)
|
|
token_store = TokenStore()
|
|
login_limiter = LoginLimiter()
|
|
|
|
|
|
def _reset_for_test() -> None:
|
|
"""单例复位(测试隔离): token/失败计数全清,注入的 TTL/窗长还原默认。
|
|
conftest autouse 前后各跑一次(照抄 _reset_rate_limit 纪律)。"""
|
|
token_store._tokens.clear()
|
|
token_store.ttl_seconds = TOKEN_TTL_SECONDS
|
|
login_limiter._fails.clear()
|
|
login_limiter.window_seconds = LOGIN_WINDOW_SECONDS
|
|
|
|
|
|
# ---- FastAPI 依赖 ------------------------------------------------------------------
|
|
|
|
@dataclass(frozen=True)
|
|
class UserContext:
|
|
user_id: int
|
|
role: str # "super" | "tenant"
|
|
tenant_slug: str | None # super 为 None
|
|
must_change_password: bool
|
|
|
|
|
|
def bearer_token(request: Request) -> str | None:
|
|
"""Authorization: Bearer <token> → token;非该形态返回 None。
|
|
|
|
公共面:get_current_user 依赖与 auth_routes 的 logout/change-password
|
|
(依赖验过后重读头取原串)共用同一解析,不各写一份漂移。"""
|
|
header = request.headers.get("authorization", "")
|
|
if not header.startswith("Bearer "):
|
|
return None
|
|
return header[len("Bearer "):].strip()
|
|
|
|
|
|
async def get_current_user(request: Request) -> UserContext:
|
|
"""Bearer 解析 → TokenStore 查验 → UserContext。
|
|
|
|
缺头/坏格式/未知/过期一律 401(静态文案,不细分原因——token 本身
|
|
就是凭据,细化文案只会泄露信息)。"""
|
|
token = bearer_token(request)
|
|
record = token_store.validate(token) if token else None
|
|
if record is None:
|
|
raise HTTPException(status_code=401, detail="未认证或凭证已失效")
|
|
return UserContext(
|
|
user_id=record.user_id,
|
|
role=record.role,
|
|
tenant_slug=record.tenant_slug,
|
|
must_change_password=record.must_change_password)
|
|
|
|
|
|
CurrentUser = Annotated[UserContext, Depends(get_current_user)]
|
|
|
|
|
|
def enforce_password_changed(user: CurrentUser) -> UserContext:
|
|
"""must_change_password=True → 409:除 login/change-password 外的受保护
|
|
端点都要过本依赖(引导/重置出的初始密码不得只当摆设——spec 强制
|
|
语义;auth_routes.logout 是 Task 2 的挂载位)。"""
|
|
if user.must_change_password:
|
|
raise HTTPException(status_code=409, detail="必须先修改初始密码")
|
|
return user
|
|
|
|
|
|
# 组合依赖: Task 5 起全部 /api/platform/* 受保护端点统一注
|
|
# `user: platform_user`(= get_current_user + enforce_password_changed,
|
|
# 链式 Depends 自动先认证后查标志)。
|
|
platform_user = Annotated[UserContext, Depends(enforce_password_changed)]
|
|
|
|
|
|
def require_super(user: platform_user) -> UserContext:
|
|
"""非 super → 403(端点级覆盖在 Task 5;本任务依赖直测)。
|
|
|
|
参数走 platform_user(认证+已改密连锁):Task 5 路由挂本依赖时
|
|
must_change 409 门自动生效,不可能静默漏挂。"""
|
|
if user.role != "super":
|
|
raise HTTPException(status_code=403, detail="需要平台超管权限")
|
|
return user
|
|
|
|
|
|
def require_tenant(slug: str):
|
|
"""依赖工厂: super 或指定租户的管理员放行,否则 403。
|
|
|
|
参数同 require_super 走 platform_user——409 门连锁,理由见其
|
|
docstring。"""
|
|
def _dep(user: platform_user) -> UserContext:
|
|
if user.role == "super" or user.tenant_slug == slug:
|
|
return user
|
|
raise HTTPException(status_code=403, detail="无权访问该租户资源")
|
|
return _dep
|