docs(harness): preserve private S3 Windows evidence and verifier
This commit is contained in:
@@ -0,0 +1,98 @@
|
||||
"""Read-only in-memory falsifiers for verify.py; does not modify archived evidence."""
|
||||
import argparse
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import warnings
|
||||
import zipfile
|
||||
|
||||
|
||||
def run(repo, directory):
|
||||
spec = importlib.util.spec_from_file_location('s3_evidence_verifier', directory / 'verify.py')
|
||||
verifier = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(verifier)
|
||||
verifier.require(__debug__, 'Optimized Python refused')
|
||||
raw = (directory / 'evidence.zip').read_bytes()
|
||||
payload = verifier.archive_payload(raw)
|
||||
positive = verifier.audit(repo, payload)
|
||||
cases = []
|
||||
|
||||
def refused(name, expected, action):
|
||||
try:
|
||||
action()
|
||||
except RuntimeError as error:
|
||||
verifier.require(expected in str(error), f'{name}: wrong refusal: {error}')
|
||||
cases.append({'name': name, 'refusal': str(error)})
|
||||
else:
|
||||
raise RuntimeError(name + ': invalid evidence was accepted')
|
||||
|
||||
def changed_json(name, change):
|
||||
candidate = dict(payload)
|
||||
data = verifier.document(candidate[name])
|
||||
change(data)
|
||||
candidate[name] = json.dumps(data).encode()
|
||||
return candidate
|
||||
|
||||
refused('changed_zip_bytes', 'Evidence ZIP SHA256 differs', lambda: verifier.archive_payload(raw + b'tampered'))
|
||||
refused('duplicate_json_key', 'Duplicate JSON key', lambda: verifier.document(b'{"source":1,"source":2}'))
|
||||
|
||||
def changed_zip(mode):
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, 'w') as archive:
|
||||
for name, content in payload.items():
|
||||
archive.writestr('../README.md' if mode == 'unsafe' and name == 'README.md' else name, content)
|
||||
if mode == 'duplicate':
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter('ignore', UserWarning)
|
||||
archive.writestr('README.md', payload['README.md'])
|
||||
malformed = buffer.getvalue()
|
||||
# Test-only pin substitution reaches ZIP structure checks. Production
|
||||
# archive_payload always requires its checked-in immutable ZIP digest.
|
||||
original_pin = verifier.ZIP_SHA
|
||||
try:
|
||||
verifier.ZIP_SHA = verifier.sha(malformed)
|
||||
verifier.archive_payload(malformed)
|
||||
finally:
|
||||
verifier.ZIP_SHA = original_pin
|
||||
|
||||
refused('duplicate_zip_member_with_test_only_pin', 'Exact unique ZIP membership', lambda: changed_zip('duplicate'))
|
||||
refused('unsafe_zip_member_with_test_only_pin', 'Unsafe ZIP path', lambda: changed_zip('unsafe'))
|
||||
|
||||
changed = dict(payload)
|
||||
changed['raw/strict-fixed-1b1c3e3.log'] = b' Compiling oa-core\n'
|
||||
refused('truncated_strict_completion', 'Closed clean strict log', lambda: verifier.audit(repo, changed))
|
||||
changed = dict(payload)
|
||||
name = 'raw/typed-fixed-1b1c3e3.log'
|
||||
changed[name] = b'\n'.join(line for line in changed[name].split(b'\n') if verifier.NEW.encode() not in line)
|
||||
refused('missing_typed_runtime_member', 'Exact ordered runtime results', lambda: verifier.audit(repo, changed))
|
||||
changed = changed_json('raw/token-fixed-1b1c3e3.result.json', lambda data: data['command'].remove('--offline'))
|
||||
refused('altered_gate_command', 'Exact command', lambda: verifier.audit(repo, changed))
|
||||
changed = changed_json('raw/typed-fixed-1b1c3e3.inputs.json', lambda data: data['inputs'][0].update(sha256='0' * 64))
|
||||
refused('altered_committed_input_hash', 'Exact538 source bindings', lambda: verifier.audit(repo, changed))
|
||||
changed = changed_json('raw/stack-diag-02-dirty.inputs.json', lambda data: data.update(head=verifier.SOURCE))
|
||||
refused('dirty_run_mislabeled_as_fixed_source', 'Dirty capture identity/stability', lambda: verifier.audit(repo, changed))
|
||||
changed = changed_json('raw/typed-90fae148.result.json', lambda data: data.update(exit=0))
|
||||
refused('historical_abort_mislabeled_success', 'Historical aborted gate', lambda: verifier.audit(repo, changed))
|
||||
changed = changed_json('raw/stack-diag-01-dirty.inputs.json', lambda data: data.update(dirty=data['dirty'].replace('STAGE entered', 'STAGE changed')))
|
||||
refused('diagnostic_patch_disagrees_with_captured_hash', 'Reconstructed diagnostic input', lambda: verifier.audit(repo, changed))
|
||||
changed = dict(payload)
|
||||
changed['raw/stack-probe-if-01-dirty.log'] = changed['raw/stack-probe-if-01-dirty.log'].replace(b'PROBE after malformed case 1:', b'PROBE after malformed case 0:')
|
||||
refused('probe_case_mislabeled', 'Dirty probe result/case', lambda: verifier.audit(repo, changed))
|
||||
result = subprocess.run([sys.executable, '-B', '-O', str(directory / 'verify.py'), '--repo', str(repo)], capture_output=True, text=True)
|
||||
verifier.require(result.returncode != 0 and 'Optimized Python refused' in result.stderr, 'Optimized verifier must refuse before audit')
|
||||
cases.append({'name': 'optimized_python_refused', 'refusal': 'Optimized Python refused'})
|
||||
return {'source': verifier.SOURCE, 'positive_audit': positive, 'negative_control_count': len(cases), 'negative_controls': cases, 'verifier_sha256': verifier.sha((directory / 'verify.py').read_bytes()), 'controls_sha256': verifier.sha(Path(__file__).read_bytes()), 'method': 'Raw archive untouched. Semantic corruptions pass directly to audit() after authentic archive validation; ZIP structural negatives temporarily substitute only the test module pin. No compiler, artifact execution or worktree write.'}
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--repo', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
print(json.dumps(run(args.repo.resolve(), Path(__file__).resolve().parent), indent=2))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user