DexDemo/frontend/venv/lib/python3.12/site-packages/dns/win32util.py

import sys

import dns._features

# pylint: disable=W0612,W0613,C0301

if sys.platform == "win32":
    import ctypes
    import ctypes.wintypes as wintypes
    import winreg  # pylint: disable=import-error
    from enum import IntEnum

    import dns.name

    # Keep pylint quiet on non-windows.
    try:
        _ = WindowsError  # pylint: disable=used-before-assignment
    except NameError:
        WindowsError = Exception

    class ConfigMethod(IntEnum):
        Registry = 1
        WMI = 2
        Win32 = 3

    class DnsInfo:
        def __init__(self):
            self.domain = None
            self.nameservers = []
            self.search = []

    _config_method = ConfigMethod.Registry

    if dns._features.have("wmi"):
        import threading

        import pythoncom  # pylint: disable=import-error
        import wmi  # pylint: disable=import-error

        # Prefer WMI by default if wmi is installed.
        _config_method = ConfigMethod.WMI

        class _WMIGetter(threading.Thread):
            # pylint: disable=possibly-used-before-assignment
            def __init__(self):
                super().__init__()
                self.info = DnsInfo()

            def run(self):
                pythoncom.CoInitialize()
                try:
                    system = wmi.WMI()
                    for interface in system.Win32_NetworkAdapterConfiguration():
                        if interface.IPEnabled and interface.DNSServerSearchOrder:
                            self.info.nameservers = list(interface.DNSServerSearchOrder)
                            if interface.DNSDomain:
                                self.info.domain = _config_domain(interface.DNSDomain)
                            if interface.DNSDomainSuffixSearchOrder:
                                self.info.search = [
                                    _config_domain(x)
                                    for x in interface.DNSDomainSuffixSearchOrder
                                ]
                            break
                finally:
                    pythoncom.CoUninitialize()

            def get(self):
                # We always run in a separate thread to avoid any issues with
                # the COM threading model.
                self.start()
                self.join()
                return self.info

    else:

        class _WMIGetter:  # type: ignore
            pass

    def _config_domain(domain):
        # Sometimes DHCP servers add a '.' prefix to the default domain, and
        # Windows just stores such values in the registry (see #687).
        # Check for this and fix it.
        if domain.startswith("."):
            domain = domain[1:]
        return dns.name.from_text(domain)

    class _RegistryGetter:
        def __init__(self):
            self.info = DnsInfo()

        def _split(self, text):
            # The windows registry has used both " " and "," as a delimiter, and while
            # it is currently using "," in Windows 10 and later, updates can seemingly
            # leave a space in too, e.g. "a, b".  So we just convert all commas to
            # spaces, and use split() in its default configuration, which splits on
            # all whitespace and ignores empty strings.
            return text.replace(",", " ").split()

        def _config_nameservers(self, nameservers):
            for ns in self._split(nameservers):
                if ns not in self.info.nameservers:
                    self.info.nameservers.append(ns)

        def _config_search(self, search):
            for s in self._split(search):
                s = _config_domain(s)
                if s not in self.info.search:
                    self.info.search.append(s)

        def _config_fromkey(self, key, always_try_domain):
            try:
                servers, _ = winreg.QueryValueEx(key, "NameServer")
            except WindowsError:
                servers = None
            if servers:
                self._config_nameservers(servers)
            if servers or always_try_domain:
                try:
                    dom, _ = winreg.QueryValueEx(key, "Domain")
                    if dom:
                        self.info.domain = _config_domain(dom)
                except WindowsError:
                    pass
            else:
                try:
                    servers, _ = winreg.QueryValueEx(key, "DhcpNameServer")
                except WindowsError:
                    servers = None
                if servers:
                    self._config_nameservers(servers)
                    try:
                        dom, _ = winreg.QueryValueEx(key, "DhcpDomain")
                        if dom:
                            self.info.domain = _config_domain(dom)
                    except WindowsError:
                        pass
            try:
                search, _ = winreg.QueryValueEx(key, "SearchList")
            except WindowsError:
                search = None
            if search is None:
                try:
                    search, _ = winreg.QueryValueEx(key, "DhcpSearchList")
                except WindowsError:
                    search = None
            if search:
                self._config_search(search)

        def _is_nic_enabled(self, lm, guid):
            # Look in the Windows Registry to determine whether the network
            # interface corresponding to the given guid is enabled.
            #
            # (Code contributed by Paul Marks, thanks!)
            #
            try:
                # This hard-coded location seems to be consistent, at least
                # from Windows 2000 through Vista.
                connection_key = winreg.OpenKey(
                    lm,
                    r"SYSTEM\CurrentControlSet\Control\Network"
                    r"\{4D36E972-E325-11CE-BFC1-08002BE10318}"
                    rf"\{guid}\Connection",
                )

                try:
                    # The PnpInstanceID points to a key inside Enum
                    (pnp_id, ttype) = winreg.QueryValueEx(
                        connection_key, "PnpInstanceID"
                    )

                    if ttype != winreg.REG_SZ:
                        raise ValueError  # pragma: no cover

                    device_key = winreg.OpenKey(
                        lm, rf"SYSTEM\CurrentControlSet\Enum\{pnp_id}"
                    )

                    try:
                        # Get ConfigFlags for this device
                        (flags, ttype) = winreg.QueryValueEx(device_key, "ConfigFlags")

                        if ttype != winreg.REG_DWORD:
                            raise ValueError  # pragma: no cover

                        # Based on experimentation, bit 0x1 indicates that the
                        # device is disabled.
                        #
                        # XXXRTH I suspect we really want to & with 0x03 so
                        # that CONFIGFLAGS_REMOVED devices are also ignored,
                        # but we're shifting to WMI as ConfigFlags is not
                        # supposed to be used.
                        return not flags & 0x1

                    finally:
                        device_key.Close()
                finally:
                    connection_key.Close()
            except Exception:  # pragma: no cover
                return False

        def get(self):
            """Extract resolver configuration from the Windows registry."""

            lm = winreg.ConnectRegistry(None, winreg.HKEY_LOCAL_MACHINE)
            try:
                tcp_params = winreg.OpenKey(
                    lm, r"SYSTEM\CurrentControlSet\Services\Tcpip\Parameters"
                )
                try:
                    self._config_fromkey(tcp_params, True)
                finally:
                    tcp_params.Close()
                interfaces = winreg.OpenKey(
                    lm,
                    r"SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces",
                )
                try:
                    i = 0
                    while True:
                        try:
                            guid = winreg.EnumKey(interfaces, i)
                            i += 1
                            key = winreg.OpenKey(interfaces, guid)
                            try:
                                if not self._is_nic_enabled(lm, guid):
                                    continue
                                self._config_fromkey(key, False)
                            finally:
                                key.Close()
                        except OSError:
                            break
                finally:
                    interfaces.Close()
            finally:
                lm.Close()
            return self.info

    class _Win32Getter(_RegistryGetter):

        def get(self):
            """Get the attributes using the Windows API."""
            # Load the IP Helper library
            # # https://learn.microsoft.com/en-us/windows/win32/api/iphlpapi/nf-iphlpapi-getadaptersaddresses
            IPHLPAPI = ctypes.WinDLL("Iphlpapi.dll")

            # Constants
            AF_UNSPEC = 0
            ERROR_SUCCESS = 0
            GAA_FLAG_INCLUDE_PREFIX = 0x00000010
            AF_INET = 2
            AF_INET6 = 23
            IF_TYPE_SOFTWARE_LOOPBACK = 24

            # Define necessary structures
            class SOCKADDRV4(ctypes.Structure):
                _fields_ = [
                    ("sa_family", wintypes.USHORT),
                    ("sa_data", ctypes.c_ubyte * 14),
                ]

            class SOCKADDRV6(ctypes.Structure):
                _fields_ = [
                    ("sa_family", wintypes.USHORT),
                    ("sa_data", ctypes.c_ubyte * 26),
                ]

            class SOCKET_ADDRESS(ctypes.Structure):
                _fields_ = [
                    ("lpSockaddr", ctypes.POINTER(SOCKADDRV4)),
                    ("iSockaddrLength", wintypes.INT),
                ]

            class IP_ADAPTER_DNS_SERVER_ADDRESS(ctypes.Structure):
                pass  # Forward declaration

            IP_ADAPTER_DNS_SERVER_ADDRESS._fields_ = [
                ("Length", wintypes.ULONG),
                ("Reserved", wintypes.DWORD),
                ("Next", ctypes.POINTER(IP_ADAPTER_DNS_SERVER_ADDRESS)),
                ("Address", SOCKET_ADDRESS),
            ]

            class IF_LUID(ctypes.Structure):
                _fields_ = [("Value", ctypes.c_ulonglong)]

            class NET_IF_NETWORK_GUID(ctypes.Structure):
                _fields_ = [("Value", ctypes.c_ubyte * 16)]

            class IP_ADAPTER_PREFIX_XP(ctypes.Structure):
                pass  # Left undefined here for simplicity

            class IP_ADAPTER_GATEWAY_ADDRESS_LH(ctypes.Structure):
                pass  # Left undefined here for simplicity

            class IP_ADAPTER_DNS_SUFFIX(ctypes.Structure):
                _fields_ = [
                    ("String", ctypes.c_wchar * 256),
                    ("Next", ctypes.POINTER(ctypes.c_void_p)),
                ]

            class IP_ADAPTER_UNICAST_ADDRESS_LH(ctypes.Structure):
                pass  # Left undefined here for simplicity

            class IP_ADAPTER_MULTICAST_ADDRESS_XP(ctypes.Structure):
                pass  # Left undefined here for simplicity

            class IP_ADAPTER_ANYCAST_ADDRESS_XP(ctypes.Structure):
                pass  # Left undefined here for simplicity

            class IP_ADAPTER_DNS_SERVER_ADDRESS_XP(ctypes.Structure):
                pass  # Left undefined here for simplicity

            class IP_ADAPTER_ADDRESSES(ctypes.Structure):
                pass  # Forward declaration

            IP_ADAPTER_ADDRESSES._fields_ = [
                ("Length", wintypes.ULONG),
                ("IfIndex", wintypes.DWORD),
                ("Next", ctypes.POINTER(IP_ADAPTER_ADDRESSES)),
                ("AdapterName", ctypes.c_char_p),
                ("FirstUnicastAddress", ctypes.POINTER(SOCKET_ADDRESS)),
                ("FirstAnycastAddress", ctypes.POINTER(SOCKET_ADDRESS)),
                ("FirstMulticastAddress", ctypes.POINTER(SOCKET_ADDRESS)),
                (
                    "FirstDnsServerAddress",
                    ctypes.POINTER(IP_ADAPTER_DNS_SERVER_ADDRESS),
                ),
                ("DnsSuffix", wintypes.LPWSTR),
                ("Description", wintypes.LPWSTR),
                ("FriendlyName", wintypes.LPWSTR),
                ("PhysicalAddress", ctypes.c_ubyte * 8),
                ("PhysicalAddressLength", wintypes.ULONG),
                ("Flags", wintypes.ULONG),
                ("Mtu", wintypes.ULONG),
                ("IfType", wintypes.ULONG),
                ("OperStatus", ctypes.c_uint),
                # Remaining fields removed for brevity
            ]

            def format_ipv4(sockaddr_in):
                return ".".join(map(str, sockaddr_in.sa_data[2:6]))

            def format_ipv6(sockaddr_in6):
                # The sa_data is:
                #
                # USHORT    sin6_port;
                # ULONG     sin6_flowinfo;
                # IN6_ADDR  sin6_addr;
                # ULONG     sin6_scope_id;
                #
                # which is 2 + 4 + 16 + 4 = 26 bytes, and we need the plus 6 below
                # to be in the sin6_addr range.
                parts = [
                    sockaddr_in6.sa_data[i + 6] << 8 | sockaddr_in6.sa_data[i + 6 + 1]
                    for i in range(0, 16, 2)
                ]
                return ":".join(f"{part:04x}" for part in parts)

            buffer_size = ctypes.c_ulong(15000)
            while True:
                buffer = ctypes.create_string_buffer(buffer_size.value)

                ret_val = IPHLPAPI.GetAdaptersAddresses(
                    AF_UNSPEC,
                    GAA_FLAG_INCLUDE_PREFIX,
                    None,
                    buffer,
                    ctypes.byref(buffer_size),
                )

                if ret_val == ERROR_SUCCESS:
                    break
                elif ret_val != 0x6F:  # ERROR_BUFFER_OVERFLOW
                    print(f"Error retrieving adapter information: {ret_val}")
                    return

            adapter_addresses = ctypes.cast(
                buffer, ctypes.POINTER(IP_ADAPTER_ADDRESSES)
            )

            current_adapter = adapter_addresses
            while current_adapter:

                # Skip non-operational adapters.
                oper_status = current_adapter.contents.OperStatus
                if oper_status != 1:
                    current_adapter = current_adapter.contents.Next
                    continue

                # Exclude loopback adapters.
                if current_adapter.contents.IfType == IF_TYPE_SOFTWARE_LOOPBACK:
                    current_adapter = current_adapter.contents.Next
                    continue

                # Get the domain from the DnsSuffix attribute.
                dns_suffix = current_adapter.contents.DnsSuffix
                if dns_suffix:
                    self.info.domain = dns.name.from_text(dns_suffix)

                current_dns_server = current_adapter.contents.FirstDnsServerAddress
                while current_dns_server:
                    sockaddr = current_dns_server.contents.Address.lpSockaddr
                    sockaddr_family = sockaddr.contents.sa_family

                    ip = None
                    if sockaddr_family == AF_INET:  # IPv4
                        ip = format_ipv4(sockaddr.contents)
                    elif sockaddr_family == AF_INET6:  # IPv6
                        sockaddr = ctypes.cast(sockaddr, ctypes.POINTER(SOCKADDRV6))
                        ip = format_ipv6(sockaddr.contents)

                    if ip:
                        if ip not in self.info.nameservers:
                            self.info.nameservers.append(ip)

                    current_dns_server = current_dns_server.contents.Next

                current_adapter = current_adapter.contents.Next

            # Use the registry getter to get the search info, since it is set at the system level.
            registry_getter = _RegistryGetter()
            info = registry_getter.get()
            self.info.search = info.search
            return self.info

    def set_config_method(method: ConfigMethod) -> None:
        global _config_method
        _config_method = method

    def get_dns_info() -> DnsInfo:
        """Extract resolver configuration."""
        if _config_method == ConfigMethod.Win32:
            getter = _Win32Getter()
        elif _config_method == ConfigMethod.WMI:
            getter = _WMIGetter()
        else:
            getter = _RegistryGetter()
        return getter.get()