qishizhe/digital-drosophila
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
digital-drosophila/tools/acceleration-publication/archive_results.py
T

184 lines
9.0 KiB
Python
Raw Permalink Blame History

"""Read-only source packing and streamed verification for exact replay evidence."""
import argparse
from datetime import datetime, timezone
import hashlib
import io
import json
from pathlib import Path
import re
import shutil
import subprocess
import tarfile
SOURCE_CODE = Path('/home/qishizhe/digital-drosophila-v100-acceleration/models/male-cns-lif')
GROUPS = {
'gpu': (Path('/home/qishizhe/phase4ar-v100-acceleration-20261001'),
'6efdec34c96cdc79f786b91bd7a99b0e8ef64d0c4dcfa525646c82659a77d44f',
'conditioning-acceleration-v1'),
'pipeline': (Path('/home/qishizhe/phase4ar-pipeline-acceleration-20261001'),
'4c598f51e927b3145891db42d9bb345c52d4aaa3b3a05ac257b207fff90cace4',
'conditioning-pipeline-acceleration-v1'),
}
REGISTRY = Path('registry/acceleration-20261001')
ARTIFACTS = Path('artifacts/acceleration-20261001')
INSTALL_LOGS = {
'create-venv.stdout.log', 'create-venv.stderr.log',
'install-candidate.stdout.log', 'install-candidate.stderr.log',
'install-candidate-attempt-02.stdout.log', 'install-candidate-attempt-02.stderr.log',
'upgrade-candidate-pip.stdout.log', 'upgrade-candidate-pip.stderr.log',
}
SECRET_RULES = [
rb'-----BEGIN (?:OPENSSH |RSA |EC )?PRIVATE KEY-----',
rb'(?:gh[pousr]_[A-Za-z0-9]{30,}|github_pat_[A-Za-z0-9_]{30,}|AKIA[A-Z0-9]{16})',
rb'https?://[^\s/]+:[^\s/@]+@',
]
def require(value, message):
if not value:
raise ValueError(message)
def sha(path):
digest = hashlib.sha256()
with path.open('rb') as stream:
for block in iter(lambda: stream.read(8 << 20), b''):
digest.update(block)
return digest.hexdigest()
def ident(path):
return dict(bytes=path.stat().st_size, sha256=sha(path))
def save(path, obj):
with path.open('x') as stream:
json.dump(obj, stream, indent=2, sort_keys=True, allow_nan=False)
stream.write('\n')
def excluded(name):
if name.startswith(('cupy-cache-', 'brain-replay/build/')):
return 'Rebuildable compiler cache/SDK intermediates, not a scientific result'
if name in INSTALL_LOGS:
return 'Environment installation output; versions and wheel report retained'
return None
def scan_text(path):
if path.suffix.lower() not in ('.json', '.jsonl', '.log', '.md', '.txt', '.csv', '.py', '.cu', '.diff'):
return
require(path.stat().st_size <= 25 * (1 << 20), 'Oversized text requires manual review')
data = path.read_bytes()
for rule in SECRET_RULES:
require(not re.search(rule, data), 'Potential credential marker: ' + path.name)
def pack(root):
require(root == Path('/home/qishizhe/digital-drosophila-acceleration-publication-20261001'), 'Independent publication clone required')
require((root / '.git').is_dir() and not root.is_symlink(), 'Ordinary independent clone required')
registry, artifacts = root / REGISTRY, root / ARTIFACTS
registry.mkdir(parents=True, exist_ok=True)
artifacts.mkdir(parents=True, exist_ok=False)
catalog = dict(schema='acceleration-publication-v1', created_utc=datetime.now(timezone.utc).isoformat(),
numerical_changes=False, training_resumed=False, complete_training_restore=False,
scope='Two completed independent engineering experiments, not the full stopped scientific run', groups={})
for group, (source, expected_manifest, code_dir) in GROUPS.items():
require(sha(source / 'artifact-manifest.json') == expected_manifest, 'Sealed source manifest changed')
manifest = json.loads((source / 'artifact-manifest.json').read_text())
for name, expected in manifest['source_files'].items():
require(sha(SOURCE_CODE / code_dir / name) == expected, 'Sealed source code changed')
require(sha(root / 'models/male-cns-lif' / code_dir / name) == expected, 'Published source differs')
members, omitted = {}, {}
for name, expected in manifest['files'].items():
path = source / name
require(path.resolve().is_relative_to(source) and not path.is_symlink(), 'Unsafe source path')
require(ident(path) == expected, 'Sealed input file changed: ' + name)
reason = excluded(name)
if reason:
omitted[name] = dict(expected, reason=reason)
else:
scan_text(path)
members[group + '/' + name] = dict(expected, source_relative=name)
for name in ('artifact-manifest.json', 'SHA256SUMS'):
require(group + '/' + name not in members, 'Duplicate source control file')
members[group + '/' + name] = dict(ident(source / name), source_relative=name)
archive = artifacts / (group + '.tar.zst')
with archive.open('xb') as destination:
compressor = subprocess.Popen(['zstd', '-q', '-T2', '-3', '-c'], stdin=subprocess.PIPE, stdout=destination)
try:
with tarfile.open(fileobj=compressor.stdin, mode='w|', format=tarfile.PAX_FORMAT) as tar:
for name, item in sorted(members.items()):
path = source / item['source_relative']
info = tarfile.TarInfo(name)
info.size, info.mode, info.mtime = item['bytes'], 0o644, 0
with path.open('rb') as stream:
tar.addfile(info, stream)
require(ident(path) == {k: item[k] for k in ('bytes', 'sha256')}, 'Input changed during packing')
compressor.stdin.close()
require(compressor.wait(timeout=120) == 0, 'Compression failed')
finally:
if compressor.poll() is None:
compressor.kill()
compressor.wait()
target = registry / group
target.mkdir(exist_ok=False)
for name in ('RESULTS.md', 'final-summary.json', 'artifact-manifest.json', 'SHA256SUMS'):
shutil.copyfile(source / name, target / name)
require(sha(source / name) == sha(target / name), 'Summary copy differs')
catalog['groups'][group] = dict(source_root=str(source), source_manifest_sha256=expected_manifest,
source_files=manifest['source_files'], archive=str(archive.relative_to(root)), **ident(archive),
member_count=len(members), member_bytes=sum(x['bytes'] for x in members.values()),
members=members, excluded=omitted, excluded_count=len(omitted))
save(registry / 'publication-catalog.json', catalog)
result = verify(root)
save(registry / 'pack-verification.json', result)
print(json.dumps(dict(complete=True, groups=result['groups'], source_files_modified=False), indent=2))
def verify(root):
catalog_path = root / REGISTRY / 'publication-catalog.json'
catalog = json.loads(catalog_path.read_text())
require(catalog['schema'] == 'acceleration-publication-v1', 'Wrong publication catalog')
results = {}
for group, info in catalog['groups'].items():
archive = root / info['archive']
require(archive.resolve().is_relative_to(root / ARTIFACTS) and not archive.is_symlink(), 'Unsafe archive path')
require(ident(archive) == {k: info[k] for k in ('bytes', 'sha256')}, 'Archive identity mismatch')
decoder = subprocess.Popen(['zstd', '-q', '-d', '-c', str(archive)], stdout=subprocess.PIPE)
found, total = set(), 0
try:
with tarfile.open(fileobj=decoder.stdout, mode='r|') as tar:
for entry in tar:
require(entry.isfile() and entry.name in info['members'] and entry.name not in found, 'Unexpected/duplicate member')
expected = info['members'][entry.name]
require(entry.size == expected['bytes'], 'Member size mismatch')
h, count = hashlib.sha256(), 0
with tar.extractfile(entry) as stream:
for block in iter(lambda: stream.read(8 << 20), b''):
h.update(block)
count += len(block)
require(count == entry.size and h.hexdigest() == expected['sha256'], 'Member hash mismatch')
found.add(entry.name)
total += count
while decoder.stdout.read(1 << 20):
pass
require(decoder.wait(timeout=30) == 0 and found == set(info['members']), 'Archive truncated or member set differs')
finally:
decoder.stdout.close()
if decoder.poll() is None:
decoder.kill()
decoder.wait()
results[group] = dict(files=len(found), original_bytes=total, archive_bytes=info['bytes'], sha256=info['sha256'], verified=True)
return dict(pass_all=True, groups=results, catalog_sha256=sha(catalog_path), verification='Full streamed members, no extraction or model execution')
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('mode', choices=('pack', 'verify'))
args = parser.parse_args()
if args.mode == 'pack':
pack(Path.cwd())
else:
print(json.dumps(verify(Path.cwd()), indent=2))
Reference in New Issue View Git Blame Copy Permalink