114 lines
5.6 KiB
Python
114 lines
5.6 KiB
Python
"""Audit the accompanying frozen-source gate archive without running its commands."""
|
|
from pathlib import Path
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import zipfile
|
|
|
|
|
|
def sha(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def verify(folder, repo):
|
|
manifest = json.loads((folder / 'manifest.json').read_bytes())
|
|
archive = (folder / 'evidence.zip').read_bytes()
|
|
assert len(archive) == manifest['archive_bytes']
|
|
assert sha(archive) == manifest['archive_sha256']
|
|
with zipfile.ZipFile(folder / 'evidence.zip') as z:
|
|
assert len(z.namelist()) == len(set(z.namelist()))
|
|
assert set(z.namelist()) == set(manifest['payloads'])
|
|
data = {name: z.read(name) for name in z.namelist()}
|
|
for name, raw in data.items():
|
|
expected = manifest['payloads'][name]
|
|
assert len(raw) == expected['bytes'] and sha(raw) == expected['sha256'], name
|
|
|
|
source_inputs = {}
|
|
gates = {}
|
|
for claim in manifest['gates']:
|
|
name = claim['name']
|
|
r = json.loads(data[f'gates/{name}.result.json'])
|
|
raw = data[f'gates/{name}.log']
|
|
input_raw = data['gates/' + r['inputs']]
|
|
inputs = json.loads(input_raw)
|
|
assert sha(raw) == r['log_sha256']
|
|
assert sha(input_raw) == r['inputs_sha256']
|
|
assert r['source'] == r['head_after'] == claim['source'] == inputs['source']
|
|
assert r['changed_inputs'] == []
|
|
assert r['exit'] == claim['exit']
|
|
assert r['command'] == claim['command']
|
|
assert r['mode'] == claim['mode'] and r['scope'] == claim['scope']
|
|
assert r['launcher_sha256'] == r['launcher_sha256_after']
|
|
launcher = data['launchers/' + r['launcher_sha256'] + '.py']
|
|
assert sha(launcher) == r['launcher_sha256']
|
|
assert len(inputs['inputs']) == claim['inputs']
|
|
paths = [item['path'] for item in inputs['inputs']]
|
|
assert len(paths) == len(set(paths))
|
|
for item in inputs['inputs']:
|
|
key = (r['source'], item['path'])
|
|
assert key not in source_inputs or source_inputs[key] == item
|
|
source_inputs[key] = item
|
|
output = raw.decode('utf-8', errors='strict')
|
|
if r['mode'] == 'list' and r['exit'] == 0:
|
|
names = re.findall(r'^([^\r\n]+): test\r?$', output, re.M)
|
|
assert names and len(names) == len(set(names))
|
|
assert names == r['listing_members']
|
|
assert len(names) == r['listing_count'] == claim['listed']
|
|
assert len(r['artifacts']) == 1
|
|
if r['mode'] == 'clippy' and r['exit'] == 0:
|
|
assert re.search(r'Finished `dev` profile', output)
|
|
assert not re.search(r'^error(?:\[|:)', output, re.M)
|
|
gates[name] = (r, output, input_raw)
|
|
|
|
tests = []
|
|
for claim in manifest['gates']:
|
|
if 'listing' not in claim:
|
|
continue
|
|
r, output, input_raw = gates[claim['name']]
|
|
listing, _, listed_inputs = gates[claim['listing']]
|
|
assert r['mode'] == 'test' and listing['mode'] == 'list'
|
|
assert r['source'] == listing['source'] and r['scope'] == listing['scope']
|
|
assert input_raw == listed_inputs and r['artifacts'] == listing['artifacts']
|
|
selected = {name for name in listing['listing_members'] if r['filter'] in name}
|
|
assert selected
|
|
outcomes = re.findall(r'^test ([^\r\n]+) \.\.\. (ok|FAILED|ignored)(?:, ([^\r\n]*))?\r?$', output, re.M)
|
|
assert len(outcomes) == len({name for name, _, _ in outcomes}) == len(selected)
|
|
assert {name for name, _, _ in outcomes} == selected
|
|
counts = [sum(status == wanted for _, status, _ in outcomes) for wanted in ('ok', 'FAILED', 'ignored')]
|
|
assert counts == claim['outcomes']
|
|
summaries = re.findall(r'^test result: (?:ok|FAILED)\. (\d+) passed; (\d+) failed; (\d+) ignored; (\d+) measured; (\d+) filtered out;', output, re.M)
|
|
assert summaries == [tuple(map(str, [*counts, 0, len(listing['listing_members']) - len(selected)]))]
|
|
ignores = [{'name': name, 'reason': reason} for name, status, reason in outcomes if status == 'ignored']
|
|
assert ignores == claim.get('ignored', [])
|
|
tests.append({'name': claim['name'], 'outcomes': counts, 'selected': len(selected)})
|
|
|
|
keys = sorted(source_inputs)
|
|
run = subprocess.run(['git', 'cat-file', '--batch'], cwd=repo,
|
|
input=''.join(f'{commit}:{path}\n' for commit, path in keys).encode(),
|
|
capture_output=True, check=True)
|
|
offset = 0
|
|
for key in keys:
|
|
end = run.stdout.index(b'\n', offset)
|
|
fields = run.stdout[offset:end].split()
|
|
assert len(fields) == 3 and fields[1] == b'blob', key
|
|
size = int(fields[2])
|
|
blob = run.stdout[end + 1:end + 1 + size]
|
|
expected = source_inputs[key]
|
|
assert size == expected['bytes'] and sha(blob) == expected['sha256'], key
|
|
offset = end + 2 + size
|
|
assert offset == len(run.stdout)
|
|
return {'payloads': len(data), 'gates': len(gates), 'source_inputs': len(keys), 'tests': tests,
|
|
'status': 'ARCHIVE_AND_RECORDED_SCOPE_VERIFIED',
|
|
'limitation': 'Does not rerun commands, reconstruct executable bytes, establish native behavior, or accept a release. Historical failures and launcher-capture limits remain in README.'}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if not __debug__:
|
|
raise SystemExit('Optimized Python is not supported')
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--repo', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
print(json.dumps(verify(Path(__file__).resolve().parent, args.repo), indent=2))
|