Require admin token for control endpoints

app/config.py

@@ -12,6 +12,7 @@ class Settings(BaseSettings):
     APP_NAME: str = "Sanctum Chronicler"
     APP_ENV: str = "development"
     DEBUG: bool = False
+    ADMIN_API_KEY: Optional[str] = None
 
     @field_validator("DEBUG", mode="before")
     @classmethod

app/main.py

@@ -1,8 +1,9 @@
 """FastAPI main application."""
 
 import asyncio
+import secrets
 from contextlib import suppress
-from fastapi import FastAPI, HTTPException, Form
+from fastapi import Depends, FastAPI, Form, Header, HTTPException
 from datetime import datetime
 import logging
 
@@ -24,6 +25,19 @@ orchestrator: AgentOrchestrator | None = None
 agent_loop_task: asyncio.Task | None = None
 
 
+async def require_admin(
+    admin_token: str | None = Header(default=None, alias="X-Admin-Token"),
+) -> None:
+    """Require the configured admin token for mutable/control endpoints."""
+    if not settings.ADMIN_API_KEY:
+        raise HTTPException(status_code=503, detail="Admin API key is not configured")
+    if not admin_token or not secrets.compare_digest(
+        admin_token,
+        settings.ADMIN_API_KEY,
+    ):
+        raise HTTPException(status_code=401, detail="Invalid admin token")
+
+
 async def agent_loop() -> None:
     """Run periodic time-based agent behavior for active sessions."""
     if not orchestrator:
@@ -80,7 +94,7 @@ async def health_check() -> dict:
     }
 
 
-@app.post("/admin/session/start")
+@app.post("/admin/session/start", dependencies=[Depends(require_admin)])
 async def start_session(channel_name: str = Form(...)) -> dict:
     """Start a new stream session."""
     if not orchestrator:
@@ -95,7 +109,7 @@ async def start_session(channel_name: str = Form(...)) -> dict:
     }
 
 
-@app.post("/admin/session/end")
+@app.post("/admin/session/end", dependencies=[Depends(require_admin)])
 async def end_session(session_id: str = Form(...)) -> dict:
     """End the current stream session."""
     if not orchestrator:
@@ -109,7 +123,7 @@ async def end_session(session_id: str = Form(...)) -> dict:
     }
 
 
-@app.post("/admin/test-message")
+@app.post("/admin/test-message", dependencies=[Depends(require_admin)])
 async def test_message(session_id: str = Form(...), message: str = Form(...), username: str = Form("test_user")) -> dict:
     """Send a test message to the orchestrator."""
     if not orchestrator:
@@ -128,7 +142,7 @@ async def test_message(session_id: str = Form(...), message: str = Form(...), us
     }
 
 
-@app.post("/admin/test-agent-response")
+@app.post("/admin/test-agent-response", dependencies=[Depends(require_admin)])
 async def test_agent_response(
     session_id: str = Form(...),
     message: str = Form(...),
@@ -153,7 +167,7 @@ async def test_agent_response(
     }
 
 
-@app.post("/admin/test-loop-inactivity")
+@app.post("/admin/test-loop-inactivity", dependencies=[Depends(require_admin)])
 async def test_loop_inactivity(
     session_id: str = Form(...),
     inactive_minutes: int = Form(16),
@@ -176,7 +190,7 @@ async def test_loop_inactivity(
     }
 
 
-@app.get("/admin/ledger")
+@app.get("/admin/ledger", dependencies=[Depends(require_admin)])
 async def get_ledger(session_id: str) -> dict:
     """Get the markdown ledger for a session."""
     if not orchestrator:
@@ -192,7 +206,7 @@ async def get_ledger(session_id: str) -> dict:
     }
 
 
-@app.get("/admin/session/status")
+@app.get("/admin/session/status", dependencies=[Depends(require_admin)])
 async def get_session_status(session_id: str) -> dict:
     """Get status for an active stream session."""
     if not orchestrator:
@@ -208,7 +222,7 @@ async def get_session_status(session_id: str) -> dict:
     }
 
 
-@app.get("/admin/loop/status")
+@app.get("/admin/loop/status", dependencies=[Depends(require_admin)])
 async def get_loop_status() -> dict:
     """Get the background agent loop runtime configuration."""
     if not orchestrator:
@@ -221,7 +235,7 @@ async def get_loop_status() -> dict:
     }
 
 
-@app.post("/admin/loop/frequency")
+@app.post("/admin/loop/frequency", dependencies=[Depends(require_admin)])
 async def set_loop_frequency(interval_seconds: float = Form(...)) -> dict:
     """Set how frequently the background agent loop runs."""
     if not orchestrator:

docker-compose.yml

@@ -32,6 +32,7 @@ services:
       APP_NAME: "Sanctum Chronicler"
       APP_ENV: ${APP_ENV:-development}
       DEBUG: ${DEBUG:-false}
+      ADMIN_API_KEY: ${ADMIN_API_KEY:-}
       DATABASE_URL: postgresql+asyncpg://sanctum:${DB_PASSWORD:-password}@sanctum-db:5432/sanctum
       TWITCH_CLIENT_ID: ${TWITCH_CLIENT_ID:-}
       TWITCH_CLIENT_SECRET: ${TWITCH_CLIENT_SECRET:-}