674 lines
26 KiB
Python
674 lines
26 KiB
Python
"""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
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from scripts import agent_exec
|
|
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["git", "-C", str(repo), *args],
|
|
check=check,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
|
|
class AgentExecAdmissionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = tempfile.TemporaryDirectory(prefix="agent-exec-admit-")
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.repo = self.root / "repo"
|
|
self.repo.mkdir()
|
|
_git(self.repo, "init", "-b", "main")
|
|
_git(self.repo, "config", "user.email", "agent-exec@example.com")
|
|
_git(self.repo, "config", "user.name", "Agent Exec")
|
|
(self.repo / "README.md").write_text("fixture\n", encoding="utf-8")
|
|
_git(self.repo, "add", "README.md")
|
|
_git(self.repo, "commit", "-m", "init")
|
|
_git(self.repo, "checkout", "-b", "feature/admission-test")
|
|
self.head = _git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
|
|
|
def _cli(self, *args: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
|
env["PYTHONPATH"] = str(SOURCE)
|
|
return subprocess.run(
|
|
["python3", "-m", "scripts.agent_exec", *args],
|
|
cwd=str(self.root),
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
|
|
def _state(self) -> dict[str, str]:
|
|
return {
|
|
"head": _git(self.repo, "rev-parse", "HEAD").stdout.strip(),
|
|
"branch": _git(self.repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip(),
|
|
"status": _git(
|
|
self.repo, "status", "--porcelain", "--untracked-files=all"
|
|
).stdout,
|
|
"index": _git(self.repo, "ls-files", "--stage").stdout,
|
|
"readme": (self.repo / "README.md").read_text(encoding="utf-8"),
|
|
}
|
|
|
|
def _transport_commit(
|
|
self,
|
|
files: dict[str, str],
|
|
*,
|
|
parent: str | None = None,
|
|
ref: str | None = None,
|
|
) -> tuple[str, str]:
|
|
parent = parent or _git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
|
ref = ref or f"refs/agent-exec/{uuid.uuid4().hex}/attempt-1"
|
|
index = self.root / f"index-{uuid.uuid4().hex}"
|
|
env = os.environ.copy()
|
|
env["GIT_INDEX_FILE"] = str(index)
|
|
subprocess.run(
|
|
["git", "-C", str(self.repo), "read-tree", parent],
|
|
check=True,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
for rel, content in files.items():
|
|
blob = subprocess.run(
|
|
["git", "-C", str(self.repo), "hash-object", "-w", "--stdin"],
|
|
input=content,
|
|
text=True,
|
|
check=True,
|
|
capture_output=True,
|
|
).stdout.strip()
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(self.repo),
|
|
"update-index",
|
|
"--add",
|
|
"--cacheinfo",
|
|
f"100644,{blob},{rel}",
|
|
],
|
|
check=True,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
tree = subprocess.run(
|
|
["git", "-C", str(self.repo), "write-tree"],
|
|
check=True,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
commit = subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(self.repo),
|
|
"-c",
|
|
"user.name=agent-exec-transport",
|
|
"-c",
|
|
"user.email=agent-exec-transport@localhost.invalid",
|
|
"commit-tree",
|
|
tree,
|
|
"-p",
|
|
parent,
|
|
"-m",
|
|
"agent-exec transport",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
_git(self.repo, "update-ref", ref, commit)
|
|
index.unlink(missing_ok=True)
|
|
return commit, ref
|
|
|
|
def _receipt(
|
|
self,
|
|
*,
|
|
parent: str,
|
|
ref: str | None,
|
|
paths: list[str],
|
|
status: str = "success",
|
|
repo: str | None = None,
|
|
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,
|
|
"repo": repo if repo is not None else str(self.repo.resolve()),
|
|
"terminal_status": status,
|
|
"reason": reason,
|
|
"base": {
|
|
"head": parent,
|
|
"tree": _git(self.repo, "rev-parse", f"{parent}^{{tree}}").stdout.strip(),
|
|
"branch": "feature/admission-test",
|
|
},
|
|
"changed_paths": paths,
|
|
}
|
|
if run_id is not None:
|
|
data["run_id"] = run_id
|
|
if not omit_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,
|
|
*,
|
|
parent: str,
|
|
ref: str,
|
|
paths: list[str],
|
|
status: str = "success",
|
|
repo: str | None = None,
|
|
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, "commit": commit},
|
|
}
|
|
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)
|
|
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
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
_git(self.repo, "add", "--", rel)
|
|
|
|
def test_apply_fast_forwards_clean_matching_transport(self) -> None:
|
|
commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
|
|
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)
|
|
)
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
|
|
self.assertIn("apply: fast-forwarded", proc.stdout)
|
|
self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout.strip(), commit)
|
|
self.assertEqual(
|
|
_git(self.repo, "rev-parse", "feature/admission-test").stdout.strip(),
|
|
commit,
|
|
)
|
|
self.assertEqual((self.repo / "src/marker.txt").read_text(encoding="utf-8"), "shipped\n")
|
|
self.assertEqual(_git(self.repo, "status", "--porcelain").stdout, "")
|
|
parents = _git(self.repo, "rev-list", "--parents", "-n", "1", "HEAD").stdout.split()
|
|
self.assertEqual(parents, [commit, self.head])
|
|
|
|
def test_apply_rejects_without_changing_state(self) -> None:
|
|
_commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
|
|
valid = dict(parent=self.head, ref=ref, paths=["src/marker.txt"])
|
|
|
|
def reject(label: str, receipt: Path) -> None:
|
|
before = self._state()
|
|
proc = self._cli(
|
|
"apply", "--repo", str(self.repo), "--receipt", str(receipt)
|
|
)
|
|
self.assertEqual(
|
|
proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout
|
|
)
|
|
self.assertIn(label, proc.stderr)
|
|
self.assertNotIn("apply: fast-forwarded", proc.stdout)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
with self.subTest(label="stale HEAD"):
|
|
(self.repo / "advance.txt").write_text("advance\n", encoding="utf-8")
|
|
_git(self.repo, "add", "advance.txt")
|
|
_git(self.repo, "commit", "-m", "advance")
|
|
reject("stale HEAD", self._receipt(**valid))
|
|
_git(self.repo, "reset", "--hard", self.head)
|
|
|
|
with self.subTest(label="dirty worktree"):
|
|
(self.repo / "README.md").write_text("dirty\n", encoding="utf-8")
|
|
reject("dirty worktree", self._receipt(**valid))
|
|
_git(self.repo, "checkout", "--", "README.md")
|
|
|
|
with self.subTest(label="non-success receipt"):
|
|
reject(
|
|
"non-success receipt",
|
|
self._receipt(**valid, status="failed-unknown"),
|
|
)
|
|
|
|
with self.subTest(label="missing transport ref"):
|
|
reject(
|
|
"missing transport",
|
|
self._receipt(
|
|
parent=self.head,
|
|
ref="refs/agent-exec/missing/attempt",
|
|
paths=["src/marker.txt"],
|
|
),
|
|
)
|
|
|
|
with self.subTest(label="omitted transport"):
|
|
reject(
|
|
"missing transport",
|
|
self._receipt(
|
|
parent=self.head,
|
|
ref=None,
|
|
paths=["src/marker.txt"],
|
|
omit_transport=True,
|
|
),
|
|
)
|
|
|
|
with self.subTest(label="mismatched transport"):
|
|
side = _git(
|
|
self.repo,
|
|
"commit-tree",
|
|
_git(self.repo, "rev-parse", "HEAD^{tree}").stdout.strip(),
|
|
"-p",
|
|
self.head,
|
|
"-m",
|
|
"side",
|
|
).stdout.strip()
|
|
_mismatch_commit, mismatch_ref = self._transport_commit(
|
|
{"src/marker.txt": "other\n"}, parent=side
|
|
)
|
|
reject(
|
|
"mismatched transport",
|
|
self._receipt(
|
|
parent=self.head, ref=mismatch_ref, paths=["src/marker.txt"]
|
|
),
|
|
)
|
|
|
|
with self.subTest(label="receipt repo mismatch"):
|
|
reject(
|
|
"receipt repo mismatch",
|
|
self._receipt(
|
|
**valid, repo=str((self.root / "other-repo").resolve())
|
|
),
|
|
)
|
|
|
|
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 local integrity record", 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"
|
|
receipt = self._receipt(
|
|
parent=self.head,
|
|
ref=ref,
|
|
paths=["src/marker.txt"],
|
|
status="failed-unknown",
|
|
reason=secret,
|
|
)
|
|
before = self._state()
|
|
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("non-success receipt", proc.stderr)
|
|
self.assertNotIn(secret, combined)
|
|
self.assertIn("***", combined)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
def test_admit_rejects_staged_paths_without_provenance(self) -> None:
|
|
self._stage("src/extra.txt", "unproven\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: no local integrity record", proc.stderr)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
def test_admit_accepts_exact_transport_paths(self) -> None:
|
|
_commit, ref = self._transport_commit(
|
|
{"src/marker.txt": "shipped\n", "src/other.txt": "also\n"}
|
|
)
|
|
self._history(
|
|
parent=self.head,
|
|
ref=ref,
|
|
paths=["src/marker.txt", "src/other.txt"],
|
|
)
|
|
self._stage("src/marker.txt", "shipped\n")
|
|
self._stage("src/other.txt", "also\n")
|
|
self._stage(".task/feature-admission.md", "host card\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: ok", proc.stdout)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
self._git_reset_index()
|
|
self._stage("src/marker.txt", "shipped\n")
|
|
partial = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(partial.returncode, agent_exec.EXIT_BLOCKED, partial.stderr)
|
|
self.assertIn("admit: no local integrity record", 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 local integrity record", 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 local integrity record", 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"])
|
|
self._stage("src/marker.txt", "rewritten\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: no local integrity record", proc.stderr)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
def test_rename_and_copy_records_keep_non_exempt_source(self) -> None:
|
|
"""Copy/rename records must not drop a non-exempt source behind an exempt dest."""
|
|
paths = agent_exec._name_status_paths(
|
|
"R100\tsrc/secret.txt\t.task/from-rename.md\n"
|
|
"C100\tlib/outside.txt\tdocs/superpowers/plans/from-copy.md\n"
|
|
)
|
|
self.assertEqual(
|
|
agent_exec._non_exempt(paths),
|
|
{"src/secret.txt", "lib/outside.txt"},
|
|
)
|
|
|
|
def test_admit_rejects_rename_of_non_exempt_source_into_exempt_path(self) -> None:
|
|
(self.repo / "src").mkdir()
|
|
(self.repo / "src" / "secret.txt").write_text("secret\n", encoding="utf-8")
|
|
_git(self.repo, "add", "src/secret.txt")
|
|
_git(self.repo, "commit", "-m", "add secret")
|
|
(self.repo / ".task").mkdir()
|
|
_git(self.repo, "mv", "src/secret.txt", ".task/hidden.md")
|
|
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("src/secret.txt", proc.stderr)
|
|
self.assertNotIn("admit: ok", proc.stdout)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
def test_admit_accepts_host_managed_exemptions(self) -> None:
|
|
self._stage(".task/feature-admission.md", "card\n")
|
|
self._stage("docs/superpowers/specs/admission.md", "spec\n")
|
|
self._stage("docs/superpowers/plans/admission.md", "plan\n")
|
|
before = self._state()
|
|
proc = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(proc.returncode, agent_exec.EXIT_SUCCESS, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: ok", proc.stdout)
|
|
self.assertEqual(self._state(), before)
|
|
|
|
self._stage("docs/superpowers/specs-extra/not-exempt.md", "nope\n")
|
|
rejected = self._cli("admit", "--repo", str(self.repo))
|
|
self.assertEqual(rejected.returncode, agent_exec.EXIT_BLOCKED, rejected.stderr)
|
|
self.assertIn("admit: no local integrity record", rejected.stderr)
|
|
|
|
def test_precommit_rejects_unproven_paths_after_branch_checks(self) -> None:
|
|
self._install_project_hook()
|
|
self._stage("src/extra.txt", "unproven\n")
|
|
proc = self._hook()
|
|
combined = proc.stdout + proc.stderr
|
|
self.assertNotEqual(proc.returncode, 0, combined)
|
|
self.assertIn("admit: no local integrity record", combined)
|
|
|
|
_git(self.repo, "checkout", "-B", "main", self.head)
|
|
self._stage("src/extra.txt", "unproven\n")
|
|
blocked = self._hook()
|
|
text = blocked.stdout + blocked.stderr
|
|
self.assertNotEqual(blocked.returncode, 0, text)
|
|
self.assertIn("集成分支", text)
|
|
self.assertNotIn("admit:", text)
|
|
|
|
def test_precommit_accepts_host_managed_exemptions(self) -> None:
|
|
self._install_project_hook()
|
|
self._stage("docs/superpowers/plans/admission.md", "plan\n")
|
|
proc = self._hook()
|
|
self.assertEqual(proc.returncode, 0, proc.stderr + proc.stdout)
|
|
self.assertIn("admit: ok", proc.stdout)
|
|
|
|
def test_make_agent_exec_apply_requires_receipt(self) -> None:
|
|
env = os.environ.copy()
|
|
env.pop("RECEIPT", None)
|
|
proc = subprocess.run(
|
|
["make", "agent-exec-apply"],
|
|
cwd=str(SOURCE),
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
combined = proc.stdout + proc.stderr
|
|
self.assertNotEqual(proc.returncode, 0, combined)
|
|
self.assertIn("RECEIPT", combined)
|
|
self.assertNotIn("apply: fast-forwarded", combined)
|
|
|
|
def _git_reset_index(self) -> None:
|
|
_git(self.repo, "reset", "--hard", self.head)
|
|
_git(self.repo, "clean", "-fd")
|
|
|
|
def _install_project_hook(self) -> None:
|
|
dest = self.repo / "scripts"
|
|
hooks = dest / "hooks"
|
|
hooks.mkdir(parents=True)
|
|
for name in ("agent_exec.py", "agent_exec_config.py", "agent_exec_adapters.py"):
|
|
shutil.copy(SOURCE / "scripts" / name, dest / name)
|
|
hook = hooks / "pre-commit"
|
|
shutil.copy(SOURCE / "scripts" / "hooks" / "pre-commit", hook)
|
|
hook.chmod(0o755)
|
|
|
|
def _hook(self) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["bash", str(self.repo / "scripts" / "hooks" / "pre-commit")],
|
|
cwd=str(self.repo),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|