#!/usr/bin/env python3 """HTTP/2 proxy integration tests, stdlib + the server's nghttp2 HPACK library. Raw frames let tests deliberately withhold stream credit, cancel a stream, and send DATA without Content-Length. No third-party Python packages are required. """ import ctypes as C import ctypes.util import hashlib import json import pathlib import socket import ssl import struct import subprocess import time import unittest from test_proxy import ProxyTests, BLOCK def frame(kind, flags, stream, payload=b''): return len(payload).to_bytes(3, 'big') + bytes([kind, flags]) + stream.to_bytes(4, 'big') + payload def integer(value, bits, prefix=0): maximum = (1 << bits) - 1 if value < maximum: return bytes([prefix | value]) result = bytes([prefix | maximum]); value -= maximum while value >= 128: result += bytes([(value & 127) | 128]); value >>= 7 return result + bytes([value]) def literal(value): value = value.encode() if isinstance(value, str) else value return integer(len(value), 7) + value class NV(C.Structure): _fields_ = [('name', C.POINTER(C.c_ubyte)), ('value', C.POINTER(C.c_ubyte)), ('namelen', C.c_size_t), ('valuelen', C.c_size_t), ('flags', C.c_uint8)] class H2: def __init__(self, sock, window=65535): self.sock = sock; self.pending = b''; self.next_id = 1 self.streams = {}; self.window = 65535; self.initial = 65535; self.hold = set() lib = ctypes.util.find_library('nghttp2') if not lib: directory = subprocess.check_output(['pkg-config', '--variable=libdir', 'libnghttp2'], text=True).strip() lib = str(next(pathlib.Path(directory).glob('libnghttp2.*dylib'))) self.lib = C.CDLL(lib); self.inflater = C.c_void_p() self.lib.nghttp2_hd_inflate_new.argtypes = [C.POINTER(C.c_void_p)] self.lib.nghttp2_hd_inflate_hd2.argtypes = [C.c_void_p, C.POINTER(NV), C.POINTER(C.c_int), C.c_void_p, C.c_size_t, C.c_int] self.lib.nghttp2_hd_inflate_hd2.restype = C.c_ssize_t self.lib.nghttp2_hd_inflate_end_headers.argtypes = [C.c_void_p] self.lib.nghttp2_hd_inflate_del.argtypes = [C.c_void_p] assert self.lib.nghttp2_hd_inflate_new(C.byref(self.inflater)) == 0 self.sock.sendall(b'PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n' + frame(4, 0, 0, struct.pack('!HI', 4, window))) def close(self): self.lib.nghttp2_hd_inflate_del(self.inflater); self.sock.close() def request(self, path, method='GET', headers=(), ended=True): sid = self.next_id; self.next_id += 2 self.streams[sid] = {'headers': [], 'body': bytearray(), 'ended': False, 'reset': None, 'window': self.initial, 'block': b''} fields = [(':method', method), (':scheme', 'https'), (':authority', 'localhost'), (':path', path), *headers] block = b''.join(b'\x00' + literal(k) + literal(v) for k, v in fields) self.sock.sendall(frame(1, 4 | int(ended), sid, block)) return sid def decode(self, block): result = []; off = 0 while True: nv = NV(); flags = C.c_int() data = C.create_string_buffer(block[off:]) n = self.lib.nghttp2_hd_inflate_hd2(self.inflater, C.byref(nv), C.byref(flags), data, len(block) - off, 1) assert n >= 0, n off += n if flags.value & 2: result.append((C.string_at(nv.name, nv.namelen).decode(), C.string_at(nv.value, nv.valuelen).decode())) if flags.value & 1: break self.lib.nghttp2_hd_inflate_end_headers(self.inflater) return result def receive(self): while len(self.pending) < 9 or len(self.pending) < 9 + int.from_bytes(self.pending[:3], 'big'): data = self.sock.recv(65536) assert data, 'HTTP/2 connection closed unexpectedly' self.pending += data size = int.from_bytes(self.pending[:3], 'big') kind, flags = self.pending[3:5]; sid = int.from_bytes(self.pending[5:9], 'big') & 0x7fffffff payload, self.pending = self.pending[9:9+size], self.pending[9+size:] assert kind != 7, f'GOAWAY {payload!r}' if kind == 4 and not flags & 1: for off in range(0, len(payload), 6): key, value = struct.unpack('!HI', payload[off:off+6]) if key == 4: for stream in self.streams.values(): stream['window'] += value - self.initial self.initial = value self.sock.sendall(frame(4, 1, 0)) if kind == 8: increment = int.from_bytes(payload, 'big') & 0x7fffffff if sid: self.streams[sid]['window'] += increment else: self.window += increment if sid in self.streams: stream = self.streams[sid] if kind in (1, 9): stream['block'] += payload if flags & 4: stream['headers'].extend(self.decode(stream['block'])); stream['block'] = b'' if kind == 0: stream['body'].extend(payload) if payload: updates = frame(8, 0, 0, len(payload).to_bytes(4, 'big')) if sid not in self.hold: updates += frame(8, 0, sid, len(payload).to_bytes(4, 'big')) self.sock.sendall(updates) if kind == 3: stream['reset'] = int.from_bytes(payload, 'big') if kind in (0, 1) and flags & 1: stream['ended'] = True return kind, sid def wait(self, sid): stream = self.streams[sid] while not stream['ended'] and stream['reset'] is None: self.receive() return stream def upload(self, sid, data, ended=True): off = 0 while off < len(data): stream = self.streams[sid] while min(stream['window'], self.window) <= 0: self.receive() n = min(16384, stream['window'], self.window, len(data) - off) self.sock.sendall(frame(0, int(ended and off+n == len(data)), sid, data[off:off+n])) stream['window'] -= n; self.window -= n; off += n if not data and ended: self.sock.sendall(frame(0, 1, sid)) class H2ProxyTests(ProxyTests): # Share fixtures, but run only the HTTP/2 tests defined here. def setUp(self): super().setUp() if not self.tls: self.skipTest('TLS is unavailable in the Intel macOS build') self.clients = [] self.h2ctx = ssl.create_default_context(cafile=str(self.cert)); self.h2ctx.set_alpn_protocols(['h2']) def tearDown(self): for client in getattr(self, 'clients', []): client.close() # HTTP/2 has an independent 65535-byte raw upload window per stream. self.h2_peak_limit = 65535 + 16384 super().tearDown() def client(self, port, window=65535): sock = socket.create_connection(('127.0.0.1', port), timeout=5) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock = self.h2ctx.wrap_socket(sock, server_hostname='localhost') self.assertEqual(sock.selected_alpn_protocol(), 'h2') client = H2(sock, window); self.clients.append(client) return client def good(self, response, status='200'): self.assertIsNone(response['reset']) self.assertEqual(dict(response['headers'])[':status'], status) return response['body'] def test_h2_multiplex_and_headers(self): log = self.root / 'h2-access.log' port = self.start(True, ['--access-log', str(log)]); client = self.client(port) delayed = [client.request('/api/delay') for _ in range(12)] self.assertTrue(self.backend.seen.wait(2)) start = time.monotonic() fast = client.request('/api/path?x=1', headers=[('cookie', 'a=1'), ('cookie', 'b=2')]) static = client.request('/hello.txt') self.assertEqual(self.good(client.wait(static)), b'assets stay responsive') result = client.wait(fast); info = json.loads(self.good(result)) self.assertLess(time.monotonic() - start, .5) self.assertEqual(info['path'], '/path?x=1') self.assertEqual(info['headers']['x-forwarded-proto'], 'https') self.assertEqual(info['headers']['cookie'], 'a=1; b=2') self.assertEqual([v for k,v in result['headers'] if k == 'set-cookie'], ['a=1','b=2']) self.assertNotIn('x-remove', dict(result['headers'])) self.assertNotIn('connection', dict(result['headers'])) for sid in delayed: self.good(client.wait(sid)) self.assertIn('127.0.0.1 - - [', log.read_text()) self.assertIn('"GET /api/path?x=1 HTTP/2" 200', log.read_text()) def test_h2_uploads_and_limits(self): port = self.start(True, ['--proxy-buffer', '4096']); client = self.client(port) body = BLOCK * 16 for known in (True, False): headers = [('content-length', str(len(body)))] if known else [] sid = client.request('/api/upload', 'POST', headers, False) client.upload(sid, body[:32768], False) self.assertTrue(self.backend.upload_seen.wait(2), 'Upload was buffered until END_STREAM') client.upload(sid, body[32768:]) info = json.loads(self.good(client.wait(sid))) self.assertEqual(info['bytes'], len(body)); self.assertEqual(info['sha256'], hashlib.sha256(body).hexdigest()) sid = client.request('/api/upload', 'POST', [('content-length', str(len(body)+1))], False) self.good(client.wait(sid), '413') self.good(client.wait(client.request('/api/ok'))) def test_h2_backpressure_and_cancel(self): port = self.start(True, ['--proxy-buffer', '4096']); client = self.client(port, 4096) slow = client.request('/api/large'); client.hold.add(slow) while len(client.streams[slow]['body']) < 4096: client.receive() start = time.monotonic(); self.good(client.wait(client.request('/api/ok'))) self.assertLess(time.monotonic() - start, .5) client.sock.sendall(frame(3, 0, slow, (8).to_bytes(4,'big'))) large = client.request('/api/large') result = client.wait(large); self.good(result) self.assertEqual(hashlib.sha256(result['body']).digest(), hashlib.sha256(BLOCK * 256).digest()) # Cancel a request with an upstream still waiting for upload bytes. for _ in range(20): sid = client.request('/api/upload', 'POST', [], False) client.upload(sid, b'partial', False) client.sock.sendall(frame(3, 0, sid, (8).to_bytes(4,'big'))) self.good(client.wait(client.request('/api/ok'))) def test_h2_framing_and_failures(self): port = self.start(True, ['--proxy-timeout-ms','500']); client = self.client(port) for path, method, expected, status in [('/chunks','GET',b'wikipedia','200'),('/eof','GET',b'close delimited','200'), ('/head','HEAD',b'','200'),('/empty','GET',b'','204'),('/modified','GET',b'','304')]: response = client.wait(client.request('/api'+path, method)) self.assertEqual(self.good(response,status), expected) for i in range(8): self.good(client.wait(client.request(f'/api/bad/{i}')), '502') self.good(client.wait(client.request('/down/')), '502') stalled = client.request('/api/stall') self.good(client.wait(client.request('/api/ok'))) self.good(client.wait(stalled), '504') bad = client.wait(client.request('/api/truncated')) self.assertIsNotNone(bad['reset']) self.good(client.wait(client.request('/api/ok'))) def test_h2_malformed_and_trailers(self): port = self.start(True); client = self.client(port) # A short or overlong Content-Length body resets just that stream. for size, data in [('9', b'short'), ('2', b'long')]: sid = client.request('/api/upload', 'POST', [('content-length',size)], False) client.upload(sid,data) result = client.wait(sid) self.assertTrue(result['reset'] is not None or dict(result['headers']).get(':status') == '400') self.good(client.wait(client.request('/api/ok'))) sid = client.request('/api/upload', 'POST', [], False) client.upload(sid,b'body',False) client.sock.sendall(frame(1,5,sid,b'\x00'+literal('x-trailer')+literal('ignored'))) self.assertEqual(json.loads(self.good(client.wait(sid)))['bytes'],4) sid = client.request('/api/ok', headers=[('x-big','a'*8200)]) self.good(client.wait(sid),'431') sid = client.request('/api/reject','POST',[('content-length','1048576')],False) self.good(client.wait(sid),'413') client.upload(sid,b'ignored',False) client.sock.sendall(frame(3,0,sid,(8).to_bytes(4,'big'))) self.good(client.wait(client.request('/api/ok'))) def test_h2_disconnect_reclaims_slots(self): port = self.start(True,['--connections','8']) for _ in range(6): client = self.client(port) streams = [client.request('/api/stall') for _ in range(4)] self.good(client.wait(client.request('/api/ok'))) client.close(); self.clients.remove(client) client = self.client(port) self.good(client.wait(client.request('/api/ok'))) def test_h2_auth_and_expect(self): port = self.start(True, ['--auth','user:pass']); client = self.client(port) response = client.wait(client.request('/api/ok')); self.good(response, '401') self.assertIn('www-authenticate', dict(response['headers'])) sid = client.request('/api/echo', 'POST', [('authorization','Basic dXNlcjpwYXNz'), ('expect','100-continue'),('content-length','4')], False) while not client.streams[sid]['headers']: client.receive() self.assertEqual(dict(client.streams[sid]['headers'])[':status'], '100') client.upload(sid,b'test'); self.assertEqual(self.good(client.wait(sid)),b'test') if __name__ == '__main__': suite = unittest.TestSuite(H2ProxyTests(name) for name in sorted(H2ProxyTests.__dict__) if name.startswith('test_h2_')) result = unittest.TextTestRunner(verbosity=2).run(suite) raise SystemExit(not result.wasSuccessful())