"""鉴权核心:密码哈希 + JWT 编解码 + API Token。""" from __future__ import annotations import hashlib import hmac import secrets from datetime import datetime, timedelta, timezone from typing import Any import jwt from passlib.context import CryptContext from app.config import settings # bcrypt 4.0.1 与 passlib 1.7.4 兼容 pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=12) def hash_password(plain: str) -> str: return pwd_ctx.hash(plain) def verify_password(plain: str, hashed: str) -> bool: try: return pwd_ctx.verify(plain, hashed) except Exception: return False # === JWT === def create_access_token(subject: str | int, extra: dict[str, Any] | None = None) -> str: now = datetime.now(timezone.utc) payload: dict[str, Any] = { "sub": str(subject), "type": "access", "iat": int(now.timestamp()), "exp": int((now + timedelta(minutes=settings.access_token_ttl_min)).timestamp()), } if extra: payload.update(extra) return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm) def create_refresh_token(subject: str | int) -> str: now = datetime.now(timezone.utc) payload = { "sub": str(subject), "type": "refresh", "iat": int(now.timestamp()), "exp": int((now + timedelta(days=settings.refresh_token_ttl_day)).timestamp()), "jti": secrets.token_urlsafe(16), } return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm) def decode_token(token: str) -> dict[str, Any]: return jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]) # === API Token(给 Android 用)=== def generate_api_token() -> tuple[str, str]: """返回 (raw_token, token_hash)。raw_token 只显示一次。""" raw = secrets.token_urlsafe(32) return raw, hash_api_token(raw) def hash_api_token(raw: str) -> str: # 简单 sha256 即可(随机性已经够) return hashlib.sha256(raw.encode()).hexdigest() def constant_time_eq(a: str, b: str) -> bool: return hmac.compare_digest(a, b)