209 lines
12 KiB
Python
209 lines
12 KiB
Python
import copy
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from harness.lib import delegation_adoption as adoption
|
|
|
|
|
|
class DelegationAdoptionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(dir="/opt/data/cache/scratch")
|
|
self.base = Path(self.tmp.name)
|
|
self.bundle = self.base / "bundle"
|
|
(self.bundle / "src").mkdir(parents=True)
|
|
(self.bundle / "evidence").mkdir()
|
|
(self.bundle / "src/value.py").write_bytes(b"VALUE = 1\n")
|
|
(self.bundle / "evidence/run.txt").write_bytes(b"run\n")
|
|
self.files = {"src/value.py": hashlib.sha256((self.bundle / "src/value.py").read_bytes()).hexdigest()}
|
|
self.bundle_sha = self.digest(self.files)
|
|
self.context = {
|
|
"schema": "gitea-agent.delegation-evidence-context/v1",
|
|
"source": {"kind": "captured_worktree", "base_revision": "a" * 40},
|
|
"stage": "implementation",
|
|
"bundle_sha256": self.bundle_sha,
|
|
"files": self.files,
|
|
"allowed_actions": ["static"],
|
|
"obligations": [{"id": "static", "kind": "static"}],
|
|
"executions": [],
|
|
}
|
|
self.result = {
|
|
"schema": "gitea-agent.delegation-evidence-result/v1",
|
|
"source": copy.deepcopy(self.context["source"]),
|
|
"stage": self.context["stage"],
|
|
"bundle_sha256": self.bundle_sha,
|
|
"claims": [{"id": "static", "status": "supported", "anchors": [{
|
|
"kind": "lines", "path": "src/value.py", "start_line": 1,
|
|
"end_line": 1, "text_sha256": hashlib.sha256(b"VALUE = 1\n").hexdigest(),
|
|
}], "execution_ids": []}],
|
|
}
|
|
self.context_path = self.base / "context.json"
|
|
self.result_path = self.base / "result.json"
|
|
self.context_path.write_text(json.dumps(self.context))
|
|
self.result_path.write_text(json.dumps(self.result))
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
@staticmethod
|
|
def digest(value):
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
def declaration(self):
|
|
return {
|
|
"mode": "delegated",
|
|
"bundle_root": str(self.bundle),
|
|
"context_path": str(self.context_path),
|
|
"result_path": str(self.result_path),
|
|
"git_dir": None,
|
|
}
|
|
|
|
def test_valid_delegated_declaration_returns_fresh_pass_result(self):
|
|
expected_source = copy.deepcopy(self.context["source"])
|
|
out = adoption.require_adoption(
|
|
self.declaration(), expected_source=expected_source, expected_stage="implementation"
|
|
)
|
|
self.assertEqual(out["status"], "pass")
|
|
self.assertEqual(out["counts"]["obligations"], 1)
|
|
|
|
def test_valid_git_tree_declaration_uses_expected_revision(self):
|
|
repo = self.base / "repo"
|
|
repo.mkdir()
|
|
(repo / "src").mkdir()
|
|
(repo / "src/value.py").write_bytes(b"VALUE = 1\n")
|
|
subprocess.run(["/usr/bin/git", "init", "-q", str(repo)], check=True)
|
|
subprocess.run(["/usr/bin/git", "-C", str(repo), "config", "user.email", "test@example.invalid"], check=True)
|
|
subprocess.run(["/usr/bin/git", "-C", str(repo), "config", "user.name", "Test"], check=True)
|
|
subprocess.run(["/usr/bin/git", "-C", str(repo), "add", "."], check=True)
|
|
subprocess.run(["/usr/bin/git", "-C", str(repo), "commit", "-q", "-m", "fixture"], check=True)
|
|
revision = subprocess.check_output(["/usr/bin/git", "-C", str(repo), "rev-parse", "HEAD"], text=True).strip()
|
|
context = copy.deepcopy(self.context)
|
|
context["source"] = {"kind": "git_tree", "revision": revision}
|
|
result = copy.deepcopy(self.result)
|
|
result["source"] = copy.deepcopy(context["source"])
|
|
self.context_path.write_text(json.dumps(context))
|
|
self.result_path.write_text(json.dumps(result))
|
|
declaration = self.declaration()
|
|
declaration["git_dir"] = str(repo / ".git")
|
|
out = adoption.require_adoption(declaration, expected_source=context["source"], expected_stage="implementation")
|
|
self.assertEqual(out["status"], "pass")
|
|
|
|
def test_explicit_none_is_the_only_non_delegated_success(self):
|
|
self.assertIsNone(adoption.require_adoption(
|
|
{"mode": "none"}, expected_source=self.context["source"], expected_stage="implementation"
|
|
))
|
|
|
|
def test_missing_declaration_is_rejected_with_sanitized_error(self):
|
|
with self.assertRaisesRegex(adoption.AdoptionError, "^delegation adoption rejected$"):
|
|
adoption.require_adoption(None, expected_source=self.context["source"], expected_stage="implementation")
|
|
|
|
def test_unsupported_or_extra_declaration_fields_are_rejected(self):
|
|
for declaration in (
|
|
{"mode": "worker"},
|
|
{"mode": "none", "bundle_root": str(self.bundle)},
|
|
{**self.declaration(), "extra": 1},
|
|
):
|
|
with self.subTest(declaration=declaration):
|
|
with self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(declaration, expected_source=self.context["source"], expected_stage="implementation")
|
|
|
|
def test_expected_source_and_stage_are_independent_and_exact(self):
|
|
for source, stage in (
|
|
({"kind": "captured_worktree", "base_revision": "b" * 40}, "implementation"),
|
|
(self.context["source"], "review"),
|
|
):
|
|
with self.subTest(source=source, stage=stage), self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=source, expected_stage=stage)
|
|
|
|
def test_stale_context_result_or_bundle_is_rejected(self):
|
|
for field, value in (("stage", "review"), ("bundle_sha256", "0" * 64)):
|
|
context = copy.deepcopy(self.context)
|
|
context[field] = value
|
|
self.context_path.write_text(json.dumps(context))
|
|
with self.subTest(field=field), self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
self.context_path.write_text(json.dumps(self.context))
|
|
self.result_path.write_text(json.dumps({**self.result, "stage": "review"}))
|
|
with self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
self.result_path.write_text(json.dumps(self.result))
|
|
(self.bundle / "src/value.py").write_bytes(b"VALUE = 2\n")
|
|
with self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
|
|
def test_empty_obligations_or_claims_cannot_be_adopted(self):
|
|
for context, result in (
|
|
({**self.context, "obligations": []}, self.result),
|
|
(self.context, {**self.result, "claims": []}),
|
|
):
|
|
self.context_path.write_text(json.dumps(context))
|
|
self.result_path.write_text(json.dumps(result))
|
|
with self.subTest(context=context, result=result), self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
self.context_path.write_text(json.dumps(self.context))
|
|
self.result_path.write_text(json.dumps(self.result))
|
|
|
|
def test_block_and_verifier_exception_fail_closed(self):
|
|
with mock.patch.object(adoption.delegation_evidence, "verify_evidence", return_value={"status": "block"}):
|
|
with self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
with mock.patch.object(adoption.delegation_evidence, "verify_evidence", side_effect=RuntimeError("secret")):
|
|
with self.assertRaisesRegex(adoption.AdoptionError, "^delegation adoption rejected$"):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
|
|
def test_verifier_is_called_again_after_a_prior_pass(self):
|
|
with mock.patch.object(adoption.delegation_evidence, "verify_evidence", wraps=adoption.delegation_evidence.verify_evidence) as verifier:
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
self.assertEqual(verifier.call_count, 2)
|
|
|
|
def test_malformed_expected_inputs_and_paths_fail_closed(self):
|
|
cases = (
|
|
({"kind": "captured_worktree", "base_revision": "A" * 40}, "implementation"),
|
|
(self.context["source"], ""),
|
|
(self.context["source"], None),
|
|
)
|
|
for source, stage in cases:
|
|
with self.subTest(source=source, stage=stage), self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption({"mode": "none"}, expected_source=source, expected_stage=stage)
|
|
for key, value in (("bundle_root", str(self.base / ".." / self.base.name / "bundle")),
|
|
("context_path", "relative.json"), ("result_path", str(self.result_path) + "\x00"),
|
|
("git_dir", 3)):
|
|
declaration = self.declaration()
|
|
declaration[key] = value
|
|
with self.subTest(key=key), self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(declaration, expected_source=self.context["source"], expected_stage="implementation")
|
|
|
|
def test_all_git_declaration_paths_reject_symlinks_at_runtime(self):
|
|
from harness.tests.common.test_delegation_consumers_integration import RealEvidence
|
|
evidence = RealEvidence(self.base / 'safe-path-fixture')
|
|
declaration = evidence.declaration()
|
|
for field in ('bundle_root', 'context_path', 'result_path', 'git_dir'):
|
|
alias = self.base / ('alias-' + field)
|
|
alias.symlink_to(declaration[field], target_is_directory=field in ('bundle_root', 'git_dir'))
|
|
bad = {**declaration, field: str(alias)}
|
|
with self.subTest(field=field), self.assertRaisesRegex(adoption.AdoptionError, '^delegation adoption rejected$'):
|
|
adoption.require_adoption(bad, expected_source=evidence.context['source'], expected_stage='spec-review')
|
|
alias_parent = self.base / 'alias-parent'
|
|
alias_parent.symlink_to(evidence.root, target_is_directory=True)
|
|
with self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption({**declaration, 'context_path': str(alias_parent / 'context.json')},
|
|
expected_source=evidence.context['source'], expected_stage='spec-review')
|
|
|
|
def test_duplicate_json_and_missing_files_fail_closed(self):
|
|
self.context_path.write_text('{"schema":"x","schema":"y"}')
|
|
with self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
self.context_path.write_text(json.dumps(self.context))
|
|
self.result_path.unlink()
|
|
with self.assertRaises(adoption.AdoptionError):
|
|
adoption.require_adoption(self.declaration(), expected_source=self.context["source"], expected_stage="implementation")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|