feat(impl): automate rework evidence manifests
This commit is contained in:
@@ -0,0 +1,217 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user