Stabilization of http2.py module

lib/core/settings.py

@@ -20,7 +20,7 @@ from lib.core.enums import OS
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-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)

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
+Coverage for the native HTTP/2 client in lib/request/http2.py: frame and HPACK
-lib/request/http2.py: the RFC 7540 frame codec, the RFC 7541 HPACK integer /
+codecs, request/response validation, connection state, flow control, retries,
-Huffman / string primitives, the HPACK Decoder/Encoder (static + dynamic table),
+URL handling, TLS policy, and the urllib-compatible H2Response wrapper.
-and the urllib-compatible H2Response wrapper.
 
-Nothing here opens a socket or negotiates TLS - only the deterministic codecs and
+Most tests are deterministic and network-free. When python-hyper/h2 is installed,
-the response adapter are exercised. Known vectors are the canonical RFC 7541
+two optional socketpair tests exercise an end-to-end local HTTP/2 peer; they never
-examples; everything else is a round-trip / invariant check.
+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 @@ from lib.request.http2 import (
     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 @@ class TestFrameCodec(unittest.TestCase):
         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
+        # the high (reserved) bit of the 31-bit stream id is rejected
-        header = encode_frame(DATA, 0, 0x80000001, b"")[:9]
+        with self.assertRaises(ValueError):
-        self.assertEqual(decode_frame_header(header), (0, DATA, 0, 1))
+            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 @@ class TestIntegerCoding(unittest.TestCase):
         # 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 @@ class TestHuffman(unittest.TestCase):
 
     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 @@ class TestStringCoding(unittest.TestCase):
         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 @@ class TestHpackDecoder(unittest.TestCase):
         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 @@ class TestHpackDecoder(unittest.TestCase):
         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 @@ class TestH2Response(unittest.TestCase):
     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):