Digital Drosophila Archive
209 lines
12 KiB
Python
209 lines
12 KiB
Python
"""Temporary Git repositories only; production verification performs no writes."""
|
|
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
import verify_git_payload as v
|
|
|
|
|
|
def data(value):
|
|
return (json.dumps(value, sort_keys=True, indent=2) + '\n').encode()
|
|
|
|
|
|
def row(name, raw):
|
|
return dict(path=name, bytes=len(raw), sha256=v.sha(raw))
|
|
|
|
|
|
class TemporaryGit(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.repo = Path(self.temp.name).resolve()
|
|
self.git('init', '-q')
|
|
self.git('config', 'user.name', 'Synthetic Test')
|
|
self.git('config', 'user.email', 'synthetic@example.invalid')
|
|
self.git('config', 'core.autocrlf', 'false')
|
|
self.pin = self.fixture()
|
|
self.stage()
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def git(self, *args, input=None):
|
|
result = subprocess.run(['git', '-C', str(self.repo), *args], input=input,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode('utf-8', 'replace'))
|
|
return result.stdout
|
|
|
|
def write(self, path, raw):
|
|
target = self.repo / path; target.parent.mkdir(parents=True, exist_ok=True); target.write_bytes(raw)
|
|
|
|
def stage(self):
|
|
self.git('add', '--', v.MODEL, v.REGISTRY)
|
|
|
|
def fixture(self, status='PASS-EXACT-LIGHT-CONTINUATION-ENGINEERING-NOT-LEARNING'):
|
|
payload, source_rows = {}, []
|
|
source_sha, input_sha = {}, {}
|
|
for name in v.SOURCES:
|
|
raw = ('# synthetic source ' + name + '\n').encode()
|
|
sealed = 'protocol/source-snapshot/' + name
|
|
payload[sealed] = raw
|
|
self.write(v.MODEL + '/' + name, raw)
|
|
source_sha[v.PROJECT + '/' + v.MODEL + '/' + name] = v.sha(raw)
|
|
input_sha[v.PROJECT + '/' + v.EXPERIMENT + '/' + sealed] = v.sha(raw)
|
|
source_rows.append(dict(model_source=v.MODEL + '/' + name, sealed_source=v.EXPERIMENT + '/' + sealed,
|
|
copy='source-snapshot/' + name, bytes=len(raw), sha256=v.sha(raw)))
|
|
payload['protocol/preregistered.json'] = data(dict(schema='phase4an-engineering-registration-v1', training=False,
|
|
source_sha256=source_sha, input_sha256=input_sha))
|
|
payload['audit/final-status.json'] = data(dict(status=status))
|
|
payload['audit/engineering-analysis.json'] = data(dict(status=status))
|
|
payload['audit/small.json'] = data({'synthetic': True})
|
|
payload['runs/a/state.npz'] = b'not copied bulk'
|
|
payload['runs/a/queue.jsonl'] = b'not copied jsonl\n'
|
|
payload['extension-build/provenance.json'] = b'not copied build'
|
|
rows = [row(name, raw) for name, raw in sorted(payload.items())]
|
|
manifest = data(dict(root=v.PROJECT + '/' + v.EXPERIMENT, files=rows))
|
|
checksums = ''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in rows)
|
|
checksums = (checksums + v.sha(manifest) + ' manifest.json\n').encode()
|
|
all_data = {**payload, 'manifest.json': manifest, 'checksums.sha256': checksums}
|
|
index, omitted = [], []
|
|
for name, raw in all_data.items():
|
|
current = row(name, raw); destination, reason = v._selection(current)
|
|
if destination is None:
|
|
omitted.append(dict(current, reason=reason))
|
|
else:
|
|
self.write(v.REGISTRY + '/' + destination, raw)
|
|
index.append(dict(source=v.EXPERIMENT + '/' + name, copy=destination, bytes=len(raw), sha256=v.sha(raw)))
|
|
self.write(v.REGISTRY + '/evidence-index.json', data(index))
|
|
self.write(v.REGISTRY + '/source-identity.json', data(source_rows))
|
|
self.write(v.REGISTRY + '/seal-verification.json', data(dict(pass_all=True, root=v.PROJECT + '/' + v.EXPERIMENT,
|
|
manifest_sha256=v.sha(manifest), checksums_sha256=v.sha(checksums), payload_file_count=len(rows), payload_bytes=sum(r['bytes'] for r in rows))))
|
|
directory = self.repo / v.REGISTRY
|
|
receipt = dict(schema='phase4an-engineering-publication-v1', publication_complete=True, registry=v.REGISTRY,
|
|
project_root=v.PROJECT, bulk_remains_at=v.EXPERIMENT, declared_status=status, model_directory_sealed=False,
|
|
used_source_count=21, model_files_copied=False, complete_bulk_backup=False, training=False,
|
|
candidate_generated=False, learning_pass_claimed=False, source_identity='source-identity.json',
|
|
experiment_manifest_sha256=v.sha(manifest), experiment_checksums_sha256=v.sha(checksums),
|
|
preregistration_sha256=v.sha(payload['protocol/preregistered.json']), final_sha256=v.sha(payload['audit/final-status.json']),
|
|
analysis_sha256=v.sha(payload['audit/engineering-analysis.json']), copied_evidence_files=len(index),
|
|
copied_evidence_bytes=sum(r['bytes'] for r in index), experiment_payload_not_copied=omitted,
|
|
files=[row(path.relative_to(directory).as_posix(), path.read_bytes()) for path in sorted(directory.rglob('*'))
|
|
if path.is_file() and path.name != 'publication-receipt.json'])
|
|
raw = data(receipt); self.write(v.REGISTRY + '/publication-receipt.json', raw)
|
|
return v.sha(raw)
|
|
|
|
def refresh_receipt(self, **updates):
|
|
path = self.repo / v.REGISTRY / 'publication-receipt.json'
|
|
receipt = json.loads(path.read_bytes()); receipt.update(updates)
|
|
receipt['files'] = [row(r['path'], (self.repo / v.REGISTRY / r['path']).read_bytes()) for r in receipt['files']]
|
|
raw = data(receipt); path.write_bytes(raw); self.pin = v.sha(raw); self.stage()
|
|
|
|
def test_index_and_commit_pass_without_touching_index(self):
|
|
before = (self.repo / '.git/index').read_bytes()
|
|
result = v.verify(self.repo, self.pin, staged=True)
|
|
self.assertTrue(result['pass_all'])
|
|
self.assertFalse(result['working_tree_payload_used'])
|
|
self.assertFalse(result['experiment_bulk_verified_by_this_tool'])
|
|
self.assertEqual(result['model_source_files'], 21)
|
|
self.assertEqual((self.repo / '.git/index').read_bytes(), before)
|
|
self.git('commit', '-qm', 'synthetic publication')
|
|
committed = v.verify(self.repo, self.pin, commit='HEAD')
|
|
self.assertEqual(committed['source'], 'commit')
|
|
self.assertEqual(committed['commit'], self.git('rev-parse', 'HEAD').decode().strip())
|
|
|
|
def test_working_tree_never_substitutes_for_staged_or_committed_blobs(self):
|
|
target = v.MODEL + '/' + v.SOURCES[0]
|
|
self.git('commit', '-qm', 'valid synthetic')
|
|
original = (self.repo / target).read_bytes()
|
|
self.write(target, b'bad working tree but index is valid\n')
|
|
self.assertTrue(v.verify(self.repo, self.pin, staged=True)['pass_all'])
|
|
self.git('add', '--', target)
|
|
self.write(target, original)
|
|
with self.assertRaisesRegex(ValueError, 'SHA/bytes'):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
self.assertTrue(v.verify(self.repo, self.pin, commit='HEAD')['pass_all'])
|
|
|
|
def test_external_receipt_sha_and_exclusive_source_mode_are_required(self):
|
|
for kwargs, pin in [({'staged': True}, '0' * 64), ({}, self.pin), ({'staged': True, 'commit': 'HEAD'}, self.pin)]:
|
|
with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
|
|
v.verify(self.repo, pin, **kwargs)
|
|
|
|
def test_extra_scientific_draft_and_registry_file_are_rejected(self):
|
|
for name in (v.MODEL + '/PROTOCOL_DRAFT.md', v.REGISTRY + '/unlisted.json'):
|
|
self.write(name, b'not in registered inventory\n'); self.git('add', '--', name)
|
|
with self.subTest(name=name), self.assertRaises(ValueError):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
self.git('rm', '--cached', '--', name)
|
|
(self.repo / name).unlink()
|
|
|
|
def test_missing_source_and_nonregular_index_entry_rejected(self):
|
|
path = v.MODEL + '/' + v.SOURCES[0]
|
|
self.git('rm', '--cached', '--', path)
|
|
with self.assertRaises(ValueError):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
self.git('add', '--', path)
|
|
oid = self.git('hash-object', '-w', '--stdin', input=b'elsewhere\n').decode().strip()
|
|
self.git('update-index', '--cacheinfo', '120000,' + oid + ',' + path)
|
|
with self.assertRaisesRegex(ValueError, 'nonregular'):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
|
|
def test_source_identity_semantics_checked_even_under_new_receipt_pin(self):
|
|
path = self.repo / v.REGISTRY / 'source-identity.json'
|
|
identities = json.loads(path.read_bytes()); identities[0]['sha256'] = 'f' * 64
|
|
path.write_bytes(data(identities)); self.refresh_receipt()
|
|
with self.assertRaisesRegex(ValueError, 'Source/snapshot'):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
|
|
def test_evidence_index_source_and_escaping_receipt_paths_fail(self):
|
|
path = self.repo / v.REGISTRY / 'evidence-index.json'
|
|
index = json.loads(path.read_bytes()); index[0]['source'] = 'wrong/source.py'
|
|
path.write_bytes(data(index)); self.refresh_receipt()
|
|
with self.assertRaisesRegex(ValueError, 'Evidence index'):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
receipt_path = self.repo / v.REGISTRY / 'publication-receipt.json'
|
|
receipt = json.loads(receipt_path.read_bytes()); receipt['files'][0]['path'] = '../escape.json'
|
|
raw = data(receipt); receipt_path.write_bytes(raw); self.pin = v.sha(raw); self.stage()
|
|
with self.assertRaisesRegex(ValueError, 'escaping'):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
|
|
def test_checksums_manifest_self_hash_is_not_just_trusted(self):
|
|
path = self.repo / v.REGISTRY / 'evidence/checksums.sha256'
|
|
text = path.read_text(); lines = text.splitlines(); lines[-1] = '0' * 64 + ' manifest.json'
|
|
raw = ('\n'.join(lines) + '\n').encode(); path.write_bytes(raw)
|
|
self.refresh_receipt(experiment_checksums_sha256=v.sha(raw))
|
|
with self.assertRaisesRegex(ValueError, 'Checksum inventory'):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
|
|
def test_declared_fail_status_is_preserved_not_relabelled_science_pass(self):
|
|
self.pin = self.fixture(status='FAIL-NUMERICAL'); self.stage()
|
|
result = v.verify(self.repo, self.pin, staged=True)
|
|
self.assertTrue(result['pass_all'])
|
|
self.assertEqual(result['declared_status'], 'FAIL-NUMERICAL')
|
|
self.assertTrue(result['publication_integrity_is_not_scientific_pass'])
|
|
self.assertFalse(result['scientific_gate_executed'])
|
|
|
|
def test_duplicate_JSON_field_in_receipt_is_rejected(self):
|
|
path = self.repo / v.REGISTRY / 'publication-receipt.json'
|
|
raw = path.read_bytes().replace(b'{', b'{"publication_complete":true,', 1)
|
|
path.write_bytes(raw); self.pin = v.sha(raw); self.stage()
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate JSON'):
|
|
v.verify(self.repo, self.pin, staged=True)
|
|
|
|
def test_cli_emits_compact_summary_without_full_payload(self):
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out):
|
|
status = v.main(['--repo', str(self.repo), '--receipt-sha256', self.pin, '--staged'])
|
|
self.assertEqual(status, 0)
|
|
result = json.loads(out.getvalue())
|
|
self.assertTrue(result['pass_all'])
|
|
self.assertNotIn('verified_blobs', result)
|
|
self.assertLess(len(out.getvalue()), 1800)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|