514 lines
27 KiB
Python
514 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only P22 record integrity; exit 2 for unmet technical readiness."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import io
|
|
import math
|
|
import wave
|
|
import zlib
|
|
from pathlib import Path
|
|
import struct
|
|
import sys
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
OUT = 'docs/plans/original-completion-2026-10-02/execution/p22-02/'
|
|
INPUT = 'docs/spec/inputs/P22-02-accepted-usage.json'
|
|
CURRENT = 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-slot-contract-r02/all-slot-current-candidate-contracts.json'
|
|
SPEC_SHA = 'f252de3c5cff7ebfa62322ae2f2df09fb80a44bc59eb286b76d4acb4c12ca875'
|
|
INPUT_SHA = '45fa52acf0c8bd4dd58c3b40b681cb92066f05d617df157fc18370a5e8816281'
|
|
|
|
def need(ok, message):
|
|
if not ok:
|
|
raise ValueError(message)
|
|
|
|
def sha(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
def semantic(value):
|
|
return sha(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False).encode())
|
|
|
|
def pairs(items):
|
|
result = {}
|
|
for key, value in items:
|
|
need(key not in result, 'duplicate JSON key: ' + key)
|
|
result[key] = value
|
|
return result
|
|
|
|
def invalid_constant(value):
|
|
raise ValueError('nonfinite JSON number: ' + value)
|
|
|
|
|
|
def finite_float(value):
|
|
number = float(value)
|
|
need(math.isfinite(number), 'nonfinite JSON number')
|
|
return number
|
|
|
|
|
|
def loads(value):
|
|
return json.loads(value, object_pairs_hook=pairs, parse_constant=invalid_constant, parse_float=finite_float)
|
|
|
|
|
|
def read(path):
|
|
return loads(raw(path))
|
|
|
|
def decode(doc, node=None):
|
|
need(type(doc) is dict and all(type(doc.get(k)) is list for k in ('keys', 'strings', 'pool')), 'wrong codec tables')
|
|
need(all(type(k) is str for k in doc['keys']) and len(set(doc['keys'])) == len(doc['keys']), 'duplicate/wrong key table')
|
|
need(all(type(s) is str for s in doc['strings']), 'wrong string table')
|
|
active = set()
|
|
def index(values, i):
|
|
need(type(i) is int and 0 <= i < len(values), 'bad reference index')
|
|
return values[i]
|
|
def walk(n):
|
|
if not isinstance(n, list):
|
|
need(n is None or type(n) in (str, int, float, bool), 'wrong scalar type')
|
|
need(type(n) is not float or math.isfinite(n), 'nonfinite codec scalar')
|
|
return n
|
|
need(bool(n) and type(n[0]) is str, 'wrong node type')
|
|
tag = n[0]
|
|
if tag == 'l':
|
|
return [walk(x) for x in n[1:]]
|
|
if tag == 'd':
|
|
need(len(n) % 2 == 1, 'malformed flat dictionary')
|
|
result = {}
|
|
for i in range(1, len(n), 2):
|
|
key = index(doc['keys'], n[i])
|
|
need(type(key) is str and key not in result, 'duplicate or invalid decoded key')
|
|
result[key] = walk(n[i + 1])
|
|
return result
|
|
need(tag in ('s', 'r') and len(n) == 2, 'unknown node')
|
|
if tag == 's':
|
|
value = index(doc['strings'], n[1])
|
|
need(type(value) is str, 'wrong shared string type')
|
|
return value
|
|
i = n[1]
|
|
value = index(doc['pool'], i)
|
|
need(i not in active and isinstance(value, list) and value and value[0] in ('d', 'l'), 'cycle or wrong shared object type')
|
|
active.add(i)
|
|
result = walk(value)
|
|
active.remove(i)
|
|
return result
|
|
result = walk(doc['files'] if node is None else node)
|
|
for i in range(len(doc['pool'])):
|
|
walk(['r', i])
|
|
return result
|
|
|
|
# A pinned historical observation is retained, never treated as current readiness.
|
|
EVIDENCE = 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-current-contract-r01/'
|
|
CHECKPOINTS = 'docs/plans/original-completion-2026-10-02/checkpoints/'
|
|
CURRENT_SHA = '5cf93b6b3fd9d0cc060075878804f65195c0d90711c969cc36e2f58f3a1917fd'
|
|
QUANTITIES = ('required_file_count', 'required_frame_count', 'required_variant_count')
|
|
PROOF_KINDS = ('fit', 'rights', 'layout', 'quantity', 'gaps', 'usability')
|
|
|
|
|
|
def safe_path(path):
|
|
need(type(path) is str and path and not Path(path).is_absolute(), 'invalid relative path')
|
|
need('..' not in Path(path).parts, 'path escape')
|
|
resolved = (ROOT / path).resolve()
|
|
need(resolved != ROOT and ROOT in resolved.parents, 'path escape or symlink escape')
|
|
return resolved
|
|
|
|
|
|
def raw(path):
|
|
return safe_path(path).read_bytes()
|
|
|
|
|
|
def pinned(path, digest):
|
|
need(type(digest) is str and len(digest) == 64 and all(c in '0123456789abcdef' for c in digest), 'invalid SHA')
|
|
value = raw(path)
|
|
need(sha(value) == digest, 'raw SHA mismatch: ' + path)
|
|
return value
|
|
|
|
|
|
def exact(a, b):
|
|
# Python equality would otherwise equate True and 1.
|
|
return semantic(a) == semantic(b)
|
|
|
|
|
|
def anchor(a):
|
|
need(type(a) is dict and set(a) == {'path', 'sha256', 'line', 'text'}, 'anchor schema')
|
|
lines = pinned(a['path'], a['sha256']).decode('utf-8').splitlines()
|
|
need(type(a['line']) is int and 1 <= a['line'] <= len(lines), 'anchor line')
|
|
need(type(a['text']) is str and a['text'].strip() and lines[a['line'] - 1] == a['text'], 'reference mismatch')
|
|
|
|
|
|
def png_format(data):
|
|
need(data[:8] == b'\x89PNG\r\n\x1a\n', 'PNG signature')
|
|
pos, chunks, payload, palette, transparent = 8, [], bytearray(), False, False
|
|
width = height = depth = color = interlace = None
|
|
while pos < len(data):
|
|
need(pos + 12 <= len(data), 'PNG truncated chunk')
|
|
size = struct.unpack('>I', data[pos:pos + 4])[0]
|
|
kind = data[pos + 4:pos + 8]
|
|
end = pos + size + 12
|
|
need(end <= len(data), 'PNG chunk length')
|
|
body = data[pos + 8:end - 4]
|
|
crc = struct.unpack('>I', data[end - 4:end])[0]
|
|
need(zlib.crc32(kind + body) & 0xffffffff == crc, 'PNG CRC')
|
|
if not chunks:
|
|
need(kind == b'IHDR' and size == 13, 'PNG IHDR')
|
|
width, height, depth, color, comp, filt, interlace = struct.unpack('>IIBBBBB', body)
|
|
need(width > 0 and height > 0 and comp == filt == 0 and interlace in (0, 1), 'PNG dimensions/encoding')
|
|
depths = {0: (1, 2, 4, 8, 16), 2: (8, 16), 3: (1, 2, 4, 8), 4: (8, 16), 6: (8, 16)}
|
|
need(color in depths and depth in depths[color], 'PNG color/depth')
|
|
else:
|
|
need(kind != b'IHDR', 'duplicate PNG IHDR')
|
|
if kind == b'PLTE':
|
|
need(not payload and size > 0 and size % 3 == 0 and size <= 768, 'PNG palette')
|
|
palette = True
|
|
if kind == b'tRNS':
|
|
transparent = True
|
|
if kind == b'IDAT':
|
|
need(not payload or chunks[-1] == b'IDAT', 'PNG noncontiguous image data')
|
|
payload.extend(body)
|
|
chunks.append(kind)
|
|
pos = end
|
|
if kind == b'IEND':
|
|
need(size == 0 and pos == len(data), 'PNG trailing bytes')
|
|
break
|
|
need(chunks[-1:] == [b'IEND'] and payload and (color != 3 or palette), 'PNG missing image/palette/end')
|
|
# Bound decoded length by IHDR; validate scanline sizes and filter tags.
|
|
channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color]
|
|
passes = [(0, 0, 1, 1)] if interlace == 0 else [(0, 0, 8, 8), (4, 0, 8, 8), (0, 4, 4, 8), (2, 0, 4, 4), (0, 2, 2, 4), (1, 0, 2, 2), (0, 1, 1, 2)]
|
|
rows = []
|
|
for x, y, dx, dy in passes:
|
|
w, h = max(0, (width - x + dx - 1) // dx), max(0, (height - y + dy - 1) // dy)
|
|
if w and h:
|
|
rows.extend([1 + (w * channels * depth + 7) // 8] * h)
|
|
expected = sum(rows)
|
|
need(expected <= 268435456, 'PNG exceeds verification memory bound')
|
|
decoder = zlib.decompressobj()
|
|
pixels = decoder.decompress(bytes(payload), expected + 1)
|
|
need(decoder.eof and not decoder.unused_data and len(pixels) == expected, 'PNG scanline length')
|
|
offset = 0
|
|
for length in rows:
|
|
need(pixels[offset] <= 4, 'PNG scanline filter')
|
|
offset += length
|
|
return {'png_width': width, 'png_height': height, 'png_color_type': color,
|
|
'png_crc_verified': True, 'png_alpha_encoding_present': color in (4, 6) or transparent}
|
|
|
|
|
|
def observe(path, data):
|
|
if path.endswith('.png'):
|
|
return png_format(data)
|
|
if path.endswith('.wav'):
|
|
with wave.open(io.BytesIO(data), 'rb') as w:
|
|
need(w.getcomptype() == 'NONE', 'WAV compression')
|
|
return {'channels': w.getnchannels(), 'sample_rate': w.getframerate(),
|
|
'sample_width_bytes': w.getsampwidth(), 'frames': w.getnframes()}
|
|
if path.endswith('.ogg'):
|
|
need(data[:4] == b'OggS' and b'\x01vorbis' in data[:128], 'Ogg/Vorbis representation')
|
|
return {'encoding': 'ogg-vorbis', 'bytes': len(data)}
|
|
data.decode('utf-8')
|
|
return {'representation': 'code', 'bytes': len(data)}
|
|
|
|
|
|
def accepted_records(accepted):
|
|
original = decode(accepted)
|
|
need(set(original) == set(accepted['source_files']), 'accepted file coverage')
|
|
records = {}
|
|
for name, rows in original.items():
|
|
need(type(rows) is list, 'decoded file must be list')
|
|
pin = accepted['source_files'][name]
|
|
need(type(pin['records']) is int and len(rows) == pin['records'] and semantic(rows) == pin['semantic_sha256'], 'decoded semantic mismatch: ' + name)
|
|
for row in rows:
|
|
need(type(row) is dict, 'accepted row type')
|
|
identity = row.get('slot_id', row.get('group_id'))
|
|
need(type(identity) is str and identity not in records, 'duplicate/invalid identity')
|
|
records[identity] = (name, row)
|
|
need(len(records) == 614 and accepted['decoded_record_count'] == 614, 'accepted coverage')
|
|
need(sum('group_id' in row for _, row in records.values()) == 2, 'state groups')
|
|
return original, records
|
|
|
|
|
|
def proof_matches(proof, kind, subject):
|
|
"""Pure proof predicate; callers supply a pinned Root/owner receipt, not booleans."""
|
|
need(type(proof) is dict and set(proof) == {'kind', 'subject', 'result', 'issuer', 'observations'}, 'proof schema')
|
|
need(proof['kind'] == kind and exact(proof['subject'], subject), 'wrong proof binding')
|
|
need(proof['result'] == 'accepted', 'proof is not accepted')
|
|
issuer = proof['issuer']
|
|
need(type(issuer) is dict and set(issuer) == {'role', 'name'}, 'proof issuer schema')
|
|
need(issuer['role'] in ('root', 'owner') and type(issuer['name']) is str and issuer['name'].strip(), 'proof authority')
|
|
need(type(proof['observations']) is list and proof['observations'], 'proof has no observations')
|
|
for observation in proof['observations']:
|
|
need(type(observation) is dict and set(observation) == {'explanation', 'anchors'}, 'observation schema')
|
|
need(type(observation['explanation']) is str and observation['explanation'].strip(), 'empty observation')
|
|
need(type(observation['anchors']) is list and observation['anchors'], 'no observation anchors')
|
|
return True
|
|
|
|
|
|
def receipt(ref, kind, subject):
|
|
# Workers cannot manufacture approval inside their own output directory.
|
|
need(type(ref) is dict and set(ref) == {'path', 'sha256', 'index'}, 'proof reference schema')
|
|
need(ref['path'].startswith(CHECKPOINTS), 'proof must be a Root/owner checkpoint')
|
|
resolved = safe_path(ref['path'])
|
|
need((ROOT / CHECKPOINTS).resolve() in resolved.parents, 'proof symlink outside checkpoints')
|
|
document = loads(pinned(ref['path'], ref['sha256']))
|
|
need(type(document) is dict and document.get('schema') == 'p22-evidence-v1', 'receipt schema')
|
|
proofs = document['proofs']
|
|
need(type(proofs) is list and type(ref['index']) is int and 0 <= ref['index'] < len(proofs), 'proof index')
|
|
proof = proofs[ref['index']]
|
|
proof_matches(proof, kind, subject)
|
|
for observation in proof['observations']:
|
|
for a in observation['anchors']:
|
|
anchor(a)
|
|
return proof
|
|
|
|
|
|
def reference_contracts(snapshot, records):
|
|
contracts, used = {}, set()
|
|
for c in snapshot['contracts']:
|
|
identity = c['id']
|
|
need(identity in records and identity not in contracts, 'contract coverage/duplicate')
|
|
name, row = records[identity]
|
|
need(c['source_file'] == name and c['accepted_record_semantic_sha256'] == semantic(row), 'contract accepted record')
|
|
need(exact(c['source_anchors'], row['source_anchors']) and exact(c['usage_condition'], row['usage_condition']), 'accepted anchors/usage omitted')
|
|
for a in c['source_anchors']:
|
|
anchor(a)
|
|
refs = c['candidate_refs']
|
|
need(len({r['candidate_ref'] for r in refs}) == len(refs), 'duplicate candidate')
|
|
need({r['candidate_ref'] for r in refs} == {r['path'] for r in row['representations']}, 'source omission')
|
|
historical = {r['path']: r for r in row['representations']}
|
|
for ref in refs:
|
|
source = ref['candidate_ref']
|
|
need(source in snapshot['candidate_pool'], 'source absent from pool')
|
|
used.add(source)
|
|
need(ref['prior_raw_sha256'] == historical[source]['sha256'], 'historical pin mismatch')
|
|
need(type(ref['prior_source_still_equal']) is bool and ref['prior_source_still_equal'] == (ref['prior_raw_sha256'] == snapshot['candidate_pool'][source]['current_raw_sha256']), 'historical applicability')
|
|
contracts[identity] = c
|
|
need(set(contracts) == set(records), 'missing accepted use')
|
|
need(used == set(snapshot['candidate_pool']) and len(used) == 19, '19-source coverage')
|
|
return contracts
|
|
|
|
|
|
def quantities_valid(values, row):
|
|
need(type(values) is dict and set(values) == set(QUANTITIES), 'quantity schema')
|
|
for key, value in values.items():
|
|
need(value is None or type(value) is int and value > 0, 'invalid quantity')
|
|
# Unknown may only be resolved by the separate bound quantity receipt.
|
|
if row.get(key) is not None:
|
|
need(exact(value, row[key]), 'required quantity mismatch')
|
|
|
|
|
|
def selection_subject(identity, row, c, selection, pins):
|
|
return {'id': identity, 'accepted_record_semantic_sha256': semantic(row),
|
|
'usage_condition': row['usage_condition'], 'source_pins': pins,
|
|
'selection': selection,
|
|
# Includes every original unmade/provisional/disclosure field, not
|
|
# just c.missing_conditions. No empty-gap denominator is possible.
|
|
'original_record': row, 'missing_conditions': c['missing_conditions']}
|
|
|
|
|
|
def asset_matches(a, row, sources):
|
|
"""Pure source/format/layout binding, shared by real and isolated tests."""
|
|
need(type(a) is dict and set(a) == {'path', 'sha256', 'format', 'region', 'code_anchor', 'kind'}, 'asset schema')
|
|
need(a['path'] in sources and a['sha256'] == sources[a['path']]['sha256'], 'selected source pin')
|
|
need(exact(a['format'], sources[a['path']]['format']), 'selected dimension/representation mismatch')
|
|
need(a['kind'] in ('image', 'audio', 'code', 'state_group'), 'fallback/provisional kind')
|
|
if 'group_id' in row:
|
|
need(a['kind'] == 'state_group', 'state group promoted to image')
|
|
if a['path'].endswith('.png'):
|
|
need(a['kind'] in ('image', 'state_group'), 'image kind mismatch')
|
|
region = a['region']
|
|
need(type(region) is list and len(region) == 4 and all(type(x) is int for x in region), 'region schema')
|
|
x, y, w, h = region
|
|
need(x >= 0 and y >= 0 and w > 0 and h > 0 and x + w <= a['format']['png_width'] and y + h <= a['format']['png_height'], 'layout region outside source')
|
|
need(a['code_anchor'] is None, 'image code representation')
|
|
elif a['path'].endswith(('.wav', '.ogg')):
|
|
need(a['kind'] == 'audio' and a['region'] is None and a['code_anchor'] is None, 'audio representation')
|
|
else:
|
|
need(a['kind'] in ('code', 'state_group') and a['region'] is None, 'code representation')
|
|
need(type(a['code_anchor']) is dict, 'missing code anchor')
|
|
need(a['code_anchor']['path'] == a['path'] and a['code_anchor']['sha256'] == a['sha256'], 'wrong code anchor')
|
|
|
|
|
|
def selection_ready(identity, row, c, choice, sources):
|
|
if type(choice) is str:
|
|
need(choice == ('candidate' if c['candidate_refs'] else 'missing'), 'candidate/missing status mismatch')
|
|
return False
|
|
need(type(choice) is dict and set(choice) == {'state', 'contract', 'proofs'}, 'selection schema')
|
|
need(choice['state'] == 'selected', 'counterfeit final state')
|
|
s = choice['contract']
|
|
fields = {'asset', 'creator', 'rights', 'name', 'replacement_chain', 'references', 'quantities', 'quantity_basis', 'usability'}
|
|
need(type(s) is dict and set(s) == fields, 'selected contract fields')
|
|
a = s['asset']
|
|
asset_matches(a, row, sources)
|
|
if a['code_anchor'] is not None:
|
|
anchor(a['code_anchor'])
|
|
need(type(s['creator']) is str and s['creator'].strip().lower() not in ('', 'unknown', 'unassigned'), 'unknown creator')
|
|
need(type(s['name']) is str and s['name'] == Path(a['path']).name, 'naming mismatch')
|
|
need(type(s['quantity_basis']) is str and s['quantity_basis'].strip(), 'quantity basis missing')
|
|
quantities_valid(s['quantities'], row)
|
|
chain = s['replacement_chain']
|
|
need(type(chain) is list and len(chain) >= 2 and chain[-1] == a['path'], 'replacement chain')
|
|
need(chain[0] == identity and all(type(x) is str and x for x in chain), 'replacement identity')
|
|
need(type(s['references']) is list and s['references'], 'missing product reference')
|
|
pins = {a['path']: a['sha256']}
|
|
for ref in s['references']:
|
|
anchor(ref)
|
|
need(ref['path'].startswith(('scripts/', 'scenes/', 'assets/')), 'product reference must be in product')
|
|
pins[ref['path']] = ref['sha256']
|
|
rights = s['rights']
|
|
need(type(rights) is dict and set(rights) == {'license', 'origin', 'documents'}, 'rights schema')
|
|
need(type(rights['license']) is str and rights['license'].strip().lower() not in ('', 'unknown', 'unassigned'), 'unknown license')
|
|
need(type(rights['origin']) is str and rights['origin'].strip(), 'rights origin')
|
|
need(type(rights['documents']) is list and rights['documents'], 'missing rights evidence')
|
|
for doc in rights['documents']:
|
|
need(type(doc) is dict and set(doc) == {'path', 'sha256'}, 'rights document schema')
|
|
pinned(doc['path'], doc['sha256'])
|
|
pins[doc['path']] = doc['sha256']
|
|
need(type(s['usability']) is dict and set(s['usability']) == {'display', 'hearing'}, 'usability schema')
|
|
for mode, explanation in s['usability'].items():
|
|
need(type(explanation) is str and explanation.strip(), 'applicability needs explanation, not boolean')
|
|
need(type(choice['proofs']) is dict and set(choice['proofs']) == set(PROOF_KINDS), 'missing required proof')
|
|
subject = selection_subject(identity, row, c, s, pins)
|
|
for kind in PROOF_KINDS:
|
|
receipt(choice['proofs'][kind], kind, subject)
|
|
return True
|
|
|
|
|
|
def technical_readiness(required, selected, unit_proofs, subject):
|
|
"""Pure reducer. No review/integration/final, no global all-false gate."""
|
|
need(type(required) is list and required and len(set(required)) == len(required), 'nonempty unique denominator')
|
|
need(type(selected) is dict and set(selected) == set(required), 'readiness exact coverage')
|
|
need(all(type(v) is bool for v in selected.values()), 'readiness booleans must be derived')
|
|
need(type(unit_proofs) is dict and set(unit_proofs) <= {'mandatory_scope', 'disclosure'}, 'unit proof schema')
|
|
for kind, proof in unit_proofs.items():
|
|
proof_matches(proof, kind, subject)
|
|
return all(selected.values()) and set(unit_proofs) == {'mandatory_scope', 'disclosure'}
|
|
|
|
|
|
def reconstruct(ledger):
|
|
"""Lossless public expansion; references are data, never omitted fields.
|
|
|
|
The caller can inspect every input/provenance field and every historical
|
|
contract. Selection is a separate plane and cannot overwrite that history.
|
|
This function performs no writes and confers no readiness.
|
|
"""
|
|
documents = {}
|
|
for source, digest in ledger['references'].items():
|
|
data = pinned(source, digest)
|
|
documents[source] = loads(data) if source.endswith('.json') else data.decode('utf-8')
|
|
return {'reference_documents': documents,
|
|
'accepted_files': decode(documents[INPUT]),
|
|
'historical_contracts': documents[CURRENT],
|
|
'selections': ledger['selections'], 'source_pins': ledger['source_pins'],
|
|
'unit_proofs': ledger['unit_proofs']}
|
|
|
|
|
|
def check(path, accepted_input=INPUT):
|
|
need(accepted_input == INPUT, 'accepted input path must match fixed contract')
|
|
spec = raw('docs/spec/acceptance/P22-02.json')
|
|
need(sha(spec.decode().strip().encode()) == SPEC_SHA, 'contract strip spec SHA')
|
|
need(sha(spec) == 'c26c0b12cf7e7a7c46e671acde2e00762904732c1cd6381a53836e3d7b3f9824', 'raw spec SHA')
|
|
accepted = loads(pinned(INPUT, INPUT_SHA))
|
|
original, records = accepted_records(accepted)
|
|
ledger = read(path)
|
|
need(type(ledger) is dict and set(ledger) == {'schema', 'references', 'reconstruction', 'selections', 'source_pins', 'unit_proofs'}, 'ledger schema')
|
|
need(ledger['schema'] == 'p22-reference-contract-v2', 'ledger version')
|
|
expected = {INPUT, CURRENT} | {str(p.relative_to(ROOT)) for p in safe_path(EVIDENCE).iterdir() if p.is_file()}
|
|
need(set(ledger['references']) == expected, 'reference omission/extra')
|
|
need(ledger['references'][INPUT] == INPUT_SHA and ledger['references'][CURRENT] == CURRENT_SHA, 'immutable reference pin')
|
|
expanded = reconstruct(ledger)
|
|
snapshot = expanded['historical_contracts']
|
|
contracts = reference_contracts(snapshot, records)
|
|
selections = ledger['selections']
|
|
need(type(selections) is dict and set(selections) == set(records), 'selection exact 614 coverage')
|
|
required_sources = set(snapshot['candidate_pool'])
|
|
for choice in selections.values():
|
|
if type(choice) is dict:
|
|
required_sources.add(choice['contract']['asset']['path'])
|
|
need(type(ledger['source_pins']) is dict and set(ledger['source_pins']) == required_sources, 'current source omission/extra')
|
|
sources = {}
|
|
for source, digest in ledger['source_pins'].items():
|
|
data = pinned(source, digest)
|
|
sources[source] = {'sha256': digest, 'format': observe(source, data)}
|
|
historic = snapshot['candidate_pool'].get(source)
|
|
if historic:
|
|
need(historic['path'] == source, 'source path mismatch')
|
|
if digest == historic['current_raw_sha256'] and source.endswith('.png'):
|
|
need(exact(sources[source]['format'], historic['observed_format']), 'observed dimension mismatch')
|
|
verify_existing_evidence(sources)
|
|
selected = {i: selection_ready(i, records[i][1], contracts[i], selections[i], sources) for i in records}
|
|
subject = {'accepted_input_raw_sha256': INPUT_SHA, 'source_pins': ledger['source_pins'],
|
|
'selection_sha256': semantic(selections), 'required_ids': sorted(records),
|
|
'original_denominator': {'packages': 30, 'delivery_units': 88, 'no_partial_credit': True},
|
|
'unresolved_scope_requires_owner_evidence': ['D08', 'D09']}
|
|
need(type(ledger['unit_proofs']) is dict, 'unit proofs type')
|
|
unit_proofs = {kind: receipt(ref, kind, subject) for kind, ref in ledger['unit_proofs'].items()}
|
|
ready = technical_readiness(list(records), selected, unit_proofs, subject)
|
|
return {'record_integrity': True, 'validation_ready': ready,
|
|
'pending_identity_count': sum(not v for v in selected.values()),
|
|
'selected_usable_count': sum(selected.values()), 'accepted_identity_count': len(records),
|
|
'candidate_source_count': len(snapshot['candidate_pool']),
|
|
'unit_proofs_missing': sorted({'mandatory_scope', 'disclosure'} - set(unit_proofs)),
|
|
'original_completion': 'not_claimed; requires formal validate, independent review, integration, final',
|
|
'acceptance_claimed': False}
|
|
|
|
|
|
def verify_existing_evidence(sources):
|
|
"""Rebind existing publisher/member/license/manifest and layout to bytes."""
|
|
packs = read(EVIDENCE + 'publisher-archive-byte-verification.json')
|
|
manifests = read('assets/kenney/sources.json')
|
|
need(sha(raw('assets/kenney/sources.json')) == packs['source_manifest_sha256'], 'Kenney manifest pin')
|
|
for pack in packs['packs']:
|
|
matching = [m for m in manifests if m['slug'] == pack['slug']]
|
|
need(len(matching) == 1, 'publisher manifest coverage')
|
|
m = matching[0]
|
|
need(m['page'] == pack['official_page'] and m['zip_sha256'] == pack['archive_sha256'] and m['download_url'] == pack['official_archive_url'], 'publisher mismatch')
|
|
for member in pack['members']:
|
|
# Publisher facts only apply to exactly equal bytes. A changed
|
|
# selection needs its own new rights evidence, never inherited CC0.
|
|
if member['repo_path'] in sources and sources[member['repo_path']]['sha256'] == member['raw_sha256']:
|
|
pinned(member['repo_path'], member['raw_sha256'])
|
|
licenses = [x for x in pack['members'] if x['archive_member'].lower() == 'license.txt']
|
|
need(len(licenses) == 1, 'publisher license member')
|
|
pinned(licenses[0]['repo_path'], licenses[0]['raw_sha256'])
|
|
ui = read(EVIDENCE + 'ui-audio-publisher-verification.json')
|
|
manifest = loads(pinned('assets/audio/ui/source_manifest.json',
|
|
'f981c2ce12ed364010217dee996908f037d2abfcc3b21b8cf501dd591f7744c9'))
|
|
need(manifest['official_page'] == ui['official_page'] and manifest['official_archive_sha256'] == ui['archive_sha256'], 'UI publisher mismatch')
|
|
pinned(manifest['license']['path'], manifest['license']['sha256'])
|
|
need(manifest['license']['name'] == 'Creative Commons Zero, CC0', 'UI license mismatch')
|
|
need(len(manifest['assets']) == len(ui['assets']) == 4, 'UI asset coverage')
|
|
for member in ui['assets']:
|
|
matches = [a for a in manifest['assets'] if a['path'] == member['repo_path']]
|
|
need(len(matches) == 1 and matches[0]['sha256'] == member['raw_sha256'] and matches[0]['bytes'] == member['bytes'], 'UI member manifest mismatch')
|
|
observed = read(EVIDENCE + 'observed-existing-asset-contracts.json')
|
|
# These layout/reference observations are valid only with their exact
|
|
# dependency pins. Changes require a refreshed external observation.
|
|
for source, digest in observed['source_pins'].items():
|
|
pinned(source, digest)
|
|
atlas = observed['pixel_atlas']
|
|
need(atlas['columns'] * atlas['cell'][0] == atlas['physical_dimensions'][0] and atlas['rows'] * atlas['cell'][1] == atlas['physical_dimensions'][1] and atlas['gap'] == 0, 'atlas layout')
|
|
for source in atlas['vendor_paths_preserved']:
|
|
fmt = sources[source]['format']
|
|
need([fmt['png_width'], fmt['png_height']] == atlas['physical_dimensions'], 'atlas dimensions/layout mismatch')
|
|
internal = read(EVIDENCE + 'internal-origin-observations.json')
|
|
for bgm in internal['bgm']:
|
|
source = sources[bgm['path']]
|
|
if source['sha256'] == bgm['raw_sha256']:
|
|
need(exact(source['format'], bgm['format_observed']), 'BGM representation mismatch')
|
|
# Origin declarations remain historical facts. They never create proof.
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--ledger', default=OUT + 'asset-contracts.json')
|
|
parser.add_argument('--verify-only', action='store_true', help='Read-only verification (also the default)')
|
|
parser.add_argument('--accepted-input', default=INPUT)
|
|
parser.add_argument('--require-ready', action='store_true')
|
|
args = parser.parse_args()
|
|
try:
|
|
result = check(args.ledger, args.accepted_input)
|
|
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
|
|
return 2 if args.require_ready and not result['validation_ready'] else 0
|
|
except (ValueError, KeyError, TypeError, IndexError, OSError, RecursionError, struct.error, zlib.error, wave.Error, UnicodeError) as exc:
|
|
print(json.dumps({'record_integrity': False, 'error': str(exc), 'acceptance_claimed': False}))
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|