212 lines
12 KiB
Python
212 lines
12 KiB
Python
"""Finite proposal controls. Run only AFTER parent review; never runs subject code.
|
|
|
|
One unchanged positive, then fourteen negatives, stopping on the first unexpected
|
|
result or timeout. Only one transient ZIP exists; retain compact logs/results.
|
|
"""
|
|
from pathlib import Path
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
|
|
FULL = 'gates/project-full-71cadfa-1'
|
|
PROJECT_NAME = 'language::analysis_source::mixed::project::tests::complete_oa_sa_forward_model_warning_and_original_authority'
|
|
CASES = [
|
|
('archive-pin', 'Top-level archive pin differs'),
|
|
('verifier-pin', 'Top-level verifier pin differs'),
|
|
('route-pair-artifact', 'Repeated integration target artifact differs'),
|
|
('source-pair-artifact', 'Repeated integration target artifact differs'),
|
|
('missing-new-outcome', 'Missing/extra runtime outcomes'),
|
|
('changed-inherited-diagnostic', 'Exact historical failure diagnostics changed'),
|
|
('forged-git-input', 'Git input bytes mismatch'),
|
|
('missing-failed-history', 'Missing/extra package payload'),
|
|
('integration-zero-test', 'Missing/extra runtime outcomes'),
|
|
('integration-ignored', 'Runtime status mismatch'),
|
|
('integration-wrong-command', 'Exact command/mode/filter differs'),
|
|
('changed-formatter-body', 'Inherited formatter block body differs'),
|
|
('changed-successor-diff', 'Exact Git correction diff differs'),
|
|
('duplicate-zip-member', 'Duplicate ZIP members'),
|
|
]
|
|
|
|
|
|
def sha(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def encode(value):
|
|
return (json.dumps(value, indent=2) + '\n').encode()
|
|
|
|
|
|
def mutate(case, payload):
|
|
if case in ('route-pair-artifact', 'source-pair-artifact'):
|
|
kind = 'route-errors' if case.startswith('route-') else 'aggregate-references'
|
|
name = 'gates/project-' + kind + '-71cadfa-1.result.json'
|
|
r = json.loads(payload[name])
|
|
old = r['artifacts'][0]['sha256']
|
|
r['artifacts'][0]['sha256'] = '0' * 64 if old != '0' * 64 else '1' * 64
|
|
payload[name] = encode(r)
|
|
return name + ': change only one repeated target artifact SHA, keeping its path/size and all outcomes.'
|
|
if case == 'missing-new-outcome':
|
|
name = FULL + '.log'
|
|
text = payload[name].decode()
|
|
text, count = re.subn(r'^test ' + re.escape(PROJECT_NAME) + r' \.\.\. ok\r?\n', '', text, flags=re.M)
|
|
assert count == 1
|
|
payload[name] = text.replace('193 passed; 16 failed', '192 passed; 16 failed').encode()
|
|
return name + ': omit one new outcome and adjust pass count; exact membership must reject.'
|
|
if case == 'changed-inherited-diagnostic':
|
|
name = FULL + '.log'
|
|
text = payload[name].decode()
|
|
old = 'field: None, line: 1, column: 9, message: "Unsupported schema version:'
|
|
assert old in text
|
|
payload[name] = text.replace(old, old.replace('column: 9', 'column: 10'), 1).encode()
|
|
return name + ': change one inherited diagnostic column, retaining name/status/header error.'
|
|
if case == 'forged-git-input':
|
|
for name in list(payload):
|
|
if name.endswith('-71cadfa-1.inputs.json'):
|
|
r = json.loads(payload[name])
|
|
r['inputs'][0]['sha256'] = '0' * 64
|
|
payload[name] = encode(r)
|
|
return 'All71 input manifests: consistently forge first input SHA; immutable Git must reject.'
|
|
if case == 'missing-failed-history':
|
|
for suffix in ('.result.json', '.log', '.inputs.json'):
|
|
del payload['gates/project-focused-b9a31bb-1' + suffix]
|
|
return 'Remove only b9 failed-focus triple; mandatory history membership must reject.'
|
|
if case in ('integration-zero-test', 'integration-ignored'):
|
|
name = 'gates/project-route-errors-71cadfa-1.log'
|
|
text = payload[name].decode()
|
|
pattern = r'^test invalid_route_source_has_located_errors_and_no_silent_nested_defaults \.\.\. ok\r?\n'
|
|
replacement = '' if case == 'integration-zero-test' else 'test invalid_route_source_has_located_errors_and_no_silent_nested_defaults ... ignored\n'
|
|
text, count = re.subn(pattern, replacement, text, flags=re.M)
|
|
assert count == 1
|
|
if case == 'integration-zero-test':
|
|
text = text.replace('running 1 test', 'running 0 tests').replace('1 passed; 0 failed; 0 ignored; 0 measured; 4 filtered out', '0 passed; 0 failed; 0 ignored; 0 measured; 5 filtered out')
|
|
else:
|
|
text = text.replace('1 passed; 0 failed; 0 ignored;', '0 passed; 0 failed; 1 ignored;')
|
|
payload[name] = text.encode()
|
|
return name + ': synthesize zero-test success or ignored outcome with matching footer.'
|
|
if case == 'integration-wrong-command':
|
|
name = 'gates/project-route-errors-71cadfa-1.result.json'
|
|
r = json.loads(payload[name])
|
|
r['command'].remove('--exact')
|
|
payload[name] = encode(r)
|
|
return name + ': remove --exact while retaining the outcome; exact argv must reject.'
|
|
if case == 'changed-formatter-body':
|
|
for short in ('71cadfa', 'aaec78f'):
|
|
name = 'gates/project-fmt-' + short + '-1.log'
|
|
text = payload[name].decode()
|
|
old = '+ Self::EngineeringProperty | Self::AnalysisCase | Self::AnalysisCheck'
|
|
assert text.count(old) == 1
|
|
payload[name] = text.replace(old, old + ' /* altered */').encode()
|
|
return 'Both fmt logs: alter the same inherited block body, retaining locations/counts.'
|
|
if case == 'changed-successor-diff':
|
|
name = 'diffs/71cadfa-to-aaec78f.patch'
|
|
payload[name] += b'\n# altered correction\n'
|
|
return name + ': append text; exact immutable Git diff must reject.'
|
|
return 'Transport-only ZIP/verifier pin corruption or duplicate ZIP member.'
|
|
|
|
|
|
def repair(payload, manifest):
|
|
# Refresh every inner receipt binding and outcome vector, then all outer
|
|
# payload hashes. ZIP and verifier pins are refreshed by materialize below.
|
|
for name in list(payload):
|
|
if name.startswith('gates/') and name.endswith('.result.json'):
|
|
stem = name[:-len('.result.json')]
|
|
r = json.loads(payload[name])
|
|
r['log_sha256'] = sha(payload[stem + '.log'])
|
|
r['inputs_sha256'] = sha(payload[stem + '.inputs.json'])
|
|
rows = re.findall(r'^test ([^\r\n]+) \.\.\. (ok|FAILED|ignored)\r?$', payload[stem + '.log'].decode(), re.M)
|
|
r['outcomes'] = [list(row) for row in rows]
|
|
payload[name] = encode(r)
|
|
manifest['payloads'] = {name: {'bytes': len(raw), 'sha256': sha(raw)} for name, raw in sorted(payload.items())}
|
|
|
|
|
|
def materialize(folder, payload, manifest, verifier, case):
|
|
(folder / 'verify.py').write_bytes(verifier)
|
|
archive = folder / 'evidence.zip'
|
|
with zipfile.ZipFile(archive, 'x', compression=zipfile.ZIP_DEFLATED) as z:
|
|
for name, raw in sorted(payload.items()):
|
|
z.writestr(name, raw)
|
|
if case == 'duplicate-zip-member':
|
|
name = sorted(payload)[0]
|
|
z.writestr(name, payload[name])
|
|
raw = archive.read_bytes()
|
|
manifest['archive'] = {'file': 'evidence.zip', 'bytes': len(raw), 'sha256': sha(raw)}
|
|
manifest['verifier'] = {'file': 'verify.py', 'bytes': len(verifier), 'sha256': sha(verifier)}
|
|
if case == 'archive-pin':
|
|
with archive.open('ab') as stream:
|
|
stream.write(b'altered-transport')
|
|
if case == 'verifier-pin':
|
|
manifest['verifier']['sha256'] = '0' * 64
|
|
(folder / 'manifest.json').write_bytes(encode(manifest))
|
|
|
|
|
|
def main():
|
|
if sys.flags.optimize:
|
|
raise SystemExit('Optimized Python is forbidden for these controls.')
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--repo', type=Path, required=True)
|
|
parser.add_argument('--package', type=Path, default=Path(__file__).resolve().parent)
|
|
parser.add_argument('--out', type=Path, required=True)
|
|
parser.add_argument('--timeout-seconds', type=int, default=300)
|
|
args = parser.parse_args()
|
|
package, out = args.package.resolve(), args.out.resolve()
|
|
assert not out.exists() and not out.is_relative_to(package) and not package.is_relative_to(out), 'Choose a fresh output outside the original package.'
|
|
assert 30 <= args.timeout_seconds <= 600
|
|
original = {path.name: path.read_bytes() for path in package.iterdir() if path.is_file()}
|
|
verifier = original['verify.py']
|
|
manifest = json.loads(original['manifest.json'])
|
|
with zipfile.ZipFile(package / 'evidence.zip') as z:
|
|
assert len(z.namelist()) == len(set(z.namelist()))
|
|
payload = {name: z.read(name) for name in z.namelist()}
|
|
out.mkdir(parents=True, exist_ok=False)
|
|
results = []
|
|
def unchanged():
|
|
assert {path.name for path in package.iterdir() if path.is_file()} == set(original)
|
|
assert all((package / name).read_bytes() == raw for name, raw in original.items()), 'Original package changed.'
|
|
def run(case, current, expected, mutation):
|
|
argv = [sys.executable, '-B', str(package / 'verify.py'), '--repo', str(args.repo.resolve()), '--package', str(current)]
|
|
try:
|
|
r = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=args.timeout_seconds)
|
|
stdout, stderr, code = r.stdout, r.stderr, r.returncode
|
|
passed = code == 0 if expected is None else code != 0 and expected in stderr.decode(errors='replace') and 'REFUSED:' in stderr.decode(errors='replace')
|
|
status = 'PASS' if passed else 'UNEXPECTED_RESULT'
|
|
except subprocess.TimeoutExpired as error:
|
|
stdout, stderr, code, status = error.stdout or b'', error.stderr or b'', None, 'TIMEOUT'
|
|
row = {'case': case, 'status': status, 'exit': code, 'expected_refusal': expected, 'mutation': mutation, 'command': argv,
|
|
'stdout_sha256': sha(stdout), 'stderr_sha256': sha(stderr), 'semantic_transport_hashes_refreshed': expected is not None and case not in ('archive-pin', 'verifier-pin')}
|
|
(out / (case + '.stdout.log')).write_bytes(stdout)
|
|
(out / (case + '.stderr.log')).write_bytes(stderr)
|
|
(out / (case + '.result.json')).write_bytes(encode(row))
|
|
results.append(row)
|
|
unchanged()
|
|
return status == 'PASS'
|
|
proceed = run('unchanged-positive', package, None, 'Original bytes, no mutation.')
|
|
if proceed:
|
|
for case, expected in CASES:
|
|
current_payload, current_manifest = dict(payload), copy.deepcopy(manifest)
|
|
mutation = mutate(case, current_payload)
|
|
repair(current_payload, current_manifest)
|
|
with tempfile.TemporaryDirectory(prefix='case-', dir=out) as temporary:
|
|
folder = Path(temporary).resolve()
|
|
assert folder.parent == out and not folder.is_symlink(), 'Temporary cleanup must remain in the named controls output.'
|
|
materialize(folder, current_payload, current_manifest, verifier, case)
|
|
proceed = run(case, folder, expected, mutation)
|
|
if not proceed:
|
|
break
|
|
summary = {'status': 'PASS' if proceed and len(results) == 15 else 'INCOMPLETE_OR_FAILED', 'planned': 15, 'executed': len(results),
|
|
'package_pins': {name: {'bytes': len(raw), 'sha256': sha(raw)} for name, raw in original.items()}, 'results': results,
|
|
'limits': 'Evidence consistency controls only. No subject/compiler/GUI execution; no retained mutated archive corpus.'}
|
|
unchanged()
|
|
(out / 'summary.json').write_bytes(encode(summary))
|
|
print(json.dumps({'status': summary['status'], 'executed': len(results), 'summary': str(out / 'summary.json')}, indent=2))
|
|
return 0 if summary['status'] == 'PASS' else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|