File size: 3,373 Bytes
d767d3e 9513328 d767d3e 9513328 d767d3e | 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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """RMI Backend — Auth middleware, API key verification, and JWT user identity.
This module is the single source of truth for auth. Routes import
`get_current_user` / `get_optional_user` from here (re-exported via
app.api.deps for convenience).
"""
from __future__ import annotations
import os
from typing import Any
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/",
]
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)
# ── JWT user identity (FastAPI dependencies) ─────────────────────────────
# Delegates to the legacy app.auth JWT logic during strangelfig migration.
# Once legacy auth.py is migrated to the new pattern, these become the
# canonical implementation. Until then they reuse the working logic so
# new routes (like app/api/v1/auth/alerts.py) can use modern Depends().
async def get_optional_user(request: Request) -> dict[str, Any] | None:
"""Return the authenticated user dict, or None if not authenticated.
Reads the Authorization: Bearer <jwt> header. Returns the user dict
(id, email, tier, role) on success, None if no/invalid token.
Use this for endpoints that work with OR without auth.
"""
from app.auth import get_current_user as _legacy_get_user # local import to avoid cycles
return await _legacy_get_user(request)
async def get_current_user(request: Request) -> dict[str, Any]:
"""Require an authenticated user. Raises 401 if missing.
Use this for endpoints that REQUIRE auth.
"""
from app.auth import require_auth as _legacy_require # local import to avoid cycles
return await _legacy_require(request)
|