#!/usr/bin/env python3 """Bounded loopback HTTP lab. Synthetic bytes; not a production downloader. Uses Python's standard library. Binds 127.0.0.1 on an ephemeral port, makes eight sequential GET requests and shuts down. Writes results.json in CWD. Only open-ended single byte ranges and identity-coded 16-byte bodies are implemented. Fault routes intentionally violate the declared contract. """ import hashlib import http.client import json import re import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path A = b"AAAAAA1111111111" B = b"BBBBBB2222222222" ETAG_A, ETAG_B = '"version-a"', '"version-b"' def strong_tag(value): # Deliberately narrow ASCII subset for these fixture tags. return isinstance(value, str) and re.fullmatch(r'"[A-Za-z0-9-]+"', value) is not None class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): routes = {'/same', '/changed', '/ignore', '/wrong-start', '/bad-length', '/denied'} if self.path not in routes: self.reply(404, b'') return if self.path == '/denied': self.reply(403, b'denied') return body, tag = (B, ETAG_B) if self.path == '/changed' else (A, ETAG_A) match = re.fullmatch(r'bytes=(\d+)-', self.headers.get('Range', '')) conditional = self.headers.get('If-Range') # If-Range false means ignore Range, before checking satisfiability. if self.path == '/ignore' or not match or (conditional is not None and conditional != tag): self.reply(200, body, ETag=tag) return start = int(match[1]) if start >= len(body): self.reply(416, b'', ETag=tag, **{'Content-Range': f'bytes */{len(body)}'}) return if self.path == '/wrong-start': start -= 1 content = body[start:] if self.path == '/bad-length': content = content[:-1] self.reply(206, content, ETag=tag, **{'Content-Range': f'bytes {start}-{len(body)-1}/{len(body)}'}) def reply(self, status, body, **headers): self.send_response(status) self.send_header('Content-Type', 'application/octet-stream') self.send_header('Content-Length', str(len(body))) self.send_header('Accept-Ranges', 'bytes') for key, value in headers.items(): self.send_header(key, value) self.end_headers() self.wfile.write(body) def consume(prefix, saved_tag, saved_total, status, headers, body): """Pure response policy. Real implementations must stream and persist safely.""" if status not in (200, 206): return 'reject-status', None if headers.get('content-encoding', 'identity') != 'identity': return 'reject-encoding', None if headers.get('content-length') != str(len(body)): return 'reject-length', None if status == 200: return 'replace', body match = re.fullmatch(r'bytes (\d+)-(\d+)/(\d+)', headers.get('content-range', '')) if not match: return 'reject-range', None start, end, total = map(int, match.groups()) # Contract: this fixture accepts one suffix reaching the saved total. if start != len(prefix) or end != total - 1 or total != saved_total or not 0 <= start <= end: return 'reject-range', None if len(body) != end - start + 1: return 'reject-length', None if not strong_tag(saved_tag) or headers.get('etag') != saved_tag: return 'reject-validator', None return 'append', prefix + body def digest(data): return hashlib.sha256(data).hexdigest() if data is not None else None def run(): # A persisted prefix is seeded, rather than simulating a dropped socket. cases = [ ('unchanged', '/same', 6, True, 206, 'append', A), ('changed-protected', '/changed', 6, True, 200, 'replace', B), ('changed-unprotected', '/changed', 6, False, 206, 'reject-validator', None), ('range-ignored', '/ignore', 6, True, 200, 'replace', A), ('offset-at-end', '/same', 16, True, 416, 'reject-status', None), ('wrong-start', '/wrong-start', 6, True, 206, 'reject-range', None), ('bad-length', '/bad-length', 6, True, 206, 'reject-length', None), ('denied', '/denied', 6, True, 403, 'reject-status', None), ] rows, captured = [], {} with ThreadingHTTPServer(('127.0.0.1', 0), Handler) as server: worker = threading.Thread(target=server.serve_forever, kwargs={'poll_interval': .01}) worker.start() try: for name, path, offset, conditional, expected_status, expected_action, expected_bytes in cases: request_headers = {'Range': f'bytes={offset}-', 'Accept-Encoding': 'identity'} if conditional: assert strong_tag(ETAG_A) request_headers['If-Range'] = ETAG_A connection = http.client.HTTPConnection('127.0.0.1', server.server_port, timeout=5) try: connection.request('GET', path, headers=request_headers) response = connection.getresponse() status = response.status headers = {k.lower(): v for k, v in response.getheaders()} body = response.read() finally: connection.close() action, output = consume(A[:offset], ETAG_A, len(A), status, headers, body) assert (status, action, output) == (expected_status, expected_action, expected_bytes), name captured[name] = body # Date/server banner and ephemeral port excluded from results. rows.append(dict(case=name, request_headers=request_headers, status=status, headers={k: headers[k] for k in ['content-length','content-range','etag'] if k in headers}, received_bytes=len(body), action=action, output_bytes=len(output) if output is not None else None, output_sha256=digest(output))) finally: server.shutdown() worker.join(timeout=5) assert not worker.is_alive() weak_tag = 'W/"version-a"' assert not strong_tag(weak_tag) # Preflight only; no ninth HTTP request. corrupt_same_length = A[:6] + captured['changed-unprotected'] corrupt_full_append = A[:6] + captured['changed-protected'] replaced = captured['changed-protected'] assert len(corrupt_same_length) == len(B) and corrupt_same_length not in (A, B) assert len(corrupt_full_append) == 22 and corrupt_full_append not in (A, B) assert replaced == B assemblies = [ dict(case='Range only, blind append', old_bytes=6, new_bytes=10, output_bytes=len(corrupt_same_length), equals_current=False, sha256=digest(corrupt_same_length)), dict(case='If-Range, blind 200 append', old_bytes=6, new_bytes=16, output_bytes=len(corrupt_full_append), equals_current=False, sha256=digest(corrupt_full_append)), dict(case='If-Range, replace on 200', old_bytes=0, new_bytes=16, output_bytes=len(replaced), equals_current=True, sha256=digest(replaced)), ] return dict(kind='loopback HTTP contract fixture', http_cases=rows, assemblies=assemblies, preflight={'weak_etag': weak_tag, 'resume_allowed': False}, input={'a_ascii':A.decode(), 'b_ascii':B.decode(), 'saved_prefix_bytes':6, 'a_sha256':digest(A),'b_sha256':digest(B)}, exclusions=['production server','CDN','browser download manager','CORS','compression', 'multipart ranges','actual interrupted socket','filesystem durability','throughput']) if __name__ == '__main__': result = run() Path('results.json').write_text(json.dumps(result, indent=2)+'\n') print('PASS: 8 loopback HTTP cases, 3 assembly comparisons, weak-tag preflight')