tricu

arboricx_readers.py (11263B)

---

 #!/usr/bin/env python3
 """Opt-in reader rewrite checks against explicit full module bundles.
 
 Independent field/truncation oracles plus full-tree differential malformed-input
 and recursive-worker probes. Does not compile, publish, or prove source identity.
 This is a frozen historical-cohort comparator (including unit suffix loss),
 not the current-candidate oracle. Use bundle_batch.py for rewritten boundaries.
 """
 import argparse
 import hashlib
 import json
 from pathlib import Path
 
 from http_headers import LEAF, TRUE, expect, number, pair, result, tree_list
 from source_order import Runner
 from stable_bundle import byte_tree
 
 
 def product(xs):
     return xs[0] if len(xs) == 1 else pair(xs[0], product(xs[1:]))
 
 
 def u32(n):
     return n.to_bytes(4, 'big')
 
 
 def string(s):
     return u32(len(s)) + s
 
 
 class Failure(Exception):
     def __init__(self, code, rest):
         self.tree = result(False, number(code), byte_tree(rest))
 
 
 class Reader:
     def __init__(self, data):
         self.data = data
 
     def take(self, n):
         if len(self.data) < n:
             raise Failure(1, self.data)
         value, self.data = self.data[:n], self.data[n:]
         return value
 
     def magic(self, value):
         original = self.data
         for expected in value:
             if not self.data:
                 raise Failure(1, original)
             if self.take(1)[0] != expected:
                 raise Failure(2, original)
 
     def string(self):
         return self.take(int.from_bytes(self.take(4), 'big'))
 
     def root(self):
         return pair(byte_tree(self.take(4)), byte_tree(self.string()))
 
     def export(self):
         return product([byte_tree(self.string()), byte_tree(self.take(4)),
                         byte_tree(self.string()), byte_tree(self.string())])
 
     def tlv(self):
         return pair(byte_tree(self.take(2)), byte_tree(self.string()))
 
     def repeated(self, worker):
         return tree_list([worker() for _ in range(int.from_bytes(self.take(4), 'big'))])
 
     def core(self):
         self.magic(b'ARBMNFST')
         self.take(2)
         self.take(2)
         fields = [byte_tree(self.string()) for _ in range(9)]
         fields += [self.repeated(lambda: byte_tree(self.string())), byte_tree(self.take(1)),
                    self.repeated(self.root), self.repeated(self.export)]
         return product(fields)
 
 
 def cases():
     core = (b'ARBMNFST' + b'\x00\x01\x00\x00' + b''.join(string(x) for x in
             [b'schema', b'bundle', b'tree', b'hash', b'domain', b'payload', b'semantics', b'eval', b'abi'])
             + u32(2) + string(b'fs') + string(b'net') + b'\0'
             + u32(1) + u32(0) + string(b'main')
             + u32(1) + string(b'main') + u32(0) + string(b'tree') + string(b'abi'))
     specs = [
         ('manifest', 'readLengthPrefixedString', string(b'a\0\xff'), lambda r: byte_tree(r.string())),
         ('manifest', 'readRootEntry', u32(7) + string(b'role'), lambda r: r.root()),
         ('manifest', 'readExportEntry', string(b'name') + u32(7) + string(b'kind') + string(b'abi'), lambda r: r.export()),
         ('manifest', 'readTLV', b'\0\x03' + string(b'a\0\xff'), lambda r: r.tlv()),
         ('manifest', 'readManifestCore', core, lambda r: r.core()),
         ('common', 'readArboricxHeader', b'ARBORICX' + bytes(range(24)),
          lambda r: (r.magic(b'ARBORICX'), product([byte_tree(r.take(n)) for n in [2, 2, 4, 8, 8]]))[1]),
         ('common', 'readSectionRecord', bytes(range(32)),
          lambda r: product([byte_tree(r.take(n)) for n in [4, 2, 2, 2, 2, 8, 8, 4]])),
         ('nodes', 'readNodeRecord', u32(5) + b'\x01\0\0\0\0', lambda r: byte_tree(r.string())),
     ]
     for module, export, encoded, oracle in specs:
         inputs = [(f'cut-{i}', encoded[:i]) for i in range(len(encoded))]
         inputs += [('exact', encoded), ('suffix', encoded + b'\0\xffTAIL')]
         if export in ('readManifestCore', 'readArboricxHeader'):
             inputs += [('bad-magic', b'!' + encoded[1:])]
         for label, data in inputs:
             r = Reader(data)
             try:
                 value = oracle(r)
                 expected = result(True, value, byte_tree(r.data))
             except Failure as error:
                 expected = error.tree
             yield module, export + '-' + label, export, [byte_tree(data)], expected
     # Historical full-manifest entrypoint drops its successful suffix. Retain
     # this independent expectation separately from the field-reader protocol.
     full = core + u32(1) + b'\0\x03' + string(b'metadata') + u32(1) + b'\0\x05' + string(b'extension')
     expected = result(True, pair(Reader(core).core(), pair(
         tree_list([pair(byte_tree(b'\0\x03'), byte_tree(b'metadata'))]), LEAF)))
     for suffix in [b'', b'TAIL']:
         yield 'manifest', 'full-suffix-' + suffix.hex(), 'readManifest', [byte_tree(full + suffix)], expected
     # Counted workers: zero, multiple items, exact suffix and failure remainder.
     for module, export, one, value in [
         ('manifest', 'readCapabilities', string(b'x'), byte_tree(b'x')),
         ('manifest', 'readRoots', u32(0) + string(b'r'), pair(byte_tree(u32(0)), byte_tree(b'r'))),
         ('manifest', 'readExports', string(b'e') + u32(0) + string(b'k') + string(b'a'),
          product([byte_tree(b'e'), byte_tree(u32(0)), byte_tree(b'k'), byte_tree(b'a')])),
         ('manifest', 'readTLVList', b'\0\x01' + string(b'v'), pair(byte_tree(b'\0\x01'), byte_tree(b'v'))),
         ('common', 'readSectionDirectory', bytes(32), product([byte_tree(bytes(n)) for n in [4,2,2,2,2,8,8,4]])),
         ('nodes', 'readNodeRecords', u32(1) + b'\0', byte_tree(b'\0')),
     ]:
         for count in [0, 1, 3]:
             yield module, export + '-' + str(count), export, [number(count), byte_tree(one*count+b'TAIL')], result(True, tree_list([value]*count), byte_tree(b'TAIL'))
         yield module, export+'-missing', export, [number(2), byte_tree(one)], result(False, number(1))
     # Noncanonical byte-list spines: compatibility comparisons, not new safety
     # guarantees. Keep counts bounded to avoid unbounded attacker lengths.
     for module, export in [('nodes','readNodeRecord'), ('manifest','readLengthPrefixedString'),
                            ('common','readSectionRecord')]:
         for i, raw in enumerate([TRUE, b'\1\1\0', tree_list([number(0)], TRUE)]):
             yield module, export+'-malformed-'+str(i), export, [raw], None
     for tag in [0, 1, 2, 3]:
         for payload in [b'', b'\0', b'\1\0\0\0\0', b'\2'+bytes(8), b'\3']:
             yield 'nodes', f'tag-{tag}-{payload.hex()}', 'nodePayloadHasTag?', [number(tag), byte_tree(payload)], TRUE if payload and payload[0] == tag else LEAF
 
     for i, records in enumerate([
         [], [b'\0'], [b'\0', b'\0'], [b'\0', b'\1'+u32(0)],
         [b'\0', b'\1'+u32(0), b'\2'+u32(0)+u32(1)],
         [b'\1'+u32(0)], [b'\0', b'\1'+u32(1)],
         [b'\0', b'\2'+u32(0)+u32(2)], [b''], [b'\3'],
         [b'\0x'], [b'\1'+bytes(3)], [b'\2'+bytes(7)],
         [b'\1'+u32(99), b'\3'],
     ]):
         encoded = tree_list([byte_tree(x) for x in records])
         invalid = any(not (x == b'\0' or (len(x) == 5 and x[0] == 1)
                           or (len(x) == 9 and x[0] == 2)) for x in records)
         bad_index = any(int.from_bytes(x[k:k+4], 'big') >= j
                         for j,x in enumerate(records) for k in range(1,len(x),4))
         code = 8 if invalid else 9 if bad_index else None
         rest = byte_tree(b'TAIL')
         expected = result(code is None, encoded if code is None else number(code), rest)
         yield 'nodes', f'validate-{i}', 'validateNodeRecords', [encoded,rest], expected
     for name in [b'a', b'b', b'missing']:
         exports = [product([byte_tree(n),byte_tree(u32(i)),byte_tree(b'k'),byte_tree(b'abi')])
                    for i,n in enumerate([b'a',b'b',b'a'])]
         value = exports[0] if name == b'a' else exports[1] if name == b'b' else None
         yield 'manifest', 'find-'+name.hex(), 'findExportByName', [tree_list(exports),byte_tree(name)], LEAF if value is None else b'\1'+value
         yield 'manifest', 'names-'+name.hex(), 'getExportNames', [tree_list(exports)], tree_list([byte_tree(n) for n in [b'a',b'b',b'a']])
     for i, raw in enumerate([LEAF, TRUE, b'\1\1\0']):
         for module, export, args in [
             ('manifest','findExportByName',[raw,byte_tree(b'a')]),
             ('manifest','getExportNames',[raw]),
             ('manifest','lookupMetadata',[raw,byte_tree(b'xx')]),
             ('common','lookupSectionRecord',[byte_tree(b'xxxx'),raw]),
             ('nodes','nodeRecordsHaveInvalidPayload?',[raw]),
             ('nodes','nodeChildIndicesInPrefix?',[raw,LEAF,number(0)]),
         ]:
             yield module, f'{export}-shape-{i}', export, args, None
 
 
 def main():
     p = argparse.ArgumentParser(description=__doc__)
     p.add_argument('tricu', type=Path)
     p.add_argument('--subject', type=Path, required=True, help='candidate directory')
     p.add_argument('--reference', type=Path, required=True, help='retained baseline directory')
     p.add_argument('--directory', type=Path, required=True)
     args = p.parse_args()
     args.directory.mkdir(parents=True)
     runner = Runner(args.tricu, args.directory)
     pins, identities = {}, {}
     for role, directory in [('subject',args.subject),('reference',args.reference)]:
         for module in ['manifest','common','nodes']:
             path = directory / ('arboricx_'+module+'.arboricx')
             pins[role,module] = runner.imported(path,role+'-'+module)
             identities[role+'-'+module] = hashlib.sha256(path.read_bytes()).hexdigest()
     report = dict(complete=False, artifacts=identities, runtime_sha256=hashlib.sha256(args.tricu.read_bytes()).hexdigest(),
                   harness_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), cases=[])
     try:
         for module,label,export,values,expected in cases():
             argv=[]
             for i,value in enumerate(values):
                 path=args.directory/(label+f'-{i}.tree');path.write_bytes(value)
                 argv+=['--arg-tree-file',path]
             before=runner.invoke(label+'-reference',pins['reference',module],export,argv)
             after=runner.invoke(label+'-subject',pins['subject',module],export,argv)
             if expected is not None:
                 expect(before,expected,label+' reference oracle')
                 expect(after,expected,label+' subject oracle')
             expect(after,before,label+' complete comparison')
             report['cases'].append(dict(label=label,oracle=expected is not None,result_sha256=hashlib.sha256(after).hexdigest()))
         # Same arity, wrong export must not satisfy the independently specified result.
         raw=byte_tree(string(b'x'));path=args.directory/'wrong.tree';path.write_bytes(raw)
         wrong=runner.invoke('wrong-subject',pins['subject','manifest'],'readRootEntry',['--arg-tree-file',path])
         assert wrong != result(True,byte_tree(b'x')), 'wrong export accepted'
         report.update(complete=True,wrong_subject_rejected=True)
     finally:
         (args.directory/'report.json').write_text(json.dumps(report,indent=2)+'\n')
     print(f'PASS: {len(report["cases"])} reader cases + wrong-subject control')
 
 
 if __name__ == '__main__':
     main()