"""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()