207 lines
12 KiB
Python
207 lines
12 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from harness.tests._support.paths import ROOT
|
|
from harness.tests._support.evidence import EvidenceManifestFixture
|
|
|
|
SCRIPT = ROOT / "harness/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.fixture = EvidenceManifestFixture().setup()
|
|
self.addCleanup(self.fixture.cleanup)
|
|
self.tmp = self.fixture.tmp
|
|
self.root = self.fixture.root
|
|
self.contract = self.fixture.contract
|
|
self.repo = self.fixture.repo
|
|
self.artifacts = self.fixture.artifacts
|
|
|
|
def invoke(self, *args, **kwargs):
|
|
return self.fixture.invoke(*args, **kwargs)
|
|
|
|
def plan(self, *args, **kwargs):
|
|
return self.fixture.plan(*args, **kwargs)
|
|
|
|
def run_manifest(self, *args, **kwargs):
|
|
return self.fixture.run_manifest(*args, **kwargs)
|
|
|
|
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', '/commands/0/timeout_seconds', '/commands/0/execution_status', '/commands/0/termination'])
|
|
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.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')
|
|
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', 'overflow.json')
|
|
self.assertEqual(result.returncode, 1)
|
|
overflow = json.loads((self.artifacts / 'overflow.json').read_text())
|
|
self.assertEqual(overflow['result'], 'failure')
|
|
self.assertEqual(overflow['commands'][0]['execution_status'], 'output_limit')
|
|
self.assertLessEqual(len(overflow['commands'][0]['output'].encode()), 16)
|
|
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', '../checkout/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.artifacts / '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_v2_validate_rejects_forged_plan_exclusion(self):
|
|
result, out = self.run_manifest()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
(self.artifacts / 'plan.json').unlink()
|
|
(self.repo / 'dirty.txt').write_text('dirty\n')
|
|
data = json.loads(out.read_text())
|
|
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)
|
|
|
|
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_archived_v1_descendant_validates_read_only(self):
|
|
result, out = self.run_manifest()
|
|
self.assertEqual(result.returncode, 0)
|
|
obj = json.loads(out.read_text())
|
|
obj['schema'] = 'evidence-manifest/v1'
|
|
del obj['artifact']
|
|
obj['excluded_paths'] = ['plan.json', 'manifest.json']
|
|
archived_plan = json.loads((self.artifacts / 'plan.json').read_text())
|
|
archived_plan['commands'][0].pop('timeout_seconds', None)
|
|
(self.repo / 'plan.json').write_text(json.dumps(archived_plan))
|
|
obj['plan']['sha256'] = hashlib.sha256((self.repo / 'plan.json').read_bytes()).hexdigest()
|
|
for key in ('timeout_seconds', 'execution_status', 'termination'):
|
|
obj['commands'][0].pop(key, None)
|
|
obj['claims'] = {key: pointers[:6] for key, pointers in obj['claims'].items()}
|
|
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()
|
|
archived_result = self.invoke('validate-legacy', '--repo-root', self.repo, '--manifest', 'manifest.json')
|
|
self.assertEqual(archived_result.returncode, 0, archived_result.stderr)
|
|
self.assertEqual(archived.read_bytes(), before)
|
|
self.assertEqual(run_git(self.repo, 'status', '--porcelain').stdout, '')
|