""" Security utilities: password hashing and JWT token management. This module provides pure functions for: - Hashing and verifying passwords using bcrypt via passlib. - Creating JWT access tokens using python-jose. - Managing a Redis-backed token blacklist for revocation. No endpoints are defined here. """ import logging import uuid as _uuid from datetime import datetime, timedelta, timezone from jose import jwt from passlib.context import CryptContext from app.config import settings logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Password hashing # --------------------------------------------------------------------------- pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(password: str) -> str: """Return a bcrypt hash of *password*.""" return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: """Return ``True`` if *plain* matches the bcrypt *hashed* value.""" return pwd_context.verify(plain, hashed) # --------------------------------------------------------------------------- # JWT tokens # --------------------------------------------------------------------------- def create_access_token(data: dict) -> str: """Create a signed JWT containing *data* plus ``exp`` and ``jti`` claims. - ``jti`` (JWT ID): unique identifier that enables token revocation. - ``exp``: expiration timestamp based on ``ACCESS_TOKEN_EXPIRE_MINUTES``. """ to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta( minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES, ) to_encode.update({ "exp": expire, "jti": str(_uuid.uuid4()), }) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) # --------------------------------------------------------------------------- # Token blacklist (Redis-backed) # --------------------------------------------------------------------------- # Each revoked token's ``jti`` is stored in Redis with a TTL equal to the # token's remaining lifetime. This means entries auto-expire exactly when # the token would have become invalid anyway — no manual cleanup needed. # # Redis survives backend restarts, so blacklisted tokens stay revoked # across deploys and multi-worker setups. # --------------------------------------------------------------------------- _BLACKLIST_PREFIX = "blacklist:" def blacklist_token(jti: str, exp: float) -> None: """Add *jti* to the Redis blacklist with a TTL derived from *exp*. *exp* is the token's ``exp`` claim (epoch timestamp). The TTL is set to ``exp - now`` so the key vanishes when the token would have expired naturally. """ from app.infrastructure.redis_client import get_redis_blacklist ttl = max(int(exp - datetime.now(timezone.utc).timestamp()), 1) try: r = get_redis_blacklist() r.setex(f"{_BLACKLIST_PREFIX}{jti}", ttl, "1") except Exception: logger.warning("Failed to blacklist token %s in Redis", jti, exc_info=True) def is_token_blacklisted(jti: str) -> bool: """Return ``True`` if *jti* has been revoked (exists in Redis).""" from app.infrastructure.redis_client import get_redis_blacklist try: r = get_redis_blacklist() return r.exists(f"{_BLACKLIST_PREFIX}{jti}") > 0 except Exception: logger.warning("Failed to check blacklist for %s in Redis", jti, exc_info=True) return False