harness: Issue 검증 증빙을 checkout 밖에 보관

This commit is contained in:
2026-10-08 01:06:16 +00:00
parent e7bcfef532
commit 6ae993887c
8 changed files with 590 additions and 56 deletions
+271 -16
View File
@@ -16,9 +16,22 @@ def run_git(cwd, *args):
class EvidenceManifestTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.repo = Path(self.tmp.name) / "repo"
self.repo.mkdir()
self.tmp = tempfile.TemporaryDirectory(dir=os.environ.get("TMPDIR", "/opt/data/cache/scratch"))
self.root = Path(self.tmp.name)
self.contract = self.root / "harness/operations/repository-cache.md"
self.contract.parent.mkdir(parents=True)
self.contract.write_text("# Repository cache 운영 지침\n")
self.repo = self.root / "repositories/owner/project/worktrees/issue-1/checkout"
self.artifacts = self.repo.parent / "verification/run-a"
self.repo.mkdir(parents=True)
self.artifacts.mkdir(parents=True)
for path in (self.root / "repositories", *self.repo.parents):
if path == self.root:
break
path.chmod(0o700)
self.repo.chmod(0o700)
self.artifacts.parent.chmod(0o700)
self.artifacts.chmod(0o700)
run_git(self.repo, "init", "-q")
run_git(self.repo, "config", "user.email", "test@example.invalid")
run_git(self.repo, "config", "user.name", "Test")
@@ -29,17 +42,19 @@ class EvidenceManifestTests(unittest.TestCase):
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)
op, *rest = args
common = [] if op == "validate-legacy" else ["--contract-path", str(self.contract), "--artifact-root", str(self.artifacts)]
return subprocess.run([sys.executable, str(SCRIPT), str(op), *common, *map(str, rest)], 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
data.update(extra); p = self.artifacts / "plan.json"; p.write_text(json.dumps(data)); p.chmod(0o600); 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
out = self.artifacts / "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)
@@ -82,7 +97,7 @@ class EvidenceManifestTests(unittest.TestCase):
"claims": ["environment"],
}]
plan = self.plan(command)
out = self.repo / "manifest.json"
out = self.artifacts / "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")
@@ -99,13 +114,15 @@ class EvidenceManifestTests(unittest.TestCase):
"max_output_bytes": 16,
"claims": ["bounded"],
}])
result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", out.name)
result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", "overflow.json")
self.assertNotEqual(result.returncode, 0)
self.assertIn("exceeds max_output_bytes", result.stderr)
self.assertFalse((self.artifacts / "overflow.json").exists())
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")
result = self.invoke("run", "--repo-root", self.repo, "--plan", plan.name, "--output", "../checkout/input.txt")
self.assertNotEqual(result.returncode, 0)
self.assertEqual((self.repo / "input.txt").read_text(), "hello\n")
@@ -141,7 +158,7 @@ class EvidenceManifestTests(unittest.TestCase):
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())
self.assertNotEqual(result.returncode, 0); self.assertFalse((self.artifacts / "manifest.json").exists())
def test_source_drift_hash_mismatch_rejected(self):
result, out = self.run_manifest(); self.assertEqual(result.returncode, 0)
@@ -186,13 +203,13 @@ class EvidenceManifestTests(unittest.TestCase):
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):
def test_v2_validate_rejects_forged_plan_exclusion(self):
result, out = self.run_manifest()
self.assertEqual(result.returncode, 0, result.stderr)
(self.repo / "plan.json").unlink()
(self.artifacts / "plan.json").unlink()
(self.repo / "dirty.txt").write_text("dirty\n")
data = json.loads(out.read_text())
data["excluded_paths"][0] = "dirty.txt"
data["excluded_paths"] = ["dirty.txt", "manifest.json"]
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)
@@ -208,10 +225,248 @@ class EvidenceManifestTests(unittest.TestCase):
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):
def test_archived_v1_descendant_validates_read_only(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)
obj=json.loads(out.read_text())
obj["schema"]="evidence-manifest/v1"
del obj["artifact"]
obj["excluded_paths"]=["plan.json", "manifest.json"]
(self.repo / "plan.json").write_bytes((self.artifacts / "plan.json").read_bytes())
archived=self.repo / "manifest.json"
archived.write_text(json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)+"\n")
run_git(self.repo, "add", "plan.json", "manifest.json"); run_git(self.repo, "commit", "-qm", "historical evidence fixture")
before=archived.read_bytes()
self.assertEqual(self.invoke("validate-legacy", "--repo-root", self.repo, "--manifest", "manifest.json").returncode, 0)
self.assertEqual(archived.read_bytes(), before)
self.assertEqual(run_git(self.repo, "status", "--porcelain").stdout, "")
class ExternalEvidenceTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(dir=os.environ.get("TMPDIR", "/opt/data/cache/scratch"))
self.root = Path(self.tmp.name)
self.contract = self.root / "harness/operations/repository-cache.md"
self.contract.parent.mkdir(parents=True)
self.contract.write_text("# Repository cache 운영 지침\n", encoding="utf-8")
self.repo = self.root / "repositories/owner/project/worktrees/issue-1/checkout"
self.artifacts = self.repo.parent / "verification/run-a"
self.repo.mkdir(parents=True)
self.artifacts.mkdir(parents=True)
for path in (self.root / "repositories", *self.repo.parents):
if path == self.root:
break
path.chmod(0o700)
self.repo.chmod(0o700)
self.artifacts.parent.chmod(0o700)
self.artifacts.chmod(0o700)
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")
self.plan = self.artifacts / "evidence-plan.json"
self.plan.write_text(json.dumps({"hash_paths": ["input.txt"], "commands": [{"id": "check", "argv": ["git", "show", "HEAD:input.txt"], "claims": ["output"]}]}))
self.plan.chmod(0o600)
self.manifest = self.artifacts / "evidence-manifest.json"
def tearDown(self):
self.tmp.cleanup()
def invoke(self, op, *args):
return subprocess.run([sys.executable, str(SCRIPT), op, "--contract-path", str(self.contract), "--repo-root", str(self.repo), "--artifact-root", str(self.artifacts), *map(str, args)], text=True, capture_output=True)
def test_ignored_source_input_is_rejected_for_run_and_validate(self):
(self.repo / ".gitignore").write_text("ignored.txt\n")
run_git(self.repo, "add", ".gitignore")
run_git(self.repo, "commit", "-qm", "ignore fixture")
ignored=self.repo / "ignored.txt"
ignored.write_text("unbound input\n")
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("not clean", result.stderr)
self.assertFalse(self.manifest.exists())
ignored.unlink()
self.assertEqual(self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name).returncode, 0)
prior=self.manifest.read_bytes()
ignored.write_text("unbound input again\n")
result=self.invoke("validate", "--manifest", self.manifest.name)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("not clean", result.stderr)
self.assertEqual(self.manifest.read_bytes(), prior)
def test_canonical_control_directory_modes_match_installed_harness(self):
# The cache contract permits a trusted NFS root with group write and
# imposes no runtime-directory mode on read-only harness code parents.
self.root.chmod(0o770)
self.contract.parent.parent.chmod(0o777)
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(self.invoke("validate", "--manifest", self.manifest.name).returncode, 0)
self.assertEqual(self.root.stat().st_mode & 0o777, 0o770)
self.assertEqual(self.contract.parent.parent.stat().st_mode & 0o777, 0o777)
def test_same_filename_in_source_and_artifacts_is_not_a_collision(self):
result=self.invoke("run", "--plan", self.plan.name, "--output", "input.txt")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual((self.repo / "input.txt").read_text(), "hello\n")
self.assertEqual(self.invoke("validate", "--manifest", "input.txt").returncode, 0)
def test_real_linked_detached_worktree_preserves_git_registration(self):
seed=self.root / "seed"
self.repo.rename(seed)
bare=self.repo.parents[2] / "repository.git"
subprocess.run(["git", "clone", "--bare", str(seed), str(bare)], check=True, capture_output=True)
subprocess.run(["git", "--git-dir", str(bare), "worktree", "add", "--detach", str(self.repo), "HEAD"], check=True, capture_output=True)
subprocess.run(["git", "--git-dir", str(bare), "worktree", "lock", str(self.repo)], check=True, capture_output=True)
before=run_git(self.repo, "worktree", "list", "--porcelain").stdout
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(self.invoke("validate", "--manifest", self.manifest.name).returncode, 0)
self.assertEqual(run_git(self.repo, "worktree", "list", "--porcelain").stdout, before)
self.assertNotEqual(run_git(self.repo, "rev-parse", "--git-common-dir").stdout.strip(), ".git")
self.assertEqual(run_git(self.repo, "status", "--porcelain").stdout, "")
def test_canonical_root_identity_and_traversal_denied_before_execution(self):
targets=(self.repo / "docs/verification/run-a", self.root / "scratch/run-a", self.repo.parent.parent / "issue-2/verification/run-a", self.artifacts.parent / "../run-a")
for target in targets:
with self.subTest(target=str(target)):
result=self.invoke("run", "--artifact-root", target, "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
self.assertFalse(self.manifest.exists())
for filename in ("../escape.json", "/absolute.json", "a/../escape.json", "./plan.json", "plan.json/"):
with self.subTest(filename=filename):
result=self.invoke("run", "--plan", filename, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
result=self.invoke("run", "--contract-path", self.root / "wrong.md", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
def test_artifact_file_and_directory_symlinks_denied(self):
alias=self.artifacts / "alias.json"
alias.symlink_to(self.plan.name)
result=self.invoke("run", "--plan", alias.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
self.assertIn("unsafe path", result.stderr)
outside=self.root / "outside"
self.artifacts.rename(outside)
self.artifacts.symlink_to(outside, target_is_directory=True)
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
self.assertIn("symlink", result.stderr)
self.assertFalse((outside / self.manifest.name).exists())
def test_private_permissions_fail_closed_without_repair(self):
self.plan.chmod(0o644)
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
self.assertEqual(self.plan.stat().st_mode & 0o777, 0o644)
self.plan.chmod(0o600)
self.artifacts.chmod(0o755)
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
self.assertEqual(self.artifacts.stat().st_mode & 0o777, 0o755)
self.assertFalse(self.manifest.exists())
def test_wrong_uid_is_denied(self):
import importlib.util
from unittest.mock import patch
spec=importlib.util.spec_from_file_location("evidence_manifest_uid_test", SCRIPT)
assert spec is not None and spec.loader is not None
module=importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
with patch.object(module.os, "getuid", return_value=self.artifacts.stat().st_uid+1):
with self.assertRaisesRegex(ValueError, "ownership"):
module.safe_entry(self.artifacts, directory=True, mode=0o700)
def test_group_writable_cache_parent_requires_exclusive_role_membership(self):
import importlib.util
from types import SimpleNamespace
from unittest.mock import patch
spec=importlib.util.spec_from_file_location("evidence_manifest_group_test", SCRIPT)
assert spec is not None and spec.loader is not None
module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
parent=self.root / "repositories"
parent.chmod(0o775)
uid=parent.stat().st_uid; gid=parent.stat().st_gid
role=SimpleNamespace(pw_name="role", pw_uid=uid, pw_gid=gid)
foreign=SimpleNamespace(pw_name="foreign", pw_uid=uid+1, pw_gid=gid)
with patch.object(module.grp, "getgrgid", return_value=SimpleNamespace(gr_mem=[])), patch.object(module.pwd, "getpwall", return_value=[role]), patch.object(module.pwd, "getpwnam", return_value=role):
module.safe_repositories(parent)
with patch.object(module.grp, "getgrgid", return_value=SimpleNamespace(gr_mem=["foreign"])), patch.object(module.pwd, "getpwall", return_value=[role, foreign]), patch.object(module.pwd, "getpwnam", side_effect=lambda name: role if name=="role" else foreign):
with self.assertRaisesRegex(ValueError, "exclusive"):
module.safe_repositories(parent)
with patch.object(module.grp, "getgrgid", side_effect=KeyError("unresolved")):
with self.assertRaisesRegex(ValueError, "unverifiable"):
module.safe_repositories(parent)
self.assertEqual(parent.stat().st_mode & 0o777, 0o775)
def test_no_clobber_rejects_rerun_before_command_execution(self):
self.assertEqual(self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name).returncode, 0)
prior=self.manifest.read_bytes()
obj=json.loads(self.plan.read_text())
marker=self.root / "rerun-executed"
obj["commands"][0]["argv"]=[sys.executable, "-c", "from pathlib import Path; Path("+repr(str(marker))+").write_text('ran')"]
self.plan.write_text(json.dumps(obj))
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2)
self.assertIn("already exists", result.stderr)
self.assertEqual(self.manifest.read_bytes(), prior)
self.assertFalse(marker.exists())
def test_v2_rejects_forged_issue_binding_and_v1_normal_validation(self):
self.assertEqual(self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name).returncode, 0)
original=json.loads(self.manifest.read_text())
for replacement in (2, True, "1"):
obj=json.loads(json.dumps(original)); obj["artifact"]["issue"]=replacement
self.manifest.write_text(json.dumps(obj, sort_keys=True, separators=(",", ":"))+"\n")
self.assertEqual(self.invoke("validate", "--manifest", self.manifest.name).returncode, 2)
original["schema"]="evidence-manifest/v1"
self.manifest.write_text(json.dumps(original, sort_keys=True, separators=(",", ":"))+"\n")
self.assertEqual(self.invoke("validate", "--manifest", self.manifest.name).returncode, 2)
def test_artifact_boundary_mutation_by_command_rejected_before_write(self):
obj=json.loads(self.plan.read_text())
obj["commands"][0]["argv"]=[sys.executable, "-c", "from pathlib import Path; Path("+repr(str(self.artifacts))+").chmod(0o755)"]
self.plan.write_text(json.dumps(obj))
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("permission", result.stderr)
self.assertFalse(self.manifest.exists())
def test_v2_stale_head_rejected_even_when_bound_file_is_unchanged(self):
self.assertEqual(self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name).returncode, 0)
(self.repo / "other.txt").write_text("new source revision\n")
run_git(self.repo, "add", "other.txt")
run_git(self.repo, "commit", "-qm", "advance head")
result=self.invoke("validate", "--manifest", self.manifest.name)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("source revision is not current HEAD", result.stderr)
def test_v2_rejects_source_symlink_alias(self):
(self.repo / "alias.txt").symlink_to("input.txt")
run_git(self.repo, "add", "alias.txt")
run_git(self.repo, "commit", "-qm", "source symlink fixture")
obj=json.loads(self.plan.read_text())
obj["hash_paths"]=["alias.txt"]
self.plan.write_text(json.dumps(obj))
result=self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("symlink", result.stderr)
self.assertFalse(self.manifest.exists())
def test_external_artifacts_leave_checkout_and_head_unchanged(self):
head = run_git(self.repo, "rev-parse", "HEAD").stdout.strip()
result = self.invoke("run", "--plan", self.plan.name, "--output", self.manifest.name)
self.assertEqual(result.returncode, 0, result.stderr)
data = json.loads(self.manifest.read_text())
self.assertEqual(data["schema"], "evidence-manifest/v2")
self.assertEqual(data["source_revision"], head)
self.assertEqual(data["artifact"]["run_id"], "run-a")
self.assertEqual(data["artifact"]["issue"], 1)
self.assertEqual(run_git(self.repo, "status", "--porcelain").stdout, "")
self.assertEqual(run_git(self.repo, "rev-parse", "HEAD").stdout.strip(), head)
self.assertEqual(self.manifest.stat().st_mode & 0o777, 0o600)
self.assertEqual(self.invoke("validate", "--manifest", self.manifest.name).returncode, 0)
if __name__ == "__main__": unittest.main()