Initial DoQ support.

9 files changed, 985 insertions, 1 deletions

diff --git a/dns/__init__.py b/dns/__init__.py
index 196be22..9abdf01 100644
--- a/dns/__init__.py
+++ b/dns/__init__.py
@@ -38,6 +38,7 @@ __all__ = [
     "node",
     "opcode",
     "query",
+    "quic",
     "rcode",
     "rdata",
     "rdataclass",
diff --git a/dns/asyncquery.py b/dns/asyncquery.py
index 107f766..15612e5 100644
--- a/dns/asyncquery.py
+++ b/dns/asyncquery.py
@@ -30,6 +30,7 @@ import dns.exception
 import dns.inet
 import dns.name
 import dns.message
+import dns.quic
 import dns.rcode
 import dns.rdataclass
 import dns.rdatatype
@@ -45,6 +46,7 @@ from dns.query import (
     _have_httpx,
     _have_http2,
     NoDOH,
+    NoDOQ,
 )
 
 if _have_httpx:
@@ -670,3 +672,62 @@ async def inbound_xfr(
                     tsig_ctx = r.tsig_ctx
                 if not retry and query.keyring and not r.had_tsig:
                     raise dns.exception.FormError("missing TSIG")
+
+
+async def quic(
+    q: dns.message.Message,
+    where: str,
+    timeout: Optional[float] = None,
+    port: int = 853,
+    source: Optional[str] = None,
+    source_port: int = 0,
+    one_rr_per_rrset: bool = False,
+    ignore_trailing: bool = False,
+    connection: Optional[dns.quic.AsyncQuicConnection] = None,
+    verify: Union[bool, str] = True,
+    backend: Optional[dns.asyncbackend.Backend] = None,
+) -> dns.message.Message:
+    """Return the response obtained after sending an asynchronous query via
+    DNS-over-QUIC.
+
+    *backend*, a ``dns.asyncbackend.Backend``, or ``None``.  If ``None``,
+    the default, then dnspython will use the default backend.
+
+    See :py:func:`dns.query.doq()` for the documentation of the other
+    parameters, exceptions, and return type of this method.
+    """
+
+    if not dns.quic.have_quic:
+        raise NoDOQ("DNS-over-QUIC is not available.")  # pragma: no cover
+
+    q.id = 0
+    wire = q.to_wire()
+    the_connection: dns.quic.AsyncQuicConnection
+    if connection:
+        cfactory = dns.quic.null_factory
+        mfactory = dns.quic.null_factory
+        the_connection = connection
+    else:
+        (cfactory, mfactory) = dns.quic.factories_for_backend(backend)
+
+    async with cfactory() as context:
+        async with mfactory(context, verify_mode=verify) as the_manager:
+            if not connection:
+                the_connection = the_manager.connect(where, port, source, source_port)
+            start = time.time()
+            stream = await the_connection.make_stream()
+            async with stream:
+                await stream.send(wire, True)
+                wire = await stream.receive(timeout)
+            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
+    return r
diff --git a/dns/dnssec.py b/dns/dnssec.py
index b325f9f..13415bd 100644
--- a/dns/dnssec.py
+++ b/dns/dnssec.py
@@ -269,7 +269,7 @@ def _validate_signature(sig: bytes, data: bytes, key: DNSKEY, chosen_hash: Any)
         keyptr = keyptr[octets:]
         dsa_y = keyptr[0:octets]
         try:
-            dsa_public_key = dsa.DSAPublicNumbers(
+            dsa_public_key = dsa.DSAPublicNumbers(  # type: ignore
                 _bytes_to_long(dsa_y),
                 dsa.DSAParameterNumbers(
                     _bytes_to_long(dsa_p), _bytes_to_long(dsa_q), _bytes_to_long(dsa_g)
diff --git a/dns/query.py b/dns/query.py
index 5663e23..871b67f 100644
--- a/dns/query.py
+++ b/dns/query.py
@@ -34,6 +34,7 @@ import dns.exception
 import dns.inet
 import dns.name
 import dns.message
+import dns.quic
 import dns.rcode
 import dns.rdataclass
 import dns.rdatatype
@@ -108,6 +109,11 @@ class NoDOH(dns.exception.DNSException):
     available."""
 
 
+class NoDOQ(dns.exception.DNSException):
+    """DNS over QUIC (DOQ) was requested but the aioquic module is not
+    available."""
+
+
 # for backwards compatibility
 TransferError = dns.xfr.TransferError
 
@@ -1059,6 +1065,86 @@ def tls(
     )
 
 
+def quic(
+    q: dns.message.Message,
+    where: str,
+    timeout: Optional[float] = None,
+    port: int = 853,
+    source: Optional[str] = None,
+    source_port: int = 0,
+    one_rr_per_rrset: bool = False,
+    ignore_trailing: bool = False,
+    connection: Optional[dns.quic.SyncQuicConnection] = None,
+    verify: Union[bool, str] = True,
+) -> dns.message.Message:
+    """Return the response obtained after sending a query via DNS-over-QUIC.
+
+    *q*, a ``dns.message.Message``, the query to send.
+
+    *where*, a ``str``, the nameserver IP address.
+
+    *timeout*, a ``float`` or ``None``, the number of seconds to wait before the query
+    times out. If ``None``, the default, wait forever.
+
+    *port*, a ``int``, the port to send the query to. The default is 443.
+
+    *source*, a ``str`` containing an IPv4 or IPv6 address, specifying the source
+    address.  The default is the wildcard address.
+
+    *source_port*, an ``int``, the port from which to send the message. The default is
+    0.
+
+    *one_rr_per_rrset*, a ``bool``. If ``True``, put each RR into its own RRset.
+
+    *ignore_trailing*, a ``bool``. If ``True``, ignore trailing junk at end of the
+    received message.
+
+    *connection*, a ``dns.quic.SyncQuicConnection``.  If provided, the
+    connection to use to send the query.
+
+    *verify*, a ``bool`` or ``str``.  If a ``True``, then TLS certificate verification
+    of the server is done using the default CA bundle; if ``False``, then no
+    verification is done; if a `str` then it specifies the path to a certificate file or
+    directory which will be used for verification.
+
+    Returns a ``dns.message.Message``.
+    """
+
+    if not dns.quic.have_quic:
+        raise NoDOQ("DNS-over-QUIC is not available.")  # pragma: no cover
+
+    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)
+        the_connection = connection
+    else:
+        manager = dns.quic.SyncQuicManager(verify_mode=verify)
+        the_manager = manager  # for type checking happiness
+
+    with manager:
+        if not connection:
+            the_connection = the_manager.connect(where, port)
+        start = time.time()
+        with the_connection.make_stream() as stream:
+            stream.send(wire, True)
+            wire = stream.receive(timeout)
+        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
+    return r
+
+
 def xfr(
     where: str,
     zone: Union[dns.name.Name, str],
diff --git a/dns/quic/__init__.py b/dns/quic/__init__.py
new file mode 100644
index 0000000..88c5862
--- /dev/null
+++ b/dns/quic/__init__.py
@@ -0,0 +1,65 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+try:
+    import aioquic.quic.configuration  # type: ignore
+
+    import dns.asyncbackend
+    from dns._asyncbackend import NullContext
+    from dns.quic._sync import SyncQuicManager, SyncQuicConnection, SyncQuicStream
+    from dns.quic._asyncio import (
+        AsyncioQuicManager,
+        AsyncioQuicConnection,
+        AsyncioQuicStream,
+    )
+    from dns.quic._common import AsyncQuicConnection, AsyncQuicManager
+
+    have_quic = True
+
+    def null_factory(*args, **kwargs):
+        return NullContext(None)
+
+    def _asyncio_manager_factory(context, *args, **kwargs):
+        return AsyncioQuicManager(*args, **kwargs)
+
+    # We have a context factory and a manager factory as for trio we need to have
+    # a nursery.
+
+    _async_factories = {"asyncio": (null_factory, _asyncio_manager_factory)}
+
+    try:
+        import trio
+        from dns.quic._trio import TrioQuicManager, TrioQuicConnection, TrioQuicStream
+
+        def _trio_context_factory():
+            return trio.open_nursery()
+
+        def _trio_manager_factory(context, *args, **kwargs):
+            return TrioQuicManager(context, *args, **kwargs)
+
+        _async_factories["trio"] = (_trio_context_factory, _trio_manager_factory)
+    except ImportError:
+        pass
+
+    def factories_for_backend(backend=None):
+        if backend is None:
+            backend = dns.asyncbackend.get_default_backend()
+        return _async_factories[backend.name()]
+
+except ImportError:
+    have_quic = False
+
+    from typing import Any
+
+    class AsyncQuicStream:  # type: ignore
+        pass
+
+    class AsyncQuicConnection:  # type: ignore
+        async def make_stream(self) -> Any:
+            raise NotImplementedError
+
+    class SyncQuicStream:  # type: ignore
+        pass
+
+    class SyncQuicConnection:  # type: ignore
+        def make_stream(self) -> Any:
+            raise NotImplementedError
diff --git a/dns/quic/_asyncio.py b/dns/quic/_asyncio.py
new file mode 100644
index 0000000..0a2e220
--- /dev/null
+++ b/dns/quic/_asyncio.py
@@ -0,0 +1,206 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+import asyncio
+import socket
+import ssl
+import struct
+import time
+
+import aioquic.quic.configuration  # type: ignore
+import aioquic.quic.connection  # type: ignore
+import aioquic.quic.events  # type: ignore
+import dns.inet
+import dns.asyncbackend
+
+from dns.quic._common import (
+    BaseQuicStream,
+    AsyncQuicConnection,
+    AsyncQuicManager,
+    QUIC_MAX_DATAGRAM,
+)
+
+
+class AsyncioQuicStream(BaseQuicStream):
+    def __init__(self, connection, stream_id):
+        super().__init__(connection, stream_id)
+        self._wake_up = asyncio.Condition()
+
+    async def _wait_for_wake_up(self):
+        async with self._wake_up:
+            await self._wake_up.wait()
+
+    async def wait_for(self, amount, expiration):
+        timeout = self._timeout_from_expiration(expiration)
+        while True:
+            if self._buffer.have(amount):
+                return
+            self._expecting = amount
+            try:
+                await asyncio.wait_for(self._wait_for_wake_up(), timeout)
+            except Exception:
+                pass
+            self._expecting = 0
+
+    async def receive(self, timeout=None):
+        expiration = self._expiration_from_timeout(timeout)
+        await self.wait_for(2, expiration)
+        (size,) = struct.unpack("!H", self._buffer.get(2))
+        await self.wait_for(size, expiration)
+        return self._buffer.get(size)
+
+    async def send(self, datagram, is_end=False):
+        data = self._encapsulate(datagram)
+        await self._connection.write(self._stream_id, data, is_end)
+
+    async def _add_input(self, data, is_end):
+        if self._common_add_input(data, is_end):
+            async with self._wake_up:
+                self._wake_up.notify()
+
+    async def close(self):
+        self._close()
+
+    # Streams are async context managers
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.close()
+        async with self._wake_up:
+            self._wake_up.notify()
+        return False
+
+
+class AsyncioQuicConnection(AsyncQuicConnection):
+    def __init__(self, connection, address, port, source, source_port, manager=None):
+        super().__init__(connection, address, port, source, source_port, manager)
+        self._socket = None
+        self._handshake_complete = asyncio.Event()
+        self._socket_created = asyncio.Event()
+        self._wake_timer = asyncio.Condition()
+        self._receiver_task = None
+        self._sender_task = None
+
+    async def _receiver(self):
+        try:
+            af = dns.inet.af_for_address(self._address)
+            backend = dns.asyncbackend.get_backend("asyncio")
+            self._socket = await backend.make_socket(
+                af, socket.SOCK_DGRAM, 0, self._source, self._peer
+            )
+            self._socket_created.set()
+            async with self._socket:
+                while not self._done:
+                    (datagram, address) = await self._socket.recvfrom(
+                        QUIC_MAX_DATAGRAM, None
+                    )
+                    if address[0] != self._peer[0] or address[1] != self._peer[1]:
+                        continue
+                    self._connection.receive_datagram(
+                        datagram, self._peer[0], time.time()
+                    )
+                    # Wake up the timer in case the sender is sleeping, as there may be
+                    # stuff to send now.
+                    async with self._wake_timer:
+                        self._wake_timer.notify_all()
+        except Exception:
+            pass
+
+    async def _wait_for_wake_timer(self):
+        async with self._wake_timer:
+            await self._wake_timer.wait()
+
+    async def _sender(self):
+        await self._socket_created.wait()
+        while not self._done:
+            datagrams = self._connection.datagrams_to_send(time.time())
+            for (datagram, address) in datagrams:
+                assert address == self._peer[0]
+                await self._socket.sendto(datagram, self._peer, None)
+            (expiration, interval) = self._get_timer_values()
+            try:
+                await asyncio.wait_for(self._wait_for_wake_timer(), interval)
+            except Exception:
+                pass
+            self._handle_timer(expiration)
+            await self._handle_events()
+
+    async def _handle_events(self):
+        count = 0
+        while True:
+            event = self._connection.next_event()
+            if event is None:
+                return
+            if isinstance(event, aioquic.quic.events.StreamDataReceived):
+                stream = self._streams.get(event.stream_id)
+                if stream:
+                    await stream._add_input(event.data, event.end_stream)
+            elif isinstance(event, aioquic.quic.events.HandshakeCompleted):
+                self._handshake_complete.set()
+            elif isinstance(
+                event, aioquic.quic.events.ConnectionTerminated
+            ) or isinstance(event, aioquic.quic.events.StreamReset):
+                self._done = True
+                self._receiver_task.cancel()
+            count += 1
+            if count > 10:
+                # yield
+                count = 0
+                await asyncio.sleep(0)
+
+    async def write(self, stream, data, is_end=False):
+        self._connection.send_stream_data(stream, data, is_end)
+        async with self._wake_timer:
+            self._wake_timer.notify_all()
+
+    def run(self):
+        if self._closed:
+            return
+        self._receiver_task = asyncio.Task(self._receiver())
+        self._sender_task = asyncio.Task(self._sender())
+
+    async def make_stream(self):
+        await self._handshake_complete.wait()
+        stream_id = self._connection.get_next_available_stream_id(False)
+        stream = AsyncioQuicStream(self, stream_id)
+        self._streams[stream_id] = stream
+        return stream
+
+    async def close(self):
+        if not self._closed:
+            self._manager.closed(self._peer[0], self._peer[1])
+            self._closed = True
+            self._connection.close()
+            async with self._wake_timer:
+                self._wake_timer.notify_all()
+            try:
+                await self._receiver_task
+            except asyncio.CancelledError:
+                pass
+            try:
+                await self._sender_task
+            except asyncio.CancelledError:
+                pass
+
+
+class AsyncioQuicManager(AsyncQuicManager):
+    def __init__(self, conf=None, verify_mode=ssl.CERT_REQUIRED):
+        super().__init__(conf, verify_mode, AsyncioQuicConnection)
+
+    def connect(self, address, port=853, source=None, source_port=0):
+        (connection, start) = self._connect(address, port, source, source_port)
+        if start:
+            connection.run()
+        return connection
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        # Copy the itertor into a list as exiting things will mutate the connections
+        # table.
+        connections = list(self._connections.values())
+        for connection in connections:
+            await connection.close()
+        return False
diff --git a/dns/quic/_common.py b/dns/quic/_common.py
new file mode 100644
index 0000000..2b14c23
--- /dev/null
+++ b/dns/quic/_common.py
@@ -0,0 +1,181 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+import socket
+import struct
+import time
+
+import aioquic.quic.configuration  # type: ignore
+import aioquic.quic.connection  # type: ignore
+import dns.inet
+from dns._asyncbackend import NullContext
+
+from typing import Any
+
+
+QUIC_MAX_DATAGRAM = 2048
+
+
+class UnexpectedEOF(Exception):
+    pass
+
+
+class Buffer:
+    def __init__(self):
+        self._buffer = b""
+        self._seen_end = False
+
+    def put(self, data, is_end):
+        if self._seen_end:
+            return
+        self._buffer += data
+        if is_end:
+            self._seen_end = True
+
+    def have(self, amount):
+        if len(self._buffer) >= amount:
+            return True
+        if self._seen_end:
+            raise UnexpectedEOF
+        return False
+
+    def seen_end(self):
+        return self._seen_end
+
+    def get(self, amount):
+        assert self.have(amount)
+        data = self._buffer[:amount]
+        self._buffer = self._buffer[amount:]
+        return data
+
+
+class BaseQuicStream:
+    def __init__(self, connection, stream_id):
+        self._connection = connection
+        self._stream_id = stream_id
+        self._buffer = Buffer()
+        self._expecting = 0
+
+    def id(self):
+        return self._stream_id
+
+    def _expiration_from_timeout(self, timeout):
+        if timeout is not None:
+            expiration = time.time() + timeout
+        else:
+            expiration = None
+        return expiration
+
+    def _timeout_from_expiration(self, expiration):
+        if expiration is not None:
+            timeout = max(expiration - time.time(), 0.0)
+        else:
+            timeout = None
+        return timeout
+
+    # Subclass must implement receive() as sync / async and which returns a message
+    # or raises UnexpectedEOF.
+
+    def _encapsulate(self, datagram):
+        l = len(datagram)
+        return struct.pack("!H", l) + datagram
+
+    def _common_add_input(self, data, is_end):
+        self._buffer.put(data, is_end)
+        return self._expecting > 0 and self._buffer.have(self._expecting)
+
+    def _close(self):
+        self._connection.close_stream(self._stream_id)
+        self._buffer.put(b"", True)  # send EOF in case we haven't seen it.
+
+
+class BaseQuicConnection:
+    def __init__(
+        self, connection, address, port, source=None, source_port=0, manager=None
+    ):
+        self._done = False
+        self._connection = connection
+        self._address = address
+        self._port = port
+        self._closed = False
+        self._manager = manager
+        self._streams = {}
+        self._af = dns.inet.af_for_address(address)
+        self._peer = dns.inet.low_level_address_tuple((address, port))
+        if source is None and source_port != 0:
+            if self._af == socket.AF_INET:
+                source = "0.0.0.0"
+            elif self._af == socket.AF_INET6:
+                source = "::"
+            else:
+                raise NotImplementedError
+        if source:
+            self._source = (source, source_port)
+        else:
+            self._source = None
+
+    def close_stream(self, stream_id):
+        del self._streams[stream_id]
+
+    def _get_timer_values(self, closed_is_special=True):
+        now = time.time()
+        expiration = self._connection.get_timer()
+        if expiration is None:
+            expiration = now + 3600  # arbitrary "big" value
+        interval = max(expiration - now, 0)
+        if self._closed and closed_is_special:
+            # lower sleep interval to avoid a race in the closing process
+            # which can lead to higher latency closing due to sleeping when
+            # we have events.
+            interval = min(interval, 0.05)
+        return (expiration, interval)
+
+    def _handle_timer(self, expiration):
+        now = time.time()
+        if expiration <= now:
+            self._connection.handle_timer(now)
+
+
+class AsyncQuicConnection(BaseQuicConnection):
+    async def make_stream(self) -> Any:
+        pass
+
+
+class BaseQuicManager:
+    def __init__(self, conf, verify_mode, connection_factory):
+        self._connections = {}
+        self._connection_factory = connection_factory
+        if conf is None:
+            verify_path = None
+            if isinstance(verify_mode, str):
+                verify_path = verify_mode
+                verify_mode = True
+            conf = aioquic.quic.configuration.QuicConfiguration(
+                alpn_protocols=["doq", "doq-i03"],
+                verify_mode=verify_mode,
+            )
+            if verify_path is not None:
+                conf.load_verify_locations(verify_path)
+        self._conf = conf
+
+    def _connect(self, address, port=853, source=None, source_port=0):
+        connection = self._connections.get((address, port))
+        if connection is not None:
+            return (connection, False)
+        qconn = aioquic.quic.connection.QuicConnection(configuration=self._conf)
+        qconn.connect(address, time.time())
+        connection = self._connection_factory(
+            qconn, address, port, source, source_port, self
+        )
+        self._connections[(address, port)] = connection
+        return (connection, True)
+
+    def closed(self, address, port):
+        try:
+            del self._connections[(address, port)]
+        except KeyError:
+            pass
+
+
+class AsyncQuicManager(BaseQuicManager):
+    def connect(self, address, port=853, source=None, source_port=0):
+        raise NotImplementedError
diff --git a/dns/quic/_sync.py b/dns/quic/_sync.py
new file mode 100644
index 0000000..be005ba
--- /dev/null
+++ b/dns/quic/_sync.py
@@ -0,0 +1,214 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+import socket
+import ssl
+import selectors
+import struct
+import threading
+import time
+
+import aioquic.quic.configuration  # type: ignore
+import aioquic.quic.connection  # type: ignore
+import aioquic.quic.events  # type: ignore
+import dns.inet
+
+from dns.quic._common import (
+    BaseQuicStream,
+    BaseQuicConnection,
+    BaseQuicManager,
+    QUIC_MAX_DATAGRAM,
+)
+
+# Avoid circularity with dns.query
+if hasattr(selectors, "PollSelector"):
+    _selector_class = selectors.PollSelector  # type: ignore
+else:
+    _selector_class = selectors.SelectSelector  # type: ignore
+
+
+class SyncQuicStream(BaseQuicStream):
+    def __init__(self, connection, stream_id):
+        super().__init__(connection, stream_id)
+        self._wake_up = threading.Condition()
+        self._lock = threading.Lock()
+
+    def wait_for(self, amount, expiration):
+        timeout = self._timeout_from_expiration(expiration)
+        while True:
+            with self._lock:
+                if self._buffer.have(amount):
+                    return
+                self._expecting = amount
+            with self._wake_up:
+                self._wake_up.wait(timeout)
+            self._expecting = 0
+
+    def receive(self, timeout=None):
+        expiration = self._expiration_from_timeout(timeout)
+        self.wait_for(2, expiration)
+        with self._lock:
+            (size,) = struct.unpack("!H", self._buffer.get(2))
+        self.wait_for(size, expiration)
+        with self._lock:
+            return self._buffer.get(size)
+
+    def send(self, datagram, is_end=False):
+        data = self._encapsulate(datagram)
+        self._connection.write(self._stream_id, data, is_end)
+
+    def _add_input(self, data, is_end):
+        if self._common_add_input(data, is_end):
+            with self._wake_up:
+                self._wake_up.notify()
+
+    def close(self):
+        with self._lock:
+            self._close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+        with self._wake_up:
+            self._wake_up.notify()
+        return False
+
+
+class SyncQuicConnection(BaseQuicConnection):
+    def __init__(self, connection, address, port, source, source_port, manager):
+        super().__init__(connection, address, port, source, source_port, manager)
+        self._socket = socket.socket(self._af, socket.SOCK_DGRAM, 0)
+        self._socket.connect(self._peer)
+        (self._send_wakeup, self._receive_wakeup) = socket.socketpair()
+        self._receive_wakeup.setblocking(False)
+        self._socket.setblocking(False)
+        if self._source is not None:
+            try:
+                self._socket.bind(
+                    dns.inet.low_level_address_tuple(self._source, self._af)
+                )
+            except Exception:
+                self._socket.close()
+                raise
+        self._handshake_complete = threading.Event()
+        self._worker_thread = None
+        self._lock = threading.Lock()
+
+    def _read(self):
+        count = 0
+        while count < 10:
+            count += 1
+            try:
+                datagram = self._socket.recv(QUIC_MAX_DATAGRAM)
+            except BlockingIOError:
+                return
+            with self._lock:
+                self._connection.receive_datagram(datagram, self._peer[0], time.time())
+
+    def _drain_wakeup(self):
+        while True:
+            try:
+                self._receive_wakeup.recv(32)
+            except BlockingIOError:
+                return
+
+    def _worker(self):
+        sel = _selector_class()
+        sel.register(self._socket, selectors.EVENT_READ, self._read)
+        sel.register(self._receive_wakeup, selectors.EVENT_READ, self._drain_wakeup)
+        while not self._done:
+            (expiration, interval) = self._get_timer_values(False)
+            items = sel.select(interval)
+            for (key, _) in items:
+                key.data()
+            with self._lock:
+                self._handle_timer(expiration)
+                datagrams = self._connection.datagrams_to_send(time.time())
+            for (datagram, _) in datagrams:
+                try:
+                    self._socket.send(datagram)
+                except BlockingIOError:
+                    # we let QUIC handle any lossage
+                    pass
+            self._handle_events()
+
+    def _handle_events(self):
+        while True:
+            with self._lock:
+                event = self._connection.next_event()
+            if event is None:
+                return
+            if isinstance(event, aioquic.quic.events.StreamDataReceived):
+                with self._lock:
+                    stream = self._streams.get(event.stream_id)
+                if stream:
+                    stream._add_input(event.data, event.end_stream)
+            elif isinstance(event, aioquic.quic.events.HandshakeCompleted):
+                self._handshake_complete.set()
+            elif isinstance(
+                event, aioquic.quic.events.ConnectionTerminated
+            ) or isinstance(event, aioquic.quic.events.StreamReset):
+                with self._lock:
+                    self._done = True
+
+    def write(self, stream, data, is_end=False):
+        with self._lock:
+            self._connection.send_stream_data(stream, data, is_end)
+        self._send_wakeup.send(b"\x01")
+
+    def run(self):
+        if self._closed:
+            return
+        self._worker_thread = threading.Thread(target=self._worker)
+        self._worker_thread.start()
+
+    def make_stream(self):
+        self._handshake_complete.wait()
+        with self._lock:
+            stream_id = self._connection.get_next_available_stream_id(False)
+            stream = SyncQuicStream(self, stream_id)
+            self._streams[stream_id] = stream
+        return stream
+
+    def close_stream(self, stream_id):
+        with self._lock:
+            super().close_stream(stream_id)
+
+    def close(self):
+        with self._lock:
+            if self._closed:
+                return
+            self._manager.closed(self._peer[0], self._peer[1])
+            self._closed = True
+            self._connection.close()
+            self._send_wakeup.send(b"\x01")
+        self._worker_thread.join()
+
+
+class SyncQuicManager(BaseQuicManager):
+    def __init__(self, conf=None, verify_mode=ssl.CERT_REQUIRED):
+        super().__init__(conf, verify_mode, SyncQuicConnection)
+        self._lock = threading.Lock()
+
+    def connect(self, address, port=853, source=None, source_port=0):
+        with self._lock:
+            (connection, start) = self._connect(address, port, source, source_port)
+            if start:
+                connection.run()
+            return connection
+
+    def closed(self, address, port):
+        with self._lock:
+            super().closed(address, port)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        # Copy the itertor into a list as exiting things will mutate the connections
+        # table.
+        connections = list(self._connections.values())
+        for connection in connections:
+            connection.close()
+        return False
diff --git a/dns/quic/_trio.py b/dns/quic/_trio.py
new file mode 100644
index 0000000..09b6950
--- /dev/null
+++ b/dns/quic/_trio.py
@@ -0,0 +1,170 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+import socket
+import ssl
+import struct
+import time
+
+import aioquic.quic.configuration  # type: ignore
+import aioquic.quic.connection  # type: ignore
+import aioquic.quic.events  # type: ignore
+import dns.inet
+import trio
+
+from dns.quic._common import (
+    BaseQuicStream,
+    AsyncQuicConnection,
+    AsyncQuicManager,
+    NullContext,
+    QUIC_MAX_DATAGRAM,
+)
+
+
+class TrioQuicStream(BaseQuicStream):
+    def __init__(self, connection, stream_id):
+        super().__init__(connection, stream_id)
+        self._wake_up = trio.Condition()
+
+    async def wait_for(self, amount):
+        while True:
+            if self._buffer.have(amount):
+                return
+            self._expecting = amount
+            async with self._wake_up:
+                await self._wake_up.wait()
+            self._expecting = 0
+
+    async def receive(self, timeout=None):
+        if timeout is None:
+            context = NullContext(None)
+        else:
+            context = trio.move_on_after(timeout)
+        with context:
+            await self.wait_for(2)
+            (size,) = struct.unpack("!H", self._buffer.get(2))
+            await self.wait_for(size)
+            return self._buffer.get(size)
+
+    async def send(self, datagram, is_end=False):
+        data = self._encapsulate(datagram)
+        await self._connection.write(self._stream_id, data, is_end)
+
+    async def _add_input(self, data, is_end):
+        if self._common_add_input(data, is_end):
+            async with self._wake_up:
+                self._wake_up.notify()
+
+    async def close(self):
+        self._close()
+
+    # Streams are async context managers
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.close()
+        async with self._wake_up:
+            self._wake_up.notify()
+        return False
+
+
+class TrioQuicConnection(AsyncQuicConnection):
+    def __init__(self, connection, address, port, source, source_port, manager=None):
+        super().__init__(connection, address, port, source, source_port, manager)
+        self._socket = trio.socket.socket(self._af, socket.SOCK_DGRAM, 0)
+        if self._source:
+            trio.socket.bind(dns.inet.low_level_address_tuple(self._source, self._af))
+        self._handshake_complete = trio.Event()
+        self._run_done = trio.Event()
+        self._worker_scope = None
+
+    async def _worker(self):
+        await self._socket.connect(self._peer)
+        while not self._done:
+            (expiration, interval) = self._get_timer_values(False)
+            with trio.CancelScope(
+                deadline=trio.current_time() + interval
+            ) as self._worker_scope:
+                datagram = await self._socket.recv(QUIC_MAX_DATAGRAM)
+                self._connection.receive_datagram(datagram, self._peer[0], time.time())
+            self._worker_scope = None
+            self._handle_timer(expiration)
+            datagrams = self._connection.datagrams_to_send(time.time())
+            for (datagram, _) in datagrams:
+                await self._socket.send(datagram)
+            await self._handle_events()
+
+    async def _handle_events(self):
+        count = 0
+        while True:
+            event = self._connection.next_event()
+            if event is None:
+                return
+            if isinstance(event, aioquic.quic.events.StreamDataReceived):
+                stream = self._streams.get(event.stream_id)
+                if stream:
+                    await stream._add_input(event.data, event.end_stream)
+            elif isinstance(event, aioquic.quic.events.HandshakeCompleted):
+                self._handshake_complete.set()
+            elif isinstance(
+                event, aioquic.quic.events.ConnectionTerminated
+            ) or isinstance(event, aioquic.quic.events.StreamReset):
+                self._done = True
+                self._socket.close()
+            count += 1
+            if count > 10:
+                # yield
+                count = 0
+                await trio.sleep(0)
+
+    async def write(self, stream, data, is_end=False):
+        self._connection.send_stream_data(stream, data, is_end)
+        if self._worker_scope is not None:
+            self._worker_scope.cancel()
+
+    async def run(self):
+        if self._closed:
+            return
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(self._worker)
+        self._run_done.set()
+
+    async def make_stream(self):
+        await self._handshake_complete.wait()
+        stream_id = self._connection.get_next_available_stream_id(False)
+        stream = TrioQuicStream(self, stream_id)
+        self._streams[stream_id] = stream
+        return stream
+
+    async def close(self):
+        if not self._closed:
+            self._manager.closed(self._peer[0], self._peer[1])
+            self._closed = True
+            self._connection.close()
+            if self._worker_scope is not None:
+                self._worker_scope.cancel()
+            await self._run_done.wait()
+
+
+class TrioQuicManager(AsyncQuicManager):
+    def __init__(self, nursery, conf=None, verify_mode=ssl.CERT_REQUIRED):
+        super().__init__(conf, verify_mode, TrioQuicConnection)
+        self._nursery = nursery
+
+    def connect(self, address, port=853, source=None, source_port=0):
+        (connection, start) = self._connect(address, port, source, source_port)
+        if start:
+            self._nursery.start_soon(connection.run)
+        return connection
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        # Copy the itertor into a list as exiting things will mutate the connections
+        # table.
+        connections = list(self._connections.values())
+        for connection in connections:
+            await connection.close()
+        return False