qh3: QUIC and HTTP/3

1.7.1 · active · verified Fri Apr 10

qh3 is a lightweight and fast Python library providing a QUIC and HTTP/3 implementation. It is a actively maintained fork of `aioquic`, designed to be embedded into other Python client and server libraries. It focuses on client-side functionality, features a minimal TLS 1.3 implementation, and does not rely on external C/OpenSSL dependencies, instead using a Rust-based cryptography backend. The library maintains a regular release cadence with updates often several times a month.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to establish an HTTP/3 client connection, send a GET request, and receive the response body. It leverages `asyncio` for asynchronous operations. Users should configure `QuicConfiguration` for certificate validation appropriate for their environment.

import asyncio
import ssl
from qh3.quic.configuration import QuicConfiguration
from qh3.asyncio.client import connect
from qh3.h3.connection import H3Connection
from qh3.h3.events import DataReceived, HeadersReceived


async def fetch_http3_url(url: str):
    # Create a QUIC configuration
    configuration = QuicConfiguration(
        is_client=True,
        alpn_protocols=["h3-29"],
        max_datagram_frame_size=65536,
        # Set up a basic TLS context. For production, use proper certificate validation.
        # In this example, we disable it for simplicity, NOT recommended for production.
        verify_mode=ssl.CERT_NONE, 
        # You can add root CAs if needed: ciphers=[...], root_cert=b"..."
    )

    print(f"Connecting to {url}...")
    async with connect(url, configuration=configuration) as client:
        print(f"Connected to {url}. Client: {client}")
        h3_connection = H3Connection(client.quic.configuration)

        # Send a GET request
        stream_id = client.quic.get_next_available_stream_id()
        headers = [
            (b":method", b"GET"),
            (b":scheme", b"https"),
            (b":authority", client.host.encode()),
            (b":path", b"/"),
            (b"user-agent", b"qh3-client"),
        ]
        h3_connection.send_headers(stream_id, headers)

        await client.transmit()

        # Receive response
        data = b''
        while True:
            events = h3_connection.handle_quic_events(client.quic.receive())
            for event in events:
                if isinstance(event, HeadersReceived):
                    print("Received headers:")
                    for k, v in event.headers:
                        print(f"  {k.decode()}: {v.decode()}")
                elif isinstance(event, DataReceived):
                    data += event.data
            if client.quic.is_idle: # Check if the QUIC connection is idle (all data received/sent)
                break
            await asyncio.sleep(0.01) # Yield to event loop
            
        print("\n--- Response Body ---")
        print(data.decode(errors='ignore')) # Ignore decoding errors for simplicity

# Example Usage:
# You need a running HTTP/3 server to test this.
# For local testing, you might use `mkcert` to generate a trusted cert for 'localhost'
# and run a server like `aioquic/examples/http3_server.py` adapted for qh3.
# This example targets a public HTTP/3 enabled endpoint for demonstration, 
# but actual connectivity depends on network and server configuration.

# Replace with a known HTTP/3 server if available
# Note: Public HTTP/3 test servers are not always stable or available.
# For best results, run a local qh3/aioquic compatible server.
# Example for aioquic server: `python -m aioquic.examples.http3_server --host 127.0.0.1 --port 4433 cert.pem key.pem`
# and then use url = "https://127.0.0.1:4433/"

# A real-world example would involve a public URL known to support HTTP/3.
# For this example, we'll try a common one, but it might require a local setup.

url = os.environ.get('QH3_TEST_URL', 'https://quic.tech:8443/')

async def main():
    await fetch_http3_url(url)

if __name__ == '__main__':
    import os
    import logging
    # Set basic logging for better visibility
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

view raw JSON →