ad111d5e69
Docker Compose with 24+ services: - PostgreSQL (PostGIS), Valkey, MinIO, Qdrant - Vault (PKI/TLS), Nginx (Reverse Proxy) - Backend Core API, Consent Service, Billing Service - RAG Service, Embedding Service - Gitea, Woodpecker CI/CD - Night Scheduler, Health Aggregator - Jitsi (Web/XMPP/JVB/Jicofo), Mailpit Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
"""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 <token>"
|
|
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")
|