#!/usr/bin/env python3 """Offline public.2 saved-evidence verifier (new unsigned publication utility). Uses only the Python standard library. Reads ZIP members in memory; never extracts, imports, or executes archived code, invokes a native experiment, or accesses a key. Hash equality verifies bytes, not truth, execution authority, or a new signature. Licensed under the edition's recorded Apache-2.0 terms. """ import argparse import base64 import hashlib import json import re import stat import sys import zipfile from pathlib import Path, PurePosixPath EDITION = 'EZA-001-LINUX-EXPERIMENTAL-EVIDENCE-001-public.2' SOURCE_EDITION = 'EZA-001-LINUX-EXPERIMENTAL-EVIDENCE-001-public.1' SOURCE_ZIP_SHA = 'd76188a10e25654bad97069ed71c5fdc8783e6f4d696ddaae9cad83c551eacdf' SOURCE_MANIFEST_SHA = 'c76344408004a35be04d13ff9fa9734d9af4e7270f224b1bc2f5c6f530a76788' ORIGINAL_ZIP_SHA = 'c560375e008991adb26536880b17a48a083ca424226f6f219e87cb18d92e25bb' ORIGINAL_MANIFEST_SHA = 'ab3602bc570a205537a0ae2e7f379de3c2582bd7a454e43dd35894832aa30b7b' VERDICTS = {'native_protocol': 'PASS', 'manual_four_file_denial': 'INCONCLUSIVE', 'isolated_scope_enforcement': 'INCONCLUSIVE', 'same_tick_mutation': 'NOT EXECUTED', 'overall_adversarial_isolation': 'INCONCLUSIVE', 'production_certification': False} OMITTED_PATHS = { 'evidence/final-release-220/audit/current-maintenance.json', 'evidence/final-release-220/audit/MAINTENANCE-REPLAY.json', 'evidence/final-release-220/audit/current-maintenance-execution.json', 'evidence/startup-prerequisite-correction-203/tools/freeze_verify.py', } PROJECTION_PATHS = { 'PUBLICATION-LINEAGE.json', 'reviews/221/PUBLICATION-LINEAGE.json', 'evidence/final-release-220/audit/CURRENT-OBSERVATION-PROVENANCE.json', 'evidence/final-release-220/audit/MANUAL-TEST-EVIDENCE.json', 'evidence/final-release-220/audit/current-physical.json', 'evidence/final-release-220/audit/post219-sudo-journal.jsonl', 'evidence/live-authority-registration-207/authority-observation.py', 'evidence/live-authority-stress-218/INDEPENDENT-VERIFICATION.json', 'evidence/live-authority-stress-218/RESULTS.json', 'evidence/live-authority-stress-218/final-physical-evidence.json', 'evidence/native-authorized-commit-217/complete-executor-journal.jsonl', 'evidence/runtime-input-acquisition-106/REPORT.txt', 'historical/original-release/CLAIM-BOUNDARIES.txt', 'historical/original-release/IDENTITY.json', } class VerificationError(Exception): pass def require(condition, message): # Do not use Python assert: checks must remain active under python -O. if not condition: raise VerificationError(message) def sha(value): return hashlib.sha256(value).hexdigest() def wire(value): return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=True, allow_nan=False).encode() def unique_object(pairs): value = {} for key, item in pairs: require(key not in value, 'duplicate JSON object key') value[key] = item return value def parse(value): return json.loads(value, object_pairs_hook=unique_object, parse_constant=lambda _: (_ for _ in ()).throw(VerificationError('nonfinite JSON value'))) def jsonl(value): return [parse(line) for line in value.splitlines() if line.strip()] def safe_path(name): path = PurePosixPath(name) return (bool(name) and not path.is_absolute() and '..' not in path.parts and '.' not in name.split('/') and '\\' not in name and str(path) == name and not name.endswith('/')) def read_package(archive, expected_archive, expected_manifest, edition): raw = Path(archive).read_bytes() require(sha(raw) == expected_archive, 'archive SHA-256 mismatch') with zipfile.ZipFile(archive) as zipped: members = zipped.infolist() names = [item.filename for item in members] require(len(names) == len(set(names)), 'duplicate ZIP member') require(sum(item.file_size for item in members) < 512 * 1024 * 1024, 'inspection size cap exceeded') prefix = edition + '/' for item in members: require(safe_path(item.filename) and item.filename.startswith(prefix), 'unsafe or unexpected ZIP member') require(not item.is_dir() and stat.S_IFMT(item.external_attr >> 16) == stat.S_IFREG, 'ZIP member is not a regular file') require(not item.flag_bits & 1, 'encrypted ZIP member') data = {name[len(prefix):]: zipped.read(name) for name in names} require('MANIFEST.json' in data, 'missing manifest') require(sha(data['MANIFEST.json']) == expected_manifest, 'manifest SHA-256 mismatch') manifest = parse(data['MANIFEST.json']) require(manifest['edition_id'] == edition, 'manifest edition mismatch') expected = {item['path']: item for item in manifest['files']} require(len(expected) == len(manifest['files']), 'duplicate manifest path') require(set(data) == set(expected) | {'MANIFEST.json'}, 'ZIP/manifest membership mismatch') for name, item in expected.items(): require(safe_path(name), 'unsafe manifest path') require(len(data[name]) == item['bytes'] and sha(data[name]) == item['sha256'], 'manifest member bytes mismatch: ' + name) return data, manifest, len(raw) def check_inventory(data, source_data=None): included = jsonl(data['INCLUDED-ORIGINAL-MEMBERS.jsonl']) omitted = jsonl(data['OMITTED-ORIGINAL-MEMBERS.jsonl']) policy = parse(data['OMISSION-POLICY.json']) provenance = parse(data['PRESERVED-BYTE-PROVENANCE.json']) projections = parse(data['PROJECTION-INVENTORY.json']) chains = [tuple(item['original_chain']) for item in included + omitted] require(len(chains) == len(set(chains)) == 20473, 'original occurrence count or uniqueness mismatch') require(policy['regular_member_occurrences'] == 20473, 'omission policy occurrence count') require(policy['included_occurrences'] == len(included), 'included count disagrees with policy') require(policy['omitted_occurrences'] == len(omitted), 'omitted count disagrees with policy') for item in included: name = item['public_path'] require(name in data and sha(data[name]) == item['sha256'] and len(data[name]) == item['bytes'], 'exact original inclusion differs: ' + name) for item in omitted: require(item['reason_code'] in policy['rules'], 'undeclared omission reason') pmap = {item['public_path']: item for item in projections} require(len(pmap) == len(projections) and PROJECTION_PATHS <= set(pmap), 'missing or duplicate affected projection') exact_paths = {item['public_path'] for item in provenance} require(len(exact_paths) == len(provenance), 'duplicate provenance path') require(not exact_paths & set(pmap), 'projection falsely included as byte-identical original') require({item['public_path'] for item in included} == exact_paths, 'original alias/provenance path mismatch') require(policy['unique_original_byte_sequences_retained'] == len(provenance), 'preserved original count') for item in provenance: name = item['public_path'] require(name in data and sha(data[name]) == item['sha256'] and len(data[name]) == item['bytes'], 'preserved provenance mismatch: ' + name) require(any(other['original_chain'] == item['original_chain'] and other['sha256'] == item['sha256'] and other['public_path'] == name for other in included), 'provenance chain lacks exact inclusion') for name, item in pmap.items(): require(item['source_edition_id'] == SOURCE_EDITION, 'projection source edition') require(item['byte_identical'] is False and item['original_signature_authenticates_projection'] is False, 'projection authentication claim') require(item['source_sha256'] != item['public_sha256'], 'projection claims unchanged digest') require(name in data and sha(data[name]) == item['public_sha256'] and len(data[name]) == item['public_bytes'], 'projection public bytes mismatch: ' + name) require(bool(item['transformation']) and bool(item['field_classes']), 'undocumented projection transformation') require(re.fullmatch('[0-9a-f]{64}', item['source_sha256']) is not None, 'invalid projection source digest') if name in PROJECTION_PATHS: if name.endswith('.json'): notices = [parse(data[name]).get('_public_projection')] elif name.endswith('.jsonl'): notices = [entry.get('_public_projection') for entry in jsonl(data[name])] else: notices = [] require('DERIVED PRIVACY PROJECTION' in data[name].decode()[:600] and item['source_sha256'] in data[name].decode()[:600], 'projected text/source lacks prominent change notice') for notice in notices: require(isinstance(notice, dict) and notice.get('classification') == 'derived privacy projection', 'projected JSON lacks change notice') require(notice['source_edition_id'] == SOURCE_EDITION and notice['source_path'] == item['source_path'] and notice['source_sha256'] == item['source_sha256'], 'projection notice source differs') require(notice['byte_identical'] is False and notice['opaque_labels_are_historical_values'] is False and notice['original_signature_authenticates_projection'] is False, 'projection notice authenticity differs') if item.get('original_chain'): matching = [other for other in omitted if other['original_chain'] == item['original_chain']] require(len(matching) == 1 and matching[0]['sha256'] == item['source_sha256'], 'projected original source not omitted') require(matching[0]['reason_code'] == 'PRIVACY-PROJECTION' and matching[0]['projection_path'] == name, 'projected original source mapping') for item in omitted: if item['reason_code'] == 'PRIVACY-PROJECTION': require(item['projection_path'] in pmap and item['sha256'] == pmap[item['projection_path']]['source_sha256'], 'omission projection alias source mismatch') require(not OMITTED_PATHS & set(data), 'private maintenance/ancillary payload retained') if source_data is not None: require(set(source_data) - set(data) == OMITTED_PATHS, 'unexpected omitted public.1 member') changed = {name for name in set(data) & set(source_data) if name != 'MANIFEST.json' and data[name] != source_data[name]} require(changed == set(pmap), 'changed inherited record lacks projection inventory entry') original_source = {tuple(item['original_chain']): (item['sha256'], item['bytes']) for name in ['INCLUDED-ORIGINAL-MEMBERS.jsonl', 'OMITTED-ORIGINAL-MEMBERS.jsonl'] for item in jsonl(source_data[name])} actual = {tuple(item['original_chain']): (item['sha256'], item['bytes']) for item in included + omitted} require(actual == original_source, 'original occurrence identities changed from pinned public.1') for name, item in pmap.items(): original = source_data[item['source_path']] require(sha(original) == item['source_sha256'] and len(original) == item['source_bytes'], 'projection source differs from frozen public.1: ' + name) for item in provenance: require(data[item['public_path']] == source_data[item['public_path']], 'retained original changed from public.1') source_physical = parse(source_data['evidence/live-authority-stress-218/final-physical-evidence.json']) public_physical = parse(data['evidence/live-authority-stress-218/final-physical-evidence.json']) for key in ['ledger_files', 'units', 'custodian_process_status', 'custodian_exe', 'mount']: require(source_physical[key] == public_physical[key], 'substantive physical evidence changed: ' + key) source_current = parse(source_data['evidence/final-release-220/audit/current-physical.json']) public_current = parse(data['evidence/final-release-220/audit/current-physical.json']) for key in ['files', 'custodian', 'process_status', 'spec', 'observed_at_ns']: require(source_current[key] == public_current[key], 'substantive current physical evidence changed: ' + key) journal_pairs = [(parse_journal(source_physical['journal']), parse_journal(public_physical['journal']))] journal_path = 'evidence/native-authorized-commit-217/complete-executor-journal.jsonl' journal_pairs.append((jsonl(source_data[journal_path]), jsonl(data[journal_path]))) identity_fields = {'_MACHINE_ID', '_BOOT_ID', '_HOSTNAME', '__SEQNUM_ID', '__CURSOR', '_public_projection'} for old_entries, new_entries in journal_pairs: require(len(old_entries) == len(new_entries), 'native journal entry membership changed') for old_entry, new_entry in zip(old_entries, new_entries): require({key: value for key, value in old_entry.items() if key not in identity_fields} == {key: value for key, value in new_entry.items() if key not in identity_fields}, 'native journal nonidentity fields or MESSAGE bytes changed') return included, omitted, provenance, pmap def check_ledger(rows, expected_head): previous = '0' * 64 for sequence, row in enumerate(rows, 1): require(row['sequence'] == sequence and row['previous'] == previous, 'ledger sequence/previous mismatch') require(sha(wire({key: value for key, value in row.items() if key != 'sha256'})) == row['sha256'], 'ledger row hash mismatch') previous = row['sha256'] require(previous == expected_head, 'ledger head mismatch') def check_state(state, current_rows): check_ledger(state['ledger_rows'], state['ledger_head']) require(state['ledger_rows'] == current_rows[:len(state['ledger_rows'])], 'state ledger differs from final prefix') require(sha(wire(state['payload'])) == state['proof']['payload'], 'state payload/proof hash mismatch') def parse_journal(value): return jsonl(value.encode() if isinstance(value, str) else value) def check_physical(physical, current_physical, rows): files = physical['ledger_files'] require(set(files) == {f'{i:08d}.json' for i in range(1, 23)} | {'checkpoint'}, 'physical ledger inventory mismatch') for name, item in files.items(): value = base64.b64decode(item['base64'], validate=True) require(len(value) == item['size'] and sha(value) == item['sha256'], 'physical file bytes mismatch') require(item['uid'] == item['gid'] == 5200 and item['mode'] == '0o600', 'physical file ownership/mode mismatch') if name == 'checkpoint': require(parse(value) == {'domain': rows[-1]['domain'], 'head': rows[-1]['sha256'], 'sequence': 22}, 'physical checkpoint mismatch') else: require(value == wire(rows[int(name[:8]) - 1]), 'physical ledger row differs from logical row') path = '/var/lib/eza-custody-001/domains/qualification-001/custody-records/' + name later = current_physical['files'][path] require(later['sha256'] == item['sha256'] and later['bytes'] == item['size'], 'current physical hash differs') require(later['uid'] == later['gid'] == 5200, 'current physical ownership differs') for field in ['device', 'inode', 'mode', 'acl']: require(later[field] == item[field], 'current physical metadata differs: ' + field) for value in [physical['custodian_process_status'], current_physical['process_status']]: for field in ['Uid', 'Gid']: require(re.search(r'^' + field + r':\s+5200\s+5200\s+5200\s+5200\s*$', value, re.M) is not None, 'custodian process identity differs') for field in ['MainPID', 'ActiveState', 'SubState', 'InvocationID']: require(physical['units']['eza-custody-001.service'][field] == current_physical['custodian'][field], 'custodian process/unit correlation differs') def check_native(data): def load(name): return parse(data['evidence/' + name]) current = load('final-release-220/audit/current-authority.json') prior = load('live-linux-adversarial-proof-219/authority-before.json') rows = current['ledger_rows'] require(len(rows) == 22, 'expected 22 ledger rows') for key in ['proof', 'payload', 'ledger_rows', 'ledger_head', 'tickets', 'config', 'issuer_binding', 'tools', 'scheduling']: require(current[key] == prior[key], 'current/prior authority differs: ' + key) check_state(current, rows) physical = load('live-authority-stress-218/final-physical-evidence.json') current_physical = load('final-release-220/audit/current-physical.json') check_physical(physical, current_physical, rows) journal = parse_journal(physical['journal']) + parse_journal(data['evidence/native-authorized-commit-217/complete-executor-journal.jsonl']) by_ticket = {} for entry in journal: try: message = parse(entry.get('MESSAGE', '')) except (ValueError, TypeError): continue if isinstance(message, dict) and 'ticket' in message: require(entry['_UID'] == entry['_GID'] == '5206', 'native journal executor UID/GID differs') require(entry['_SYSTEMD_UNIT'] == 'eza-executor-001.service', 'native journal unit differs') by_ticket.setdefault(message['ticket'], {})[entry['__CURSOR']] = (entry, message) event_count = 0 previous_proof = rows[0]['data']['proof'] for start_index in range(1, 22, 3): registration, start, end = [row['data'] for row in rows[start_index:start_index + 3]] receipt, proof = end['receipt'], end['proof'] require([row['event'] for row in rows[start_index:start_index + 3]] == ['REGISTER', 'START', 'RESULT'], 'transaction ledger event ordering') require(start['before'] == previous_proof, 'transaction preceding proof mismatch') require(registration['ticket'] == start['ticket'] == end['ticket'] == receipt['ticket'], 'transaction ticket mismatch') require(start['txid'] == receipt['txid'] == receipt['begin']['data']['txid'] == receipt['done']['data']['txid'], 'transaction ID mismatch') require(receipt['binding_sha256'] == sha(wire(start['binding'])), 'receipt binding hash mismatch') require(receipt['begin']['data']['binding'] == start['binding'], 'receipt binding mismatch') for event in [receipt['begin'], receipt['done']]: require(sha(wire({key: value for key, value in event.items() if key != 'sha256'})) == event['sha256'], 'receipt BEGIN/DONE event hash mismatch') event_count += 1 require(receipt['begin']['kind'] == 'BEGIN' and receipt['done']['kind'] == 'DONE', 'receipt event kinds') require(receipt['begin']['previous'] == previous_proof['head'], 'BEGIN preceding witness head mismatch') require(receipt['begin']['data']['before']['sha256'] == previous_proof['payload'], 'BEGIN prior payload mismatch') require(receipt['done']['sha256'] == proof['head'] and receipt['begin']['sha256'] == proof['commitment'], 'receipt/proof head or commitment mismatch') require(receipt['done']['data']['anchor']['sha256'] == proof['payload'], 'receipt/proof payload mismatch') require(receipt['begin']['sequence'] == previous_proof['sequence'] + 1 and receipt['done']['sequence'] == proof['sequence'], 'witness event sequence mismatch') require(proof['revision'] == previous_proof['revision'] + 1, 'revision transition mismatch') entries = by_ticket.get(registration['ticket'], {}) require(len(entries) == 1, 'ticket has missing or ambiguous native journal entry') _, message = next(iter(entries.values())) for key in ['receipt', 'result', 'proof', 'ticket']: require(message[key] == end[key], 'journal/ledger mismatch: ' + key) require(message['head'] == rows[start_index + 2]['sha256'] and message['sequence'] == rows[start_index + 2]['sequence'], 'journal ledger anchor mismatch') require(message['domain'] == rows[start_index + 2]['domain'] and message['status'] == 'RECORDED' and message['authority'] == 'custodian', 'journal recorded authority mismatch') previous_proof = proof require(event_count == 14 and previous_proof == current['proof'], 'event count/final proof mismatch') cases = [] for case, sequence, folder in [ ('authorized', 10, 'native-authorized-commit-217'), ('no-authority', 13, 'live-authority-stress-218/no-authority'), ('out-of-scope-replay', 16, 'live-authority-stress-218/out-of-scope'), ('consumed-grant', 19, 'live-authority-stress-218/consumed-grant'), ('spent-handle', 22, 'live-authority-stress-218/spent-handle'), ]: registration, start, end = [row['data'] for row in rows[sequence - 3:sequence]] record = end['result']['response']['record'] receipt_bytes = data['evidence/' + folder + '/NATIVE-RECEIPT.json'] require(receipt_bytes == wire(end['receipt']), 'saved case receipt bytes differ') before = load(folder + ('/authority-before.json' if case == 'authorized' else '/authority-before.stdout')) after = load(folder + ('/authority-after.json' if case == 'authorized' else '/authority-after.stdout')) for state in [before, after]: check_state(state, rows) require(before['proof'] == start['before'] and after['proof'] == end['proof'], 'case state proof differs from ledger') require(len(after['ledger_rows']) == sequence, 'case final ledger length') result = load(folder + ('/COMMIT-RESULT.json' if case == 'authorized' else '/NATIVE-RESULT.json')) require(result == (end['result'] if case == 'authorized' else record), 'case result differs from ledger/journal') if case == 'authorized': require(before['payload']['accesses'] == [] and before['payload']['consumed_roots'] == [] and len(before['payload']['prepared']) == 1, 'authorized preconditions differ') require(after['payload']['prepared'] == {} and len(after['payload']['consumed_roots']) == 1, 'grant consumption/prepared handle transition differs') require(after['payload']['accesses'] == [{'action': 'write', 'resource': 'object:X', 'value': 'linux-isolation-207-authorized'}], 'authorized effect differs') require(before['payload']['state']['resources'] == {'object:X': 'initial', 'object:Y': 'initial'}, 'authorized initial resource state') require(after['payload']['state']['resources'] == {'object:X': 'linux-isolation-207-authorized', 'object:Y': 'initial'}, 'authorized final resource state') require(record['reason_code'] == 'EZA-AUTHORIZED' and record['protected_effect_count'] == 1 and record['execution'] == 'SUCCEEDED', 'authorized result differs') require(after['payload']['consumed_roots'] == record['authority_lineage'], 'consumed grant differs from authorized lineage') else: require(len(after['ledger_rows']) == len(before['ledger_rows']) + 3, 'refusal ledger growth') for key in ['accesses', 'consumed_roots', 'prepared']: require(before['payload'][key] == after['payload'][key], 'refusal changed protected effect or handle state') for key in ['resources', 'grants', 'revoked']: require(before['payload']['state'][key] == after['payload']['state'][key], 'refusal changed protected resource/authority state') require(record['protected_effect_count'] == 0 and record['execution'] == 'NOT_EXECUTED', 'refusal protected effect differs') require(record['blocking_reason_code'] == {'no-authority': 'EZA-AUTHORITY-ABSENT', 'out-of-scope-replay': 'EZA-REPLAY-DETECTED', 'consumed-grant': 'EZA-REPLAY-DETECTED', 'spent-handle': 'EZA-REQUEST-BINDING-MISMATCH'}[case], 'refusal reason differs') cases.append({'case': case, 'ledger_sequence': sequence, 'receipt_sha256': sha(receipt_bytes), 'result_record': record, 'status': 'PASS'}) return cases, journal def check_lineage(data, projections, cases): lineage = parse(data['PUBLICATION-LINEAGE.json']) require(lineage['public_edition_id'] == EDITION, 'lineage edition mismatch') require(lineage['verdicts'] == VERDICTS, 'claim boundaries changed') require(lineage['evidence_archive_sha256'] == ORIGINAL_ZIP_SHA and lineage['evidence_manifest_sha256'] == ORIGINAL_MANIFEST_SHA, 'historical lineage digest changed') require(lineage['public_derivation']['original_signature_authenticates_public_edition'] is False, 'lineage signature claim') require(lineage['public_derivation']['original_signature_authenticates_projections'] is False, 'lineage projection signature claim') require(lineage['public_derivation']['full_maintenance_replay_publicly_reproducible'] is False, 'maintenance replay limitation missing') require(lineage['public_derivation']['opaque_aliases_are_actual_historical_values'] is False, 'opaque alias interpretation missing') require(lineage['public_derivation']['retirement_or_current_operational_status_established'] is False, 'unproved operational status claim') require('authority' in lineage['authority_boundary'].lower(), 'missing authority boundary') ids, references = {}, [] reference_keys = {'requirement_id', 'experiment_id', 'observation_id', 'claim_id', 'source_artifact_id', 'original_archive_artifact_id', 'artifact_id', 'verification_id', 'limitation_id'} def walk(value): if isinstance(value, dict): if 'id' in value: require(value['id'] not in ids, 'duplicate lineage ID') ids[value['id']] = value for key, item in value.items(): if key in reference_keys or key.endswith('_ids'): for ref in item if isinstance(item, list) else [item]: if isinstance(ref, str) and ref.startswith('eza-001/'): references.append(ref) walk(item) elif isinstance(value, list): for item in value: walk(item) walk(lineage) require(all(ref in ids for ref in references), 'unresolved internal lineage reference') artifacts = {item['id']: item for item in lineage['artifacts']} present, unavailable, projected = 0, 0, 0 for artifact in artifacts.values(): representation = artifact['public_representation'] path = representation['path'] if path is None: require(representation['status'] == 'omitted-reference-only', 'unavailable lineage artifact not explicit') unavailable += 1 continue require(path in data and sha(data[path]) == representation['sha256'] and len(data[path]) == representation['bytes'], 'lineage public artifact bytes differ') if path in projections: require(representation['status'] == 'derived-projection', 'projected lineage artifact status') # Historical nodes retain the original hash; newly added projection nodes use projected hash. require(artifact['sha256'] in {projections[path]['source_sha256'], projections[path]['public_sha256']}, 'lineage source/projected hash conflation') projected += 1 else: require(artifact['sha256'] == representation['sha256'], 'unprojected lineage historical hash differs') present += 1 for relation in lineage['relationships']: for key in ['requirement_id', 'experiment_id', 'claim_id']: require(relation[key] in ids, 'relationship endpoint missing') if relation['observation_id'] is not None: require(relation['observation_id'] in ids, 'relationship observation missing') for aid in relation['evidence_artifact_ids']: require(aid in artifacts, 'relationship evidence artifact missing') require(bool(relation['limitation_ids']), 'relationship lacks limitations') native_observations = {case['case'] for case in cases} if relation['observation_id'] and relation['observation_id'].rsplit('/', 1)[-1] in native_observations: require(bool(relation['verification_ids']), 'native-case relationship lacks verification') for case in cases: observations = [item for item in lineage['observations'] if item['id'].endswith('/observation/' + case['case'])] require(len(observations) == 1 and observations[0]['recorded_result'] == case['result_record'], 'lineage observation differs from raw saved case') require(observations[0]['externally_attested'] is False, 'new external attestation inferred') return lineage, {'internal_references': len(references), 'included_artifact_references': present, 'unavailable_artifact_references': unavailable, 'projected_artifact_references': projected, 'relationships': len(lineage['relationships'])} def check_text_and_privacy(data, journal): disclosure = parse(data['DISCLOSURE.json']) require(set(disclosure['source_disclosure_set']) == PROJECTION_PATHS | OMITTED_PATHS, 'disclosure scope mismatch') require(set(disclosure['projection_paths']) == PROJECTION_PATHS and set(disclosure['omitted_paths']) == OMITTED_PATHS, 'disclosure disposition differs from actual privacy scope') require(disclosure['alias_policy']['synthetic'] is True and disclosure['alias_policy']['historical_values'] is False, 'opaque alias policy missing synthetic/nonhistorical status') require(disclosure['raw_identity_mapping_included'] is False and disclosure['authority_from_retrieval'] is False, 'disclosure privacy or authority boundary differs') require(disclosure['accepted_ssh_authentication_records_omitted'] == 470, 'authentication omission scope mismatch') magic = [b'\x7fELF', b'\xcf\xfa\xed\xfe', b'\xfe\xed\xfa\xcf', b'\xca\xfe\xba\xbe', b'MZ', b'PK\x03\x04', b'\x1f\x8b'] decoded = 0 for name, value in data.items(): value.decode('utf-8') require(not any(value.startswith(prefix) for prefix in magic), 'redistributed native/archive payload') require(not name.endswith(('.so', '.dll', '.dylib', '.whl', '.egg', '.tar', '.tar.gz', '.tar.xz', '.exe', '.pyc')), 'redistributed binary extension') for token in re.findall(rb'(?