Digital Drosophila Archive
338 lines
19 KiB
Python
338 lines
19 KiB
Python
"""Fetch only externally anchored Phase 4AK small evidence and model source.
|
|
|
|
No SSH connection is made at import. The API accepts an already authenticated
|
|
SFTP client; the CLI rejects unknown host keys. This is not a trajectory backup
|
|
or a local re-verification of remote bulk payloads/scientific gates.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import stat
|
|
|
|
REMOTE_ROOT = '/data/digital-drosophila'
|
|
MODEL_RELATIVE = 'models/male-cns-lif/apl-timing-validation-v1'
|
|
EXPERIMENT_RELATIVE = 'experiments/phase4ak-apl-timing-validation'
|
|
REGISTRY_RELATIVE = 'registry/phase4ak-apl-timing-validation'
|
|
MODEL_ROOT = REMOTE_ROOT + '/' + MODEL_RELATIVE
|
|
REGISTRY_ROOT = REMOTE_ROOT + '/' + REGISTRY_RELATIVE
|
|
DEFAULT_DESTINATION = Path(__file__).resolve().parents[2] / 'reports/phase4ak-apl-timing-validation'
|
|
MAX_FILE_BYTES = 4 * 1024**2
|
|
MAX_TOTAL_BYTES = 1024**3
|
|
EVIDENCE_SUFFIXES = frozenset(('.json', '.csv', '.md', '.log', '.sha256'))
|
|
SOURCE_SUFFIXES = frozenset(('.py', '.json', '.md', '.yaml', '.yml', '.toml', '.txt', '.sha256', '.cpp', '.cu', '.h', '.hpp'))
|
|
CONTROLS = ('manifest.json', 'checksums.sha256')
|
|
GENERATED = ('seal-verification.json', 'evidence-index.json')
|
|
SHA = re.compile(r'[0-9a-f]{64}\Z')
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def sha(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def unique_object(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
require(key not in result, 'Duplicate JSON key: ' + key)
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def json_bytes(data):
|
|
def bad(value):
|
|
raise ValueError('Nonfinite JSON constant: ' + value)
|
|
return json.loads(data.decode('utf-8'), object_pairs_hook=unique_object, parse_constant=bad)
|
|
|
|
|
|
def relative_name(value):
|
|
require(isinstance(value, str) and value and not any(c in value for c in ('\\', '\0', '\r', '\n', ':')),
|
|
'Invalid relative filename')
|
|
path = PurePosixPath(value)
|
|
require(not path.is_absolute() and path.as_posix() == value and
|
|
all(part not in ('', '.', '..', '__pycache__') and not part.endswith((' ', '.')) for part in path.parts),
|
|
'Noncanonical/escaping/cache filename: ' + value)
|
|
devices = {'CON', 'PRN', 'AUX', 'NUL', *(f'COM{i}' for i in range(1, 10)), *(f'LPT{i}' for i in range(1, 10))}
|
|
require(all(part.split('.')[0].upper() not in devices for part in path.parts), 'Reserved local filename')
|
|
return path
|
|
|
|
|
|
def file_row(row, *, small=False):
|
|
require(isinstance(row, dict) and set(row) == {'path', 'bytes', 'sha256'}, 'Invalid hash-inventory row')
|
|
relative_name(row['path'])
|
|
require(type(row['bytes']) is int and row['bytes'] >= 0 and isinstance(row['sha256'], str) and
|
|
SHA.fullmatch(row['sha256']), 'Invalid file size/SHA')
|
|
if small:
|
|
require(row['bytes'] <= MAX_FILE_BYTES, 'Receipt file exceeds 4 MiB: ' + row['path'])
|
|
return row
|
|
|
|
|
|
class BoundedSFTP:
|
|
def __init__(self, sftp):
|
|
self.sftp = sftp
|
|
self.directories = set()
|
|
|
|
def read(self, remote, size=None, checksum=None):
|
|
require(remote.startswith(REGISTRY_ROOT + '/') or remote.startswith(MODEL_ROOT + '/'),
|
|
'Remote request outside fixed AK registry/model scope')
|
|
# Never list a remote directory or discover unregistered sidecars.
|
|
parent = PurePosixPath(remote).parent
|
|
for directory in reversed((parent, *parent.parents)):
|
|
name = str(directory)
|
|
if name not in self.directories:
|
|
require(stat.S_ISDIR(self.sftp.lstat(name).st_mode), 'Remote symlink/non-directory: ' + name)
|
|
self.directories.add(name)
|
|
info = self.sftp.lstat(remote)
|
|
require(stat.S_ISREG(info.st_mode), 'Remote symlink/nonregular file: ' + remote)
|
|
require(0 <= info.st_size <= MAX_FILE_BYTES and (size is None or info.st_size == size),
|
|
'Remote size outside expected small-file limit: ' + remote)
|
|
chunks, count = [], 0
|
|
with self.sftp.open(remote, 'rb') as source:
|
|
while True:
|
|
block = source.read(min(1 << 20, MAX_FILE_BYTES + 1 - count))
|
|
if not block:
|
|
break
|
|
count += len(block)
|
|
require(count <= MAX_FILE_BYTES, 'Remote file grew beyond download limit')
|
|
chunks.append(block)
|
|
data = b''.join(chunks)
|
|
require(len(data) == info.st_size and (size is None or len(data) == size), 'Remote size changed during read')
|
|
if checksum is not None:
|
|
require(sha(data) == checksum, 'Remote file SHA mismatch: ' + remote)
|
|
return data
|
|
|
|
|
|
def validate_receipt(data, expected_sha):
|
|
require(sha(data) == expected_sha, 'External publication-receipt SHA mismatch')
|
|
receipt = json_bytes(data)
|
|
require(isinstance(receipt, dict) and receipt.get('publication_complete') is True and
|
|
receipt.get('registry') == REGISTRY_RELATIVE and receipt.get('project_root') == REMOTE_ROOT,
|
|
'Wrong/incomplete AK publication receipt')
|
|
for key in ('source_trees_verified_before_and_after_copy', 'publication_success_is_not_scientific_pass'):
|
|
require(receipt.get(key) is True, 'Missing publication qualification: ' + key)
|
|
for key in ('complete_bulk_backup', 'scientific_gate_executed_by_publisher',
|
|
'scientific_gate_certified_by_publication', 'candidate_generated', 'learning_achieved'):
|
|
require(receipt.get(key) is False, 'Unexpected publication claim: ' + key)
|
|
require(receipt.get('limits') == dict(max_selected_evidence_file_bytes=MAX_FILE_BYTES,
|
|
selected_suffixes=sorted(EVIDENCE_SUFFIXES)), 'Unexpected publication selection limits')
|
|
for key in ('model_manifest_sha256', 'experiment_manifest_sha256'):
|
|
require(isinstance(receipt.get(key), str) and SHA.fullmatch(receipt[key]), 'Missing manifest anchor: ' + key)
|
|
rows = {}
|
|
require(isinstance(receipt.get('files'), list) and receipt['files'], 'Empty receipt inventory')
|
|
for row in receipt['files']:
|
|
file_row(row, small=True)
|
|
name = row['path']
|
|
require(name not in rows, 'Duplicate receipt file: ' + name)
|
|
allowed = name in GENERATED or name in ('model/manifest.json', 'model/checksums.sha256') or (
|
|
name.startswith('evidence/') and PurePosixPath(name).suffix.lower() in EVIDENCE_SUFFIXES)
|
|
require(allowed, 'Receipt attempts a non-evidence/sidecar download: ' + name)
|
|
rows[name] = row
|
|
required = {'model/manifest.json', 'model/checksums.sha256', 'evidence/manifest.json',
|
|
'evidence/checksums.sha256', 'evidence/audit/final-status.json',
|
|
'evidence/regression/old-integrity-after.json', *GENERATED}
|
|
require(required.issubset(rows), 'Receipt omits required AK evidence/seal controls')
|
|
return receipt, rows
|
|
|
|
|
|
def validate_manifest(manifest_data, checksums_data, expected_root, expected_sha):
|
|
require(sha(manifest_data) == expected_sha, 'Receipt/manifest SHA mismatch')
|
|
manifest = json_bytes(manifest_data)
|
|
require(isinstance(manifest, dict) and set(manifest) == {'root', 'files'} and
|
|
manifest['root'] == expected_root and isinstance(manifest['files'], list) and manifest['files'],
|
|
'Wrong seal manifest schema/root')
|
|
rows = {}
|
|
for row in manifest['files']:
|
|
file_row(row)
|
|
require(row['path'] not in rows and row['path'] not in CONTROLS, 'Duplicate/control manifest payload')
|
|
rows[row['path']] = row
|
|
checksums = {}
|
|
for line in checksums_data.decode('utf-8').splitlines():
|
|
require(len(line) >= 67 and line[64:66] == ' ' and SHA.fullmatch(line[:64]), 'Invalid checksum line')
|
|
name = line[66:]
|
|
relative_name(name)
|
|
require(name not in checksums, 'Duplicate checksum filename')
|
|
checksums[name] = line[:64]
|
|
require(checksums == {**{name: row['sha256'] for name, row in rows.items()}, 'manifest.json': expected_sha},
|
|
'Manifest self-SHA/checksum inventory mismatch')
|
|
return rows
|
|
|
|
|
|
def local_name(registry_name):
|
|
if registry_name.startswith('evidence/'):
|
|
return registry_name[len('evidence/'):]
|
|
if registry_name.startswith('model/'):
|
|
return registry_name
|
|
return 'registry/' + registry_name
|
|
|
|
|
|
def _check_local_target(destination):
|
|
require(destination.is_absolute(), 'Report destination must be absolute')
|
|
for part in (destination, *destination.parents):
|
|
require(not part.is_symlink() and not getattr(part, 'is_junction', lambda: False)(),
|
|
'Local symlink/junction destination is forbidden')
|
|
require(not destination.exists(), 'Report destination exists; no overwrite/resume')
|
|
|
|
|
|
def _write(path, data):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open('xb') as output:
|
|
output.write(data)
|
|
require(path.stat().st_size == len(data) and sha(path.read_bytes()) == sha(data), 'Local write verification failed')
|
|
|
|
|
|
def fetch(sftp, destination, receipt_sha256, *, max_total_bytes=MAX_TOTAL_BYTES):
|
|
"""SFTP read-only API. Destination is exclusive; receipt SHA is out-of-band.
|
|
|
|
Registry evidence/* mirrors into the report root, model/* into model/, and
|
|
receipt/generated publication metadata into registry/. Only source-whitelisted
|
|
model payloads are added; every transferred file is individually SHA checked.
|
|
"""
|
|
require(isinstance(receipt_sha256, str) and SHA.fullmatch(receipt_sha256.lower()), 'External receipt SHA256 is required')
|
|
receipt_sha256 = receipt_sha256.lower()
|
|
require(type(max_total_bytes) is int and max_total_bytes > 0, 'Invalid total download budget')
|
|
destination = Path(destination)
|
|
_check_local_target(destination)
|
|
remote = BoundedSFTP(sftp)
|
|
receipt_bytes = remote.read(REGISTRY_ROOT + '/publication-receipt.json', checksum=receipt_sha256)
|
|
receipt, rows = validate_receipt(receipt_bytes, receipt_sha256)
|
|
cached = {}
|
|
for prefix in ('model/', 'evidence/'):
|
|
for control in CONTROLS:
|
|
name = prefix + control; row = rows[name]
|
|
cached[name] = remote.read(REGISTRY_ROOT + '/' + name, row['bytes'], row['sha256'])
|
|
model_rows = validate_manifest(cached['model/manifest.json'], cached['model/checksums.sha256'],
|
|
MODEL_ROOT, receipt['model_manifest_sha256'])
|
|
experiment_rows = validate_manifest(cached['evidence/manifest.json'], cached['evidence/checksums.sha256'],
|
|
REMOTE_ROOT + '/' + EXPERIMENT_RELATIVE, receipt['experiment_manifest_sha256'])
|
|
# Model source is fetched from its sealed model path, not the registry. Bind
|
|
# its current controls to the receipt before reading any source payload.
|
|
for control in CONTROLS:
|
|
row = rows['model/' + control]
|
|
remote.read(MODEL_ROOT + '/' + control, row['bytes'], row['sha256'])
|
|
for name, row in rows.items():
|
|
if name.startswith('evidence/') and name[len('evidence/'):] not in CONTROLS:
|
|
original = experiment_rows.get(name[len('evidence/'):])
|
|
require(original is not None and all(row[key] == original[key] for key in ('bytes', 'sha256')),
|
|
'Registry evidence disagrees with experiment seal: ' + name)
|
|
for name in GENERATED:
|
|
row = rows[name]
|
|
cached[name] = remote.read(REGISTRY_ROOT + '/' + name, row['bytes'], row['sha256'])
|
|
proof = json_bytes(cached['seal-verification.json'])
|
|
for key, root, manifest_rows, prefix in (('model', MODEL_ROOT, model_rows, 'model/'),
|
|
('experiment', REMOTE_ROOT + '/' + EXPERIMENT_RELATIVE, experiment_rows, 'evidence/')):
|
|
item = proof.get(key, {})
|
|
require(item.get('pass_all') is True and item.get('root') == root and
|
|
item.get('manifest_sha256') == receipt[key + '_manifest_sha256'] and
|
|
item.get('checksums_sha256') == rows[prefix + 'checksums.sha256']['sha256'] and
|
|
item.get('payload_file_count') == len(manifest_rows) and
|
|
item.get('payload_bytes') == sum(row['bytes'] for row in manifest_rows.values()),
|
|
'Publication seal-verification disagrees with manifest: ' + key)
|
|
index = json_bytes(cached['evidence-index.json'])
|
|
require(isinstance(index, list), 'Invalid publication evidence index')
|
|
indexed = set()
|
|
for item in index:
|
|
require(isinstance(item, dict) and set(item) == {'source', 'copy', 'bytes', 'sha256'}, 'Invalid evidence-index row')
|
|
name = item['copy']
|
|
require(name in rows and name not in GENERATED and name not in indexed, 'Duplicate/unexpected evidence-index copy')
|
|
indexed.add(name)
|
|
expected_source = (EXPERIMENT_RELATIVE + '/' + name[len('evidence/'):] if name.startswith('evidence/')
|
|
else MODEL_RELATIVE + '/' + name[len('model/'):])
|
|
require(item['source'] == expected_source and all(item[key] == rows[name][key] for key in ('bytes', 'sha256')),
|
|
'Evidence-index source/hash mismatch')
|
|
require(indexed == set(rows) - set(GENERATED), 'Evidence-index coverage mismatch')
|
|
plan = [(REGISTRY_ROOT + '/' + name, local_name(name), row, cached.get(name)) for name, row in sorted(rows.items())]
|
|
omitted = []
|
|
for name, row in sorted(model_rows.items()):
|
|
if PurePosixPath(name).suffix.lower() in SOURCE_SUFFIXES and row['bytes'] <= MAX_FILE_BYTES:
|
|
plan.append((MODEL_ROOT + '/' + name, 'model/' + name, row, None))
|
|
else:
|
|
omitted.append(dict(row, reason='model_non_source_or_over_4_MiB'))
|
|
paths = ['registry/publication-receipt.json', 'fetch-verification.json', 'fetch-failure.json'] + [entry[1] for entry in plan]
|
|
lowered = [relative_name(name).as_posix().casefold() for name in paths]
|
|
require(len(set(lowered)) == len(lowered) and not any(
|
|
parent.as_posix().casefold() in set(lowered) for name in paths for parent in PurePosixPath(name).parents if str(parent) != '.'),
|
|
'Local case-insensitive/path-prefix collision')
|
|
planned_bytes = len(receipt_bytes) + sum(entry[2]['bytes'] for entry in plan)
|
|
require(planned_bytes <= max_total_bytes, 'Selected evidence exceeds total download budget')
|
|
_check_local_target(destination)
|
|
destination.mkdir(parents=True, exist_ok=False)
|
|
try:
|
|
_write(destination / 'registry/publication-receipt.json', receipt_bytes)
|
|
downloaded = []
|
|
for source, local, row, data in plan:
|
|
if data is None:
|
|
data = remote.read(source, row['bytes'], row['sha256'])
|
|
require(len(data) == row['bytes'] and sha(data) == row['sha256'], 'Cached/downloaded file hash mismatch')
|
|
_write(destination / local, data)
|
|
downloaded.append(dict(remote=source, local=local, bytes=len(data), sha256=sha(data)))
|
|
final = json_bytes((destination / 'audit/final-status.json').read_bytes())
|
|
require(final.get('status') == receipt.get('declared_audit_status'), 'Receipt/final-status declaration mismatch')
|
|
for key, value in receipt.get('declared_gate_results', {}).items():
|
|
require(final.get(key) == value and type(final.get(key)) is type(value), 'Receipt/final gate flag mismatch')
|
|
for key, value in receipt.get('declared_validation_counts', {}).items():
|
|
require(final.get(key) == value and type(final.get(key)) is type(value), 'Receipt/final coverage mismatch')
|
|
# Re-read immutable anchors, not unselected experiment payloads.
|
|
remote.read(REGISTRY_ROOT + '/publication-receipt.json', len(receipt_bytes), receipt_sha256)
|
|
for control in CONTROLS:
|
|
row = rows['model/' + control]
|
|
remote.read(MODEL_ROOT + '/' + control, row['bytes'], row['sha256'])
|
|
result = dict(fetch_complete=True, publication_receipt_sha256=receipt_sha256,
|
|
external_receipt_anchor_required=True, declared_audit_status=final['status'],
|
|
model_manifest_sha256=receipt['model_manifest_sha256'], experiment_manifest_sha256=receipt['experiment_manifest_sha256'],
|
|
both_manifest_self_checksums_verified=True, every_downloaded_file_sha256_verified=True,
|
|
files=downloaded, file_count=len(downloaded), selected_bytes=planned_bytes,
|
|
model_payload_not_downloaded=omitted, all_remote_experiment_bulk_rehashed_locally=False,
|
|
full_remote_seal_verification_is_publisher_receipt_claim=True, complete_bulk_backup=False,
|
|
scientific_gate_executed=False, scientific_pass_certified_by_fetch=False,
|
|
fetch_success_is_not_scientific_pass=True, candidate_generated=False, learning_achieved=False,
|
|
AJ_interpretation_sidecar_requested=False, existing_files_modified=False)
|
|
_write(destination / 'fetch-verification.json', (json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + '\n').encode())
|
|
return result
|
|
except Exception as error:
|
|
failure = destination / 'fetch-failure.json'
|
|
if not failure.exists():
|
|
_write(failure, (json.dumps(dict(fetch_complete=False, type=type(error).__name__, message=str(error),
|
|
partial_outputs_preserved=True, existing_files_modified=False), allow_nan=False) + '\n').encode())
|
|
raise
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--receipt-sha256', required=True, help='Out-of-band SHA printed by the verified AK publisher')
|
|
parser.add_argument('--destination', type=Path, default=DEFAULT_DESTINATION)
|
|
parser.add_argument('--host', default='192.168.10.51')
|
|
parser.add_argument('--user', default='qishizhe')
|
|
parser.add_argument('--port', type=int, default=22)
|
|
parser.add_argument('--password-env', default='MALECNS_REMOTE_PASSWORD')
|
|
parser.add_argument('--known-hosts', type=Path, default=Path.home() / '.ssh/known_hosts',
|
|
help='Explicit existing user known_hosts; unknown hosts are always rejected')
|
|
args = parser.parse_args()
|
|
require(SHA.fullmatch(args.receipt_sha256.lower()), 'External receipt SHA256 is required')
|
|
_check_local_target(args.destination)
|
|
require(args.known_hosts.is_file(), 'Known-hosts file is missing; configure trust externally, never auto-add')
|
|
import paramiko
|
|
client = paramiko.SSHClient()
|
|
client.load_system_host_keys(str(args.known_hosts))
|
|
client.set_missing_host_key_policy(paramiko.RejectPolicy())
|
|
try:
|
|
client.connect(args.host, port=args.port, username=args.user, password=os.environ.get(args.password_env),
|
|
timeout=15, auth_timeout=15, banner_timeout=15)
|
|
with client.open_sftp() as sftp:
|
|
result = fetch(sftp, args.destination, args.receipt_sha256)
|
|
print(json.dumps({key: result[key] for key in ('fetch_complete', 'publication_receipt_sha256',
|
|
'declared_audit_status', 'file_count', 'fetch_success_is_not_scientific_pass')}), flush=True)
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|