diff --git a/scripts/agent_exec.py b/scripts/agent_exec.py index 333521e0..44f1a4df 100644 --- a/scripts/agent_exec.py +++ b/scripts/agent_exec.py @@ -2,18 +2,23 @@ from __future__ import annotations import argparse +import contextlib +import errno import fcntl import hashlib +import hmac import json import os import re import select import shutil import signal +import stat import subprocess import sys import time import uuid +from collections.abc import Iterator from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -1971,11 +1976,18 @@ def _finish_run( ) history_path: Path | None = None try: - history_path = _append_execution_history(repo, receipt, secret_values) - receipt["history"] = { + history_path, attestation = _append_execution_history( + repo, receipt, secret_values + ) + recorded: dict[str, Any] = { "path": str(history_path), "status": "recorded", + "record_id": receipt.get("run_id"), + "run_id": receipt.get("run_id"), } + if attestation: + recorded["attestation"] = attestation + receipt["history"] = recorded except Exception as exc: receipt["history"] = { "path": str(history_path) if history_path is not None else None, @@ -1995,32 +2007,32 @@ def _append_execution_history( repo: Path, receipt: dict[str, Any], secret_values: set[str], -) -> Path: - common = _git( - repo.resolve(), - "rev-parse", - "--path-format=absolute", - "--git-common-dir", - check=False, - ) - if common.returncode != 0 or not common.stdout.strip(): - raise RuntimeError( - "unable to resolve git common dir: " - + (common.stderr.strip() or common.stdout.strip()) - ) - common_dir = Path(common.stdout.strip()) - if not common_dir.is_absolute(): - common_dir = (repo.resolve() / common_dir).resolve() - history_dir = common_dir / "agent-exec" +) -> tuple[Path, str | None]: + repo = repo.resolve() + history_dir = _git_common_dir(repo) / "agent-exec" if history_dir.is_symlink(): raise RuntimeError("agent-exec history directory must not be a symlink") history_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + if history_dir.is_symlink() or not history_dir.is_dir(): + raise RuntimeError("agent-exec history directory must not be a symlink") os.chmod(history_dir, 0o700) history_path = history_dir / "execution-history.jsonl" - record = _execution_history_record(repo, receipt) - line = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":")) - line = _redact_text(line, secret_values) + "\n" - fd = os.open(history_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + if history_path.is_symlink(): + raise RuntimeError("agent-exec history file must not be a symlink") + attestation: str | None = None + if receipt.get("terminal_status") == "success": + _bind_success_transport_commit(repo, receipt) + key = _ensure_attestation_key(repo) + record = _execution_history_record(repo, receipt) + line, attestation = _signed_history_line(record, key, secret_values) + else: + record = _execution_history_record(repo, receipt) + line = _redact_text(_canonical_json(record), secret_values) + "\n" + fd = os.open( + history_path, + os.O_WRONLY | os.O_CREAT | os.O_APPEND | os.O_NOFOLLOW | os.O_CLOEXEC, + 0o600, + ) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "a", encoding="utf-8", closefd=False) as handle: @@ -2033,7 +2045,7 @@ def _append_execution_history( fcntl.flock(fd, fcntl.LOCK_UN) finally: os.close(fd) - return history_path + return history_path, attestation def _execution_history_record(repo: Path, receipt: dict[str, Any]) -> dict[str, Any]: @@ -2073,6 +2085,9 @@ def _execution_history_record(repo: Path, receipt: dict[str, Any]) -> dict[str, "ref": transport.get("ref"), "parent": transport.get("parent"), } + commit = transport.get("commit") + if isinstance(commit, str) and commit: + safe_transport["commit"] = commit terminal = receipt.get("terminal_status") return { "schema": HISTORY_SCHEMA, @@ -2338,7 +2353,12 @@ def _valid_transport_ref(ref: object) -> str | None: return ref -def _history_file(repo: Path) -> Path | None: +_ATTESTATION_KEY_NAME = "attestation.key" +_ATTESTATION_KEY_BYTES = 32 + + +def _git_common_dir(repo: Path) -> Path: + repo = repo.resolve() common = _git( repo, "rev-parse", @@ -2347,11 +2367,200 @@ def _history_file(repo: Path) -> Path | None: check=False, ) if common.returncode != 0 or not common.stdout.strip(): - raise RuntimeError(common.stderr.strip() or "unable to resolve git common dir") + raise RuntimeError( + "unable to resolve git common dir: " + + (common.stderr.strip() or common.stdout.strip() or "empty") + ) common_dir = Path(common.stdout.strip()) if not common_dir.is_absolute(): common_dir = (repo / common_dir).resolve() - history_dir = common_dir / "agent-exec" + return common_dir + + +def _canonical_json(data: dict[str, Any]) -> str: + return json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _history_attestation(key: bytes, record: dict[str, Any]) -> str: + body = {field: value for field, value in record.items() if field != "attestation"} + return hmac.new(key, _canonical_json(body).encode("utf-8"), hashlib.sha256).hexdigest() + + +def _history_attestation_ok(key: bytes, record: dict[str, Any]) -> bool: + presented = record.get("attestation") + if not isinstance(presented, str) or len(presented) != 64: + return False + return hmac.compare_digest(_history_attestation(key, record), presented) + + +def _signed_history_line( + record: dict[str, Any], key: bytes, secret_values: set[str] +) -> tuple[str, str]: + text = _redact_text(_canonical_json(record), secret_values) + body = json.loads(text) + if not isinstance(body, dict): + raise RuntimeError("history record redaction produced non-object JSON") + mac = _history_attestation(key, body) + body["attestation"] = mac + return _canonical_json(body) + "\n", mac + + +def _bind_success_transport_commit(repo: Path, receipt: dict[str, Any]) -> None: + transport = receipt.get("transport") + if not isinstance(transport, dict): + raise RuntimeError("success history requires transport") + ref = transport.get("ref") + parent = transport.get("parent") + if not isinstance(ref, str) or not isinstance(parent, str): + raise RuntimeError("success history requires transport ref and parent") + commit = _resolve_commit(repo, ref) + resolved_parent = _resolve_commit(repo, parent) + if commit is None or resolved_parent is None: + raise RuntimeError("success transport ref did not resolve") + if _commit_parents(repo, commit) != [resolved_parent]: + raise RuntimeError("success transport parent does not match the commit") + transport["commit"] = commit + transport["parent"] = resolved_parent + + +def _symlink_oserror(exc: OSError, message: str) -> OSError | RuntimeError: + if exc.errno == errno.ELOOP: + return RuntimeError(message) + return exc + + +@contextlib.contextmanager +def _locked_agent_exec_dir(repo: Path, *, create: bool) -> Iterator[int | None]: + """Lock the real agent-exec directory. Missing dirs yield None unless creating. + + Callers share this lock so the first key write cannot race another reader. + """ + history_dir = _git_common_dir(repo) / "agent-exec" + if history_dir.is_symlink(): + raise RuntimeError("agent-exec history directory must not be a symlink") + if not create and not history_dir.exists(): + yield None + return + if create: + history_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + if history_dir.is_symlink() or not history_dir.is_dir(): + raise RuntimeError("agent-exec history directory must not be a symlink") + os.chmod(history_dir, 0o700) + elif not history_dir.is_dir(): + raise RuntimeError("agent-exec history directory must not be a symlink") + try: + fd = os.open( + history_dir, + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, + ) + except OSError as exc: + raise _symlink_oserror( + exc, "agent-exec history directory must not be a symlink" + ) from exc + try: + info = os.fstat(fd) + if not stat.S_ISDIR(info.st_mode): + raise RuntimeError("agent-exec history directory must not be a symlink") + if stat.S_IMODE(info.st_mode) != 0o700: + raise RuntimeError("agent-exec history directory mode must be 0700") + fcntl.flock(fd, fcntl.LOCK_EX) + try: + yield fd + finally: + fcntl.flock(fd, fcntl.LOCK_UN) + finally: + os.close(fd) + + +def _read_key_at(dir_fd: int) -> bytes: + fd = os.open( + _ATTESTATION_KEY_NAME, + os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, + dir_fd=dir_fd, + ) + try: + current = os.fstat(fd) + if not stat.S_ISREG(current.st_mode): + raise RuntimeError("attestation key must be a regular file") + if stat.S_IMODE(current.st_mode) != 0o600: + raise RuntimeError("attestation key mode must be 0600") + if current.st_size != _ATTESTATION_KEY_BYTES: + raise RuntimeError("attestation key must be 32 bytes") + os.lseek(fd, 0, os.SEEK_SET) + key = os.read(fd, _ATTESTATION_KEY_BYTES) + if len(key) != _ATTESTATION_KEY_BYTES: + raise RuntimeError("attestation key must be 32 bytes") + return key + finally: + os.close(fd) + + +def _key_is_symlink(dir_fd: int) -> bool: + try: + info = os.stat(_ATTESTATION_KEY_NAME, dir_fd=dir_fd, follow_symlinks=False) + except FileNotFoundError: + return False + return stat.S_ISLNK(info.st_mode) + + +def _read_attestation_key(repo: Path) -> bytes | None: + """Read the local attestation key. Never creates one and never follows links.""" + with _locked_agent_exec_dir(repo, create=False) as dir_fd: + if dir_fd is None: + return None + if _key_is_symlink(dir_fd): + raise RuntimeError("attestation key must not be a symlink") + try: + return _read_key_at(dir_fd) + except FileNotFoundError: + return None + except OSError as exc: + raise _symlink_oserror(exc, "attestation key must not be a symlink") from exc + + +def _ensure_attestation_key(repo: Path) -> bytes: + """Create or read the 32-byte key under the Git common dir. + + This binds controller history to later apply/admit checks. It does not + protect against the same OS user deliberately reading the key file. + """ + with _locked_agent_exec_dir(repo, create=True) as dir_fd: + if dir_fd is None: + raise RuntimeError("agent-exec history directory is unavailable") + if _key_is_symlink(dir_fd): + raise RuntimeError("attestation key must not be a symlink") + flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC + try: + fd = os.open(_ATTESTATION_KEY_NAME, flags, 0o600, dir_fd=dir_fd) + except OSError as exc: + raise _symlink_oserror(exc, "attestation key must not be a symlink") from exc + try: + os.fchmod(fd, 0o600) + current = os.fstat(fd) + if not stat.S_ISREG(current.st_mode): + raise RuntimeError("attestation key must be a regular file") + if current.st_size == 0: + key = os.urandom(_ATTESTATION_KEY_BYTES) + written = os.write(fd, key) + if written != _ATTESTATION_KEY_BYTES: + raise RuntimeError("attestation key write was short") + os.fsync(fd) + return key + if current.st_size != _ATTESTATION_KEY_BYTES: + raise RuntimeError("attestation key must be 32 bytes") + os.lseek(fd, 0, os.SEEK_SET) + key = os.read(fd, _ATTESTATION_KEY_BYTES) + if len(key) != _ATTESTATION_KEY_BYTES: + raise RuntimeError("attestation key must be 32 bytes") + if stat.S_IMODE(os.fstat(fd).st_mode) != 0o600: + raise RuntimeError("attestation key mode must be 0600") + return key + finally: + os.close(fd) + + +def _history_file(repo: Path) -> Path | None: + history_dir = _git_common_dir(repo) / "agent-exec" history_path = history_dir / "execution-history.jsonl" if history_dir.is_symlink() or history_path.is_symlink(): return None @@ -2381,7 +2590,10 @@ def _transport_covers( *, head: str, required: set[str], + key: bytes | None, ) -> bool: + if key is None or not _history_attestation_ok(key, record): + return False if record.get("schema") != HISTORY_SCHEMA: return False if record.get("terminal_status") != "success": @@ -2393,20 +2605,70 @@ def _transport_covers( return False ref = _valid_transport_ref(transport.get("ref")) parent = transport.get("parent") - if ref is None or not isinstance(parent, str): + signed_commit = transport.get("commit") + if ref is None or not isinstance(parent, str) or not isinstance(signed_commit, str): + return False + if _resolve_commit(repo, ref) != signed_commit: return False if _resolve_commit(repo, parent) != head: return False - commit = _resolve_commit(repo, ref) - if commit is None or _commit_parents(repo, commit) != [head]: + if _commit_parents(repo, signed_commit) != [head]: return False changed = record.get("changed_paths") if not isinstance(changed, list) or any(not isinstance(path, str) for path in changed): return False - diff_paths = _diff_paths(repo, head, commit) + diff_paths = _diff_paths(repo, head, signed_commit) if _non_exempt(changed) != required or _non_exempt(diff_paths) != required: return False - return _index_matches_commit(repo, commit, required) + return _index_matches_commit(repo, signed_commit, required) + + +def _apply_signed_history_ok( + repo: Path, + receipt: dict[str, Any], + *, + commit: str, + base_head: str, +) -> bool: + key = _read_attestation_key(repo) + if key is None: + return False + run_id = receipt.get("run_id") + transport = receipt.get("transport") + changed = receipt.get("changed_paths") + if not isinstance(run_id, str) or not run_id or not isinstance(transport, dict): + return False + ref = transport.get("ref") + parent = transport.get("parent") + claimed = transport.get("commit") + if not isinstance(ref, str) or parent != base_head or claimed != commit: + return False + if _resolve_commit(repo, ref) != commit: + return False + if _commit_parents(repo, commit) != [base_head]: + return False + if not isinstance(changed, list) or any(not isinstance(path, str) for path in changed): + return False + if _non_exempt(changed) != _non_exempt(_diff_paths(repo, base_head, commit)): + return False + for record in _load_history_records(repo): + if record.get("run_id") != run_id or record.get("terminal_status") != "success": + continue + if not _same_repo(repo, record.get("repo")): + continue + record_paths = record.get("changed_paths") + if record_paths != changed: + continue + rec_transport = record.get("transport") + if not isinstance(rec_transport, dict): + continue + if rec_transport.get("ref") != ref or rec_transport.get("parent") != parent: + continue + if rec_transport.get("commit") != commit: + continue + if _history_attestation_ok(key, record): + return True + return False def cmd_apply(repo: Path, receipt_path: Path) -> int: @@ -2472,6 +2734,15 @@ def cmd_apply(repo: Path, receipt_path: Path) -> int: ): return _reject("apply: mismatched transport", secrets) + try: + authorized = _apply_signed_history_ok( + repo, receipt, commit=commit, base_head=resolved_base + ) + except Exception as exc: + return _reject(f"apply: no signed history: {exc}", secrets) + if not authorized: + return _reject("apply: no signed history", secrets) + merge = _git(repo, "merge", "--ff-only", "--quiet", commit, check=False) if merge.returncode != 0: return _reject( @@ -2500,8 +2771,11 @@ def cmd_admit(repo: Path) -> int: if history_path is not None and history_path.is_file(): history_text = history_path.read_text(encoding="utf-8") secrets = _secret_set(history_text) + key = _read_attestation_key(repo) for record in _load_history_records(repo): - if _transport_covers(repo, record, head=head, required=required): + if _transport_covers( + repo, record, head=head, required=required, key=key + ): print("admit: ok") return EXIT_SUCCESS except Exception as exc: diff --git a/tests/test_agent_exec.py b/tests/test_agent_exec.py index 1dc166da..f079e019 100644 --- a/tests/test_agent_exec.py +++ b/tests/test_agent_exec.py @@ -5,6 +5,8 @@ Set RUN_REAL_AGENT_EXEC=1 and AGENT_EXEC_REAL_PROFILE to exactly one profile. """ from __future__ import annotations +import hashlib +import hmac import json import os import re @@ -2342,6 +2344,124 @@ class AgentExecHistoryTests(unittest.TestCase): self.assertEqual(records[0]["terminal_status"], "invalid-config") self.assertEqual(records[0]["attempts"], []) + def _run_marker_success(self, name: str) -> tuple[Path, dict[str, object]]: + _write_fake_worker( + self.bindir, + name, + body=textwrap.dedent( + """\ + workspace="$1" + result="$3" + mkdir -p "$workspace/src" + printf 'ok\\n' > "$workspace/src/marker.txt" + printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result" + exit 0 + """ + ), + ) + _write_config( + self.repo, + profiles={name: {"adapter": "custom", "executable": name}}, + role_order=[name], + ) + unit = _write_unit(self.root / f"{name}.json", unit_id=f"unit-{name}") + receipt_path = self.receipts / f"{name}.json" + code = agent_exec.cmd_run( + self.repo, + role="implementation", + unit_path=unit, + receipt_path=receipt_path, + ) + receipt_text = ( + receipt_path.read_text(encoding="utf-8") if receipt_path.is_file() else "" + ) + self.assertEqual(code, agent_exec.EXIT_SUCCESS, receipt_text) + return receipt_path, json.loads(receipt_text) + + def test_success_history_is_mode_correct_and_signed(self) -> None: + receipt_path, receipt = self._run_marker_success("signedhistory") + key_path = self.history_path.parent / "attestation.key" + self.assertTrue(key_path.is_file()) + self.assertFalse(key_path.is_symlink()) + self.assertFalse(key_path.parent.is_symlink()) + self.assertEqual(stat.S_IMODE(key_path.parent.stat().st_mode), 0o700) + self.assertEqual(stat.S_IMODE(key_path.stat().st_mode), 0o600) + key = key_path.read_bytes() + self.assertEqual(len(key), 32) + self.assertNotIn(key, receipt_path.read_bytes()) + self.assertNotIn(key, self.history_path.read_bytes()) + + records = [ + json.loads(line) + for line in self.history_path.read_text(encoding="utf-8").splitlines() + ] + self.assertEqual(len(records), 1) + record = records[0] + transport = record["transport"] + ref = transport["ref"] + commit = _git(self.repo, "rev-parse", ref).stdout.strip() + self.assertEqual(record["terminal_status"], "success") + self.assertEqual(record["run_id"], receipt["run_id"]) + self.assertEqual(transport["parent"], receipt["transport"]["parent"]) + self.assertEqual(transport["commit"], commit) + self.assertEqual(receipt["transport"]["commit"], commit) + self.assertEqual(record["changed_paths"], receipt["changed_paths"]) + body = {field: value for field, value in record.items() if field != "attestation"} + payload = json.dumps( + body, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + self.assertEqual( + hmac.new(key, payload, hashlib.sha256).hexdigest(), + record["attestation"], + ) + identity = receipt["history"] + self.assertEqual(identity["status"], "recorded") + self.assertEqual(identity["run_id"], receipt["run_id"]) + self.assertEqual(identity["record_id"], record["record_id"]) + self.assertEqual(identity["attestation"], record["attestation"]) + self.assertNotIn(key.hex(), json.dumps(identity)) + + def test_attestation_key_symlink_is_not_followed(self) -> None: + history_dir = self.history_path.parent + history_dir.mkdir(mode=0o700) + outside = self.root / "outside-key" + outside.write_bytes(b"A" * 32) + key_path = history_dir / "attestation.key" + key_path.symlink_to(outside) + receipt_path, receipt = self._run_marker_success("symlinkkey") + self.assertTrue(key_path.is_symlink()) + self.assertEqual(outside.read_bytes(), b"A" * 32) + self.assertEqual(receipt["history"]["status"], "failed") + self.assertNotIn(b"A" * 32, receipt_path.read_bytes()) + + def test_attestation_key_first_run_race_shares_one_key(self) -> None: + found: list[bytes] = [] + errors: list[BaseException] = [] + barrier = threading.Barrier(8) + + def load() -> None: + try: + barrier.wait(timeout=5) + found.append(agent_exec._ensure_attestation_key(self.repo)) + except BaseException as exc: # noqa: BLE001 - race failures are the assertion + errors.append(exc) + + threads = [threading.Thread(target=load) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + self.assertFalse(thread.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(len(found), 8) + self.assertEqual(len({item for item in found}), 1) + self.assertEqual(len(found[0]), 32) + key_path = self.history_path.parent / "attestation.key" + self.assertFalse(key_path.is_symlink()) + self.assertEqual(key_path.read_bytes(), found[0]) + self.assertEqual(stat.S_IMODE(key_path.stat().st_mode), 0o600) + self.assertEqual(stat.S_IMODE(key_path.parent.stat().st_mode), 0o700) + # Opt-in real CLI probes (U5). Default unittest must not contact providers. U5_PROFILES = ( diff --git a/tests/test_agent_exec_admission.py b/tests/test_agent_exec_admission.py index bf29adaf..c97bb854 100644 --- a/tests/test_agent_exec_admission.py +++ b/tests/test_agent_exec_admission.py @@ -1,6 +1,8 @@ """Provenance admission: apply a success transport, or admit exact staged paths.""" from __future__ import annotations +import hashlib +import hmac import json import os import shutil @@ -148,6 +150,8 @@ class AgentExecAdmissionTests(unittest.TestCase): reason: str | None = None, transport_parent: str | None = None, omit_transport: bool = False, + run_id: str | None = None, + commit: str | None = None, ) -> Path: data: dict[str, object] = { "schema": agent_exec.RECEIPT_SCHEMA, @@ -161,16 +165,65 @@ class AgentExecAdmissionTests(unittest.TestCase): }, "changed_paths": paths, } + if run_id is not None: + data["run_id"] = run_id if not omit_transport: - data["transport"] = { + transport: dict[str, object] = { "ref": ref, "parent": parent if transport_parent is None else transport_parent, "attempt_id": "attempt-1", } + if commit is not None: + transport["commit"] = commit + data["transport"] = transport path = self.root / f"receipt-{uuid.uuid4().hex}.json" path.write_text(json.dumps(data), encoding="utf-8") return path + def _common_dir(self) -> Path: + return Path( + _git( + self.repo, + "rev-parse", + "--path-format=absolute", + "--git-common-dir", + ).stdout.strip() + ) + + def _history_path(self) -> Path: + return self._common_dir() / "agent-exec" / "execution-history.jsonl" + + def _attestation_key(self) -> bytes: + dest = self._common_dir() / "agent-exec" + dest.mkdir(mode=0o700, exist_ok=True) + os.chmod(dest, 0o700) + path = dest / "attestation.key" + if path.is_symlink(): + raise RuntimeError("test attestation key must not be a symlink") + if path.exists(): + key = path.read_bytes() + if len(key) != 32: + raise RuntimeError("test attestation key must be 32 bytes") + return key + key = os.urandom(32) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + try: + os.write(fd, key) + os.fchmod(fd, 0o600) + finally: + os.close(fd) + return key + + @staticmethod + def _sign_history_record(record: dict[str, object], key: bytes) -> dict[str, object]: + body = {field: value for field, value in record.items() if field != "attestation"} + payload = json.dumps( + body, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + signed = dict(body) + signed["attestation"] = hmac.new(key, payload, hashlib.sha256).hexdigest() + return signed + def _history( self, *, @@ -179,26 +232,39 @@ class AgentExecAdmissionTests(unittest.TestCase): paths: list[str], status: str = "success", repo: str | None = None, - ) -> None: - record = { + commit: str | None = None, + run_id: str | None = None, + sign: bool | None = None, + ) -> dict[str, object]: + if commit is None: + commit = _git(self.repo, "rev-parse", f"{ref}^{{commit}}").stdout.strip() + identity = run_id or uuid.uuid4().hex + record: dict[str, object] = { "schema": agent_exec.HISTORY_SCHEMA, "terminal_status": status, + "record_id": identity, + "run_id": identity, "repo": repo if repo is not None else str(self.repo.resolve()), "changed_paths": paths, - "transport": {"ref": ref, "parent": parent}, + "transport": {"ref": ref, "parent": parent, "commit": commit}, } - common = Path( - _git( - self.repo, - "rev-parse", - "--path-format=absolute", - "--git-common-dir", - ).stdout.strip() - ) - dest = common / "agent-exec" + if sign is None: + sign = status == "success" + if sign: + record = self._sign_history_record(record, self._attestation_key()) + dest = self._history_path().parent dest.mkdir(mode=0o700, exist_ok=True) - with (dest / "execution-history.jsonl").open("a", encoding="utf-8") as handle: - handle.write(json.dumps(record, sort_keys=True) + "\n") + os.chmod(dest, 0o700) + line = json.dumps( + record, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + with self._history_path().open("a", encoding="utf-8") as handle: + handle.write(line + "\n") + return record + + def _reset_branch(self) -> None: + _git(self.repo, "reset", "--hard", self.head) + _git(self.repo, "clean", "-fd") def _stage(self, rel: str, content: str) -> None: path = self.repo / rel @@ -208,9 +274,23 @@ class AgentExecAdmissionTests(unittest.TestCase): def test_apply_fast_forwards_clean_matching_transport(self) -> None: commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"}) - receipt = self._receipt( - parent=self.head, ref=ref, paths=["src/marker.txt"] + run_id = uuid.uuid4().hex + signed = self._history( + parent=self.head, + ref=ref, + paths=["src/marker.txt"], + commit=commit, + run_id=run_id, ) + receipt = self._receipt( + parent=self.head, + ref=ref, + paths=["src/marker.txt"], + run_id=run_id, + commit=commit, + ) + self.assertEqual(signed["transport"]["commit"], commit) + self.assertIn("attestation", signed) proc = self._cli( "apply", "--repo", str(self.repo), "--receipt", str(receipt) ) @@ -309,6 +389,64 @@ class AgentExecAdmissionTests(unittest.TestCase): ), ) + def test_apply_rejects_fabricated_unsigned_or_tampered_history(self) -> None: + commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"}) + run_id = uuid.uuid4().hex + paths = ["src/marker.txt"] + + def reject(label: str) -> None: + self._reset_branch() + before = self._state() + receipt = self._receipt( + parent=self.head, + ref=ref, + paths=paths, + run_id=run_id, + commit=commit, + ) + proc = self._cli( + "apply", "--repo", str(self.repo), "--receipt", str(receipt) + ) + combined = proc.stdout + proc.stderr + self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined) + self.assertIn("apply: no signed history", proc.stderr) + self.assertNotIn("apply: fast-forwarded", proc.stdout) + self.assertEqual(self._state(), before, label) + + with self.subTest(label="fabricated receipt"): + reject("fabricated receipt") + + with self.subTest(label="unsigned history"): + self._history( + parent=self.head, + ref=ref, + paths=paths, + commit=commit, + run_id=run_id, + sign=False, + ) + reject("unsigned history") + + with self.subTest(label="tampered history"): + self._reset_branch() + signed = self._history( + parent=self.head, + ref=ref, + paths=paths, + commit=commit, + run_id=run_id, + ) + history_path = self._history_path() + lines = history_path.read_text(encoding="utf-8").splitlines() + tampered = json.loads(lines[-1]) + self.assertEqual(tampered["attestation"], signed["attestation"]) + tampered["record_id"] = "tampered-record" + lines[-1] = json.dumps( + tampered, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + history_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + reject("tampered history") + def test_apply_redacts_secret_shaped_receipt_text(self) -> None: _commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"}) secret = "sk-admissionsecret" @@ -360,6 +498,69 @@ class AgentExecAdmissionTests(unittest.TestCase): self.assertEqual(partial.returncode, agent_exec.EXIT_BLOCKED, partial.stderr) self.assertIn("admit: no provenance", partial.stderr) + def test_admit_rejects_tampered_record_and_moved_ref(self) -> None: + commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"}) + signed = self._history(parent=self.head, ref=ref, paths=["src/marker.txt"], commit=commit) + history_path = self._history_path() + lines = history_path.read_text(encoding="utf-8").splitlines() + tampered = json.loads(lines[-1]) + self.assertEqual(tampered["attestation"], signed["attestation"]) + tampered["record_id"] = "tampered-record" + lines[-1] = json.dumps( + tampered, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + history_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + self._stage("src/marker.txt", "shipped\n") + before = self._state() + proc = self._cli("admit", "--repo", str(self.repo)) + combined = proc.stdout + proc.stderr + self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined) + self.assertIn("admit: no provenance", proc.stderr) + self.assertNotIn("admit: ok", proc.stdout) + self.assertEqual(self._state(), before) + + self._reset_branch() + tree = _git(self.repo, "rev-parse", f"{commit}^{{tree}}").stdout.strip() + moved = _git( + self.repo, + "commit-tree", + tree, + "-p", + self.head, + "-m", + "moved ref", + ).stdout.strip() + self.assertNotEqual(moved, commit) + restored = self._sign_history_record( + { + "schema": agent_exec.HISTORY_SCHEMA, + "terminal_status": "success", + "record_id": signed["record_id"], + "run_id": signed["run_id"], + "repo": str(self.repo.resolve()), + "changed_paths": ["src/marker.txt"], + "transport": {"commit": commit, "parent": self.head, "ref": ref}, + }, + self._attestation_key(), + ) + history_path.write_text( + json.dumps( + restored, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + + "\n", + encoding="utf-8", + ) + _git(self.repo, "update-ref", ref, moved) + self._stage("src/marker.txt", "shipped\n") + moved_before = self._state() + moved_proc = self._cli("admit", "--repo", str(self.repo)) + moved_text = moved_proc.stdout + moved_proc.stderr + self.assertEqual(moved_proc.returncode, agent_exec.EXIT_BLOCKED, moved_text) + self.assertIn("admit: no provenance", moved_proc.stderr) + self.assertNotIn("admit: ok", moved_proc.stdout) + self.assertEqual(self._state(), moved_before) + self.assertEqual(_git(self.repo, "rev-parse", ref).stdout.strip(), moved) + def test_admit_rejects_different_content_at_transport_path(self) -> None: _commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"}) self._history(parent=self.head, ref=ref, paths=["src/marker.txt"])