301 lines
20 KiB
Python
301 lines
20 KiB
Python
"""Offline archive/source/receipt checks. Never executes archived helpers or native binaries."""
|
|
from pathlib import Path, PurePosixPath, PureWindowsPath
|
|
import argparse
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import zipfile
|
|
|
|
KIND = 'process-aggregate-removal-native-evidence'
|
|
PHASES = {'baseline': 'baseline.arcadia', 'removed': 'expected-removed.arcadia'}
|
|
V2 = 'process-removal-v2-preserved-trivia'
|
|
|
|
|
|
def pin(data):
|
|
return {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()}
|
|
|
|
|
|
def safe_name(name):
|
|
p = PurePosixPath(name)
|
|
assert name and not p.is_absolute() and str(p) == name, 'Unsafe member: ' + name
|
|
assert '\\' not in name and ':' not in name and not any(x in ('.', '..', '') for x in name.split('/')), 'Unsafe member: ' + name
|
|
|
|
|
|
def check_oracles(files, fixture, prefix='fixtures/'):
|
|
revision = fixture['fixture_revision']
|
|
assert revision in ('process-removal-v1-literal', V2)
|
|
assert fixture['selection'] == fixture['removed'] == ['a', 'c', 'approved']
|
|
assert fixture['changed'] == ['assessment']
|
|
assert fixture['caller_definitions'] == ['caller', 'outer']
|
|
assert fixture['remaining_implicit_order'] == ['e1', 'e2']
|
|
assert set(fixture['files']) == set(PHASES.values()), 'Wrong fixture inventory'
|
|
for name, expected in fixture['files'].items():
|
|
assert pin(files[prefix + name]) == expected, 'Fixture pin mismatch: ' + name
|
|
baseline = files[prefix + 'baseline.arcadia']
|
|
assert pin(baseline) == {'bytes': 2609, 'sha256': '9965f4024584d64ac81442724aeb77b018b0cfae94e50bd412c5dcbbac49f6bf'}
|
|
actual = files[prefix + 'expected-removed.arcadia']
|
|
if revision != V2:
|
|
# Preserve the original incorrect oracle as history, never acceptance.
|
|
assert pin(actual) == {'bytes': 2186, 'sha256': '117b550f0c10753f53febb3ca6beb2c8088688ffa707f21eac654573e4a3d29c'}
|
|
return
|
|
text = baseline.decode('utf-8')
|
|
assert '\r' not in text
|
|
pattern = r'(?m)^ (?:oa_occurrence (?:a|c) |sequence_link approved )[^\n]*'
|
|
# Independent fixture-specific lexical rules; never execute archived code.
|
|
def trivia(line):
|
|
return re.sub(r'\S', '', re.sub(r'"(?:\\.|[^"\\])*"', '', line))
|
|
rows = re.findall(pattern, text)
|
|
assert [trivia(row) for row in rows] == [' ' * 22, ' ' * 20, ' ' * 16]
|
|
expected = re.sub(pattern, lambda m: '\n\n' + trivia(m.group()) + '\n\n', text)
|
|
assert expected.count('steps [@a,@e1,@c,@e2]') == 1
|
|
expected = expected.replace('steps [@a,@e1,@c,@e2]', 'steps [@e1,@e2]')
|
|
assert actual == expected.encode('utf-8'), 'V2 differs from independently specified delta'
|
|
assert pin(actual) == {'bytes': 2258, 'sha256': 'b4a2b07ad4c796f555c61945ca42b1c221b454b35488c60fefc86f96f7aa71ba'}
|
|
|
|
|
|
def check_source(files, actual, phase, prefix='fixtures/'):
|
|
assert phase in PHASES, 'Unknown source phase'
|
|
assert actual == files[prefix + PHASES[phase]], 'Exact full source differs from ' + phase
|
|
|
|
|
|
def verify(root):
|
|
manifest = json.loads((root / 'manifest.json').read_text(encoding='utf-8'))
|
|
assert manifest['kind'] == KIND, 'Wrong package kind'
|
|
assert pin((root / 'native-evidence.zip').read_bytes()) == manifest['archive'], 'Archive pin mismatch'
|
|
with zipfile.ZipFile(root / 'native-evidence.zip') as archive:
|
|
names = archive.namelist()
|
|
assert len(names) == len(set(names)) and set(names) == set(manifest['members']), 'Archive member inventory mismatch'
|
|
for name in names:
|
|
safe_name(name)
|
|
assert ((archive.getinfo(name).external_attr >> 16) & 0o170000) != 0o120000, 'Symlink member'
|
|
files = {name: archive.read(name) for name in names}
|
|
for name, data in files.items():
|
|
assert pin(data) == manifest['members'][name], 'Member pin mismatch: ' + name
|
|
assert PurePosixPath(name).suffix.lower() not in ('.exe', '.dll'), 'Unexpected archived binary'
|
|
for name in manifest['excluded_binary_bytes']:
|
|
safe_name(name)
|
|
assert name not in files and PurePosixPath(name).suffix.lower() in ('.exe', '.dll'), 'Invalid omitted binary entry'
|
|
assert (root / 'verify.py').read_bytes() == files['helpers/native-evidence-verify.py'], 'External verifier differs from pinned helper'
|
|
|
|
def js(name):
|
|
return json.loads(files[name].decode('utf-8-sig'))
|
|
|
|
def present_pin(name):
|
|
return pin(files[name]) if name in files else manifest['excluded_binary_bytes'].get(name)
|
|
|
|
def build_binding(prefix, source):
|
|
receipt = js(prefix + 'result.json')
|
|
proof = js(prefix + 'source-proof.json')
|
|
inputs = js(prefix + 'inputs.json')
|
|
assert receipt['exitCode'] == 0 and receipt['inputsUnchanged'] is True, 'Build not successful/unchanged'
|
|
assert proof['status'].startswith('PASS') and proof['result'] == pin(files[prefix + 'result.json']), 'Build proof result binding'
|
|
assert proof['inputMap'] == receipt['inputs'] == pin(files[prefix + 'inputs.json']), 'Build input-map binding'
|
|
assert proof['inputCount'] == receipt['inputCount'] == len(inputs), 'Build input count'
|
|
assert proof['commit'] == receipt['head'] == source and re.fullmatch('[0-9a-f]{40}', source), 'Build source identity'
|
|
return receipt
|
|
|
|
fixture = js('fixtures/fixture-manifest.json')
|
|
check_oracles(files, fixture)
|
|
for name in files:
|
|
if name.startswith('oracle-revisions/') and name.endswith('/fixture-manifest.json'):
|
|
check_oracles(files, js(name), name.removesuffix('fixture-manifest.json'))
|
|
derivation = 'oracle-revisions/oracle-v2/derive-source-oracle-v2.ps1'
|
|
assert pin(files[derivation]) == js('oracle-revisions/oracle-v2/fixture-manifest.json')['derivation']
|
|
config = js('review-config.json')
|
|
final_source = config['final_source']
|
|
assert re.fullmatch('[0-9a-f]{40}', final_source)
|
|
assert config['runs'] == manifest['runs'] and config['preflights'] == manifest['preflights']
|
|
for entries, prefix, pattern in [(config['runs'], 'runs', r'run-[0-9]+'), (config['preflights'], 'preflight', r'fixture-preflight-[0-9]+')]:
|
|
assert entries and len({x['name'] for x in entries}) == len(entries)
|
|
assert all(re.fullmatch(pattern, x['name']) and type(x['accepted']) is bool for x in entries)
|
|
assert {name.split('/')[1] for name in files if name.startswith(prefix + '/')} == {x['name'] for x in entries}, 'Undeclared archived attempts'
|
|
|
|
gates = config['gates']
|
|
assert len(gates) == len(set(gates)) and {'process-removal-tests-1', 'process-removal-tests-2'} <= set(gates)
|
|
assert all(re.fullmatch(r'process-removal-(?:tests|format|clippy|build)-[0-9]+', g) for g in gates)
|
|
assert {n.split('/')[1] for n in files if n.startswith('gates/')} == set(gates)
|
|
gate_reports = []
|
|
for gate in gates:
|
|
prefix = 'gates/' + gate + '/'
|
|
result, proof, inputs = (js(prefix + f) for f in ('result.json', 'source-proof.json', 'inputs.json'))
|
|
assert result['inputsUnchanged'] is True and 'finish' in result
|
|
assert proof['status'].startswith('PASS') and proof['result'] == pin(files[prefix + 'result.json'])
|
|
assert proof['inputMap'] == result['inputs'] == pin(files[prefix + 'inputs.json'])
|
|
assert proof['inputCount'] == result['inputCount'] == len(inputs)
|
|
assert proof['commit'] == result['head'] and re.fullmatch('[0-9a-f]{40}', result['head'])
|
|
for stream in ('stdout', 'stderr'):
|
|
assert result[stream] == pin(files[prefix + stream + '.log'])
|
|
gate_reports.append({'gate': gate, 'source': result['head'], 'exitCode': result['exitCode']})
|
|
# The original failed gates must remain failed historical facts.
|
|
for gate, source in [('process-removal-tests-1', 'fe637dcd277b17653139ffa234e601b29775c565'),
|
|
('process-removal-tests-2', 'f20497b29bf637619bc35ffe41274fc0f66c387b')]:
|
|
receipt = js('gates/' + gate + '/result.json')
|
|
assert receipt['head'] == source and receipt['exitCode'] == 101
|
|
|
|
reports, preflights = [], []
|
|
accepted_cli_bindings = set()
|
|
for entry in config['preflights']:
|
|
prefix = 'preflight/' + entry['name'] + '/'
|
|
problems = []
|
|
try:
|
|
current = js(prefix + 'result.json')
|
|
started = js(prefix + 'started.json')
|
|
fixture_name = prefix + 'evidence/fixture-manifest.json'
|
|
current_fixture = js(fixture_name)
|
|
check_oracles(files, current_fixture, prefix + 'fixtures/')
|
|
revision = current_fixture.get('fixture_revision', 'removal-v1-source-derived')
|
|
if 'fixture_revision' in current:
|
|
assert current['fixture_revision'] == revision and current['fixture_manifest'] == pin(files[fixture_name]), 'Preflight fixture revision binding'
|
|
assert current['fixture_comparator'] == pin(files[prefix + 'evidence/verify-captured.py']), 'Preflight comparator pin'
|
|
assert started['checks'] == [] and all(current[key] == started[key] for key in started if key != 'checks'), 'Preflight started/result changed identity'
|
|
assert current['status'].startswith('PASS') and current['cliUnchanged'] is True, 'Preflight not passed'
|
|
receipt = build_binding(prefix + 'evidence/', current['source'])
|
|
assert current['receipt'] == pin(files[prefix + 'evidence/result.json']) and current['proof'] == pin(files[prefix + 'evidence/source-proof.json']), 'Preflight proof pins'
|
|
cli = [a for a in receipt['artifacts'] if a['cargo']['target']['name'] == 'arcadia' and not a['cargo']['profile']['test']]
|
|
assert len(cli) == 1 and cli[0]['readback'] == current['cli'] == present_pin(prefix + 'bin/arcadia.exe'), 'CLI artifact binding'
|
|
assert len(current['checks']) == len(PHASES) and {x['fixture'] for x in current['checks']} == set(PHASES.values()), 'Incomplete fixture preflight'
|
|
for check in current['checks']:
|
|
name = check['fixture']
|
|
assert check['exitCode'] == 0 and check['json_report'] is True and check['fixtureUnchanged'] is True, 'Fixture check failed: ' + name
|
|
assert check['stdout'] == pin(files[prefix + name + '.stdout.json']) and check['stderr'] == pin(files[prefix + name + '.stderr.log']), 'CLI output pins'
|
|
assert pin(files[prefix + 'fixtures/' + name]) == current_fixture['files'][name], 'Preflight fixture differs'
|
|
raw = js(prefix + name + '.stdout.json')
|
|
assert raw['operation'] == 'check' and raw['exit_code'] == 0 and raw['written'] is False and not any(x['severity'] == 'error' for x in raw['diagnostics']), 'Raw CLI report rejects fixture'
|
|
for runtime in current['runtime']:
|
|
assert present_pin(prefix + 'bin/' + PureWindowsPath(runtime['source']).name) == {k: runtime[k] for k in ('bytes', 'sha256')}, 'CLI runtime pin'
|
|
except (AssertionError, KeyError, ValueError, TypeError) as error:
|
|
problems.append(str(error))
|
|
preflights.append({'preflight': entry['name'], 'reviewer_accepted': entry['accepted'], 'checks_pass': not problems, 'problems': problems})
|
|
if entry['accepted']:
|
|
assert current_fixture['fixture_revision'] == V2, 'Historical v1 cannot be accepted'
|
|
assert current['source'] == final_source, 'Accepted preflight source differs'
|
|
assert not problems, json.dumps(preflights[-1])
|
|
accepted_cli_bindings.add((current['source'], pin(files[fixture_name])['sha256']))
|
|
|
|
for entry in config['runs']:
|
|
prefix = 'runs/' + entry['name'] + '/'
|
|
problems, captures, sources = [], 0, []
|
|
# Report the actual discovered count even when an earlier historical
|
|
# source failure prevents the remaining per-run checks from passing.
|
|
for name in files:
|
|
if name.startswith(prefix) and name.endswith('.json'):
|
|
try:
|
|
record = js(name)
|
|
captures += int(isinstance(record, dict) and 'images' in record)
|
|
except (ValueError, UnicodeError):
|
|
pass
|
|
try:
|
|
prepared, started, finished = (js(prefix + name + '.json') for name in ('prepared', 'started', 'finished'))
|
|
fixture_name = prefix + 'evidence/fixture-manifest.json'
|
|
run_fixture = js(fixture_name)
|
|
check_oracles(files, run_fixture, prefix + 'evidence/')
|
|
revision = run_fixture.get('fixture_revision', 'removal-v1-source-derived')
|
|
if 'fixture_revision' in prepared:
|
|
assert prepared['fixture_revision'] == revision and prepared['fixture_manifest'] == pin(files[fixture_name]), 'Run fixture revision binding'
|
|
assert started['prepared'] == pin(files[prefix + 'prepared.json']), 'Started/prepared binding'
|
|
for rel, expected in prepared['immutable'].items():
|
|
safe_name(rel)
|
|
assert present_pin(prefix + rel) == expected, 'Immutable pin: ' + rel
|
|
assert prepared['helper'] == pin(files[prefix + 'evidence/native-harness.py']), 'Native helper binding'
|
|
for helper in ('capture.js', 'verify-captured.py'):
|
|
rel = 'evidence/' + helper
|
|
assert prepared['immutable'].get(rel) == pin(files[prefix + rel]), 'Captured observation/source helper is not immutable: ' + helper
|
|
for name in [*run_fixture['files'], 'fixture-manifest.json']:
|
|
assert prepared['immutable'].get('evidence/' + name) == pin(files[prefix + 'evidence/' + name]), 'Run oracle is not immutable: ' + name
|
|
receipt = build_binding(prefix + 'evidence/', prepared['source'])
|
|
gui = [a for a in receipt['artifacts'] if a['cargo']['target']['name'] == 'arcadia-studio' and not a['cargo']['profile']['test']]
|
|
assert len(gui) == 1 and gui[0]['readback'] == prepared['artifact'] == present_pin(prefix + 'bin/arcadia-studio.exe'), 'GUI artifact binding'
|
|
for runtime in prepared['runtime_sources']:
|
|
assert present_pin(prefix + 'bin/' + PureWindowsPath(runtime['path']).name) == {k: runtime[k] for k in ('bytes', 'sha256')}, 'GUI runtime pin'
|
|
assert started['command'] == prepared['command'] and started['cwd'] == prepared['cwd'], 'Launch command/cwd changed'
|
|
identity = started['identity']
|
|
assert identity['pid'] == started['pid'] and identity['created_filetime'] > 0 and identity['running'] is True, 'Missing initial live process identity'
|
|
assert PureWindowsPath(identity['image']) == PureWindowsPath(prepared['command'][0]), 'Wrong process image'
|
|
assert finished['exitCode'] == 0 and finished['immutableUnchanged'] is True, 'Normal exit/immutability not established'
|
|
for stream in ('stdout', 'stderr'):
|
|
assert finished[stream] == pin(files[prefix + 'app.' + stream + '.log']), 'Final log binding'
|
|
assert finished['working'] == pin(files[prefix + 'working.arcadia']), 'Final source binding'
|
|
assert prepared.get('initial_pin', started['workingBefore']) == started['workingBefore'], 'Initial source pin changed'
|
|
origin = entry.get('reopen_from')
|
|
if prefix + 'reopen-origin.json' in files:
|
|
old = js(prefix + 'reopen-origin.json')
|
|
declared = {'run': old['from_run'], 'file': old['from_file']}
|
|
assert origin is None or origin == declared, 'Contradictory reopen origins'
|
|
origin = declared
|
|
assert old['from_pin'] == old['copied_pin'] == started['workingBefore'], 'Reopen-origin pins disagree'
|
|
if origin:
|
|
assert origin['run'] != entry['name'] and origin['run'] in {x['name'] for x in config['runs']}, 'Unknown/self reopen origin'
|
|
safe_name(origin['file'])
|
|
assert pin(files['runs/' + origin['run'] + '/' + origin['file']]) == started['workingBefore'], 'Reopen source is not the exact captured copy'
|
|
else:
|
|
assert started['workingBefore'] == run_fixture['files']['baseline.arcadia'], 'Nonbaseline start needs explicit reopen_from'
|
|
phases = entry.get('source_phases', {})
|
|
actual_sources = [name for name in files if name.startswith(prefix) and name.endswith('.arcadia') and '/evidence/' not in name and name != prefix + 'baseline.arcadia']
|
|
assert set(phases) <= {name[len(prefix):] for name in actual_sources}, 'Phase config references missing source'
|
|
for name in actual_sources:
|
|
rel = name[len(prefix):]
|
|
sidecar = name[:-8] + '.json'
|
|
phase = phases.get(rel)
|
|
if sidecar in files and 'phase' in js(sidecar):
|
|
saved = js(sidecar)
|
|
assert phase is None or phase == saved['phase'], 'Contradictory snapshot phase'
|
|
phase = saved['phase']
|
|
assert saved['pin'] == pin(files[name]) and saved['exitCode'] == 0, 'Snapshot check/pin failed'
|
|
evaluated = js(name[:-8] + '.evaluator.stdout.json')
|
|
assert evaluated['status'] == 'PASS' and evaluated['phase'] == phase and evaluated['sha256'] == pin(files[name])['sha256'], 'Raw source evaluator report mismatch'
|
|
assert not files[name[:-8] + '.evaluator.stderr.log'], 'Source evaluator stderr is nonempty'
|
|
check_source(files, files[name], phase, prefix + 'evidence/')
|
|
sources.append({'file': rel, 'phase': phase})
|
|
start_time = datetime.datetime.fromisoformat(started['started'])
|
|
finish_time = datetime.datetime.fromisoformat(finished['finished'])
|
|
assert start_time <= finish_time, 'Reversed process lifetime'
|
|
for name in files:
|
|
if not name.startswith(prefix) or not name.endswith('.json'):
|
|
continue
|
|
record = js(name)
|
|
if not isinstance(record, dict) or 'images' not in record:
|
|
continue
|
|
assert record['images'], 'Empty capture record'
|
|
assert record['window']['app'].casefold() == ('process:' + identity['image']).casefold(), 'Capture uses a different executable surface'
|
|
observed = datetime.datetime.fromisoformat(record['time'])
|
|
assert start_time <= observed <= finish_time, 'Capture is outside process lifetime'
|
|
for image in record['images']:
|
|
safe_name(image['file'])
|
|
assert prefix + image['file'] in files and image['width'] > 0 and image['height'] > 0, 'Missing/invalid capture image'
|
|
assert captures > 0, 'No captured native observations'
|
|
if entry['accepted']:
|
|
assert run_fixture['fixture_revision'] == V2, 'Historical v1 cannot be accepted'
|
|
assert prepared['source'] == final_source, 'Accepted native source differs'
|
|
assert (prepared['source'], pin(files[fixture_name])['sha256']) in accepted_cli_bindings, 'No accepted current-source/current-oracle CLI fixture preflight'
|
|
except (AssertionError, KeyError, ValueError, TypeError) as error:
|
|
problems.append(str(error))
|
|
reports.append({'run': entry['name'], 'reviewer_accepted': entry['accepted'], 'checks_pass': not problems, 'problems': problems, 'capture_records': captures, 'source_checks': sources})
|
|
if entry['accepted']:
|
|
assert not problems, json.dumps(reports[-1], indent=2)
|
|
return {
|
|
'status': 'PASS offline archive, exact source and receipt consistency checks',
|
|
'gates': gate_reports,
|
|
'runs': reports,
|
|
'preflights': preflights,
|
|
'accepted_run_count': sum(entry['accepted'] for entry in config['runs']),
|
|
'limitations': [
|
|
'Human visual, keyboard, selection and interaction claims are not inferred from screenshots.',
|
|
'Omitted executable/DLL bytes cannot be rehashed offline.',
|
|
'Build proof/input map consistency checked; Git blobs are not independently available offline.',
|
|
'accepted=false failures are retained and reported, never silently promoted.',
|
|
'No native history-depth introspection, usability acceptance, release or publication assertion.',
|
|
],
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('directory', type=Path)
|
|
args = parser.parse_args()
|
|
print(json.dumps(verify(args.directory.resolve()), indent=2))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|