Code that reproduces Yahoo's HTTP/2 issue.

yahoo-http2-issue.py

"""
Code that reproduces Yahoo's HTTP/2 issue.

Requires you to run `python -m pip install h2` before executing.

Example output:
```
Testing 'google.com' with 'send_unknown_setting=False'...
Result: <ResponseReceived stream_id:1, headers:[(b':status', b'301'), (b'location', b...

Testing 'google.com' with 'send_unknown_setting=True'...
Result: <ResponseReceived stream_id:1, headers:[(b':status', b'301'), (b'location', b...

Testing 'yahoo.com' with 'send_unknown_setting=False'...
Result: <ResponseReceived stream_id:1, headers:[(b':status', b'301'), (b'date', b'Tue...

Testing 'yahoo.com' with 'send_unknown_setting=True'...
Result: <ConnectionTerminated error_code:ErrorCodes.PROTOCOL_ERROR, last_stream_id:0,...
```
"""
import socket
import ssl
import time
import h2.connection as h2_conn
import h2.events as h2_events
import h2.settings as h2_settings


def main():
    # These all work fine.
    func("google.com", send_unknown_setting=False)
    func("google.com", send_unknown_setting=True)
    func("yahoo.com", send_unknown_setting=False)

    # This *does not* work fine.
    func("yahoo.com", send_unknown_setting=True)


def func(host: str, send_unknown_setting: bool) -> None:
    print(f"\nTesting '{host}' with 'send_unknown_setting={send_unknown_setting}'...")

    ctx = ssl.create_default_context()
    ctx.set_alpn_protocols(["h2"])
    sock = socket.create_connection((host, 443))
    sock = ctx.wrap_socket(sock, server_hostname=host)

    # By default sends 'SETTINGS_ENABLE_CONNECT_PROTOCOL=0'
    settings = h2_settings.Settings()

    if not send_unknown_setting:
        # Don't send 'SETTINGS_ENABLE_CONNECT_PROTOCOL'
        settings._settings.pop(h2_settings.SettingCodes.ENABLE_CONNECT_PROTOCOL, None)

    conn = h2_conn.H2Connection()
    conn.local_settings = settings

    conn.initiate_connection()
    stream_id = conn.get_next_available_stream_id()
    conn.send_headers(
        stream_id,
        [
            (b":method", b"GET"),
            (b":scheme", b"https"),
            (b":authority", host.encode()),
            (b":path", b"/"),
        ],
        end_stream=True,
    )

    sock.sendall(conn.data_to_send())

    received_response = False
    while not received_response:
        received_data = sock.recv(1024)
        for event in conn.receive_data(received_data):
            if isinstance(
                event, (h2_events.ConnectionTerminated, h2_events.ResponseReceived)
            ):
                print(f"Result: {str(event)[:77]}...")
                received_response = True

        time.sleep(0.1)


if __name__ == "__main__":
    main()