Digital Drosophila Archive
230 lines
15 KiB
Python
230 lines
15 KiB
Python
"""In-memory SFTP / temporary local files only; no real server access."""
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import stat
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import fetch_phase4an as f
|
|
|
|
|
|
def data(value):
|
|
return (json.dumps(value, sort_keys=True, allow_nan=False) + '\n').encode()
|
|
|
|
|
|
def row(name, raw):
|
|
return dict(path=name, bytes=len(raw), sha256=f.sha(raw))
|
|
|
|
|
|
class FakeSFTP:
|
|
def __init__(self, files):
|
|
self.files, self.opened, self.symlinks, self.hook = files, [], set(), None
|
|
|
|
def lstat(self, name):
|
|
if name in self.symlinks: return types.SimpleNamespace(st_mode=stat.S_IFLNK | 0o777, st_size=1)
|
|
if name in self.files: return types.SimpleNamespace(st_mode=stat.S_IFREG | 0o644, st_size=len(self.files[name]))
|
|
if any(x.startswith(name.rstrip('/') + '/') for x in self.files): return types.SimpleNamespace(st_mode=stat.S_IFDIR | 0o755, st_size=0)
|
|
raise FileNotFoundError(name)
|
|
|
|
def open(self, name, mode):
|
|
if mode != 'rb': raise AssertionError('Remote writes are forbidden')
|
|
self.opened.append(name)
|
|
if self.hook: self.hook(name)
|
|
return io.BytesIO(self.files[name])
|
|
|
|
|
|
def fixture(status=f.PASS):
|
|
payload, sources, inputs, identities = {}, {}, {}, []
|
|
for name in f.SOURCE_FILES:
|
|
content = ('# synthetic ' + name).encode(); path = 'protocol/source-snapshot/' + name
|
|
payload[path] = content
|
|
sources[f.REMOTE_ROOT + '/' + f.MODEL_REL + '/' + name] = f.sha(content)
|
|
inputs[f.REMOTE_ROOT + '/' + f.EXP_REL + '/' + path] = f.sha(content)
|
|
identities.append(dict(model_source=f.MODEL_REL + '/' + name, sealed_source=f.EXP_REL + '/' + path,
|
|
copy='source-snapshot/' + name, bytes=len(content), sha256=f.sha(content)))
|
|
payload['protocol/preregistered.json'] = data(dict(source_sha256=sources, input_sha256=inputs))
|
|
prereg = f.sha(payload['protocol/preregistered.json'])
|
|
numerical, covered = status != 'FAIL-NUMERICAL', status == f.PASS
|
|
analysis = dict(status=status, registration_sha256=prereg, source_sha256=sources, expected_calls=32,
|
|
analyzed_calls=32, executed_ticks=11074, numerical_exact=numerical, nonvacuous_temporal_coverage=covered,
|
|
pass_all=numerical and covered, training=False, candidate_generated=False, learning_pass_claimed=False)
|
|
payload['audit/engineering-analysis.json'] = data(analysis)
|
|
analysis_sha = f.sha(payload['audit/engineering-analysis.json'])
|
|
final = dict(status=status, analysis_sha256=analysis_sha, registration_sha256=prereg, analysis_recomputed_exactly=True,
|
|
completed_calls=32, executed_ticks=11074, training=False, learned_candidate=False, learning_pass_claimed=False)
|
|
payload['audit/final-status.json'] = data(final)
|
|
payload.update({'runs/example/state.npz': b'bulk', 'runs/example/queue.jsonl': b'bulk queue',
|
|
'extension-build/build.log': b'not selected', 'logs/readable.log': b'small log'})
|
|
manifest_rows = [row(name, raw) for name, raw in sorted(payload.items())]
|
|
manifest = data(dict(root=f.REMOTE_ROOT + '/' + f.EXP_REL, files=manifest_rows))
|
|
checksums = (''.join(r['sha256'] + ' ' + r['path'] + '\n' for r in manifest_rows) + f.sha(manifest) + ' manifest.json\n').encode()
|
|
controls = {'manifest.json': manifest, 'checksums.sha256': checksums}
|
|
registry, evidence_index, omitted = {}, [], []
|
|
for name, raw in sorted({**payload, **controls}.items()):
|
|
r = row(name, raw); target, reason = f.selected(r)
|
|
if target:
|
|
registry[target] = raw
|
|
evidence_index.append(dict(source=f.EXP_REL + '/' + name, copy=target, bytes=len(raw), sha256=f.sha(raw)))
|
|
else: omitted.append(dict(r, reason=reason))
|
|
proof = dict(pass_all=True, root=f.REMOTE_ROOT + '/' + f.EXP_REL, manifest_sha256=f.sha(manifest), checksums_sha256=f.sha(checksums),
|
|
payload_file_count=len(manifest_rows), payload_bytes=sum(r['bytes'] for r in manifest_rows))
|
|
registry['seal-verification.json'] = data(proof)
|
|
registry['evidence-index.json'] = data(evidence_index)
|
|
registry['source-identity.json'] = data(identities)
|
|
receipt = dict(schema='phase4an-engineering-publication-v1', publication_complete=True, project_root=f.REMOTE_ROOT,
|
|
registry=f.REGISTRY_REL, declared_status=status, used_sources_verified_before_and_after=True,
|
|
complete_checksums_reference_bulk_not_copied=True, publication_success_is_not_engineering_or_learning_pass=True,
|
|
model_directory_sealed=False, model_files_copied=False, complete_bulk_backup=False, numerical_gate_rerun_by_publisher=False,
|
|
training=False, candidate_generated=False, learning_pass_claimed=False, existing_files_modified=False, helper_globals_modified=False,
|
|
selection_policy=dict(max_file_bytes=f.MAX_FILE_BYTES, suffixes=sorted(f.SUFFIXES), source_snapshot_names=list(f.SOURCE_FILES),
|
|
excluded_tree='extension-build', all_jsonl_excluded=True),
|
|
used_source_count=21, completed_calls=32, executed_ticks=11074, preregistration_sha256=prereg,
|
|
source_identity='source-identity.json', bulk_remains_at=f.EXP_REL, experiment_manifest_sha256=f.sha(manifest),
|
|
experiment_checksums_sha256=f.sha(checksums), final_sha256=f.sha(payload['audit/final-status.json']), analysis_sha256=analysis_sha,
|
|
publisher_sha256='a'*64, helper_sha256='b'*64, files=[row(name, raw) for name, raw in sorted(registry.items())],
|
|
experiment_payload_not_copied=omitted, copied_evidence_files=len(evidence_index),
|
|
copied_evidence_bytes=sum(r['bytes'] for r in evidence_index), declared_numerical_exact=numerical,
|
|
declared_live_coverage=covered, declared_engineering_pass=numerical and covered)
|
|
registry['publication-receipt.json'] = data(receipt)
|
|
return FakeSFTP({f.REGISTRY_ROOT + '/' + name: raw for name, raw in registry.items()}), prereg, f.sha(registry['publication-receipt.json'])
|
|
|
|
|
|
class FetchTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='phase4an-fetch-test-'); self.addCleanup(self.temp.cleanup)
|
|
self.destination = Path(self.temp.name).resolve() / 'report'
|
|
self.sftp, self.prereg, self.pin = fixture()
|
|
|
|
def fetch(self, **kwargs):
|
|
with patch.object(f, 'PREREG_SHA', self.prereg):
|
|
return f.fetch(self.sftp, self.destination, self.pin, **kwargs)
|
|
|
|
def receipt(self): return f.json_bytes(self.sftp.files[f.REGISTRY_ROOT + '/publication-receipt.json'])
|
|
|
|
def resign(self, receipt):
|
|
raw = data(receipt); self.sftp.files[f.REGISTRY_ROOT + '/publication-receipt.json'] = raw; self.pin = f.sha(raw)
|
|
|
|
def change_selected(self, name, content):
|
|
self.sftp.files[f.REGISTRY_ROOT + '/' + name] = content
|
|
receipt = self.receipt(); next(r for r in receipt['files'] if r['path'] == name).update(bytes=len(content), sha256=f.sha(content))
|
|
self.resign(receipt)
|
|
|
|
def test_exact_small_registry_mirror_and_receipt_hashes_no_model_or_bulk(self):
|
|
result = self.fetch()
|
|
self.assertTrue(result['fetch_complete']); self.assertTrue(result['exact_receipt_inventory_downloaded'])
|
|
self.assertFalse(result['engineering_gate_executed']); self.assertFalse(result['learning_pass_claimed'])
|
|
self.assertTrue((self.destination / 'evidence/audit/final-status.json').is_file())
|
|
self.assertTrue((self.destination / 'source-snapshot/runtime_light.py').is_file())
|
|
self.assertEqual(len(result['files']), len(self.receipt()['files']) + 1)
|
|
self.assertEqual(result['sftp_bytes_read_including_anchor_reread'], result['downloaded_bytes'] + len(self.sftp.files[f.REGISTRY_ROOT + '/publication-receipt.json']))
|
|
for r in result['files']:
|
|
self.assertEqual(f.sha((self.destination / r['path']).read_bytes()), r['sha256'])
|
|
self.assertTrue(all(name.startswith(f.REGISTRY_ROOT + '/') for name in self.sftp.opened))
|
|
self.assertFalse(any(name.endswith(('.npz', '.npy', '.jsonl')) or '/extension-build/' in name for name in self.sftp.opened))
|
|
|
|
def test_failure_and_incomplete_declarations_never_promoted(self):
|
|
for status in ('FAIL-NUMERICAL', 'INCOMPLETE-COVERAGE'):
|
|
with self.subTest(status=status), tempfile.TemporaryDirectory() as directory:
|
|
sftp, prereg, pin = fixture(status)
|
|
with patch.object(f, 'PREREG_SHA', prereg): result = f.fetch(sftp, Path(directory).resolve() / 'report', pin)
|
|
self.assertEqual(result['declared_status'], status)
|
|
self.assertTrue(result['fetch_success_is_not_engineering_or_learning_pass'])
|
|
|
|
def test_external_receipt_pin_and_fixed_prereg_required_before_payload(self):
|
|
for pin in ('', 'x'*64, '0'*64):
|
|
self.sftp.opened.clear()
|
|
with self.assertRaises(ValueError): f.fetch(self.sftp, self.destination, pin)
|
|
self.assertFalse(self.destination.exists())
|
|
self.assertTrue(all(name.endswith('/publication-receipt.json') for name in self.sftp.opened))
|
|
with self.assertRaisesRegex(ValueError, 'authority/scope'): f.fetch(self.sftp, self.destination, self.pin)
|
|
|
|
def test_forbidden_paths_and_binary_or_extra_source_fail_before_more_reads(self):
|
|
for name in ('../bad.json', 'evidence/x/../bad.json', 'evidence/C:/bad.json', 'evidence/CON.log',
|
|
'source-snapshot/unregistered.py', 'evidence/state.npz', 'evidence/queue.jsonl', 'evidence/extension-build/info.json'):
|
|
with self.subTest(name=name):
|
|
self.sftp, self.prereg, self.pin = fixture(); receipt = self.receipt()
|
|
receipt['files'].append(dict(path=name, bytes=1, sha256='a'*64)); self.resign(receipt)
|
|
with self.assertRaises(ValueError): self.fetch()
|
|
self.assertFalse(self.destination.exists()); self.assertEqual(len(self.sftp.opened), 1)
|
|
|
|
def test_case_prefix_collision_and_receipt_self_inventory_rejected(self):
|
|
for name in ('EVIDENCE/audit/final-status.json', 'evidence/audit', 'publication-receipt.json', 'fetch-verification.json'):
|
|
with self.subTest(name=name):
|
|
self.sftp, self.prereg, self.pin = fixture(); receipt = self.receipt()
|
|
receipt['files'].append(dict(path=name, bytes=1, sha256='a'*64)); self.resign(receipt)
|
|
with self.assertRaises(ValueError): self.fetch()
|
|
self.assertFalse(self.destination.exists())
|
|
|
|
def test_budget_is_hard_cap_and_cannot_be_raised(self):
|
|
with self.assertRaises(ValueError): self.fetch(max_total_bytes=f.MAX_TOTAL_BYTES + 1)
|
|
self.assertFalse(self.sftp.opened)
|
|
with self.assertRaisesRegex(ValueError, 'budget'): self.fetch(max_total_bytes=1)
|
|
self.assertFalse(self.destination.exists())
|
|
|
|
def test_manifest_checksum_link_checked_even_after_receipt_resigned(self):
|
|
name = 'evidence/checksums.sha256'; raw = self.sftp.files[f.REGISTRY_ROOT + '/' + name]
|
|
lines = raw.decode().splitlines(); lines[-1] = '0'*64 + ' manifest.json'; changed = ('\n'.join(lines)+'\n').encode()
|
|
self.change_selected(name, changed); receipt = self.receipt(); receipt['experiment_checksums_sha256'] = f.sha(changed); self.resign(receipt)
|
|
with self.assertRaisesRegex(ValueError, 'self-SHA'): self.fetch()
|
|
self.assertFalse(self.destination.exists())
|
|
|
|
def test_source_identity_and_index_crosslinks_not_just_receipt_flags(self):
|
|
for name, mutation in (('source-identity.json', 'model_source'), ('evidence-index.json', 'source')):
|
|
with self.subTest(name=name):
|
|
self.sftp, self.prereg, self.pin = fixture(); values = f.json_bytes(self.sftp.files[f.REGISTRY_ROOT + '/' + name])
|
|
values[0][mutation] = 'wrong/source.py'; self.change_selected(name, data(values))
|
|
with self.assertRaises(ValueError): self.fetch()
|
|
self.assertFalse(self.destination.exists())
|
|
|
|
def test_exact_omitted_and_selection_inventory_cannot_drop_evidence(self):
|
|
receipt = self.receipt(); receipt['experiment_payload_not_copied'].pop(); self.resign(receipt)
|
|
with self.assertRaisesRegex(ValueError, 'Omitted'): self.fetch()
|
|
self.assertFalse(self.destination.exists())
|
|
self.sftp, self.prereg, self.pin = fixture(); receipt = self.receipt()
|
|
receipt['files'] = [r for r in receipt['files'] if r['path'] != 'evidence/logs/readable.log']; self.resign(receipt)
|
|
with self.assertRaisesRegex(ValueError, 'exact sealed policy'): self.fetch()
|
|
|
|
def test_existing_or_alias_destination_denied_without_remote_reads(self):
|
|
self.destination.mkdir()
|
|
with self.assertRaisesRegex(ValueError, 'exists'): self.fetch()
|
|
self.assertFalse(self.sftp.opened)
|
|
with self.assertRaises(ValueError): f.check_destination(self.destination / '..' / 'other')
|
|
|
|
def test_remote_symlink_directory_and_file_denied(self):
|
|
for remote in (f.REGISTRY_ROOT, f.REGISTRY_ROOT + '/publication-receipt.json'):
|
|
self.sftp.symlinks.add(remote)
|
|
with self.assertRaisesRegex(ValueError, 'symlink'): self.fetch()
|
|
self.assertFalse(self.sftp.opened); self.sftp.symlinks.clear()
|
|
|
|
def test_payload_corruption_preserves_partial_and_prevents_retry(self):
|
|
remote = f.REGISTRY_ROOT + '/evidence/logs/readable.log'; self.sftp.files[remote] = b'x' * len(self.sftp.files[remote])
|
|
with self.assertRaisesRegex(ValueError, 'SHA'): self.fetch()
|
|
self.assertTrue((self.destination / 'fetch-failure.json').is_file())
|
|
self.assertFalse((self.destination / 'fetch-verification.json').exists())
|
|
with self.assertRaisesRegex(ValueError, 'exists'): self.fetch()
|
|
|
|
def test_remote_receipt_changed_during_download_fails_final_anchor(self):
|
|
def change(name):
|
|
if name.endswith('/source-snapshot/runtime_light.py'):
|
|
self.sftp.files[f.REGISTRY_ROOT + '/publication-receipt.json'] += b' '
|
|
self.sftp.hook = change
|
|
with self.assertRaises(ValueError): self.fetch()
|
|
self.assertTrue((self.destination / 'fetch-failure.json').exists())
|
|
self.assertFalse((self.destination / 'fetch-verification.json').exists())
|
|
|
|
def test_cli_host_trust_credentials_and_default_destination_are_explicit(self):
|
|
source = Path(f.__file__).read_text(encoding='utf-8')
|
|
self.assertIn('paramiko.RejectPolicy()', source)
|
|
self.assertIn('client.load_system_host_keys(str(args.known_hosts))', source)
|
|
self.assertIn('allow_agent=False, look_for_keys=False', source)
|
|
self.assertNotIn('AutoAddPolicy', source)
|
|
self.assertNotIn("add_argument('--password'", source)
|
|
self.assertEqual(f.DEFAULT_DESTINATION, Path(f.__file__).resolve().parents[2] / 'reports/phase4an-engineering-validation')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|