1
pip install aioquic

1
#!/usr/bin/env python3
2
# http3_client.py -- HTTP/3 client demo using aioquic
3
import asyncio
4
from aioquic.asyncio.client import connect
5
from aioquic.h3.connection import H3Connection
6
from aioquic.h3.events import HeadersReceived, DataReceived
7
from aioquic.quic.configuration import QuicConfiguration
8

9

10
async def http3_get(url: str):
11
    """
12
    发起 HTTP/3 GET 请求
13
    需要: pip install aioquic
14
    """
15
    from urllib.parse import urlparse
16

17
    parsed = urlparse(url)
18
    host = parsed.hostname
19
    port = parsed.port or 443
20
    path = parsed.path or "/"
21

22
    # 创建 QUIC 配置
23
    configuration = QuicConfiguration()
24

25
    # 连接本地自签名服务器时需要跳过证书验证
26
    configuration.verify_mode = 0
27

28
    async with connect(host, port, configuration=configuration) as protocol:
29
        # 创建 HTTP/3 连接
30
        h3 = H3Connection(protocol._quic)
31

32
        # 获取下一个可用的流 ID
33
        stream_id = protocol._quic.get_next_available_stream_id()
34

35
        # 发送请求头
36
        h3.send_headers(
37
            stream_id=stream_id,
38
            headers=[
39
                (b":method", b"GET"),
40
                (b":scheme", b"https"),
41
                (b":authority", f"{host}:{port}".encode()),
42
                (b":path", path.encode()),
43
                (b"user-agent", b"aioquic-h3-demo/1.0"),
44
            ],
45
        )
46
        # 发送空请求体，结束流
47
        h3.send_data(stream_id=stream_id, data=b"", end_stream=True)
48
        protocol.transmit()
49

50
        # 接收响应
51
        response_headers = {}
52
        response_body = b""
53

54
        # 通过 QUIC 事件驱动接收 H3 事件
55
        while True:
56
            event = protocol._quic.next_event()
57
            if event is None:
58
                await asyncio.sleep(0.01)
59
                continue
60
            for h3_event in h3.handle_event(event):
61
                if isinstance(h3_event, HeadersReceived):
62
                    for name, value in h3_event.headers:
63
                        response_headers[name.decode()] = value.decode()
64
                elif isinstance(h3_event, DataReceived):
65
                    response_body += h3_event.data
66
                    if h3_event.stream_ended:
67
                        return response_headers, response_body
68

69

70
if __name__ == "__main__":
71
    headers, body = asyncio.run(http3_get("https://localhost:8443/"))
72
    print(f"Status: {headers.get(':status', 'unknown')}")
73
    print(f"Content-Type: {headers.get('content-type', 'unknown')}")
74
    print(f"Body length: {len(body)} bytes")
75
    print(f"Body: {body.decode()[:100]}...")

1
# 在一个终端捕获 QUIC 流量（Caddy 监听在 8443 端口）
2
sudo tcpdump -i lo -n udp port 8443 -w quic_capture.pcap
3

4
# 在另一个终端发起 HTTP/3 请求
5
curl --http3 -k https://localhost:8443/
6

7
# 停止 tcpdump 后用 Wireshark 分析
8
wireshark quic_capture.pcap

1
No.  Time      Source        Dest          Protocol  Info
2
1    0.000     ::1           ::1           UDP       54321 → 8443 Len=1250 (QUIC Initial)
3
2    0.025     ::1           ::1           UDP       8443 → 54321 Len=1250 (QUIC Initial)
4
3    0.026     ::1           ::1           UDP       8443 → 54321 Len=1250 (QUIC Handshake)
5
4    0.051     ::1           ::1           UDP       54321 → 8443 Len=100  (QUIC Handshake)
6
5    0.075     ::1           ::1           UDP       8443 → 54321 Len=1250 (QUIC 1-RTT Protected)