From d208a89d9d590bffb8269e6bfc27b45ceced0374 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Mon, 17 Aug 2026 16:24:37 +0000 Subject: [PATCH] fix(hermes): close the zero-evidence fail-open in absence checks evaluate_names_absent returned PASS when its step exited 0 with no output, so five mandatory checks - the ones asserting that provider API keys, forge credentials, a cluster-admin binding, and shared coordinator state are absent - could report a pass on no evidence and turn a NO_GO into a GO. Both name rules now resolve their step through one guard in _line_step, so zero observations are NOT_RUN. Regressions pin all five real catalog specs plus both reachable silence paths: a POSIX pipeline whose status comes from its last stage, and a drifted kubectl -o jsonpath. The pool claim projection emits one = line per template volume so a volume without a PVC still counts as an observation rather than reading as drift. Also closes the review's reachable hardening and evidence defects: - pin Gitea paths to atlas/titan-iac on an exact segment boundary and reject relative segments, including percent-encoded ones - forbid impersonation structurally in every mode and vantage; the inner command of kubectl exec is re-checked rather than exempted, and validate_catalog no longer guards only the operator vantage - drop flux and helm from the binary allowlist; they had no pinned release digest, so no allowlisted binary can now be admitted that the executor would refuse to attest - remove the inert --concurrency and --expect-telegram-sessions flags and the dead concurrency bound; Telegram continuity stays mandatory - read the ephemeral pull index page by page, treat the create response as an authoritative source for the pull number, close every number either source names, and surface residue_ref plus exact manual_cleanup commands when creation is uncertain - keep executable_path and executable_sha256 on unrecorded bulk-evidence steps so withholding bytes never withholds binary attestation - revert the repo-wide hygiene legacy-exception mechanism; the contract change here is purely additive and the four pre-existing over-cap files are left to the canonical contract change in PR #14/#15 - correct the runbook ruff format scope so the documented command passes Split hermes_handoff_arming.py out of hermes_handoff_ephemeral.py to keep both modules under the 500-line cap. All 16 handoff modules hold at least 95% line and branch coverage; the mutation gate is 13/13. Co-Authored-By: Claude Opus 5 --- docs/hermes_full_handoff_acceptance.md | 97 ++++- scripts/ops/hermes_handoff_acceptance.py | 11 - scripts/ops/hermes_handoff_arming.py | 154 ++++++++ scripts/ops/hermes_handoff_catalog.py | 1 - scripts/ops/hermes_handoff_checks_workers.py | 9 +- scripts/ops/hermes_handoff_ephemeral.py | 358 ++++++++---------- scripts/ops/hermes_handoff_evaluators.py | 20 +- scripts/ops/hermes_handoff_exec.py | 19 +- scripts/ops/hermes_handoff_policy.py | 85 ++--- scripts/ops/hermes_handoff_run.py | 9 +- testing/quality_contract.json | 25 +- testing/quality_handoff_mutation.py | 52 +++ testing/quality_hygiene.py | 25 +- .../tests/test_hermes_handoff_acceptance.py | 4 +- .../tests/test_hermes_handoff_ephemeral.py | 141 ++++++- testing/tests/test_hermes_handoff_exec.py | 3 - .../test_hermes_handoff_mutation_gate.py | 13 +- testing/tests/test_hermes_handoff_policy.py | 76 +++- testing/tests/test_hermes_handoff_run.py | 27 +- .../test_hermes_handoff_zero_evidence.py | 243 ++++++++++++ testing/tests/test_quality_contract.py | 33 +- 21 files changed, 977 insertions(+), 428 deletions(-) create mode 100644 scripts/ops/hermes_handoff_arming.py create mode 100644 testing/tests/test_hermes_handoff_zero_evidence.py diff --git a/docs/hermes_full_handoff_acceptance.md b/docs/hermes_full_handoff_acceptance.md index be13c1f0..5533ef2d 100644 --- a/docs/hermes_full_handoff_acceptance.md +++ b/docs/hermes_full_handoff_acceptance.md @@ -58,13 +58,24 @@ Default mode permits a narrow read grammar: projections, `auth reconcile`, config operations, all mutations, and every dry-run mutation trick are rejected before spawn. - Git is read-only. `fetch`, configuration/helper overrides, alternate worktrees, - external diff helpers, and push are rejected. Helm values are not readable. + external diff helpers, and push are rejected. +- Gitea paths are pinned to `atlas/titan-iac` on an exact segment boundary, so a + look-alike repository such as `atlas/titan-iac-evil` is refused, and any + relative segment — including a percent-encoded `%2e%2e` — is rejected before + the request is built. - Shell commands must be rendered from frozen templates, use fixed executable paths, and cannot name service-account, Vault, runtime-access, SSH, or other credential roots. -- Outer impersonation is rejected. A reviewed inner command can use impersonation - only after an explicit `kubectl exec --` boundary; it cannot substitute for the - independent self identity. +- Impersonation is structurally impossible. `--as`, `--as-group`, and `--as-uid` + are refused for every argv, in every mode, from every vantage, and the inner + command of a `kubectl exec` is re-checked under the same rule rather than being + granted an exemption. `validate_catalog` refuses an impersonating step from any + vantage, not only the operator. A future audited self-probe that genuinely needs + impersonation is a reviewed code change, not a configuration flag. +- The binary allowlist is exactly the set with pinned release digests: + `kubectl`, `git`, `sh`, `hermes`, and the Gitea client. `flux` and `helm` are + not on it; Flux and Helm status is read through `kubectl` instead, so no + allowlisted binary can be admitted that the executor would refuse to attest. - `PATH` is replaced with a fixed path. Every outer executable is resolved to an allowed real path, checked for unsafe write mode, and compared with a pinned SHA-256 digest. A caller-authored `kubectl`, `git`, or helper cannot fabricate @@ -73,9 +84,14 @@ Default mode permits a narrow read grammar: The executor uses a new process group, drains stdout and stderr concurrently, and retains at most the configured bytes while the child is running. A monotonic absolute timeout terminates and reaps the group, including descendants that keep -pipes open. Timeout, deadline, output, and concurrency values reject booleans, -zero/negative values, non-finite numbers, and values outside fixed bounds. +pipes open. Timeout, deadline, and output-byte values reject booleans, +zero/negative values, non-finite numbers, and values outside fixed bounds. There +is no concurrency knob: `run_catalog` is strictly sequential and says so rather +than accepting a bound it would ignore. Truncation discards the affected stream and makes every evaluator `NOT_RUN`. +Bulk-evidence steps have their bytes replaced before they are recorded, but they +keep the resolved executable path and its attested SHA-256, so withholding output +never withholds the provenance of the binary that produced it. Credential screening runs before truncation. Command, output, error, identity, reason, evidence, JSON, and human-summary fields are screened recursively, @@ -99,6 +115,17 @@ forge administrative denial check performs a read-only GET and accepts only an explicit authorization refusal; generic `404`, `not found`, and `no route` text are not denial evidence. +An absence is only evidence when the probe observed something. Both name rules +resolve their step through one guard, so a step that exits `0` with no lines — or +with only separators — is `NOT_RUN`, never `PASS`. This is reachable in practice +two ways, and both are pinned by regressions: a POSIX pipeline reports the status +of its *last* stage, so a missing upstream binary in the frozen `env_names` +template exits `0` with no bytes; and `kubectl -o jsonpath` exits `0` when the +structure it walks has drifted, whether a parent field, a leaf field, or a +container filter. Because all five absence checks are mandatory, silence costs +the release its `GO`. The pool claim projection emits one `=` line +per template volume so a volume without a PVC still counts as an observation. + The strict evaluators reject duplicate-key JSON, raw control characters, `NaN`/infinity, boolean/number equality coercion, future timestamps, stale provider evidence, and every malformed routing or Flux row. Routing freshness is @@ -143,11 +170,22 @@ then pinned to that exact ref. It allows only its push, one exact draft-create payload, cleanup of the exactly discovered PR number, and deletion of that ref. The run reserves at least 300 seconds (up to 6000) for cleanup. It verifies the -pushed SHA, always discovers the PR after create (including ambiguous create -responses), and requires exactly one open draft with the expected base, head ref, -base SHA, and head SHA. Cleanup requires successful PR close, successful branch deletion, -a successful empty `ls-remote`, and a strict re-read of the same closed PR. -Malformed or partial cleanup is `FAIL`; it is never inferred from an error. +pushed SHA, then identifies the pull request from two independent sources: the +create response itself, and the pull index read page by page to its last page +(50 per page, 20 pages maximum) rather than a single truncated first page. Either +source alone names a pull request this run definitely made, because preflight +proved the ephemeral ref did not exist. Every number found from either source is +registered and closed, so a failed or unusable index can no longer leave an +orphan open. A `GO` still requires exactly one open draft with the expected base, +head ref, base SHA, and head SHA, and the two sources must agree. + +Cleanup requires successful PR close, successful branch deletion, a successful +empty `ls-remote`, and a strict re-read of the same closed PR. Malformed or +partial cleanup is `FAIL`; it is never inferred from an error. When anything is +uncertain the `draft-pull-request` evidence carries `residue_ref` and an exact +`manual_cleanup` command list — the `PATCH .../pulls/ --field state=closed` +calls for every number found, the `DELETE .../branches/` call, and, when no +number could be determined at all, the exact ref to sweep the index for by hand. Force, force-with-lease, fan-out, protected refs, other repositories, other PR numbers, arbitrary PATCH payloads, and repository administration remain denied. @@ -224,28 +262,51 @@ rollback. ## Quality and release evidence -The canonical quality contract explicitly manages, lints, and tracks all 15 +The canonical quality contract explicitly manages, lints, and tracks all 16 `scripts/ops/hermes_handoff_*.py` modules. It enforces at least 95% line and branch coverage per tracked file and the 500-physical-line source cap. Contract tests derive the module set so a future file cannot silently escape the gate. -Four pre-existing over-cap test files are temporarily locked to their exact -line count and SHA-256; any byte edit invalidates the exception and must split -the file below 500 lines. PR #14 supplies the intended additive test splits. +This change is purely additive to the contract: it registers its own modules and +changes no repo-wide hygiene semantics. + +The gate's `hygiene` step is red on `main` and stays red here. Exactly four +pre-existing test files exceed the 500-line cap — +`scripts/tests/test_dashboards_render_atlas.py` (516), +`testing/tests/test_hermes_chat_quality.py` (2242), +`testing/tests/test_hermes_cli_lanes.py` (1971), and +`testing/tests/test_hermes_coordinator.py` (510). All four are byte-identical at +merge-base `30259b52` and at current `origin/main`, none is touched by this +change, and the same four are the only issues a pristine `main` reports. Splitting +them, or grandfathering them, is a repo-wide contract decision that belongs to the +canonical contract change carried by PR #14/#15, not to this acceptance harness. +Every other gate step — docs, smell, unit, coverage — is green, and every file +this change adds or edits is under the cap. ```sh python3 -m testing.quality_gate --profile local --build-dir build python3 -m testing.quality_handoff_mutation python3 -m ruff check --select F,B,SIM,C4,UP --ignore B017,UP015,UP035 \ - scripts/ops/hermes_handoff_*.py testing/quality_*.py testing/tests/test_hermes_handoff_*.py + scripts/ops/hermes_handoff_*.py testing/quality_coverage.py \ + testing/quality_handoff_mutation.py testing/tests/test_hermes_handoff_*.py python3 -m ruff format --check \ - scripts/ops/hermes_handoff_*.py testing/quality_*.py testing/tests/test_hermes_handoff_*.py + scripts/ops/hermes_handoff_*.py testing/quality_coverage.py \ + testing/quality_handoff_mutation.py testing/tests/test_hermes_handoff_*.py python3 -m compileall -q scripts/ops testing kustomize build services/hermes | kubectl apply --dry-run=client --validate=false -f - git diff --check origin/main...HEAD ``` +The format scope names the files this change owns. A wider `testing/quality_*.py` +glob picks up `testing/quality_gate.py`, which is unmodified here and not +format-clean, so the documented command would fail for a reason this change did +not cause. `ruff format` is evidence, not a gate step: the canonical gate runs +`ruff check` only. + The deterministic mutation gate must kill every policy, redaction, verdict, -numeric, truncation, and cleanup mutant. The adversarial suite includes large +numeric, truncation, provenance, and cleanup mutant, including one that restores +the zero-evidence pass, one that restores the bare-prefix repository pin, one +that stops re-checking a `kubectl exec` inner command, and one that drops binary +attestation from an unrecorded step. The adversarial suite includes large concurrent stdout/stderr, malformed JSON and bytes, zero state, partial failure, absolute timeout, descendant-held pipes, malformed cleanup, ambiguous creates, and no-subprocess/no-mutation spies. Render/client dry-run, exact conflict checks, diff --git a/scripts/ops/hermes_handoff_acceptance.py b/scripts/ops/hermes_handoff_acceptance.py index 6b35d49f..6dbc69d2 100755 --- a/scripts/ops/hermes_handoff_acceptance.py +++ b/scripts/ops/hermes_handoff_acceptance.py @@ -123,9 +123,6 @@ def build_parser() -> argparse.ArgumentParser: default=DEFAULT_MAX_BYTES, help="Per-capture byte budget.", ) - parser.add_argument( - "--concurrency", type=int, default=1, help="Validated scheduling bound." - ) parser.add_argument( "--baseline-commit", default=Targets.baseline_commit, @@ -190,12 +187,6 @@ def build_parser() -> argparse.ArgumentParser: default=0, help="Chat tenant ordinal that carries the Telegram identity.", ) - parser.add_argument( - "--expect-telegram-sessions", - action="store_true", - default=True, - help="Require the selected chat tenant to hold durable Telegram-sourced sessions.", - ) parser.add_argument( "--pool-worker-env", action="append", @@ -243,7 +234,6 @@ def targets_from_args(arguments: argparse.Namespace, now: dt.datetime) -> Target pool_replicas=arguments.pool_replicas, chat_config_revision=arguments.chat_config_revision, chat_ordinal=arguments.chat_ordinal, - expect_telegram_sessions=arguments.expect_telegram_sessions, pool_worker_env=tuple(arguments.pool_worker_env or defaults.pool_worker_env), max_evidence_age_seconds=arguments.max_evidence_age, now=now, @@ -329,7 +319,6 @@ def main(argv: Sequence[str] | None = None) -> int: command_timeout=arguments.command_timeout, deadline_seconds=arguments.deadline, max_bytes=arguments.max_output_bytes, - concurrency=arguments.concurrency, ) except ValueError as exc: report = unavailable_report( diff --git a/scripts/ops/hermes_handoff_arming.py b/scripts/ops/hermes_handoff_arming.py new file mode 100644 index 00000000..5aa596c4 --- /dev/null +++ b/scripts/ops/hermes_handoff_arming.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +"""Arming, provenance, and push-target rules for the armed ephemeral mode. + +Nothing here touches the network or spawns a process. It is the half of armed +mode that must be true before a single command exists: the confirmation phrase, +the fixed repository and base, the one ephemeral ref grammar, and a local +worktree whose HEAD and ``origin/main`` are byte-exact with the reviewed release. +""" + +from __future__ import annotations + +import configparser +import re +from dataclasses import dataclass +from pathlib import Path + +from hermes_handoff_policy import ( + EXPECTED_BASE, + EXPECTED_REMOTE, + EXPECTED_REPO, + arm_ephemeral_policy, +) + +CONFIRMATION = "ARM EPHEMERAL HERMES HANDOFF PUSH" +REF_PREFIX = "ephemeral/hermes-handoff-acceptance" +REF_RE = re.compile(rf"{REF_PREFIX}/[a-z0-9][a-z0-9-]{{7,63}}\Z") +TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9-]{7,63}\Z") +SHA_RE = re.compile(r"[0-9a-f]{40}\Z") +EXPECTED_HEAD_REF = "feature/hermes-full-handoff-acceptance" +EXPECTED_REMOTE_URL = "https://scm.bstein.dev/atlas/titan-iac.git" +PROTECTED_NAMES = frozenset( + { + "default", + "develop", + "head", + "main", + "master", + "prod", + "production", + "release", + "stable", + "trunk", + } +) + + +class ArmingError(ValueError): + """Raised when an ephemeral run is not safe to start.""" + + +@dataclass(frozen=True) +class ArmRequest: + """Every fixed and caller-confirmed input to one armed run.""" + + repo: str + remote: str + token: str + confirmation: str + base: str = EXPECTED_BASE + expected_head: str = "" + expected_base_sha: str = "" + + @property + def ref(self) -> str: + return f"{REF_PREFIX}/{self.token}" + + +def normalise_branch(name: str) -> str: + """Return a branch name stripped of ref decoration, for comparison.""" + return name.strip().removeprefix("refs/heads/").strip("/").lower() + + +def assert_push_target_allowed(ref: str) -> None: + """Raise unless ``ref`` is the one ephemeral branch grammar.""" + if normalise_branch(ref) in PROTECTED_NAMES: + raise ArmingError(f"refusing to push to protected branch {ref!r}") + if not REF_RE.fullmatch(ref): + raise ArmingError(f"push target {ref!r} is not an ephemeral acceptance ref") + + +def bounded_read(path: Path, limit: int = 1024 * 1024) -> str: + data = path.read_bytes() + if not data or len(data) > limit: + raise ArmingError(f"local Git metadata is empty or exceeds {limit} bytes") + return data.decode("utf-8", "strict") + + +def _attest_worktree(root: Path, expected_head: str, expected_base_sha: str) -> None: + """Bind cwd, common Git metadata, branch, remote, and exact local HEAD.""" + resolved = root.resolve(strict=True) + marker = resolved / ".git" + if not marker.is_file(): + raise ArmingError( + "armed mode must run at the root of the existing linked worktree" + ) + line = bounded_read(marker, 4096).strip() + if not line.startswith("gitdir: "): + raise ArmingError("linked-worktree Git metadata is malformed") + gitdir = Path(line.removeprefix("gitdir: ")).resolve(strict=True) + common = (gitdir / bounded_read(gitdir / "commondir", 4096).strip()).resolve( + strict=True + ) + head = bounded_read(gitdir / "HEAD", 4096).strip() + expected_ref = f"refs/heads/{EXPECTED_HEAD_REF}" + if head != f"ref: {expected_ref}": + raise ArmingError("armed mode must run from the existing PR #19 feature branch") + ref_path = common / expected_ref + actual_head = bounded_read(ref_path, 4096).strip() if ref_path.is_file() else "" + if actual_head != expected_head: + raise ArmingError( + "local worktree HEAD does not match the exact reviewed PR head" + ) + remote_ref = common / "refs" / "remotes" / EXPECTED_REMOTE / EXPECTED_BASE + remote_main = bounded_read(remote_ref, 4096).strip() if remote_ref.is_file() else "" + if remote_main != expected_base_sha: + raise ArmingError("local origin/main does not match the exact release base SHA") + parser = configparser.RawConfigParser(interpolation=None) + try: + parser.read_string(bounded_read(common / "config")) + remote_url = parser.get(f'remote "{EXPECTED_REMOTE}"', "url") + except (configparser.Error, KeyError) as exc: + raise ArmingError("local Git remote metadata is malformed") from exc + if remote_url != EXPECTED_REMOTE_URL: + raise ArmingError("origin is not the fixed Atlas titan-iac HTTPS repository") + + +def preflight( + request: ArmRequest, expected_repo: str, worktree: Path | None = None +) -> None: + """Validate arming and local provenance before any subprocess/network call.""" + if request.confirmation != CONFIRMATION: + raise ArmingError("confirmation phrase does not match") + if expected_repo != EXPECTED_REPO or request.repo != EXPECTED_REPO: + raise ArmingError( + "armed repository is not the fixed Atlas titan-iac repository" + ) + if ( + request.remote != EXPECTED_REMOTE + or normalise_branch(request.base) != EXPECTED_BASE + ): + raise ArmingError("armed remote and pull-request base are fixed to origin/main") + if not TOKEN_RE.fullmatch(request.token): + raise ArmingError( + "ephemeral token must be 8-64 lowercase alphanumeric or dash characters" + ) + if not SHA_RE.fullmatch(request.expected_head) or not SHA_RE.fullmatch( + request.expected_base_sha + ): + raise ArmingError("exact reviewed head and release base SHAs are required") + assert_push_target_allowed(request.ref) + _attest_worktree( + worktree or Path.cwd(), request.expected_head, request.expected_base_sha + ) + arm_ephemeral_policy(request.ref) diff --git a/scripts/ops/hermes_handoff_catalog.py b/scripts/ops/hermes_handoff_catalog.py index 9dc5c98a..59334dec 100644 --- a/scripts/ops/hermes_handoff_catalog.py +++ b/scripts/ops/hermes_handoff_catalog.py @@ -152,7 +152,6 @@ class Targets: routing_tail_bytes: int = 512 * 1024 listing_bytes: int = 512 * 1024 git_askpass: str = "/opt/coordinator/gitea_askpass.sh" - expect_telegram_sessions: bool = True now: Any = None extra: dict[str, Any] = field(default_factory=dict) diff --git a/scripts/ops/hermes_handoff_checks_workers.py b/scripts/ops/hermes_handoff_checks_workers.py index eba7df6d..d62c719c 100644 --- a/scripts/ops/hermes_handoff_checks_workers.py +++ b/scripts/ops/hermes_handoff_checks_workers.py @@ -102,12 +102,15 @@ def _pool(targets: Targets) -> list[CheckSpec]: f"statefulset/{targets.pool_statefulset}", "-o", projection( - 'jsonpath={range .spec.template.spec.volumes[*]}{.persistentVolumeClaim.claimName}{"\\n"}{end}' + "jsonpath={range .spec.template.spec.volumes[*]}{.name}{'='}" + "{.persistentVolumeClaim.claimName}{'\\n'}{end}".replace( + "'", '"' + ) ), ) ], - {"step": "claims", "names": targets.coordinator_claims}, - rationale="Two writers on one SQLite state file is the duplicate-work and lost-result failure this pool exists to avoid.", + {"step": "claims", "contains": targets.coordinator_claims}, + rationale="Two writers on one SQLite state file is the duplicate-work and lost-result failure this pool exists to avoid. One line per template volume keeps a drifted or empty projection NOT_RUN instead of a silent pass.", ), *_pool_env(targets), ] diff --git a/scripts/ops/hermes_handoff_ephemeral.py b/scripts/ops/hermes_handoff_ephemeral.py index fdcfdbad..ca1ea594 100644 --- a/scripts/ops/hermes_handoff_ephemeral.py +++ b/scripts/ops/hermes_handoff_ephemeral.py @@ -1,161 +1,57 @@ #!/usr/bin/env python3 -"""Strictly preflighted, discoverable, and self-cleaning armed mode.""" +"""Discoverable, residue-surfacing, self-cleaning armed ephemeral run. + +The create response — not the index — is the authoritative name of a pull +request this harness made, and the index is read to its last page. When either +source is uncertain the run fails closed and reports the exact residue plus the +exact commands that remove it, so an operator is never left guessing what was +created. + +Arming, provenance, and push-target rules live in :mod:`hermes_handoff_arming` +and are re-exported here for callers that treat armed mode as one surface. +""" from __future__ import annotations -import configparser -import re from collections.abc import Sequence -from dataclasses import dataclass -from pathlib import Path from typing import Any +from hermes_handoff_arming import ( + CONFIRMATION, + REF_PREFIX, + ArmingError, + ArmRequest, + assert_push_target_allowed, + bounded_read, + normalise_branch, + preflight, +) from hermes_handoff_exec import Outcome, Runner, Vantage from hermes_handoff_model import EPHEMERAL, FAIL, NOT_RUN, PASS, CheckResult, CheckSpec from hermes_handoff_policy import ( EXPECTED_BASE, - EXPECTED_REMOTE, - EXPECTED_REPO, GITEA_CLIENT, - arm_ephemeral_policy, register_ephemeral_pull, ) from hermes_handoff_rules import strict_json -CONFIRMATION = "ARM EPHEMERAL HERMES HANDOFF PUSH" -REF_PREFIX = "ephemeral/hermes-handoff-acceptance" -REF_RE = re.compile(rf"{REF_PREFIX}/[a-z0-9][a-z0-9-]{{7,63}}\Z") -TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9-]{7,63}\Z") -SHA_RE = re.compile(r"[0-9a-f]{40}\Z") +__all__ = [ + "CONFIRMATION", + "DISCOVERY_PAGE_BUDGET", + "DISCOVERY_PAGE_SIZE", + "REF_PREFIX", + "ArmRequest", + "ArmingError", + "assert_push_target_allowed", + "bounded_read", + "normalise_branch", + "preflight", + "run_armed", +] + DRAFT_TITLE_PREFIX = "WIP: " -EXPECTED_HEAD_REF = "feature/hermes-full-handoff-acceptance" -EXPECTED_REMOTE_URL = "https://scm.bstein.dev/atlas/titan-iac.git" -PROTECTED_NAMES = frozenset( - { - "default", - "develop", - "head", - "main", - "master", - "prod", - "production", - "release", - "stable", - "trunk", - } -) - - -class ArmingError(ValueError): - """Raised when an ephemeral run is not safe to start.""" - - -@dataclass(frozen=True) -class ArmRequest: - """Every fixed and caller-confirmed input to one armed run.""" - - repo: str - remote: str - token: str - confirmation: str - base: str = EXPECTED_BASE - expected_head: str = "" - expected_base_sha: str = "" - - @property - def ref(self) -> str: - return f"{REF_PREFIX}/{self.token}" - - -def normalise_branch(name: str) -> str: - """Return a branch name stripped of ref decoration, for comparison.""" - return name.strip().removeprefix("refs/heads/").strip("/").lower() - - -def assert_push_target_allowed(ref: str) -> None: - """Raise unless ``ref`` is the one ephemeral branch grammar.""" - if normalise_branch(ref) in PROTECTED_NAMES: - raise ArmingError(f"refusing to push to protected branch {ref!r}") - if not REF_RE.fullmatch(ref): - raise ArmingError(f"push target {ref!r} is not an ephemeral acceptance ref") - - -def _bounded_read(path: Path, limit: int = 1024 * 1024) -> str: - data = path.read_bytes() - if not data or len(data) > limit: - raise ArmingError(f"local Git metadata is empty or exceeds {limit} bytes") - return data.decode("utf-8", "strict") - - -def _attest_worktree(root: Path, expected_head: str, expected_base_sha: str) -> None: - """Bind cwd, common Git metadata, branch, remote, and exact local HEAD.""" - resolved = root.resolve(strict=True) - marker = resolved / ".git" - if not marker.is_file(): - raise ArmingError( - "armed mode must run at the root of the existing linked worktree" - ) - line = _bounded_read(marker, 4096).strip() - if not line.startswith("gitdir: "): - raise ArmingError("linked-worktree Git metadata is malformed") - gitdir = Path(line.removeprefix("gitdir: ")).resolve(strict=True) - common = (gitdir / _bounded_read(gitdir / "commondir", 4096).strip()).resolve( - strict=True - ) - head = _bounded_read(gitdir / "HEAD", 4096).strip() - expected_ref = f"refs/heads/{EXPECTED_HEAD_REF}" - if head != f"ref: {expected_ref}": - raise ArmingError("armed mode must run from the existing PR #19 feature branch") - ref_path = common / expected_ref - actual_head = _bounded_read(ref_path, 4096).strip() if ref_path.is_file() else "" - if actual_head != expected_head: - raise ArmingError( - "local worktree HEAD does not match the exact reviewed PR head" - ) - remote_ref = common / "refs" / "remotes" / EXPECTED_REMOTE / EXPECTED_BASE - remote_main = ( - _bounded_read(remote_ref, 4096).strip() if remote_ref.is_file() else "" - ) - if remote_main != expected_base_sha: - raise ArmingError("local origin/main does not match the exact release base SHA") - parser = configparser.RawConfigParser(interpolation=None) - try: - parser.read_string(_bounded_read(common / "config")) - remote_url = parser.get(f'remote "{EXPECTED_REMOTE}"', "url") - except (configparser.Error, KeyError) as exc: - raise ArmingError("local Git remote metadata is malformed") from exc - if remote_url != EXPECTED_REMOTE_URL: - raise ArmingError("origin is not the fixed Atlas titan-iac HTTPS repository") - - -def preflight( - request: ArmRequest, expected_repo: str, worktree: Path | None = None -) -> None: - """Validate arming and local provenance before any subprocess/network call.""" - if request.confirmation != CONFIRMATION: - raise ArmingError("confirmation phrase does not match") - if expected_repo != EXPECTED_REPO or request.repo != EXPECTED_REPO: - raise ArmingError( - "armed repository is not the fixed Atlas titan-iac repository" - ) - if ( - request.remote != EXPECTED_REMOTE - or normalise_branch(request.base) != EXPECTED_BASE - ): - raise ArmingError("armed remote and pull-request base are fixed to origin/main") - if not TOKEN_RE.fullmatch(request.token): - raise ArmingError( - "ephemeral token must be 8-64 lowercase alphanumeric or dash characters" - ) - if not SHA_RE.fullmatch(request.expected_head) or not SHA_RE.fullmatch( - request.expected_base_sha - ): - raise ArmingError("exact reviewed head and release base SHAs are required") - assert_push_target_allowed(request.ref) - _attest_worktree( - worktree or Path.cwd(), request.expected_head, request.expected_base_sha - ) - arm_ephemeral_policy(request.ref) +DISCOVERY_PAGE_SIZE = 50 +DISCOVERY_PAGE_BUDGET = 20 def _spec(identifier: str, title: str) -> CheckSpec: @@ -246,17 +142,74 @@ def _pr_problems(payload: Any, request: ArmRequest) -> list[str]: return problems -def _discover(outcome: Outcome, request: ArmRequest) -> tuple[list[int], str]: +def _created_number(outcome: Outcome, request: ArmRequest) -> tuple[int, str]: + """Return the number the create response itself reports. + + The create response is the only source that names a pull request the harness + definitely made. Relying on the index alone means an index that fails or + answers ambiguously leaves an unnamed, unclosed pull request behind. + """ if not outcome.ok: - return ([], f"pull-request discovery failed: {_detail(outcome)}") + return (0, f"pull-request creation failed: {_detail(outcome)}") try: payload = strict_json(outcome.stdout) except ValueError as exc: - return ([], f"pull-request discovery returned malformed JSON: {exc}") - if not isinstance(payload, list): - return ([], "pull-request discovery did not return an array") + return (0, f"pull-request creation returned malformed JSON: {exc}") + problems = _pr_problems(payload, request) + if problems: + return (0, f"created pull request is invalid: {', '.join(problems)}") + return (payload["number"], "") + + +def _index_pages( + runner: Runner, vantage: Vantage, api: str, reserve: float +) -> tuple[list[Outcome], list[Any], str]: + """Read every page of the pull index; a truncated page can hide a match.""" + outcomes: list[Outcome] = [] + rows: list[Any] = [] + for page in range(1, DISCOVERY_PAGE_BUDGET + 1): + outcome = runner.run( + ( + GITEA_CLIENT, + "GET", + f"{api}/pulls?state=all&limit={DISCOVERY_PAGE_SIZE}&page={page}", + ), + vantage, + reserve_seconds=reserve, + ) + outcomes.append(outcome) + if not outcome.ok: + return (outcomes, [], f"discovery page {page} failed: {_detail(outcome)}") + try: + payload = strict_json(outcome.stdout) + except ValueError as exc: + return (outcomes, [], f"discovery page {page} is malformed JSON: {exc}") + if not isinstance(payload, list): + return (outcomes, [], f"discovery page {page} did not return an array") + rows.extend(payload) + if len(payload) < DISCOVERY_PAGE_SIZE: + return (outcomes, rows, "") + return (outcomes, [], "pull-request discovery exceeded its page budget") + + +def _residue(request: ArmRequest, numbers: Sequence[int], api: str) -> dict[str, Any]: + """Name the exact residue and the exact commands that remove it by hand.""" + manual = [ + f"{GITEA_CLIENT} PATCH {api}/pulls/{number} --field state=closed" + for number in numbers + ] + if not numbers: + manual.append( + f"{GITEA_CLIENT} GET {api}/pulls?state=all&limit={DISCOVERY_PAGE_SIZE}" + f" then close by hand every pull whose head ref is {request.ref}" + ) + manual.append(f"{GITEA_CLIENT} DELETE {api}/branches/{request.ref}") + return {"residue_ref": request.ref, "manual_cleanup": manual} + + +def _match_pulls(rows: Sequence[Any], request: ArmRequest) -> tuple[list[int], str]: matches: list[int] = [] - for index, item in enumerate(payload): + for index, item in enumerate(rows): if ( not isinstance(item, dict) or not isinstance(item.get("head"), dict) @@ -354,33 +307,34 @@ def _cleanup( ) +def _nothing_created(branch: CheckResult, draft_reason: str) -> list[CheckResult]: + """Report the two results that follow a run which created no artifact.""" + return [ + branch, + _result( + "draft-pull-request", + "A draft pull request can be opened and closed", + NOT_RUN, + draft_reason, + ), + _result( + "cleanup-verified", + "Every ephemeral artifact is removed", + NOT_RUN, + "nothing was created", + ), + ] + + def run_armed( runner: Runner, vantage: Vantage, request: ArmRequest ) -> list[CheckResult]: """Create, discover, validate, and clean one exact ephemeral branch/PR.""" guard = _protected_refusal_result() + branch_title = "A unique ephemeral feature branch can be pushed and removed" if guard.status != PASS: - return [ - guard, - _result( - "feature-branch-push", - "A unique ephemeral feature branch can be pushed and removed", - NOT_RUN, - "protected-ref guard failed", - ), - _result( - "draft-pull-request", - "A draft pull request can be opened and closed", - NOT_RUN, - "protected-ref guard failed", - ), - _result( - "cleanup-verified", - "Every ephemeral artifact is removed", - NOT_RUN, - "nothing was created", - ), - ] + unrun = _result("feature-branch-push", branch_title, NOT_RUN, "guard failed") + return [guard, *_nothing_created(unrun, "protected-ref guard failed")] api = f"/api/v1/repos/{request.repo}" reserve = min(6000.0, max(300.0, runner.command_timeout * 10)) existing = runner.run( @@ -392,28 +346,8 @@ def run_armed( reason = ( _detail(existing) if not existing.ok else "the ephemeral ref already exists" ) - return [ - guard, - _result( - "feature-branch-push", - "A unique ephemeral feature branch can be pushed and removed", - FAIL, - reason, - [existing], - ), - _result( - "draft-pull-request", - "A draft pull request can be opened and closed", - NOT_RUN, - "branch preflight failed", - ), - _result( - "cleanup-verified", - "Every ephemeral artifact is removed", - NOT_RUN, - "nothing was created", - ), - ] + refused = _result("feature-branch-push", branch_title, FAIL, reason, [existing]) + return [guard, *_nothing_created(refused, "branch preflight failed")] local_head = runner.run( ("git", "rev-parse", "HEAD"), vantage, reserve_seconds=reserve ) @@ -441,7 +375,7 @@ def run_armed( ] branch_result = _result( "feature-branch-push", - "A unique ephemeral feature branch can be pushed and removed", + branch_title, PASS if branch_ok else FAIL, "the exact reviewed HEAD was pushed" if branch_ok @@ -450,7 +384,7 @@ def run_armed( {"ref": request.ref, "head": request.expected_head}, ) numbers: list[int] = [] - discovery_error = "" + problems: list[str] = ["branch creation was not verified"] create_outcomes: list[Outcome] = [] if branch_ok: created = runner.run( @@ -470,26 +404,36 @@ def run_armed( vantage, reserve_seconds=reserve, ) - discovered = runner.run( - (GITEA_CLIENT, "GET", f"{api}/pulls?state=all&limit=50"), - vantage, - reserve_seconds=reserve, + number, create_error = _created_number(created, request) + pages, rows, index_error = _index_pages(runner, vantage, api, reserve) + found, match_error = ( + _match_pulls(rows, request) if not index_error else ([], "") ) - create_outcomes = [created, discovered] - numbers, discovery_error = _discover(discovered, request) - for number in numbers: - register_ephemeral_pull(number) - else: - discovery_error = "branch creation was not verified" + create_outcomes = [created, *pages] + # Either source alone names a pull request this run definitely made: the + # ephemeral ref was proven absent in preflight, so any pull on it is ours. + numbers = sorted({item for item in (number, *found) if item}) + problems = [text for text in (index_error, match_error) if text] + if number and found and found != [number]: + problems.append(f"created #{number} is not the discovered {found}") + if not numbers: + problems.append( + create_error or "no exact draft pull request was identified" + ) + for item in numbers: + register_ephemeral_pull(item) + exact = len(numbers) == 1 and not problems draft_result = _result( "draft-pull-request", "A draft pull request can be opened and closed against the ephemeral branch", - PASS if len(numbers) == 1 and not discovery_error else FAIL, + PASS if exact else FAIL, f"discovered exact draft pull request #{numbers[0]}" - if len(numbers) == 1 and not discovery_error - else discovery_error, + if exact + else "; ".join(problems), create_outcomes, - {"pull_requests": numbers}, + {"pull_requests": numbers} + if exact + else {"pull_requests": numbers, **_residue(request, numbers, api)}, ) - cleanup = _cleanup(runner, vantage, request, api, numbers, discovery_error) + cleanup = _cleanup(runner, vantage, request, api, numbers, "; ".join(problems)) return [guard, branch_result, draft_result, cleanup] diff --git a/scripts/ops/hermes_handoff_evaluators.py b/scripts/ops/hermes_handoff_evaluators.py index 61e9e06b..ea0a21b8 100644 --- a/scripts/ops/hermes_handoff_evaluators.py +++ b/scripts/ops/hermes_handoff_evaluators.py @@ -7,6 +7,9 @@ an authorization review alone — ``kubectl auth can-i`` reports what RBAC says, not what the API server does — so a real request must have been made and refused before it will pass. +An absence is only evidence when the probe observed something, so a name step +that exits 0 with no lines is ``NOT_RUN``: drift produces silence, not a pass. + Importing this module also registers the JSON-shaped rules, so a caller only has to import one place to have the full catalog vocabulary available. """ @@ -129,16 +132,20 @@ def _matched(observed: set[str], patterns: tuple[str, ...]) -> set[str]: def _line_step( spec: CheckSpec, outcomes: dict[str, Outcome] ) -> tuple[set[str], str] | Evaluation: + """Return a step's distinct lines, or the NOT_RUN that silence must produce.""" key = spec.expect["step"] outcome = outcomes[key] if not outcome.ok: return Evaluation(NOT_RUN, f"step {key} failed (rc={outcome.returncode})") - return (set(lines(outcome)), key) + observed = set(lines(outcome)) + if not observed: + return Evaluation(NOT_RUN, f"step {key} exited 0 with no lines to examine") + return (observed, key) @evaluator("names_absent") def evaluate_names_absent(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Evaluation: - """Pass when no forbidden name or substring appears in a step's lines.""" + """Pass when a step observed names and none of them is forbidden.""" blocked = missing_steps(spec, outcomes) if blocked: return blocked @@ -179,8 +186,6 @@ def evaluate_names_present(spec: CheckSpec, outcomes: dict[str, Outcome]) -> Eva return Evaluation( FAIL, f"required names absent: {', '.join(missing)}", evidence ) - if not observed: - return Evaluation(NOT_RUN, "step produced no lines to examine", evidence) return Evaluation(PASS, "every required name is present", evidence) @@ -317,11 +322,8 @@ def evaluate_vantages_agree( @evaluator("not_armed") def evaluate_not_armed(spec: CheckSpec, _outcomes: dict[str, Outcome]) -> Evaluation: """Report the fixed NOT_RUN a default, read-only run must produce.""" - return Evaluation( - NOT_RUN, - spec.expect.get("reason", "ephemeral mutation mode is not armed"), - {"armed": False}, - ) + reason = spec.expect.get("reason", "ephemeral mutation mode is not armed") + return Evaluation(NOT_RUN, reason, {"armed": False}) # This validator is one compact, linear state machine so every exit observes diff --git a/scripts/ops/hermes_handoff_exec.py b/scripts/ops/hermes_handoff_exec.py index 08888b13..d3edbe09 100644 --- a/scripts/ops/hermes_handoff_exec.py +++ b/scripts/ops/hermes_handoff_exec.py @@ -37,7 +37,6 @@ DEFAULT_CLEANUP_RESERVE = 90.0 MAX_COMMAND_TIMEOUT = 600.0 MAX_RUN_DEADLINE = 7200.0 MAX_OUTPUT_BYTES = 8 * 1024 * 1024 -MAX_CONCURRENCY = 16 READ_CHUNK = 64 * 1024 PIPE_DRAIN_SECONDS = 0.25 @@ -52,8 +51,6 @@ FIXED_ENVIRONMENT = {"GIT_ASKPASS": "/opt/coordinator/gitea_askpass.sh", "LC_ALL": "C", "PATH": SAFE_PATH} EXPECTED_PATHS = { "kubectl": {"/opt/data/tools/bin/kubectl", "/usr/local/bin/kubectl", "/usr/bin/kubectl"}, - "flux": {"/opt/data/tools/bin/flux", "/usr/local/bin/flux", "/usr/bin/flux"}, - "helm": {"/opt/data/tools/bin/helm", "/usr/local/bin/helm", "/usr/bin/helm"}, "git": {"/usr/bin/git", "/bin/git"}, "hermes": {"/opt/hermes/.venv/bin/hermes"}, "sh": {"/usr/bin/sh", "/bin/sh"}, @@ -71,7 +68,7 @@ DEADLINE_ERROR = "deadline-exceeded" TIMEOUT_ERROR = "timeout" OPERATOR = "operator" SELF = "self" -CONTEXT_AWARE_BINARIES = ("kubectl", "flux", "helm") +CONTEXT_AWARE_BINARIES = ("kubectl",) POD_COMMAND_PATHS = {"kubectl": "/usr/local/bin/kubectl", "hermes": "/opt/hermes/.venv/bin/hermes", "git": "/usr/bin/git", "sh": "/bin/sh", GITEA_CLIENT: GITEA_CLIENT} # fmt: on @@ -187,18 +184,6 @@ def _byte_budget(value: Any) -> int: return value -def _concurrency(value: Any) -> int: - if ( - isinstance(value, bool) - or not isinstance(value, int) - or not 1 <= value <= MAX_CONCURRENCY - ): - raise ValueError( - f"concurrency must be an integer between 1 and {MAX_CONCURRENCY}" - ) - return value - - def build_environment( vantage: Vantage, source: Mapping[str, str] | None = None ) -> dict[str, str]: @@ -408,7 +393,6 @@ class Runner: command_timeout: float = DEFAULT_COMMAND_TIMEOUT, deadline_seconds: float = DEFAULT_RUN_DEADLINE, max_bytes: int = DEFAULT_MAX_BYTES, - concurrency: int = 1, clock: Callable[[], float] = time.monotonic, executor: Callable[..., Capture] | None = None, spawn: Callable[..., subprocess.CompletedProcess[Any]] | None = None, @@ -423,7 +407,6 @@ class Runner: "deadline_seconds", deadline_seconds, 0.05, MAX_RUN_DEADLINE ) self.max_bytes = _byte_budget(max_bytes) - self.concurrency = _concurrency(concurrency) self._clock = clock self._executor = executor or ( _injected_executor(spawn) if spawn else execute_bounded diff --git a/scripts/ops/hermes_handoff_policy.py b/scripts/ops/hermes_handoff_policy.py index fc1a7097..11c8bc1c 100644 --- a/scripts/ops/hermes_handoff_policy.py +++ b/scripts/ops/hermes_handoff_policy.py @@ -3,12 +3,19 @@ Before spawn it admits only credential-safe, structurally read-only grammar; shell is limited to frozen templates with absolute tool paths. + +Impersonation is structurally impossible: ``--as`` and its siblings are refused +for every argv, in every mode, from every vantage, and the inner command of a +``kubectl exec`` is re-checked under the same rule rather than being trusted. +Only binaries with a pinned release digest are on the allowlist, so nothing can +be admitted here that the executor would then refuse to attest. """ from __future__ import annotations import re from collections.abc import Iterator +from urllib.parse import unquote READ_ONLY = "read-only" ARMED = "ephemeral-armed" @@ -27,9 +34,7 @@ IMPERSONATION_ARGS = ("--as", "--as-group", "--as-uid") FORBIDDEN_BINARIES = frozenset( {"age", "curl", "gpg", "openssl", "sops", "vault", "wget"} ) -ALLOWED_BINARIES = frozenset( - {"kubectl", "flux", "helm", "git", "hermes", "sh", GITEA_CLIENT} -) +ALLOWED_BINARIES = frozenset({"kubectl", "git", "hermes", "sh", GITEA_CLIENT}) TRUSTED_EXECUTABLE_ROOTS = ( "/bin/", "/usr/bin/", @@ -99,10 +104,6 @@ UNSAFE_PROJECTION_PARTS = ( SAFE_ENV_VALUE = 'env[?(@.name=="hermes_auto_router_profile")].value' SAFE_JSON_FIELDS = frozenset(["authenticated", "transport", "state", "model", "effort", "latency_ms", "checked_at", "api_provider", "auth_method", "subscription_type"]) # fmt: skip -FLUX_READ_SUBCOMMANDS = frozenset( - {"check", "events", "get", "stats", "trace", "tree", "version"} -) -HELM_READ_SUBCOMMANDS = frozenset({"history", "list", "status", "version"}) GIT_READ_SUBCOMMANDS = frozenset( { "cat-file", @@ -118,7 +119,6 @@ GIT_READ_SUBCOMMANDS = frozenset( "status", } ) -GIT_ARMED_SUBCOMMANDS = frozenset({"push"}) GIT_FORCE_FLAGS = frozenset( {"-f", "--force", "--force-with-lease", "--mirror", "--all", "--tags"} ) @@ -270,15 +270,16 @@ def _binary_name(value: str) -> str: return value.rsplit("/", 1)[-1] -def _reject_forbidden_arguments( - argv: tuple[str, ...], allow_impersonation: bool -) -> None: - permitted = IMPERSONATION_ARGS if allow_impersonation else () +def _reject_forbidden_arguments(argv: tuple[str, ...]) -> None: + """Refuse credential and impersonation arguments up to the ``--`` boundary. + + Everything after ``--`` is the inner command of a ``kubectl exec``, which + :func:`_check_kubectl` submits to :func:`check_argv` again, so it is held to + exactly this rule rather than being exempted from it. + """ boundary = argv.index("--") if "--" in argv else len(argv) for argument in argv[:boundary]: for forbidden in FORBIDDEN_ARGS: - if forbidden in permitted: - continue if argument == forbidden or argument.startswith(f"{forbidden}="): raise PolicyError(f"forbidden argument: {forbidden}") if argument == "--raw" or argument.startswith("--raw="): @@ -332,7 +333,7 @@ def _check_projection(output: str) -> None: ) -def _check_kubectl(argv: tuple[str, ...], mode: str, allow_impersonation: bool) -> None: +def _check_kubectl(argv: tuple[str, ...], mode: str) -> None: words = list(positionals(argv)) if not words or words[0] not in KUBECTL_SUBCOMMANDS: raise PolicyError("kubectl subcommand is outside the read-only boundary") @@ -351,7 +352,7 @@ def _check_kubectl(argv: tuple[str, ...], mode: str, allow_impersonation: bool) raise PolicyError( "kubectl exec inner command must use a fixed trusted path" ) - check_argv(inner, mode, allow_impersonation=True) + check_argv(inner, mode) return resources = words[1].split(",") if len(words) > 1 else [] if any(_resource_name(resource) in SECRET_RESOURCES for resource in resources): @@ -362,21 +363,6 @@ def _check_kubectl(argv: tuple[str, ...], mode: str, allow_impersonation: bool) _check_projection(output) -def _first_positional(binary: str, argv: tuple[str, ...]) -> str: - word = next(positionals(argv), None) - if word is None: - raise PolicyError(f"{binary} requires a subcommand") - return word - - -def _check_read_only_tool( - binary: str, argv: tuple[str, ...], allowed: frozenset[str] -) -> None: - subcommand = _first_positional(binary, argv) - if subcommand not in allowed: - raise PolicyError(f"{binary} {subcommand} is outside the read-only boundary") - - def _check_git(argv: tuple[str, ...], mode: str) -> None: unsafe = False for item in argv[1:]: @@ -390,7 +376,9 @@ def _check_git(argv: tuple[str, ...], mode: str) -> None: unsafe = True if unsafe: raise PolicyError("git configuration and helper overrides are forbidden") - subcommand = _first_positional("git", argv) + subcommand = next(positionals(argv), None) + if subcommand is None: + raise PolicyError("git requires a subcommand") if subcommand in GIT_READ_SUBCOMMANDS: return if subcommand == "push" and mode == ARMED: @@ -416,17 +404,34 @@ def _check_hermes(argv: tuple[str, ...]) -> None: ) +def _reject_dot_segments(path: str) -> None: + """Refuse any relative segment, including a percent-encoded one. + + The repository pin below is a textual prefix test, so a single ``..`` would + let a pinned path address another repository or the token endpoints once a + router normalised it. Decode first: ``%2e%2e`` is the same segment. + """ + decoded = unquote(unquote(path)).split("?", 1)[0] + if any(segment in {"", ".", ".."} for segment in decoded.split("/")[1:]): + raise PolicyError("gitea_api.py paths may not contain relative segments") + + def _check_gitea(argv: tuple[str, ...], mode: str) -> None: if len(argv) < 3: raise PolicyError("gitea_api.py requires a method and a path") method, path = argv[1].upper(), argv[2] if not path.startswith("/api/v1/") or not _SAFE_API_PATH_RE.fullmatch(path): raise PolicyError("gitea_api.py requires a bounded API path") - if path != "/api/v1/user" and not path.startswith(f"/api/v1/repos/{EXPECTED_REPO}"): + _reject_dot_segments(path) + repo_root = f"/api/v1/repos/{EXPECTED_REPO}" + if ( + path != "/api/v1/user" + and path != repo_root + and not path.startswith((f"{repo_root}/", f"{repo_root}?")) + ): raise PolicyError("gitea_api.py is pinned to the Atlas titan-iac repository") if method in GITEA_READ_METHODS: return - repo_root = f"/api/v1/repos/{EXPECTED_REPO}" if mode == ARMED: fields = [ argv[index + 1] for index, item in enumerate(argv[:-1]) if item == "--field" @@ -463,9 +468,7 @@ def _check_shell(argv: tuple[str, ...]) -> None: raise PolicyError("sh is only reachable as sh -c ") -def check_argv( - argv: tuple[str, ...], mode: str = READ_ONLY, allow_impersonation: bool = False -) -> None: +def check_argv(argv: tuple[str, ...], mode: str = READ_ONLY) -> None: """Raise :class:`PolicyError` unless ``argv`` is safe in ``mode``.""" if mode not in MODES: raise PolicyError(f"unknown mode: {mode}") @@ -476,16 +479,12 @@ def check_argv( raise PolicyError(f"forbidden binary: {binary}") if binary not in ALLOWED_BINARIES: raise PolicyError(f"binary is not on the harness allowlist: {binary}") - _reject_forbidden_arguments(argv, allow_impersonation) + _reject_forbidden_arguments(argv) _reject_secret_paths(argv) if binary == "sh": _check_shell(argv) elif binary == "kubectl": - _check_kubectl(argv, mode, allow_impersonation) - elif binary == "flux": - _check_read_only_tool("flux", argv, FLUX_READ_SUBCOMMANDS) - elif binary == "helm": - _check_read_only_tool("helm", argv, HELM_READ_SUBCOMMANDS) + _check_kubectl(argv, mode) elif binary == "git": _check_git(argv, mode) elif binary == "hermes": diff --git a/scripts/ops/hermes_handoff_run.py b/scripts/ops/hermes_handoff_run.py index a3b21cd7..45459dc8 100644 --- a/scripts/ops/hermes_handoff_run.py +++ b/scripts/ops/hermes_handoff_run.py @@ -97,9 +97,10 @@ def validate_catalog(specs: Sequence[CheckSpec]) -> list[str]: problems.append( f"{spec.id}: step {step.key} has unknown kind {step.kind!r}" ) - if step.vantage == OPERATOR and set(step.argv) & set(IMPERSONATION_ARGS): + if set(step.argv) & set(IMPERSONATION_ARGS): problems.append( - f"{spec.id}: step {step.key} impersonates from the operator vantage" + f"{spec.id}: step {step.key} impersonates from the " + f"{step.vantage} vantage" ) try: addressed = ( @@ -152,7 +153,7 @@ def validate_targets(targets: Targets) -> list[str]: problems.append("deployment revision must be a positive integer string") if targets.node_count != 3 or targets.pool_replicas != 3: problems.append("node and pool counts are fixed at exactly three") - if not targets.expect_telegram_sessions or not targets.chat_config_revision.strip(): + if not targets.chat_config_revision.strip(): problems.append( "Telegram continuity and its chat config revision are mandatory" ) @@ -386,6 +387,8 @@ def run_check( truncated=outcome.truncated, duration_ms=outcome.duration_ms, error=outcome.error, + executable_path=outcome.executable_path, + executable_sha256=outcome.executable_sha256, ) ) evaluation = evaluate(spec, outcomes) diff --git a/testing/quality_contract.json b/testing/quality_contract.json index 9c3de171..a14e045d 100644 --- a/testing/quality_contract.json +++ b/testing/quality_contract.json @@ -15,6 +15,7 @@ ], "managed_modules": [ "scripts/ops/hermes_handoff_acceptance.py", + "scripts/ops/hermes_handoff_arming.py", "scripts/ops/hermes_handoff_catalog.py", "scripts/ops/hermes_handoff_checks_access.py", "scripts/ops/hermes_handoff_checks_delivery.py", @@ -52,6 +53,7 @@ ], "lint_paths": [ "scripts/ops/hermes_handoff_acceptance.py", + "scripts/ops/hermes_handoff_arming.py", "scripts/ops/hermes_handoff_catalog.py", "scripts/ops/hermes_handoff_checks_access.py", "scripts/ops/hermes_handoff_checks_delivery.py", @@ -150,24 +152,6 @@ ], "hygiene": { "max_lines": 500, - "legacy_line_exceptions": { - "scripts/tests/test_dashboards_render_atlas.py": { - "lines": 516, - "sha256": "afe4923391b157d03bcf0899208f77eeab79a2762da3575a365530b3074c5f25" - }, - "testing/tests/test_hermes_chat_quality.py": { - "lines": 2242, - "sha256": "71e9cafd14460b5132d07dd951514756d091009c23db7703db1575b05e34e75f" - }, - "testing/tests/test_hermes_cli_lanes.py": { - "lines": 1971, - "sha256": "992bf3788c33252e37a695f5307c31b1e8c0646e00a715e3e55f6901788abb62" - }, - "testing/tests/test_hermes_coordinator.py": { - "lines": 510, - "sha256": "290240cf391abc755eaa0abb12d247ad1718e7c9295a2092f349a7eaba4bfe55" - } - }, "line_limit_globs": [ "scripts/ops/hermes_handoff_*.py", "testing/**/*.py", @@ -221,6 +205,7 @@ "minimum_branch_percent": 95.0, "branch_tracked_files": [ "scripts/ops/hermes_handoff_acceptance.py", + "scripts/ops/hermes_handoff_arming.py", "scripts/ops/hermes_handoff_catalog.py", "scripts/ops/hermes_handoff_checks_access.py", "scripts/ops/hermes_handoff_checks_delivery.py", @@ -237,11 +222,11 @@ "scripts/ops/hermes_handoff_run.py", "ci/scripts/semgrep_report.py", "testing/quality_coverage.py", - "testing/quality_handoff_mutation.py", - "testing/quality_hygiene.py" + "testing/quality_handoff_mutation.py" ], "tracked_files": [ "scripts/ops/hermes_handoff_acceptance.py", + "scripts/ops/hermes_handoff_arming.py", "scripts/ops/hermes_handoff_catalog.py", "scripts/ops/hermes_handoff_checks_access.py", "scripts/ops/hermes_handoff_checks_delivery.py", diff --git a/testing/quality_handoff_mutation.py b/testing/quality_handoff_mutation.py index a0c63fa0..95d2df42 100644 --- a/testing/quality_handoff_mutation.py +++ b/testing/quality_handoff_mutation.py @@ -138,6 +138,58 @@ MUTATIONS = ( "p={'number':1,'state':'open','draft':True,'merged':False,'base':{'ref':'main','sha':'2'*40},'head':{'ref':q.ref,'sha':'1'*40}}\n" "assert e._closed_pr_problems(p,q)\n", ), + Mutation( + "absence-check-passes-on-zero-evidence", + "hermes_handoff_evaluators.py", + " if not observed:\n" + ' return Evaluation(NOT_RUN, f"step {key} exited 0 with no lines to examine")\n', + " if False and not observed:\n return Evaluation(NOT_RUN, 'mutant')\n", + "import hermes_handoff_evaluators as v\n" + "from hermes_handoff_exec import Outcome\nfrom hermes_handoff_model import *\n" + "s=CheckSpec('x','x','x','names_absent',steps=(Step('e','self',('kubectl',)),)," + "expect={'step':'e','names':('GITEA_TOKEN',)})\n" + "assert v.evaluate(s,{'e':Outcome(('x',),'x',0,'')}).status == NOT_RUN\n", + ), + Mutation( + "repository-pin-is-a-bare-prefix", + "hermes_handoff_policy.py", + ' and not path.startswith((f"{repo_root}/", f"{repo_root}?"))\n', + " and not path.startswith(repo_root)\n", + "from hermes_handoff_policy import *\n" + "try: check_argv((GITEA_CLIENT,'GET','/api/v1/repos/atlas/titan-iac-evil/pulls'))\n" + "except PolicyError: raise SystemExit(0)\n" + "raise SystemExit(1)\n", + ), + Mutation( + "exec-inner-command-is-not-rechecked", + "hermes_handoff_policy.py", + " check_argv(inner, mode)\n return\n", + " return\n", + "from hermes_handoff_policy import *\n" + "argv=('kubectl','exec','pod','--','/usr/local/bin/kubectl','get','ns','-o','name'," + "'--as','system:admin')\n" + "try: check_argv(argv)\n" + "except PolicyError: raise SystemExit(0)\n" + "raise SystemExit(1)\n", + ), + Mutation( + "unrecorded-step-loses-its-binary-attestation", + "hermes_handoff_run.py", + " executable_path=outcome.executable_path,\n" + " executable_sha256=outcome.executable_sha256,\n", + "", + "import hermes_handoff_run as r\nfrom hermes_handoff_exec import Outcome\n" + "from hermes_handoff_model import *\n" + "class V:\n name='self'\n" + "class N:\n" + " def run(self,argv,_v,_b=None): return Outcome(argv,'self',0,'locked'," + "executable_path='/usr/bin/git',executable_sha256='a'*64)\n" + "s=CheckSpec('x','x','x','stdout_matches'," + "steps=(Step('s','self',('git','status'),record=False),)," + "expect={'step':'s','equals':'locked'})\n" + "o=r.run_check(N(),s,{'self':V()}).outcomes[0]\n" + "assert o.executable_sha256 == 'a'*64 and o.executable_path == '/usr/bin/git'\n", + ), ) diff --git a/testing/quality_hygiene.py b/testing/quality_hygiene.py index 6b68de3a..5060ca02 100644 --- a/testing/quality_hygiene.py +++ b/testing/quality_hygiene.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib import re from collections.abc import Iterable from pathlib import Path @@ -17,18 +16,6 @@ def _expand_globs(root: Path, patterns: Iterable[str]) -> list[Path]: return sorted(matched) -def _is_unchanged_legacy( - config: dict[str, Any], root: Path, path: Path, line_count: int -) -> bool: - """Allow an over-cap legacy file only while its exact bytes stay unchanged.""" - relative = path.relative_to(root).as_posix() - record = config.get("legacy_line_exceptions", {}).get(relative) - if not isinstance(record, dict) or record.get("lines") != line_count: - return False - digest = hashlib.sha256(path.read_bytes()).hexdigest() - return record.get("sha256") == digest - - def run_check(contract: dict[str, Any], root: Path) -> list[str]: """Return human-readable issues for naming and file-size rules.""" config = contract.get("hygiene", {}) @@ -37,12 +24,8 @@ def run_check(contract: dict[str, Any], root: Path) -> list[str]: for path in _expand_globs(root, config.get("line_limit_globs", [])): line_count = sum(1 for _ in path.open("r", encoding="utf-8")) - if line_count > max_lines and not _is_unchanged_legacy( - config, root, path, line_count - ): - issues.append( - f"file exceeds {max_lines} LOC: {path.relative_to(root)} ({line_count})" - ) + if line_count > max_lines: + issues.append(f"file exceeds {max_lines} LOC: {path.relative_to(root)} ({line_count})") for rule in config.get("naming_rules", []): pattern = re.compile(rule["pattern"]) @@ -65,9 +48,7 @@ def count_files_over_line_limit(contract: dict[str, Any], root: Path) -> int: count = 0 for path in _expand_globs(root, config.get("line_limit_globs", [])): line_count = sum(1 for _ in path.open("r", encoding="utf-8")) - if line_count > max_lines and not _is_unchanged_legacy( - config, root, path, line_count - ): + if line_count > max_lines: count += 1 return count diff --git a/testing/tests/test_hermes_handoff_acceptance.py b/testing/tests/test_hermes_handoff_acceptance.py index 4f89db35..de4a411c 100644 --- a/testing/tests/test_hermes_handoff_acceptance.py +++ b/testing/tests/test_hermes_handoff_acceptance.py @@ -124,7 +124,6 @@ def test_the_parser_exposes_the_documented_controls() -> None: "21", "--pool-worker-env", "HERMES_POOL_LEASE_SECONDS", - "--expect-telegram-sessions", ] ) @@ -134,7 +133,8 @@ def test_the_parser_exposes_the_documented_controls() -> None: assert targets.expected_suspensions == ("flux-system/parked",) assert targets.node_count == 21 assert targets.pool_worker_env == ("HERMES_POOL_LEASE_SECONDS",) - assert targets.expect_telegram_sessions is True + assert not hasattr(arguments, "concurrency") + assert not hasattr(arguments, "expect_telegram_sessions") def test_defaults_fall_back_to_the_shipped_targets() -> None: diff --git a/testing/tests/test_hermes_handoff_ephemeral.py b/testing/tests/test_hermes_handoff_ephemeral.py index 9f54601a..c1f993f8 100644 --- a/testing/tests/test_hermes_handoff_ephemeral.py +++ b/testing/tests/test_hermes_handoff_ephemeral.py @@ -15,6 +15,7 @@ from testing.tests.test_hermes_handoff_support import ( ) ephemeral = load_handoff_module("hermes_handoff_ephemeral") +arming = load_handoff_module("hermes_handoff_arming") exec_module = load_handoff_module("hermes_handoff_exec") policy = load_handoff_module("hermes_handoff_policy") model = load_handoff_module("hermes_handoff_model") @@ -84,16 +85,16 @@ def linked_worktree(tmp_path: Path) -> Path: (root / ".git").write_text(f"gitdir: {gitdir}\n", encoding="utf-8") (gitdir / "commondir").write_text("../..\n", encoding="utf-8") (gitdir / "HEAD").write_text( - f"ref: refs/heads/{ephemeral.EXPECTED_HEAD_REF}\n", encoding="utf-8" + f"ref: refs/heads/{arming.EXPECTED_HEAD_REF}\n", encoding="utf-8" ) - ref = common / "refs" / "heads" / ephemeral.EXPECTED_HEAD_REF + ref = common / "refs" / "heads" / arming.EXPECTED_HEAD_REF ref.parent.mkdir(parents=True) ref.write_text(f"{HEAD}\n", encoding="utf-8") remote_ref = common / "refs" / "remotes" / "origin" / "main" remote_ref.parent.mkdir(parents=True) remote_ref.write_text(f"{MAIN}\n", encoding="utf-8") (common / "config").write_text( - f'[remote "origin"]\n\turl = {ephemeral.EXPECTED_REMOTE_URL}\n', + f'[remote "origin"]\n\turl = {arming.EXPECTED_REMOTE_URL}\n', encoding="utf-8", ) return root @@ -152,7 +153,7 @@ def test_worktree_attestation_rejects_each_malformed_local_source( wrong_sha = linked_worktree(tmp_path / "sha") gitdir = Path((wrong_sha / ".git").read_text().split(": ", 1)[1].strip()) common = (gitdir / (gitdir / "commondir").read_text().strip()).resolve() - (common / "refs" / "heads" / ephemeral.EXPECTED_HEAD_REF).write_text("0" * 40) + (common / "refs" / "heads" / arming.EXPECTED_HEAD_REF).write_text("0" * 40) roots.append(wrong_sha) wrong_main = linked_worktree(tmp_path / "main") gitdir = Path((wrong_main / ".git").read_text().split(": ", 1)[1].strip()) @@ -175,7 +176,7 @@ def test_worktree_attestation_rejects_each_malformed_local_source( oversized = tmp_path / "oversized" oversized.write_text("x" * 5, encoding="utf-8") with pytest.raises(ephemeral.ArmingError): - ephemeral._bounded_read(oversized, 4) + ephemeral.bounded_read(oversized, 4) def test_guard_failure_stops_before_runner_calls(monkeypatch) -> None: @@ -253,26 +254,24 @@ def test_pull_request_parsers_reject_every_malformed_or_ambiguous_shape() -> Non assert {"number is malformed", "draft does not match", "base is malformed"} <= set( problems ) - assert ephemeral._discover(exec_module.Outcome((), "x", returncode=1), request())[1] - for body in ( - "not-json", - "{}", - "[null]", - json.dumps([{"head": {"ref": "other"}, "base": {}}]), - ): - numbers, error = ephemeral._discover( + assert ephemeral._created_number( + exec_module.Outcome((), "x", returncode=1), request() + ) == (0, "pull-request creation failed: exit status 1") + for body in ("not-json", "{}"): + number, error = ephemeral._created_number( exec_module.Outcome((), "x", 0, body), request() ) + assert number == 0 and error + assert ephemeral._created_number( + exec_module.Outcome((), "x", 0, json.dumps(pr_payload())), request() + ) == (42, "") + for rows in ([None], [{"head": {"ref": "other"}, "base": {}}]): + numbers, error = ephemeral._match_pulls(rows, request()) assert numbers == [] and error invalid = pr_payload() invalid["draft"] = False - assert ephemeral._discover( - exec_module.Outcome((), "x", 0, json.dumps([invalid])), request() - )[1] - assert ( - "discovered 0" - in ephemeral._discover(exec_module.Outcome((), "x", 0, "[]"), request())[1] - ) + assert ephemeral._match_pulls([invalid], request())[1] + assert "discovered 0" in ephemeral._match_pulls([], request())[1] @pytest.mark.parametrize( @@ -335,3 +334,105 @@ def test_read_only_runner_cannot_reach_push() -> None: results = by_id(ephemeral.run_armed(runner, VANTAGE, request())) assert results["ephemeral.feature-branch-push"].status == model.FAIL assert not any(call["argv"][1] == "push" for call in spawn.calls) + + +def push_prelude(ref_line: str) -> list: + """The four scripted results that get an armed run to a verified push.""" + return [ + completed(), + completed(stdout=f"{HEAD}\n"), + completed(), + completed(stdout=ref_line), + ] + + +def test_pull_request_discovery_reads_every_page() -> None: + """A full first page must not end discovery: the match can be on page two.""" + ref_line = f"{HEAD}\trefs/heads/{request().ref}\n" + filler = [ + { + "number": index, + "state": "open", + "draft": True, + "merged": False, + "base": {"ref": "main", "sha": MAIN}, + "head": {"ref": "feature/other", "sha": HEAD}, + } + for index in range(1, ephemeral.DISCOVERY_PAGE_SIZE + 1) + ] + runner, spawn = armed_runner( + [ + *push_prelude(ref_line), + completed(stdout=json.dumps(pr_payload())), + completed(stdout=json.dumps(filler)), + completed(stdout=json.dumps([pr_payload()])), + completed(stdout=json.dumps(pr_payload(state="closed"))), + completed(), + completed(), + completed(stdout=json.dumps(pr_payload(state="closed"))), + ] + ) + + results = by_id(ephemeral.run_armed(runner, VANTAGE, request())) + + assert results["ephemeral.draft-pull-request"].status == model.PASS + pages = [call["argv"][2] for call in spawn.calls if "pulls?state=all" in str(call)] + assert any("page=1" in page for page in pages) + assert any("page=2" in page for page in pages) + + +def test_uncertain_discovery_still_closes_the_created_pull_and_names_residue() -> None: + """The create response names the pull, so an unusable index cannot orphan it.""" + ref_line = f"{HEAD}\trefs/heads/{request().ref}\n" + runner, spawn = armed_runner( + [ + *push_prelude(ref_line), + completed(stdout=json.dumps(pr_payload())), + completed(returncode=1), + completed(stdout=json.dumps(pr_payload(state="closed"))), + completed(), + completed(), + completed(stdout=json.dumps(pr_payload(state="closed"))), + ] + ) + + results = by_id(ephemeral.run_armed(runner, VANTAGE, request())) + draft = results["ephemeral.draft-pull-request"] + + assert draft.status == model.FAIL + assert draft.evidence["pull_requests"] == [42] + assert draft.evidence["residue_ref"] == request().ref + assert any("pulls/42" in line for line in draft.evidence["manual_cleanup"]) + assert any("branches/" in line for line in draft.evidence["manual_cleanup"]) + # The pull the harness created is closed, not left behind. + assert any( + call["argv"][1] == "PATCH" and call["argv"][2].endswith("/pulls/42") + for call in spawn.calls + ) + assert results["ephemeral.cleanup-verified"].status == model.FAIL + + +def test_a_wholly_unidentifiable_create_names_the_ref_for_manual_cleanup() -> None: + """When neither source names a pull, the operator gets the exact ref to sweep.""" + ref_line = f"{HEAD}\trefs/heads/{request().ref}\n" + runner, _ = armed_runner( + [ + *push_prelude(ref_line), + completed(stdout="not-json"), + completed(returncode=1), + completed(), + completed(), + ] + ) + + draft = by_id(ephemeral.run_armed(runner, VANTAGE, request()))[ + "ephemeral.draft-pull-request" + ] + + assert draft.status == model.FAIL + assert draft.evidence["pull_requests"] == [] + assert draft.evidence["residue_ref"] == request().ref + assert any( + request().ref in line and "close by hand" in line + for line in draft.evidence["manual_cleanup"] + ) diff --git a/testing/tests/test_hermes_handoff_exec.py b/testing/tests/test_hermes_handoff_exec.py index c192448d..fb54d3a5 100644 --- a/testing/tests/test_hermes_handoff_exec.py +++ b/testing/tests/test_hermes_handoff_exec.py @@ -226,8 +226,6 @@ def test_combined_output_joins_only_the_streams_that_carry_text() -> None: {"deadline_seconds": -1}, {"max_bytes": -1}, {"max_bytes": 10**100}, - {"concurrency": 0}, - {"concurrency": 17}, ], ) def test_execution_bounds_reject_nonfinite_nonpositive_or_unbounded_inputs( @@ -255,7 +253,6 @@ def test_every_reviewer_unsafe_argv_is_refused_without_spawning() -> None: ("git", "fetch", "origin"), ("git", "config", "core.sshCommand", "x"), ("git", "-ccore.sshCommand=x", "ls-remote", "origin"), - ("helm", "get", "values", "x"), ] for command in commands: assert runner.run(command, OPERATOR).error.startswith("policy:") diff --git a/testing/tests/test_hermes_handoff_mutation_gate.py b/testing/tests/test_hermes_handoff_mutation_gate.py index 1e84cfb8..61c60705 100644 --- a/testing/tests/test_hermes_handoff_mutation_gate.py +++ b/testing/tests/test_hermes_handoff_mutation_gate.py @@ -4,7 +4,13 @@ from __future__ import annotations from pathlib import Path -from testing.quality_handoff_mutation import Mutation, MutationReport, main, run_gate +from testing.quality_handoff_mutation import ( + MUTATIONS, + Mutation, + MutationReport, + main, + run_gate, +) ROOT = Path(__file__).parents[2] @@ -38,8 +44,11 @@ def test_invalid_anchor_baseline_syntax_and_survivor_are_reported() -> None: def test_main_reports_a_perfect_score(capsys) -> None: + """Every mutant must die, and the count must track the declared set.""" assert main() == 0 - assert "9/9 (100.00%)" in capsys.readouterr().out + total = len(MUTATIONS) + assert total >= 13 + assert f"{total}/{total} (100.00%)" in capsys.readouterr().out def test_main_reports_mutation_issues(monkeypatch, capsys) -> None: diff --git a/testing/tests/test_hermes_handoff_policy.py b/testing/tests/test_hermes_handoff_policy.py index 90cef791..41f1fc7e 100644 --- a/testing/tests/test_hermes_handoff_policy.py +++ b/testing/tests/test_hermes_handoff_policy.py @@ -9,8 +9,8 @@ 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, inner: bool = False) -> None: - policy.check_argv(argv, mode or policy.READ_ONLY, allow_impersonation=inner) +def check(*argv: str, mode: str | None = None) -> None: + policy.check_argv(argv, mode or policy.READ_ONLY) @pytest.mark.parametrize( @@ -26,8 +26,6 @@ def check(*argv: str, mode: str | None = None, inner: bool = False) -> None: "-o", policy.projection("jsonpath={.status.phase}"), ), - ("flux", "get", "kustomizations", "--all-namespaces"), - ("helm", "list", "--all-namespaces"), ("git", "merge-base", "--is-ancestor", "a", "b"), ("git", "ls-remote", "origin", "refs/heads/main"), ("hermes", "kanban", "list", "--json"), @@ -63,8 +61,10 @@ def test_read_only_commands_are_permitted(argv: tuple[str, ...]) -> None: ("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"), @@ -103,22 +103,41 @@ def test_a_resource_named_after_a_subcommand_cannot_smuggle_a_delete() -> None: check("kubectl", "delete", "-n", "get", "configmap", "x") -def test_impersonation_is_operator_forbidden_and_in_pod_permitted() -> None: - with pytest.raises(policy.PolicyError, match="forbidden argument"): - check("kubectl", "get", "namespaces", "--as", "system:admin") - check( +@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", - "namespaces", + "ns", "-o", "name", - "--as", + flag, "system:admin", - inner=True, ) + with pytest.raises(policy.PolicyError, match="forbidden argument"): + check(*inner, mode=mode) -def test_exec_validates_its_inner_command_and_tolerates_impersonation_inside() -> None: +def test_exec_validates_its_inner_command() -> None: script = policy.render_shell("env_names") policy._RENDERED.add(script) check( @@ -134,13 +153,42 @@ def test_exec_validates_its_inner_command_and_tolerates_impersonation_inside() - "ns", "-o", "name", - "--as", - "system:admin", ) 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") diff --git a/testing/tests/test_hermes_handoff_run.py b/testing/tests/test_hermes_handoff_run.py index 881ffd55..4d87fb3e 100644 --- a/testing/tests/test_hermes_handoff_run.py +++ b/testing/tests/test_hermes_handoff_run.py @@ -135,7 +135,6 @@ def test_target_validation_rejects_every_vacuous_or_unbound_release_input() -> N deployment_revision="0", node_count=1, pool_replicas=1, - expect_telegram_sessions=False, chat_config_revision="", pool_worker_env=(), dependency_heads=((14, "bad"), (14, "bad")), @@ -335,6 +334,25 @@ def test_a_check_whose_vantage_is_absent_is_not_run() -> None: assert result.outcomes[0].error == "vantage node is unavailable" +def test_validate_catalog_rejects_impersonation_from_every_vantage() -> None: + """The operator vantage was the only one guarded; now none is exempt.""" + for vantage in (catalog.OPERATOR, catalog.SELF, catalog.CHAT): + spec = check( + "stdout_matches", + {"step": "s", "equals": "x"}, + steps=( + model.Step( + key="s", + vantage=vantage, + argv=("kubectl", "get", "ns", "-o", "name", "--as", "system:admin"), + ), + ), + ) + problems = harness_run.validate_catalog([spec]) + assert any("impersonates" in problem for problem in problems) + assert any(vantage in problem for problem in problems) + + def test_bulk_evidence_is_evaluated_but_replaced_before_it_is_recorded() -> None: spec = check( "stdout_matches", @@ -357,6 +375,13 @@ def test_bulk_evidence_is_evaluated_but_replaced_before_it_is_recorded() -> None assert result.status == model.PASS assert result.outcomes[0].stdout == harness_run.UNRECORDED assert result.outcomes[0].stderr == harness_run.UNRECORDED + # Withholding bulk bytes must not withhold the binary that produced them. + recorded = result.outcomes[0] + assert recorded.executable_path + assert recorded.executable_sha256 + assert recorded.executable_path == "kubectl" + assert recorded.executable_sha256 == "a" * 64 + assert recorded.as_dict()["executable_sha256"] == recorded.executable_sha256 def test_an_expired_deadline_leaves_the_report_saying_so() -> None: diff --git a/testing/tests/test_hermes_handoff_zero_evidence.py b/testing/tests/test_hermes_handoff_zero_evidence.py new file mode 100644 index 00000000..8e47768a --- /dev/null +++ b/testing/tests/test_hermes_handoff_zero_evidence.py @@ -0,0 +1,243 @@ +"""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 ``=`` 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 diff --git a/testing/tests/test_quality_contract.py b/testing/tests/test_quality_contract.py index fecae836..3c9ab2e3 100644 --- a/testing/tests/test_quality_contract.py +++ b/testing/tests/test_quality_contract.py @@ -2,17 +2,13 @@ from __future__ import annotations -import hashlib from pathlib import Path import textwrap from testing.quality_contract import load_contract from testing.quality_coverage import run_check as run_coverage_check from testing.quality_docs import run_check as run_docs_check -from testing.quality_hygiene import ( - count_files_over_line_limit, - run_check as run_hygiene_check, -) +from testing.quality_hygiene import run_check as run_hygiene_check def test_bundled_contract_exposes_local_and_jenkins_profiles(): @@ -28,7 +24,7 @@ def test_handoff_modules_are_exactly_managed_linted_and_covered(): expected = { path.as_posix() for path in Path("scripts/ops").glob("hermes_handoff_*.py") } - assert len(expected) == 15 + assert len(expected) == 16 assert expected <= set(contract["managed_modules"]) assert expected <= set(contract["lint_paths"]) assert expected <= set(contract["coverage"]["tracked_files"]) @@ -112,31 +108,6 @@ def test_hygiene_check_enforces_line_limit_and_name_rules(tmp_path: Path): ) -def test_hygiene_legacy_exception_is_exact_and_invalidated_by_any_edit(tmp_path: Path): - legacy = tmp_path / "legacy.py" - legacy.write_text("line\n" * 4, encoding="utf-8") - contract = { - "hygiene": { - "max_lines": 3, - "line_limit_globs": ["*.py"], - "legacy_line_exceptions": { - "legacy.py": { - "lines": 4, - "sha256": hashlib.sha256(legacy.read_bytes()).hexdigest(), - } - }, - } - } - assert run_hygiene_check(contract, tmp_path) == [] - assert count_files_over_line_limit(contract, tmp_path) == 0 - - legacy.write_text("edit\n" * 4, encoding="utf-8") - assert run_hygiene_check(contract, tmp_path) == [ - "file exceeds 3 LOC: legacy.py (4)" - ] - assert count_files_over_line_limit(contract, tmp_path) == 1 - - def test_coverage_check_enforces_per_file_floor(tmp_path: Path): build_dir = tmp_path / "build" build_dir.mkdir()