"""安全模块:密码哈希 + JWT""" import os import secrets import warnings from datetime import datetime, timedelta, timezone from typing import Optional import bcrypt from jose import jwt, JWTError # JWT 密钥:生产环境必须通过环境变量 JWT_SECRET_KEY 设置 SECRET_KEY = os.getenv("JWT_SECRET_KEY") if not SECRET_KEY: warnings.warn( "未设置 JWT_SECRET_KEY 环境变量,使用临时随机密钥(重启后所有 Token 将失效)。" "生产环境请务必设置 JWT_SECRET_KEY 环境变量。", stacklevel=2, ) SECRET_KEY = secrets.token_urlsafe(32) assert SECRET_KEY is not None SECRET_KEY_TYPED: str = SECRET_KEY ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_HOURS = int(os.getenv("ACCESS_TOKEN_EXPIRE_HOURS", "24")) def hash_password(password: str) -> str: pwd_bytes = password.encode("utf-8") # bcrypt 限制72字节,截断 pwd_bytes = pwd_bytes[:72] return bcrypt.hashpw(pwd_bytes, bcrypt.gensalt()).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: try: return bcrypt.checkpw(plain.encode("utf-8")[:72], hashed.encode("utf-8")) except Exception: return False def create_access_token( data: dict, expires_hours: int = ACCESS_TOKEN_EXPIRE_HOURS ) -> str: to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta(hours=expires_hours) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY_TYPED, algorithm=ALGORITHM) def decode_access_token(token: str) -> Optional[dict]: try: return jwt.decode(token, SECRET_KEY_TYPED, algorithms=[ALGORITHM]) except JWTError: return None