Hướng Dẫn Sử Dụng HTTP/3 Trong Python

pip install aioquic

openssl req -x509 -newkey rsa:2048 -nodes -days 365 \
  -keyout key.pem -out cert.pem -subj "/CN=localhost"

import argparse
import asyncio
import json
import logging
from typing import Dict, Optional
from aioquic.asyncio import QuicConnectionProtocol, serve
from aioquic.h3.connection import H3_ALPN, H3Connection
from aioquic.h3.events import DataReceived, HeadersReceived, H3Event
from aioquic.quic.configuration import QuicConfiguration

LOG = logging.getLogger("h3server")

class Http3ServerProtocol(QuicConnectionProtocol):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._http: Optional[H3Connection] = None
        self._bodies: Dict[int, bytearray] = {}
        self._req_meta: Dict[int, tuple[str, str]] = {}

    def handle_h3_event(self, event: H3Event) -> None:
        if isinstance(event, HeadersReceived):
            headers = {k.lower(): v for k, v in event.headers}
            method = headers.get(b":method", b"").decode()
            path = headers.get(b":path", b"/").decode()
            sid = event.stream_id
            LOG.info("Headers on stream %d: %s %s", sid, method, path)
            self._req_meta[sid] = (method, path)
            if method in ("POST", "PUT", "PATCH"):
                self._bodies[sid] = bytearray()
                if event.stream_ended:
                    payload = bytes(self._bodies.pop(sid))
                    self._respond(sid, method, path, payload)
                    self._req_meta.pop(sid, None)
            else:
                self._respond(sid, method, path, body=b"")
                self._req_meta.pop(sid, None)
        elif isinstance(event, DataReceived):
            sid = event.stream_id
            if sid not in self._bodies:
                return
            self._bodies[sid].extend(event.data)
            if event.stream_ended:
                method, path = self._req_meta.get(sid, ("", "/"))
                payload = bytes(self._bodies.pop(sid, bytearray()))
                self._respond(sid, method, path, payload)
                self._req_meta.pop(sid, None)

    def _respond(self, stream_id: int, method: Optional[str], path: Optional[str], body: bytes) -> None:
        status = b"200"
        content_type = b"text/plain"
        if method == "GET" and path == "/":
            payload = b"hello from aioquic h3 \n"
        elif method == "POST" and path == "/echo":
            try:
                obj = json.loads(body.decode() or "{}")
                payload = (json.dumps(obj, indent=2) + "\n").encode()
                content_type = b"application/json"
            except Exception:
                payload = body or b"(empty)\n"
                content_type = b"application/octet-stream"
        else:
            payload = b"ok\n"
        headers = [
            (b":status", status),
            (b"server", b"aioquic-tutorial"),
            (b"content-type", content_type),
            (b"alt-svc", b'h3=":4433"; ma=3600'),
        ]
        self._http.send_headers(stream_id, headers)
        self._http.send_data(stream_id, payload, end_stream=True)
        LOG.info("Responded on stream %d (%s bytes)", stream_id, len(payload))

async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="::1")
    parser.add_argument("--port", type=int, default=4433)
    parser.add_argument("--certificate", default="cert.pem")
    parser.add_argument("--private-key", dest="private_key", default="key.pem")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO)
    config = QuicConfiguration(
        is_client=False,
        alpn_protocols=H3_ALPN,
    )
    config.load_cert_chain(args.certificate, args.private_key)
    server = await serve(
        host=args.host,
        port=args.port,
        configuration=config,
        create_protocol=Http3ServerProtocol,
    )
    logging.info("HTTP/3 server listening on %s:%d", args.host, args.port)
    try:
        await asyncio.Future()
    except KeyboardInterrupt:
        pass
    finally:
        server.close()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

python server.py --host ::1 --port 4433 --certificate cert.pem --private-key key.pem

INFO:root:HTTP/3 server listening on ::1:4433

curl -v --http3 -k https://localhost:4433/
curl -v --http3 -k https://localhost:4433/echo -d '{"hello": "world"}' \
     -H 'content-type: application/json'

import argparse
import asyncio
import logging
import json
from typing import Optional
from aioquic.asyncio import connect
from aioquic.asyncio.protocol import QuicConnectionProtocol
from aioquic.h3.connection import H3_ALPN, H3Connection
from aioquic.h3.events import HeadersReceived, DataReceived
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import QuicEvent

LOG = logging.getLogger("h3client")

class H3ClientProtocol(QuicConnectionProtocol):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._http = H3Connection(self._quic)
        self._stream_id: Optional[int] = None
        self._response_headers = None
        self._response_body = bytearray()
        self._done = asyncio.Event()

    def request(self, method: str, path: str, body: bytes = b""):
        self._stream_id = self._quic.get_next_available_stream_id()
        headers = [
            (b":method", method.encode()),
            (b":scheme", b"https"),
            (b":authority", b"localhost"),
            (b":path", path.encode()),
        ]
        self._http.send_headers(self._stream_id, headers)
        self._http.send_data(self._stream_id, body or b"", end_stream=True)
        self.transmit()

    def quic_event_received(self, event: QuicEvent):
        for h3_event in self._http.handle_event(event):
            if isinstance(h3_event, HeadersReceived):
                self._response_headers = h3_event.headers
            elif isinstance(h3_event, DataReceived):
                self._response_body.extend(h3_event.data)
                if h3_event.stream_ended:
                    self._done.set()

    async def wait_response(self):
        await self._done.wait()
        return self._response_headers, bytes(self._response_body)

async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="localhost")
    parser.add_argument("--port", type=int, default=4433)
    parser.add_argument("--path", default="/")
    parser.add_argument("--insecure", action="store_true", default=True)
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO)
    config = QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN)
    if args.insecure:
        config.verify_mode = False
    async with connect(
        args.host,
        args.port,
        configuration=config,
        create_protocol=H3ClientProtocol,
    ) as protocol:
        protocol.request("GET", args.path)
        headers, body = await protocol.wait_response()
        LOG.info("Response headers: %s", headers)
        LOG.info("Response body:\n%s", body.decode(errors="ignore"))

if __name__ == "__main__":
    asyncio.run(main())