Function _http3

dns/query.py:676–734

(
    q: dns.message.Message,
    where: str,
    url: str,
    timeout: float | None = None,
    port: int = 443,
    source: str | None = None,
    source_port: int = 0,
    one_rr_per_rrset: bool = False,
    ignore_trailing: bool = False,
    verify: bool | str | ssl.SSLContext = True,
    post: bool = True,
    connection: dns.quic.SyncQuicConnection | None = None,
)

def _http3(
    q: dns.message.Message,
    where: str,
    url: str,
    timeout: float | None = None,
    port: int = 443,
    source: str | None = None,
    source_port: int = 0,
    one_rr_per_rrset: bool = False,
    ignore_trailing: bool = False,
    verify: bool | str | ssl.SSLContext = True,
    post: bool = True,
    connection: dns.quic.SyncQuicConnection | None = None,
) -> dns.message.Message:
    if not dns.quic.have_quic:
        raise NoDOH("DNS-over-HTTP3 is not available.")  # pragma: no cover

    url_parts = urllib.parse.urlparse(url)
    hostname = url_parts.hostname
    assert hostname is not None
    if url_parts.port is not None:
        port = url_parts.port

    q.id = 0
    wire = q.to_wire()
    the_connection: dns.quic.SyncQuicConnection
    the_manager: dns.quic.SyncQuicManager
    if connection:
        manager: contextlib.AbstractContextManager = contextlib.nullcontext(None)
    else:
        manager = dns.quic.SyncQuicManager(
            verify_mode=verify, server_name=hostname, h3=True  # pyright: ignore
        )
        the_manager = manager  # for type checking happiness

    with manager:
        if connection:
            the_connection = connection
        else:
            the_connection = the_manager.connect(  # pyright: ignore
                where, port, source, source_port
            )
        (start, expiration) = _compute_times(timeout)
        with the_connection.make_stream(timeout) as stream:  # pyright: ignore
            stream.send_h3(url, wire, post)
            wire = stream.receive(_remaining(expiration))
            _check_status(stream.headers(), where, wire)
        finish = time.time()
    r = dns.message.from_wire(
        wire,
        keyring=q.keyring,
        request_mac=q.request_mac,
        one_rr_per_rrset=one_rr_per_rrset,
        ignore_trailing=ignore_trailing,
    )
    r.time = max(finish - start, 0.0)
    if not q.is_response(r):
        raise BadResponse