"""Password hashing and JWT token helpers.""" import hashlib import uuid from datetime import UTC, datetime, timedelta import bcrypt import jwt from app.core.config import settings ACCESS_TOKEN_TYPE = "access" REFRESH_TOKEN_TYPE = "refresh" # --- Passwords --------------------------------------------------------------- def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() def verify_password(password: str, hashed: str) -> bool: try: return bcrypt.checkpw(password.encode(), hashed.encode()) except ValueError: return False # --- JWT --------------------------------------------------------------------- def _create_token( subject: str | int, expires_delta: timedelta, token_type: str, ) -> tuple[str, dict]: now = datetime.now(UTC) payload: dict = { "sub": str(subject), "iat": now, "exp": now + expires_delta, "type": token_type, "jti": uuid.uuid4().hex, } token = jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm) return token, payload def create_access_token( subject: str | int, expires_delta: timedelta | None = None ) -> str: token, _ = _create_token( subject, expires_delta or timedelta(minutes=settings.access_token_expire_minutes), ACCESS_TOKEN_TYPE, ) return token def create_refresh_token(subject: str | int) -> tuple[str, dict]: """Return ``(token, payload)`` so the caller can persist jti/exp for revocation.""" return _create_token( subject, timedelta(days=settings.refresh_token_expire_days), REFRESH_TOKEN_TYPE, ) def decode_token(token: str) -> dict: """Decode and validate a JWT. Raises ``jwt.PyJWTError`` on failure.""" return jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) def hash_token(token: str) -> str: """Stable hash used to store refresh tokens without keeping the raw value.""" return hashlib.sha256(token.encode()).hexdigest()