"""密码哈希与 JWT 签发/校验。""" from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Any import bcrypt import jwt from app.core.config import settings from app.core.exceptions import AuthError # ---------- 密码 ---------- def hash_password(raw: str) -> str: return bcrypt.hashpw(raw.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(raw: str, hashed: str) -> bool: if not raw or not hashed: return False try: return bcrypt.checkpw(raw.encode("utf-8"), hashed.encode("utf-8")) except ValueError: return False # ---------- JWT ---------- def create_access_token(payload: dict[str, Any], expires_minutes: int | None = None) -> str: minutes = expires_minutes or settings.JWT_EXPIRE_MINUTES body = dict(payload) body["exp"] = datetime.now(timezone.utc) + timedelta(minutes=minutes) body["iat"] = datetime.now(timezone.utc) return jwt.encode(body, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM) def decode_access_token(token: str) -> dict[str, Any]: try: return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM]) except jwt.ExpiredSignatureError as exc: raise AuthError("登录已过期,请重新登录") from exc except jwt.PyJWTError as exc: raise AuthError("令牌无效") from exc