pyr0ball
321abe0646
app/config.py: centralized Settings (DEMO_MODE, CLOUD_MODE, ports, etc.) app/middleware/demo.py: DemoModeMiddleware — caps sessions (429), blocks export (403), adds X-Linnet-Mode header app/middleware/cloud.py: CloudAuthMiddleware — requires X-CF-Session on /session/* routes, 401 without it app/services/session_store.py: active_session_count() for demo cap app/main.py: wires middleware conditionally, extends CORS for cloud origins compose.test.yml: hermetic pytest runner in Docker (CF_VOICE_MOCK=1) compose.demo.yml: DEMO_MODE=true, ports 8523/8524, demo.circuitforge.tech/linnet compose.cloud.yml: CLOUD_MODE=true, ports 8522/8527, menagerie.circuitforge.tech/linnet docker/web/Dockerfile: two-stage build (node:20 → nginx:alpine), VITE_BASE_URL/VITE_API_BASE ARGs docker/web/nginx.conf: SSE + WS proxy, SPA routing (dev/demo) docker/web/nginx.cloud.conf: adds X-CF-Session forwarding, /linnet/ alias for path-strip Caddy routing manage.sh: profile arg (dev|demo|cloud|test), start/stop/restart/status/test/logs/build/open per profile tests/test_profiles.py: 8 tests — demo export block, session cap, cloud auth gate, mode headers
41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
# app/middleware/cloud.py — CLOUD_MODE auth
|
|
#
|
|
# When CLOUD_MODE=true, all /session/* routes require the X-CF-Session header
|
|
# (injected by Caddy from the cf_session cookie set by the website auth flow).
|
|
# The header value is forwarded opaquely — Linnet trusts it as an opaque user ID.
|
|
# Full Heimdall JWT validation is a v1.0 addition (tracked in linnet#16).
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from fastapi import Request, Response
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.responses import JSONResponse
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Paths that don't require auth even in cloud mode
|
|
_PUBLIC_PATHS = {"/health", "/docs", "/openapi.json", "/redoc"}
|
|
|
|
|
|
class CloudAuthMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request: Request, call_next) -> Response:
|
|
path = request.url.path
|
|
|
|
if path in _PUBLIC_PATHS or not path.startswith("/session"):
|
|
return await call_next(request)
|
|
|
|
session_token = request.headers.get(settings.cloud_session_header, "").strip()
|
|
if not session_token:
|
|
return JSONResponse(
|
|
status_code=401,
|
|
content={"detail": "Authentication required. Sign in at circuitforge.tech."},
|
|
)
|
|
|
|
# Attach the user identity to request state so endpoints can use it
|
|
request.state.cf_user = session_token
|
|
response = await call_next(request)
|
|
response.headers["X-Linnet-Mode"] = "cloud"
|
|
return response
|