from datetime import datetime, timedelta, timezone import os import bcrypt from jose import JWTError, jwt # ── Config (override via environment variables) ─────────────────────────────── _raw_secret = os.getenv("JWT_SECRET_KEY", "") if not _raw_secret: import warnings _raw_secret = "lms_dev_jwt_secret_change_in_production" warnings.warn( "JWT_SECRET_KEY is not set — using insecure default. " "Set it in .env before going to production.", RuntimeWarning, stacklevel=1, ) SECRET_KEY: str = _raw_secret ALGORITHM: str = os.getenv("JWT_ALGORITHM", "HS256") ACCESS_TOKEN_EXPIRE_MINUTES: int = int( os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60") ) # ── Password hashing (use bcrypt directly — passlib has compat issues) ──────── def hash_password(plain: str) -> str: return bcrypt.hashpw(plain.encode(), bcrypt.gensalt()).decode() def verify_password(plain: str, hashed: str) -> bool: return bcrypt.checkpw(plain.encode(), hashed.encode()) # ── JWT ─────────────────────────────────────────────────────────────────────── def create_access_token(user_id: int) -> str: expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) payload = {"sub": str(user_id), "exp": expire} return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM) def decode_access_token(token: str) -> int: """ Decode the JWT and return the user_id (int). Raises JWTError on any validation failure. """ payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) sub = payload.get("sub") if sub is None: raise JWTError("Missing subject claim.") return int(sub)