atlas-iac/testing/tests/test_hermes_handoff_zero_evidence.py

"""Regressions for the zero-evidence fail-open in the absence checks.

``evaluate_names_absent`` used to return ``PASS`` when its step exited 0 with no
output, so five mandatory checks — the ones asserting that provider API keys,
forge credentials, cluster-admin bindings, and shared coordinator state are
*absent* — could report a pass on no evidence at all and turn a ``NO_GO`` into a
``GO``. These tests pin the real catalog specs, not synthetic look-alikes, and
reproduce both ways silence actually arrives: a POSIX pipeline whose exit status
comes from its last stage, and a ``kubectl -o jsonpath`` over drifted structure.
"""

from __future__ import annotations

import datetime
import shutil
import subprocess

import pytest

from testing.tests.test_hermes_handoff_support import load_handoff_module, outcome

catalog = load_handoff_module("hermes_handoff_catalog")
evaluators = load_handoff_module("hermes_handoff_evaluators")
harness_run = load_handoff_module("hermes_handoff_run")
model = load_handoff_module("hermes_handoff_model")
policy = load_handoff_module("hermes_handoff_policy")

# Every mandatory check whose whole claim is that something is not there.
ABSENCE_CHECKS = (
    "access.no-cluster-admin-binding-for-agent",
    "forge.worker-holds-no-git-credential-variable",
    "identity.no-provider-api-key-in-manifest",
    "identity.no-provider-api-key-in-pod",
    "pool.coordinator-retains-sole-state-ownership",
)


def real_catalog() -> dict:
    targets = catalog.Targets(now=datetime.datetime.now(datetime.UTC))
    return {spec.id: spec for spec in harness_run.build_catalog(targets)}


def test_the_five_absence_checks_are_exactly_the_mandatory_names_absent_set() -> None:
    """If a sixth absence check appears it must be covered here too."""
    specs = real_catalog()
    found = {
        identifier
        for identifier, spec in specs.items()
        if spec.rule == "names_absent" and spec.mandatory
    }
    assert found == set(ABSENCE_CHECKS)


@pytest.mark.parametrize("identifier", ABSENCE_CHECKS)
def test_a_real_absence_check_never_passes_on_zero_observations(
    identifier: str,
) -> None:
    """rc=0 with no lines is drift, so the verdict must be NOT_RUN, not PASS."""
    spec = real_catalog()[identifier]
    step_key = spec.expect["step"]

    silent = evaluators.evaluate(spec, {step_key: outcome(stdout="")})
    assert silent.status == model.NOT_RUN
    assert "no lines" in silent.reason
    assert silent.status != model.PASS

    whitespace = evaluators.evaluate(spec, {step_key: outcome(stdout="\n  \n\t\n")})
    assert whitespace.status == model.NOT_RUN


@pytest.mark.parametrize("identifier", ABSENCE_CHECKS)
def test_a_real_absence_check_still_passes_on_real_evidence(identifier: str) -> None:
    """The guard must not turn a healthy observation into a false NOT_RUN."""
    spec = real_catalog()[identifier]
    step_key = spec.expect["step"]
    benign = "clusterrolebinding.rbac.authorization.k8s.io/system:basic-user\nPATH\n"

    passed = evaluators.evaluate(spec, {step_key: outcome(stdout=benign)})
    assert passed.status == model.PASS
    assert passed.evidence["observed_count"] == 2


@pytest.mark.parametrize("identifier", ABSENCE_CHECKS)
def test_a_real_absence_check_still_fails_on_a_forbidden_name(identifier: str) -> None:
    spec = real_catalog()[identifier]
    step_key = spec.expect["step"]
    offender = next(
        iter(spec.expect.get("names", ()) or spec.expect.get("contains", ()))
    )

    verdict = evaluators.evaluate(
        spec, {step_key: outcome(stdout=f"OTHER\n{offender}")}
    )
    assert verdict.status == model.FAIL
    assert offender in verdict.reason


def test_zero_evidence_on_a_mandatory_absence_check_forces_no_go() -> None:
    """A NOT_RUN on a mandatory check must not be rounded up to a GO.

    Every other check is held at PASS so the verdict turns on the five absence
    checks alone: silence must cost the release its GO.
    """
    specs = real_catalog()
    silent, green = [], []
    for identifier, spec in specs.items():
        status = model.PASS
        if identifier in ABSENCE_CHECKS:
            evaluation = evaluators.evaluate(
                spec, {spec.expect["step"]: outcome(stdout="")}
            )
            assert evaluation.status == model.NOT_RUN
            status = evaluation.status
        silent.append(model.CheckResult(spec=spec, status=status))
        green.append(model.CheckResult(spec=spec, status=model.PASS))

    assert all(result.blocking for result in silent if result.status == model.NOT_RUN)
    assert model.Report(mode="read-only", started_at="t", results=silent).decision == (
        model.NO_GO
    )
    assert (
        model.Report(mode="read-only", started_at="t", results=green).decision
        == model.GO
    )


def test_a_broken_shell_pipeline_exits_zero_with_no_output() -> None:
    """The harness' own frozen template is reachable with rc=0 and 0 bytes.

    A POSIX pipeline reports the status of its *last* stage, so a missing or
    renamed upstream binary produces success-with-no-evidence rather than an
    error. This is one of the two live paths that made the fail-open reachable.
    """
    script = policy.render_shell("env_names").replace("/usr/bin/env", "/usr/bin/absent")

    completed = subprocess.run(  # noqa: S603
        ["/bin/sh", "-c", script],
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
    )

    assert completed.returncode == 0
    assert completed.stdout == ""

    spec = real_catalog()["identity.no-provider-api-key-in-pod"]
    verdict = evaluators.evaluate(
        spec, {"env": outcome(stdout=completed.stdout, returncode=completed.returncode)}
    )
    assert verdict.status == model.NOT_RUN


DRIFTED_JSONPATH_OUTPUT = {
    # A renamed parent — `containerz` — makes kubectl print nothing at all.
    "renamed parent": "",
    # A renamed leaf — `.nam` — makes it print only the separators.
    "renamed leaf": "\n" * 6,
    # A container filter that matches nothing behaves the same way.
    "unmatched filter": "",
}


@pytest.mark.parametrize("shape", sorted(DRIFTED_JSONPATH_OUTPUT))
def test_drifted_kubectl_jsonpath_output_is_not_run(shape: str) -> None:
    """The exact bytes a drifted ``-o jsonpath`` emits must never read as a pass.

    ``kubectl -o jsonpath`` exits 0 when the structure it walks is missing, so a
    renamed field, a renamed leaf, or an unmatched container filter all deliver
    success with no usable evidence. Pinned as literal output so the regression
    holds with no cluster and no network; the live rc=0 behaviour is asserted
    separately by :func:`test_a_live_drifted_projection_still_exits_zero`.
    """
    spec = real_catalog()["identity.no-provider-api-key-in-manifest"]

    verdict = evaluators.evaluate(
        spec, {"env": outcome(stdout=DRIFTED_JSONPATH_OUTPUT[shape])}
    )

    assert verdict.status == model.NOT_RUN
    assert "no lines" in verdict.reason


def test_a_live_drifted_projection_still_exits_zero() -> None:
    """Confirm against the real tool that drift is success, not an error.

    kubectl has no offline JSONPath renderer, so this reads the live agent
    workload — a plain, read-only ``get`` with a name-only projection — and skips
    when no cluster is reachable. The hermetic assertion above does not skip.
    """
    kubectl = shutil.which("kubectl")
    if kubectl is None:  # pragma: no cover - kubectl ships with the release image
        pytest.skip("kubectl is not installed in this environment")
    base = ["--namespace", "hermes", "get", "deploy/hermes-agent", "-o"]
    healthy = 'jsonpath={range .spec.template.spec.containers[*]}{.name}{"\\n"}{end}'
    probe = subprocess.run(  # noqa: S603
        [kubectl, *base, healthy],
        capture_output=True,
        text=True,
        timeout=60,
        check=False,
    )
    if probe.returncode != 0:
        pytest.skip("no read-only cluster access in this environment")
    assert probe.stdout.strip()

    for drifted in (
        'jsonpath={range .spec.template.spec.containerz[*]}{.name}{"\\n"}{end}',
        'jsonpath={range .spec.template.spec.containers[*]}{.nam}{"\\n"}{end}',
    ):
        result = subprocess.run(  # noqa: S603
            [kubectl, *base, drifted],
            capture_output=True,
            text=True,
            timeout=60,
            check=False,
        )
        assert result.returncode == 0
        assert result.stdout.strip() == ""
        spec = real_catalog()["identity.no-provider-api-key-in-manifest"]
        verdict = evaluators.evaluate(spec, {"env": outcome(stdout=result.stdout)})
        assert verdict.status == model.NOT_RUN


def test_every_absence_probe_yields_one_line_per_observed_item() -> None:
    """A healthy probe must produce evidence, or the guard would misfire.

    The pool claim projection emits ``<volume>=<claim>`` so a template volume
    without a PVC still counts as an observation; an empty result therefore only
    happens when the object or the projection has drifted.
    """
    spec = real_catalog()["pool.coordinator-retains-sole-state-ownership"]
    argv = spec.steps[0].argv
    assert '{.name}{"="}' in argv[-1]

    verdict = evaluators.evaluate(spec, {"claims": outcome(stdout="tmp=\nconfig=\n")})
    assert verdict.status == model.PASS
    assert verdict.evidence["observed_count"] == 2

    leaked = evaluators.evaluate(
        spec, {"claims": outcome(stdout="state=hermes-agent-home\n")}
    )
    assert leaked.status == model.FAIL