import hashlib import json import os import subprocess import sys import tempfile import unittest from pathlib import Path SCRIPT = Path(__file__).resolve().parents[2] / "agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py" def run_git(cwd, *args): return subprocess.run(["git", *args], cwd=cwd, text=True, capture_output=True, check=True) class EvidenceManifestTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.repo = Path(self.tmp.name) / "repo" self.repo.mkdir() run_git(self.repo, "init", "-q") run_git(self.repo, "config", "user.email", "test@example.invalid") run_git(self.repo, "config", "user.name", "Test") (self.repo / "input.txt").write_text("hello\n") run_git(self.repo, "add", "input.txt") run_git(self.repo, "commit", "-qm", "initial") def tearDown(self): self.tmp.cleanup() def invoke(self, *args, env=None): return subprocess.run([sys.executable, str(SCRIPT), *map(str, args)], cwd=self.repo, text=True, capture_output=True, env=env) def plan(self, commands=None, **extra): if commands is None: commands = [{"id": "check", "purpose": "show input", "argv": ["git", "show", "HEAD:input.txt"], "expected_exit_code": 0, "claims": ["output"]}] data = {"hash_paths": ["input.txt"], "commands": commands} data.update(extra); p = self.repo / "plan.json"; p.write_text(json.dumps(data)); return p def run_manifest(self, plan=None): if plan is None: plan = self.plan() out = self.repo / "manifest.json"; result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", "manifest.json"); return result, out def test_run_captures_success_claim_and_hash(self): result, out = self.run_manifest(); self.assertEqual(result.returncode, 0, result.stderr) manifest = json.loads(out.read_text()) self.assertEqual(manifest["commands"][0]["argv"], ["git", "show", "HEAD:input.txt"]) self.assertEqual(manifest["commands"][0]["exit_code"], 0) self.assertEqual(manifest["commands"][0]["claim_ids"], ["output"]) self.assertEqual( manifest["claims"]["output"], [ "/commands/0/argv", "/commands/0/exit_code", "/commands/0/expected_exit_code", "/commands/0/output", "/commands/0/started_at_utc", "/commands/0/finished_at_utc", ], ) self.assertEqual(manifest["files"][0]["sha256"], hashlib.sha256(b"hello\n").hexdigest()) self.assertEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", "manifest.json").returncode, 0) def test_unexpected_exit_is_recorded_and_nonzero(self): result, out = self.run_manifest(self.plan([{"argv": [sys.executable, "-c", "print('no'); raise SystemExit(3)"], "expected_exit_code": 0, "claims": ["result"]}])) self.assertNotEqual(result.returncode, 0); self.assertEqual(json.loads(out.read_text())["commands"][0]["exit_code"], 3) def test_failed_run_preserves_prior_manifest(self): result, out = self.run_manifest() self.assertEqual(result.returncode, 0, result.stderr) prior = out.read_bytes() bad = self.plan([{"argv": [sys.executable, "-c", "print('Authorization: Bearer ***')"]}]) result = self.invoke("run", "--repo-root", self.repo, "--plan", bad, "--output", out) self.assertNotEqual(result.returncode, 0) self.assertEqual(out.read_bytes(), prior) def test_command_environment_is_allowlisted_and_output_is_bounded(self): env = os.environ.copy() env["UNLISTED_SECRET"] = "must-not-reach-child" command = [{ "argv": [sys.executable, "-c", "import os; print(os.getenv('UNLISTED_SECRET'))"], "claims": ["environment"], }] plan = self.plan(command) out = self.repo / "manifest.json" result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", out.name, env=env) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(out.read_text())["commands"][0]["output"], "None\n") env["LEAK"] = "SECRET_VALUE" plan = self.plan(command, command_env_allowlist=["LEAK"]) result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", out.name, env=env) self.assertNotEqual(result.returncode, 0) self.assertNotIn("SECRET_VALUE", out.read_text()) prior = out.read_bytes() plan = self.plan([{ "argv": [sys.executable, "-c", "print('x' * 100000)"], "max_output_bytes": 16, "claims": ["bounded"], }]) result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", out.name) self.assertNotEqual(result.returncode, 0) self.assertEqual(out.read_bytes(), prior) def test_output_collision_and_password_argument_are_rejected(self): plan = self.plan() result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", "input.txt") self.assertNotEqual(result.returncode, 0) self.assertEqual((self.repo / "input.txt").read_text(), "hello\n") result, out = self.run_manifest(self.plan([{ "argv": [sys.executable, "-c", "from pathlib import Path; Path('ran.txt').write_text('ran')", "--password", "visible-value"], "claims": ["password-leak"], }])) self.assertNotEqual(result.returncode, 0) self.assertFalse(out.exists()) self.assertFalse((self.repo / "ran.txt").exists()) def test_argv_only_and_traversal_rejected(self): result, out = self.run_manifest(self.plan([{"argv": ["echo", "shell=true"]}])) self.assertNotEqual(result.returncode, 0); self.assertFalse(out.exists()) result, out = self.run_manifest(self.plan(hash_paths=["../outside"])) self.assertNotEqual(result.returncode, 0); self.assertFalse(out.exists()) result, out = self.run_manifest(self.plan(hash_paths=["nested/../input.txt"])) self.assertNotEqual(result.returncode, 0); self.assertFalse(out.exists()) def test_empty_plan_and_malformed_command_record_are_rejected(self): result, out = self.run_manifest(self.plan([])) self.assertNotEqual(result.returncode, 0) self.assertFalse(out.exists()) result, out = self.run_manifest() self.assertEqual(result.returncode, 0, result.stderr) data = json.loads(out.read_text()) data["commands"][0]["duration_seconds"] = -1 out.write_text(json.dumps(data, sort_keys=True, separators=(",", ":")) + "\n") self.assertNotEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", out.name).returncode, 0) def test_secret_or_header_leak_fails_closed(self): env = os.environ.copy(); env["TEST_SECRET"] = "never-write-this" p = self.plan([{"argv": [sys.executable, "-c", "import os; print(os.environ['TEST_SECRET'])"]}], secret_env_names=["TEST_SECRET"]) result = self.invoke("run", "--repo-root", self.repo, "--plan", "plan.json", "--output", "manifest.json", env=env) self.assertNotEqual(result.returncode, 0); self.assertFalse((self.repo / "manifest.json").exists()) def test_source_drift_hash_mismatch_rejected(self): result, out = self.run_manifest(); self.assertEqual(result.returncode, 0) (self.repo / "input.txt").write_text("changed\n") self.assertNotEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", out.name).returncode, 0) def test_command_source_mutation_is_rejected_before_manifest_write(self): command = [{ "argv": [sys.executable, "-c", "from pathlib import Path; Path('input.txt').write_text('changed\\n')"], "claims": ["mutation"], }] result, out = self.run_manifest(self.plan(command)) self.assertNotEqual(result.returncode, 0) self.assertFalse(out.exists()) def test_command_commit_drift_is_rejected_before_manifest_write(self): script = ( "from pathlib import Path; import subprocess; " "Path('other.txt').write_text('committed\\n'); " "subprocess.run(['git','add','other.txt'],check=True); " "subprocess.run(['git','commit','-qm','drift'],check=True)" ) result, out = self.run_manifest(self.plan([{ "argv": [sys.executable, "-c", script], "claims": ["commit-drift"], }])) self.assertNotEqual(result.returncode, 0) self.assertFalse(out.exists()) def test_binary_source_hash_is_byte_accurate(self): payload = b"\xff\x00\x80binary\n" (self.repo / "binary.bin").write_bytes(payload) run_git(self.repo, "add", "binary.bin") run_git(self.repo, "commit", "-qm", "binary") result, out = self.run_manifest(self.plan(hash_paths=["binary.bin"])) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(json.loads(out.read_text())["files"][0]["sha256"], hashlib.sha256(payload).hexdigest()) def test_invalid_claim_pointer_rejected(self): result, out = self.run_manifest(self.plan([{"argv": ["true"], "claims": ["bad"]}])) self.assertEqual(result.returncode, 0) data=json.loads(out.read_text()); data["claims"]["bad"]="/missing"; out.write_text(json.dumps(data, sort_keys=True, separators=(",", ":"))+"\n") self.assertNotEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", out.name).returncode, 0) def test_validate_rejects_forged_plan_exclusion(self): result, out = self.run_manifest() self.assertEqual(result.returncode, 0, result.stderr) (self.repo / "plan.json").unlink() (self.repo / "dirty.txt").write_text("dirty\n") data = json.loads(out.read_text()) data["excluded_paths"][0] = "dirty.txt" out.write_text(json.dumps(data, sort_keys=True, separators=(",", ":")) + "\n") self.assertNotEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", out.name).returncode, 0) def test_validate_rejects_empty_or_shell_argv(self): result, out = self.run_manifest() self.assertEqual(result.returncode, 0, result.stderr) data = json.loads(out.read_text()) data["commands"][0]["argv"] = [] out.write_text(json.dumps(data, sort_keys=True, separators=(",", ":")) + "\n") self.assertNotEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", out.name).returncode, 0) data["commands"][0]["argv"] = ["bash", "-c", "true"] out.write_text(json.dumps(data, sort_keys=True, separators=(",", ":")) + "\n") self.assertNotEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", out.name).returncode, 0) def test_evidence_only_descendant_validates(self): result, out = self.run_manifest(); self.assertEqual(result.returncode, 0) (self.repo / "evidence.txt").write_text("evidence\n"); run_git(self.repo, "add", "evidence.txt"); run_git(self.repo, "commit", "-qm", "evidence") self.assertEqual(self.invoke("validate", "--repo-root", self.repo, "--manifest", "manifest.json").returncode, 0) if __name__ == "__main__": unittest.main()