interaction_inspection.py (7629B)
1 #!/usr/bin/env python3 2 """Read-only IO inspection CLI against independently assembled indexed bundles.""" 3 import argparse 4 import hashlib 5 import json 6 import os 7 from pathlib import Path 8 import struct 9 import subprocess 10 import tempfile 11 12 from stable_bundle import assembled, sections 13 14 ROOT = Path(__file__).resolve().parent.parent 15 16 17 def main(): 18 parser = argparse.ArgumentParser(description=__doc__) 19 parser.add_argument('tricu', type=Path) 20 exe = parser.parse_args().tricu.resolve() 21 manifest, _ = sections(ROOT / 'test/fixtures/id.arboricx') 22 at = 12 23 for _ in range(9): 24 size = struct.unpack_from('>I', manifest, at)[0] 25 at += 4 + size 26 assert struct.unpack_from('>I', manifest, at)[0] == 0 27 prefix = manifest[:at + 5] 28 29 def text(value): 30 data = value.encode() 31 return struct.pack('>I', len(data)) + data 32 33 nodes, indices = [], {} 34 35 def intern(*children): 36 if children not in indices: 37 indices[children] = len(nodes) 38 nodes.append(children) 39 return indices[children] 40 41 leaf = intern() 42 one = intern(leaf) 43 44 def number(n): 45 return intern(one if n % 2 else leaf, number(n // 2)) if n else leaf 46 47 def string(s): 48 result = leaf 49 for c in reversed(s): 50 result = intern(number(ord(c)), result) 51 return result 52 53 def action(tag, payload): 54 return intern(number(tag), payload) 55 56 def envelope(a, version=1): 57 return intern(string('tricuIO'), intern(number(version), a)) 58 59 def bundle(exports, extra=()): 60 records = [bytes([len(cs)]) + b''.join(struct.pack('>I', c) for c in cs) for cs in nodes + list(extra)] 61 m = prefix + struct.pack('>I', len(exports)) 62 for _, index in exports: 63 m += struct.pack('>I', index) + text('default') 64 m += struct.pack('>I', len(exports)) 65 for name, index in exports: 66 m += text(name) + struct.pack('>I', index) + text('term') + text('arboricx.abi.tree.v1') 67 m += struct.pack('>II', 0, 0) 68 return assembled(m, struct.pack('>Q', len(records)) + b''.join(struct.pack('>I', len(r)) + r for r in records)) 69 70 with tempfile.TemporaryDirectory(prefix='interaction-inspection-') as tmp: 71 root = Path(tmp) 72 artifact = root / 'input.arboricx' 73 victim = root / 'must-remain' 74 victim.write_text('untouched') 75 absent_store = root / 'must-not-create-store' 76 # If inspection accidentally executes effects, this deletes the victim. 77 dangerous = action(26, string(str(victim))) 78 k = action(10, string('MUST-NOT-PRINT')) # deliberately not a callable continuation 79 bind = action(1, intern(dangerous, k)) 80 main_root = envelope(bind) 81 artifact.write_bytes(bundle([('main', main_root), ('resume\n\x1b[2J', k), ('alias', k)])) 82 83 def snapshot(): 84 return {str(p.relative_to(root)): (p.read_bytes(), p.stat().st_mtime_ns) 85 for p in root.rglob('*') if p.is_file()} 86 87 def run(*args, ok=True): 88 before = snapshot() 89 reply = subprocess.run([str(exe), '--store', str(absent_store), 'arboricx', 'inspect', 90 '--file', str(artifact), *map(str, args), '+RTS', '-N1', '-M512m', '-RTS'], 91 cwd=root, capture_output=True, text=True, timeout=20) 92 assert (reply.returncode == 0) == ok, (reply.stdout, reply.stderr) 93 assert snapshot() == before, 'inspection mutated files' 94 assert not absent_store.exists(), 'inspection populated a store' 95 assert '\x1b' not in reply.stdout, 'terminal control injection' 96 return reply 97 98 output = run('--as', 'io').stdout 99 assert 'DeleteFile' in output and 'continuation (not applied)' in output 100 assert 'MUST-NOT-PRINT' not in output 101 assert '\\n\\ESC' in output 102 report = json.loads(run('--as', 'io', '--json').stdout) 103 assert report['schema'] == 'tricu.interaction-inspection.v1' 104 assert report['bundleSha256'] == hashlib.sha256(artifact.read_bytes()).hexdigest() 105 assert [o['path'] for o in report['occurrences']] == [ 106 'root', 'root.action', 'root.action.action', 'root.action.action.path', 'root.action.resume'] 107 assert report['occurrences'][-1]['aliases'] == ['resume\n\x1b[2J', 'alias'] 108 assert report == json.loads(run('--as', 'io', '--json').stdout) 109 narrow = json.loads(run('--as', 'io', '--json', '--max-nodes', 2).stdout) 110 assert len(narrow['occurrences']) == 2 and narrow['occurrenceLimitReached'] 111 shallow = json.loads(run('--as', 'io', '--json', '--max-depth', 0).stdout) 112 assert shallow['occurrences'][0]['status'] == 'truncated' 113 run(ok=False) # protocol selection is required 114 run('--as', 'contract', ok=False) 115 run('--as', 'io', '--name', 'absent', ok=False) 116 run('--as', 'io', '--max-bytes', 1, ok=False) 117 run('--as', 'io', '--max-nodes', 0, ok=False) 118 run('--as', 'io', '--max-depth', 257, ok=False) 119 run('--as', 'io', '--preview-items', 0, ok=False) 120 121 for tree, expected in [(leaf, 'invalid'), (envelope(leaf, 2), 'unknown'), 122 (envelope(action(999, leaf)), 'unknown'), 123 (envelope(action(1, leaf)), 'invalid')]: 124 artifact.write_bytes(bundle([('main', tree)])) 125 failure = json.loads(run('--as', 'io', '--json', ok=False).stdout) 126 assert expected in [o['status'] for o in failure['occurrences']] 127 128 # Huge logical value is reachable, but remains an opaque DAG root. 129 huge = leaf 130 for _ in range(200): 131 huge = intern(huge, huge) 132 artifact.write_bytes(bundle([('main', envelope(action(0, huge)))])) 133 large = json.loads(run('--as', 'io', '--json').stdout) 134 assert len(large['occurrences']) == 3 135 assert large['storedNodes'] < 500 136 # Validate all stored nodes, not only those demanded by the outline. 137 artifact.write_bytes(bundle([('main', main_root)], extra=[(len(nodes),)])) 138 assert 'references child' in run('--as', 'io', ok=False).stderr 139 artifact.write_bytes(b'not a bundle') 140 assert 'invalid magic' in run('--as', 'io', ok=False).stderr 141 # Nonregular inputs must not block on FIFOs. 142 artifact.unlink() 143 os.mkfifo(artifact) 144 reply = subprocess.run([str(exe), 'arboricx', 'inspect', '--file', str(artifact), '--as', 'io', 145 '+RTS', '-N1', '-RTS'], capture_output=True, text=True, timeout=10) 146 assert reply.returncode != 0 and 'regular file' in reply.stderr 147 artifact.unlink() 148 # Native stable compiler smoke is separate from independent wire tests. 149 subprocess.run([str(exe), 'arboricx', 'compile', '--file', 150 str(ROOT / 'demos/interaction-inspection.tri'), '--store', str(root / 'compile-store'), 151 '--output', str(artifact), '+RTS', '-N1', '-M512m', '-RTS'], 152 cwd=root, capture_output=True, text=True, check=True, timeout=120) 153 compiled = json.loads(run('--as', 'io', '--json').stdout) 154 assert compiled['occurrences'][-1]['aliases'] == ['afterRead'] 155 assert compiled['occurrences'][2]['summary'] == 'ReadFile' 156 assert compiled['occurrences'][3]['summary'] == '"settings.txt"' 157 print('PASS interaction inspection: independent bundle layouts, JSON/outline, no effects/store writes, limits and malformed inputs') 158 159 160 if __name__ == '__main__': 161 main()