arboricx_deployment.py (12711B)
1 #!/usr/bin/env python3 2 """Real registry runners: socket/security controls and opt-in size/lifetime probes. 3 4 Fixed port 8080 must be free. --workload exercises 4 KiB, 64 KiB and 1 MiB 5 publication/readback plus mixed concurrent requests. --handler-policy also checks 6 intentional redaction; --fault-runner injects ordinary logging failure. Reports 7 are single-run observations, not statistically controlled timing comparisons. 8 """ 9 import argparse 10 from concurrent.futures import ThreadPoolExecutor 11 import json 12 import threading 13 import hashlib 14 from pathlib import Path 15 import socket 16 import subprocess 17 import tempfile 18 import time 19 import urllib.error 20 import urllib.request 21 22 23 def main(): 24 parser = argparse.ArgumentParser(description=__doc__) 25 parser.add_argument('runner', type=Path) 26 parser.add_argument('--handler-policy', action='store_true') 27 parser.add_argument('--io-version', type=int, choices=[1,2], default=1, 28 help='explicit authority expectation; v2 has OS process authority') 29 parser.add_argument('--workload', action='store_true') 30 parser.add_argument('--fault-runner', type=Path) 31 parser.add_argument('--report', type=Path) 32 args = parser.parse_args() 33 runner = str(args.runner.resolve()) 34 measurements = [] 35 peak_kib = [0] 36 stop = threading.Event() 37 def watch(proc): 38 while not stop.wait(.02): 39 try: 40 for line in Path(f'/proc/{proc.pid}/status').read_text().splitlines(): 41 if line.startswith(('VmRSS:', 'VmHWM:')): 42 peak_kib[0] = max(peak_kib[0], int(line.split()[1])) 43 except (OSError, ValueError): 44 pass 45 opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) 46 47 def request(path, data=None, status=200): 48 req = urllib.request.Request('http://127.0.0.1:8080' + path, data=data) 49 try: 50 response = opener.open(req, timeout=180) 51 except urllib.error.HTTPError as error: 52 response = error 53 with response: 54 assert response.headers.get('Connection', '').lower() == 'close' 55 assert response.status == status, (path, response.status) 56 return response.read() 57 58 with tempfile.TemporaryDirectory(prefix='arboricx-deployment-') as temporary: 59 root = Path(temporary) 60 # Startup failures must exit nonzero rather than return an ignored Result. 61 (root / 'store').write_text('not a directory') 62 failed = subprocess.run([runner], cwd=root, capture_output=True, timeout=30) 63 assert failed.returncode != 0, failed 64 (root / 'store').unlink() 65 with socket.socket() as occupied: 66 occupied.bind(('0.0.0.0', 8080)) 67 occupied.listen() 68 failed = subprocess.run([runner], cwd=root, capture_output=True, timeout=30) 69 assert failed.returncode != 0, failed 70 71 payload = b'opaque registry bytes\x00\xff\n' 72 digest = hashlib.sha256(payload).hexdigest() 73 route = '/_arboricx/bundle/hash/' + digest 74 # Second start checks persistence without a compiler cache in the cwd. 75 for iteration in range(2): 76 with (root / 'server.log').open('w+') as log: 77 proc = subprocess.Popen([runner], cwd=root, stdout=log, stderr=log) 78 stop.clear() 79 monitor = threading.Thread(target=watch, args=(proc,), daemon=True); monitor.start() 80 try: 81 for _ in range(100): 82 if proc.poll() is not None: 83 log.seek(0) 84 raise AssertionError(log.read()) 85 try: 86 assert request('/_arboricx/health') == b'OK\n' 87 break 88 except (OSError, urllib.error.URLError): 89 time.sleep(.1) 90 else: 91 raise AssertionError('server startup timeout') 92 assert b'Arboricx Server' in request('/') 93 if iteration == 0: 94 reply = request('/_arboricx/bundle', payload, status=201) 95 assert reply.strip() == digest.encode(), reply 96 assert request(route) == payload 97 request('/_arboricx/bundle/hash/not-a-hash', status=400) 98 missing = request('/_arboricx/bundle/hash/' + '0' * 64, status=404) 99 if args.handler_policy: 100 assert missing == b'Artifact unavailable\n' 101 # Wrong-digest symlinks never expose arbitrary bytes. V1 also 102 # confines paths; v2 instead authenticates bytes with OS authority. 103 outside = root / 'outside'; outside.write_bytes(b'PRIVATE') 104 denied_hash = 'b'*64 105 denied = root / 'store/objects/bbb' / (denied_hash+'.arboricx') 106 denied.parent.mkdir(parents=True, exist_ok=True) 107 if not denied.exists(): 108 denied.symlink_to(outside) 109 denied_reply = request('/_arboricx/bundle/hash/'+denied_hash, status=404) 110 assert b'PRIVATE' not in denied_reply 111 if args.handler_policy: 112 assert denied_reply == b'Artifact unavailable\n' 113 # A v1 grant rejects publication through an escaping shard. 114 # HostIO deliberately permits it under OS process authority. 115 outside_dir = root / 'outside-dir'; outside_dir.mkdir(exist_ok=True) 116 rejected_body = b'reject publication outside grant' 117 rejected_hash = hashlib.sha256(rejected_body).hexdigest() 118 shard = root / 'store/objects' / rejected_hash[:3] 119 if not shard.exists(): 120 shard.symlink_to(outside_dir, target_is_directory=True) 121 if args.io_version == 1: 122 upload_error = request('/_arboricx/bundle', rejected_body, status=400) 123 assert list(outside_dir.iterdir()) == [] 124 if args.handler_policy: 125 assert upload_error == b'Upload failed\n' 126 else: 127 # Explicitly retire the v1 confinement expectation, not 128 # digest authentication or raw target validation controls. 129 assert request('/_arboricx/bundle', rejected_body, status=201) == rejected_hash.encode() 130 assert (outside_dir/(rejected_hash+'.arboricx')).read_bytes() == rejected_body 131 assert request('/_arboricx/bundle/hash/'+rejected_hash) == rejected_body 132 request('/_arboricx/package/../bad', status=400) 133 package = root / 'store/packages/example' 134 package.mkdir(parents=True, exist_ok=True) 135 record = b'arboricx.package-release.v1\nexample\t1.0\t' + b'a' * 64 + b'\t' + digest.encode() + b'\n' 136 (package / '1.0').write_bytes(record) 137 assert request('/_arboricx/package/example/1.0') == record 138 request('/_arboricx/package/example/missing', status=404) 139 if (args.workload or args.handler_policy) and iteration == 0: 140 # Same live listener, mixed concurrent requests. No 141 # shared log-state collector is installed by the app. 142 tasks = [(route, 200, payload), ('/_arboricx/health', 200, b'OK\n'), 143 ('/_arboricx/bundle/hash/bad', 400, b'expected 64 lowercase hexadecimal hash characters')]*4 144 with ThreadPoolExecutor(max_workers=4) as pool: 145 replies = list(pool.map(lambda task: request(task[0], status=task[1]), tasks)) 146 assert replies == [task[2] for task in tasks] 147 # Header/body policy is still the existing HTTP layer. 148 def raw_request(raw, wanted): 149 with socket.create_connection(('127.0.0.1', 8080), timeout=30) as client: 150 client.sendall(raw) 151 client.shutdown(socket.SHUT_WR) 152 chunks = [] 153 while True: 154 part = client.recv(65536) 155 if not part: break 156 chunks.append(part) 157 response = b''.join(chunks) 158 assert response.startswith(f'HTTP/1.1 {wanted} '.encode()), response[:200] 159 headers, body = response.split(b'\r\n\r\n', 1) 160 length = next(int(v.split(b':',1)[1]) for v in headers.split(b'\r\n') if v.lower().startswith(b'content-length:')) 161 assert len(body) == length 162 raw_request(b'NOT HTTP\r\n\r\n', 400) 163 raw_request(b'POST /_arboricx/bundle HTTP/1.1\r\nHost: local\r\nContent-Length: 1048577\r\n\r\n', 413) 164 for size in (4096, 65536, 1048576): 165 large = (bytes(range(256))*(size//256)) 166 hash = hashlib.sha256(large).hexdigest() 167 begin = time.monotonic() 168 assert request('/_arboricx/bundle', large, 201) == hash.encode() 169 uploaded = time.monotonic()-begin 170 begin = time.monotonic() 171 assert request('/_arboricx/bundle/hash/'+hash) == large 172 measurements.append(dict(bytes=size, upload_seconds=uploaded, 173 download_seconds=time.monotonic()-begin, observed_peak_kib=peak_kib[0])) 174 175 finally: 176 proc.terminate() 177 try: 178 proc.wait(timeout=5) 179 except subprocess.TimeoutExpired: 180 proc.kill() 181 proc.wait() 182 stop.set(); monitor.join(timeout=2) 183 assert {p.name for p in root.iterdir()} == {'store', 'server.log', 'outside', 'outside-dir'} 184 if args.fault_runner: 185 with (root / 'server.log').open('w+') as log: 186 proc = subprocess.Popen([str(args.fault_runner.resolve())], cwd=root, stdout=log, stderr=log) 187 try: 188 for _ in range(100): 189 if proc.poll() is not None: 190 raise AssertionError('fault runner exited') 191 try: 192 assert request('/_arboricx/health', status=500) == b'Internal Server Error\n' 193 break 194 except (OSError, urllib.error.URLError): 195 time.sleep(.1) 196 else: 197 raise AssertionError('fault runner startup timeout') 198 # Logger failure is not rollback: publication precedes log. 199 body = b'published before failed logging' 200 digest = hashlib.sha256(body).hexdigest() 201 assert request('/_arboricx/bundle', body, 500) == b'Internal Server Error\n' 202 assert (root/'store/objects'/digest[:3]/(digest+'.arboricx')).read_bytes() == body 203 assert request('/_arboricx/health', status=500) == b'Internal Server Error\n' 204 finally: 205 proc.terminate() 206 try: proc.wait(timeout=5) 207 except subprocess.TimeoutExpired: 208 proc.kill(); proc.wait() 209 if args.report: 210 args.report.write_text(json.dumps(dict(measurements=measurements, peak_kib=peak_kib[0], runner=runner, io_version=args.io_version, 211 handler_policy=args.handler_policy, 212 fault_runner=str(args.fault_runner.resolve()) if args.fault_runner else None, 213 limits=['single samples, not throughput estimates', 'Linux observed RSS/high-water mark, not allocation volume', 214 ('bounded FIFO window of 16 connections; fixed port 8080' if args.io_version == 2 else 'existing fork-per-connection scheduling; fixed port 8080'), 'termination is not a graceful-shutdown test']), indent=2)+'\n') 215 print('PASS: deployment startup failures, health, binary upload/download, path controls, packages, restart') 216 217 218 if __name__ == '__main__': 219 main()