250 lines
12 KiB
Python
250 lines
12 KiB
Python
"""Opt-in semantic controls for the reviewed v2 verifier; never execute archived code.
|
|
|
|
Creates a NEW control directory, retains every case, and runs only copied verify.py
|
|
with read-only Git access. Ordinary archive/payload/reference hashes are repaired.
|
|
Do not run until the parent has reviewed this driver and the successor verifier.
|
|
"""
|
|
from pathlib import Path
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import ntpath
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import zipfile
|
|
|
|
LOCAL = Path(__file__).resolve().parent
|
|
PACKAGE = LOCAL / 'peer-union-e34-evidence-proposal-v2-20260923'
|
|
FULL = 'gates/combined-workspace-full-e34-1'
|
|
OLD = 'gates/combined-desktop-diagnostic-1'
|
|
SUPPLEMENT = 'supplement/combined-rebuilt-inventory-e34-2/'
|
|
|
|
|
|
def sha(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def pin(raw):
|
|
return {'bytes': len(raw), 'sha256': sha(raw)}
|
|
|
|
|
|
def read(data, name):
|
|
return json.loads(data[name])
|
|
|
|
|
|
def write(data, name, value):
|
|
data[name] = (json.dumps(value, indent=2) + '\n').encode()
|
|
|
|
|
|
def parse(text):
|
|
return {'listing_members': re.findall(r'^([^\r\n]+): test\r?$', text, re.M),
|
|
'outcomes': [list(t) for t in re.findall(r'^test ([^\r\n]+) \.\.\. (ok|FAILED|ignored)(?:, ([^\r\n]*))?\r?$', text, re.M)],
|
|
'summaries': [list(t) for t in re.findall(r'^test result: (ok|FAILED)\. (\d+) passed; (\d+) failed; (\d+) ignored; (\d+) measured; (\d+) filtered out;', text, re.M)],
|
|
'listing_totals': [list(t) for t in re.findall(r'^(\d+) tests?, (\d+) benchmarks?\r?$', text, re.M)],
|
|
'text_sha256': sha(text.encode())}
|
|
|
|
|
|
def refresh_full(data, manifest, update_policy=False):
|
|
text = data[FULL + '.log'].decode()
|
|
headers = list(re.finditer(r'^\s+(?:Running (?P<kind>[^\r\n]+) \((?P<exe>[^\r\n]+\.exe)\)|Doc-tests (?P<doc>[^\r\n]+))\r?$', text, re.M))
|
|
sections = []
|
|
for index, match in enumerate(headers):
|
|
body = text[match.end():headers[index + 1].start() if index + 1 < len(headers) else len(text)]
|
|
sections.append({'key': ntpath.normpath(match['exe']) if match['exe'] else 'doctest:' + match['doc'],
|
|
'kind': match['kind'] or 'doctest', **parse(body)})
|
|
result = read(data, FULL + '.result.json')
|
|
result['workspace_sections'] = sections
|
|
write(data, FULL + '.result.json', result)
|
|
if update_policy:
|
|
manifest['ignored'] = [{'section': section['key'], 'name': n, 'reason': reason}
|
|
for section in sections for n, status, reason in section['outcomes'] if status == 'ignored']
|
|
|
|
|
|
def replace_once(raw, old, new):
|
|
assert raw.count(old) == 1, old
|
|
return raw.replace(old, new, 1)
|
|
|
|
|
|
def repair_references(original, data):
|
|
"""Repair ordinary payload-digest references transitively, preserving other bytes.
|
|
|
|
This deliberately does not repair policy constants or consistency statements:
|
|
aggregate/leaf contradictions and mutated outcome identities must remain visible.
|
|
"""
|
|
history = {name: {sha(raw)} for name, raw in original.items()}
|
|
for _ in range(16):
|
|
replacements = {}
|
|
for name, raw in data.items():
|
|
current = pin(raw)
|
|
for old_hash in history[name]:
|
|
if old_hash != current['sha256']:
|
|
assert old_hash not in replacements or replacements[old_hash] == current
|
|
replacements[old_hash] = current
|
|
history[name].add(current['sha256'])
|
|
|
|
def repaired(value):
|
|
if isinstance(value, dict):
|
|
out = {k: repaired(v) for k, v in value.items()}
|
|
if value.get('sha256') in replacements and 'bytes' in value:
|
|
out['bytes'] = replacements[value['sha256']]['bytes']
|
|
return out
|
|
if isinstance(value, list):
|
|
return [repaired(v) for v in value]
|
|
if isinstance(value, str) and value in replacements:
|
|
return replacements[value]['sha256']
|
|
return value
|
|
|
|
changed = False
|
|
for name in data:
|
|
if not name.endswith('.json'):
|
|
continue
|
|
before = read(data, name)
|
|
after = repaired(before)
|
|
if after != before:
|
|
write(data, name, after)
|
|
changed = True
|
|
if not changed:
|
|
return
|
|
raise AssertionError('Digest repair did not converge')
|
|
|
|
|
|
def mutate(kind, data, manifest):
|
|
if kind == 'historical_failed_identity':
|
|
raw = data[OLD + '.log']
|
|
rows = parse(raw.decode())['outcomes']
|
|
bad = next(n for n, status, _ in rows if status == 'FAILED')
|
|
good = next(n for n, status, _ in rows if status == 'ok')
|
|
raw = replace_once(raw, f'test {bad} ... FAILED'.encode(), f'test {bad} ... ok'.encode())
|
|
data[OLD + '.log'] = replace_once(raw, f'test {good} ... ok'.encode(), f'test {good} ... FAILED'.encode())
|
|
elif kind == 'historical_panic_details':
|
|
raw = data[OLD + '.log']
|
|
line = next(line for line in raw.splitlines(keepends=True) if b'panicked at' in line)
|
|
data[OLD + '.log'] = replace_once(raw, line, b'')
|
|
elif kind in ('ignored_identity', 'ignored_reason'):
|
|
raw = data[FULL + '.log']
|
|
r = read(data, FULL + '.result.json')
|
|
section = next(s for s in r['workspace_sections'] if any(status == 'ignored' for _, status, _ in s['outcomes']))
|
|
bad, _, reason = next(row for row in section['outcomes'] if row[1] == 'ignored')
|
|
if kind == 'ignored_identity':
|
|
good = next(n for n, status, _ in section['outcomes'] if status == 'ok')
|
|
raw = replace_once(raw, f'test {bad} ... ignored, {reason}'.encode(), f'test {bad} ... ok'.encode())
|
|
raw = replace_once(raw, f'test {good} ... ok'.encode(), f'test {good} ... ignored, {reason}'.encode())
|
|
else:
|
|
raw = replace_once(raw, f'test {bad} ... ignored, {reason}'.encode(), f'test {bad} ... ignored, changed control reason'.encode())
|
|
data[FULL + '.log'] = raw
|
|
refresh_full(data, manifest, update_policy=True)
|
|
elif kind == 'supplement_leaf_exit':
|
|
name = SUPPLEMENT + 'arcadia-dc4fafbbc0215f88.result.json'
|
|
value = read(data, name)
|
|
value['exit'] = 101
|
|
write(data, name, value)
|
|
elif kind == 'supplement_before':
|
|
name = SUPPLEMENT + 'before.json'
|
|
value = read(data, name)
|
|
value['environment']['RUST_LOG'] = 'control-change'
|
|
write(data, name, value)
|
|
elif kind == 'rebuilt_artifact_rollback':
|
|
value = read(data, FULL + '.result.json')
|
|
listing = read(data, 'gates/combined-workspace-list-e34-1.result.json')
|
|
old = next(a for a in listing['artifacts'] if ntpath.basename(a['path']) == 'arcadia-dc4fafbbc0215f88.exe')
|
|
current = next(a for a in value['artifacts'] if a['path'] == old['path'])
|
|
assert current['sha256'] != old['sha256']
|
|
current.update(old)
|
|
write(data, FULL + '.result.json', value)
|
|
elif kind in ('outcome_membership', 'duplicate_outcome'):
|
|
raw = data[FULL + '.log']
|
|
result = read(data, FULL + '.result.json')
|
|
section = next(s for s in result['workspace_sections'] if sum(status == 'ok' for _, status, _ in s['outcomes']) >= 2)
|
|
first, second = [n for n, status, _ in section['outcomes'] if status == 'ok'][:2]
|
|
replacement = 'control::not_in_inventory' if kind == 'outcome_membership' else second
|
|
data[FULL + '.log'] = replace_once(raw, f'test {first} ... ok'.encode(), f'test {replacement} ... ok'.encode())
|
|
refresh_full(data, manifest)
|
|
elif kind == 'filtered_command':
|
|
value = read(data, FULL + '.result.json')
|
|
value['command'].insert(value['command'].index('--'), 'control_filter')
|
|
value['filter'] = 'control_filter'
|
|
write(data, FULL + '.result.json', value)
|
|
elif kind == 'source_input_digest':
|
|
changed = 0
|
|
for name in list(data):
|
|
if not name.endswith('.json'):
|
|
continue
|
|
value = read(data, name)
|
|
if (isinstance(value, dict) and value.get('source') == manifest['source']
|
|
and isinstance(value.get('inputs'), list) and len(value['inputs']) == 760):
|
|
value['inputs'][0]['sha256'] = '0' * 64
|
|
write(data, name, value)
|
|
changed += 1
|
|
assert changed >= 8
|
|
elif kind == 'native_error':
|
|
name = 'gates/combined-native-e34-1.log'
|
|
data[name] += b'error: injected semantic control compiler failure\n'
|
|
else:
|
|
raise AssertionError(kind)
|
|
|
|
|
|
CASES = (
|
|
'historical_failed_identity', 'historical_panic_details', 'ignored_identity',
|
|
'ignored_reason', 'supplement_leaf_exit', 'supplement_before',
|
|
'rebuilt_artifact_rollback', 'outcome_membership', 'duplicate_outcome',
|
|
'filtered_command', 'source_input_digest', 'native_error',
|
|
)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--repo', type=Path, required=True)
|
|
parser.add_argument('--out', type=Path, required=True, help='New directory; existing paths are rejected')
|
|
args = parser.parse_args()
|
|
assert not args.out.exists()
|
|
manifest = json.loads((PACKAGE / 'manifest.json').read_bytes())
|
|
assert pin((PACKAGE / 'verify.py').read_bytes()) == manifest['verifier']
|
|
assert pin((PACKAGE / 'evidence.zip').read_bytes()) == manifest['archive']
|
|
with zipfile.ZipFile(PACKAGE / 'evidence.zip') as archive:
|
|
assert len(archive.namelist()) == len(set(archive.namelist()))
|
|
original = {n: archive.read(n) for n in archive.namelist()}
|
|
assert {n: pin(raw) for n, raw in original.items()} == manifest['payloads']
|
|
args.out.mkdir(parents=True)
|
|
results = []
|
|
for kind in ('positive', *CASES):
|
|
case = args.out / kind
|
|
shutil.copytree(PACKAGE, case)
|
|
if kind != 'positive':
|
|
data = original.copy()
|
|
changed_manifest = copy.deepcopy(manifest)
|
|
mutate(kind, data, changed_manifest)
|
|
repair_references(original, data)
|
|
assert data != original
|
|
with zipfile.ZipFile(case / 'evidence.zip', 'w', compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
|
|
for name, raw in sorted(data.items()):
|
|
archive.writestr(name, raw)
|
|
changed_manifest['payloads'] = {n: pin(raw) for n, raw in sorted(data.items())}
|
|
changed_manifest['archive'] = pin((case / 'evidence.zip').read_bytes())
|
|
(case / 'manifest.json').write_text(json.dumps(changed_manifest, indent=2) + '\n', encoding='utf-8', newline='\n')
|
|
# The only executable under test is reviewed top-level verify.py, never ZIP contents.
|
|
command = [sys.executable, str((case / 'verify.py').resolve()), '--repo', str(args.repo.resolve())]
|
|
run = subprocess.run(command, capture_output=True)
|
|
(case / 'stdout.txt').write_bytes(run.stdout)
|
|
(case / 'stderr.txt').write_bytes(run.stderr)
|
|
if kind == 'positive':
|
|
accepted = run.returncode == 0 and json.loads(run.stdout)['status'] == 'RECORDED_E34_GATES_VERIFIED'
|
|
else:
|
|
accepted = run.returncode != 0 and b'AssertionError' in run.stderr
|
|
results.append({'case': kind, 'expected': 'accept' if kind == 'positive' else 'reject',
|
|
'exit': run.returncode, 'control_passed': accepted, 'command': command})
|
|
print(json.dumps(results[-1]), flush=True)
|
|
report = {'verifier': manifest['verifier'], 'archive': manifest['archive'],
|
|
'results': results, 'passed': all(r['control_passed'] for r in results),
|
|
'archived_code_executed': False}
|
|
(args.out / 'report.json').write_text(json.dumps(report, indent=2) + '\n', encoding='utf-8', newline='\n')
|
|
return 0 if report['passed'] else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if not __debug__:
|
|
raise SystemExit('Optimized Python is not supported')
|
|
sys.exit(main())
|