feat: bind operator mode to public origin (Closes #609)

README.md

@@ -163,11 +163,12 @@ export GITEA_TOKEN='<read-notification-and-issue-write-token>'
 export STACKCHAIN_DASHBOARD_AUTH_MODE='operator'
 export STACKCHAIN_DASHBOARD_ACCESS_TOKEN='<operator-sign-in-secret>'
 export STACKCHAIN_DASHBOARD_SESSION_SECRET='<independent-cookie-signing-secret>'
+export STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN='https://forge.example.com'
 # Optional; defaults to STACKCHAIN_STATE_DIR/sessions.sqlite3.
 export STACKCHAIN_SESSION_DB='/var/lib/stackchain-dashboard/sessions.sqlite3'
 # Optional; defaults to STACKCHAIN_STATE_DIR/security-events.sqlite3.
 export STACKCHAIN_SECURITY_EVENT_DB='/var/lib/stackchain-dashboard/security-events.sqlite3'
-# Recommended behind a proxy; WebAuthn assertions must match these public values.
+# Optional compatibility overrides; defaults derive from the required public origin.
 export STACKCHAIN_PASSKEY_RP_ID='forge.example.com'
 export STACKCHAIN_PASSKEY_ORIGIN='https://forge.example.com'
 # Optional; defaults to eight hours.

conftest.py

@@ -6,4 +6,5 @@ def explicit_local_dashboard_auth_mode(monkeypatch):
     """Keep tests intentional now that deployed authentication fails closed."""
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "insecure-local")
     monkeypatch.delenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", raising=False)
-    monkeypatch.delenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", raising=False)
+    monkeypatch.delenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", raising=False)
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://test")

src/dashboard_auth.py

@@ -10,6 +10,7 @@ import os
 import secrets
 import time
 from dataclasses import dataclass
+from urllib.parse import urlsplit
 
 from fastapi import Request
 
@@ -47,6 +48,33 @@ def mode() -> str:
     return os.getenv("STACKCHAIN_DASHBOARD_AUTH_MODE", OPERATOR_MODE).strip().lower()
 
 
+def public_origin() -> str | None:
+    """Return the canonical HTTPS browser origin, or None when it is invalid."""
+    configured = os.getenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "").strip()
+    if not configured:
+        return None
+    try:
+        parsed = urlsplit(configured)
+        port = parsed.port
+    except ValueError:
+        return None
+    if (
+        parsed.scheme != "https"
+        or not parsed.hostname
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.path not in {"", "/"}
+        or parsed.query
+        or parsed.fragment
+    ):
+        return None
+    hostname = parsed.hostname.lower()
+    authority = f"[{hostname}]" if ":" in hostname else hostname
+    if port is not None and port != 443:
+        authority += f":{port}"
+    return f"https://{authority}"
+
+
 def configuration_error() -> str | None:
     configured_mode = mode()
     if configured_mode == INSECURE_LOCAL_MODE:
@@ -59,6 +87,8 @@ def configuration_error() -> str | None:
         return "operator mode requires two sufficiently long secrets"
     if hmac.compare_digest(operator_secret, signing_secret):
         return "operator and session secrets must be independent"
+    if public_origin() is None:
+        return "operator mode requires a canonical HTTPS public origin"
     return None
 
 
@@ -274,5 +304,5 @@ def application_path(request: Request) -> str:
 
 def same_origin(request: Request) -> bool:
     origin = request.headers.get("origin", "")
-    expected = f"{request.url.scheme}://{request.url.netloc}"
+    expected = public_origin()
-    return bool(origin) and hmac.compare_digest(origin.rstrip("/"), expected)
+    return bool(origin and expected) and hmac.compare_digest(origin.rstrip("/"), expected)

src/main.py

@@ -13,7 +13,7 @@ from contextlib import asynccontextmanager
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Literal
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlsplit
 
 from fastapi import FastAPI, Header, HTTPException, Path as PathParam, Query, Request, Response
 from fastapi.exceptions import RequestValidationError
@@ -340,10 +340,12 @@ def _security_event_store() -> SecurityEventStore:
 
 
 def _passkey_relying_party(request: Request) -> tuple[str, str]:
-    rp_id = os.getenv("STACKCHAIN_PASSKEY_RP_ID", request.url.hostname or "")
+    canonical_origin = dashboard_auth.public_origin()
+    canonical_host = urlsplit(canonical_origin).hostname if canonical_origin else None
+    rp_id = os.getenv("STACKCHAIN_PASSKEY_RP_ID", canonical_host or request.url.hostname or "")
     origin = os.getenv(
         "STACKCHAIN_PASSKEY_ORIGIN",
-        f"{request.url.scheme}://{request.url.netloc}",
+        canonical_origin or f"{request.url.scheme}://{request.url.netloc}",
     )
     return rp_id, origin
 
@@ -886,6 +888,15 @@ def _share_target_login_redirect(request: Request) -> str:
 @app.middleware("http")
 async def require_operator_session(request: Request, call_next):
     path = dashboard_auth.application_path(request)
+    canonical_origin = dashboard_auth.public_origin()
+    if dashboard_auth.mode() == dashboard_auth.OPERATOR_MODE and canonical_origin:
+        expected_authority = urlsplit(canonical_origin).netloc
+        if request.url.netloc.lower() != expected_authority:
+            return JSONResponse(
+                {"detail": "Request host does not match the configured public origin"},
+                status_code=421,
+                headers={"Cache-Control": "no-store"},
+            )
     if path == "/healthz":
         return await call_next(request)
 
@@ -1043,6 +1054,8 @@ async def enforce_browser_security_boundary(request: Request, call_next):
         "camera=(), microphone=(), geolocation=(), payment=(), usb=()"
     )
     response.headers["X-Frame-Options"] = "DENY"
+    if dashboard_auth.mode() == dashboard_auth.OPERATOR_MODE:
+        response.headers["Strict-Transport-Security"] = "max-age=31536000"
     return response
 
 

tests/test_dashboard_auth.py

@@ -6,6 +6,7 @@ from urllib.parse import parse_qs, urlsplit
 
 import httpx
 import pytest
+from fastapi import Request
 
 from src import main
 from src.session_store import SessionStoreError
@@ -15,6 +16,7 @@ from src.views import FRONTEND_BUILD
 @pytest.fixture
 def access_control(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://test")
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
     monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
@@ -1364,6 +1366,78 @@ async def test_mutation_requires_same_origin_and_session_csrf(access_control, mo
     assert calls == 1
 
 
+@pytest.mark.anyio
+async def test_operator_mode_rejects_unconfigured_or_foreign_public_origin_before_auth(
+    monkeypatch,
+):
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
+    monkeypatch.setenv(
+        "STACKCHAIN_DASHBOARD_SESSION_SECRET",
+        "a-separate-session-signing-secret-with-enough-entropy",
+    )
+    monkeypatch.delenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", raising=False)
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        missing = await client.get("/login")
+        monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "http://test")
+        insecure = await client.get("/login")
+        monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://dashboard.example")
+        foreign_host = await client.get("/login")
+
+    assert missing.status_code == 503
+    assert insecure.status_code == 503
+    assert foreign_host.status_code == 421
+
+
+@pytest.mark.anyio
+async def test_canonical_origin_drives_csrf_and_passkey_defaults(access_control, monkeypatch):
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://dashboard.example")
+    monkeypatch.delenv("STACKCHAIN_PASSKEY_RP_ID", raising=False)
+    monkeypatch.delenv("STACKCHAIN_PASSKEY_ORIGIN", raising=False)
+    calls = 0
+
+    async def mark_read(notification_id):
+        nonlocal calls
+        calls += 1
+        return {"id": notification_id, "read": True}
+
+    monkeypatch.setattr(main, "mark_notification_read", mark_read)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://dashboard.example"
+    ) as client:
+        await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+        csrf = client.cookies["stackchain_csrf"]
+        valid = await client.patch(
+            "/api/v1/notifications/7/read",
+            headers={
+                "Origin": "https://dashboard.example",
+                "X-CSRF-Token": csrf,
+            },
+        )
+
+    request = Request({
+        "type": "http",
+        "method": "GET",
+        "scheme": "https",
+        "path": "/",
+        "query_string": b"",
+        "headers": [(b"host", b"attacker.invalid")],
+        "server": ("attacker.invalid", 443),
+        "client": ("127.0.0.1", 1),
+    })
+    assert main._passkey_relying_party(request) == (
+        "dashboard.example",
+        "https://dashboard.example",
+    )
+    assert valid.status_code == 200
+    assert calls == 1
+
+
 @pytest.mark.anyio
 async def test_logout_clears_session_and_blocks_private_routes(access_control):
     transport = httpx.ASGITransport(app=main.app)

tests/test_security_headers.py

@@ -30,6 +30,7 @@ async def test_security_boundary_covers_pages_health_static_and_api_errors(monke
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "access-token-with-at-least-thirty-two-characters")
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "session-secret-with-at-least-thirty-two-characters")
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://test")
     transport = httpx.ASGITransport(app=main.app)
 
     async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
@@ -43,6 +44,7 @@ async def test_security_boundary_covers_pages_health_static_and_api_errors(monke
     assert [response.status_code for response in responses] == [200, 200, 200, 401]
     for response in responses:
         assert_browser_security_boundary(response)
+        assert response.headers["strict-transport-security"] == "max-age=31536000"
 
 
 @pytest.mark.anyio
@@ -57,3 +59,13 @@ async def test_security_boundary_covers_fail_closed_authentication_response(monk
 
     assert response.status_code == 503
     assert_browser_security_boundary(response)
+
+
+@pytest.mark.anyio
+async def test_insecure_local_mode_does_not_emit_hsts():
+    transport = httpx.ASGITransport(app=main.app, client=("127.0.0.1", 123))
+    async with httpx.AsyncClient(transport=transport, base_url="http://localhost") as client:
+        response = await client.get("/healthz")
+
+    assert response.status_code == 200
+    assert "strict-transport-security" not in response.headers