atlas-iac/testing/tests/test_hermes_handoff_policy.py

"""Contracts for the fail-closed command policy of the handoff harness."""

from __future__ import annotations

import pytest

from testing.tests.test_hermes_handoff_support import load_handoff_module

policy = load_handoff_module("hermes_handoff_policy")


def check(*argv: str, mode: str | None = None) -> None:
    policy.check_argv(argv, mode or policy.READ_ONLY)


@pytest.mark.parametrize(
    "argv",
    [
        ("kubectl", "--namespace", "hermes", "get", "pods", "-o", "name"),
        ("kubectl", "auth", "can-i", "get", "secrets", "--all-namespaces"),
        ("kubectl", "get", "nodes", "-o", "name"),
        (
            "kubectl",
            "get",
            "pods",
            "-o",
            policy.projection("jsonpath={.status.phase}"),
        ),
        ("git", "merge-base", "--is-ancestor", "a", "b"),
        ("git", "ls-remote", "origin", "refs/heads/main"),
        ("hermes", "kanban", "list", "--json"),
        ("hermes", "sessions", "list", "--source", "telegram"),
        ("hermes", "status"),
        (policy.GITEA_CLIENT, "GET", "/api/v1/user"),
    ],
)
def test_read_only_commands_are_permitted(argv: tuple[str, ...]) -> None:
    check(*argv)


@pytest.mark.parametrize(
    "argv",
    [
        (),
        ("vault", "read", "kv/x"),
        ("curl", "https://example.dev"),
        ("rm", "-rf", "/"),
        ("kubectl", "get", "pods", "--token", "abc"),
        ("kubectl", "get", "pods"),
        ("kubectl", "get", "pods", "--as=system:admin"),
        ("kubectl", "get", "secrets", "-o", "name"),
        ("kubectl", "get", "pods,secrets/x", "-o", "name"),
        ("kubectl", "describe", "pod/x"),
        ("kubectl", "config", "view", "--raw"),
        ("kubectl", "auth", "reconcile", "-f", "x"),
        ("kubectl", "delete", "pod/x", "--dry-run=client"),
        ("kubectl", "delete", "pod/x", "--", "--dry-run=server"),
        ("kubectl", "get", "pods", "-o=json"),
        ("kubectl", "get", "pods", "-o", "jsonpath={.spec.containers[*].env[*].value}"),
        ("kubectl", "port-forward", "svc/x", "80"),
        ("kubectl", "exec", "pod"),
        ("kubectl", "exec", "pod", "--"),
        ("kubectl",),
        ("flux", "get", "kustomizations", "--all-namespaces"),
        ("flux", "reconcile", "kustomization", "x"),
        ("flux",),
        ("helm", "list", "--all-namespaces"),
        ("helm", "upgrade", "x"),
        ("helm", "get", "values", "x"),
        ("git", "commit", "-m", "x"),
        ("git", "push", "origin", "HEAD:refs/heads/x"),
        ("git", "fetch", "origin"),
        ("git", "config", "core.sshCommand", "x"),
        ("git", "-ccore.sshCommand=x", "ls-remote", "origin"),
        ("git", "--git-dir=/tmp/x", "status"),
        ("hermes", "kanban", "complete", "t_1"),
        ("hermes",),
        (policy.GITEA_CLIENT, "POST", "/api/v1/x"),
        (policy.GITEA_CLIENT, "GET"),
        ("sh", "-c", "rm -rf /"),
        ("sh", "echo"),
        ("git", "log", "/etc/shadow"),
    ],
)
def test_unsafe_commands_are_refused(argv: tuple[str, ...]) -> None:
    with pytest.raises(policy.PolicyError):
        check(*argv)


def test_unknown_mode_is_refused() -> None:
    with pytest.raises(policy.PolicyError, match="unknown mode"):
        policy.check_argv(("kubectl", "get", "pods"), "whatever")


def test_dry_run_mutation_commands_are_never_constructible() -> None:
    with pytest.raises(policy.PolicyError, match="dry-run"):
        check("kubectl", "patch", "deploy/x", "--dry-run=server")


def test_a_resource_named_after_a_subcommand_cannot_smuggle_a_delete() -> None:
    """`-n get` must not make a delete look like a read."""
    with pytest.raises(policy.PolicyError):
        check("kubectl", "delete", "-n", "get", "configmap", "x")


@pytest.mark.parametrize("flag", ["--as", "--as-group", "--as-uid"])
@pytest.mark.parametrize("mode", ["read-only", "ephemeral-armed"])
def test_impersonation_is_refused_from_every_vantage_and_mode(
    flag: str, mode: str
) -> None:
    """No vantage may impersonate: not the operator, and not inside a pod.

    The inner command of ``kubectl exec`` used to be granted impersonation
    unconditionally, so a self/switchyard/node/chat step carrying ``--as`` was
    accepted. The inner argv is now re-checked under the same rule.
    """
    for argv in (
        ("kubectl", "get", "namespaces", "-o", "name", flag, "system:admin"),
        ("kubectl", "get", "ns", "-o", "name", f"{flag}=system:admin"),
    ):
        with pytest.raises(policy.PolicyError, match="forbidden argument"):
            check(*argv, mode=mode)
    inner = (
        "kubectl",
        "exec",
        "pod",
        "--",
        "/usr/local/bin/kubectl",
        "get",
        "ns",
        "-o",
        "name",
        flag,
        "system:admin",
    )
    with pytest.raises(policy.PolicyError, match="forbidden argument"):
        check(*inner, mode=mode)


def test_exec_validates_its_inner_command() -> None:
    script = policy.render_shell("env_names")
    policy._RENDERED.add(script)
    check(
        "kubectl", "exec", "--namespace", "hermes", "pod", "--", "/bin/sh", "-c", script
    )
    check(
        "kubectl",
        "exec",
        "pod",
        "--",
        "/usr/local/bin/kubectl",
        "get",
        "ns",
        "-o",
        "name",
    )
    with pytest.raises(policy.PolicyError):
        check("kubectl", "exec", "pod", "--", "bash", "-c", "id")


def test_the_repository_pin_has_an_exact_boundary_and_rejects_dot_segments() -> None:
    """A prefix match without a boundary let a look-alike repository through."""
    for path in (
        "/api/v1/repos/atlas/titan-iac-evil/pulls",
        "/api/v1/repos/atlas/titan-iacx",
        "/api/v1/repos/atlas/titan-iac../pulls",
    ):
        with pytest.raises(policy.PolicyError):
            check(policy.GITEA_CLIENT, "GET", path)
    for path in (
        "/api/v1/repos/atlas/titan-iac/../../user/tokens",
        "/api/v1/repos/atlas/titan-iac/%2e%2e/%2e%2e/user/tokens",
        "/api/v1/repos/atlas/titan-iac/%252e%252e/user/tokens",
        "/api/v1/repos/atlas/titan-iac/./pulls",
        "/api/v1/repos/atlas/titan-iac//pulls",
    ):
        with pytest.raises(policy.PolicyError, match="relative segments"):
            check(policy.GITEA_CLIENT, "GET", path)
    check(policy.GITEA_CLIENT, "GET", "/api/v1/repos/atlas/titan-iac")
    check(policy.GITEA_CLIENT, "GET", "/api/v1/repos/atlas/titan-iac/pulls?state=all")


def test_every_allowlisted_binary_is_release_attestable() -> None:
    """An allowlisted binary with no pinned digest is dead, misleading config."""
    exec_module = load_handoff_module("hermes_handoff_exec")
    assert set(exec_module.EXPECTED_SHA256) == set(policy.ALLOWED_BINARIES)
    assert set(exec_module.EXPECTED_PATHS) == set(policy.ALLOWED_BINARIES)
    assert set(exec_module.POD_COMMAND_PATHS) == set(policy.ALLOWED_BINARIES)
    assert set(exec_module.CONTEXT_AWARE_BINARIES) <= set(policy.ALLOWED_BINARIES)


def test_shell_templates_reject_credential_roots() -> None:
    with pytest.raises(policy.PolicyError, match="credential paths"):
        policy.shell("json_fields", path="/runtime-access/token", fields="state")
    with pytest.raises(policy.PolicyError, match="credential-bearing"):
        policy.shell("json_fields", path="/tmp/health.json", fields="state,token")


def test_armed_mode_opens_exactly_the_mutations_it_should() -> None:
    ref = "ephemeral/hermes-handoff-acceptance/acceptance1"
    policy.arm_ephemeral_policy(ref)
    policy.register_ephemeral_pull(9)
    check("git", "push", "origin", f"HEAD:refs/heads/{ref}", mode=policy.ARMED)
    root = "/api/v1/repos/atlas/titan-iac"
    check(
        policy.GITEA_CLIENT,
        "POST",
        f"{root}/pulls",
        "--field",
        "title=WIP: Hermes handoff acceptance ephemeral probe",
        "--field",
        f"head={ref}",
        "--field",
        "base=main",
        "--field",
        "body=Ephemeral acceptance probe. Closed and deleted by the harness.",
        mode=policy.ARMED,
    )
    check(
        policy.GITEA_CLIENT,
        "PATCH",
        f"{root}/pulls/9",
        "--field",
        "state=closed",
        mode=policy.ARMED,
    )
    check(policy.GITEA_CLIENT, "DELETE", f"{root}/branches/{ref}", mode=policy.ARMED)
    with pytest.raises(policy.PolicyError, match="may not force"):
        check(
            "git",
            "push",
            "--force",
            "origin",
            f"HEAD:refs/heads/{ref}",
            mode=policy.ARMED,
        )
    with pytest.raises(policy.PolicyError, match="may not force"):
        check("git", "push", "--all", "origin", mode=policy.ARMED)
    with pytest.raises(policy.PolicyError, match="not permitted"):
        check(policy.GITEA_CLIENT, "PUT", f"{root}/pulls/9", mode=policy.ARMED)


def test_arming_and_projection_registration_fail_closed() -> None:
    with pytest.raises(policy.PolicyError, match="exact ephemeral ref"):
        policy.arm_ephemeral_policy("main")
    policy._ARMED_REF = ""
    with pytest.raises(policy.PolicyError, match="registration is malformed"):
        policy.register_ephemeral_pull(9)
    policy.arm_ephemeral_policy("ephemeral/hermes-handoff-acceptance/acceptance1")

    for candidate in (
        "wide",
        "jsonpath={.status.token}",
        "jsonpath={.metadata.annotations}",
        "jsonpath={.metadata.labels.other}",
        "jsonpath={.spec.containers[*].env[*].value}",
    ):
        with pytest.raises(policy.PolicyError):
            policy.projection(candidate)

    with pytest.raises(policy.PolicyError):
        check("/usr/bin/awk", "x")
    with pytest.raises(policy.PolicyError):
        check("git", "--exec-path=/tmp", "status")


def test_every_frozen_template_renders_and_is_accepted() -> None:
    parameters = {
        "account": "hermes-agent",
        "askpass": "/opt/coordinator/gitea_askpass.sh",
        "fields": "state,model",
        "limit": "200",
        "path": "/host-etc/passwd",
        "url": "https://scm.example.dev/atlas/repo.git",
    }
    for name in policy.SHELL_TEMPLATES:
        required = policy.template_parameters(name)
        argv = policy.shell(name, **{key: parameters[key] for key in required})
        policy.check_argv(argv)
        assert argv[0] == "sh"


def test_template_rendering_rejects_unknown_names_and_unsafe_parameters() -> None:
    with pytest.raises(policy.PolicyError, match="unknown shell template"):
        policy.render_shell("nope")
    with pytest.raises(policy.PolicyError, match="expects parameters"):
        policy.render_shell("json_fields")
    with pytest.raises(policy.PolicyError, match="unsafe shell parameter"):
        policy.render_shell("json_fields", path="/tmp;rm", fields="state")
    with pytest.raises(policy.PolicyError, match="unsafe shell parameter"):
        policy.render_shell("json_fields", path="$(id)", fields="state")


def test_positionals_skip_flags_and_stop_at_the_boundary() -> None:
    argv = ("kubectl", "--namespace", "hermes", "get", "pods", "--", "ignored")
    assert list(policy.positionals(argv)) == ["get", "pods"]
    assert list(policy.positionals(("kubectl", "--all-namespaces", "get"))) == ["get"]


def test_a_subcommand_must_precede_its_resource_arguments() -> None:
    with pytest.raises(policy.PolicyError):
        check("kubectl", "pods", "get")


@pytest.mark.parametrize(
    "argv",
    [
        ("/usr/bin/../../tmp/kubectl", "get", "pods"),
        ("/tmp/reviewer/kubectl", "get", "pods"),
        ("kubectl", "get", "pods", "-o", "go-template=x"),
        ("kubectl", "get", "pods", "-o", "jsonpath={.status.token}"),
        ("kubectl", "get", "pods", "-o", "jsonpath={.metadata.annotations}"),
        ("kubectl", "get", "pods", "-o", "jsonpath={.spec.containers[*].env[*].value}"),
        ("git", "-C", "/tmp", "status"),
        ("git", "--work-tree=/tmp", "status"),
        (policy.GITEA_CLIENT, "GET", "not-an-api-path"),
        (policy.GITEA_CLIENT, "GET", "/api/v1/repos/atlas/other"),
    ],
)
def test_provenance_projection_and_repository_bypasses_are_rejected(argv) -> None:
    with pytest.raises(policy.PolicyError):
        check(*argv)


def test_armed_writes_are_pinned_to_exact_remote_ref_and_endpoints() -> None:
    ref = "ephemeral/hermes-handoff-acceptance/acceptance1"
    policy.arm_ephemeral_policy(ref)
    policy.register_ephemeral_pull(9)
    with pytest.raises(policy.PolicyError, match="pinned to origin"):
        check(
            "git",
            "push",
            "upstream",
            "HEAD:refs/heads/ephemeral/hermes-handoff-acceptance/acceptance1",
            mode=policy.ARMED,
        )
    with pytest.raises(policy.PolicyError, match="preflighted ephemeral ref"):
        check("git", "push", "origin", "HEAD:refs/heads/other", mode=policy.ARMED)
    with pytest.raises(policy.PolicyError):
        check(
            policy.GITEA_CLIENT,
            "DELETE",
            "/api/v1/repos/atlas/titan-iac",
            mode=policy.ARMED,
        )
    with pytest.raises(policy.PolicyError):
        check(
            policy.GITEA_CLIENT,
            "PATCH",
            "/api/v1/repos/atlas/titan-iac/pulls/8",
            "--field",
            "state=closed",
            mode=policy.ARMED,
        )
    with pytest.raises(policy.PolicyError):
        check(
            policy.GITEA_CLIENT,
            "POST",
            "/api/v1/repos/atlas/titan-iac/pulls",
            "--field",
            "title=WIP: wrong branch",
            "--field",
            "head=main",
            "--field",
            "base=main",
            mode=policy.ARMED,
        )