photoanalyzer/tests/integration/test_trusted_hosts.py

"""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()