Files
ARCADIA-Studio/harness/experiments/analysis-source-s3-windows-20260922/negative_controls.py
T

99 lines
5.6 KiB
Python

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