test_http2.py

#!/usr/bin/env python

"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission

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.

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,
    DATA,
    HEADERS,
    FLAG_END_HEADERS,
    FLAG_END_STREAM,
    decode_frame_header,
    decode_integer,
    decode_string,
    encode_frame,
    encode_integer,
    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]
        self.assertEqual(decode_frame_header(header), (3, HEADERS, FLAG_END_HEADERS, 1))

    def test_payload_is_appended_verbatim(self):
        frame = encode_frame(DATA, 0, 1, b"hello")
        self.assertEqual(frame[9:], b"hello")

    def test_reserved_stream_bit_is_masked(self):
        # 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]
        length, _, flags, _ = decode_frame_header(header)
        self.assertEqual(length, 0)
        self.assertEqual(flags, FLAG_END_STREAM)

    def test_oversized_payload_rejected(self):
        with self.assertRaises(ValueError):
            encode_frame(DATA, 0, 1, b"x" * (0xFFFFFF + 1))

    def test_bad_header_length_rejected(self):
        with self.assertRaises(ValueError):
            decode_frame_header(b"123")


class TestIntegerCoding(unittest.TestCase):
    def test_rfc_c11_small(self):
        # RFC 7541 C.1.1: 10 with a 5-bit prefix fits in the prefix
        self.assertEqual(list(encode_integer(10, 5)), [10])

    def test_rfc_c12_multibyte(self):
        # RFC 7541 C.1.2: 1337 with a 5-bit prefix
        self.assertEqual(list(encode_integer(1337, 5)), [31, 154, 10])
        self.assertEqual(decode_integer(bytearray([31, 154, 10]), 0, 5), (1337, 3))

    def test_rfc_c13_full_byte_prefix(self):
        # RFC 7541 C.1.3: 42 starting from a full (8-bit prefix at an octet boundary)
        self.assertEqual(list(encode_integer(42, 8)), [42])

    def test_roundtrip_across_prefixes(self):
        for prefix in (4, 5, 6, 7, 8):
            for value in (0, 1, 2, 30, 31, 32, 127, 128, 255, 256, 16384, 1000000):
                encoded = bytearray(encode_integer(value, prefix))
                decoded, pos = decode_integer(encoded, 0, prefix)
                self.assertEqual(decoded, value)
                self.assertEqual(pos, len(encoded))

    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):
        # RFC 7541 C.4.1
        self.assertEqual(binascii.hexlify(huffman_encode(b"www.example.com")), b"f1e3c2e5f23a6ba0ab90f4ff")

    def test_empty(self):
        self.assertEqual(huffman_encode(b""), b"")
        self.assertEqual(huffman_decode(b""), b"")

    def test_roundtrip(self):
        for sample in (b"a", b"hello world", b"/index.html?a=1&b=2",
                       b"GET", b"application/json", b"ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
                       bytes(bytearray(range(256)))):
            self.assertEqual(huffman_decode(huffman_encode(sample)), sample)

    def test_shrinks_typical_text(self):
        sample = b"www.example.com"
        self.assertLess(len(huffman_encode(sample)), len(sample))

    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(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):
        encoded = encode_string(b"custom-value")
        self.assertTrue(bytearray(encoded)[0] & 0x80)          # huffman flag set for compressible text
        self.assertEqual(decode_string(bytearray(encoded), 0), (b"custom-value", len(encoded)))

    def test_literal_branch_when_huffman_would_not_shrink(self):
        encoded = encode_string(_b(0xFF))
        self.assertFalse(bytearray(encoded)[0] & 0x80)         # falls back to a literal string
        self.assertEqual(decode_string(bytearray(encoded), 0), (_b(0xFF), len(encoded)))

    def test_disable_huffman(self):
        encoded = encode_string(b"abc", huffman=False)
        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):
        # 0x82/0x86/0x84 -> static indices 2, 6, 4
        self.assertEqual(
            Decoder().decode(_b(0x82, 0x86, 0x84)),
            [(b":method", b"GET"), (b":scheme", b"http"), (b":path", b"/")],
        )

    def test_static_lookup_bounds(self):
        d = Decoder()
        self.assertEqual(d._get(1), (b":authority", b""))
        self.assertEqual(d._get(2), (b":method", b"GET"))
        self.assertEqual(d._get(STATIC_LEN), STATIC_TABLE[-1])

    def test_index_zero_rejected(self):
        with self.assertRaises(H2CompressionError):
            Decoder()._get(0)

    def test_index_out_of_range_rejected(self):
        with self.assertRaises(H2CompressionError):
            Decoder()._get(STATIC_LEN + 1)     # no dynamic entries yet

    def test_literal_incremental_indexing_populates_dynamic_table(self):
        # 0x40 = literal with incremental indexing, new name
        block = bytearray([0x40]) + encode_string(b"custom-key") + encode_string(b"custom-value")
        d = Decoder()
        self.assertEqual(d.decode(bytes(block)), [(b"custom-key", b"custom-value")])
        # entry is now addressable at the first dynamic index (STATIC_LEN + 1)
        self.assertEqual(d._get(STATIC_LEN + 1), (b"custom-key", b"custom-value"))
        self.assertEqual(d._size, 32 + len(b"custom-key") + len(b"custom-value"))

    def test_literal_without_indexing_does_not_touch_dynamic_table(self):
        block = bytearray([0x00]) + encode_string(b"k") + encode_string(b"v")
        d = Decoder()
        self.assertEqual(d.decode(bytes(block)), [(b"k", b"v")])
        self.assertEqual(d.dynamic, [])

    def test_dynamic_table_eviction(self):
        d = Decoder(max_size=40)               # each 2+2 byte entry costs 32+2+2 = 36
        d._add(b"aa", b"bb")
        self.assertEqual(len(d.dynamic), 1)
        d._add(b"cc", b"dd")                   # 72 > 40 -> oldest evicted
        self.assertEqual(d.dynamic, [(b"cc", b"dd")])
        self.assertEqual(d._size, 36)

    def test_dynamic_size_update_clears(self):
        d = Decoder()
        d._add(b"x", b"y")
        d.decode(_b(0x20))                     # 0x20 = dynamic table size update to 0
        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):
        headers = [
            (b":method", b"GET"),
            (b":scheme", b"https"),
            (b":path", b"/a/b?c=d"),
            (b":authority", b"example.com"),
            (b"user-agent", b"sqlmap"),
            (b"accept", b""),                  # empty value
            (b"x-custom", b"\x00\x01\xff"),    # non-ASCII value
        ]
        self.assertEqual(Decoder().decode(Encoder().encode(headers)), headers)

    def test_encoder_output_is_bytes(self):
        self.assertIsInstance(Encoder().encode([(b"a", b"b")]), bytes)


class TestH2Response(unittest.TestCase):
    def _make(self, status=200, headers=None, body=b"body"):
        headers = headers if headers is not None else [(b":status", b"200"), (b"content-type", b"text/html")]
        return H2Response("https://target/x", status, headers, body)

    def test_basic_fields(self):
        r = self._make()
        self.assertEqual(r.code, 200)
        self.assertEqual(r.status, 200)
        self.assertEqual(r.msg, "OK")
        self.assertEqual(r.http_version, "HTTP/2.0")
        self.assertEqual(r.geturl(), "https://target/x")

    def test_unknown_status_message(self):
        self.assertEqual(self._make(status=799).msg, "")

    def test_pseudo_headers_stripped(self):
        r = self._make()
        self.assertNotIn(":status", r.info())
        self.assertEqual(r.info().get("content-type"), "text/html")

    def test_read_full_then_empty(self):
        r = self._make(body=b"hello")
        self.assertEqual(r.read(), b"hello")
        self.assertEqual(r.read(), b"")            # offset exhausted

    def test_read_in_chunks(self):
        r = self._make(body=b"abcdef")
        self.assertEqual(r.read(2), b"ab")
        self.assertEqual(r.read(3), b"cde")
        self.assertEqual(r.read(10), b"f")         # asking past the end returns the remainder
        self.assertEqual(r.read(10), b"")

    def test_str_header_names_accepted(self):
        # headers may arrive already decoded to str (not only bytes)
        r = H2Response("https://t/", 200, [("content-type", "application/json")], b"{}")
        self.assertEqual(r.info().get("content-type"), "application/json")

    def test_mimetools_style_headers_list(self):
        # patchHeaders() relies on a '.headers' list of "Name: value\r\n" lines being present
        r = self._make()
        self.assertTrue(hasattr(r.info(), "headers"))
        self.assertIn("content-type: text/html\r\n", r.info().headers)

    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):
        for code in (301, 302, 303, 307, 308):
            self.assertIn(code, REDIRECT_CODES)
        self.assertNotIn(200, REDIRECT_CODES)

    def test_static_table_length(self):
        self.assertEqual(STATIC_LEN, len(STATIC_TABLE))
        self.assertEqual(STATIC_LEN, 61)           # RFC 7541 Appendix A


if __name__ == "__main__":
    unittest.main(verbosity=2)