chengkai3
90 lines
2.4 KiB
Python
90 lines
2.4 KiB
Python
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import bcrypt
|
|
import jwt
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import InvalidHash, VerifyMismatchError
|
|
from fastapi import HTTPException, status
|
|
|
|
from .config import get_settings
|
|
|
|
password_hasher = PasswordHasher()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return password_hasher.hash(password)
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
# Backward compatibility: legacy quiz users use BCrypt hashes.
|
|
if password_hash.startswith(("$2a$", "$2b$", "$2y$")):
|
|
try:
|
|
return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
|
|
except ValueError:
|
|
return False
|
|
|
|
try:
|
|
return password_hasher.verify(password_hash, password)
|
|
except (VerifyMismatchError, InvalidHash):
|
|
return False
|
|
|
|
|
|
def create_access_token(
|
|
*,
|
|
user_id: str,
|
|
role_codes: list[str],
|
|
permission_codes: list[str],
|
|
expires_minutes: int | None = None,
|
|
) -> tuple[str, int]:
|
|
settings = get_settings()
|
|
now = datetime.now(timezone.utc)
|
|
minutes = expires_minutes or settings.access_token_expire_minutes
|
|
expires_at = now + timedelta(minutes=minutes)
|
|
|
|
payload = {
|
|
"sub": user_id,
|
|
"roles": role_codes,
|
|
"permissions": permission_codes,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expires_at.timestamp()),
|
|
}
|
|
token = jwt.encode(
|
|
payload,
|
|
settings.jwt_secret_key,
|
|
algorithm=settings.jwt_algorithm,
|
|
)
|
|
return token, int((expires_at - now).total_seconds())
|
|
|
|
|
|
def decode_access_token(token: str) -> dict[str, Any]:
|
|
settings = get_settings()
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
settings.jwt_secret_key,
|
|
algorithms=[settings.jwt_algorithm],
|
|
)
|
|
except jwt.PyJWTError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired access token",
|
|
) from exc
|
|
|
|
if not payload.get("sub"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid access token payload",
|
|
)
|
|
return payload
|
|
|
|
|
|
def create_refresh_token() -> str:
|
|
return secrets.token_urlsafe(48)
|
|
|
|
|
|
def hash_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|