io_artifact_service.py (11944B)
1 #!/usr/bin/env python3 2 """Actual v2 artifact handlers: exact bytes, integrity, lexical routing and limits.""" 3 import argparse 4 import hashlib 5 import json 6 from pathlib import Path 7 import subprocess 8 from source_order import Runner, field 9 from http_headers import number, result, tree_list, pair 10 from io_objects import text, problem 11 12 ROOT = Path(__file__).resolve().parent.parent 13 14 15 def octets(data): 16 return tree_list([number(v) for v in data]) 17 18 19 def main(): 20 p = argparse.ArgumentParser(description=__doc__) 21 p.add_argument('tricu', type=Path) 22 p.add_argument('--directory', type=Path, required=True) 23 a = p.parse_args() 24 root = a.directory.resolve(); root.mkdir(parents=True, exist_ok=False) 25 runner = Runner(a.tricu, root) 26 bundle = root/'service.arboricx' 27 runner.run('compile', ['compile', '--file', ROOT/'test/io/artifact-service.tri', 28 '--store', runner.store, '--output', bundle, '--compile-timeout', '600'], seconds=660) 29 pin = runner.imported(bundle, 'import') 30 checks = [] 31 32 def arguments(label, values): 33 args = ['--io'] 34 for i, value in enumerate(values): 35 path = root/f'{label}-{i}.tree'; path.write_bytes(value) 36 args += ['--arg-tree-file', path] 37 return args 38 39 def check(label, name, values, expected): 40 actual = runner.invoke(label, pin, name, arguments(label, values)) 41 assert actual == expected, label 42 checks.append(label) 43 44 def path(directory, data): 45 digest = hashlib.sha256(data).hexdigest() 46 return directory/'objects'/digest[:3]/(digest+'.arboricx') 47 48 left, right = root/'left', root/'right' 49 for directory in [left, right]: 50 for suffix in ['new', 'existing']: 51 check(directory.name+'-'+suffix, 'prepare', [text(str(directory))], result(True, b'\0')) 52 limit = number(8192) 53 for index, data in enumerate([b'', b'\0\xff\x80\r\n', bytes(range(256))*16+b'end']): 54 digest = text(hashlib.sha256(data).hexdigest()) 55 for suffix in ['new', 'existing']: 56 check(f'publish-{index}-{suffix}', 'publish', [text(str(left)), limit, octets(data)], result(True, digest)) 57 assert path(left, data).read_bytes() == data 58 check(f'load-{index}', 'load', [text(str(left)), limit, digest], result(True, octets(data))) 59 corrupt = b'X'+data[1:] if data else b'X' 60 path(left, data).write_bytes(corrupt) 61 check(f'corrupt-load-{index}', 'load', [text(str(left)), limit, digest], 62 problem('integrity-failed', 'artifacts/load', 'digest mismatch', 'unknown')) 63 check(f'corrupt-winner-{index}', 'publish', [text(str(left)), limit, octets(data)], 64 problem('invalid-data', 'storage/verify', 'different bytes', 'unknown')) 65 assert path(left, data).read_bytes() == corrupt 66 path(left, data).write_bytes(data+b'extra') 67 check(f'trailing-winner-{index}', 'publish', [text(str(left)), limit, octets(data)], 68 problem('invalid-data', 'storage/verify', 'different bytes', 'unknown')) 69 path(left, data).write_bytes(data) 70 data = b'lexical' 71 check('nearest-route', 'routed', [text(str(left)), text(str(right)), limit, octets(data)], 72 result(True, text(hashlib.sha256(data).hexdigest()))) 73 assert path(right, data).read_bytes() == data and not path(left, data).exists() 74 data = b'captured' 75 check('captured-route', 'captured', [text(str(left)), text(str(right)), limit, octets(data)], 76 result(True, text(hashlib.sha256(data).hexdigest()))) 77 assert path(left, data).read_bytes() == data and not path(right, data).exists() 78 check('forward-unknown', 'forwarded', [text(str(left))], tree_list([text('other'), number(42)])) 79 absent = root/'untouched' 80 diagnostic = tree_list([text('invalid-digest'), text('expected 64 lowercase hexadecimal hash characters')]) 81 for index, invalid in enumerate(['../outside', 'A'*64, '0'*63, '0'*65, '\0'*64]): 82 for name in ['load', 'rawLoad']: 83 check(f'{name}-invalid-{index}', name, [text(str(absent)), limit, text(invalid)], result(False, diagnostic)) 84 check('budget-before-paths', 'publish', [text(str(absent)), number(0), octets(b'x')], 85 problem('invalid-bound', 'storage/hash', 'bytes exceed limit', 'none')) 86 assert not absent.exists() 87 data = b'captured' 88 check('read-budget', 'load', [text(str(left)), number(1), text(hashlib.sha256(data).hexdigest())], 89 problem('invalid-bound', 'storage/read', 'file exceeds byte limit', 'unknown')) 90 for label, malformed in [('wide', tree_list([number(256)])), ('improper', b'\1\0')]: 91 args = arguments('raw-'+label, [text(str(absent)), limit, malformed]) 92 command = [str(a.tricu.resolve()), 'arboricx', 'run', '--store', str(runner.store), 93 '--module', pin, '--name', 'rawPublish', *args] 94 reply = subprocess.run(command, capture_output=True, text=True, timeout=30) 95 assert reply.returncode != 0 and 'IO task failed' in reply.stderr, reply 96 actual = runner.invoke('view-'+label, pin, 'rejectedView', args[1:]) 97 wanted = pair(number(103), tree_list([text('invalid-request'), text('artifacts'), text('publish'), 98 text('expected octet' if label == 'wide' else 'not a list'), tree_list([])])) 99 assert actual == wanted, label 100 checks.append('raw-'+label) 101 assert not absent.exists() 102 blocked = root/'blocked'; blocked.write_bytes(b'keep') 103 actual = runner.invoke('blocked-root', pin, 'prepare', arguments('blocked', [text(str(blocked))])) 104 assert field(actual, 1, 0, 0) == text('not-directory') 105 assert blocked.read_bytes() == b'keep' 106 checks.append('blocked-root') 107 data = b'concurrent winner' 108 args = arguments('race', [text(str(right)), limit, octets(data)]) 109 command = [str(a.tricu.resolve()), 'arboricx', 'run', '--store', str(runner.store), 110 '--module', pin, '--name', 'publish', '--format', 'ternary', *args, 111 '+RTS', '-N1', '-M2g', '-RTS'] 112 processes = [subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) for _ in range(2)] 113 try: 114 for process in processes: 115 stdout, stderr = process.communicate(timeout=60) 116 assert process.returncode == 0, stderr 117 assert bytes(int(c) for c in stdout.strip().decode()) == result(True, text(hashlib.sha256(data).hexdigest())) 118 finally: 119 for process in processes: 120 if process.poll() is None: 121 process.kill(); process.wait() 122 assert path(right, data).read_bytes() == data 123 checks.append('concurrent-winner') 124 data = b'must not write with old grants' 125 args = arguments('grants', [text(str(right)), limit, octets(data)]) 126 reply = subprocess.run([str(a.tricu.resolve()), 'arboricx', 'run', '--store', str(runner.store), 127 '--module', pin, '--name', 'publish', *args, '--allow-write', str(right)], 128 capture_output=True, text=True, timeout=30) 129 assert reply.returncode != 0 and 'v1 permission grants are not accepted' in reply.stderr 130 assert not path(right, data).exists() 131 checks.append('v1-grants-rejected-before-effects') 132 def response(status, body, mime='text/plain; charset=utf-8'): 133 phrase = {200:'OK', 201:'Created', 400:'Bad Request', 404:'Not Found', 500:'Internal Server Error'}[status] 134 return octets((f'HTTP/1.1 {status} {phrase}\r\nContent-Type: {mime}\r\n' 135 f'Content-Length: {len(body)}\r\nConnection: close\r\n\r\n').encode()+body) 136 def route(label, method, target, body, expected): 137 check(label, 'route', [text(str(left)), limit, text(method), text(target), octets(body)], expected) 138 route('health', 'GET', '/_arboricx/health', b'', response(200, b'OK\n')) 139 data = b'route binary\0\xff' 140 digest = hashlib.sha256(data).hexdigest() 141 route('route-publish', 'POST', '/_arboricx/bundle', data, response(201, digest.encode())) 142 route('route-get', 'GET', '/_arboricx/bundle/hash/'+digest, b'', 143 response(200, data, 'application/vnd.arboricx.bundle')) 144 path(left, data).write_bytes(b'corrupt') 145 route('route-corrupt-get', 'GET', '/_arboricx/bundle/hash/'+digest, b'', response(404, b'Artifact unavailable\n')) 146 route('route-corrupt-put', 'POST', '/_arboricx/bundle', data, response(400, b'Upload failed\n')) 147 assert path(left, data).read_bytes() == b'corrupt' 148 data = b'published despite logger failure' 149 check('failed-log-after-publication', 'routeFailedLog', [text(str(left)), limit, octets(data)], 150 response(500, b'Internal Server Error\n')) 151 assert path(left, data).read_bytes() == data 152 route('after-failed-request', 'GET', '/_arboricx/health', b'', response(200, b'OK\n')) 153 package = b'captured package bytes' 154 (left/'packages'/'example').mkdir(parents=True) 155 (left/'packages'/'example'/'1').write_bytes(package) 156 (right/'packages'/'example').mkdir(parents=True) 157 (right/'packages'/'example'/'1').write_bytes(b'wrong root') 158 target = '/_arboricx/package/example/1' 159 check('captured-package', 'capturedPackage', [text(str(left)), text(str(right)), limit, text(target)], 160 response(200, package, 'application/vnd.arboricx.package-release')) 161 route('package-route', 'GET', target, b'', response(200, package, 'application/vnd.arboricx.package-release')) 162 route('package-invalid', 'GET', '/_arboricx/package/../escape', b'', response(400, b'Invalid package name/version')) 163 check('stdout-status', 'routeStdout', [text(str(left)), limit], response(200, b'OK\n')) 164 assert 'registry status=200\n' in (root/'stdout-status.log').read_text() 165 data = b'published before cancellation' 166 check('request-cancellation', 'routeCancelledLog', [text(str(left)), limit, octets(data)], 167 tree_list([text('cancelled'), text('logger stopped'), tree_list([])])) 168 assert path(left, data).read_bytes() == data 169 for size in [0, 1, 4095, 4096, 4097, 65536, 1048576]: 170 data = (bytes(range(256))*((size+255)//256))[:size] 171 source = root/f'stream-source-{size}'; source.write_bytes(data) 172 digest = hashlib.sha256(data).hexdigest() 173 for mode in ['new', 'existing']: 174 check(f'stream-{size}-{mode}', 'streamPublish', [text(str(right)), text(str(source)), number(size)], 175 result(True, text(digest))) 176 assert path(right, data).read_bytes() == data 177 assert list((right/'tmp').iterdir()) == [] 178 if size >= 65536: 179 record = runner.records[-1] 180 assert record['peak_kib'] < 128*1024, 'stream RSS retention regression' 181 assert record['rts_process']['maximum_live_bytes'] < 16*1024*1024, 'stream heap retention regression' 182 data = b'stream corruption test' 183 source = root/'stream-negative'; source.write_bytes(data) 184 check('stream-short', 'streamPublish', [text(str(right)), text(str(source)), number(len(data)+1)], 185 result(False, text('Incomplete artifact stream'))) 186 check('stream-excess', 'streamPublish', [text(str(right)), text(str(source)), number(len(data)-1)], 187 result(False, text('Artifact exceeds declared length'))) 188 assert not path(right, data).exists() 189 path(right, data).parent.mkdir(exist_ok=True) 190 path(right, data).write_bytes(b'X'+data[1:]) 191 check('stream-corrupt-winner', 'streamPublish', [text(str(right)), text(str(source)), number(len(data))], 192 result(False, text('Existing artifact digest mismatch'))) 193 assert path(right, data).read_bytes() == b'X'+data[1:] 194 assert list((right/'tmp').iterdir()) == [] 195 assert not list(left.rglob('.tricu-*')) and not list(right.rglob('.tricu-*')) 196 (root/'report.json').write_text(json.dumps(dict(checks=checks, records=runner.records), indent=2)+'\n') 197 print(f'PASS {len(checks)} v2 artifact service controls; deployment unchanged') 198 199 200 if __name__ == '__main__': 201 main()