Bind operator mode to a canonical public origin #610
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -7,3 +7,4 @@ def explicit_local_dashboard_auth_mode(monkeypatch):
|
|||
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.setenv("STACKCHAIN_DASHBOARD_PUBLIC_ORIGIN", "https://test")
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
19
src/main.py
19
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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user