"""Optional JWT authentication helper. If JWT_SECRET is configured and an Authorization header is present, the token is verified. If no header is present or JWT_SECRET is empty, the request is allowed through (public access). """ import logging from typing import Optional from fastapi import HTTPException, Request from jose import JWTError, jwt from config import settings logger = logging.getLogger("rag-service.auth") def optional_jwt_auth(request: Request) -> Optional[dict]: """ Validate the JWT from the Authorization header if present. Returns the decoded token payload, or None if no auth was provided. Raises HTTPException 401 if a token IS provided but is invalid. """ auth_header: Optional[str] = request.headers.get("authorization") if not auth_header: return None if not settings.JWT_SECRET: # No secret configured -- skip validation return None # Expect "Bearer " parts = auth_header.split() if len(parts) != 2 or parts[0].lower() != "bearer": raise HTTPException(status_code=401, detail="Invalid Authorization header format") token = parts[1] try: payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"]) return payload except JWTError as exc: logger.warning("JWT verification failed: %s", exc) raise HTTPException(status_code=401, detail="Invalid or expired token")