streaming_protocol.py (9179B)
1 #!/usr/bin/env python3 2 """Real-socket negative controls for the opt-in streaming upload composition. 3 4 Uses fixed port 8080; requires a Nix-built runner. No production fault flags. 5 Checks exact binary objects/digests, staged visibility, EOF cleanup, authority 6 and commit failures, framing rejection, and version-specific admission limits. 7 """ 8 import argparse 9 import hashlib 10 from pathlib import Path 11 import socket 12 import subprocess 13 import tempfile 14 import time 15 16 17 def main(): 18 parser = argparse.ArgumentParser(description=__doc__) 19 parser.add_argument('runner', type=Path) 20 parser.add_argument('--io-version', type=int, choices=[1, 2], default=1) 21 args = parser.parse_args() 22 runner = str(args.runner.resolve()) 23 passed = [] 24 def connect(): 25 return socket.create_connection(('127.0.0.1', 8080), timeout=30) 26 def headers(length): 27 return (f'POST /_arboricx/bundle HTTP/1.1\r\nHost: local\r\nContent-Length: {length}\r\n\r\n').encode() 28 def response(client, expected): 29 parts = [] 30 while True: 31 part = client.recv(65536) 32 if not part: break 33 parts.append(part) 34 header, body = b''.join(parts).split(b'\r\n\r\n', 1) 35 assert header.startswith(f'HTTP/1.1 {expected} '.encode()), header 36 fields = dict(line.split(b': ', 1) for line in header.split(b'\r\n')[1:]) 37 assert fields[b'Connection'] == b'close' 38 assert int(fields[b'Content-Length']) == len(body) 39 return body 40 def request(wire, expected): 41 with connect() as client: 42 client.sendall(wire); client.shutdown(socket.SHUT_WR) 43 return response(client, expected) 44 def wait_for(predicate, message): 45 for _ in range(300): 46 if predicate(): return 47 time.sleep(.01) 48 raise AssertionError(message) 49 50 with tempfile.TemporaryDirectory(prefix='tricu-streaming-protocol-') as directory: 51 root = Path(directory) 52 with (root/'server.log').open('w+') as log: 53 proc = subprocess.Popen([runner], cwd=root, stdout=log, stderr=log) 54 try: 55 for _ in range(100): 56 if proc.poll() is not None: 57 log.seek(0); raise AssertionError(log.read()) 58 try: 59 assert request(b'GET /_arboricx/health HTTP/1.1\r\nHost: local\r\n\r\n', 200) == b'OK\n' 60 break 61 except OSError: 62 time.sleep(.1) 63 else: raise AssertionError('startup timeout') 64 staging = root/'store/tmp' 65 def object_path(body): 66 digest = hashlib.sha256(body).hexdigest() 67 return digest, root/'store/objects'/digest[:3]/(digest+'.arboricx') 68 def empty_stage(): 69 assert list(staging.iterdir()) == [] 70 for size in (0, 1, 4095, 4096, 4097, 8193): 71 body = (bytes(range(256))*((size+255)//256))[:size] 72 digest, path = object_path(body) 73 with connect() as client: 74 client.sendall(headers(size)) 75 # Deliberately fragment bodies across odd-sized writes. 76 for offset in range(0, size, 997): 77 client.sendall(body[offset:offset+997]); time.sleep(.001) 78 client.shutdown(socket.SHUT_WR) 79 assert response(client, 201) == digest.encode() 80 assert path.read_bytes() == body 81 empty_stage(); passed.append(f'binary-{size}') 82 83 # No object is visible while only a prefix has been accepted. 84 prefix = b'not a committed artifact\x00\xff'*200 85 digest, path = object_path(prefix) 86 with connect() as client: 87 client.sendall(headers(len(prefix)+100)+prefix) 88 wait_for(lambda: len(list(staging.iterdir())) == 1, 'no staging file') 89 assert not path.exists() 90 client.shutdown(socket.SHUT_WR) 91 assert response(client, 400) == b'Upload failed\n' 92 assert not path.exists(); empty_stage(); passed.append('early-eof-invisible') 93 94 for name, wire, status in [ 95 ('oversized', headers(1048577), 413), 96 ('duplicate-length', b'POST /_arboricx/bundle HTTP/1.1\r\nHost: local\r\nContent-Length: 0\r\nContent-Length: 0\r\n\r\n', 400), 97 ('transfer-encoding', b'POST /_arboricx/bundle HTTP/1.1\r\nHost: local\r\nTransfer-Encoding: chunked\r\n\r\n', 400)]: 98 request(wire, status); empty_stage(); passed.append(name) 99 100 # V1 grants reject an escaping staging symlink; v2 follows OS authority. 101 outside = root/'outside'; outside.mkdir() 102 staging.rmdir(); staging.symlink_to(outside, target_is_directory=True) 103 try: 104 if args.io_version == 1: 105 assert request(headers(0), 400) == b'Upload failed\n' 106 else: 107 assert request(headers(0), 201) == hashlib.sha256(b'').hexdigest().encode() 108 assert list(outside.iterdir()) == [] 109 finally: 110 staging.unlink(); staging.mkdir() 111 passed.append('open-denied' if args.io_version == 1 else 'os-authority-symlink') 112 113 # An unusable staging path fails under either authority model. 114 staging.rmdir(); staging.write_bytes(b'unchanged') 115 try: 116 assert request(headers(0), 400) == b'Upload failed\n' 117 assert staging.read_bytes() == b'unchanged' 118 finally: 119 staging.unlink(); staging.mkdir() 120 passed.append('open-failed') 121 122 body = b'commit failure must leave the original target intact' 123 digest, path = object_path(body) 124 path.mkdir(parents=True); (path/'marker').write_bytes(b'unchanged') 125 assert request(headers(len(body))+body, 400) == b'Upload failed\n' 126 assert (path/'marker').read_bytes() == b'unchanged' 127 empty_stage(); passed.append('commit-failed') 128 129 clients = [] 130 try: 131 for _ in range(16): 132 client = connect(); clients.append(client) 133 client.sendall(headers(2)+b'A') 134 wait_for(lambda: len(list(staging.iterdir())) == 16, 'sessions did not open') 135 if args.io_version == 1: 136 assert request(headers(0), 400) == b'Upload failed\n' 137 assert len(list(staging.iterdir())) == 16 138 assert request(b'GET /_arboricx/health HTTP/1.1\r\nHost: local\r\n\r\n', 200) == b'OK\n' 139 else: 140 # HostIO bounds connections, not a separate native upload 141 # table. The seventeenth connection waits in the backlog 142 # until the oldest owned child is joined. 143 with connect() as queued: 144 queued.sendall(headers(0)); queued.shutdown(socket.SHUT_WR) 145 queued.settimeout(.2) 146 try: 147 unexpected = queued.recv(1) 148 except socket.timeout: 149 pass 150 else: 151 raise AssertionError(('connection admitted past window', unexpected)) 152 assert len(list(staging.iterdir())) == 16 153 clients[0].shutdown(socket.SHUT_WR) 154 assert response(clients[0], 400) == b'Upload failed\n' 155 queued.settimeout(30) 156 assert response(queued, 201) == hashlib.sha256(b'').hexdigest().encode() 157 wait_for(lambda: len(list(staging.iterdir())) == 15, 'oldest connection not released') 158 for client in (clients if args.io_version == 1 else clients[1:]): 159 client.shutdown(socket.SHUT_WR) 160 assert response(client, 400) == b'Upload failed\n' 161 finally: 162 for client in clients: client.close() 163 wait_for(lambda: not list(staging.iterdir()), 'abandoned sessions not removed') 164 assert proc.poll() is None 165 assert request(b'GET /_arboricx/health HTTP/1.1\r\nHost: local\r\n\r\n', 200) == b'OK\n' 166 passed.append('session-limit-and-abandonment' if args.io_version == 1 else 'fifo-admission-and-abandonment') 167 finally: 168 proc.terminate() 169 try: proc.wait(timeout=5) 170 except subprocess.TimeoutExpired: 171 proc.kill(); proc.wait() 172 print(f'PASS IO v{args.io_version}:', ', '.join(passed)) 173 174 175 if __name__ == '__main__': 176 main()