62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
"""Password hashing and JWT helpers."""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import jwt
|
|
|
|
from app.core.config import get_settings
|
|
|
|
_PBKDF2_ALGORITHM = "sha256"
|
|
_PBKDF2_ITERATIONS = 310_000
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""Hash a password using salted PBKDF2-HMAC."""
|
|
|
|
salt = secrets.token_bytes(16)
|
|
digest = hashlib.pbkdf2_hmac(_PBKDF2_ALGORITHM, password.encode(), salt, _PBKDF2_ITERATIONS)
|
|
return f"pbkdf2_sha256${_PBKDF2_ITERATIONS}${salt.hex()}${digest.hex()}"
|
|
|
|
|
|
def verify_password(password: str, encoded: str | None) -> bool:
|
|
"""Verify a password hash without revealing whether the account exists."""
|
|
|
|
if not encoded:
|
|
return False
|
|
try:
|
|
scheme, iterations, salt_hex, digest_hex = encoded.split("$", 3)
|
|
if scheme != "pbkdf2_sha256":
|
|
return False
|
|
candidate = hashlib.pbkdf2_hmac(
|
|
_PBKDF2_ALGORITHM,
|
|
password.encode(),
|
|
bytes.fromhex(salt_hex),
|
|
int(iterations),
|
|
)
|
|
return hmac.compare_digest(candidate.hex(), digest_hex)
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
def create_access_token(subject: int, *, role: str) -> str:
|
|
settings = get_settings()
|
|
expires_at = datetime.now(UTC) + timedelta(minutes=settings.jwt_expire_minutes)
|
|
payload: dict[str, Any] = {"sub": str(subject), "role": role, "exp": expires_at}
|
|
return jwt.encode(payload, settings.jwt_secret_key.get_secret_value(), algorithm=settings.jwt_algorithm)
|
|
|
|
|
|
def decode_access_token(token: str) -> dict[str, Any] | None:
|
|
settings = get_settings()
|
|
try:
|
|
return jwt.decode(
|
|
token,
|
|
settings.jwt_secret_key.get_secret_value(),
|
|
algorithms=[settings.jwt_algorithm],
|
|
)
|
|
except jwt.PyJWTError:
|
|
return None
|