night-shift: gateway token auth + Origin rejection + tests

src/personal_llm/config.py

@@ -23,6 +23,7 @@ class Settings(BaseSettings):
     personal_llm_chroma_dir: str = "./data/chroma"
     personal_llm_workspace_dir: str = "./data/workspace"
     personal_llm_voice_dir: str = "./data/voice"
+    personal_llm_gateway_token_path: str = "./data/gateway_token"
 
     retrieval_top_k: int = 8
     retrieval_min_similarity: float = 0.25

src/personal_llm/interfaces/api.py

@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+import secrets
 import tempfile
 from pathlib import Path
 
-from fastapi import FastAPI, File, HTTPException, UploadFile
+from fastapi import FastAPI, File, HTTPException, Request, UploadFile
+from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
 from personal_llm import __version__
@@ -24,6 +26,46 @@
 
 app = FastAPI(title="Personal LLM", version=__version__)
 
+GATEWAY_TOKEN_HEADER = "x-dreamos-token"
+_gateway_token: str | None = None
+
+
+def _load_or_create_gateway_token() -> str:
+    """Shared secret DreamOS (or any other local caller) must send back on every request.
+
+    Generated once and persisted to disk so it survives a gateway restart without the
+    caller needing to re-pair.
+    """
+    global _gateway_token
+    if _gateway_token is not None:
+        return _gateway_token
+    path = Path(get_settings().personal_llm_gateway_token_path)
+    existing = path.read_text().strip() if path.exists() else ""
+    if existing:
+        _gateway_token = existing
+    else:
+        _gateway_token = secrets.token_hex(32)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(_gateway_token)
+    return _gateway_token
+
+
+@app.middleware("http")
+async def gateway_auth(request: Request, call_next):
+    """CSRF hardening (MASTER-FIX-PLAN.md S3 / Phase 3 item 12).
+
+    Multipart and form-encoded POSTs are reachable cross-origin without a CORS
+    preflight, so a browser page could otherwise submit them straight to this
+    gateway. Any request carrying an Origin header - which every real browser
+    request does, and no local non-browser caller does - is rejected outright,
+    and every request must also present the shared token.
+    """
+    if "origin" in request.headers:
+        return JSONResponse(status_code=403, content={"detail": "Cross-origin requests are not allowed."})
+    if request.headers.get(GATEWAY_TOKEN_HEADER) != _load_or_create_gateway_token():
+        return JSONResponse(status_code=401, content={"detail": "Missing or invalid X-DreamOS-Token."})
+    return await call_next(request)
+
 
 async def _save_upload_to_temp(upload: UploadFile) -> str:
     suffix = Path(upload.filename or "upload").suffix

tests/test_api.py

@@ -0,0 +1,104 @@
+"""Gateway auth: shared token + Origin rejection (MASTER-FIX-PLAN.md Phase 3 item 12).
+
+FakeRouter/store/vectors fixtures (conftest.py) keep this offline and keyless -
+no network call, no API key, ever.
+"""
+
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+from personal_llm.engine import Engine
+from personal_llm.interfaces import api
+from personal_llm.voice import SpeechToText, TextToSpeech
+
+TOKEN = "test-token"
+
+
+def _fake_engine(store, vectors, router) -> Engine:
+    return Engine(store=store, vectors=vectors, router=router, stt=SpeechToText(), tts=TextToSpeech())
+
+
+def _client(monkeypatch, store, vectors, router):
+    monkeypatch.setattr(api, "build_engine", lambda: _fake_engine(store, vectors, router))
+    monkeypatch.setattr(api, "_load_or_create_gateway_token", lambda: TOKEN)
+    return TestClient(api.app)
+
+
+def test_tokenless_request_is_rejected(monkeypatch, store, vectors, router):
+    client = _client(monkeypatch, store, vectors, router)
+
+    resp = client.get("/stats")
+
+    assert resp.status_code == 401
+
+
+def test_wrong_token_is_rejected(monkeypatch, store, vectors, router):
+    client = _client(monkeypatch, store, vectors, router)
+
+    resp = client.get("/stats", headers={api.GATEWAY_TOKEN_HEADER: "not-the-token"})
+
+    assert resp.status_code == 401
+
+
+def test_correct_token_allows_stats(monkeypatch, store, vectors, router):
+    client = _client(monkeypatch, store, vectors, router)
+
+    resp = client.get("/stats", headers={api.GATEWAY_TOKEN_HEADER: TOKEN})
+
+    assert resp.status_code == 200
+    assert resp.json() == store.stats()
+
+
+def test_origin_header_is_rejected_even_with_a_valid_token(monkeypatch, store, vectors, router):
+    client = _client(monkeypatch, store, vectors, router)
+
+    resp = client.get(
+        "/stats",
+        headers={api.GATEWAY_TOKEN_HEADER: TOKEN, "Origin": "https://evil.example"},
+    )
+
+    assert resp.status_code == 403
+
+
+def test_ask_endpoint_requires_token(monkeypatch, store, vectors, router):
+    client = _client(monkeypatch, store, vectors, router)
+
+    resp = client.post("/ask", json={"question": "anything"})
+
+    assert resp.status_code == 401
+
+
+def test_ask_endpoint_round_trips_with_a_valid_token(monkeypatch, store, vectors, router):
+    client = _client(monkeypatch, store, vectors, router)
+
+    resp = client.post(
+        "/ask",
+        json={"question": "anything"},
+        headers={api.GATEWAY_TOKEN_HEADER: TOKEN},
+    )
+
+    assert resp.status_code == 200
+    assert "text" in resp.json()
+
+
+def test_voice_ask_endpoint_requires_token(monkeypatch, store, vectors, router):
+    client = _client(monkeypatch, store, vectors, router)
+
+    resp = client.post("/voice/ask", files={"file": ("clip.wav", b"not real audio", "audio/wav")})
+
+    assert resp.status_code == 401
+
+
+def test_load_or_create_gateway_token_persists_across_calls(tmp_path, monkeypatch):
+    token_path = tmp_path / "gateway_token"
+    monkeypatch.setattr(api, "_gateway_token", None)
+    monkeypatch.setattr(api, "get_settings", lambda: type("S", (), {"personal_llm_gateway_token_path": str(token_path)})())
+
+    first = api._load_or_create_gateway_token()
+
+    monkeypatch.setattr(api, "_gateway_token", None)
+    second = api._load_or_create_gateway_token()
+
+    assert first == second
+    assert token_path.read_text().strip() == first