"""Password hashing and JWT access tokens for registered users. Separate from app/core/security.py, which handles admin-API-key auth and rate limiting -- unrelated concerns (a machine-to-machine secret vs. a human user's own credentials). """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta import jwt from argon2 import PasswordHasher from argon2.exceptions import VerifyMismatchError from app.core.config import get_settings _hasher = PasswordHasher() _JWT_ALGORITHM = "HS256" ACCESS_TOKEN_TTL = timedelta(hours=24) class InvalidTokenError(Exception): """Raised for any access token that's missing, malformed, expired, or signed with a different key -- callers treat all of these identically (401), so this collapses PyJWT's several exception types into one.""" def normalize_email(email: str) -> str: """Emails are matched and stored case-insensitively (RFC 5321 technically allows a case-sensitive local part, but no mainstream provider honors that, and forcing lowercase avoids needing a functional unique index).""" return email.strip().lower() def hash_password(password: str) -> str: return _hasher.hash(password) def verify_password(password: str, hashed_password: str) -> bool: try: return _hasher.verify(hashed_password, password) except VerifyMismatchError: return False # Computed once at import time, verified against on a login attempt whose # email doesn't exist at all -- so rejecting a nonexistent account takes # roughly the same time as rejecting a wrong password on a real one. # Without this, a login endpoint's response time itself reveals which # emails are registered, letting an attacker enumerate accounts without # ever guessing a password. _DUMMY_HASH = _hasher.hash("not-a-real-password-used-only-for-timing") def burn_time_like_a_password_check() -> None: verify_password("irrelevant", _DUMMY_HASH) def create_access_token(user_id: uuid.UUID) -> str: settings = get_settings() now = datetime.now(UTC) payload = {"sub": str(user_id), "iat": now, "exp": now + ACCESS_TOKEN_TTL} return jwt.encode(payload, settings.AURORA_SECRET_KEY, algorithm=_JWT_ALGORITHM) def decode_access_token(token: str) -> uuid.UUID: settings = get_settings() try: payload = jwt.decode(token, settings.AURORA_SECRET_KEY, algorithms=[_JWT_ALGORITHM]) return uuid.UUID(payload["sub"]) except (jwt.InvalidTokenError, KeyError, ValueError) as exc: raise InvalidTokenError(str(exc)) from exc