Publish mixed-syntax evidence and State multiselect audit
This commit is contained in:
@@ -0,0 +1,211 @@
|
||||
"""Real-package verifier controls. Never execute a subject or archived helper."""
|
||||
from pathlib import Path
|
||||
import argparse
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import warnings
|
||||
import zipfile
|
||||
|
||||
VERIFIER = '2902f0c5e4922b942b3e8b9d0b288810ef6ad8dc71444d28ef8bfa756c4c5338'
|
||||
SOURCE = 'a578ac98a06772a60bec00ae287e65650680d119'
|
||||
FULL = 'gates/mixed-direct-full-a578-1'
|
||||
ORIGINAL = 'gates/mixed-list-a578-1.result.json'
|
||||
|
||||
|
||||
def pin(path):
|
||||
raw = path.read_bytes()
|
||||
return {'bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()}
|
||||
|
||||
|
||||
def encode(value):
|
||||
return (json.dumps(value, indent=2, ensure_ascii=False) + '\n').encode()
|
||||
|
||||
|
||||
def exclusive_json(path, value):
|
||||
with path.open('xb') as stream:
|
||||
stream.write(encode(value))
|
||||
|
||||
|
||||
def main():
|
||||
if sys.flags.optimize:
|
||||
raise SystemExit('Optimized Python is not valid for these assertion-based controls')
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--repo', type=Path, required=True)
|
||||
parser.add_argument('--package', type=Path, required=True)
|
||||
parser.add_argument('--verifier', type=Path, required=True)
|
||||
parser.add_argument('--out', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
repo, package, verifier, out = [p.resolve() for p in (args.repo, args.package, args.verifier, args.out)]
|
||||
assert pin(verifier)['sha256'] == VERIFIER, 'Use the separately reviewed external verifier'
|
||||
assert verifier.parent != package and not verifier.is_relative_to(out)
|
||||
out.mkdir(exist_ok=False)
|
||||
original_pins = {str(p): pin(p) for p in package.iterdir() if p.is_file()}
|
||||
published = repo / 'harness/experiments/analysis-mixed-syntax-20260923'
|
||||
original_pins.update({str(p): pin(p) for p in published.iterdir() if p.is_file()})
|
||||
gates = repo / '.local/analysis-mixed-syntax-gates-20260923'
|
||||
original_pins.update({str(p): pin(p) for p in gates.rglob('*') if p.is_file()})
|
||||
verifier_pin = pin(verifier)
|
||||
original_pins[str(verifier)] = verifier_pin
|
||||
with (out / 'control-script.py').open('xb') as stream:
|
||||
stream.write(Path(__file__).read_bytes())
|
||||
with (out / 'reviewed-verifier.py').open('xb') as stream:
|
||||
stream.write(verifier.read_bytes())
|
||||
with zipfile.ZipFile(package / 'evidence.zip') as archive:
|
||||
assert len(archive.namelist()) == len(set(archive.namelist()))
|
||||
original_payload = {name: archive.read(name) for name in archive.namelist()}
|
||||
original_manifest = json.loads((package / 'manifest.json').read_bytes())
|
||||
new_test = next(n for n in original_manifest['verified_summary']['new_42_passes']
|
||||
if n.startswith('language::analysis_source::mixed::'))
|
||||
cases = [
|
||||
('unchanged-real-package', None),
|
||||
('missing-new-outcome', 'Missing/extra runtime outcomes'),
|
||||
('changed-new-outcome', 'Runtime status mismatch'),
|
||||
('changed-known-failure-cause', 'Failure cause changed'),
|
||||
('forged-current-git-input', 'Git input bytes mismatch'),
|
||||
('duplicate-zip-member', 'Duplicate ZIP members'),
|
||||
('missing-intermediate-strict', 'Required checkpoint gates missing/ambiguous'),
|
||||
('changed-final-formatting-location', 'Final formatting failure scope changed'),
|
||||
('changed-clock-receipt', 'Direct historical binding differs'),
|
||||
('aliased-exact-clock-receipt', 'Gate not closed'),
|
||||
('direct-extra-filter-argv', 'Unexpected direct executable argv'),
|
||||
('direct-old-before-artifact', 'Direct artifact changed or differs'),
|
||||
('direct-missing-bound-postcheck', 'Direct bound-file postcheck coverage differs'),
|
||||
('direct-fake-cargo-line', 'Direct output contains a fabricated Cargo invocation line'),
|
||||
('direct-recorded-launch-failure', 'Direct invocation did not close cleanly'),
|
||||
('missing-direct-listing', 'gates/mixed-direct-list-a578-1.result.json'),
|
||||
]
|
||||
results = []
|
||||
for label, refusal in cases:
|
||||
case = out / label
|
||||
case.mkdir()
|
||||
payload, manifest = dict(original_payload), copy.deepcopy(original_manifest)
|
||||
duplicate = None
|
||||
detail = ''
|
||||
|
||||
def change_receipt(name, update):
|
||||
value = json.loads(payload[name])
|
||||
update(value)
|
||||
payload[name] = encode(value)
|
||||
|
||||
def change_log(stem, transform):
|
||||
name = stem + '.log'
|
||||
payload[name] = transform(payload[name])
|
||||
change_receipt(stem + '.result.json', lambda r: r.update(log_sha256=hashlib.sha256(payload[name]).hexdigest()))
|
||||
|
||||
def remove_gate(stem):
|
||||
for suffix in ('.result.json', '.inputs.json', '.log'):
|
||||
del payload[stem + suffix]
|
||||
referenced = set()
|
||||
for name, raw in payload.items():
|
||||
if name.startswith('gates/') and name.endswith('.result.json'):
|
||||
r = json.loads(raw)
|
||||
referenced.add('gates/' + r['launcher_copy'].replace('\\', '/'))
|
||||
for key in ('bound_original_listing', 'bound_focused', 'bound_listing'):
|
||||
if isinstance(r.get(key), dict):
|
||||
referenced.add('gates/' + r[key]['launcher_copy'].replace('\\', '/'))
|
||||
for name in list(payload):
|
||||
if name.startswith('gates/launchers/') and name not in referenced:
|
||||
del payload[name]
|
||||
|
||||
if label in ('missing-new-outcome', 'changed-new-outcome'):
|
||||
pattern = rb'(?m)^test ' + re.escape(new_test.encode()) + rb' \.\.\. ok\r?\n'
|
||||
assert len(re.findall(pattern, payload[FULL + '.log'])) == 1
|
||||
replacement = b'' if label == 'missing-new-outcome' else ('test ' + new_test + ' ... FAILED\n').encode()
|
||||
change_log(FULL, lambda raw: re.sub(pattern, replacement, raw, count=1))
|
||||
detail = new_test
|
||||
elif label == 'changed-known-failure-cause':
|
||||
old = b'Unsupported schema version: expected arcadia 3, 4, 5, 6 or 7.'
|
||||
assert payload[FULL + '.log'].count(old) == 16
|
||||
change_log(FULL, lambda raw: raw.replace(old, b'Different semantic failure for this control.', 1))
|
||||
elif label == 'forged-current-git-input':
|
||||
inputs = []
|
||||
for name, raw in list(payload.items()):
|
||||
if name.startswith('gates/') and name.endswith('.inputs.json') and json.loads(raw)['source'] == SOURCE:
|
||||
value = json.loads(raw)
|
||||
row = next(r for r in value['inputs'] if r['path'] == 'crates/oa-core/src/language.rs')
|
||||
row['sha256'] = '0' * 64
|
||||
payload[name] = encode(value)
|
||||
inputs.append(name)
|
||||
for name in inputs:
|
||||
digest = hashlib.sha256(payload[name]).hexdigest()
|
||||
change_receipt(name[:-len('.inputs.json')] + '.result.json', lambda r, d=digest: r.update(inputs_sha256=d))
|
||||
detail = 'All ' + str(len(inputs)) + ' final-source input copies and their receipt hashes changed; immutable Git bytes unchanged.'
|
||||
elif label == 'duplicate-zip-member':
|
||||
duplicate = FULL + '.log'
|
||||
elif label == 'missing-intermediate-strict':
|
||||
remove_gate('gates/mixed-clippy-eeb1-1')
|
||||
elif label == 'changed-final-formatting-location':
|
||||
stem = 'gates/mixed-fmt-a578-1'
|
||||
assert re.search(rb'(?m)^Diff in [^\r\n]+:\d+:\r?$', payload[stem + '.log'])
|
||||
change_log(stem, lambda raw: re.sub(rb'(?m)^(Diff in [^\r\n]+:)(\d+)(:\r?)$',
|
||||
lambda m: m[1] + str(int(m[2]) + 1).encode() + m[3], raw, count=1))
|
||||
elif label == 'changed-clock-receipt':
|
||||
payload[ORIGINAL] += b'\n' # Same JSON values, different exact receipt bytes.
|
||||
elif label == 'aliased-exact-clock-receipt':
|
||||
payload['gates/mixed-list-a578-clock-alias.result.json'] = payload[ORIGINAL]
|
||||
detail = 'Keep original for direct bindings; identical reversed-clock receipt under a second name must not inherit exception.'
|
||||
elif label == 'direct-extra-filter-argv':
|
||||
change_receipt(FULL + '.result.json', lambda r: r['command'].append('analysis_source'))
|
||||
elif label == 'direct-old-before-artifact':
|
||||
change_receipt(FULL + '.result.json', lambda r: r['artifact_before'].update(
|
||||
sha256='ae4974da9a58c487e1fe1a4d1461743b106fb7495f3c5a84a20a8999cc9a7ce2'))
|
||||
elif label == 'direct-missing-bound-postcheck':
|
||||
change_receipt(FULL + '.result.json', lambda r: r['bound_files_after'].pop())
|
||||
elif label == 'direct-fake-cargo-line':
|
||||
path = json.loads(payload[FULL + '.result.json'])['artifacts'][0]['path']
|
||||
change_log(FULL, lambda raw: ('Running unittests src/lib.rs (' + path + ')\n').encode() + raw)
|
||||
elif label == 'direct-recorded-launch-failure':
|
||||
change_receipt(FULL + '.result.json', lambda r: r.update(launch_error='control launch failure',
|
||||
verification_failures=['Executable launch failed']))
|
||||
elif label == 'missing-direct-listing':
|
||||
remove_gate('gates/mixed-direct-list-a578-1')
|
||||
|
||||
if refusal is None:
|
||||
(case / 'evidence.zip').write_bytes((package / 'evidence.zip').read_bytes())
|
||||
(case / 'manifest.json').write_bytes((package / 'manifest.json').read_bytes())
|
||||
else:
|
||||
manifest['payloads'] = {name: {'bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()}
|
||||
for name, raw in sorted(payload.items())}
|
||||
with zipfile.ZipFile(case / 'evidence.zip', 'x', compression=zipfile.ZIP_DEFLATED) as archive:
|
||||
for name, raw in sorted(payload.items()):
|
||||
archive.writestr(name, raw)
|
||||
if duplicate:
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter('ignore', UserWarning)
|
||||
archive.writestr(duplicate, payload[duplicate])
|
||||
exclusive_json(case / 'manifest.json', manifest)
|
||||
command = [sys.executable, '-B', str(verifier), '--repo', str(repo), '--package', str(case)]
|
||||
started = time.monotonic()
|
||||
with (case / 'stdout.log').open('xb') as stdout, (case / 'stderr.log').open('xb') as stderr:
|
||||
process = subprocess.run(command, cwd=repo, stdout=stdout, stderr=stderr, timeout=180, check=False)
|
||||
elapsed = time.monotonic() - started
|
||||
output = (case / 'stdout.log').read_text() + (case / 'stderr.log').read_text()
|
||||
accepted = process.returncode == 0
|
||||
matched = (accepted if refusal is None else not accepted and refusal in output and 'REFUSED:' in output)
|
||||
result = {'case': label, 'expected_accept': refusal is None, 'observed_accept': accepted,
|
||||
'exit': process.returncode, 'elapsed_monotonic_seconds': elapsed,
|
||||
'expected_refusal_fragment': refusal, 'actual_refusal': output.strip() if not accepted else None,
|
||||
'matched': matched, 'detail': detail, 'command': command,
|
||||
'files': {p.name: pin(p) for p in case.iterdir() if p.is_file()}}
|
||||
exclusive_json(case / 'control-result.json', result)
|
||||
results.append(result)
|
||||
print(label + ': ' + ('EXPECTED ACCEPT' if accepted else output.strip()), flush=True)
|
||||
assert matched, 'Unexpected verifier outcome: ' + label
|
||||
assert all(pin(Path(path)) == expected for path, expected in original_pins.items()), 'Original evidence changed'
|
||||
exclusive_json(out / 'controls.json', {
|
||||
'status': 'PASS_REAL_ARCHIVE_POSITIVE_AND_NEGATIVE_CONTROLS',
|
||||
'scope': 'Receipt/Git verifier controls only; no subject, compiler or archived helper executed.',
|
||||
'original_package': str(package), 'external_verifier': {'path': str(verifier), **verifier_pin},
|
||||
'control_script': pin(Path(__file__)), 'python': {'path': sys.executable, **pin(Path(sys.executable))},
|
||||
'original_evidence_unchanged': original_pins, 'cases': results,
|
||||
})
|
||||
print('PASS ' + str(len(results)) + '/' + str(len(cases)) + ' actual-package controls; original evidence unchanged.', flush=True)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user