feat(middleware): mode-aware security headers (#437) [stacked on #474]

src/pinky_daemon/api.py

@@ -687,21 +687,15 @@ def create_api(
         )
 
     # ── Security Headers ──────────────────────────────────
-    @app.middleware("http")
-    async def security_headers_middleware(request: Request, call_next):
-        if request.headers.get("upgrade", "").lower() == "websocket":
-            return await call_next(request)
-        response = await call_next(request)
-        response.headers["X-Content-Type-Options"] = "nosniff"
-        response.headers["X-Frame-Options"] = "DENY"
-        response.headers["X-XSS-Protection"] = "1; mode=block"
-        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
-        response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
-        # HSTS only when behind TLS (reverse proxy sets X-Forwarded-Proto)
-        proto = request.headers.get("x-forwarded-proto", "")
-        if proto == "https" or request.url.scheme == "https":
-            response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
-        return response
+    # Mode-aware helmet-style headers (#437). Trusted = minimal back-compat
+    # (nosniff + X-XSS-Protection disable). LAN adds clickjacking +
+    # referrer + permissions-policy + report-only CSP. Web adds strict CSP
+    # + HSTS-when-verified-HTTPS. See pinky_daemon.middleware.security_headers
+    # for the full matrix and the HSTS verification rules.
+    from pinky_daemon.middleware.security_headers import (
+        build_security_headers_middleware,
+    )
+    app.middleware("http")(build_security_headers_middleware())
 
     session_store = SessionStore(db_path=db_path.replace(".db", "_sessions.db"))
     session_event_store = SessionEventStore(db_path=db_path.replace(".db", "_sessions.db"))

src/pinky_daemon/middleware/__init__.py

@@ -0,0 +1,7 @@
+"""HTTP middleware modules for the pinky daemon.
+
+Currently houses :mod:`pinky_daemon.middleware.security_headers` (#437).
+Future cross-cutting middleware introduced by the #433 hardening track
+(rate limiting #438, CSRF #443, elevated-session enforcement #439) should
+live here too.
+"""

src/pinky_daemon/middleware/security_headers.py

@@ -0,0 +1,238 @@
+"""Security headers middleware (#437).
+
+Helmet-style HTTP response headers, gated by deployment mode. The ``trusted``
+posture stays minimal (we don't want to break the existing Mac Mini SPA in
+ways the operator wouldn't notice); ``lan`` adds clickjacking + referrer
+hardening; ``web`` adds HSTS and a Content Security Policy.
+
+Header matrix
+-------------
+============================  ========  =====  =====
+Header                        trusted   lan    web
+============================  ========  =====  =====
+X-Content-Type-Options        ✓         ✓      ✓
+X-Frame-Options               ✗         ✓      ✓
+Referrer-Policy               ✗         ✓      ✓
+Permissions-Policy            ✗         ✓      ✓
+Content-Security-Policy       ✗         basic  strict
+Strict-Transport-Security     ✗         ✗      ✓ (verified-HTTPS only)
+X-XSS-Protection              ``0``     ``0``  ``0``
+============================  ========  =====  =====
+
+``X-XSS-Protection: 0`` is the modern recommendation: the legacy auditor in
+old browsers introduced its own XSS class and OWASP / MDN now recommend
+disabling it explicitly. See OWASP "Secure Headers" project, 2024 guidance.
+
+CSP enforcement
+---------------
+Lands in **report-only** mode by default (``Content-Security-Policy-Report-Only``).
+Operators flip to enforcing via :data:`CSP_ENFORCING_ENV`. Rationale: the
+Svelte SPA uses inline styles deliberately; flipping straight to enforcing
+is a recipe for breaking the UI on a Friday. Soak in report-only first,
+fix any genuine surprises, then enforce.
+
+HSTS
+----
+Only emitted when the request's HTTPS-ness can be **verified**. A bare
+``X-Forwarded-Proto: https`` from an untrusted peer must NOT trigger HSTS,
+because an attacker who can MITM your HTTP traffic could pin HSTS for a
+year on a victim's browser. The check uses
+:func:`pinky_daemon.net.client_identity.resolve_client_identity` so trust
+evaluation matches the rest of the #433 track.
+
+Part of #433. Issue #437.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from collections.abc import Awaitable, Callable
+
+from fastapi import Request
+
+from pinky_daemon.config import DeploymentMode
+from pinky_daemon.net.client_identity import (
+    peer_is_trusted,
+    raw_peer_from_request,
+)
+
+__all__ = [
+    "CSP_ENFORCING_ENV",
+    "build_csp",
+    "build_security_headers_middleware",
+    "headers_for_mode",
+    "is_verified_https",
+]
+
+logger = logging.getLogger(__name__)
+
+#: Env var that flips CSP from report-only to enforcing.
+CSP_ENFORCING_ENV = "PINKY_CSP_ENFORCING"
+
+
+# ── Header builders ───────────────────────────────────────────────
+
+
+_PERMISSIONS_POLICY_BASIC = "camera=(), microphone=(), geolocation=()"
+_PERMISSIONS_POLICY_STRICT = (
+    "camera=(), microphone=(), geolocation=(), interest-cohort=(), "
+    "payment=(), usb=(), magnetometer=(), accelerometer=(), gyroscope=()"
+)
+
+
+def build_csp(mode: DeploymentMode) -> str:
+    """Build the Content-Security-Policy string for ``mode``.
+
+    Both ``lan`` and ``web`` policies allow ``'unsafe-inline'`` for
+    ``style-src`` because the Svelte SPA uses inline styles deliberately
+    (per CLAUDE.md frontend conventions). Removing it requires a refactor
+    that's out of scope for #437 — call out as known compromise.
+
+    ``web`` also locks down ``object-src`` and adds ``upgrade-insecure-requests``.
+    """
+    common = [
+        "default-src 'self'",
+        "script-src 'self'",
+        "style-src 'self' 'unsafe-inline'",  # Svelte inline styles, see CLAUDE.md
+        "img-src 'self' data:",
+        "font-src 'self' data:",
+        # SSE + websocket to same origin
+        "connect-src 'self' ws: wss:",
+        "frame-ancestors 'none'",
+        "base-uri 'self'",
+        "form-action 'self'",
+    ]
+    if mode is DeploymentMode.WEB:
+        return "; ".join(
+            common + ["object-src 'none'", "upgrade-insecure-requests"]
+        )
+    return "; ".join(common)
+
+
+def headers_for_mode(mode: DeploymentMode) -> dict[str, str]:
+    """Return the static (non-HSTS) header set for a given posture.
+
+    HSTS is intentionally absent — its inclusion depends on per-request
+    HTTPS-verification logic that lives in the middleware closure.
+    Callers that just want to inspect "what headers would mode X emit" use
+    this helper for the static slice.
+    """
+    out: dict[str, str] = {
+        "X-Content-Type-Options": "nosniff",
+        "X-XSS-Protection": "0",
+    }
+    if mode is DeploymentMode.TRUSTED:
+        return out
+    # lan + web share clickjacking + referrer + permissions-policy.
+    out["X-Frame-Options"] = "DENY"
+    out["Referrer-Policy"] = "strict-origin-when-cross-origin"
+    out["Permissions-Policy"] = (
+        _PERMISSIONS_POLICY_STRICT
+        if mode is DeploymentMode.WEB
+        else _PERMISSIONS_POLICY_BASIC
+    )
+    return out
+
+
+# ── HSTS / HTTPS verification ─────────────────────────────────────
+
+
+def is_verified_https(request: Request) -> bool:
+    """``True`` when the request can be **verified** to have arrived over HTTPS.
+
+    Two acceptance paths:
+
+    1. Direct TLS — ``request.url.scheme == "https"``. The ASGI server
+       (uvicorn behind a TLS-terminating socket) sets this; can't be
+       spoofed by header.
+    2. Trusted reverse proxy — ``X-Forwarded-Proto: https`` AND the
+       immediate peer is one of the configured trusted proxies (#442).
+
+    A bare ``X-Forwarded-Proto: https`` from an untrusted peer is **not**
+    enough — that's the HSTS-poisoning vector we explicitly want to avoid.
+    """
+    if request.url.scheme == "https":
+        return True
+
+    proto = request.headers.get("x-forwarded-proto", "").lower().strip()
+    if proto != "https":
+        return False
+
+    # The header says HTTPS — only trust it when it came from a configured
+    # reverse proxy. Bare X-Forwarded-Proto from an arbitrary peer must NOT
+    # be enough; that's the HSTS-poisoning vector.
+    state = request.app.state if request.app else None
+    trusted = (
+        getattr(state, "trusted_proxies", None) if state else None
+    ) or ()
+    raw_peer = raw_peer_from_request(request)
+    return peer_is_trusted(raw_peer, trusted)
+
+
+# ── Middleware factory ────────────────────────────────────────────
+
+
+def _csp_enforcing_from_env(env: dict | None = None) -> bool:
+    source = env if env is not None else os.environ
+    raw = (source.get(CSP_ENFORCING_ENV) or "").strip().lower()
+    return raw in ("1", "true", "yes", "on")
+
+
+def build_security_headers_middleware(*, csp_enforcing: bool | None = None):
+    """Return an ASGI middleware closure that emits hardened response headers.
+
+    Args:
+        csp_enforcing: When ``True``, emit ``Content-Security-Policy``
+            (enforcing). When ``False``, emit ``Content-Security-Policy-Report-Only``.
+            When ``None`` (default), read :data:`CSP_ENFORCING_ENV` per
+            request — slow path but lets tests flip behavior without
+            recreating the middleware.
+
+    The middleware reads ``app.state.deployment_mode`` per request so the
+    same closure works in tests that swap modes. Websocket upgrades pass
+    through untouched (header injection breaks WS handshakes in some
+    clients).
+    """
+
+    async def security_headers_mw(
+        request: Request,
+        call_next: Callable[[Request], Awaitable],
+    ):
+        # Don't touch WebSocket upgrades — Starlette handles those before
+        # response middleware anyway, but be explicit.
+        if request.headers.get("upgrade", "").lower() == "websocket":
+            return await call_next(request)
+
+        response = await call_next(request)
+
+        state = request.app.state if request.app else None
+        mode = getattr(state, "deployment_mode", DeploymentMode.TRUSTED)
+
+        for name, value in headers_for_mode(mode).items():
+            response.headers[name] = value
+
+        # CSP only in lan/web. Trusted is purely back-compat.
+        if mode is not DeploymentMode.TRUSTED:
+            csp_value = build_csp(mode)
+            enforcing = (
+                csp_enforcing
+                if csp_enforcing is not None
+                else _csp_enforcing_from_env()
+            )
+            header_name = (
+                "Content-Security-Policy"
+                if enforcing
+                else "Content-Security-Policy-Report-Only"
+            )
+            response.headers[header_name] = csp_value
+
+        # HSTS only in web mode and only when HTTPS is verified.
+        if mode is DeploymentMode.WEB and is_verified_https(request):
+            response.headers["Strict-Transport-Security"] = (
+                "max-age=31536000; includeSubDomains"
+            )
+
+        return response
+
+    return security_headers_mw

src/pinky_daemon/net/client_identity.py

@@ -65,7 +65,10 @@
     "ClientIdentity",
     "get_client_identity",
     "parse_trusted_proxies",
+    "peer_is_trusted",
+    "raw_peer_from_request",
     "resolve_client_identity",
+    "trusted_proxies_from_env",
 ]
 
 logger = logging.getLogger(__name__)
@@ -181,7 +184,7 @@ def _parse_ip(token: str) -> IPAddress | None:
         return None
 
 
-def _peer_is_trusted(peer: IPAddress, trusted: tuple[IPNetwork, ...]) -> bool:
+def peer_is_trusted(peer: IPAddress, trusted: tuple[IPNetwork, ...]) -> bool:
     """True when ``peer`` falls inside any configured trusted-proxy CIDR."""
     for net in trusted:
         # IPv4 vs IPv6 mismatch raises TypeError on `in` — guard explicitly so
@@ -258,7 +261,7 @@ def _warn_throttled(key: str, message: str) -> None:
 # ── Resolver ──────────────────────────────────────────────────────
 
 
-def _raw_peer_from_request(request: Request) -> IPAddress:
+def raw_peer_from_request(request: Request) -> IPAddress:
     """Pull the immediate TCP peer off a Starlette/FastAPI request.
 
     Falls back to the unspecified IPv4 address (``0.0.0.0``) when ``client``
@@ -291,7 +294,7 @@ def _client_from_chain(
     # Walk from the right (closest hop = last entry per RFC 7239 / XFF
     # convention). Skip any hop that is itself a trusted proxy.
     for hop in reversed(chain):
-        if not _peer_is_trusted(hop, trusted):
+        if not peer_is_trusted(hop, trusted):
             return hop
     # Whole chain is trusted proxies — the leftmost entry is the client.
     return chain[0]
@@ -307,7 +310,7 @@ def resolve_client_identity(
     See module docstring for the full spec. This is the single function every
     IP-consuming feature in the hardening track should call.
     """
-    raw_peer = _raw_peer_from_request(request)
+    raw_peer = raw_peer_from_request(request)
 
     # Trusted-LAN posture: never honor forwarded headers, full stop.
     if mode is DeploymentMode.TRUSTED:
@@ -342,7 +345,7 @@ def resolve_client_identity(
     # Did anyone send a forwarding header at all?
     sent_forwarding = bool(forwarded_header or xff_header)
 
-    if not _peer_is_trusted(raw_peer, trusted_proxies):
+    if not peer_is_trusted(raw_peer, trusted_proxies):
         # Untrusted peer claiming to be a proxy — drop their chain and log
         # so audit can correlate suspected spoofing attempts.
         if sent_forwarding: