From 8ea46974ff2e47a7fb2ea03a36bfb206342d9c19 Mon Sep 17 00:00:00 2001 From: timmy Date: Wed, 12 Aug 2026 01:54:51 +0000 Subject: [PATCH] feat: bind operator mode to public origin (Closes #609) --- README.md | 3 +- conftest.py | 3 +- src/dashboard_auth.py | 34 +++++++++++++++- src/main.py | 19 +++++++-- tests/test_dashboard_auth.py | 74 ++++++++++++++++++++++++++++++++++ tests/test_security_headers.py | 12 ++++++ 6 files changed, 138 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 97f7cfe..b5ad7f1 100644 --- a/README.md +++ b/README.md @@ -163,11 +163,12 @@ export GITEA_TOKEN='' export STACKCHAIN_DASHBOARD_AUTH_MODE='operator' export STACKCHAIN_DASHBOARD_ACCESS_TOKEN='' export STACKCHAIN_DASHBOARD_SESSION_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. diff --git a/conftest.py b/conftest.py index 4c13c15..c77eba8 100644 --- a/conftest.py +++ b/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) \ No newline at end of file + monkeypatch.delenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", raising=False) + monkeypatch.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://test") diff --git a/src/dashboard_auth.py b/src/dashboard_auth.py index 562826d..576943e 100644 --- a/src/dashboard_auth.py +++ b/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}" - return bool(origin) and hmac.compare_digest(origin.rstrip("/"), expected) + expected = public_origin() + return bool(origin and expected) and hmac.compare_digest(origin.rstrip("/"), expected) diff --git a/src/main.py b/src/main.py index ba8a15e..60dc9f3 100644 --- a/src/main.py +++ b/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 diff --git a/tests/test_dashboard_auth.py b/tests/test_dashboard_auth.py index f297778..cdd567a 100644 --- a/tests/test_dashboard_auth.py +++ b/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) diff --git a/tests/test_security_headers.py b/tests/test_security_headers.py index ee3ff73..c51de7b 100644 --- a/tests/test_security_headers.py +++ b/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 -- 2.43.0