atlas-iac/testing/tests/test_hermes_public_host_continuity.py

"""Public chat/triage hostnames stay served on every layer during a rename.

PR #34 renamed the chat and triage public hosts in place: the legacy names left
the certificate SANs, the Ingress rules, the CoreDNS overrides and the Keycloak
ensure script in one change. The legacy hosts began answering 404 with Traefik's
default certificate while the renamed hosts could not finish a login, because
Keycloak matches ``redirect_uri`` exactly and still held the old callback. Chat
and triage were unreachable on every hostname at once.

These tests pin the contract that makes that outage impossible to reintroduce
silently: a public host is either served by all four layers or by none of them.
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest
import yaml

REPO = Path(__file__).parents[2]
HERMES = REPO / "services/hermes"
KEYCLOAK = REPO / "services/keycloak"
COREDNS = REPO / "infrastructure/core/coredns-custom.yaml"
ENSURE_SCRIPT = KEYCLOAK / "scripts/hermes_access_oidc_ensure.sh"

# Every hostname the chat/triage surfaces must answer on, with the oauth2-proxy
# backend that serves it. Both the renamed and the legacy names belong here
# until a deliberate retirement change removes a row from this table.
PUBLIC_HOSTS = {
    "chat.bstein.dev": "oauth2-proxy-hermes-chat",
    "chat.hermes.bstein.dev": "oauth2-proxy-hermes-chat",
    "triage.bstein.dev": "oauth2-proxy-hermes-triage",
    "triage.hermes.bstein.dev": "oauth2-proxy-hermes-triage",
}

# The agent surface was deliberately untouched by the rename.
AGENT_HOST = "agent.hermes.bstein.dev"


def _docs(path: Path) -> list[dict]:
    return [doc for doc in yaml.safe_load_all(path.read_text()) if doc]


def _named(path: Path, kind: str, name: str) -> dict:
    for doc in _docs(path):
        if doc.get("kind") == kind and doc["metadata"]["name"] == name:
            return doc
    raise AssertionError(f"{kind}/{name} missing from {path}")


@pytest.fixture(scope="module")
def certificate() -> dict:
    return _named(HERMES / "agent-certificate.yaml", "Certificate", "hermes-sites-tls")


@pytest.fixture(scope="module")
def sites_ingress() -> dict:
    return _named(HERMES / "agent-ingress.yaml", "Ingress", "hermes-sites")


@pytest.fixture(scope="module")
def coredns_hosts() -> set[str]:
    block = yaml.safe_load(COREDNS.read_text())["data"]["bstein-dev.server"]
    return {
        line.split()[1]
        for line in block.splitlines()
        if len(line.split()) == 2 and re.fullmatch(r"[\d.]+", line.split()[0])
    }


@pytest.fixture(scope="module")
def ensure_script() -> str:
    return ENSURE_SCRIPT.read_text()


@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
def test_host_is_on_the_shared_certificate(host: str, certificate: dict):
    """A host without a SAN serves Traefik's default self-signed certificate."""
    assert host in certificate["spec"]["dnsNames"]


@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
def test_host_resolves_inside_the_cluster(host: str, coredns_hosts: set[str]):
    assert host in coredns_hosts


@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
def test_host_has_an_ingress_rule_and_tls_entry(host: str, sites_ingress: dict):
    """A host without a rule answers 404 even though DNS and TLS look healthy."""
    spec = sites_ingress["spec"]
    assert host in {name for entry in spec["tls"] for name in entry["hosts"]}

    rule = next((item for item in spec["rules"] if item["host"] == host), None)
    assert rule is not None, f"no hermes-sites rule serves {host}"

    backends = {
        path["backend"]["service"]["name"] for path in rule["http"]["paths"]
    }
    assert backends == {PUBLIC_HOSTS[host]}


@pytest.mark.parametrize("host", sorted(PUBLIC_HOSTS))
def test_host_is_registered_with_keycloak(host: str, ensure_script: str):
    """Keycloak matches redirect_uri exactly, so every served host needs one."""
    assert f"https://{host}" in ensure_script


def test_agent_surface_was_not_touched_by_the_rename(
    certificate: dict, coredns_hosts: set[str], ensure_script: str
):
    assert AGENT_HOST in certificate["spec"]["dnsNames"]
    assert AGENT_HOST in coredns_hosts
    assert f"https://{AGENT_HOST}" in ensure_script


def test_ensure_script_registers_legacy_and_renamed_origins_together(
    ensure_script: str,
):
    """The renamed proxies must carry both origins; the agent proxy only one."""
    for client, canonical, legacy in (
        ("hermes-chat-proxy", "chat.bstein.dev", "chat.hermes.bstein.dev"),
        ("hermes-triage-proxy", "triage.bstein.dev", "triage.hermes.bstein.dev"),
    ):
        call = re.search(
            rf'ensure_proxy_client "{client}".*?(?=\nensure_)',
            ensure_script,
            re.DOTALL,
        )
        assert call, f"{client} is never ensured"
        assert f"https://{canonical}" in call.group(0)
        assert f"https://{legacy}" in call.group(0)


@pytest.mark.parametrize(
    "deployment", ["oauth2-proxy-hermes-chat", "oauth2-proxy-hermes-triage"]
)
def test_multi_host_proxies_derive_their_callback_from_the_request(deployment: str):
    """A pinned callback host breaks logins started on the other hostname.

    oauth2-proxy returns ``--redirect-url`` verbatim when it carries a host, so
    a login started on the legacy host would send the browser to the canonical
    host's callback. The CSRF cookie uses the ``__Host-`` prefix and cannot
    cross origins, so the callback fails to find it. Leaving the URL host-less
    makes oauth2-proxy build it from the request host instead.
    """
    spec = _named(HERMES / "oauth2-proxy.yaml", "Deployment", deployment)
    args = spec["spec"]["template"]["spec"]["containers"][0]["args"]
    redirect = next(a for a in args if a.startswith("--redirect-url="))
    value = redirect.split("=", 1)[1]
    assert value.startswith("/"), f"{deployment} pins a callback host: {value}"

    # Derivation only trusts X-Forwarded-Host behind a declared reverse proxy.
    assert "--reverse-proxy=true" in args
    assert any(a.startswith("--trusted-proxy-ip=") for a in args)


def test_ensure_job_is_rerun_whenever_the_script_changes():
    """The Job is immutable, so a stale name silently skips the rerun."""
    job = _named(
        KEYCLOAK / "bootstrap-jobs/hermes-access-oidc-client-job.yaml",
        "Job",
        # The suffix moves with every rerun; resolve it from the manifest.
        _job_name(),
    )
    assert job["spec"]["template"]["spec"]["containers"][0]["command"] == [
        "/scripts/hermes_access_oidc_ensure.sh"
    ]


def _job_name() -> str:
    path = KEYCLOAK / "bootstrap-jobs/hermes-access-oidc-client-job.yaml"
    name = yaml.safe_load(path.read_text())["metadata"]["name"]
    assert re.fullmatch(r"hermes-access-oidc-client-ensure-\d+", name), name
    return name