atlas-iac/testing/tests/test_hermes_handoff_exec.py

"""Contracts for bounded, screened command execution in the handoff harness."""

from __future__ import annotations

import hashlib
import subprocess
import time

import pytest

from testing.tests.test_hermes_handoff_support import (
    FakeClock,
    FakeSpawn,
    load_handoff_module,
)

runner_module = load_handoff_module("hermes_handoff_exec")
policy = load_handoff_module("hermes_handoff_policy")


def completed(stdout: str = "", stderr: str = "", returncode: int = 0):
    return subprocess.CompletedProcess(
        args=[], returncode=returncode, stdout=stdout, stderr=stderr
    )


def build(results: list, **kwargs) -> tuple:
    clock = FakeClock()
    spawn = FakeSpawn(results)
    runner = runner_module.Runner(
        clock=clock,
        spawn=spawn,
        environ={},
        attestor=lambda command, _environment: runner_module.Attestation(
            command, "a" * 64
        ),
        **kwargs,
    )
    return (runner, spawn, clock)


OPERATOR = runner_module.operator_vantage()
SAFE_GET = ("kubectl", "get", "pods", "-o", "name")


def test_a_successful_command_is_captured_and_screened() -> None:
    runner, spawn, _ = build([completed(stdout="password=a-long-enough-secret\n")])

    outcome = runner.run(SAFE_GET, OPERATOR)

    assert outcome.ok and outcome.ran
    assert "[redacted]" in outcome.stdout
    assert outcome.returncode == 0
    assert spawn.calls[0]["argv"] == list(SAFE_GET)
    assert runner.commands_run == 1


def test_a_failing_command_is_not_ok_but_still_ran() -> None:
    runner, _, _ = build(
        [completed(stderr="Error from server (Forbidden)", returncode=1)]
    )

    outcome = runner.run(SAFE_GET, OPERATOR)

    assert outcome.ran and not outcome.ok
    assert "Forbidden" in outcome.combined


def test_a_policy_violation_never_reaches_a_subprocess() -> None:
    runner, spawn, _ = build([])

    outcome = runner.run(("kubectl", "delete", "pods", "--all"), OPERATOR)

    assert outcome.error and outcome.error.startswith("policy:")
    assert not outcome.ran
    assert spawn.calls == []
    assert runner.commands_run == 0


def test_a_timeout_is_recorded_rather_than_raised() -> None:
    runner, _, _ = build([subprocess.TimeoutExpired(cmd="kubectl", timeout=1.0)])

    outcome = runner.run(SAFE_GET, OPERATOR)

    assert outcome.error == runner_module.TIMEOUT_ERROR
    assert not outcome.ran


def test_a_missing_binary_is_recorded_rather_than_raised() -> None:
    runner, _, _ = build([FileNotFoundError(2, "No such file or directory")])

    outcome = runner.run(SAFE_GET, OPERATOR)

    assert outcome.error and outcome.error.startswith("spawn:")


def test_the_run_deadline_stops_further_probes_without_spawning() -> None:
    runner, spawn, clock = build([completed(stdout="first")], deadline_seconds=10.0)

    assert runner.run(SAFE_GET, OPERATOR).ok
    clock.advance(11.0)
    second = runner.run(("kubectl", "get", "nodes", "-o", "name"), OPERATOR)

    assert second.error == runner_module.DEADLINE_ERROR
    assert len(spawn.calls) == 1
    assert runner.remaining_seconds < 0


def test_the_command_timeout_is_clamped_by_the_remaining_deadline() -> None:
    runner, spawn, clock = build(
        [completed()], command_timeout=30.0, deadline_seconds=10.0
    )
    clock.advance(6.0)

    runner.run(SAFE_GET, OPERATOR)

    assert spawn.calls[0]["timeout"] == 4.0


def test_output_is_bounded_per_command_and_can_be_raised_per_call() -> None:
    runner, _, _ = build(
        [completed(stdout="y" * 200), completed(stdout="y" * 200)], max_bytes=50
    )

    small = runner.run(SAFE_GET, OPERATOR)
    large = runner.run(SAFE_GET, OPERATOR, max_bytes=4096)

    assert small.truncated
    assert not large.truncated


def test_the_child_environment_is_rebuilt_from_an_allowlist() -> None:
    source = {
        "PATH": "/bin",
        "KUBECONFIG": "/tmp/kc",
        "ANTHROPIC_API_KEY": "leak",
        "HOME": "/root",
    }
    environment = runner_module.build_environment(OPERATOR, source)

    assert environment["PATH"] == runner_module.SAFE_PATH
    assert environment["KUBECONFIG"] == "/tmp/kc"
    assert environment["GITEA_BASE_URL"] == "https://scm.bstein.dev"
    assert environment["GIT_CONFIG_GLOBAL"] == "/dev/null"
    assert "HOME" not in environment
    assert "ANTHROPIC_API_KEY" not in environment


def test_a_vantage_can_override_the_environment_it_needs() -> None:
    vantage = runner_module.operator_vantage(kubeconfig="/tmp/operator.yaml")
    environment = runner_module.build_environment(vantage, {"KUBECONFIG": "/tmp/other"})

    assert environment["KUBECONFIG"] == "/tmp/operator.yaml"
    assert "kubeconfig-pinned" in vantage.description


def test_the_operator_context_is_inserted_only_for_context_aware_tools() -> None:
    vantage = runner_module.operator_vantage(context="atlas-operator")

    assert vantage.wrap(("kubectl", "get", "pods"))[:3] == (
        "kubectl",
        "--context",
        "atlas-operator",
    )
    assert vantage.wrap(("git", "status")) == ("git", "status")
    assert vantage.wrap(()) == ()


def test_the_pod_vantage_execs_into_the_container_and_drops_the_context() -> None:
    operator = runner_module.operator_vantage(context="atlas-operator")
    vantage = runner_module.pod_vantage("hermes", "hermes-agent-1", "hermes", operator)

    wrapped = vantage.wrap(SAFE_GET)

    assert wrapped[:3] == ("kubectl", "--context", "atlas-operator")
    assert wrapped[3] == "exec"
    assert wrapped[-6:] == (
        "--",
        "/usr/local/bin/kubectl",
        "get",
        "pods",
        "-o",
        "name",
    )
    assert vantage.description == "in-pod hermes/hermes-agent-1[hermes]"
    policy.check_argv(wrapped)


def test_a_pod_vantage_without_an_operator_still_addresses_the_pod() -> None:
    vantage = runner_module.pod_vantage("hermes", "pod-a", "hermes")

    assert vantage.wrap(("kubectl", "version"))[0] == "kubectl"
    assert vantage.env == {}


def test_outcome_serialisation_screens_the_command_line() -> None:
    outcome = runner_module.Outcome(
        argv=("git", "clone", "https://user:a-long-enough-secret@scm.example.dev/x"),
        vantage="operator",
        returncode=0,
        stdout="ok",
    )

    payload = outcome.as_dict()

    assert "a-long-enough-secret" not in payload["command"]
    assert payload["vantage"] == "operator"
    assert payload["stdout"] == "ok"


def test_combined_output_joins_only_the_streams_that_carry_text() -> None:
    assert (
        runner_module.Outcome(argv=(), vantage="x", stdout="a", stderr="b").combined
        == "a\nb"
    )
    assert runner_module.Outcome(argv=(), vantage="x", stdout="a").combined == "a"
    assert runner_module.Outcome(argv=(), vantage="x").combined == ""


@pytest.mark.parametrize(
    "kwargs",
    [
        {"command_timeout": float("nan")},
        {"command_timeout": float("inf")},
        {"command_timeout": 0},
        {"deadline_seconds": -1},
        {"max_bytes": -1},
        {"max_bytes": 10**100},
    ],
)
def test_execution_bounds_reject_nonfinite_nonpositive_or_unbounded_inputs(
    kwargs,
) -> None:
    with pytest.raises(ValueError):
        runner_module.Runner(**kwargs)


def test_execution_bounds_reject_boolean_and_nonnumeric_inputs() -> None:
    for value in (True, "1", None):
        with pytest.raises(ValueError, match="numeric"):
            runner_module.Runner(command_timeout=value)


def test_every_reviewer_unsafe_argv_is_refused_without_spawning() -> None:
    runner, spawn, _ = build([])
    commands = [
        ("kubectl", "auth", "reconcile", "-f", "x"),
        ("kubectl", "config", "view", "--raw"),
        ("kubectl", "get", "secret/x", "-o", "name"),
        ("kubectl", "get", "--raw=/api/v1/secrets"),
        ("kubectl", "delete", "pod/x", "--dry-run=client"),
        ("kubectl", "delete", "pod/x", "--", "--dry-run=server"),
        ("git", "fetch", "origin"),
        ("git", "config", "core.sshCommand", "x"),
        ("git", "-ccore.sshCommand=x", "ls-remote", "origin"),
    ]
    for command in commands:
        assert runner.run(command, OPERATOR).error.startswith("policy:")
    assert spawn.calls == []
    assert runner.commands_run == 0


def test_live_capture_bounds_concurrent_stdout_and_stderr() -> None:
    capture = runner_module.execute_bounded(
        (
            "/usr/bin/python3",
            "-c",
            "import os;os.write(1,b'x'*2000000);os.write(2,b'y'*2000000)",
        ),
        {"LC_ALL": "C"},
        timeout=5,
        max_bytes=1024,
    )
    assert capture.returncode == 0
    assert capture.stdout_truncated and capture.stderr_truncated
    assert len(capture.stdout) == len(capture.stderr) == 1024


def test_absolute_timeout_kills_the_process_group_and_reaps_quickly() -> None:
    started = time.monotonic()
    capture = runner_module.execute_bounded(
        (
            "/usr/bin/python3",
            "-c",
            "import os,signal,time;"
            "p=os.fork();"
            "signal.signal(signal.SIGTERM,signal.SIG_IGN);"
            "time.sleep(30)",
        ),
        {"LC_ALL": "C"},
        timeout=0.15,
        max_bytes=1024,
    )
    assert capture.timed_out
    assert time.monotonic() - started < 1.5


def test_descendant_holding_pipes_is_bounded_after_the_leader_exits() -> None:
    started = time.monotonic()
    capture = runner_module.execute_bounded(
        (
            "/usr/bin/python3",
            "-c",
            "import os,signal,time;"
            "p=os.fork();"
            "(os._exit(0) if p else None);"
            "signal.signal(signal.SIGTERM,signal.SIG_IGN);"
            "time.sleep(30)",
        ),
        {"LC_ALL": "C"},
        timeout=2,
        max_bytes=1024,
    )
    assert not capture.timed_out
    assert time.monotonic() - started < 1.5


def test_the_scm_client_pin_is_derivable_from_the_merged_source() -> None:
    """The attested client digest must equal the ConfigMap-generated source.

    ``/opt/scm/gitea_api.py`` is mounted from ConfigMap ``hermes-scm-boundary-v2``,
    which is generated verbatim from ``services/hermes/scm-common/scripts/
    gitea_api.py``. Pinning anything else would make the harness unable to
    attest the client that actually ships in the agent pod.
    """
    from pathlib import Path

    client = policy.GITEA_CLIENT
    assert client == "/opt/scm/gitea_api.py"
    assert runner_module.EXPECTED_PATHS[client] == {client}
    assert runner_module.POD_COMMAND_PATHS[client] == client
    source = (
        Path(__file__).resolve().parents[2]
        / "services/hermes/scm-common/scripts/gitea_api.py"
    )
    digest = hashlib.sha256(source.read_bytes()).hexdigest()
    assert runner_module.EXPECTED_SHA256[client] == {digest}


def test_no_phantom_askpass_or_coordinator_path_is_configured() -> None:
    assert "GIT_ASKPASS" not in runner_module.FIXED_ENVIRONMENT
    assert "/opt/coordinator" not in runner_module.SAFE_PATH
    assert "/opt/scm" in runner_module.SAFE_PATH.split(":")
    environment = runner_module.build_environment(OPERATOR, {})
    assert "GIT_ASKPASS" not in environment


def test_attestation_rejects_a_reviewer_authored_executable(tmp_path) -> None:
    fake = tmp_path / "kubectl"
    fake.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
    fake.chmod(0o755)
    with pytest.raises(OSError, match="trusted paths"):
        runner_module.attest_executable(str(fake), {"PATH": str(tmp_path)})


def test_attestation_checks_presence_mode_digest_and_success(
    tmp_path, monkeypatch
) -> None:
    fake = tmp_path / "tool"
    fake.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
    fake.chmod(0o755)
    digest = hashlib.sha256(fake.read_bytes()).hexdigest()
    monkeypatch.setitem(runner_module.EXPECTED_PATHS, "tool", {str(fake)})
    monkeypatch.setitem(runner_module.EXPECTED_SHA256, "tool", {digest})
    assert (
        runner_module.attest_executable("tool", {"PATH": str(tmp_path)}).sha256
        == digest
    )
    with pytest.raises(OSError, match="unavailable"):
        runner_module.attest_executable("missing", {"PATH": str(tmp_path)})
    fake.chmod(0o777)
    with pytest.raises(OSError, match="unsafe ownership mode"):
        runner_module.attest_executable("tool", {"PATH": str(tmp_path)})
    fake.chmod(0o755)
    monkeypatch.setitem(runner_module.EXPECTED_SHA256, "tool", {"0" * 64})
    with pytest.raises(OSError, match="digest"):
        runner_module.attest_executable("tool", {"PATH": str(tmp_path)})


def test_runner_attestation_or_resolved_policy_failure_never_spawns() -> None:
    runner, spawn, _ = build([])
    runner._attestor = lambda _command, _environment: (_ for _ in ()).throw(
        OSError("no")
    )
    assert runner.run(SAFE_GET, OPERATOR).error.startswith("attestation:")
    runner._attestor = lambda _command, _environment: runner_module.Attestation(
        "/tmp/reviewer-authored/kubectl", "a" * 64
    )
    assert runner.run(SAFE_GET, OPERATOR).error.startswith("policy:")
    assert spawn.calls == []