atlas-iac/scripts/ops/hermes_handoff_checks_delivery.py

#!/usr/bin/env python3
"""Acceptance checks for forge authority, node accounts, images, and regressions.

What the platform is allowed to do outside its own namespace: which forge
authority a worker holds, what a node account is and is not, how an image is
built and pinned, and whether the fixes the release depends on are actually
running rather than merely merged.

Where a property has no side-effect-free live probe — a merge that must never be
attempted, a branch protection the harness identity cannot read — the entry says
so in its rationale and asserts the nearest fact that *is* observable, rather
than quietly asserting nothing.
"""

from __future__ import annotations

from hermes_handoff_catalog import (
    FORGE_CREDENTIAL_NAMES,
    INIT_NAME_PROJECTION,
    NODE_NAME_PROJECTION,
    OPERATOR,
    SELF,
    Targets,
    check,
    step,
)
from hermes_handoff_model import ATTEMPT, CheckSpec
from hermes_handoff_policy import GITEA_CLIENT, projection, shell


def _forge(targets: Targets) -> list[CheckSpec]:
    api = f"/api/v1/repos/{targets.repo}"
    return [
        check(
            "forge.worker-holds-no-git-credential-variable",
            "No forge credential is exported into the worker process",
            "forge",
            "names_absent",
            [step("env", SELF, *shell("env_names"))],
            {"step": "env", "names": FORGE_CREDENTIAL_NAMES},
            rationale="Authentication reaches Git through an askpass helper that reads runtime state, never through the environment.",
        ),
        check(
            "forge.identity-is-not-an-administrator",
            "The forge identity the worker uses holds no administrative rights",
            "forge",
            "json_field",
            [step("user", SELF, GITEA_CLIENT, "GET", "/api/v1/user")],
            {"step": "user", "fields": {"is_admin": False}},
        ),
        check(
            "forge.repository-authority-is-push-only",
            "The worker can push a branch but cannot administer the repository",
            "forge",
            "json_field",
            [step("repo", SELF, GITEA_CLIENT, "GET", api)],
            {
                "step": "repo",
                "fields": {
                    "permissions.admin": False,
                    "permissions.push": True,
                    "permissions.pull": True,
                },
            },
            rationale="Merge, approve, close, and protection changes all require authority this identity is proven not to hold.",
        ),
        check(
            "forge.administrative-route-is-refused",
            "An administrative forge route is refused for this identity",
            "forge",
            "denied",
            [
                step(
                    "attempt",
                    SELF,
                    GITEA_CLIENT,
                    "GET",
                    f"{api}/branch_protections",
                    kind=ATTEMPT,
                )
            ],
            rationale="A live refusal, not a configuration reading. The same lack of authority is what blocks a merge route.",
        ),
    ]


def _nodes(targets: Targets) -> list[CheckSpec]:
    return [
        check(
            "nodes.hardening-covers-every-node",
            "The node account audit runs on every node in the cluster",
            "nodes",
            "distinct_count",
            [
                step(
                    "coverage",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    "pods",
                    "--selector",
                    f"app={targets.node_daemonset}",
                    "-o",
                    NODE_NAME_PROJECTION,
                )
            ],
            {
                "step": "coverage",
                "minimum": targets.node_count,
                "total_equals": targets.node_count,
            },
            rationale="A hardened fleet with one unenrolled node is an unhardened fleet.",
        ),
    ]


def _build(targets: Targets) -> list[CheckSpec]:
    pipeline = f"{targets.remote}/main:{targets.builder_pipeline}"
    return [
        check(
            "build.builder-service-account-is-tokenless",
            "The image builder service account mounts no API token",
            "build",
            "stdout_matches",
            [
                step(
                    "serviceaccount",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.builder_namespace,
                    "get",
                    "serviceaccount",
                    targets.builder_serviceaccount,
                    "-o",
                    projection("jsonpath={.automountServiceAccountToken}"),
                )
            ],
            {"step": "serviceaccount", "equals": "false"},
        ),
        check(
            "build.builder-runs-unprivileged-with-exact-capabilities",
            "The build pod is unprivileged and adds only the expected capabilities",
            "build",
            "names_present",
            [step("pipeline", OPERATOR, "git", "show", pipeline, record=False)],
            {
                "step": "pipeline",
                "names": (
                    targets.builder_capabilities,
                    "privileged: false",
                    "allowPrivilegeEscalation: false",
                    "automountServiceAccountToken: false",
                    'drop: ["ALL"]',
                ),
            },
            rationale="Exact lines, not a substring sweep: a widened capability set has to change one of these lines to take effect.",
        ),
        check(
            "build.harbor-immutability-rule-is-applied",
            "The Harbor immutable-tag rule for the agent image has been applied",
            "build",
            "lines_match",
            [
                step(
                    "job",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.harbor_namespace,
                    "get",
                    "job",
                    targets.harbor_immutability_job,
                    "-o",
                    projection("jsonpath={.status.succeeded}"),
                )
            ],
            {"step": "job", "pattern": r"^[1-9][0-9]*$", "minimum": 1},
            rationale="The job verifies the rule it creates, so a completed run is evidence the rule matched the expected contract.",
        ),
        _release_lineage(targets),
    ]


def _release_lineage(targets: Targets) -> CheckSpec:
    """Bind the exact reviewed release from Git through the Ready pods."""
    container = targets.agent_container
    namespace = targets.namespace
    deployment = f"deploy/{targets.agent_deployment}"
    deployment_projection = projection(
        (
            "jsonpath={.metadata.generation}{'\\t'}{.status.observedGeneration}{'\\t'}"
            "{.spec.replicas}{'\\t'}{.status.readyReplicas}{'\\t'}"
            "{.metadata.annotations.deployment\\.kubernetes\\.io/revision}{'\\t'}"
            f'{{.spec.template.spec.containers[?(@.name=="{container}")].image}}'
        ).replace("'", '"')
    )
    pod_projection = projection(
        (
            "jsonpath={range .items[*]}{.metadata.name}{'\\t'}{.status.phase}{'\\t'}"
            "{range .status.conditions[?(@.type==\"Ready\")]}{.status}{end}{'\\t'}"
            f'{{.spec.containers[?(@.name=="{container}")].image}}{{"\\t"}}'
            f'{{.status.containerStatuses[?(@.name=="{container}")].imageID}}{{"\\t"}}'
            "{.metadata.labels.pod-template-hash}{'\\n'}{end}"
        ).replace("'", '"')
    )
    replicaset_projection = projection(
        (
            "jsonpath={range .items[*]}{.metadata.name}{'\\t'}"
            "{.metadata.annotations.deployment\\.kubernetes\\.io/revision}{'\\t'}"
            "{.status.readyReplicas}{'\\t'}{.status.availableReplicas}{'\\t'}"
            f'{{.spec.template.spec.containers[?(@.name=="{container}")].image}}{{"\\t"}}'
            "{.metadata.labels.pod-template-hash}{'\\n'}{end}"
        ).replace("'", '"')
    )
    flux_projection = projection(
        (
            "jsonpath={.metadata.name}{'\\t'}{.metadata.generation}{'\\t'}"
            "{.status.observedGeneration}{'\\t'}{.spec.suspend}{'\\t'}"
            "{range .status.conditions[?(@.type==\"Ready\")]}{.status}{end}{'\\t'}"
            "{.status.lastAppliedRevision}"
        ).replace("'", '"')
    )
    return check(
        "release.exact-lineage-is-running",
        "Remote main, reviewed PR, image, build, Flux, and running revision are identical",
        "release",
        "release_lineage",
        [
            step(
                "remote-main",
                OPERATOR,
                "git",
                "ls-remote",
                targets.remote,
                "refs/heads/main",
            ),
            step(
                "reviewed-pr",
                OPERATOR,
                GITEA_CLIENT,
                "GET",
                f"/api/v1/repos/{targets.repo}/pulls/{targets.reviewed_pr_number}",
            ),
            step(
                "build-source",
                OPERATOR,
                "git",
                "rev-parse",
                f"{targets.remote}/main",
            ),
            step(
                "deployment",
                OPERATOR,
                "kubectl",
                "--namespace",
                namespace,
                "get",
                deployment,
                "-o",
                deployment_projection,
            ),
            step(
                "pods",
                OPERATOR,
                "kubectl",
                "--namespace",
                namespace,
                "get",
                "pods",
                "--selector",
                f"app={targets.agent_deployment}",
                "-o",
                pod_projection,
            ),
            step(
                "replicasets",
                OPERATOR,
                "kubectl",
                "--namespace",
                namespace,
                "get",
                "replicasets",
                "--selector",
                f"app={targets.agent_deployment}",
                "-o",
                replicaset_projection,
            ),
            step(
                "flux",
                OPERATOR,
                "kubectl",
                "--namespace",
                "flux-system",
                "get",
                "kustomization/hermes",
                "-o",
                flux_projection,
            ),
        ],
        {
            "main_sha": targets.remote_main_sha,
            "head_sha": targets.reviewed_head_sha,
            "head_ref": targets.reviewed_head_ref,
            "pr_number": targets.reviewed_pr_number,
            "image": targets.agent_image,
            "build_sha": targets.build_sha,
            "deployment_revision": targets.deployment_revision,
        },
        rationale="A PASS requires one exact release identity across every source and running object.",
    )


def _reliability(targets: Targets) -> list[CheckSpec]:
    return [
        check(
            "reliability.finalization-and-replay-patches-are-deployed",
            "The agent workload runs the finalization, replay, and session patches",
            "reliability",
            "names_present",
            [
                step(
                    "init",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"deploy/{targets.agent_deployment}",
                    "-o",
                    INIT_NAME_PROJECTION,
                )
            ],
            {"step": "init", "names": targets.agent_init_containers},
            rationale="The regression suites live in the repository; this asserts the fixes they cover are actually running.",
        ),
        check(
            "reliability.regression-suites-are-present-on-main",
            "The decomposition and worker-recovery regression suites are on main",
            "reliability",
            "names_present",
            [
                step(
                    "suites",
                    OPERATOR,
                    "git",
                    "ls-tree",
                    "--name-only",
                    f"{targets.remote}/main",
                    "testing/tests/",
                )
            ],
            {
                "step": "suites",
                "names": (
                    "testing/tests/test_hermes_cli_lanes.py",
                    "testing/tests/test_hermes_coordinator.py",
                    "testing/tests/test_hermes_worker_recovery.py",
                ),
            },
        ),
    ]


def delivery_checks(targets: Targets) -> list[CheckSpec]:
    """Return the forge, node, build, and reliability checks."""
    return [
        *_forge(targets),
        *_nodes(targets),
        *_build(targets),
        *_reliability(targets),
    ]