codekingpro/portable-devtools

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license2#3# Support for Discovery of Designated Resolvers4 5import socket6import time7from urllib.parse import urlparse8 9import dns.asyncbackend10import dns.inet11import dns.name12import dns.nameserver13import dns.query14import dns.rdtypes.svcbbase15 16# The special name of the local resolver when using DDR17_local_resolver_name = dns.name.from_text("_dns.resolver.arpa")18 19 20#21# Processing is split up into I/O independent and I/O dependent parts to22# make supporting sync and async versions easy.23#24 25 26class _SVCBInfo:27    def __init__(self, bootstrap_address, port, hostname, nameservers):28        self.bootstrap_address = bootstrap_address29        self.port = port30        self.hostname = hostname31        self.nameservers = nameservers32 33    def ddr_check_certificate(self, cert):34        """Verify that the _SVCBInfo's address is in the cert's subjectAltName (SAN)"""35        for name, value in cert["subjectAltName"]:36            if name == "IP Address" and value == self.bootstrap_address:37                return True38        return False39 40    def make_tls_context(self):41        ssl = dns.query.ssl42        ctx = ssl.create_default_context()43        ctx.minimum_version = ssl.TLSVersion.TLSv1_244        return ctx45 46    def ddr_tls_check_sync(self, lifetime):47        ctx = self.make_tls_context()48        expiration = time.time() + lifetime49        with socket.create_connection(50            (self.bootstrap_address, self.port), lifetime51        ) as s:52            with ctx.wrap_socket(s, server_hostname=self.hostname) as ts:53                ts.settimeout(dns.query._remaining(expiration))54                ts.do_handshake()55                cert = ts.getpeercert()56                return self.ddr_check_certificate(cert)57 58    async def ddr_tls_check_async(self, lifetime, backend=None):59        if backend is None:60            backend = dns.asyncbackend.get_default_backend()61        ctx = self.make_tls_context()62        expiration = time.time() + lifetime63        async with await backend.make_socket(64            dns.inet.af_for_address(self.bootstrap_address),65            socket.SOCK_STREAM,66            0,67            None,68            (self.bootstrap_address, self.port),69            lifetime,70            ctx,71            self.hostname,72        ) as ts:73            cert = await ts.getpeercert(dns.query._remaining(expiration))74            return self.ddr_check_certificate(cert)75 76 77def _extract_nameservers_from_svcb(answer):78    bootstrap_address = answer.nameserver79    if not dns.inet.is_address(bootstrap_address):80        return []81    infos = []82    for rr in answer.rrset.processing_order():83        nameservers = []84        param = rr.params.get(dns.rdtypes.svcbbase.ParamKey.ALPN)85        if param is None:86            continue87        alpns = set(param.ids)88        host = rr.target.to_text(omit_final_dot=True)89        port = None90        param = rr.params.get(dns.rdtypes.svcbbase.ParamKey.PORT)91        if param is not None:92            port = param.port93        # For now we ignore address hints and address resolution and always use the94        # bootstrap address95        if b"h2" in alpns:96            param = rr.params.get(dns.rdtypes.svcbbase.ParamKey.DOHPATH)97            if param is None or not param.value.endswith(b"{?dns}"):98                continue99            path = param.value[:-6].decode()100            if not path.startswith("/"):101                path = "/" + path102            if port is None:103                port = 443104            url = f"https://{host}:{port}{path}"105            # check the URL106            try:107                urlparse(url)108                nameservers.append(dns.nameserver.DoHNameserver(url, bootstrap_address))109            except Exception:110                # continue processing other ALPN types111                pass112        if b"dot" in alpns:113            if port is None:114                port = 853115            nameservers.append(116                dns.nameserver.DoTNameserver(bootstrap_address, port, host)117            )118        if b"doq" in alpns:119            if port is None:120                port = 853121            nameservers.append(122                dns.nameserver.DoQNameserver(bootstrap_address, port, True, host)123            )124        if len(nameservers) > 0:125            infos.append(_SVCBInfo(bootstrap_address, port, host, nameservers))126    return infos127 128 129def _get_nameservers_sync(answer, lifetime):130    """Return a list of TLS-validated resolver nameservers extracted from an SVCB131    answer."""132    nameservers = []133    infos = _extract_nameservers_from_svcb(answer)134    for info in infos:135        try:136            if info.ddr_tls_check_sync(lifetime):137                nameservers.extend(info.nameservers)138        except Exception:139            pass140    return nameservers141 142 143async def _get_nameservers_async(answer, lifetime):144    """Return a list of TLS-validated resolver nameservers extracted from an SVCB145    answer."""146    nameservers = []147    infos = _extract_nameservers_from_svcb(answer)148    for info in infos:149        try:150            if await info.ddr_tls_check_async(lifetime):151                nameservers.extend(info.nameservers)152        except Exception:153            pass154    return nameservers155