import logging from datetime import datetime, timedelta, timezone from jose import JWTError, jwt from passlib.context import CryptContext from core.config import JWT_SECRET_KEY logger = logging.getLogger("app") SECRET_KEY = JWT_SECRET_KEY ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 REFRESH_TOKEN_EXPIRE_DAYS = 30 pwd_context = CryptContext(schemes=["bcrypt"]) # Verified against when a login names a user that doesn't exist, so the response time # is the same either way. Without it, "no such user" returns before bcrypt runs and # the timing difference tells an attacker which usernames are real. DUMMY_PASSWORD_HASH = pwd_context.hash("not-a-real-password") def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) # All token timestamps are timezone-aware UTC. They used to be naive datetime.now(), # which jose serialises as if it were already UTC — with TZ=Europe/Paris that handed # out tokens living 1-2 hours longer than intended, drifting again at each DST change. # # 'iat' is carried so a password reset can invalidate everything issued before it; # see token_valid_after on the User model. It is written as a float rather than a # datetime: jose truncates datetimes to whole seconds, and at that resolution a token # minted just *after* a reset is indistinguishable from one minted just before, so # logging in immediately after a password change would be rejected. RFC 7519 permits # a non-integer NumericDate. def create_access_token(user_id: str) -> str: now = datetime.now(timezone.utc) return jwt.encode( { "sub": user_id, "iat": now.timestamp(), "exp": now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES), "type": "access", }, SECRET_KEY, algorithm=ALGORITHM, ) def create_refresh_token(user_id: str) -> str: now = datetime.now(timezone.utc) return jwt.encode( { "sub": user_id, "iat": now.timestamp(), "exp": now + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS), "type": "refresh", }, SECRET_KEY, algorithm=ALGORITHM, ) def _decode(token: str, expected_type: str) -> dict | None: """Verified payload, or None if the signature, expiry or type is wrong.""" try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) except JWTError: return None if payload.get("type") != expected_type: return None return payload def decode_refresh_token_payload(token: str) -> dict | None: return _decode(token, "refresh") def decode_access_token_payload(token: str) -> dict | None: return _decode(token, "access") def decode_refresh_token(token: str) -> str | None: payload = _decode(token, "refresh") return payload.get("sub") if payload else None def decode_access_token(token: str) -> str | None: payload = _decode(token, "access") return payload.get("sub") if payload else None def token_issued_before(payload: dict, cutoff: datetime | None) -> bool: """True if this token predates a revocation cutoff (i.e. should be rejected).""" if cutoff is None: return False iat = payload.get("iat") # Tokens minted before 'iat' was introduced have no issue time, so they can't be # proven to postdate the cutoff — treat them as revoked rather than trusted. if iat is None: return True if cutoff.tzinfo is None: cutoff = cutoff.replace(tzinfo=timezone.utc) return datetime.fromtimestamp(iat, timezone.utc) < cutoff