README.md

@@ -69,6 +69,34 @@ not a loopback name (DNS rebinding), when `Origin` is any other origin, when
 thumbnail), or when the body exceeds `PHOTO_PIPELINE_MAX_REQUEST_BYTES`. There is no
 CORS middleware at all, so no other origin can read a response.
 
+### Reaching it through a hostname or proxy (US08-01)
+
+| variable | meaning |
+|---|---|
+| `PHOTO_PIPELINE_ALLOWED_HOSTS` | comma-separated extra names the app answers to; empty means loopback only |
+| `PHOTO_PIPELINE_ACCESS_SECRET` | traded for the session cookie at `GET /api/v1/session` via `X-Access-Secret` |
+| `PHOTO_PIPELINE_TRUSTED_PROXIES` | comma-separated peer addresses whose `X-Forwarded-Proto`/`X-Forwarded-Host` are believed |
+
+Being reachable *was* the authentication: whoever could open `127.0.0.1:8000` owned
+the library. So naming any non-loopback host — or binding to one, `0.0.0.0` included
+— makes the access secret mandatory, and `serve` refuses to start without it rather
+than publishing the library. Loopback-only deployments need no secret and behave
+exactly as before.
+
+```bash
+curl -sc /tmp/pp.jar -H "X-Access-Secret: $PHOTO_PIPELINE_ACCESS_SECRET" \
+  https://photos.example.com/api/v1/session
+```
+
+The browser asks for the secret once per tab and keeps it in `sessionStorage`.
+Wrong secrets are rate-limited (5 per minute) and logged with the caller's address
+only. `Host` and `Origin` are judged against the configured names; the *external*
+scheme and host come from the forwarded headers only when the request arrived from a
+`PHOTO_PIPELINE_TRUSTED_PROXIES` address, so a client cannot declare its own origin,
+and the session cookie is marked `Secure` when that external scheme is HTTPS. Health
+endpoints stay reachable without the secret so an orchestrator can restart the
+container; nothing else does.
+
 ## Testing
 
 One offline command runs the whole suite (unit, integration, and browser

frontend/js/api.js

@@ -7,9 +7,28 @@ export const BASE = "/api/v1";
 // what makes it proof that the caller is this app and not another page.
 let csrfToken = null;
 
+// A deployment reachable through a proxy trades an operator secret for that cookie.
+// Kept per tab: sessionStorage dies with the tab, and the secret never enters a URL.
+const SECRET_KEY = "pp_access_secret";
+
+async function bootstrap(secret) {
+  return fetch(BASE + "/session", {
+    credentials: "same-origin",
+    headers: secret ? { "X-Access-Secret": secret } : {},
+  });
+}
+
 async function session() {
   if (csrfToken === null) {
-    const response = await fetch(BASE + "/session", { credentials: "same-origin" });
+    let response = await bootstrap(sessionStorage.getItem(SECRET_KEY));
+    if (response.status === 401) {
+      sessionStorage.removeItem(SECRET_KEY);
+      const secret = prompt("Access secret");
+      if (secret) {
+        response = await bootstrap(secret);
+        if (response.ok) sessionStorage.setItem(SECRET_KEY, secret);
+      }
+    }
     const body = await response.json().catch(() => null);
     csrfToken = (body && body.csrf_token) || null;
   }

photo_pipeline/__main__.py

@@ -209,15 +209,25 @@ def main(argv: Sequence[str] | None = None) -> int:
             lock.release()
         return 0
 
+    import sys
+
     import uvicorn
 
-    from photo_pipeline.api.app import create_app
+    from photo_pipeline.api.app import ConfigurationRefused, create_app
+
+    # An exposed deployment without an access secret must not reach the port at all,
+    # and the operator needs a sentence, not a traceback (US08-01).
+    try:
+        app = create_app(config)
+    except ConfigurationRefused as error:
+        print(str(error), file=sys.stderr)
+        return 4
 
     lock = LibraryLock(config, "api")
     if (held := _acquire(lock, allow_legacy=args.allow_legacy)) is not None:
         return held
     try:
-        uvicorn.run(create_app(config), host=config.host, port=config.port)
+        uvicorn.run(app, host=config.host, port=config.port)
     finally:
         lock.release()
     return 0

photo_pipeline/api/app.py

@@ -35,7 +35,13 @@ from photo_pipeline.api.routes import (
     uploads,
     workflow,
 )
-from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Session
+from photo_pipeline.api.security import (
+    DEFAULT_HEADERS,
+    FailureLimiter,
+    SecurityMiddleware,
+    Session,
+    trust_refusal,
+)
 
 # Registers the safety_score / analysis job handlers on import.
 import photo_pipeline.jobs.domain_handlers  # noqa: F401
@@ -84,9 +90,16 @@ def _install_error_handlers(app: FastAPI) -> None:
         return _envelope(500, "internal_error", "internal error")
 
 
+class ConfigurationRefused(RuntimeError):
+    """The configuration would serve the library to callers it cannot authenticate."""
+
+
 def create_app(config: Config | None = None) -> FastAPI:
     config = config or Config.from_env()
     configure_logging(config.log_level, config.log_format)
+    # Before anything is built, let alone bound to a port (US08-01).
+    if (why := trust_refusal(config)) is not None:
+        raise ConfigurationRefused(why)
 
     @asynccontextmanager
     async def lifespan(app: FastAPI):
@@ -114,6 +127,10 @@ def create_app(config: Config | None = None) -> FastAPI:
     # One session per process: the browser exchanges it for a cookie + CSRF token,
     # and every other origin is refused before a route ever runs (US07-02).
     app.state.session = Session.create()
+    app.state.access_limiter = FailureLimiter()
+    # Also set in the lifespan, but the bootstrap route reads it, and a caller can
+    # arrive before anything else has touched app.state.
+    app.state.config = config
     app.add_middleware(SecurityMiddleware, session=app.state.session, config=config)
     _install_error_handlers(app)
     app.include_router(session_routes.router, prefix="/api/v1")

photo_pipeline/api/routes/session.py

@@ -3,21 +3,52 @@
 It sets the ``HttpOnly``/``SameSite=Strict`` session cookie and returns the CSRF
 token in the body. A foreign page can call this — it just cannot read the answer,
 because the app sends no CORS headers — and the cookie it received is never attached
-to a request that foreign page initiates.
+to a request that foreign page initiated.
+
+When an access secret is configured (mandatory as soon as the app is reachable from
+another machine, US08-01) this is also the authentication gate: the secret buys the
+cookie, and every route behind it keeps asking for exactly the session and CSRF token
+it asked for before. Wrong secrets are counted, and a burst of them stops being
+answered — otherwise a proxy-exposed deployment could be guessed at indefinitely.
 """
 
 from __future__ import annotations
 
+import logging
+import secrets
+
 from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 
-from photo_pipeline.api.security import SESSION_COOKIE
+from photo_pipeline.api.security import ACCESS_SECRET_HEADER, SESSION_COOKIE
 
 router = APIRouter(tags=["session"])
 
+log = logging.getLogger(__name__)
+
+
+def _refuse(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
 
 @router.get("/session")
 def start_session(request: Request) -> JSONResponse:
+    config = request.app.state.config
+    secret = config.access_secret
+    if secret is not None:
+        limiter = request.app.state.access_limiter
+        if limiter.blocked():
+            return _refuse(429, "too_many_attempts", "too many failed attempts; retry later")
+        offered = request.headers.get(ACCESS_SECRET_HEADER, "")
+        if not secrets.compare_digest(offered, secret.get_secret_value()):
+            limiter.record_failure()
+            # The client address is the whole record: the offered secret, the issued
+            # session, and the request body all stay out of the log.
+            log.warning(
+                "access secret rejected", extra={"client": _client(request), "path": "/session"}
+            )
+            return _refuse(401, "access_denied", "a valid access secret is required")
+
     session = request.app.state.session
     response = JSONResponse({"csrf_token": session.csrf_token})
     response.set_cookie(
@@ -25,6 +56,13 @@ def start_session(request: Request) -> JSONResponse:
         session.id,
         httponly=True,
         samesite="strict",
+        # HTTPS outside means the cookie must never travel over a plain hop, even one
+        # this process cannot see. Loopback http keeps working unchanged.
+        secure=request.scope.get("state", {}).get("external_scheme") == "https",
         path="/",
     )
     return response
+
+
+def _client(request: Request) -> str:
+    return request.client.host if request.client else "unknown"

photo_pipeline/api/security.py

@@ -19,6 +19,14 @@ The defenses stack, because each one alone has a hole:
   only in the bootstrap response body, which a foreign page cannot read (no CORS) —
   so possessing it proves the caller is same-origin.
 
+Behind a reverse proxy (US08-01) the same stack holds with two substitutions: the
+allowed host set comes from configuration instead of being the loopback names, and
+the host/scheme the policy judges is the *external* one, which is only read from
+``X-Forwarded-*`` when the request actually arrived from a configured proxy. The
+loopback check was standing in for authentication, so naming a non-loopback host
+also makes an access secret mandatory — ``trust_refusal`` refuses to start without
+one, and the secret is what the bootstrap endpoint trades for the session cookie.
+
 ``evaluate`` is a pure function over the request metadata: the whole policy is one
 table that a unit test can enumerate, and the middleware only applies its verdict.
 """
@@ -26,6 +34,7 @@ table that a unit test can enumerate, and the middleware only applies its verdic
 from __future__ import annotations
 
 import secrets
+import time
 from collections.abc import Mapping
 from dataclasses import dataclass
 from urllib.parse import urlsplit
@@ -35,6 +44,7 @@ from starlette.responses import JSONResponse
 
 SESSION_COOKIE = "pp_session"
 CSRF_HEADER = "x-csrf-token"
+ACCESS_SECRET_HEADER = "x-access-secret"
 API_PREFIX = "/api/v1"
 SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 # Reachable without a session: liveness/readiness (an orchestrator has no cookie)
@@ -96,6 +106,28 @@ def split_host(value: str) -> tuple[str, str]:
     return host, port
 
 
+def external_view(
+    *,
+    client: str | None,
+    headers: Mapping[str, str],
+    scheme: str,
+    trusted_proxies: frozenset[str],
+) -> tuple[str, str]:
+    """The ``(scheme, host)`` the caller used, as opposed to the one this hop saw.
+
+    Forwarded headers are a client-supplied claim. Believing them from anyone lets a
+    request declare its own origin — and origin is half of this module's evidence —
+    so they count only when the connection came from a configured proxy.
+    """
+    host = headers.get("host", "")
+    if client is None or client not in trusted_proxies:
+        return scheme, host
+    # A chain appends: the first entry is what the original client asked for.
+    forwarded_proto = headers.get("x-forwarded-proto", "").split(",")[0].strip().lower()
+    forwarded_host = headers.get("x-forwarded-host", "").split(",")[0].strip()
+    return forwarded_proto or scheme, forwarded_host or host
+
+
 def evaluate(
     *,
     method: str,
@@ -103,20 +135,26 @@ def evaluate(
     headers: Mapping[str, str],
     session: Session,
     allowed_hosts: frozenset[str] = LOOPBACK_HOSTS,
+    scheme: str = "http",
     max_request_bytes: int,
 ) -> Refusal | None:
-    """Why this request must be refused, or ``None`` when it may proceed."""
+    """Why this request must be refused, or ``None`` when it may proceed.
+
+    ``headers["host"]`` and ``scheme`` are the external ones (see ``external_view``);
+    the allowed origins are the allowed hosts under that scheme and port, so there is
+    no second list that can drift away from the first.
+    """
     host_header = headers.get("host", "")
     host, port = split_host(host_header)
     if host.lower() not in allowed_hosts:
-        return Refusal(403, "host_not_allowed", "request host is not a local address")
+        return Refusal(403, "host_not_allowed", "request host is not an allowed address")
 
     origin = headers.get("origin")
     if origin is not None and origin != "":
         parts = urlsplit(origin)
         origin_host, origin_port = split_host(parts.netloc)
         if (
-            parts.scheme not in ("http", "https")
+            parts.scheme != scheme
             or origin_host.lower() not in allowed_hosts
             or origin_port != port
         ):
@@ -142,6 +180,52 @@ def evaluate(
     return None
 
 
+def exposed_hosts(config) -> list[str]:
+    """Configured names by which this application is reachable from another machine."""
+    names = {str(config.host).lower()}
+    names.update(split_host(name)[0].lower() for name in config.allowed_hosts)
+    return sorted(names - LOOPBACK_HOSTS)
+
+
+def trust_refusal(config) -> str | None:
+    """Why this configuration must not serve at all, or ``None``.
+
+    Reaching the app used to prove ownership of it. The moment a configuration makes
+    it reachable from elsewhere that stops being true, so serving without a secret
+    would publish the library — refuse at startup rather than at the first request,
+    when the operator is no longer watching (US08-01).
+    """
+    exposed = exposed_hosts(config)
+    if exposed and config.access_secret is None:
+        return (
+            f"refusing to serve: {', '.join(exposed)} is reachable from outside this "
+            "machine, so PHOTO_PIPELINE_ACCESS_SECRET must be set"
+        )
+    return None
+
+
+class FailureLimiter:
+    """Bounded failed access-secret attempts, so the secret cannot be guessed online.
+
+    ponytail: one counter for the whole process rather than per client address —
+    behind a proxy every attempt arrives from the same address anyway. Per-caller
+    buckets if the app is ever exposed without one.
+    """
+
+    def __init__(self, limit: int = 5, window: float = 60.0) -> None:
+        self.limit = limit
+        self.window = window
+        self._failures: list[float] = []
+
+    def blocked(self) -> bool:
+        now = time.monotonic()
+        self._failures = [at for at in self._failures if now - at < self.window]
+        return len(self._failures) >= self.limit
+
+    def record_failure(self) -> None:
+        self._failures.append(time.monotonic())
+
+
 class SecurityMiddleware:
     """Pure-ASGI so the SSE stream keeps streaming (BaseHTTPMiddleware buffers)."""
 
@@ -150,7 +234,12 @@ class SecurityMiddleware:
         self.session = session
         self.config = config
         self.max_request_bytes = config.max_request_bytes
-        self.allowed_hosts = frozenset(LOOPBACK_HOSTS | {str(config.host).lower()})
+        self.allowed_hosts = frozenset(
+            LOOPBACK_HOSTS
+            | {str(config.host).lower()}
+            | {split_host(name)[0].lower() for name in config.allowed_hosts}
+        )
+        self.trusted_proxies = frozenset(config.trusted_proxies)
 
     async def __call__(self, scope, receive, send) -> None:
         if scope["type"] != "http":
@@ -161,12 +250,22 @@ class SecurityMiddleware:
         # policy never has to parse a Cookie header.
         lookup = dict(headers)
         lookup["cookie-session"] = _cookie(headers.get("cookie", ""), SESSION_COOKIE)
+        client = scope.get("client")
+        scheme, lookup["host"] = external_view(
+            client=client[0] if client else None,
+            headers=headers,
+            scheme=scope.get("scheme", "http"),
+            trusted_proxies=self.trusted_proxies,
+        )
+        # What the session cookie's Secure flag is decided from, one hop later.
+        scope.setdefault("state", {})["external_scheme"] = scheme
         refusal = evaluate(
             method=scope.get("method", "GET"),
             path=scope.get("path", "/"),
             headers=lookup,
             session=self.session,
             allowed_hosts=self.allowed_hosts,
+            scheme=scheme,
             max_request_bytes=self.max_request_bytes,
         )
         if refusal is None:

photo_pipeline/config.py

@@ -7,6 +7,9 @@ real external call needs them.
 
 pydantic-settings would do this too, but a prefix-scan over the declared fields
 is a few lines and one fewer dependency.
+
+Tuple-valued settings are lists in one variable: library roots are ``os.pathsep``
+separated because they are paths, everything else is comma separated.
 """
 
 from __future__ import annotations
@@ -20,6 +23,7 @@ from pydantic import BaseModel, ConfigDict, SecretStr
 ENV_PREFIX = "PHOTO_PIPELINE_"
 ENV_FILE_VAR = f"{ENV_PREFIX}ENV_FILE"
 DEFAULT_ENV_FILE = Path(".env")
+COMMA_LIST_FIELDS = frozenset({"allowed_hosts", "trusted_proxies"})
 
 # The archived CLI's variable names, so the configuration file an operator already
 # has keeps working. The vision provider reads the OpenAI SDK's names, and the
@@ -85,6 +89,19 @@ class Config(BaseModel):
     log_level: str = "INFO"
     log_format: str = "json"  # "json" or "text"
 
+    # Trust boundary (US08-01). Empty means loopback only, which is what the app did
+    # before there was a setting: a request whose Host is not a loopback name is
+    # refused, and no secret is needed because nothing outside this machine can call.
+    # Naming a real hostname here is what makes the app reachable through a reverse
+    # proxy, and it is exactly then that ``access_secret`` becomes mandatory.
+    allowed_hosts: tuple[str, ...] = ()
+    # Addresses whose ``X-Forwarded-Proto``/``X-Forwarded-Host`` may be believed. A
+    # client that is not the proxy can otherwise declare its own origin.
+    trusted_proxies: tuple[str, ...] = ()
+    # Exchanged for the session cookie at the bootstrap endpoint. Once set it is
+    # required even on loopback, so a development setup cannot half-enable it.
+    access_secret: SecretStr | None = None
+
     # Largest request body the API accepts. Every endpoint takes small JSON commands;
     # anything larger is a mistake or an attempt to exhaust memory (US07-02).
     max_request_bytes: int = 1_048_576
@@ -131,5 +148,10 @@ class Config(BaseModel):
             raw = env.get(ENV_PREFIX + name.upper())
             if not raw:
                 continue
-            data[name] = raw.split(os.pathsep) if name == "library_roots" else raw
+            if name == "library_roots":
+                data[name] = raw.split(os.pathsep)
+            elif name in COMMA_LIST_FIELDS:
+                data[name] = [part.strip() for part in raw.split(",") if part.strip()]
+            else:
+                data[name] = raw
         return cls(**data)

tests/integration/test_trusted_hosts.py

@@ -0,0 +1,253 @@
+"""US08-01: the configurable trust boundary and its authentication gate.
+
+Until now, reaching the app proved ownership of it: it answered only to loopback
+names. A container behind a reverse proxy answers to a real hostname, so these tests
+pin the two halves that replace that proof — the app refuses to start exposed without
+an access secret, and the secret is the only way to obtain the session every other
+route already required (US07-02, unchanged and re-asserted here).
+
+The suite's ``conftest`` bootstraps a session for any ``TestClient`` automatically,
+which is precisely what an unauthenticated caller does not get; ``raw_client``
+pre-seeds a placeholder CSRF header to opt out of that convenience.
+"""
+
+from __future__ import annotations
+
+import pytest
+from starlette.testclient import TestClient
+
+from photo_pipeline.api.app import ConfigurationRefused, create_app
+from photo_pipeline.api.security import ACCESS_SECRET_HEADER, CSRF_HEADER, SESSION_COOKIE
+from photo_pipeline.config import Config
+
+SECRET = "operator-secret-value"
+HOSTNAME = "photos.example.com"
+# What Starlette reports as the peer address of an in-process request.
+TESTCLIENT_ADDRESS = "testclient"
+
+# One of each route class: a read, a mutation, and a media endpoint.
+PROTECTED = [
+    ("GET", "/api/v1/workflow", None),
+    ("POST", "/api/v1/albums/proposals", {}),
+    ("GET", "/api/v1/assets/unknown-asset/thumbnail?size=256", None),
+]
+
+
+def config(tmp_path, **overrides) -> Config:
+    return Config(data_dir=tmp_path / "data", **overrides)
+
+
+def raw_client(app, base_url="http://127.0.0.1") -> TestClient:
+    client = TestClient(app, base_url=base_url)
+    client.headers[CSRF_HEADER] = "placeholder"
+    return client
+
+
+def exchange(client, secret=SECRET, headers=None):
+    return client.get("/api/v1/session", headers={ACCESS_SECRET_HEADER: secret, **(headers or {})})
+
+
+# ── startup: exposure without a secret is refused, loopback is unchanged ──────
+
+
+@pytest.mark.parametrize(
+    "exposure,exposed",
+    [({"allowed_hosts": (HOSTNAME,)}, HOSTNAME), ({"host": "0.0.0.0"}, "0.0.0.0")],
+)
+def test_an_exposed_configuration_refuses_to_serve_without_a_secret(tmp_path, exposure, exposed):
+    with pytest.raises(ConfigurationRefused) as refused:
+        create_app(config(tmp_path, **exposure))
+    assert "PHOTO_PIPELINE_ACCESS_SECRET" in str(refused.value)
+    # The message names what is exposed, so the operator knows which setting did it.
+    assert exposed in str(refused.value)
+
+
+def test_the_serve_command_reports_the_refusal_instead_of_binding(tmp_path, monkeypatch, capsys):
+    """Exit before the port, the lock, and the database, with a sentence not a trace."""
+    from photo_pipeline.__main__ import main
+
+    monkeypatch.setenv("PHOTO_PIPELINE_DATA_DIR", str(tmp_path / "data"))
+    monkeypatch.setenv("PHOTO_PIPELINE_ALLOWED_HOSTS", HOSTNAME)
+    monkeypatch.delenv("PHOTO_PIPELINE_ACCESS_SECRET", raising=False)
+
+    assert main(["serve"]) == 4
+    assert "PHOTO_PIPELINE_ACCESS_SECRET" in capsys.readouterr().err
+
+
+def test_an_exposed_configuration_with_a_secret_starts(tmp_path):
+    app = create_app(config(tmp_path, allowed_hosts=(HOSTNAME,), access_secret=SECRET))
+    with raw_client(app, base_url=f"http://{HOSTNAME}") as client:
+        assert exchange(client).status_code == 200
+
+
+def test_a_loopback_configuration_still_needs_no_secret(tmp_path):
+    """An unset trust boundary must behave exactly as it did before this story."""
+    with raw_client(create_app(config(tmp_path))) as client:
+        response = client.get("/api/v1/session")
+        assert response.status_code == 200
+        assert response.json()["csrf_token"]
+        assert "secure" not in response.headers["set-cookie"].lower()
+
+
+# ── the exchange: secret in, session out ─────────────────────────────────────
+
+
+@pytest.fixture
+def gated(tmp_path):
+    app = create_app(
+        config(
+            tmp_path,
+            allowed_hosts=(HOSTNAME,),
+            access_secret=SECRET,
+            trusted_proxies=(TESTCLIENT_ADDRESS,),
+        )
+    )
+    with raw_client(app, base_url=f"http://{HOSTNAME}") as client:
+        yield client
+
+
+def test_the_secret_buys_the_session_and_the_session_buys_the_routes(gated):
+    response = exchange(gated)
+    assert response.status_code == 200
+    cookie = response.headers["set-cookie"].lower()
+    assert "httponly" in cookie and "samesite=strict" in cookie
+    gated.headers[CSRF_HEADER] = response.json()["csrf_token"]
+
+    # The session and CSRF requirements behind the gate are the ones US07-02 set.
+    assert gated.get("/api/v1/workflow").status_code == 200
+    assert gated.post("/api/v1/albums/proposals", json={}).status_code == 200
+    refused = gated.post("/api/v1/albums/proposals", json={}, headers={CSRF_HEADER: "guessed"})
+    assert refused.json()["error"]["code"] == "csrf_failed"
+
+
+@pytest.mark.parametrize("offered", ["", "wrong-secret", SECRET + "x", SECRET.upper()])
+def test_a_wrong_secret_buys_nothing(gated, offered):
+    response = exchange(gated, secret=offered)
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "access_denied"
+    assert "set-cookie" not in response.headers
+
+
+def test_a_refusal_never_echoes_the_secret_or_the_session(gated, caplog):
+    with caplog.at_level("WARNING"):
+        response = exchange(gated, secret="wrong-secret")
+    assert SECRET not in response.text and "wrong-secret" not in response.text
+    assert SECRET not in caplog.text
+    # Logged as an event with its caller, without the session it did not get.
+    assert "access secret rejected" in caplog.text
+
+
+def test_guessing_is_rate_limited(gated):
+    codes = [exchange(gated, secret=f"guess-{n}").status_code for n in range(6)]
+    assert codes.count(401) == 5 and codes[-1] == 429
+    assert gated.get("/api/v1/session").status_code == 429
+    # The right secret is refused too while the limiter holds: that is the point.
+    blocked = exchange(gated)
+    assert blocked.status_code == 429
+    assert SECRET not in blocked.text
+
+
+def test_every_route_class_is_unreachable_without_the_secret(gated):
+    for method, path, body in PROTECTED:
+        response = gated.request(method, path, json=body)
+        assert response.status_code == 401, path
+        assert response.json()["error"]["code"] == "unauthenticated", path
+    # Health stays open: an orchestrator restarting the container holds no secret.
+    assert gated.get("/api/v1/health/live").status_code == 200
+    assert gated.get("/api/v1/health/ready").status_code == 200
+
+
+def test_a_session_from_another_process_is_not_replayable(tmp_path):
+    """Sessions live in the process, so a cookie captured from a previous one — a
+    restarted container, or a second deployment — must not open this one."""
+    settings = dict(allowed_hosts=(HOSTNAME,), access_secret=SECRET)
+    first, second = (create_app(config(tmp_path / str(n), **settings)) for n in (1, 2))
+    with raw_client(first, base_url=f"http://{HOSTNAME}") as client:
+        exchange(client)
+        stolen = client.cookies[SESSION_COOKIE]
+
+    with raw_client(second, base_url=f"http://{HOSTNAME}") as client:
+        client.cookies.set(SESSION_COOKIE, stolen, domain=HOSTNAME)
+        response = client.get("/api/v1/workflow")
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "unauthenticated"
+
+
+def test_a_cross_site_request_is_still_refused_behind_the_gate(gated):
+    gated.headers[CSRF_HEADER] = exchange(gated).json()["csrf_token"]
+    refused = gated.post(
+        "/api/v1/albums/proposals", json={}, headers={"Origin": "https://evil.example"}
+    )
+    assert refused.json()["error"]["code"] == "origin_not_allowed"
+    embedded = gated.get(
+        "/api/v1/assets/unknown-asset/thumbnail?size=256", headers={"Sec-Fetch-Site": "cross-site"}
+    )
+    assert embedded.json()["error"]["code"] == "cross_site_blocked"
+
+
+def test_an_unconfigured_host_is_refused_even_with_a_valid_session(gated):
+    gated.headers[CSRF_HEADER] = exchange(gated).json()["csrf_token"]
+    for host in ("other.example.com", "192.168.1.10"):
+        response = gated.get("/api/v1/workflow", headers={"Host": host})
+        assert response.status_code == 403, host
+        assert response.json()["error"]["code"] == "host_not_allowed", host
+
+
+# ── forwarded headers: believed from the proxy, ignored from anyone else ──────
+
+
+def test_a_trusted_proxys_https_makes_the_cookie_secure(gated):
+    """The proxy speaks HTTPS outward and HTTP to this app, so only the header knows."""
+    assert "secure" in exchange(gated, headers={"X-Forwarded-Proto": "https"}).headers[
+        "set-cookie"
+    ].lower()
+    assert "secure" not in exchange(gated).headers["set-cookie"].lower()
+
+
+def test_the_external_scheme_is_part_of_the_accepted_origin(gated):
+    gated.headers[CSRF_HEADER] = exchange(gated).json()["csrf_token"]
+    allowed = gated.post(
+        "/api/v1/albums/proposals",
+        json={},
+        headers={"X-Forwarded-Proto": "https", "Origin": f"https://{HOSTNAME}"},
+    )
+    assert allowed.status_code == 200
+    # The scheme is part of the origin: the same name over plain HTTP is not it.
+    refused = gated.post(
+        "/api/v1/albums/proposals",
+        json={},
+        headers={"X-Forwarded-Proto": "https", "Origin": f"http://{HOSTNAME}"},
+    )
+    assert refused.json()["error"]["code"] == "origin_not_allowed"
+
+
+def test_a_trusted_proxys_forwarded_host_is_the_host_that_is_judged(tmp_path):
+    """The proxy terminates the operator's hostname and dials this app by address."""
+    app = create_app(
+        config(
+            tmp_path,
+            allowed_hosts=(HOSTNAME,),
+            access_secret=SECRET,
+            trusted_proxies=(TESTCLIENT_ADDRESS,),
+        )
+    )
+    with raw_client(app, base_url="http://10.0.0.5") as client:
+        forwarded = {"X-Forwarded-Host": HOSTNAME}
+        assert exchange(client, headers=forwarded).status_code == 200
+        # Without the header the address it was dialled by is not an allowed name.
+        assert exchange(client).json()["error"]["code"] == "host_not_allowed"
+
+
+def test_forwarded_headers_from_an_untrusted_client_are_ignored(tmp_path):
+    """Otherwise any caller could declare the hostname and scheme of its choosing."""
+    app = create_app(config(tmp_path, allowed_hosts=(HOSTNAME,), access_secret=SECRET))
+    with raw_client(app, base_url="http://evil.example") as client:
+        forged = exchange(client, headers={"X-Forwarded-Host": HOSTNAME})
+        assert forged.json()["error"]["code"] == "host_not_allowed"
+
+    with raw_client(app, base_url=f"http://{HOSTNAME}") as client:
+        # A forged scheme would flip the cookie's Secure flag on a plain connection,
+        # which is how a cookie gets set and then never sent again.
+        response = exchange(client, headers={"X-Forwarded-Proto": "https"})
+        assert response.status_code == 200
+        assert "secure" not in response.headers["set-cookie"].lower()

tests/story_traceability.json

@@ -174,10 +174,13 @@
     "US07-07": [
       "tests/e2e/test_release_gate.py",
       "tests/e2e/test_release_journey.py"
+    ],
+    "US08-01": [
+      "tests/unit/test_security_policy.py",
+      "tests/integration/test_trusted_hosts.py"
     ]
   },
   "planned": [
-    "US08-01",
     "US08-02",
     "US08-03",
     "US08-04",

tests/unit/test_security_policy.py

@@ -13,18 +13,31 @@ import pytest
 
 from photo_pipeline.api.security import (
     CSRF_HEADER,
+    LOOPBACK_HOSTS,
     PUBLIC_PATHS,
+    FailureLimiter,
     Session,
     evaluate,
+    exposed_hosts,
+    external_view,
     split_host,
+    trust_refusal,
 )
+from photo_pipeline.config import Config
 
 SESSION = Session(id="session-id", csrf_token="csrf-token")
 HOST = "127.0.0.1:8000"
 LIMIT = 1024
+HOSTNAME = "photos.example.com"
 
 
-def check(method="GET", path="/api/v1/workflow", **headers):
+def check(
+    method="GET",
+    path="/api/v1/workflow",
+    allowed_hosts=LOOPBACK_HOSTS,
+    scheme="http",
+    **headers,
+):
     """Evaluate a request that is authenticated and same-origin unless overridden."""
     sent = {
         "host": HOST,
@@ -38,6 +51,8 @@ def check(method="GET", path="/api/v1/workflow", **headers):
         path=path,
         headers=sent,
         session=SESSION,
+        allowed_hosts=allowed_hosts,
+        scheme=scheme,
         max_request_bytes=LIMIT,
     )
 
@@ -155,6 +170,115 @@ def test_refusals_name_no_path_secret_or_internal():
         assert "/" not in refusal.message
 
 
+# ── US08-01: the same table with a configured trust boundary ─────────────────
+
+CONFIGURED = frozenset(LOOPBACK_HOSTS | {HOSTNAME})
+
+
+def test_a_configured_host_is_accepted_and_its_neighbours_are_not():
+    assert check(host=HOSTNAME, allowed_hosts=CONFIGURED) is None
+    for host in ("other.example.com", f"evil-{HOSTNAME}", "192.168.1.10"):
+        refusal = check(host=host, allowed_hosts=CONFIGURED)
+        assert (refusal.status, refusal.code) == (403, "host_not_allowed"), host
+
+
+def test_the_loopback_default_refuses_a_host_nobody_configured():
+    """The default set is what the app enforced before there was a setting."""
+    refusal = check(host=HOSTNAME)
+    assert (refusal.status, refusal.code) == (403, "host_not_allowed")
+
+
+def test_the_origin_must_match_the_external_scheme():
+    for scheme in ("http", "https"):
+        assert (
+            check(
+                method="POST",
+                host=HOSTNAME,
+                origin=f"{scheme}://{HOSTNAME}",
+                allowed_hosts=CONFIGURED,
+                scheme=scheme,
+            )
+            is None
+        )
+    # An HTTPS deployment whose caller claims plain HTTP is a different origin.
+    refusal = check(
+        method="POST",
+        host=HOSTNAME,
+        origin=f"http://{HOSTNAME}",
+        allowed_hosts=CONFIGURED,
+        scheme="https",
+    )
+    assert (refusal.status, refusal.code) == (403, "origin_not_allowed")
+
+
+def view(client, *, trusted=(), **headers):
+    sent = {name.replace("_", "-"): value for name, value in headers.items()}
+    return external_view(
+        client=client,
+        headers={"host": HOST, **sent},
+        scheme="http",
+        trusted_proxies=frozenset(trusted),
+    )
+
+
+def test_forwarded_headers_are_ignored_without_a_trusted_proxy():
+    forged = {"x_forwarded_proto": "https", "x_forwarded_host": HOSTNAME}
+    assert view("10.0.0.9", **forged) == ("http", HOST)
+    assert view(None, **forged) == ("http", HOST)
+    # Configuring *a* proxy does not trust a caller that is not it.
+    assert view("10.0.0.9", trusted=("10.0.0.1",), **forged) == ("http", HOST)
+
+
+def test_a_trusted_proxy_defines_the_external_scheme_and_host():
+    assert view(
+        "10.0.0.1", trusted=("10.0.0.1",), x_forwarded_proto="https", x_forwarded_host=HOSTNAME
+    ) == ("https", HOSTNAME)
+    # A chain: the first entry is what the original client asked for.
+    assert view(
+        "10.0.0.1",
+        trusted=("10.0.0.1",),
+        x_forwarded_proto="https, http",
+        x_forwarded_host=f"{HOSTNAME}, inner.internal",
+    ) == ("https", HOSTNAME)
+    # Trusted but silent: this hop's own view stands.
+    assert view("10.0.0.1", trusted=("10.0.0.1",)) == ("http", HOST)
+
+
+@pytest.mark.parametrize(
+    "settings,exposed",
+    [
+        ({}, []),
+        ({"host": "127.0.0.1"}, []),
+        ({"allowed_hosts": ("localhost", "127.0.0.1")}, []),
+        ({"allowed_hosts": (f"{HOSTNAME}:8443",)}, [HOSTNAME]),
+        ({"host": "0.0.0.0", "allowed_hosts": (HOSTNAME,)}, ["0.0.0.0", HOSTNAME]),
+    ],
+)
+def test_exposed_hosts_names_only_what_another_machine_can_reach(settings, exposed):
+    assert exposed_hosts(Config(**settings)) == exposed
+
+
+def test_an_exposed_configuration_without_a_secret_must_not_serve():
+    refusal = trust_refusal(Config(allowed_hosts=(HOSTNAME,)))
+    assert HOSTNAME in refusal and "PHOTO_PIPELINE_ACCESS_SECRET" in refusal
+    assert trust_refusal(Config(allowed_hosts=(HOSTNAME,), access_secret="s")) is None
+    # Loopback-only, with and without a secret, is unchanged.
+    assert trust_refusal(Config()) is None
+    assert trust_refusal(Config(access_secret="s")) is None
+
+
+def test_failed_attempts_are_bounded_per_window():
+    limiter = FailureLimiter(limit=2, window=60.0)
+    assert not limiter.blocked()
+    limiter.record_failure()
+    assert not limiter.blocked()
+    limiter.record_failure()
+    assert limiter.blocked()
+    # Attempts age out, so a locked-out operator is not locked out forever.
+    limiter._failures = [-120.0, -120.0]
+    assert not limiter.blocked()
+
+
 @pytest.mark.parametrize(
     "value,expected",
     [