"""密码哈希与 JWT 签发/校验""" from datetime import datetime, timedelta, timezone from typing import Any, Dict, Optional import bcrypt from jose import JWTError, jwt from app.core.config import get_settings def hash_password(plain: str) -> str: return bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(plain: str, password_hash: str) -> bool: try: return bcrypt.checkpw(plain.encode("utf-8"), password_hash.encode("utf-8")) except (ValueError, TypeError): return False def create_access_token(sub: str, extra: Optional[Dict[str, Any]] = None) -> str: settings = get_settings() payload: Dict[str, Any] = {"sub": sub} if extra: payload.update(extra) expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) payload["exp"] = expire return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM) def decode_token(token: str) -> Dict[str, Any]: settings = get_settings() return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) class TokenError(Exception): """JWT 无效或过期。""" def safe_decode_token(token: str) -> Dict[str, Any]: try: return decode_token(token) except JWTError as exc: raise TokenError(str(exc)) from exc