48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
"""密码哈希与 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
|