"""安全工具:密码哈希、JWT 签发/校验、API Key 生成。""" import secrets import datetime as dt from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from .config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") API_KEY_PREFIX = "pica_" # 便于人眼识别 def hash_password(plain: str) -> str: return pwd_context.hash(plain) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def create_access_token(user_id: int) -> str: """签发会话 JWT。短期、用于调用本服务的受保护页面接口。""" now = dt.datetime.utcnow() payload = { "sub": str(user_id), "iat": now, "exp": now + dt.timedelta(seconds=settings.jwt_expires_seconds), } return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm) def decode_access_token(token: str) -> Optional[int]: """校验 JWT,返回 user_id 或 None。""" try: payload = jwt.decode( token, settings.jwt_secret, algorithms=[settings.jwt_algorithm] ) sub = payload.get("sub") return int(sub) if sub is not None else None except (JWTError, ValueError, TypeError): return None def generate_api_key() -> str: """生成一个随机 API Key 明文。32 字节熵。""" return API_KEY_PREFIX + secrets.token_urlsafe(32)