File size: 1,916 Bytes
6993919 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """RMI Backend — Auth middleware and API key verification."""
import os
from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
RMI_AUTH_TOKEN = os.getenv("RMI_AUTH_TOKEN", "")
PUBLIC_WRITE_PREFIXES = [
"/api/v1/auth/",
"/api/v1/x402/",
"/api/v1/x402-tools/",
"/api/v1/x402-databus/",
"/api/v1/databus/",
"/api/v1/alerts/",
"/api/v1/admin/",
"/api/v1/content/",
"/api/v1/bulletin/",
"/api/v1/rag/permanence/",
"/api/v1/token/",
"/api/v1/ai/",
"/api/v1/premium/",
"/api/v1/rag/",
"/api/v1/protect/",
"/api/v1/wallet-manager/",
]
# Auth bypass paths
PUBLIC_GET_PREFIXES = [
"/api/v1/token/",
"/api/v1/databus/",
"/api/v1/alerts/",
"/api/v1/x402-databus/",
"/api/v1/x402-tools/",
]
def is_public_path(path: str, method: str) -> bool:
"""Check if a path is publicly accessible without auth."""
if path in ("/health", "/ready", "/docs", "/openapi.json", "/redoc", "/", "/favicon.ico"):
return True
if path.startswith("/ws/") or not path.startswith("/api/"):
return True
if method in ("POST", "PUT", "DELETE", "PATCH"):
return any(path.startswith(p) for p in PUBLIC_WRITE_PREFIXES)
return True # GET/HEAD always public
class AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
path = request.url.path
method = request.method
if is_public_path(path, method):
return await call_next(request)
api_key = request.headers.get("X-API-Key", "")
if RMI_AUTH_TOKEN and api_key != RMI_AUTH_TOKEN:
return JSONResponse(
status_code=401,
content={"detail": "Unauthorized - valid X-API-Key header required for write operations"},
)
return await call_next(request)
|