diff --git a/harness/agents/impl/skills/gitea-issue-rework/SKILL.md b/harness/agents/impl/skills/gitea-issue-rework/SKILL.md index cc85de1..fe609a1 100644 --- a/harness/agents/impl/skills/gitea-issue-rework/SKILL.md +++ b/harness/agents/impl/skills/gitea-issue-rework/SKILL.md @@ -1,7 +1,7 @@ --- name: gitea-issue-rework description: Gitea PR 검토 지적을 계약에 맞춰 재작업하고 검증한다. -version: 0.2.6 +version: 0.2.7 author: genichin, Hermes Agent license: MIT platforms: [linux] @@ -163,6 +163,10 @@ Reviewer가 rework head를 새 snapshot으로 다시 검토해야 하며 이 com ## 절차 +### Evidence manifest automation + +모든 baseline, 통합·precommit 및 readback 검증은 `scripts/evidence_manifest.py`와 `templates/evidence-plan.json`을 사용해 exact argv, exit code, 제한된 output, 허용된 환경, source revision, 파일 hash와 표준 JSON-pointer claims를 기록한다. 명령은 argv 배열만 허용하고 명시된 최소 child-environment allowlist에서 실행하며 output 상한을 넘는 process는 중단한다. Secret·header·private-key·password 값이나 plan 금지 패턴이 발견되면 redaction하지 않고 fail-closed한다. `run`은 baseline부터 최종 precommit까지 사용하고, `validate`는 plan과 manifest를 제외한 clean tree에서 PR claim 생성 전에 실행한다. `validate`는 source ancestry와 hash를 다시 검증하며 evidence-only descendant에서는 기록된 hash path가 source blob과 동일해야 한다. 수동으로 매칭되지 않는 PR 성공·검증 주장을 만들지 않는다. Comment publish 전 remote immutable artifact의 digest를 authoritative readback으로 비교한다. + ### 1. Target, identity와 review 고정 1. Exact target identifier를 검증하고 Issue lock을 획득한다. diff --git a/harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py b/harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py new file mode 100644 index 0000000..bb29395 --- /dev/null +++ b/harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py @@ -0,0 +1,319 @@ +#!/usr/bin/env python3 +"""Create and validate bounded, fail-closed evidence manifests.""" +from __future__ import annotations +import argparse, datetime as dt, hashlib, json, math, os, platform, re, shutil, subprocess, sys, tempfile, time +from pathlib import Path + +HEX40 = re.compile(r"^[0-9a-f]{40}$") +HEX64 = re.compile(r"^[0-9a-f]{64}$") +ID = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{0,127}$") +ISO = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d+)?Z$") +SAFE_ENV = ("os", "kernel", "architecture", "python", "uv") +SECRET_PATTERNS = [ + re.compile(r"(?:cookie|set-cookie|authorization|proxy-authorization|x-api-key|api-key|x-auth-token)\s*:", re.I), + re.compile(r"\bbearer\s+[^\s]+", re.I), + re.compile(r"-----BEGIN [^-]*PRIVATE KEY-----", re.I), + re.compile(r"\b(?:password|passwd|secret|token)\s*[:=]", re.I), + re.compile(r"(?:^|\s)--?(?:password|passwd|secret|token)(?:\s+|=)\S+", re.I), +] +SAFE_CHILD_ENV = ("PATH", "HOME", "TMPDIR", "TMP", "TEMP", "LANG", "LC_ALL", "VIRTUAL_ENV", "UV_CACHE_DIR") +SHELL_NAMES = {"sh", "bash", "dash", "zsh", "ksh", "fish", "cmd", "cmd.exe", "powershell", "powershell.exe", "pwsh", "pwsh.exe"} + + +def die(msg): + print(str(msg), file=sys.stderr); return 2 + + +def git(root: Path, *args, check=True): + p = subprocess.run(["git", f"-c", f"safe.directory={root}", *args], cwd=root, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if check and p.returncode: raise ValueError(p.stderr.decode("utf-8", "replace").strip() or "git failed") + return p + + +def inside(root, value, *, allow_absolute=False): + if not isinstance(value, str) or not value or (Path(value).is_absolute() and not allow_absolute): + raise ValueError("paths must be nonempty repository-relative paths") + p = Path(value) + if ".." in p.parts: raise ValueError("path traversal is forbidden") + if p.parts and p.parts[0] == ".git": raise ValueError(".git paths are forbidden") + candidate = (root / p).resolve() + try: candidate.relative_to(root.resolve()) + except ValueError: raise ValueError(f"path outside repository: {value}") + if (root / p).is_symlink() and not candidate.is_relative_to(root.resolve()): raise ValueError("symlink escapes repository") + return candidate, candidate.relative_to(root.resolve()).as_posix() + + +def utc_now(): return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z") +def canonical(obj): return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n" +def text_bytes(b): return b.decode("utf-8", "replace") + + +def scan_sensitive(text, secrets, forbidden): + for s in secrets: + if s and s in text: return "secret value" + for pattern in SECRET_PATTERNS + forbidden: + if pattern.search(text): return f"forbidden pattern: {pattern.pattern}" + return None + + +def validate_regexes(values): + if not isinstance(values, list) or any(not isinstance(x, str) for x in values): raise ValueError("forbidden_regexes must be strings") + try: return [re.compile(x) for x in values] + except re.error as e: raise ValueError(f"invalid forbidden regex: {e}") + + +def validate_argv(argv): + if not isinstance(argv,list) or not argv or any(not isinstance(x,str) for x in argv): raise ValueError("commands require nonempty argv string arrays") + executable=Path(argv[0]).name.lower() + if executable in SHELL_NAMES or any(x.lower() in ("shell", "shell=true") for x in argv): raise ValueError("shell commands are forbidden") + + +def run_bounded(argv, root, env, max_bytes): + proc=subprocess.Popen(argv,cwd=root,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,shell=False,env=env,bufsize=0) + stream=proc.stdout + if stream is None: raise ValueError("command output pipe unavailable") + data=bytearray() + try: + while True: + chunk=stream.read(min(65536,max_bytes-len(data)+1)) + if not chunk: break + data.extend(chunk) + if len(data)>max_bytes: + proc.kill(); proc.wait() + raise ValueError("command output exceeds max_output_bytes") + return proc.wait(),bytes(data) + finally: + stream.close() + if proc.poll() is None: + proc.kill(); proc.wait() + + +def status_paths(root): + raw = git(root, "status", "--porcelain=v1", "-z").stdout + out=[] + for entry in raw.split(b"\0"): + if not entry: continue + path = entry[3:] if len(entry) >= 3 else b"" + if entry[:2] in (b"R ", b" R", b"R?"): + path = path.split(b"\0",1)[-1] + out.append(path.decode("utf-8", "surrogateescape")) + return out + + +def check_clean(root, allowed): + bad = [p for p in status_paths(root) if p not in allowed] + if bad: raise ValueError("source tree is not clean: " + ", ".join(bad)) + + +def plan_checks(plan, root): + if not isinstance(plan, dict): raise ValueError("plan must be an object") + for key in ("hash_paths", "commands", "limitations", "secret_env_names", "forbidden_regexes", "command_env_allowlist"): + if key in plan and not isinstance(plan[key], list): raise ValueError(f"{key} must be a list") + names = plan.get("secret_env_names", []) + if any(not isinstance(n, str) for n in names): raise ValueError("secret_env_names must be strings") + secrets = [os.environ.get(n, "") for n in names if os.environ.get(n, "")] + forbidden = validate_regexes(plan.get("forbidden_regexes", [])) + requested = plan.get("environment", list(SAFE_ENV[:4])) + if not isinstance(requested, list) or len(set(requested)) != len(requested) or any(x not in SAFE_ENV for x in requested): raise ValueError("invalid environment allowlist") + child_names=plan.get("command_env_allowlist",list(SAFE_CHILD_ENV)) + if any(not isinstance(x,str) or not x or x not in SAFE_CHILD_ENV for x in child_names): raise ValueError("command_env_allowlist may contain only safe operational names") + child_names=list(dict.fromkeys([*child_names,*names])) + hash_paths=plan.get("hash_paths",[]) + if not hash_paths or any(not isinstance(x,str) or not x for x in hash_paths): raise ValueError("hash_paths must be a nonempty string list") + command_plans=plan.get("commands",[]) + if not command_plans: raise ValueError("commands must be nonempty") + seen=set(); command_ids=set(); commands=[] + for index,spec in enumerate(command_plans): + if not isinstance(spec, dict): raise ValueError("command must be object") + argv=spec.get("argv"); validate_argv(argv) + expected=spec.get("expected_exit_code",0); maxb=spec.get("max_output_bytes",65536) + if not isinstance(expected,int) or isinstance(expected,bool) or not isinstance(maxb,int) or maxb < 0: raise ValueError("invalid command bounds") + claims=spec.get("claims",[]) + if not isinstance(claims,list) or not claims or any(not isinstance(x,str) or not ID.fullmatch(x) for x in claims): raise ValueError("claims must be a nonempty list of valid IDs") + if any(x in seen for x in claims): raise ValueError("duplicate claim ID") + command_id=spec.get("id",f"command-{index}"); purpose=spec.get("purpose","") + if not isinstance(command_id,str) or not ID.fullmatch(command_id) or command_id in command_ids: raise ValueError("invalid or duplicate command ID") + if not isinstance(purpose,str): raise ValueError("command purpose must be a string") + command_ids.add(command_id); seen.update(claims); commands.append((spec,argv,expected,maxb,claims)) + return secrets, forbidden, requested, child_names, commands + + +def tracked_blob(root, source, rel): + info=git(root,"ls-tree",source,"--",rel).stdout.decode("utf-8","replace").strip().split() + if len(info)<3 or info[1] not in ("blob",) or info[0] not in ("100644","100755"): raise ValueError(f"hash path is not tracked regular file: {rel}") + oid=info[2]; blob=git(root,"cat-file","blob",oid).stdout + return oid, hashlib.sha256(blob).hexdigest() + + +def atomic_write(path, data): + if not path.parent.is_dir(): raise ValueError("output parent must already exist") + fd,tmp=tempfile.mkstemp(prefix="."+path.name+".", dir=path.parent) + try: + with os.fdopen(fd,"w",encoding="utf-8",newline="") as f: + f.write(data); f.flush(); os.fsync(f.fileno()) + os.replace(tmp,path) + dir_fd = os.open(path.parent, os.O_RDONLY) + try: os.fsync(dir_fd) + finally: os.close(dir_fd) + finally: + try: os.unlink(tmp) + except FileNotFoundError: pass + + +def run(args): + root=Path(args.repo_root).resolve(); plan_path,plan_rel=inside(root,args.plan); out,out_rel=inside(root,args.output) + if plan_path.suffix.lower()!=".json": raise ValueError("plan must be a JSON file") + if not out.parent.is_dir(): raise ValueError("output parent must already exist") + plan_bytes=plan_path.read_bytes(); plan_sha256=hashlib.sha256(plan_bytes).hexdigest() + plan=json.loads(plan_bytes.decode("utf-8")); secrets,forbidden,env_keys,child_names,commandspec=plan_checks(plan,root) + for _,argv,_,_,_ in commandspec: + violation=scan_sensitive("\n".join(argv),secrets,forbidden) + if violation: raise ValueError("sensitive command rejected before execution: "+violation) + check_clean(root,{plan_rel,out_rel}) + source=git(root,"rev-parse","HEAD").stdout.decode().strip() + if not HEX40.fullmatch(source): raise ValueError("source revision is not full SHA") + files=[] + for item in plan.get("hash_paths",[]): + path,rel=inside(root,item) + if not path.is_file() or path.is_symlink(): raise ValueError(f"hash path is not a regular file: {item}") + oid,sha=tracked_blob(root,source,rel) + if hashlib.sha256(path.read_bytes()).hexdigest()!=sha: raise ValueError(f"source hash mismatch: {rel}") + files.append({"path":rel,"sha256":sha,"blob_oid":oid}) + bound_paths={record["path"] for record in files} + if plan_rel==out_rel or plan_rel in bound_paths or out_rel in bound_paths: raise ValueError("plan/output paths must not overlap bound source paths") + child_env={name:os.environ[name] for name in child_names if name in os.environ} + result_commands=[]; claim_map={} + for i,(spec,argv,expected,maxb,claim_ids) in enumerate(commandspec): + started=utc_now(); t=time.monotonic(); returncode,raw=run_bounded(argv,root,child_env,maxb); finished=utc_now() + output=text_bytes(raw); violation=scan_sensitive("\n".join(argv)+"\n"+output,secrets,forbidden) + if violation: raise ValueError("sensitive evidence rejected: "+violation) + c={"id":spec.get("id",f"command-{i}"),"purpose":spec.get("purpose",""),"argv":argv,"exit_code":returncode,"expected_exit_code":expected,"max_output_bytes":maxb,"output":output,"started_at_utc":started,"finished_at_utc":finished,"duration_seconds":time.monotonic()-t,"claim_ids":claim_ids} + result_commands.append(c) + for cid in claim_ids: + claim_map[cid]=[ + f"/commands/{i}/argv", + f"/commands/{i}/exit_code", + f"/commands/{i}/expected_exit_code", + f"/commands/{i}/output", + f"/commands/{i}/started_at_utc", + f"/commands/{i}/finished_at_utc", + ] + env={k:None for k in env_keys} + if "os" in env: env["os"]=platform.system() + if "kernel" in env: env["kernel"]=platform.release() + if "architecture" in env: env["architecture"]=platform.machine() + if "python" in env: env["python"]=platform.python_version() + if "uv" in env: + p=subprocess.run(["uv","--version"],stdout=subprocess.PIPE,stderr=subprocess.PIPE) if shutil.which("uv") else None + env["uv"]=text_bytes((p.stdout+p.stderr).strip()) if p and p.returncode==0 else None + # Commands are evidence producers, not trusted source mutators. Recheck the + # tree and every bound blob before persisting their claims. + check_clean(root,{plan_rel,out_rel}) + if git(root,"rev-parse","HEAD").stdout.decode().strip()!=source: raise ValueError("HEAD changed during verification") + if not plan_path.is_file() or hashlib.sha256(plan_path.read_bytes()).hexdigest()!=plan_sha256: raise ValueError("plan changed during verification") + for record in files: + if hashlib.sha256((root/record["path"]).read_bytes()).hexdigest()!=record["sha256"]: + raise ValueError(f"source changed during verification: {record['path']}") + success=all(c["exit_code"]==c["expected_exit_code"] for c in result_commands) + manifest={"schema":"evidence-manifest/v1","source_revision":source,"source_tree_clean":True,"excluded_paths":[plan_rel,out_rel],"plan":{"path":plan_rel,"sha256":plan_sha256},"generated_at":utc_now(),"environment":env,"commands":result_commands,"files":files,"claims":claim_map,"redaction":{"checked":True,"violations":[]},"limitations":plan.get("limitations",[]),"result":"success" if success else "failure"} + atomic_write(out,canonical(manifest)) + return 0 if success else 1 + + +def pointer(obj,ptr): + if not isinstance(ptr,str) or not ptr.startswith("/"): raise ValueError("invalid claim pointer") + cur=obj + for part in ptr[1:].split("/"): + part=part.replace("~1","/").replace("~0","~") + if isinstance(cur,list) and part.isdigit() and int(part)c["max_output_bytes"]: raise ValueError("invalid command bounds") + if not isinstance(c["duration_seconds"],(int,float)) or isinstance(c["duration_seconds"],bool) or not math.isfinite(c["duration_seconds"]) or c["duration_seconds"]<0: raise ValueError("invalid command duration") + if not isinstance(c["claim_ids"],list) or any(not isinstance(x,str) or not ID.fullmatch(x) for x in c["claim_ids"]): raise ValueError("invalid command fields") + if c["id"]!=spec.get("id",f"command-{i}") or c["purpose"]!=spec.get("purpose","") or c["argv"]!=planned_argv or c["expected_exit_code"]!=planned_exit or c["max_output_bytes"]!=planned_max or c["claim_ids"]!=planned_claims: raise ValueError("command record does not match plan") + if not ISO.fullmatch(c["started_at_utc"]) or not ISO.fullmatch(c["finished_at_utc"]): raise ValueError("invalid timestamp") + for cid in c["claim_ids"]: + if cid in claim_seen or cid not in obj["claims"]: raise ValueError("duplicate or missing claim") + claim_seen.add(cid) + pointers=obj["claims"][cid] + expected_pointers=[ + f"/commands/{i}/argv", + f"/commands/{i}/exit_code", + f"/commands/{i}/expected_exit_code", + f"/commands/{i}/output", + f"/commands/{i}/started_at_utc", + f"/commands/{i}/finished_at_utc", + ] + if pointers!=expected_pointers: raise ValueError("claim pointers are not canonical") + for ptr in pointers: pointer(obj,ptr) + violation=scan_sensitive("\n".join(c["argv"])+"\n"+c["output"],plan_secrets,plan_forbidden) + if violation: raise ValueError("redaction violation") + if claim_seen != set(obj["claims"]): raise ValueError("unreferenced claim") + expected="success" if all(c["exit_code"]==c["expected_exit_code"] for c in obj["commands"]) else "failure" + if obj["result"]!=expected: raise ValueError("result inconsistent") + return 0 + + +def main(): + p=argparse.ArgumentParser(); sub=p.add_subparsers(dest="op",required=True) + r=sub.add_parser("run"); r.add_argument("--repo-root",required=True); r.add_argument("--plan",required=True); r.add_argument("--output",required=True) + v=sub.add_parser("validate"); v.add_argument("--repo-root",required=True); v.add_argument("--manifest",required=True) + a=p.parse_args() + try: return run(a) if a.op=="run" else validate(a) + except (OSError, ValueError, TypeError, KeyError, AttributeError, UnicodeError, json.JSONDecodeError, re.error, subprocess.SubprocessError) as e: return die(e) +if __name__=="__main__": sys.exit(main()) diff --git a/harness/agents/impl/skills/gitea-issue-rework/templates/evidence-plan.json b/harness/agents/impl/skills/gitea-issue-rework/templates/evidence-plan.json new file mode 100644 index 0000000..790d763 --- /dev/null +++ b/harness/agents/impl/skills/gitea-issue-rework/templates/evidence-plan.json @@ -0,0 +1,18 @@ +{ + "environment": ["os", "kernel", "architecture", "python"], + "command_env_allowlist": ["PATH", "HOME", "TMPDIR", "LANG", "VIRTUAL_ENV", "UV_CACHE_DIR"], + "secret_env_names": ["PROJECT_TEST_TOKEN"], + "forbidden_regexes": [], + "hash_paths": ["src/project.py", "tests/test_project.py"], + "commands": [ + { + "id": "focused-tests", + "purpose": "Run the focused verification", + "argv": ["python3", "-m", "unittest", "tests.test_project", "-v"], + "expected_exit_code": 0, + "max_output_bytes": 65536, + "claims": ["focused-result"] + } + ], + "limitations": [] +} diff --git a/harness/tests/impl/test_evidence_manifest.py b/harness/tests/impl/test_evidence_manifest.py new file mode 100644 index 0000000..5b5cb3e --- /dev/null +++ b/harness/tests/impl/test_evidence_manifest.py @@ -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()