diff --git a/lib/core/settings.py b/lib/core/settings.py index 70cf154e89a..966483adc99 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.8.6" +VERSION = "1.10.8.7" TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable" TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34} VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE) diff --git a/lib/request/http2.py b/lib/request/http2.py index 7af598c4193..289b4dbaca4 100644 --- a/lib/request/http2.py +++ b/lib/request/http2.py @@ -5,16 +5,17 @@ See the file 'LICENSE' for copying permission """ -# Native, dependency-free HTTP/2 client (RFC 7540) with HPACK (RFC 7541), replacing the optional -# 'httpx[http2]' third-party stack. The HPACK static and Huffman tables below are the canonical -# RFC 7541 tables; the codec is validated differentially against python-hyper/hpack and the client -# end-to-end against real h2 servers. Pure standard library, Python 2.7 / 3.x. +# Native, dependency-free HTTP/2 client (RFC 9113) with HPACK (RFC 7541). Runtime code uses only +# the standard library. The accompanying tests optionally use python-hyper/hpack and hyper-h2 for +# differential and local peer validation. The implementation keeps Python 2.7-compatible syntax, +# but this rewrite was executed and tested on Python 3.13 only. import base64 import socket import ssl import struct import threading +from collections import OrderedDict try: from http.client import responses as _HTTP_RESPONSES @@ -22,8 +23,9 @@ from httplib import responses as _HTTP_RESPONSES try: - from urllib.parse import urljoin, urlsplit + from urllib.parse import quote, urljoin, urlsplit except ImportError: + from urllib import quote from urlparse import urljoin, urlsplit from email.message import Message as _Message @@ -140,115 +142,234 @@ ) STATIC_LEN = len(STATIC_TABLE) +# HTTP/2 frame codec and client implementation (RFC 9113; HPACK remains RFC 7541). +# The public API is compatible with the original module. Optional `verify` and +# `ssl_context` arguments were added so TLS verification is opt-in. By default +# verification is disabled (the tool is used against test/self-signed targets). -# HTTP/2 frame codec (RFC 7540 section 4.1) - the zero-table-risk brick. Pure stdlib, py2/py3, ASCII. +try: + text_type = unicode +except NameError: + text_type = str -# frame types (RFC 7540 s6) -DATA, HEADERS, RST_STREAM, SETTINGS, PUSH_PROMISE, PING, GOAWAY, WINDOW_UPDATE, CONTINUATION = 0x0, 0x1, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0x9 -# flags +try: + integer_types = (int, long) +except NameError: + integer_types = (int,) + +# Frame types +DATA = 0x0 +HEADERS = 0x1 +PRIORITY = 0x2 +RST_STREAM = 0x3 +SETTINGS = 0x4 +PUSH_PROMISE = 0x5 +PING = 0x6 +GOAWAY = 0x7 +WINDOW_UPDATE = 0x8 +CONTINUATION = 0x9 + +# Flags FLAG_END_STREAM = 0x1 FLAG_ACK = 0x1 FLAG_END_HEADERS = 0x4 FLAG_PADDED = 0x8 FLAG_PRIORITY = 0x20 +# Settings +SETTINGS_HEADER_TABLE_SIZE = 0x1 +SETTINGS_ENABLE_PUSH = 0x2 +SETTINGS_MAX_CONCURRENT_STREAMS = 0x3 +SETTINGS_INITIAL_WINDOW_SIZE = 0x4 +SETTINGS_MAX_FRAME_SIZE = 0x5 +SETTINGS_MAX_HEADER_LIST_SIZE = 0x6 + +# Error codes +NO_ERROR = 0x0 +PROTOCOL_ERROR = 0x1 +INTERNAL_ERROR = 0x2 +FLOW_CONTROL_ERROR = 0x3 +SETTINGS_TIMEOUT = 0x4 +STREAM_CLOSED = 0x5 +FRAME_SIZE_ERROR = 0x6 +REFUSED_STREAM = 0x7 +CANCEL = 0x8 +COMPRESSION_ERROR = 0x9 +CONNECT_ERROR = 0xa +ENHANCE_YOUR_CALM = 0xb +INADEQUATE_SECURITY = 0xc +HTTP_1_1_REQUIRED = 0xd + CONNECTION_PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" +DEFAULT_MAX_FRAME_SIZE = 16384 +MAX_FRAME_SIZE = 0xffffff +MAX_WINDOW = (1 << 31) - 1 +DEFAULT_WINDOW = 65535 +LOCAL_INITIAL_WINDOW = 16 * 1024 * 1024 +MAX_RESPONSE_SIZE = 100 * 1024 * 1024 +MAX_HEADER_BLOCK_SIZE = 1024 * 1024 +MAX_HEADER_LIST_SIZE = 1024 * 1024 +MAX_HPACK_INTEGER = MAX_WINDOW +MAX_STREAM_ID = (1 << 31) - 1 +H2_POOL_LIMIT = 32 +MAX_INFORMATIONAL_RESPONSES = 32 +MAX_INFORMATIONAL_HEADER_BYTES = 1024 * 1024 +MAX_RESPONSE_FRAMES = 100000 + +_TOKEN_BYTES = frozenset(bytearray(b"!#$%&'*+-.^_`|~0123456789abcdefghijklmnopqrstuvwxyz")) +_METHOD_TOKEN_BYTES = frozenset(bytearray( + b"!#$%&'*+-.^_`|~0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" +)) +_CONNECTION_FIELDS = frozenset(( + b"connection", b"proxy-connection", b"keep-alive", b"transfer-encoding", b"upgrade" +)) +_SENSITIVE_REDIRECT_FIELDS = frozenset((b"authorization", b"cookie", b"proxy-authorization")) +_CONTENT_REDIRECT_FIELDS = frozenset((b"content-length", b"content-type", b"content-encoding")) + + +class H2Error(IOError): + pass + + +class H2ProtocolError(H2Error): + pass + + +class H2CompressionError(H2ProtocolError): + pass + + +class H2TransportError(H2Error): + pass + + +class _UnprocessedStream(H2Error): + """The peer explicitly guaranteed that the request stream was not processed.""" + + +def _as_bytes(value): + if isinstance(value, bytes): + return value + if isinstance(value, bytearray): + return bytes(value) + if isinstance(value, text_type): + return value.encode("latin-1") + raise TypeError("expected bytes/text value, got %s" % type(value).__name__) + def encode_frame(ftype, flags, stream_id, payload=b""): - """Serialize an HTTP/2 frame (RFC 7540 s4.1): 24-bit length + type + flags + 31-bit stream id. + """Serialize an HTTP/2 frame. >>> decode_frame_header(encode_frame(HEADERS, FLAG_END_HEADERS, 1, b'abc')[:9]) (3, 1, 4, 1) """ - if len(payload) > 0xffffff: + payload = _as_bytes(payload) + if not (0 <= ftype <= 0xff and 0 <= flags <= 0xff): + raise ValueError("frame type and flags must fit in one octet") + if not (0 <= stream_id <= MAX_STREAM_ID): + raise ValueError("stream id must be between 0 and 2^31-1") + if len(payload) > MAX_FRAME_SIZE: raise ValueError("frame payload exceeds 24-bit length") - header = struct.pack("!I", len(payload))[1:] # 24-bit length (drop MSB of the 32-bit pack) - header += struct.pack("!BBI", ftype, flags, stream_id & 0x7fffffff) # type, flags, R(1)+stream(31) + header = struct.pack("!I", len(payload))[1:] + header += struct.pack("!BBI", ftype, flags, stream_id) return header + payload + def decode_frame_header(nine): - """Parse the 9-byte frame header into (length, type, flags, stream_id); the reserved high bit of the stream id is masked off. + """Parse a 9-byte frame header into (length, type, flags, stream_id). - >>> decode_frame_header(encode_frame(DATA, 0, 0x80000001, b'')[:9]) + >>> decode_frame_header(encode_frame(DATA, 0, 1, b'')[:9]) (0, 0, 0, 1) """ if len(nine) != 9: raise ValueError("frame header must be exactly 9 bytes") length = struct.unpack("!I", b"\x00" + nine[:3])[0] ftype, flags, stream_id = struct.unpack("!BBI", nine[3:9]) - return length, ftype, flags, stream_id & 0x7fffffff + return length, ftype, flags, stream_id & MAX_STREAM_ID + # ---------- Huffman ---------- def huffman_encode(data): - """Huffman-encode a byte string per the RFC 7541 static table (s5.2), padding with EOS 1-bits. + """Huffman-encode a byte string using the RFC 7541 static code. >>> huffman_decode(huffman_encode(b'www.example.com')) == b'www.example.com' True - >>> huffman_encode(b'') == b'' - True """ + data = _as_bytes(data) if not data: return b"" acc = 0 nbits = 0 - for b in bytearray(data): - acc = (acc << HUFFMAN_LENGTHS[b]) | HUFFMAN_CODES[b] - nbits += HUFFMAN_LENGTHS[b] + for item in bytearray(data): + acc = (acc << HUFFMAN_LENGTHS[item]) | HUFFMAN_CODES[item] + nbits += HUFFMAN_LENGTHS[item] pad = (8 - nbits % 8) % 8 - acc = (acc << pad) | ((1 << pad) - 1) # pad with 1-bits (EOS prefix) + acc = (acc << pad) | ((1 << pad) - 1) total = (nbits + pad) // 8 out = bytearray() - for i in range(total - 1, -1, -1): - out.append((acc >> (8 * i)) & 0xff) + for index in range(total - 1, -1, -1): + out.append((acc >> (8 * index)) & 0xff) return bytes(out) + _HUFF_ROOT = {} + + def _build_huffman_trie(): for sym in range(256): - code, length = HUFFMAN_CODES[sym], HUFFMAN_LENGTHS[sym] + code = HUFFMAN_CODES[sym] + length = HUFFMAN_LENGTHS[sym] node = _HUFF_ROOT - for i in range(length - 1, -1, -1): - bit = (code >> i) & 1 - if i == 0: - node[bit] = sym # leaf: int symbol + for index in range(length - 1, -1, -1): + bit = (code >> index) & 1 + if index == 0: + node[bit] = sym else: - node = node.setdefault(bit, {}) + child = node.get(bit) + if child is None: + child = {} + node[bit] = child + elif not isinstance(child, dict): + raise AssertionError("invalid prefix Huffman table") + node = child + + _build_huffman_trie() + def huffman_decode(data): + data = _as_bytes(data) out = bytearray() node = _HUFF_ROOT - consumed = 0 # bits into the current (partial) symbol - for byte in bytearray(data): - for i in range(7, -1, -1): - bit = (byte >> i) & 1 + consumed = 0 + tail = 0 + for item in bytearray(data): + for index in range(7, -1, -1): + bit = (item >> index) & 1 nxt = node.get(bit) if nxt is None: - raise ValueError("invalid Huffman sequence") + raise H2CompressionError("invalid Huffman sequence") consumed += 1 + tail = (tail << 1) | bit if isinstance(nxt, dict): node = nxt else: out.append(nxt) node = _HUFF_ROOT consumed = 0 - # RFC 7541 5.2: any leftover partial path must be EOS padding: all 1-bits and fewer than 8 + tail = 0 if node is not _HUFF_ROOT: - if consumed >= 8: - raise ValueError("Huffman padding too long") - # walk back is unnecessary: padding is all-ones, i.e. we must have only taken '1' branches - # since the last leaf; verify by re-deriving is overkill - reference cross-check guards it + if consumed > 7 or tail != (1 << consumed) - 1: + raise H2CompressionError("invalid Huffman EOS padding") return bytes(out) -# ---------- integer / string (RFC 7541 5.1 / 5.2) ---------- -def encode_integer(value, prefix_bits, first_byte=0): - """Encode an integer with an N-bit prefix (RFC 7541 s5.1); the C.1.2 example is 1337 / 5-bit prefix. - >>> list(encode_integer(10, 5)) - [10] - >>> list(encode_integer(1337, 5)) - [31, 154, 10] - """ +# ---------- HPACK integer/string ---------- +def encode_integer(value, prefix_bits, first_byte=0): + if not isinstance(value, integer_types) or value < 0: + raise ValueError("HPACK integer must be non-negative") + if not 1 <= prefix_bits <= 8: + raise ValueError("invalid HPACK prefix width") mask = (1 << prefix_bits) - 1 if value < mask: return bytearray([first_byte | value]) @@ -260,12 +381,12 @@ def encode_integer(value, prefix_bits, first_byte=0): out.append(value) return out -def decode_integer(data, pos, prefix_bits): - """Decode an N-bit-prefixed integer, returning (value, new_pos) (RFC 7541 s5.1). - >>> decode_integer(bytearray([31, 154, 10]), 0, 5) - (1337, 3) - """ +def decode_integer(data, pos, prefix_bits, max_value=MAX_HPACK_INTEGER): + if not 1 <= prefix_bits <= 8: + raise H2CompressionError("invalid HPACK prefix width") + if pos < 0 or pos >= len(data): + raise H2CompressionError("truncated HPACK integer") mask = (1 << prefix_bits) - 1 value = data[pos] & mask pos += 1 @@ -273,41 +394,62 @@ def decode_integer(data, pos, prefix_bits): return value, pos shift = 0 while True: - b = data[pos] + if pos >= len(data): + raise H2CompressionError("truncated HPACK integer") + item = data[pos] pos += 1 - value += (b & 0x7f) << shift + addend = (item & 0x7f) << shift + if addend > max_value - value: + raise H2CompressionError("HPACK integer exceeds implementation limit") + value += addend + if not (item & 0x80): + return value, pos shift += 7 - if not (b & 0x80): - break - return value, pos + if shift > 28: + raise H2CompressionError("HPACK integer encoding is too long") + def encode_string(value, huffman=True): + value = _as_bytes(value) if huffman: encoded = huffman_encode(value) - if len(encoded) < len(value): # only use Huffman when it actually shrinks - return encode_integer(len(encoded), 7, 0x80) + encoded + if len(encoded) < len(value): + return encode_integer(len(encoded), 7, 0x80) + bytearray(encoded) return encode_integer(len(value), 7, 0x00) + bytearray(value) + def decode_string(data, pos): - huffman = bool(data[pos] & 0x80) - length, pos = decode_integer(data, pos, 7) - raw = bytes(data[pos:pos + length]) - pos += length - return (huffman_decode(raw) if huffman else raw), pos + if pos < 0 or pos >= len(data): + raise H2CompressionError("truncated HPACK string") + use_huffman = bool(data[pos] & 0x80) + length, pos = decode_integer(data, pos, 7, max_value=MAX_HEADER_BLOCK_SIZE) + end = pos + length + if end > len(data): + raise H2CompressionError("truncated HPACK string payload") + raw = bytes(data[pos:end]) + return (huffman_decode(raw) if use_huffman else raw), end + -# ---------- dynamic table + decoder/encoder ---------- class Decoder(object): - def __init__(self, max_size=4096): + def __init__(self, max_size=4096, max_header_list_size=MAX_HEADER_LIST_SIZE): + self.max_allowed_size = max_size self.max_size = max_size - self.dynamic = [] # newest first: [(name, value), ...] + self.max_header_list_size = max_header_list_size + self.dynamic = [] self._size = 0 - def _entry_size(self, name, value): + @staticmethod + def _entry_size(name, value): return 32 + len(name) + len(value) def _add(self, name, value): + entry_size = self._entry_size(name, value) + if entry_size > self.max_size: + self.dynamic = [] + self._size = 0 + return self.dynamic.insert(0, (name, value)) - self._size += self._entry_size(name, value) + self._size += entry_size self._evict() def _evict(self): @@ -317,30 +459,29 @@ def _evict(self): def _get(self, index): if index <= 0: - raise ValueError("invalid header index 0") + raise H2CompressionError("invalid HPACK header index 0") if index <= STATIC_LEN: return STATIC_TABLE[index - 1] - index -= STATIC_LEN + 1 - if index >= len(self.dynamic): - raise ValueError("dynamic index out of range") - return self.dynamic[index] + dynamic_index = index - STATIC_LEN - 1 + if dynamic_index < 0 or dynamic_index >= len(self.dynamic): + raise H2CompressionError("HPACK dynamic index out of range") + return self.dynamic[dynamic_index] def decode(self, data): - """Decode an HPACK header block into a list of (name, value) byte pairs (RFC 7541 s6). - - >>> Decoder().decode(bytes(bytearray([0x82, 0x86, 0x84]))) == [(b':method', b'GET'), (b':scheme', b'http'), (b':path', b'/')] - True - """ - data = bytearray(data) + data = bytearray(_as_bytes(data)) pos = 0 headers = [] - n = len(data) - while pos < n: - byte = data[pos] - if byte & 0x80: # 6.1 indexed + list_size = 0 + saw_header = False + while pos < len(data): + first = data[pos] + if first & 0x80: + saw_header = True index, pos = decode_integer(data, pos, 7) - headers.append(self._get(index)) - elif byte & 0x40: # 6.2.1 literal + incremental indexing + name, value = self._get(index) + headers.append((name, value)) + elif first & 0x40: + saw_header = True index, pos = decode_integer(data, pos, 6) if index: name = self._get(index)[0] @@ -349,11 +490,17 @@ def decode(self, data): value, pos = decode_string(data, pos) self._add(name, value) headers.append((name, value)) - elif byte & 0x20: # 6.3 dynamic table size update + elif first & 0x20: + if saw_header: + raise H2CompressionError("dynamic table size update is not at block start") new_size, pos = decode_integer(data, pos, 5) + if new_size > self.max_allowed_size: + raise H2CompressionError("dynamic table size exceeds advertised maximum") self.max_size = new_size self._evict() - else: # 6.2.2 without / 6.2.3 never indexed (4-bit prefix) + continue + else: + saw_header = True index, pos = decode_integer(data, pos, 4) if index: name = self._get(index)[0] @@ -361,76 +508,119 @@ def decode(self, data): name, pos = decode_string(data, pos) value, pos = decode_string(data, pos) headers.append((name, value)) + + name, value = headers[-1] + list_size += self._entry_size(name, value) + if list_size > self.max_header_list_size: + raise H2CompressionError("decoded header list exceeds configured limit") return headers + class Encoder(object): - # Minimal, always-valid: emit each header as a literal WITHOUT indexing + Huffman-coded strings. - # (Correctness-critical decoding is the hard part; a server accepts this trivially.) + """A stateless, literal-without-indexing HPACK encoder.""" + def encode(self, headers): out = bytearray() for name, value in headers: - out += encode_integer(0, 4, 0x00) # 0000 0000 : literal w/o indexing, new name + name = _as_bytes(name) + value = _as_bytes(value) + out += encode_integer(0, 4, 0x00) out += encode_string(name) out += encode_string(value) return bytes(out) -SETTINGS_ENABLE_PUSH = 0x2 -SETTINGS_INITIAL_WINDOW_SIZE = 0x4 -BIG_WINDOW = (1 << 31) - 1 - -# Upper bound on the response bytes (body or header block) buffered per stream. The client advertises a -# ~2GB flow-control window, so without this a large (or hostile) server would drive the whole body into -# memory and OOM the process. Mirrors the HTTP/1.1 path's MAX_CONNECTION_TOTAL_SIZE (100MB) cap in -# connect.py; a stream that exceeds it is truncated (body) or abandoned (headers) and its connection retired. -MAX_RESPONSE_SIZE = 100 * 1024 * 1024 -def _recv_exact(sock, n): - buf = b"" - while len(buf) < n: - chunk = sock.recv(n - len(buf)) +def _recv_exact(sock, count): + if count < 0: + raise ValueError("negative receive size") + chunks = [] + received = 0 + while received < count: + chunk = sock.recv(count - received) if not chunk: - raise IOError("connection closed by peer") - buf += chunk - return buf + raise H2TransportError("connection closed by peer") + chunks.append(chunk) + received += len(chunk) + return b"".join(chunks) -def _read_frame(sock): + +def _read_frame(sock, max_frame_size=DEFAULT_MAX_FRAME_SIZE): length, ftype, flags, sid = decode_frame_header(_recv_exact(sock, 9)) - return ftype, flags, sid, (_recv_exact(sock, length) if length else b"") + if length > max_frame_size: + raise H2ProtocolError("received frame exceeds local maximum frame size") + payload = _recv_exact(sock, length) if length else b"" + return ftype, flags, sid, payload + + +def _idna_host(host): + if isinstance(host, bytes): + try: + host = host.decode("ascii") + except UnicodeDecodeError: + try: + host = host.decode("utf-8") + except UnicodeDecodeError: + raise H2Error("hostname bytes are neither ASCII nor UTF-8") + if not isinstance(host, text_type): + host = text_type(host) + if host.startswith("[") and host.endswith("]"): + host = host[1:-1] + if not host: + raise H2Error("hostname is required") + if ":" in host: # IPv6 literal, optionally with a zone identifier + return host + try: + return host.encode("idna").decode("ascii") + except (UnicodeError, AttributeError): + raise H2Error("invalid internationalized hostname %r" % host) + + +def _format_authority_host(host): + host = _idna_host(host) + if ":" in host: + return "[%s]" % host + return host + + +def _authority_for_host(host, port): + authority = _format_authority_host(host) + if port != 443: + authority += ":%d" % port + return authority -def _tob(x): - return x if isinstance(x, bytes) else x.encode("latin-1") def _connect_socket(host, port, proxy, timeout): - # Direct TCP, or an HTTP CONNECT tunnel through an (optionally authenticated) proxy. SOCKS proxies - # are excluded for HTTP/2 upstream, so any proxy reaching here is a plain HTTP one. proxy is a - # (proxy_host, proxy_port, "user:pass"-or-None) tuple. if not proxy: return socket.create_connection((host, port), timeout=timeout) proxy_host, proxy_port, proxy_cred = proxy raw = socket.create_connection((proxy_host, proxy_port), timeout=timeout) try: - request = "CONNECT %s:%d HTTP/1.1\r\nHost: %s:%d\r\n" % (host, port, host, port) + tunnel_host = _format_authority_host(host) + authority = "%s:%d" % (tunnel_host, port) + request = "CONNECT %s HTTP/1.1\r\nHost: %s\r\n" % (authority, authority) if proxy_cred: - token = base64.b64encode(proxy_cred.encode("latin-1")).decode("ascii") + token = base64.b64encode(_as_bytes(proxy_cred)).decode("ascii") request += "Proxy-Authorization: Basic %s\r\n" % token request += "\r\n" raw.sendall(request.encode("latin-1")) - response = b"" - while b"\r\n\r\n" not in response: - chunk = raw.recv(4096) + # Read only through the CONNECT header terminator. A large recv() can consume + # bytes from the tunneled TLS handshake that immediately follows the proxy + # response, and those bytes cannot be pushed back before wrap_socket(). + response = bytearray() + while not bytes(response).endswith(b"\r\n\r\n"): + chunk = raw.recv(1) if not chunk: - raise IOError("proxy closed the connection during CONNECT") - response += chunk + raise H2Error("proxy closed connection during CONNECT") + response.extend(chunk) if len(response) > 65536: - raise IOError("oversized proxy CONNECT response") - - status_line = response.split(b"\r\n", 1)[0].decode("latin-1", "replace") + raise H2Error("oversized proxy CONNECT response") + status_line = bytes(response).split(b"\r\n", 1)[0].decode("latin-1", "replace") fields = status_line.split(None, 2) code = int(fields[1]) if len(fields) >= 2 and fields[1].isdigit() else 0 - if not (200 <= code < 300): - raise IOError("proxy CONNECT failed: %s" % status_line) + if not 200 <= code < 300: + raise H2Error("proxy CONNECT failed: %s" % status_line) return raw except Exception: try: @@ -439,281 +629,858 @@ def _connect_socket(host, port, proxy, timeout): pass raise -class _UnprocessedStream(IOError): - """Raised when the server made it clear our stream was NOT processed (GOAWAY with last-stream-id below - ours), so the request is always safe to retry on a fresh connection.""" + +def _make_ssl_context(verify, ssl_context): + context = ssl_context + if context is None: + if verify: + creator = getattr(ssl, "create_default_context", None) + if creator is None: + raise H2Error("certificate verification requires ssl.create_default_context") + context = creator() + else: + creator = getattr(ssl, "_create_unverified_context", None) + if creator is not None: + context = creator() + else: + context = ssl.SSLContext(ssl.PROTOCOL_SSLv23) + context.verify_mode = ssl.CERT_NONE + tls_version = getattr(ssl, "TLSVersion", None) + if tls_version is not None and hasattr(context, "minimum_version"): + if context.minimum_version < tls_version.TLSv1_2: + context.minimum_version = tls_version.TLSv1_2 + else: + for option_name in ("OP_NO_SSLv2", "OP_NO_SSLv3", "OP_NO_TLSv1", "OP_NO_TLSv1_1"): + option = getattr(ssl, option_name, 0) + if option: + context.options |= option + no_compression = getattr(ssl, "OP_NO_COMPRESSION", 0) + if no_compression: + context.options |= no_compression + setter = getattr(context, "set_alpn_protocols", None) + if setter is None: + raise H2Error("this Python/OpenSSL build does not support ALPN") + setter(["h2"]) + return context + + +def _iter_header_items(headers): + if headers is None: + return [] + if hasattr(headers, "items"): + return list(headers.items()) + return list(headers) + + +def _validate_regular_name(name): + if not name or name.startswith(b":"): + raise H2ProtocolError("invalid regular field name") + for item in bytearray(name): + if item not in _TOKEN_BYTES: + raise H2ProtocolError("invalid HTTP field name %r" % name) + + +def _validate_field_value(value): + if b"\x00" in value or b"\r" in value or b"\n" in value: + raise H2ProtocolError("invalid control character in HTTP field value") + if value[:1] in (b" ", b"\t") or value[-1:] in (b" ", b"\t"): + raise H2ProtocolError("HTTP/2 field values may not have leading/trailing whitespace") + + +def _validate_pseudo_value(name, value): + _validate_field_value(value) + for item in bytearray(value): + if item < 0x21 or item > 0x7e: + raise H2ProtocolError("invalid octet in %s pseudo-field" % name) + + +def _validate_method(method): + if not method: + raise H2ProtocolError("method is required") + for item in bytearray(method): + if item not in _METHOD_TOKEN_BYTES: + raise H2ProtocolError("invalid HTTP method token") + + +def _normalize_request_headers(headers, body): + raw_items = [] + connection_tokens = set() + for key, value in _iter_header_items(headers): + name = _as_bytes(key).lower() + val = _as_bytes(value) + if name.startswith(b":"): + raise H2ProtocolError("callers may not inject HTTP/2 pseudo-fields") + _validate_regular_name(name) + _validate_field_value(val) + raw_items.append((name, val)) + if name == b"connection": + for token in val.split(b","): + token = token.strip().lower() + if token: + _validate_regular_name(token) + connection_tokens.add(token) + + normalized = [] + content_lengths = [] + content_length_added = False + for name, value in raw_items: + if name == b"host" or name == b"proxy-authorization": + continue + if name in _CONNECTION_FIELDS or name in connection_tokens: + continue + if name == b"te" and value.strip().lower() != b"trailers": + raise H2ProtocolError("HTTP/2 permits TE only with value 'trailers'") + if name == b"content-length": + content_lengths.append(value.strip()) + if content_length_added: + continue + content_length_added = True + normalized.append((name, value)) + + if content_lengths: + if len(set(content_lengths)) != 1 or not content_lengths[0].isdigit(): + raise H2ProtocolError("invalid or conflicting content-length fields") + expected = len(body) + if int(content_lengths[0]) != expected: + raise H2ProtocolError("content-length does not match request body") + return normalized + + +def _request_header_list(method, path, authority, headers, body): + method = _as_bytes(method) + authority = _as_bytes(authority) + path = _as_bytes(path) + _validate_method(method) + if not authority: + raise H2ProtocolError("authority is required") + _validate_pseudo_value(":authority", authority) + if method == b"CONNECT": + request = [(b":method", method), (b":authority", authority)] + else: + if not path: + raise H2ProtocolError("non-CONNECT requests require a non-empty :path") + _validate_pseudo_value(":path", path) + request = [(b":method", method), (b":scheme", b"https"), (b":path", path), (b":authority", authority)] + request.extend(_normalize_request_headers(headers, body)) + return request + + +def _field_section_size(headers): + return sum(32 + len(name) + len(value) for name, value in headers) + + +def _header_frames(stream_id, block, max_frame_size, end_stream): + if max_frame_size < DEFAULT_MAX_FRAME_SIZE: + raise H2ProtocolError("peer advertised an invalid maximum frame size") + if not block: + return [encode_frame(HEADERS, FLAG_END_HEADERS | (FLAG_END_STREAM if end_stream else 0), stream_id, b"")] + frames = [] + pos = 0 + first = True + while pos < len(block): + chunk = block[pos:pos + max_frame_size] + pos += len(chunk) + final = pos == len(block) + if first: + flags = (FLAG_END_STREAM if end_stream else 0) | (FLAG_END_HEADERS if final else 0) + frames.append(encode_frame(HEADERS, flags, stream_id, chunk)) + first = False + else: + flags = FLAG_END_HEADERS if final else 0 + frames.append(encode_frame(CONTINUATION, flags, stream_id, chunk)) + return frames + + +def _data_frames(stream_id, body, max_frame_size): + frames = [] + pos = 0 + while pos < len(body): + chunk = body[pos:pos + max_frame_size] + pos += len(chunk) + flags = FLAG_END_STREAM if pos == len(body) else 0 + frames.append(encode_frame(DATA, flags, stream_id, chunk)) + return frames + + +def _extract_header_fragment(ftype, flags, payload): + if ftype == CONTINUATION: + return payload + pos = 0 + pad_length = 0 + if flags & FLAG_PADDED: + if not payload: + raise H2ProtocolError("padded HEADERS frame has no pad length") + pad_length = bytearray(payload)[0] + pos = 1 + if flags & FLAG_PRIORITY: + if len(payload) - pos < 5: + raise H2ProtocolError("truncated HEADERS priority fields") + pos += 5 + if pad_length > len(payload) - pos: + raise H2ProtocolError("HEADERS padding exceeds frame payload") + end = len(payload) - pad_length + return payload[pos:end] + + +def _extract_data(flags, payload): + if not (flags & FLAG_PADDED): + return payload + if not payload: + raise H2ProtocolError("padded DATA frame has no pad length") + pad_length = bytearray(payload)[0] + if pad_length > len(payload) - 1: + raise H2ProtocolError("DATA padding exceeds frame payload") + return payload[1:len(payload) - pad_length] + + +def _parse_content_length(headers, trailers=False): + values = [] + for name, value in headers: + if _as_bytes(name).lower() != b"content-length": + continue + if trailers: + raise H2ProtocolError("content-length is not permitted in trailers") + for part in _as_bytes(value).split(b","): + part = part.strip() + if not part or not part.isdigit(): + raise H2ProtocolError("invalid content-length field") + values.append(int(part)) + if not values: + return None + expected = values[0] + if any(value != expected for value in values[1:]): + raise H2ProtocolError("conflicting content-length fields") + return expected + + +def _validate_response_field_section(headers, trailers=False): + regular_seen = False + status = None + for name, value in headers: + name = _as_bytes(name) + value = _as_bytes(value) + _validate_field_value(value) + if name.startswith(b":"): + if trailers or regular_seen or name != b":status" or status is not None: + raise H2ProtocolError("invalid response pseudo-field ordering or duplication") + if len(value) != 3 or not value.isdigit(): + raise H2ProtocolError("invalid :status pseudo-field") + status = int(value) + if not 100 <= status <= 599: + raise H2ProtocolError("response status is outside the valid range") + if status == 101: + raise H2ProtocolError("101 Switching Protocols is not valid in HTTP/2") + else: + regular_seen = True + if name.lower() != name: + raise H2ProtocolError("uppercase HTTP/2 field name") + _validate_regular_name(name) + if name in _CONNECTION_FIELDS: + raise H2ProtocolError("connection-specific field in HTTP/2 response") + content_length = _parse_content_length(headers, trailers=trailers) + if trailers: + return None, None + if status is None: + raise H2ProtocolError("response field section lacks :status") + return status, content_length + + +class _ResponseState(object): + def __init__(self, stream_id, request_method): + self.stream_id = stream_id + self.request_method = _as_bytes(request_method) + self.block = None + self.block_end_stream = False + self.headers = None + self.trailers = [] + self.informational_count = 0 + self.informational_bytes = 0 + self.status = None + self.expected_content_length = None + self.body_forbidden = False + self.body = bytearray() + self.recv_window = LOCAL_INITIAL_WINDOW + self.complete = False + class _H2Connection(object): - """A single HTTP/2 connection reused for sequential (one-stream-at-a-time) requests within a thread. - - Multiplexing is intentionally NOT used - one stream is fully consumed before the next is opened - which - preserves request<->response isolation (clean time-based latency, no desync), exactly like the - thread-local HTTP/1.1 keep-alive pool. Reuse amortizes the TCP+TLS+preface cost across all of a thread's - requests to a host. Correctness note: only the HPACK Decoder (server->client dynamic table) is stateful, - so it is kept per-connection and fed responses in order; the Encoder is literal-without-indexing - (stateless), hence a fresh one per request is safe on a reused socket.""" - - def __init__(self, host, port, proxy, timeout): - self.host, self.port, self.proxy = host, port, proxy - self.dec = Decoder() # persistent server->client HPACK table - self.next_sid = 1 # odd, strictly increasing per RFC 7540 + """A sequential HTTP/2 connection with an explicit two-stream timing primitive.""" + + def __init__(self, host, port, proxy, timeout, verify=False, ssl_context=None): + self.host = _idna_host(host) + self.port = port + self.proxy = proxy + self.dec = Decoder() + self.next_sid = 1 self.usable = True - ctx = ssl._create_unverified_context() - ctx.set_alpn_protocols(["h2"]) - raw = _connect_socket(host, port, proxy, timeout) + self.sock = None + self.peer_max_frame_size = DEFAULT_MAX_FRAME_SIZE + self.peer_initial_window = DEFAULT_WINDOW + self.peer_max_concurrent_streams = None + self.peer_max_header_list_size = None + self.conn_send_window = DEFAULT_WINDOW + self.conn_recv_window = LOCAL_INITIAL_WINDOW + self.stream_send_windows = {} + self.local_max_frame_size = DEFAULT_MAX_FRAME_SIZE + self._our_settings_acked = False + context = _make_ssl_context(verify, ssl_context) + raw = _connect_socket(self.host, port, proxy, timeout) try: - raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) # coalesced-pair writes must not be Nagle-buffered - except (OSError, socket.error): - pass - self.sock = ctx.wrap_socket(raw, server_hostname=host) + try: + raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + except (OSError, socket.error): + pass + self.sock = context.wrap_socket(raw, server_hostname=self.host) + except Exception: + try: + raw.close() + except Exception: + pass + raise try: - if self.sock.selected_alpn_protocol() != "h2": - raise IOError("server did not negotiate h2 (ALPN=%r)" % self.sock.selected_alpn_protocol()) + selected = getattr(self.sock, "selected_alpn_protocol", lambda: None)() + if selected != "h2": + raise H2Error("server did not negotiate h2 (ALPN=%r)" % selected) + tls_version = getattr(self.sock, "version", lambda: None)() + if tls_version not in ("TLSv1.2", "TLSv1.3"): + raise H2Error("HTTP/2 requires TLS 1.2 or newer (negotiated %r)" % tls_version) + compression = getattr(self.sock, "compression", lambda: None)() + if compression is not None: + raise H2Error("TLS compression is not permitted for HTTP/2") self.sock.settimeout(timeout) - # connection preface + client SETTINGS (disable server push + advertise a large per-stream window) - # + bump conn window. ENABLE_PUSH=0 keeps servers from opening pushed streams whose HPACK header - # block we would otherwise have to decode to keep the dynamic table in sync (skipping it desyncs - # the decoder and corrupts every later header) - this client has no use for pushed responses. - self.sock.sendall(CONNECTION_PREFACE) - self.sock.sendall(encode_frame(SETTINGS, 0, 0, struct.pack("!HIHI", SETTINGS_ENABLE_PUSH, 0, SETTINGS_INITIAL_WINDOW_SIZE, BIG_WINDOW))) - self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, 0, struct.pack("!I", BIG_WINDOW - 65535))) + settings_payload = struct.pack( + "!HIHIHI", + SETTINGS_ENABLE_PUSH, 0, + SETTINGS_INITIAL_WINDOW_SIZE, LOCAL_INITIAL_WINDOW, + SETTINGS_MAX_HEADER_LIST_SIZE, MAX_HEADER_LIST_SIZE, + ) + self.sock.sendall(CONNECTION_PREFACE + encode_frame(SETTINGS, 0, 0, settings_payload)) + self.sock.sendall(encode_frame( + WINDOW_UPDATE, 0, 0, struct.pack("!I", LOCAL_INITIAL_WINDOW - DEFAULT_WINDOW) + )) + self._receive_initial_settings() except Exception: self.close() raise def close(self): self.usable = False - try: - self.sock.close() - except Exception: - pass + sock = getattr(self, "sock", None) + if sock is not None: + try: + sock.close() + except Exception: + pass + self.sock = None def __del__(self): self.close() + def _connection_error(self, message): + self.usable = False + raise H2ProtocolError(message) + + def _receive_initial_settings(self): + ftype, flags, sid, payload = _read_frame(self.sock, self.local_max_frame_size) + if ftype != SETTINGS or sid != 0 or (flags & FLAG_ACK): + self._connection_error("the server connection preface did not start with SETTINGS") + self._handle_settings(flags, sid, payload) + + def _handle_settings(self, flags, sid, payload): + if sid != 0: + self._connection_error("SETTINGS frame used a non-zero stream id") + if flags & FLAG_ACK: + if payload: + self._connection_error("SETTINGS acknowledgement carried a payload") + self._our_settings_acked = True + return + if len(payload) % 6: + self._connection_error("SETTINGS payload length is not a multiple of 6") + data = bytearray(payload) + for pos in range(0, len(data), 6): + setting_id, value = struct.unpack("!HI", bytes(data[pos:pos + 6])) + if setting_id == SETTINGS_ENABLE_PUSH: + if value != 0: + self._connection_error("a server may set SETTINGS_ENABLE_PUSH only to 0") + elif setting_id == SETTINGS_MAX_CONCURRENT_STREAMS: + self.peer_max_concurrent_streams = value + elif setting_id == SETTINGS_INITIAL_WINDOW_SIZE: + if value > MAX_WINDOW: + self._connection_error("invalid SETTINGS_INITIAL_WINDOW_SIZE") + delta = value - self.peer_initial_window + for stream_id in list(self.stream_send_windows): + updated = self.stream_send_windows[stream_id] + delta + if updated > MAX_WINDOW: + self._connection_error("stream send window overflow after SETTINGS") + self.stream_send_windows[stream_id] = updated + self.peer_initial_window = value + elif setting_id == SETTINGS_MAX_FRAME_SIZE: + if not DEFAULT_MAX_FRAME_SIZE <= value <= MAX_FRAME_SIZE: + self._connection_error("invalid SETTINGS_MAX_FRAME_SIZE") + self.peer_max_frame_size = value + elif setting_id == SETTINGS_MAX_HEADER_LIST_SIZE: + self.peer_max_header_list_size = value + elif setting_id == SETTINGS_HEADER_TABLE_SIZE: + # Our encoder never uses the dynamic table, so every value is safe. + pass + self.sock.sendall(encode_frame(SETTINGS, FLAG_ACK, 0, b"")) + + def _handle_ping(self, flags, sid, payload): + if sid != 0 or len(payload) != 8: + self._connection_error("invalid PING frame") + if not (flags & FLAG_ACK): + self.sock.sendall(encode_frame(PING, FLAG_ACK, 0, payload)) + + def _handle_window_update(self, sid, payload): + if len(payload) != 4: + self._connection_error("invalid WINDOW_UPDATE length") + increment = struct.unpack("!I", payload)[0] & MAX_WINDOW + if increment == 0: + self._connection_error("WINDOW_UPDATE increment is zero") + if sid == 0: + if self.conn_send_window > MAX_WINDOW - increment: + self._connection_error("connection flow-control window overflow") + self.conn_send_window += increment + elif sid in self.stream_send_windows: + current = self.stream_send_windows[sid] + if current > MAX_WINDOW - increment: + self._connection_error("stream flow-control window overflow") + self.stream_send_windows[sid] = current + increment + elif sid % 2 == 0 or sid >= self.next_sid: + self._connection_error("WINDOW_UPDATE referenced an idle stream") + + def _allocate_stream_ids(self, count): + if count <= 0: + raise ValueError("stream count must be positive") + last = self.next_sid + 2 * (count - 1) + if last > MAX_STREAM_ID: + self.usable = False + raise H2Error("HTTP/2 stream-id space exhausted") + sids = [self.next_sid + 2 * index for index in range(count)] + self.next_sid += 2 * count + if self.next_sid > MAX_STREAM_ID: + self.usable = False + return sids + + def _send_headers(self, sid, request_headers, end_stream): + if self.peer_max_header_list_size is not None: + if _field_section_size(request_headers) > self.peer_max_header_list_size: + raise H2Error("request header list exceeds peer-advertised limit") + block = Encoder().encode(request_headers) + return b"".join(_header_frames(sid, block, self.peer_max_frame_size, end_stream)) + + def _finish_response(self, state): + actual = len(state.body) + if state.body_forbidden and actual: + self._connection_error("response semantics forbid a message body") + if (not state.body_forbidden and state.expected_content_length is not None and + actual != state.expected_content_length): + self._connection_error( + "response content-length mismatch (expected %d, received %d)" % + (state.expected_content_length, actual) + ) + state.complete = True + + def _complete_header_block(self, state): + try: + headers = self.dec.decode(bytes(state.block)) + except H2CompressionError: + self.usable = False + raise + end_stream = state.block_end_stream + state.block = None + state.block_end_stream = False + + if state.headers is None: + status, content_length = _validate_response_field_section(headers, trailers=False) + if 100 <= status < 200: + if end_stream: + self._connection_error("informational response ended the stream") + if content_length is not None: + self._connection_error("informational response carried content-length") + state.informational_count += 1 + state.informational_bytes += _field_section_size(headers) + if state.informational_count > MAX_INFORMATIONAL_RESPONSES: + self.usable = False + raise H2Error("too many informational responses") + if state.informational_bytes > MAX_INFORMATIONAL_HEADER_BYTES: + self.usable = False + raise H2Error("informational response headers exceed configured limit") + return + state.headers = headers + state.status = status + state.expected_content_length = content_length + state.body_forbidden = state.request_method == b"HEAD" or status in (204, 205, 304) + if status == 204 and content_length is not None: + self._connection_error("204 response must not carry content-length") + if status == 205 and content_length not in (None, 0): + self._connection_error("205 response content-length must be zero") + if state.request_method == b"CONNECT" and 200 <= status < 300 and content_length is not None: + self._connection_error("successful CONNECT response must not carry content-length") + if (not state.body_forbidden and content_length is not None and + content_length > MAX_RESPONSE_SIZE): + self.usable = False + raise H2Error("declared response body exceeds configured limit") + if end_stream: + self._finish_response(state) + else: + _validate_response_field_section(headers, trailers=True) + if not end_stream: + self._connection_error("trailing field section did not end the stream") + if _field_section_size(state.headers) + _field_section_size(headers) > MAX_HEADER_LIST_SIZE: + self.usable = False + raise H2Error("aggregate response headers exceed configured limit") + state.trailers.extend(headers) + state.headers.extend(headers) + self._finish_response(state) + + def _process_header_frame(self, ftype, flags, sid, payload, states, continuation_sid): + if continuation_sid is not None: + if ftype != CONTINUATION or sid != continuation_sid: + self._connection_error("frame interleaved inside a CONTINUATION sequence") + elif ftype == CONTINUATION: + self._connection_error("unexpected CONTINUATION frame") + + state = states.get(sid) + if state is None: + self._connection_error("header frame for an unknown stream") + if state.complete: + self._connection_error("header frame received after END_STREAM") + + if ftype == HEADERS: + if state.block is not None: + self._connection_error("new HEADERS before the prior field section completed") + state.block = bytearray() + state.block_end_stream = bool(flags & FLAG_END_STREAM) + state.block.extend(_extract_header_fragment(ftype, flags, payload)) + if len(state.block) > MAX_HEADER_BLOCK_SIZE: + self.usable = False + raise H2Error("compressed response header block exceeds configured limit") + + if flags & FLAG_END_HEADERS: + self._complete_header_block(state) + return None + return sid + + def _process_data_frame(self, flags, sid, payload, states): + state = states.get(sid) + if state is None: + self._connection_error("DATA frame for an unknown stream") + if state.headers is None: + self._connection_error("DATA received before final response headers") + if state.complete: + self._connection_error("DATA received after END_STREAM") + flow_len = len(payload) + if flow_len > self.conn_recv_window: + self._connection_error("peer exceeded the connection receive window") + if flow_len > state.recv_window: + self._connection_error("peer exceeded the stream receive window") + self.conn_recv_window -= flow_len + state.recv_window -= flow_len + data = _extract_data(flags, payload) + if state.body_forbidden and data: + self._connection_error("response semantics forbid a message body") + if (state.expected_content_length is not None and not state.body_forbidden and + len(state.body) + len(data) > state.expected_content_length): + self._connection_error("response body exceeds declared content-length") + if len(state.body) + len(data) > MAX_RESPONSE_SIZE: + self.usable = False + raise H2Error("response body exceeds configured limit") + state.body.extend(data) + + end_stream = bool(flags & FLAG_END_STREAM) + if flow_len: + if end_stream: + try: + self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, 0, struct.pack("!I", flow_len))) + self.conn_recv_window += flow_len + except (socket.error, ssl.SSLError, IOError): + self.usable = False + else: + self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, 0, struct.pack("!I", flow_len))) + self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, sid, struct.pack("!I", flow_len))) + self.conn_recv_window += flow_len + state.recv_window += flow_len + if end_stream: + self._finish_response(state) + + def _process_frame(self, frame, states, continuation_sid): + ftype, flags, sid, payload = frame + if continuation_sid is not None and not (ftype == CONTINUATION and sid == continuation_sid): + self._connection_error("non-CONTINUATION frame interleaved in a field section") + + if ftype == SETTINGS: + self._handle_settings(flags, sid, payload) + elif ftype == PING: + self._handle_ping(flags, sid, payload) + elif ftype == WINDOW_UPDATE: + self._handle_window_update(sid, payload) + elif ftype == GOAWAY: + if sid != 0 or len(payload) < 8: + self._connection_error("invalid GOAWAY frame") + self.usable = False + last_sid = struct.unpack("!I", payload[:4])[0] & MAX_STREAM_ID + active = sorted(states) + unprocessed = [stream_id for stream_id in active if stream_id > last_sid] + if len(unprocessed) == len(active): + raise _UnprocessedStream("GOAWAY confirms active stream(s) were not processed") + if unprocessed: + raise H2Error("GOAWAY partially processed a multi-stream exchange") + elif ftype == RST_STREAM: + if sid == 0 or len(payload) != 4: + self._connection_error("invalid RST_STREAM frame") + if sid in states: + error_code = struct.unpack("!I", payload)[0] + if len(states) == 1 and error_code == REFUSED_STREAM: + raise _UnprocessedStream("peer refused the stream before processing") + raise H2Error("stream %d reset by peer (error %d)" % (sid, error_code)) + if sid % 2 == 0 or sid >= self.next_sid: + self._connection_error("RST_STREAM referenced an idle stream") + elif ftype == PUSH_PROMISE: + self._connection_error("server push received despite SETTINGS_ENABLE_PUSH=0") + elif ftype in (HEADERS, CONTINUATION): + continuation_sid = self._process_header_frame( + ftype, flags, sid, payload, states, continuation_sid + ) + elif ftype == DATA: + if sid == 0: + self._connection_error("DATA frame used stream 0") + self._process_data_frame(flags, sid, payload, states) + elif ftype == PRIORITY: + if sid == 0 or len(payload) != 5: + self._connection_error("invalid PRIORITY frame") + # Unknown extension frames are ignored, as required by HTTP/2. + return continuation_sid + def exchange(self, method, path, authority, headers, body, timeout): if not self.usable: - raise IOError("HTTP/2 connection no longer usable") - - sid = self.next_sid - self.next_sid += 2 - if self.next_sid >= BIG_WINDOW: # stream-id space nearly exhausted -> retire after this - self.usable = False + raise H2Error("HTTP/2 connection is not usable") + if self.peer_max_concurrent_streams == 0: + raise H2Error("peer currently permits no new streams") self.sock.settimeout(timeout) + body = b"" if body is None else _as_bytes(body) + method = _as_bytes(method) + sid = self._allocate_stream_ids(1)[0] + self.stream_send_windows[sid] = self.peer_initial_window + try: + state = _ResponseState(sid, method) + states = {sid: state} + request_headers = _request_header_list(method, path, authority, headers, body) + self.sock.sendall(self._send_headers(sid, request_headers, not body)) + + body_pos = 0 + continuation_sid = None + frames_seen = 0 + while not state.complete: + while body_pos < len(body): + available = min( + self.peer_max_frame_size, + self.conn_send_window, + self.stream_send_windows[sid], + len(body) - body_pos, + ) + if available <= 0: + break + chunk = body[body_pos:body_pos + available] + body_pos += len(chunk) + flags = FLAG_END_STREAM if body_pos == len(body) else 0 + self.sock.sendall(encode_frame(DATA, flags, sid, chunk)) + self.conn_send_window -= len(chunk) + self.stream_send_windows[sid] -= len(chunk) + + if state.complete: + break + frame = _read_frame(self.sock, self.local_max_frame_size) + frames_seen += 1 + if frames_seen > MAX_RESPONSE_FRAMES: + self.usable = False + raise H2Error("response exceeded the frame-count limit") + continuation_sid = self._process_frame(frame, states, continuation_sid) - req = [(b":method", _tob(method)), (b":scheme", b"https"), (b":path", _tob(path)), (b":authority", _tob(authority))] - for k, v in (headers or {}).items(): - req.append((_tob(k).lower(), _tob(v))) - hblock = Encoder().encode(req) - self.sock.sendall(encode_frame(HEADERS, FLAG_END_HEADERS | (0 if body else FLAG_END_STREAM), sid, hblock)) - if body: - self.sock.sendall(encode_frame(DATA, FLAG_END_STREAM, sid, _tob(body))) - - header_block, resp_headers, resp_body, done = b"", None, bytearray(), False - while not done: - ftype, flags, fsid, payload = _read_frame(self.sock) - if ftype == SETTINGS: - if not (flags & FLAG_ACK): - self.sock.sendall(encode_frame(SETTINGS, FLAG_ACK, 0, b"")) - elif ftype == PING: - if not (flags & FLAG_ACK): - self.sock.sendall(encode_frame(PING, FLAG_ACK, 0, payload)) - elif ftype == PUSH_PROMISE: - self.usable = False # we advertised ENABLE_PUSH=0; a push would desync HPACK - raise _UnprocessedStream("unexpected PUSH_PROMISE despite SETTINGS_ENABLE_PUSH=0") - elif ftype == GOAWAY: - self.usable = False # server won't accept new streams -> retire connection - last_sid = (struct.unpack("!I", payload[4:8])[0] & 0x7fffffff) if len(payload) >= 8 else 0 - if sid > last_sid: # our stream was not processed -> safe to retry fresh - raise _UnprocessedStream("GOAWAY (last stream %d) before stream %d was processed" % (last_sid, sid)) - elif ftype == RST_STREAM and fsid == sid: + if continuation_sid is not None: + self._connection_error("stream ended with an incomplete field section") + if body_pos < len(body): self.usable = False - raise IOError("stream reset by server (error %d)" % struct.unpack("!I", payload[:4])[0]) - elif ftype in (HEADERS, CONTINUATION) and fsid == sid: - p = payload - if ftype == HEADERS: - if flags & FLAG_PADDED: - p = p[1:len(p) - bytearray(payload)[0]] - if flags & FLAG_PRIORITY: - p = p[5:] - header_block += p - if len(header_block) > MAX_RESPONSE_SIZE: # hostile/endless header block -> bail rather than OOM - self.usable = False - raise IOError("oversized HTTP/2 header block") - if flags & FLAG_END_HEADERS: - resp_headers = self.dec.decode(header_block) - if flags & FLAG_END_STREAM: - done = True - elif ftype == DATA and fsid == sid: - p = payload - if flags & FLAG_PADDED: - p = p[1:len(p) - bytearray(payload)[0]] - resp_body += p - if len(resp_body) > MAX_RESPONSE_SIZE: # cap like the HTTP/1.1 path; stop reading and retire the - self.usable = False # connection (leftover frames abandoned) instead of OOM - break - if payload: # replenish stream + connection windows - self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, sid, struct.pack("!I", len(payload)))) - self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, 0, struct.pack("!I", len(payload)))) - if flags & FLAG_END_STREAM: - done = True - status = None - for n, v in (resp_headers or []): - if _tob(n) == b":status": - status = int(v) - break - return status, resp_headers, bytes(resp_body) + try: + self.sock.sendall(encode_frame(RST_STREAM, 0, sid, struct.pack("!I", CANCEL))) + except Exception: + pass + if state.status is None: + self._connection_error("stream ended without final response headers") + return state.status, state.headers, bytes(state.body) + finally: + self.stream_send_windows.pop(sid, None) def exchange_pair(self, requests, timeout): - """Timeless-timing primitive (Van Goethem et al., USENIX Security 2020). Send TWO requests - multiplexed and COALESCED into a single TCP write, then read frames until BOTH streams reach - END_STREAM, recording the order in which they finished. Because both requests ride the same - packet on the same connection, network jitter hits them equally and cancels - only the server's - relative processing time decides which END_STREAM lands first, so a sub-millisecond server-side - delta is readable that absolute wall-clock timing (drowned by jitter) cannot resolve. - - `requests` is a 2-list of dicts {method, path, authority, headers, body}. Returns - (finish_order, results) where finish_order is the list of stream ids in completion order and - results maps sid -> (status, headers, body). Requires the server to process the two streams - CONCURRENTLY; a serializing front-proxy defeats it (callers must calibrate - see h2_timeless_probe).""" + """Send exactly two small requests in one application-level write. + + A single sendall() reduces userspace scheduling skew, but TCP/TLS may still split + the bytes into multiple records or packets; packet-level coalescing is not guaranteed. + """ if not self.usable: - raise IOError("HTTP/2 connection no longer usable") + raise H2Error("HTTP/2 connection is not usable") + requests = list(requests) + if len(requests) != 2: + raise ValueError("exchange_pair requires exactly two requests") + if self.peer_max_concurrent_streams is not None and self.peer_max_concurrent_streams < 2: + raise H2Error("peer does not permit two concurrent streams") self.sock.settimeout(timeout) - - sids, out = [], b"" - for r in requests: - sid = self.next_sid - self.next_sid += 2 - sids.append(sid) - req = [(b":method", _tob(r.get("method", "GET"))), (b":scheme", b"https"), - (b":path", _tob(r["path"])), (b":authority", _tob(r.get("authority") or self.host))] - for k, v in (r.get("headers") or {}).items(): - req.append((_tob(k).lower(), _tob(v))) - body = r.get("body") - out += encode_frame(HEADERS, FLAG_END_HEADERS | (0 if body else FLAG_END_STREAM), sid, Encoder().encode(req)) - if body: - out += encode_frame(DATA, FLAG_END_STREAM, sid, _tob(body)) - if self.next_sid >= BIG_WINDOW: - self.usable = False - self.sock.sendall(out) # THE crux: one write -> one TCP segment -> simultaneous arrival - - state = dict((sid, {"hb": b"", "headers": None, "body": bytearray()}) for sid in sids) - finish_order = [] - remaining = set(sids) - while remaining: - ftype, flags, fsid, payload = _read_frame(self.sock) - if ftype == SETTINGS: - if not (flags & FLAG_ACK): - self.sock.sendall(encode_frame(SETTINGS, FLAG_ACK, 0, b"")) - elif ftype == PING: - if not (flags & FLAG_ACK): - self.sock.sendall(encode_frame(PING, FLAG_ACK, 0, payload)) - elif ftype == PUSH_PROMISE: - self.usable = False # we advertised ENABLE_PUSH=0; a push would desync HPACK - raise _UnprocessedStream("unexpected PUSH_PROMISE despite SETTINGS_ENABLE_PUSH=0") - elif ftype == GOAWAY: - # Routine on a long-lived connection (server retires it after its per-connection request cap). - # The pair did not complete cleanly, so it must be re-sent on a fresh connection; flag it - # retry-safe (idempotent boolean read) rather than crashing the extraction. - self.usable = False - raise _UnprocessedStream("GOAWAY during timeless pair") - elif ftype == RST_STREAM and fsid in state: - self.usable = False - raise _UnprocessedStream("stream reset during timeless pair") - elif ftype in (HEADERS, CONTINUATION) and fsid in state: - p = payload - if ftype == HEADERS: - if flags & FLAG_PADDED: - p = p[1:len(p) - bytearray(payload)[0]] - if flags & FLAG_PRIORITY: - p = p[5:] - state[fsid]["hb"] += p - if len(state[fsid]["hb"]) > MAX_RESPONSE_SIZE: - self.usable = False - raise IOError("oversized HTTP/2 header block during timeless pair") - if flags & FLAG_END_HEADERS: - state[fsid]["headers"] = self.dec.decode(state[fsid]["hb"]) - if flags & FLAG_END_STREAM and fsid in remaining: - finish_order.append(fsid); remaining.discard(fsid) - elif ftype == DATA and fsid in state: - p = payload - if flags & FLAG_PADDED: - p = p[1:len(p) - bytearray(payload)[0]] - state[fsid]["body"] += p - if len(state[fsid]["body"]) > MAX_RESPONSE_SIZE: # cap the buffered body (mirrors exchange()) - self.usable = False - raise IOError("oversized response during timeless pair") - if payload: - self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, fsid, struct.pack("!I", len(payload)))) - self.sock.sendall(encode_frame(WINDOW_UPDATE, 0, 0, struct.pack("!I", len(payload)))) - if flags & FLAG_END_STREAM and fsid in remaining: - finish_order.append(fsid); remaining.discard(fsid) - - results = {} + sids = self._allocate_stream_ids(2) for sid in sids: - status = None - for n, v in (state[sid]["headers"] or []): - if _tob(n) == b":status": - status = int(v); break - results[sid] = (status, state[sid]["headers"], bytes(state[sid]["body"])) - return finish_order, results + self.stream_send_windows[sid] = self.peer_initial_window + try: + states = {} + outbound = [] + total_body = 0 + for sid, request in zip(sids, requests): + body = b"" if request.get("body") is None else _as_bytes(request.get("body")) + method = _as_bytes(request.get("method", "GET")) + states[sid] = _ResponseState(sid, method) + if len(body) > self.peer_initial_window: + raise H2Error("timeless-pair body exceeds the peer stream window") + total_body += len(body) + request_headers = _request_header_list( + method, + request["path"], + request.get("authority") or _authority_for_host(self.host, self.port), + request.get("headers"), + body, + ) + outbound.append(self._send_headers(sid, request_headers, not body)) + if body: + outbound.extend(_data_frames(sid, body, self.peer_max_frame_size)) + self.stream_send_windows[sid] -= len(body) + if total_body > self.conn_send_window: + raise H2Error("timeless-pair bodies exceed the connection flow-control window") + self.conn_send_window -= total_body + self.sock.sendall(b"".join(outbound)) + + finish_order = [] + continuation_sid = None + frames_seen = 0 + while len(finish_order) < 2: + before = dict((sid, states[sid].complete) for sid in sids) + frame = _read_frame(self.sock, self.local_max_frame_size) + frames_seen += 1 + if frames_seen > MAX_RESPONSE_FRAMES: + self.usable = False + raise H2Error("timeless pair exceeded the frame-count limit") + continuation_sid = self._process_frame(frame, states, continuation_sid) + for sid in sids: + if states[sid].complete and not before[sid] and sid not in finish_order: + finish_order.append(sid) + if continuation_sid is not None: + self._connection_error("pair ended with an incomplete field section") + results = {} + for sid in sids: + state = states[sid] + if state.status is None: + self._connection_error("pair stream ended without final response headers") + results[sid] = (state.status, state.headers, bytes(state.body)) + return finish_order, results + finally: + for sid in sids: + self.stream_send_windows.pop(sid, None) -# Thread-local pool: one live connection per (host, port, proxy) per thread. Mirrors keepalive.py's model -# (one connection per host per thread) so streams never interleave across threads and time-based -# measurements stay clean. _h2_pool = threading.local() -def _pooledExchange(host, port, proxy, method, path, authority, headers, body, timeout): + +def _pool_key(host, port, proxy, verify, ssl_context): + proxy_key = tuple(proxy) if isinstance(proxy, list) else proxy + context_key = ssl_context if ssl_context is not None else bool(verify) + return _idna_host(host), port, proxy_key, context_key + + +def _get_pool(): pool = getattr(_h2_pool, "connections", None) - if pool is None: - pool = _h2_pool.connections = {} - key = (host, port, proxy) + if pool is None or not isinstance(pool, OrderedDict): + pool = OrderedDict() + _h2_pool.connections = pool + return pool + + +def _pool_put(pool, key, conn): + old = pool.pop(key, None) + if old is not None and old is not conn: + old.close() + pool[key] = conn + while len(pool) > H2_POOL_LIMIT: + _, stale = pool.popitem(last=False) + stale.close() + + +def close_pooled_connections(): + pool = _get_pool() + for conn in list(pool.values()): + conn.close() + pool.clear() - conn = pool.get(key) + +def _retryable_after_stale_connection(method): + return _as_bytes(method) in (b"GET", b"HEAD") + + +def _pooledExchange(host, port, proxy, method, path, authority, headers, body, timeout, + verify=False, ssl_context=None): + pool = _get_pool() + key = _pool_key(host, port, proxy, verify, ssl_context) + conn = pool.pop(key, None) reused = conn is not None and conn.usable - if not reused: + if conn is None or not conn.usable: if conn is not None: conn.close() - conn = pool[key] = _H2Connection(host, port, proxy, timeout) - + conn = _H2Connection(host, port, proxy, timeout, verify=verify, ssl_context=ssl_context) + _pool_put(pool, key, conn) try: result = conn.exchange(method, path, authority, headers, body, timeout) - except _UnprocessedStream: # explicitly not processed -> always safe to retry fresh - conn.close(); pool.pop(key, None) - conn = pool[key] = _H2Connection(host, port, proxy, timeout) - result = conn.exchange(method, path, authority, headers, body, timeout) - except (socket.error, ssl.SSLError, IOError): - conn.close(); pool.pop(key, None) - if reused: # stale keep-alive socket (server closed idle conn) -> reopen once - conn = pool[key] = _H2Connection(host, port, proxy, timeout) - result = conn.exchange(method, path, authority, headers, body, timeout) - else: + except _UnprocessedStream: + conn.close() + pool.pop(key, None) + replacement = _H2Connection(host, port, proxy, timeout, verify=verify, ssl_context=ssl_context) + _pool_put(pool, key, replacement) + try: + result = replacement.exchange(method, path, authority, headers, body, timeout) + except Exception: + replacement.close() + pool.pop(key, None) raise - if not conn.usable: # GOAWAY / id-exhaustion mid-exchange -> don't keep it pooled - conn.close(); pool.pop(key, None) + conn = replacement + except (socket.error, ssl.SSLError, H2TransportError): + conn.close() + pool.pop(key, None) + if not (reused and _retryable_after_stale_connection(method)): + raise + replacement = _H2Connection(host, port, proxy, timeout, verify=verify, ssl_context=ssl_context) + _pool_put(pool, key, replacement) + try: + result = replacement.exchange(method, path, authority, headers, body, timeout) + except Exception: + replacement.close() + pool.pop(key, None) + raise + conn = replacement + except (H2Error, IOError): + conn.close() + pool.pop(key, None) + raise + if not conn.usable: + conn.close() + pool.pop(key, None) return result -def h2_request(host, port=443, method="GET", path="/", authority=None, headers=None, body=None, timeout=30, proxy=None): - """One-shot request on a throwaway connection (kept for direct/back-compat callers; the engine path - goes through open_url -> the reusing pool).""" - conn = _H2Connection(host, port, proxy, timeout) + +def h2_request(host, port=443, method="GET", path="/", authority=None, headers=None, body=None, + timeout=30, proxy=None, verify=False, ssl_context=None): + conn = _H2Connection(host, port, proxy, timeout, verify=verify, ssl_context=ssl_context) try: - return conn.exchange(method, path, authority or host, headers, body, timeout) + return conn.exchange(method, path, authority or _authority_for_host(host, port), headers, body, timeout) finally: conn.close() class H2Response(object): - """A urllib-response-compatible wrapper around a native HTTP/2 response, so the rest of sqlmap's - request pipeline can consume it exactly like a urllib response (code/msg/info()/read()/geturl()). - - >>> r = H2Response('https://x/', 200, [(b':status', b'200'), (b'content-type', b'text/html')], b'body') - >>> (r.code, r.msg, r.read() == b'body', r.geturl()) - (200, 'OK', True, 'https://x/') - >>> ':status' in r.info() - False - """ + """A small urllib-response-compatible wrapper for a buffered HTTP/2 response.""" def __init__(self, url, status, headers, body): self.url = url @@ -723,15 +1490,12 @@ def __init__(self, url, status, headers, body): self._body = body self._offset = 0 self._info = _Message() - for name, value in (headers or []): + for name, value in headers or []: name = name.decode("latin-1") if isinstance(name, bytes) else name value = value.decode("latin-1") if isinstance(value, bytes) else value - if not name.startswith(":"): # drop HTTP/2 pseudo-headers (:status etc.) + if not name.startswith(":"): self._info[name] = value - # expose a mimetools.Message-style '.headers' list so patchHeaders() treats this object - # uniformly across Python 2/3 (email.message.Message lacks it, and Python 2 iteration over a - # bare Message falls back to integer indexing) - self._info.headers = ["%s: %s\r\n" % (name, value) for (name, value) in self._info.items()] + self._info.headers = ["%s: %s\r\n" % item for item in self._info.items()] def info(self): return self._info @@ -739,8 +1503,11 @@ def info(self): def geturl(self): return self.url + def getcode(self): + return self.code + def read(self, amt=None): - if amt is None: + if amt is None or amt < 0: data = self._body[self._offset:] self._offset = len(self._body) else: @@ -751,38 +1518,93 @@ def read(self, amt=None): def close(self): pass + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + return False + + +def _headers_without(headers, names): + names = set(names) + result = [] + for key, value in _iter_header_items(headers): + name = _as_bytes(key).lower() + if name not in names: + result.append((key, value)) + return result + + +def _url_port(parts): + try: + return parts.port or 443 + except ValueError as ex: + raise H2Error("invalid URL port: %s" % ex) + -def open_url(url, method="GET", headers=None, body=None, timeout=30, follow_redirects=True, max_redirects=10, proxy=None): - """Fetch url over native HTTP/2 (https only), following redirects like a browser (mirroring the - previous httpx follow_redirects=True), and return an H2Response. Raises IOError on a transport or - ALPN-negotiation failure. Connection-level and h2-forbidden request headers are stripped.""" - forbidden = ("host", "connection", "keep-alive", "proxy-connection", "transfer-encoding", "upgrade", "content-length") - req_headers = {} - for key in (headers or {}): - name = key.decode("latin-1") if isinstance(key, bytes) else key - if name.lower() not in forbidden: - req_headers[key] = headers[key] +def _origin_tuple(parts): + host = _idna_host(parts.hostname or "").lower() + return parts.scheme.lower(), host, _url_port(parts) - for _ in range(max_redirects + 1): + +def _quote_target_component(value, safe): + if isinstance(value, text_type): + try: + value.encode("ascii") + except UnicodeEncodeError: + value = value.encode("utf-8") + return quote(value, safe=safe) + + +def open_url(url, method="GET", headers=None, body=None, timeout=30, follow_redirects=True, + max_redirects=10, proxy=None, verify=False, ssl_context=None): + req_headers = headers or {} + current_method = method + current_body = body + + for redirect_count in range(max_redirects + 1): parts = urlsplit(url) - if parts.scheme != "https": - raise IOError("native HTTP/2 client supports 'https://' targets only (got %r)" % parts.scheme) - path = parts.path or "/" + if parts.scheme.lower() != "https": + raise H2Error("native HTTP/2 client supports only https URLs") + if not parts.hostname: + raise H2Error("URL has no hostname") + host = _idna_host(parts.hostname) + port = _url_port(parts) + path = _quote_target_component(parts.path or "/", "/%:@!$&'()*+,;=-._~") if parts.query: - path += "?" + parts.query - status, resp_headers, resp_body = _pooledExchange(parts.hostname, parts.port or 443, proxy, method, path, - parts.netloc.split("@")[-1], req_headers, body, timeout) - if follow_redirects and status in REDIRECT_CODES: - location = None - for name, value in (resp_headers or []): - if (name.decode("latin-1") if isinstance(name, bytes) else name).lower() == "location": - location = value.decode("latin-1") if isinstance(value, bytes) else value - break - if location: - url = urljoin(url, location) - if status in (301, 302, 303): # per RFC 7231, these degrade to GET - method, body = "GET", None - continue - return H2Response(url, status, resp_headers, resp_body) + path += "?" + _quote_target_component(parts.query, "/?%:@!$&'()*+,;=-._~") + authority = _authority_for_host(host, port) + status, response_headers, response_body = _pooledExchange( + host, port, proxy, current_method, path, authority, + req_headers, current_body, timeout, verify=verify, ssl_context=ssl_context + ) + if not (follow_redirects and status in REDIRECT_CODES): + return H2Response(url, status, response_headers, response_body) + + location = None + for name, value in response_headers or []: + if _as_bytes(name).lower() == b"location": + location = value.decode("latin-1") if isinstance(value, bytes) else value + break + if not location: + return H2Response(url, status, response_headers, response_body) + if redirect_count == max_redirects: + break - raise IOError("too many HTTP/2 redirects") + new_url = urljoin(url, location) + old_parts = parts + new_parts = urlsplit(new_url) + if _origin_tuple(old_parts) != _origin_tuple(new_parts): + req_headers = _headers_without(req_headers, _SENSITIVE_REDIRECT_FIELDS) + + method_bytes = _as_bytes(current_method) + change_to_get = status == 303 and method_bytes != b"HEAD" + change_to_get = change_to_get or (status in (301, 302) and method_bytes == b"POST") + if change_to_get: + current_method = "GET" + current_body = None + req_headers = _headers_without(req_headers, _CONTENT_REDIRECT_FIELDS) + url = new_url + + raise H2Error("too many HTTP/2 redirects") diff --git a/tests/test_http2.py b/tests/test_http2.py index 7c762648176..7ac0b225554 100644 --- a/tests/test_http2.py +++ b/tests/test_http2.py @@ -4,31 +4,46 @@ Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org) See the file 'LICENSE' for copying permission -Unit coverage for the PURE (network-free) parts of the native HTTP/2 client in -lib/request/http2.py: the RFC 7540 frame codec, the RFC 7541 HPACK integer / -Huffman / string primitives, the HPACK Decoder/Encoder (static + dynamic table), -and the urllib-compatible H2Response wrapper. +Coverage for the native HTTP/2 client in lib/request/http2.py: frame and HPACK +codecs, request/response validation, connection state, flow control, retries, +URL handling, TLS policy, and the urllib-compatible H2Response wrapper. -Nothing here opens a socket or negotiates TLS - only the deterministic codecs and -the response adapter are exercised. Known vectors are the canonical RFC 7541 -examples; everything else is a round-trip / invariant check. +Most tests are deterministic and network-free. When python-hyper/h2 is installed, +two optional socketpair tests exercise an end-to-end local HTTP/2 peer; they never +use the external network. Known vectors are the canonical RFC 7541 examples. stdlib unittest only (no pytest / no pip); works on Python 2.7 and 3.x. """ import binascii import os +import socket +import ssl +import struct import sys +import threading import unittest sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from _testutils import bootstrap bootstrap() +from lib.request import http2 as _http2 from lib.request.http2 import ( Decoder, Encoder, + H2CompressionError, + H2Error, + H2ProtocolError, + H2TransportError, H2Response, + MAX_INFORMATIONAL_RESPONSES, + DEFAULT_MAX_FRAME_SIZE, + DEFAULT_WINDOW, + LOCAL_INITIAL_WINDOW, + SETTINGS_ENABLE_PUSH, + GOAWAY, + CONTINUATION, REDIRECT_CODES, STATIC_LEN, STATIC_TABLE, @@ -44,14 +59,91 @@ encode_string, huffman_decode, huffman_encode, + _H2Connection, + _authority_for_host, + _make_ssl_context, + _parse_content_length, + _request_header_list, + _validate_response_field_section, ) +try: + from h2.config import H2Configuration + from h2.connection import H2Connection as ReferenceH2Connection + from h2.events import DataReceived, RequestReceived, StreamEnded + HAVE_H2 = True +except ImportError: + HAVE_H2 = False + def _b(*ints): # build a bytes object from ints (identical on Python 2 and 3) return bytes(bytearray(ints)) +class _FakeSocket(object): + def __init__(self, incoming=b""): + self.incoming = bytearray(incoming) + self.sent = [] + self.timeout = None + self.closed = False + + def recv(self, size): + if not self.incoming: + return b"" + chunk = bytes(self.incoming[:size]) + del self.incoming[:size] + return chunk + + def sendall(self, data): + self.sent.append(bytes(data)) + + def settimeout(self, timeout): + self.timeout = timeout + + def close(self): + self.closed = True + + +def _bare_connection(incoming=b""): + conn = object.__new__(_H2Connection) + conn.host = "example.test" + conn.port = 443 + conn.proxy = None + conn.dec = Decoder() + conn.next_sid = 1 + conn.usable = True + conn.sock = _FakeSocket(incoming) + conn.peer_max_frame_size = DEFAULT_MAX_FRAME_SIZE + conn.peer_initial_window = DEFAULT_WINDOW + conn.peer_max_concurrent_streams = None + conn.peer_max_header_list_size = None + conn.conn_send_window = DEFAULT_WINDOW + conn.conn_recv_window = LOCAL_INITIAL_WINDOW + conn.stream_send_windows = {} + conn.local_max_frame_size = DEFAULT_MAX_FRAME_SIZE + conn._our_settings_acked = False + return conn + + +def _response_block(status=b"200", extra=()): + return Encoder().encode([(b":status", status)] + list(extra)) + + +def _frames_from_bytes(data): + result = [] + pos = 0 + while pos < len(data): + length, ftype, flags, sid = decode_frame_header(data[pos:pos + 9]) + pos += 9 + payload = data[pos:pos + length] + pos += length + result.append((ftype, flags, sid, payload)) + if pos != len(data): + raise AssertionError("trailing frame bytes") + return result + + class TestFrameCodec(unittest.TestCase): def test_roundtrip(self): header = encode_frame(HEADERS, FLAG_END_HEADERS, 1, b"abc")[:9] @@ -62,9 +154,9 @@ def test_payload_is_appended_verbatim(self): self.assertEqual(frame[9:], b"hello") def test_reserved_stream_bit_is_masked(self): - # the high (reserved) bit of the 31-bit stream id must be dropped on both ends - header = encode_frame(DATA, 0, 0x80000001, b"")[:9] - self.assertEqual(decode_frame_header(header), (0, DATA, 0, 1)) + # the high (reserved) bit of the 31-bit stream id is rejected + with self.assertRaises(ValueError): + encode_frame(DATA, 0, 0x80000001, b"") def test_zero_length_payload(self): header = encode_frame(DATA, FLAG_END_STREAM, 1, b"")[:9] @@ -107,6 +199,18 @@ def test_first_byte_bits_preserved(self): # a caller-supplied opcode in the high bits must survive a small value self.assertEqual(bytearray(encode_integer(5, 7, 0x80))[0], 0x80 | 5) + def test_negative_value_rejected(self): + with self.assertRaises(ValueError): + encode_integer(-1, 5) + + def test_truncated_multibyte_integer_rejected(self): + with self.assertRaises(H2CompressionError): + decode_integer(bytearray([31, 0x80]), 0, 5) + + def test_overlong_integer_rejected(self): + with self.assertRaises(H2CompressionError): + decode_integer(bytearray([31, 0x80, 0x80, 0x80, 0x80, 0x80, 0]), 0, 5) + class TestHuffman(unittest.TestCase): def test_known_vector_www_example_com(self): @@ -129,9 +233,13 @@ def test_shrinks_typical_text(self): def test_padding_too_long_rejected(self): # 0xfe walks eight 1-bits into a long (unterminated) code -> more than a byte of padding - with self.assertRaises(ValueError): + with self.assertRaises(H2CompressionError): huffman_decode(_b(0xFE)) + def test_non_eos_padding_rejected(self): + with self.assertRaises(H2CompressionError): + huffman_decode(b"\x00") + class TestStringCoding(unittest.TestCase): def test_huffman_branch_roundtrip(self): @@ -149,6 +257,10 @@ def test_disable_huffman(self): self.assertFalse(bytearray(encoded)[0] & 0x80) self.assertEqual(decode_string(bytearray(encoded), 0), (b"abc", len(encoded))) + def test_truncated_string_rejected(self): + with self.assertRaises(H2CompressionError): + decode_string(bytearray(b"\x03ab"), 0) + class TestHpackDecoder(unittest.TestCase): def test_indexed_static_entries(self): @@ -165,11 +277,11 @@ def test_static_lookup_bounds(self): self.assertEqual(d._get(STATIC_LEN), STATIC_TABLE[-1]) def test_index_zero_rejected(self): - with self.assertRaises(ValueError): + with self.assertRaises(H2CompressionError): Decoder()._get(0) def test_index_out_of_range_rejected(self): - with self.assertRaises(ValueError): + with self.assertRaises(H2CompressionError): Decoder()._get(STATIC_LEN + 1) # no dynamic entries yet def test_literal_incremental_indexing_populates_dynamic_table(self): @@ -202,6 +314,21 @@ def test_dynamic_size_update_clears(self): self.assertEqual(d.max_size, 0) self.assertEqual(d.dynamic, []) + def test_oversized_dynamic_size_update_rejected(self): + update = bytes(encode_integer(4097, 5, 0x20)) + with self.assertRaises(H2CompressionError): + Decoder().decode(update) + + def test_late_dynamic_size_update_rejected(self): + block = _b(0x82) + bytes(encode_integer(0, 5, 0x20)) + with self.assertRaises(H2CompressionError): + Decoder().decode(block) + + def test_decoded_header_list_limit(self): + block = bytearray([0x00]) + encode_string(b"name") + encode_string(b"value") + with self.assertRaises(H2CompressionError): + Decoder(max_header_list_size=1).decode(bytes(block)) + class TestHpackEncoderRoundTrip(unittest.TestCase): def test_roundtrip_through_decoder(self): @@ -267,6 +394,500 @@ def test_mimetools_style_headers_list(self): def test_close_is_noop(self): self.assertIsNone(self._make().close()) + def test_negative_read_reads_all_remaining(self): + r = self._make(body=b"abcdef") + self.assertEqual(r.read(2), b"ab") + self.assertEqual(r.read(-1), b"cdef") + self.assertEqual(r.read(), b"") + + +class TestRequestValidation(unittest.TestCase): + def test_method_case_is_preserved(self): + headers = _request_header_list("foo", "/", "example.test", None, b"") + self.assertEqual(headers[0], (b":method", b"foo")) + + def test_lowercase_connect_is_not_rewritten(self): + headers = _request_header_list("connect", "/x", "example.test", None, b"") + self.assertIn((b":scheme", b"https"), headers) + self.assertIn((b":path", b"/x"), headers) + + def test_exact_connect_uses_connect_pseudo_fields(self): + headers = _request_header_list("CONNECT", "ignored", "example.test:443", None, b"") + self.assertEqual(headers[:2], [(b":method", b"CONNECT"), (b":authority", b"example.test:443")]) + self.assertNotIn((b":scheme", b"https"), headers) + + def test_invalid_method_rejected(self): + with self.assertRaises(H2ProtocolError): + _request_header_list("GE\rT", "/", "example.test", None, b"") + + def test_invalid_path_rejected(self): + with self.assertRaises(H2ProtocolError): + _request_header_list("GET", "/bad\x00path", "example.test", None, b"") + + def test_invalid_authority_rejected(self): + with self.assertRaises(H2ProtocolError): + _request_header_list("GET", "/", "x\r\ny", None, b"") + + def test_connection_fields_and_nominees_are_removed(self): + headers = _request_header_list( + "GET", "/", "example.test", + [("Connection", "x-remove"), ("X-Remove", "yes"), ("X-Keep", "yes")], b"" + ) + self.assertNotIn((b"connection", b"x-remove"), headers) + self.assertNotIn((b"x-remove", b"yes"), headers) + self.assertIn((b"x-keep", b"yes"), headers) + + def test_invalid_te_rejected(self): + with self.assertRaises(H2ProtocolError): + _request_header_list("GET", "/", "example.test", {"TE": "gzip"}, b"") + + def test_content_length_mismatch_rejected(self): + with self.assertRaises(H2ProtocolError): + _request_header_list("POST", "/", "example.test", {"Content-Length": "4"}, b"abc") + + def test_duplicate_content_length_is_collapsed(self): + headers = _request_header_list( + "POST", "/", "example.test", + [("Content-Length", "3"), ("content-length", "3")], b"abc" + ) + self.assertEqual([item for item in headers if item[0] == b"content-length"], [(b"content-length", b"3")]) + + +class TestAuthorityHandling(unittest.TestCase): + def test_default_port_is_omitted(self): + self.assertEqual(_authority_for_host("example.test", 443), "example.test") + + def test_nondefault_port_is_included(self): + self.assertEqual(_authority_for_host("example.test", 8443), "example.test:8443") + + def test_ipv6_is_bracketed(self): + self.assertEqual(_authority_for_host("2001:db8::1", 8443), "[2001:db8::1]:8443") + + def test_idna_host_is_ascii(self): + self.assertEqual(_authority_for_host(u"t\u00e4st.example", 443), "xn--tst-qla.example") + + +class TestResponseValidation(unittest.TestCase): + def test_identical_content_lengths_are_accepted(self): + headers = [(b":status", b"200"), (b"content-length", b"3"), (b"content-length", b"3")] + self.assertEqual(_parse_content_length(headers), 3) + + def test_comma_joined_identical_content_lengths_are_accepted(self): + self.assertEqual(_parse_content_length([(b"content-length", b"3, 3")]), 3) + + def test_conflicting_content_lengths_are_rejected(self): + with self.assertRaises(H2ProtocolError): + _parse_content_length([(b"content-length", b"1"), (b"content-length", b"2")]) + + def test_invalid_status_range_is_rejected(self): + with self.assertRaises(H2ProtocolError): + _validate_response_field_section([(b":status", b"799")]) + + def test_content_length_in_trailers_is_rejected(self): + with self.assertRaises(H2ProtocolError): + _validate_response_field_section([(b"content-length", b"0")], trailers=True) + + +class TestConnectionState(unittest.TestCase): + def _exchange(self, incoming, method="GET", body=None): + conn = _bare_connection(incoming) + return conn, conn.exchange(method, "/", "example.test", {}, body, 1) + + def test_end_stream_headers_wait_for_continuation(self): + block = _response_block(b"200", [(b"x-test", b"ok")]) + cut = max(1, len(block) // 2) + incoming = ( + encode_frame(HEADERS, FLAG_END_STREAM, 1, block[:cut]) + + encode_frame(CONTINUATION, FLAG_END_HEADERS, 1, block[cut:]) + ) + conn, result = self._exchange(incoming) + self.assertEqual(result[0], 200) + self.assertIn((b"x-test", b"ok"), result[1]) + self.assertEqual(conn.sock.incoming, bytearray()) + + def test_content_length_short_body_is_rejected(self): + incoming = ( + encode_frame(HEADERS, FLAG_END_HEADERS, 1, _response_block(b"200", [(b"content-length", b"4")])) + + encode_frame(DATA, FLAG_END_STREAM, 1, b"abc") + ) + with self.assertRaises(H2ProtocolError): + self._exchange(incoming) + + def test_content_length_long_body_is_rejected(self): + incoming = ( + encode_frame(HEADERS, FLAG_END_HEADERS, 1, _response_block(b"200", [(b"content-length", b"2")])) + + encode_frame(DATA, FLAG_END_STREAM, 1, b"abc") + ) + with self.assertRaises(H2ProtocolError): + self._exchange(incoming) + + def test_204_body_is_rejected(self): + incoming = ( + encode_frame(HEADERS, FLAG_END_HEADERS, 1, _response_block(b"204")) + + encode_frame(DATA, FLAG_END_STREAM, 1, b"not allowed") + ) + with self.assertRaises(H2ProtocolError): + self._exchange(incoming) + + def test_204_content_length_is_rejected(self): + incoming = encode_frame( + HEADERS, FLAG_END_HEADERS | FLAG_END_STREAM, 1, + _response_block(b"204", [(b"content-length", b"0")]) + ) + with self.assertRaises(H2ProtocolError): + self._exchange(incoming) + + def test_head_body_is_rejected(self): + incoming = ( + encode_frame(HEADERS, FLAG_END_HEADERS, 1, _response_block(b"200")) + + encode_frame(DATA, FLAG_END_STREAM, 1, b"body") + ) + with self.assertRaises(H2ProtocolError): + self._exchange(incoming, method="HEAD") + + def test_informational_responses_are_bounded(self): + early = encode_frame(HEADERS, FLAG_END_HEADERS, 1, _response_block(b"103")) + incoming = early * (MAX_INFORMATIONAL_RESPONSES + 1) + with self.assertRaises(H2Error): + self._exchange(incoming) + + def test_receive_flow_control_is_enforced(self): + incoming = ( + encode_frame(HEADERS, FLAG_END_HEADERS, 1, _response_block(b"200")) + + encode_frame(DATA, FLAG_END_STREAM, 1, b"ab") + ) + conn = _bare_connection(incoming) + conn.conn_recv_window = 1 + with self.assertRaises(H2ProtocolError): + conn.exchange("GET", "/", "example.test", {}, None, 1) + + def test_large_request_is_fragmented(self): + incoming = encode_frame( + HEADERS, FLAG_END_HEADERS | FLAG_END_STREAM, 1, _response_block(b"200") + ) + conn, result = self._exchange(incoming, method="POST", body=b"x" * 20000) + self.assertEqual(result[0], 200) + frames = _frames_from_bytes(b"".join(conn.sock.sent)) + lengths = [len(payload) for ftype, _, _, payload in frames if ftype == DATA] + self.assertEqual(lengths, [16384, 3616]) + + def test_goaway_uses_first_payload_word_as_last_stream_id(self): + incoming = ( + encode_frame(GOAWAY, 0, 0, struct.pack("!II", 1, 0)) + + encode_frame(HEADERS, FLAG_END_HEADERS | FLAG_END_STREAM, 1, _response_block(b"200")) + ) + conn, result = self._exchange(incoming) + self.assertEqual(result[0], 200) + self.assertFalse(conn.usable) + + def test_trailers_are_decoded_as_separate_field_section(self): + incoming = ( + encode_frame(HEADERS, FLAG_END_HEADERS, 1, _response_block(b"200", [(b"content-length", b"4")])) + + encode_frame(DATA, 0, 1, b"body") + + encode_frame(HEADERS, FLAG_END_HEADERS | FLAG_END_STREAM, 1, Encoder().encode([(b"x-trailer", b"done")])) + ) + _, result = self._exchange(incoming) + self.assertEqual(result[2], b"body") + self.assertIn((b"x-trailer", b"done"), result[1]) + + def test_server_settings_enable_push_zero_is_accepted(self): + conn = _bare_connection() + payload = struct.pack("!HI", SETTINGS_ENABLE_PUSH, 0) + conn._handle_settings(0, 0, payload) + self.assertTrue(conn.sock.sent) + + def test_server_settings_enable_push_one_is_rejected(self): + conn = _bare_connection() + payload = struct.pack("!HI", SETTINGS_ENABLE_PUSH, 1) + with self.assertRaises(H2ProtocolError): + conn._handle_settings(0, 0, payload) + + +class TestTlsPolicy(unittest.TestCase): + def setUp(self): + if not hasattr(ssl.SSLContext, "set_alpn_protocols"): + self.skipTest("ALPN is unavailable") + + def test_verification_is_disabled_by_default_context(self): + context = _make_ssl_context(False, None) + self.assertEqual(context.verify_mode, ssl.CERT_NONE) + self.assertFalse(getattr(context, "check_hostname", False)) + + def test_verification_can_be_enabled(self): + if not hasattr(ssl, "create_default_context"): + self.skipTest("verified contexts are unavailable") + context = _make_ssl_context(True, None) + self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED) + + def test_tls_minimum_is_12_when_supported(self): + context = _make_ssl_context(False, None) + tls_version = getattr(ssl, "TLSVersion", None) + if tls_version is None or not hasattr(context, "minimum_version"): + self.skipTest("minimum_version is unavailable") + self.assertGreaterEqual(context.minimum_version, tls_version.TLSv1_2) + + +class TestRedirectHandling(unittest.TestCase): + def _run_with_fake_exchange(self, responses, **kwargs): + calls = [] + original = _http2._pooledExchange + + def fake(host, port, proxy, method, path, authority, headers, body, timeout, + verify=False, ssl_context=None): + calls.append((host, port, method, path, authority, list(_http2._iter_header_items(headers)), body)) + return responses[len(calls) - 1] + + _http2._pooledExchange = fake + try: + response = _http2.open_url(**kwargs) + finally: + _http2._pooledExchange = original + return response, calls + + def test_custom_method_is_not_case_normalized_on_redirect(self): + responses = [ + (302, [(b":status", b"302"), (b"location", b"/next")], b""), + (200, [(b":status", b"200")], b"ok"), + ] + response, calls = self._run_with_fake_exchange( + responses, url="https://example.test/start", method="post", headers={}, body=b"x" + ) + self.assertEqual(response.status, 200) + self.assertEqual(calls[1][2], "post") + self.assertEqual(calls[1][6], b"x") + + def test_post_302_becomes_get_and_drops_entity_headers(self): + responses = [ + (302, [(b":status", b"302"), (b"location", b"/next")], b""), + (200, [(b":status", b"200")], b"ok"), + ] + _, calls = self._run_with_fake_exchange( + responses, + url="https://example.test/start", + method="POST", + headers=[("Content-Type", "text/plain"), ("X-Keep", "yes")], + body=b"x", + ) + self.assertEqual(calls[1][2], "GET") + self.assertIsNone(calls[1][6]) + names = [key.lower() for key, _ in calls[1][5]] + self.assertNotIn("content-type", names) + self.assertIn("x-keep", names) + + def test_cross_origin_redirect_strips_all_credentials(self): + responses = [ + (307, [(b":status", b"307"), (b"location", b"https://other.test/final")], b""), + (200, [(b":status", b"200")], b"ok"), + ] + _, calls = self._run_with_fake_exchange( + responses, + url="https://example.test/start", + method="GET", + headers=[("Authorization", "a"), ("Cookie", "b"), ("Authorization", "c"), ("X-Keep", "yes")], + ) + names = [key.lower() for key, _ in calls[1][5]] + self.assertNotIn("authorization", names) + self.assertNotIn("cookie", names) + self.assertIn("x-keep", names) + + def test_idna_and_nondefault_port_authority(self): + responses = [(200, [(b":status", b"200")], b"ok")] + _, calls = self._run_with_fake_exchange( + responses, url=u"https://t\u00e4st.example:8443/path", method="GET" + ) + self.assertEqual(calls[0][0], "xn--tst-qla.example") + self.assertEqual(calls[0][1], 8443) + self.assertEqual(calls[0][4], "xn--tst-qla.example:8443") + + +class TestPoolRetries(unittest.TestCase): + class _Connection(object): + def __init__(self, outcomes): + self.outcomes = outcomes + self.usable = True + self.closed = False + + def exchange(self, *args): + outcome = self.outcomes.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + def close(self): + self.closed = True + self.usable = False + + def setUp(self): + _http2.close_pooled_connections() + + def tearDown(self): + _http2.close_pooled_connections() + + def _install_factory(self, connections): + original = _http2._H2Connection + + def factory(*args, **kwargs): + return connections.pop(0) + + _http2._H2Connection = factory + return original + + def test_reused_get_is_retried_after_transport_eof(self): + first = self._Connection([ + (200, [(b":status", b"200")], b"first"), + H2TransportError("stale"), + ]) + second = self._Connection([(200, [(b":status", b"200")], b"second")]) + original = self._install_factory([first, second]) + try: + _http2._pooledExchange("x", 443, None, "GET", "/", "x", {}, None, 1) + result = _http2._pooledExchange("x", 443, None, "GET", "/", "x", {}, None, 1) + finally: + _http2._H2Connection = original + self.assertEqual(result[2], b"second") + self.assertTrue(first.closed) + + def test_reused_post_is_not_retried_after_transport_eof(self): + first = self._Connection([ + (200, [(b":status", b"200")], b"first"), + H2TransportError("stale"), + ]) + original = self._install_factory([first]) + try: + _http2._pooledExchange("x", 443, None, "POST", "/", "x", {}, b"a", 1) + with self.assertRaises(H2TransportError): + _http2._pooledExchange("x", 443, None, "POST", "/", "x", {}, b"a", 1) + finally: + _http2._H2Connection = original + + +@unittest.skipUnless(HAVE_H2, "python-hyper/h2 is not installed") +class TestLocalH2Peer(unittest.TestCase): + class _WrappedSocket(object): + def __init__(self, raw): + self.raw = raw + + def selected_alpn_protocol(self): + return "h2" + + def version(self): + return "TLSv1.2" + + def compression(self): + return None + + def __getattr__(self, name): + return getattr(self.raw, name) + + class _FakeContext(object): + def wrap_socket(self, raw, server_hostname=None): + return TestLocalH2Peer._WrappedSocket(raw) + + def _with_peer(self, server_target, client_target): + client_raw, server_raw = socket.socketpair() + errors = [] + + def run_server(): + try: + server_target(server_raw) + except Exception as ex: + errors.append(ex) + finally: + server_raw.close() + + original_connect = _http2._connect_socket + original_context = _http2._make_ssl_context + _http2._connect_socket = lambda *args, **kwargs: client_raw + _http2._make_ssl_context = lambda *args, **kwargs: self._FakeContext() + thread = threading.Thread(target=run_server) + thread.start() + try: + result = client_target() + finally: + _http2._connect_socket = original_connect + _http2._make_ssl_context = original_context + client_raw.close() + thread.join(5) + if thread.is_alive(): + self.fail("local HTTP/2 peer did not terminate") + if errors: + raise errors[0] + return result + + def test_large_post_flow_control_and_method_preservation(self): + observed = {"headers": None, "body": bytearray()} + + def server(raw): + h2 = ReferenceH2Connection(config=H2Configuration(client_side=False, header_encoding=None)) + h2.initiate_connection() + raw.sendall(h2.data_to_send()) + done = False + while not done: + data = raw.recv(65536) + if not data: + break + for event in h2.receive_data(data): + if isinstance(event, RequestReceived): + observed["headers"] = event.headers + elif isinstance(event, DataReceived): + observed["body"].extend(event.data) + h2.acknowledge_received_data(event.flow_controlled_length, event.stream_id) + elif isinstance(event, StreamEnded): + h2.send_headers(event.stream_id, [(b":status", b"200"), (b"content-length", b"2")]) + h2.send_data(event.stream_id, b"ok", end_stream=True) + done = True + outbound = h2.data_to_send() + if outbound: + raw.sendall(outbound) + + def client(): + conn = _H2Connection("example.test", 443, None, 5) + try: + return conn.exchange("foo", "/upload", "example.test", {}, b"x" * 100000, 5) + finally: + conn.close() + + result = self._with_peer(server, client) + self.assertEqual(result[0], 200) + self.assertEqual(result[2], b"ok") + self.assertEqual(len(observed["body"]), 100000) + self.assertIn((b":method", b"foo"), observed["headers"]) + + def test_exchange_pair_reports_completion_order(self): + def server(raw): + h2 = ReferenceH2Connection(config=H2Configuration(client_side=False, header_encoding=None)) + h2.initiate_connection() + raw.sendall(h2.data_to_send()) + ended = [] + while len(ended) < 2: + data = raw.recv(65536) + if not data: + break + for event in h2.receive_data(data): + if isinstance(event, StreamEnded): + ended.append(event.stream_id) + outbound = h2.data_to_send() + if outbound: + raw.sendall(outbound) + for sid in reversed(ended): + h2.send_headers(sid, [(b":status", b"200")], end_stream=True) + raw.sendall(h2.data_to_send()) + + def client(): + conn = _H2Connection("example.test", 443, None, 5) + try: + return conn.exchange_pair([ + {"method": "GET", "path": "/first"}, + {"method": "GET", "path": "/second"}, + ], 5) + finally: + conn.close() + + order, results = self._with_peer(server, client) + self.assertEqual(order, [3, 1]) + self.assertEqual(results[1][0], 200) + self.assertEqual(results[3][0], 200) + class TestConstants(unittest.TestCase): def test_redirect_codes(self):