141 lines
9.1 KiB
Python
141 lines
9.1 KiB
Python
"""Read archive/original bytes only; never import or execute archived helpers."""
|
|
from pathlib import Path, PurePosixPath, PureWindowsPath
|
|
from collections import Counter
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import zipfile
|
|
|
|
root = Path(__file__).resolve().parents[2]
|
|
folder = Path(__file__).resolve().parent
|
|
receipt = json.loads((folder / 'archive-result.json').read_bytes())
|
|
raw = (folder / receipt['archive']).read_bytes()
|
|
sha = lambda b: hashlib.sha256(b).hexdigest()
|
|
failures = []
|
|
checks = 0
|
|
def check(ok, message):
|
|
global checks
|
|
checks += 1
|
|
if not ok:
|
|
failures.append(message)
|
|
def safe(name):
|
|
pos = PurePosixPath(name)
|
|
win = PureWindowsPath(name)
|
|
return not pos.is_absolute() and not win.is_absolute() and not win.drive and '..' not in pos.parts and '\\' not in name
|
|
def verify_zip(z, label):
|
|
names = z.namelist()
|
|
check(len(names) == len(set(names)), label + ': duplicate member')
|
|
check(z.testzip() is None, label + ': CRC')
|
|
check(all(safe(n) for n in names), label + ': unsafe member path')
|
|
check(all(not n.endswith('/') for n in names), label + ': unexpected directory member')
|
|
check(all(PurePosixPath(n).suffix.lower() not in {'.exe', '.dll', '.pyc'} for n in names), label + ': executable/cache member')
|
|
check(all(not ({'node_modules', 'target', '.git', '__pycache__', 'profile', 'profiles', 'cache', 'caches'} & set(PurePosixPath(n).parts)) for n in names), label + ': out-of-scope directory')
|
|
return names
|
|
|
|
check(sha(raw) == receipt['sha256'], 'whole archive SHA')
|
|
check(len(raw) == receipt['bytes'], 'whole archive bytes')
|
|
prefix = 'source-rename-completed-evidence/'
|
|
z = zipfile.ZipFile(io.BytesIO(raw))
|
|
names = verify_zip(z, 'outer')
|
|
check(len(names) == receipt['members'] == 77, 'outer member count')
|
|
check(all(n.startswith(prefix) for n in names), 'outer prefix')
|
|
payload = {n[len(prefix):]: z.read(n) for n in names}
|
|
check(sum(map(len, payload.values())) == receipt['payloadBytes'], 'outer payload bytes')
|
|
manifest = json.loads(payload['PAYLOAD-MANIFEST.json'])
|
|
entries = manifest['members']
|
|
check(manifest['selfExcluded'] == 'PAYLOAD-MANIFEST.json', 'manifest self exclusion')
|
|
check(len(entries) == len({e['path'] for e in entries}), 'manifest duplicate entry')
|
|
check(set(e['path'] for e in entries) == set(payload) - {'PAYLOAD-MANIFEST.json'}, 'manifest complete membership')
|
|
for entry in entries:
|
|
data = payload[entry['path']]
|
|
check(len(data) == entry['bytes'] and sha(data) == entry['sha256'], 'manifest member: ' + entry['path'])
|
|
|
|
source_maps = [
|
|
('.local/source-rename-latency-diagnosis-20260912/', 'diagnosis/'),
|
|
('.local/source-rename-preview-clipping-20260912/', 'clipping-stage/'),
|
|
('.local/source-rename-controller-independent-review-20260912/', 'reviews/'),
|
|
]
|
|
source_entries = json.loads(payload['READBACK-SOURCES.json'])
|
|
check(len(source_entries) == receipt['sourceReadbacks'] == 73, 'readback count')
|
|
original_deltas = []
|
|
for entry in source_entries:
|
|
source = entry['path'].replace('\\', '/')
|
|
check(safe(source), 'unsafe original path: ' + source)
|
|
if source.endswith('/build_archive.py'):
|
|
member = 'archive-tools/build_archive.py'
|
|
elif source in {'.local/source-rename-integration-20260912/gate.py', '.local/source-rename-integration-20260912/focused.py'}:
|
|
member = 'helpers-current-readback/' + source.rsplit('/', 1)[1]
|
|
elif source.startswith('.local/source-rename-integration-20260912/'):
|
|
member = 'runs/' + source.removeprefix('.local/source-rename-integration-20260912/')
|
|
else:
|
|
mapped = [target + source.removeprefix(base) for base, target in source_maps if source.startswith(base)]
|
|
check(len(mapped) == 1, 'original target mapping: ' + source)
|
|
if not mapped:
|
|
continue
|
|
member = mapped[0]
|
|
archived = payload[member]
|
|
check(len(archived) == entry['bytes'] and sha(archived) == entry['sha256'], 'readback archive binding: ' + member)
|
|
actual = (root / source).read_bytes()
|
|
if actual != archived:
|
|
original_deltas.append({'path': source, 'currentSha256': sha(actual), 'archiveSha256': sha(archived)})
|
|
check(not original_deltas, 'original source byte differences')
|
|
|
|
summary = json.loads(payload['SUMMARY.json'])
|
|
runs = receipt['completedRuns']
|
|
check(runs == ['attempt-6', 'clipping-negative-1', 'clipping-positive-diagnostic-1', 'nested-sync-diagnostic-1', 'editor-patch-check-1'], 'completed run scope')
|
|
check({PurePosixPath(n).parts[1] for n in payload if n.startswith('runs/')} == set(runs), 'actual run membership')
|
|
nested = {}
|
|
commands = []
|
|
for run in runs:
|
|
start = f'runs/{run}/'
|
|
manifest = json.loads(payload[start + 'inputs.json'])
|
|
data = payload[start + 'inputs.zip']
|
|
check(sha(data) == manifest['archiveSha256'], run + ': input ZIP binding')
|
|
inner = zipfile.ZipFile(io.BytesIO(data))
|
|
inner_names = verify_zip(inner, run)
|
|
expected = manifest['files']
|
|
check(len(expected) == len({e['path'] for e in expected}), run + ': duplicate input manifest entry')
|
|
check(set(inner_names) == {e['path'] for e in expected}, run + ': input membership')
|
|
for entry in expected:
|
|
item = inner.read(entry['path'])
|
|
check(len(item) == entry['bytes'] and sha(item) == entry['sha256'], run + ': ' + entry['path'])
|
|
nested[run] = {'members': len(inner_names), 'sha256': sha(data), 'topLevelCounts': dict(Counter(PurePosixPath(n).parts[0] for n in inner_names))}
|
|
recorded = [r for r in summary['checks'] if r['run'] == run and 'inputArchiveSha256' in r]
|
|
check(len(recorded) == 1 and recorded[0]['inputArchiveSha256'] == sha(data) and recorded[0]['inputMembers'] == len(inner_names), run + ': summary input binding')
|
|
phases = ['clippy', 'tests', 'timing', 'gui'] if run == 'attempt-6' else ['']
|
|
for phase in phases:
|
|
stem = phase + '-' if phase else ''
|
|
result = json.loads(payload[start + stem + 'result.json'])
|
|
check(bool(result['finished']) and result['inputsUnchanged'] and result['started'] < result['finished'], run + ': completion ' + phase)
|
|
check(result['exitCode'] == (101 if run == 'clipping-negative-1' else 0), run + ': exit ' + phase)
|
|
for stream in ['stdout', 'stderr']:
|
|
check(sha(payload[start + stem + stream + '.log']) == result[stream + 'Sha256'], run + ': log ' + phase + stream)
|
|
recorded = [r for r in summary['checks'] if r['run'] == run and r.get('phase') == (phase or 'command')]
|
|
check(len(recorded) == 1 and all(recorded[0][key] == result[key] for key in ['command', 'finished', 'exitCode']), run + ': command summary ' + phase)
|
|
commands.append({'run': run, 'phase': phase or 'command', 'exitCode': result['exitCode']})
|
|
if run == 'editor-patch-check-1':
|
|
corrected = inner.read('crates/oa-desktop/src/app/source_rename/editor_patch.rs')
|
|
check(sha(corrected) == summary['correctedHelperFromCheckInputSha256'] == '049c9774c1ca28cda01f9a8e024604bef282366aa52c53c17c1918dbd09c3441', 'corrected helper binding')
|
|
check(len(commands) == receipt['commandReceipts'] == 8, 'command count')
|
|
check(len(nested) == receipt['inputArchivesVerified'] == 5, 'input archive count')
|
|
complete = json.loads(payload['runs/attempt-6/complete.json'])
|
|
check(complete['inputsUnchanged'] and complete['gates'] == ['clippy', 'tests', 'timing', 'gui'], 'gate6 closure')
|
|
check(sha(payload['diagnosis/candidate/editor_patch.rs']) == 'f301b8a043916b69dccb176da527a50ee85124b7e8b1da18921e4a0374a5153b', 'original candidate binding')
|
|
check('source glyph under scrollbar' in payload['runs/clipping-negative-1/stderr.log'].decode() and 'cluster=92..93' in payload['runs/clipping-negative-1/stderr.log'].decode(), 'negative glyph failure receipt')
|
|
check(json.loads(payload['diagnosis/nested-summary.json'])['logSha256'] == sha(payload['runs/nested-sync-diagnostic-1/stdout.log']), 'nested timing summary log binding')
|
|
for entry in json.loads(payload['clipping-stage/staged-hashes.json']):
|
|
member = 'clipping-stage/' + PureWindowsPath(entry['Path']).name
|
|
check(sha(payload[member]) == entry['Hash'].lower(), 'staged layout source binding: ' + member)
|
|
external = []
|
|
for entry in summary['externalBindings']:
|
|
data = (root / entry['path']).read_bytes()
|
|
check(not entry['included'], 'external mistakenly marked included')
|
|
check(len(data) == entry['bytes'] and sha(data) == entry['sha256'], 'external binding: ' + entry['path'])
|
|
external.append({'path': entry['path'], 'sha256': sha(data), 'bytes': len(data)})
|
|
binary = json.loads(payload['runs/attempt-6/gui-result.json'])['binary']
|
|
check(external[1]['sha256'] == binary['sha256'] and external[1]['bytes'] == binary['bytes'], 'external binary gate receipt')
|
|
|
|
result = {'status': 'PASS' if not failures else 'FAIL', 'checks': checks, 'archiveSha256': sha(raw), 'bytes': len(raw), 'members': len(names), 'payloadBytes': sum(map(len, payload.values())), 'nestedInputs': nested, 'commandReceipts': commands, 'originalReadbacks': len(source_entries), 'originalDeltas': original_deltas, 'externalBindings': external, 'failures': failures, 'scope': 'Independent byte/CRC/membership and receipt verification only; no archived or production code executed.'}
|
|
(folder / 'independent-verification.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
|
|
print(json.dumps(result, indent=2))
|