* Add basic DDR support.
async def getsockname(self):
raise NotImplementedError
+ async def getpeercert(self, timeout):
+ raise NotImplementedError
+
async def __aenter__(self):
return self
async def getsockname(self):
return self.transport.get_extra_info("sockname")
+ async def getpeercert(self, timeout):
+ raise NotImplementedError
+
class StreamSocket(dns._asyncbackend.StreamSocket):
def __init__(self, af, reader, writer):
async def getsockname(self):
return self.writer.get_extra_info("sockname")
+ async def getpeercert(self, timeout):
+ return self.writer.get_extra_info("peercert")
+
try:
import anyio
--- /dev/null
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+#
+# Support for Discovery of Designated Resolvers
+
+from urllib.parse import urlparse
+import socket
+import time
+
+import dns.asyncbackend
+import dns.inet
+import dns.name
+import dns.nameserver
+import dns.query
+import dns.rdtypes.svcbbase
+
+
+# The special name of the local resolver when using DDR
+_local_resolver_name = dns.name.from_text("_dns.resolver.arpa")
+
+
+#
+# Processing is split up into I/O independent and I/O dependent parts to
+# make supporting sync and async versions easy.
+#
+
+
+class _SVCBInfo:
+ def __init__(self, bootstrap_address, port, hostname, nameservers):
+ self.bootstrap_address = bootstrap_address
+ self.port = port
+ self.hostname = hostname
+ self.nameservers = nameservers
+
+ def ddr_check_certificate(self, cert):
+ """Verify that the _SVCBInfo's address is in the cert's subjectAltName (SAN)"""
+ for name, value in cert["subjectAltName"]:
+ if name == "IP Address" and value == self.bootstrap_address:
+ return True
+ return False
+
+ def make_tls_context(self):
+ ssl = dns.query.ssl
+ ctx = ssl.create_default_context()
+ ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+ return ctx
+
+ def ddr_tls_check_sync(self, lifetime):
+ ctx = self.make_tls_context()
+ expiration = time.time() + lifetime
+ with socket.create_connection(
+ (self.bootstrap_address, self.port), lifetime
+ ) as s:
+ with ctx.wrap_socket(s, server_hostname=self.hostname) as ts:
+ ts.settimeout(dns.query._remaining(expiration))
+ ts.do_handshake()
+ cert = ts.getpeercert()
+ return self.ddr_check_certificate(cert)
+
+ async def ddr_tls_check_async(self, lifetime, backend=None):
+ if backend is None:
+ backend = dns.asyncbackend.get_default_backend()
+ ctx = self.make_tls_context()
+ expiration = time.time() + lifetime
+ async with await backend.make_socket(
+ dns.inet.af_for_address(self.bootstrap_address),
+ socket.SOCK_STREAM,
+ 0,
+ None,
+ (self.bootstrap_address, self.port),
+ lifetime,
+ ctx,
+ self.hostname,
+ ) as ts:
+ cert = await ts.getpeercert(dns.query._remaining(expiration))
+ return self.ddr_check_certificate(cert)
+
+
+def _extract_nameservers_from_svcb(answer):
+ bootstrap_address = answer.nameserver
+ if not dns.inet.is_address(bootstrap_address):
+ return []
+ infos = []
+ for rr in answer.rrset.processing_order():
+ nameservers = []
+ param = rr.params.get(dns.rdtypes.svcbbase.ParamKey.ALPN)
+ if param is None:
+ continue
+ alpns = set(param.ids)
+ host = rr.target.to_text(omit_final_dot=True)
+ port = None
+ param = rr.params.get(dns.rdtypes.svcbbase.ParamKey.PORT)
+ if param is not None:
+ port = param.port
+ # For now we ignore address hints and address resolution and always use the
+ # bootstrap address
+ if b"h2" in alpns:
+ param = rr.params.get(dns.rdtypes.svcbbase.ParamKey.DOHPATH)
+ if param is None or not param.value.endswith(b"{?dns}"):
+ continue
+ path = param.value[:-6].decode()
+ if not path.startswith("/"):
+ path = "/" + path
+ if port is None:
+ port = 443
+ url = f"https://{host}:{port}{path}"
+ # check the URL
+ try:
+ urlparse(url)
+ nameservers.append(dns.nameserver.DoHNameserver(url, bootstrap_address))
+ except Exception:
+ # continue processing other ALPN types
+ pass
+ if b"dot" in alpns:
+ if port is None:
+ port = 853
+ nameservers.append(
+ dns.nameserver.DoTNameserver(bootstrap_address, port, host)
+ )
+ if b"doq" in alpns:
+ if port is None:
+ port = 853
+ nameservers.append(
+ dns.nameserver.DoQNameserver(bootstrap_address, port, True, host)
+ )
+ if len(nameservers) > 0:
+ infos.append(_SVCBInfo(bootstrap_address, port, host, nameservers))
+ return infos
+
+
+def _get_nameservers_sync(answer, lifetime):
+ """Return a list of TLS-validated resolver nameservers extracted from an SVCB
+ answer."""
+ nameservers = []
+ infos = _extract_nameservers_from_svcb(answer)
+ for info in infos:
+ try:
+ if info.ddr_tls_check_sync(lifetime):
+ nameservers.extend(info.nameservers)
+ except Exception:
+ pass
+ return nameservers
+
+
+async def _get_nameservers_async(answer, lifetime):
+ """Return a list of TLS-validated resolver nameservers extracted from an SVCB
+ answer."""
+ nameservers = []
+ infos = _extract_nameservers_from_svcb(answer)
+ for info in infos:
+ try:
+ if await info.ddr_tls_check_async(lifetime):
+ nameservers.extend(info.nameservers)
+ except Exception:
+ pass
+ return nameservers
async def getsockname(self):
return self.socket.getsockname()
+ async def getpeercert(self, timeout):
+ raise NotImplementedError
+
class StreamSocket(dns._asyncbackend.StreamSocket):
def __init__(self, family, stream, tls=False):
else:
return self.stream.socket.getsockname()
+ async def getpeercert(self, timeout):
+ if self.tls:
+ with _maybe_timeout(timeout):
+ await self.stream.do_handshake()
+ return self.stream.getpeercert()
+ else:
+ raise NotImplementedError
+
try:
import httpx
import dns.asyncbackend
import dns.asyncquery
+import dns._ddr
import dns.exception
import dns.name
import dns.query
canonical_name = e.canonical_name
return canonical_name
+ async def try_ddr(self, lifetime: float = 5.0) -> None:
+ """Try to update the resolver's nameservers using Discovery of Designated
+ Resolvers (DDR). If successful, the resolver will subsequently use
+ DNS-over-HTTPS or DNS-over-TLS for future queries.
+
+ *lifetime*, a float, is the maximum time to spend attempting DDR. The default
+ is 5 seconds.
+
+ If the SVCB query is successful and results in a non-empty list of nameservers,
+ then the resolver's nameservers are set to the returned servers in priority
+ order.
+
+ The current implementation does not use any address hints from the SVCB record,
+ nor does it resolve addresses for the SCVB target name, rather it assumes that
+ the bootstrap nameserver will always be one of the addresses and uses it.
+ A future revision to the code may offer fuller support. The code verifies that
+ the bootstrap nameserver is in the Subject Alternative Name field of the
+ TLS certficate.
+ """
+ try:
+ expiration = time.time() + lifetime
+ answer = await self.resolve(
+ dns._ddr._local_resolver_name, "svcb", lifetime=lifetime
+ )
+ timeout = dns.query._remaining(expiration)
+ nameservers = await dns._ddr._get_nameservers_async(answer, timeout)
+ if len(nameservers) > 0:
+ self.nameservers = nameservers
+ except Exception:
+ pass
+
default_resolver = None
return await get_default_resolver().canonical_name(name)
+async def try_ddr(timeout: float = 5.0) -> None:
+ """Try to update the default resolver's nameservers using Discovery of Designated
+ Resolvers (DDR). If successful, the resolver will subsequently use
+ DNS-over-HTTPS or DNS-over-TLS for future queries.
+
+ See :py:func:`dns.resolver.Resolver.try_ddr` for more information.
+ """
+ return await get_default_resolver().try_ddr(timeout)
+
+
async def zone_for_name(
name: Union[dns.name.Name, str],
rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
def __str__(self):
raise NotImplementedError
+ def kind(self) -> str:
+ raise NotImplementedError
+
def is_always_max_size(self) -> bool:
raise NotImplementedError
self.url = url
self.bootstrap_address = bootstrap_address
+ def kind(self):
+ return "DoH"
+
def is_always_max_size(self) -> bool:
return True
IPV4HINT = 4
ECH = 5
IPV6HINT = 6
+ DOHPATH = 7
@classmethod
def _maximum(cls):
from urllib.parse import urlparse
import contextlib
+import random
import socket
import sys
import threading
import time
-import random
import warnings
+import dns._ddr
import dns.exception
import dns.edns
import dns.flags
import dns.rcode
import dns.rdataclass
import dns.rdatatype
+import dns.rdtypes.svcbbase
import dns.reversename
import dns.tsig
# pylint: enable=redefined-outer-name
+ def try_ddr(self, lifetime: float = 5.0) -> None:
+ """Try to update the resolver's nameservers using Discovery of Designated
+ Resolvers (DDR). If successful, the resolver will subsequently use
+ DNS-over-HTTPS or DNS-over-TLS for future queries.
+
+ *lifetime*, a float, is the maximum time to spend attempting DDR. The default
+ is 5 seconds.
+
+ If the SVCB query is successful and results in a non-empty list of nameservers,
+ then the resolver's nameservers are set to the returned servers in priority
+ order.
+
+ The current implementation does not use any address hints from the SVCB record,
+ nor does it resolve addresses for the SCVB target name, rather it assumes that
+ the bootstrap nameserver will always be one of the addresses and uses it.
+ A future revision to the code may offer fuller support. The code verifies that
+ the bootstrap nameserver is in the Subject Alternative Name field of the
+ TLS certficate.
+ """
+ try:
+ expiration = time.time() + lifetime
+ answer = self.resolve(
+ dns._ddr._local_resolver_name, "SVCB", lifetime=lifetime
+ )
+ timeout = dns.query._remaining(expiration)
+ nameservers = dns._ddr._get_nameservers_sync(answer, timeout)
+ if len(nameservers) > 0:
+ self.nameservers = nameservers
+ except Exception:
+ pass
+
#: The default resolver.
default_resolver: Optional[Resolver] = None
return get_default_resolver().canonical_name(name)
+def try_ddr(lifetime: float = 5.0) -> None:
+ """Try to update the default resolver's nameservers using Discovery of Designated
+ Resolvers (DDR). If successful, the resolver will subsequently use
+ DNS-over-HTTPS or DNS-over-TLS for future queries.
+
+ See :py:func:`dns.resolver.Resolver.try_ddr` for more information.
+ """
+ return get_default_resolver().try_ddr(lifetime)
+
+
def zone_for_name(
name: Union[dns.name.Name, str],
rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
.. autofunction:: dns.asyncresolver.resolve_address
.. autofunction:: dns.asyncresolver.resolve_name
.. autofunction:: dns.asyncresolver.canonical_name
+.. autofunction:: dns.asyncresolver.try_ddr
.. autofunction:: dns.asyncresolver.zone_for_name
.. autodata:: dns.asyncresolver.default_resolver
.. autofunction:: dns.asyncresolver.get_default_resolver
.. autofunction:: dns.resolver.resolve_address
.. autofunction:: dns.resolver.resolve_name
.. autofunction:: dns.resolver.canonical_name
+.. autofunction:: dns.resolver.try_ddr
.. autofunction:: dns.resolver.zone_for_name
.. autofunction:: dns.resolver.query
.. autodata:: dns.resolver.default_resolver
* DNSSEC zone signing with NSEC records is now supported. Thank you
very much (again!) Jakob Schlyter!
+* The resolver and async resolver now have the ``try_ddr()`` method, which will try to
+ use Discovery of Designated Resolvers (DDR) to upgrade the connection from the stub
+ resolver to the recursive server so that it uses DNS-over-HTTPS, DNS-over-TLS, or
+ DNS-over-QUIC. This feature is currently experimental as the standard is still in
+ draft stage, although the DDR has been deployed already.
+
* Curio support has been removed.
2.3.0
--- /dev/null
+#!/usr/bin/env python3
+
+# Using Discovery of Designated Resolvers (synchronous I/O)
+
+import dns.resolver
+
+res = dns.resolver.Resolver(configure=False)
+res.nameservers = ["1.1.1.1"]
+# Invoke try_ddr() to attempt to upgrade the connection via DDR
+res.try_ddr()
+# Do a sample resolution
+for rr in res.resolve("www.google.com", "A"):
+ print(rr.address)
+# Note that the nameservers have been upgraded
+print(res.nameservers)
+
+
+# Using Discovery of Designated Resolvers (asynchronous I/O)
+
+# We show using asyncio, but if you comment out asyncio lines
+# and uncomment the trio lines, it will work with trio too.
+
+import asyncio
+
+# import trio
+
+import dns.asyncresolver
+
+
+async def amain():
+ res = dns.asyncresolver.Resolver(configure=False)
+ res.nameservers = ["8.8.8.8"]
+ await res.try_ddr()
+
+ for rr in await res.resolve("www.google.com", "A"):
+ print(rr.address)
+
+ print(res.nameservers)
+
+
+asyncio.run(amain())
+# trio.run(amain)
--- /dev/null
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+import asyncio
+import time
+
+import pytest
+
+import dns.asyncbackend
+import dns.asyncresolver
+import dns.resolver
+import dns.nameserver
+
+import tests.util
+
+
+@pytest.mark.skipif(
+ not tests.util.is_internet_reachable(), reason="Internet not reachable"
+)
+def test_basic_ddr_sync():
+ for nameserver in ["1.1.1.1", "8.8.8.8"]:
+ res = dns.resolver.Resolver(configure=False)
+ res.nameservers = [nameserver]
+ res.try_ddr()
+ for nameserver in res.nameservers:
+ assert isinstance(nameserver, dns.nameserver.Nameserver)
+ assert nameserver.kind() != "Do53"
+
+
+@pytest.mark.skipif(
+ not tests.util.is_internet_reachable(), reason="Internet not reachable"
+)
+def test_basic_ddr_async():
+ async def run():
+ dns.asyncbackend._default_backend = None
+ for nameserver in ["1.1.1.1", "8.8.8.8"]:
+ res = dns.asyncresolver.Resolver(configure=False)
+ res.nameservers = [nameserver]
+ await res.try_ddr()
+ for nameserver in res.nameservers:
+ assert isinstance(nameserver, dns.nameserver.Nameserver)
+ assert nameserver.kind() != "Do53"
+
+ asyncio.run(run())
projects / thirdparty / dnspython.git / commitdiff
