17046250819

Committed 18 Aug 2025 04:12PM UTC coverage: 96.648% (+1.9%) from 94.721%

Build # 17046250819

Build Type

push

github

Committed by

funilrys

Commit Message

Bump verstion to v4.3.0

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

154 existing lines in 14 files now uncovered.

11967 of 12382 relevant lines covered (96.65%)

8.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.66

/PyFunceble/query/dns/query_tool.py

"""
The tool to check the availability or syntax of domain, IP or URL.

::


    ██████╗ ██╗   ██╗███████╗██╗   ██╗███╗   ██╗ ██████╗███████╗██████╗ ██╗     ███████╗
    ██╔══██╗╚██╗ ██╔╝██╔════╝██║   ██║████╗  ██║██╔════╝██╔════╝██╔══██╗██║     ██╔════╝
    ██████╔╝ ╚████╔╝ █████╗  ██║   ██║██╔██╗ ██║██║     █████╗  ██████╔╝██║     █████╗
    ██╔═══╝   ╚██╔╝  ██╔══╝  ██║   ██║██║╚██╗██║██║     ██╔══╝  ██╔══██╗██║     ██╔══╝
    ██║        ██║   ██║     ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
    ╚═╝        ╚═╝   ╚═╝      ╚═════╝ ╚═╝  ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝

Provides an interface for the query.

Author:
    Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom

Special thanks:
    https://pyfunceble.github.io/#/special-thanks

Contributors:
    https://pyfunceble.github.io/#/contributors

Project link:
    https://github.com/funilrys/PyFunceble

Project documentation:
    https://docs.pyfunceble.com

Project homepage:
    https://pyfunceble.github.io/

License:
::


    Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024, 2025 Nissar Chababy

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        https://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

# pylint: disable=too-many-lines

import copy
import functools
import ipaddress
import random
import socket
import time
from typing import Dict, List, Optional, Union

import dns.exception
import dns.message
import dns.name
import dns.query
import dns.rdataclass
import dns.rdatatype

import PyFunceble.facility
import PyFunceble.storage
from PyFunceble.query.dns.nameserver import Nameservers
from PyFunceble.query.record.dns import DNSQueryToolRecord


class DNSQueryTool:
    """
    Provides our query tool.
    """

    # pylint: disable=too-many-public-methods

    STD_PROTOCOL: str = "UDP"
    STD_TIMEOUT: float = 5.0
    STD_FOLLOW_NAMESERVER_ORDER: bool = True
    STD_TRUST_SERVER: bool = False
    STD_DELAY: float = 0.0

    SUPPORTED_PROTOCOL: List[str] = ["TCP", "UDP", "HTTPS", "TLS"]

    value2rdata_type: Dict[int, str] = {
        x.value: x.name for x in dns.rdatatype.RdataType
    }
    rdata_type2value: Dict[str, int] = {
        x.name: x.value for x in dns.rdatatype.RdataType
    }

    nameservers: Optional[Nameservers] = None
    _query_record_type: int = dns.rdatatype.RdataType.ANY

    _subject: Optional[str] = None
    _follow_nameserver_order: bool = True
    _preferred_protocol: str = "UDP"
    _query_timeout: float = 5.0
    _trust_server: bool = False
    _delay: float = 0.0

    dns_name: Optional[str] = None

    query_message: Optional[dns.message.QueryMessage] = None
    lookup_record: Optional[DNSQueryToolRecord] = None

    def __init__(
        self,
        *,
        nameservers: Optional[List[str]] = None,
        follow_nameserver_order: Optional[bool] = None,
        preferred_protocol: Optional[str] = None,
        trust_server: Optional[bool] = None,
        delay: Optional[bool] = None,
    ) -> None:
        self.nameservers = Nameservers()

        if preferred_protocol is not None:
            self.preferred_protocol = preferred_protocol
        else:
            self.guess_and_set_preferred_protocol()

        if nameservers is not None:
            self.nameservers.set_nameservers(nameservers)
        else:  # pragma: no cover ## I'm not playing with system resolver.
            self.nameservers.guess_and_set_nameservers()

        if follow_nameserver_order is not None:
            self.follow_nameserver_order = follow_nameserver_order
        else:
            self.guess_and_set_follow_nameserver_order()

        if trust_server is not None:
            self.trust_server = trust_server
        else:
            self.guess_and_set_trust_server()

        if delay is not None:
            self.delay = delay
        else:
            self.guess_and_set_delay()

    def prepare_query(func):  # pylint: disable=no-self-argument
        """
        Prepare the query after running the decorated method.
        """

        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            result = func(self, *args, **kwargs)  # pylint: disable=not-callable

            if self.subject and self.query_record_type:
                self.dns_name = self.get_dns_name_from_subject_and_query_type()

                if self.dns_name:
                    self.query_message = dns.message.make_query(
                        self.dns_name, self.query_record_type
                    )
                else:
                    self.query_message = None

            return result

        return wrapper

    def update_lookup_record(func):  # pylint: disable=no-self-argument
        """
        Ensures that a clean record is generated after the execution of
        the decorated method.
        """

        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            result = func(self, *args, **kwargs)  # pylint: disable=not-callable

            if self.lookup_record is None or self.subject != self.lookup_record.subject:
                self.lookup_record = DNSQueryToolRecord()
                self.lookup_record.subject = self.subject

            if self.dns_name != self.lookup_record.dns_name:
                self.lookup_record.dns_name = self.dns_name

            if (
                self.get_human_query_record_type()
                != self.lookup_record.query_record_type
            ):
                self.lookup_record.query_record_type = (
                    self.get_human_query_record_type()
                )

            if (
                self.follow_nameserver_order
                != self.lookup_record.follow_nameserver_order
            ):
                self.lookup_record.follow_nameserver_order = (
                    self.follow_nameserver_order
                )

            if self.query_timeout != self.lookup_record.query_timeout:
                self.lookup_record.query_timeout = self.query_timeout

            if self.preferred_protocol != self.lookup_record.preferred_protocol:
                self.lookup_record.preferred_protocol = self.preferred_protocol

            return result

        return wrapper

    def ensure_subject_is_given(func):  # pylint: disable=no-self-argument
        """
        Ensures that the subject to work with is given before running the
        decorated method.

        :raise TypeError:
            If :code:`self.subject` is not a :py:class:`str`.
        """

        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            if not isinstance(self.subject, str):
                raise TypeError(
                    f"<self.subject> should be {str}, {type(self.subject)} given."
                )

            return func(self, *args, **kwargs)  # pylint: disable=not-callable

        return wrapper

    def update_lookup_record_response(func):  # pylint: disable=no-self-argument
        """
        Ensures that the response of the decorated method is set as response
        in our record.
        """

        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            result = func(self, *args, **kwargs)  # pylint: disable=not-callable

            if result != self.lookup_record.response:
                self.lookup_record.response = result

            return result

        return wrapper

    def ignore_if_query_message_is_missing(func):  # pylint: disable=no-self-argument
        """
        Ignores the call to the decorated method if the query message is
        missing. Otherwise, return an empty list.
        """

        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            if self.query_message:
                return func(self, *args, **kwargs)  # pylint: disable=not-callable
            return []  # pragma: no cover ## Safety

        return wrapper

    @ensure_subject_is_given
    def get_dns_name_from_subject_and_query_type(self):
        """
        Provides the dns name based on the current subject and query type.
        """

        try:
            if self.get_human_query_record_type().lower() == "ptr":
                try:
                    return dns.name.from_text(
                        ipaddress.ip_address(self.subject).reverse_pointer
                    )
                except ValueError:
                    return dns.name.from_text(self.subject)
            return dns.name.from_text(self.subject)
        except (
            dns.name.LabelTooLong,
            dns.name.EmptyLabel,
            dns.name.BadEscape,
            dns.name.NameTooLong,
        ):
            return None

    @property
    def subject(self) -> Optional[str]:
        """
        Provides the current state of the :code:`_subject` attribute.
        """

        return self._subject

    @subject.setter
    @prepare_query
    @update_lookup_record
    def subject(self, value: str) -> None:
        """
        Sets the subject to work with.

        :param value:
            The subject to set.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`str`.
        :raise ValueError:
            When the given :code:`value` is empty.
        """

        if not isinstance(value, str):
            raise TypeError(f"<value> should be {str}, {type(value)} given.")

        if not value:
            raise ValueError("<value> should not be empty.")

        self._subject = value

    def set_subject(self, value: str) -> "DNSQueryTool":
        """
        Sets the subject to work with.

        :param value:
            The subject to set.
        """

        self.subject = value

        return self

    def set_nameservers(
        self, value: List[str]
    ) -> "DNSQueryTool":  # pragma: no cover ## Underlying already tested.
        """
        Sets the nameservers to work with.

        :raise TypeError:
            When the given :code:`value` is not a list of :py:class:`str`.
        :raise ValueError:
            When the given :code:`value` is empty.
        """

        self.nameservers.set_nameservers(value)

        return self

    @property
    def follow_nameserver_order(self) -> bool:
        """
        Provides the current state of the :code:`_follow_nameserver_order`
        attribute.
        """

        return self._follow_nameserver_order

    @follow_nameserver_order.setter
    @update_lookup_record
    def follow_nameserver_order(self, value: bool) -> None:
        """
        Updates the :code:`follow_nameserver_order` variable.

        :param value:
            The value to set.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`bool`.
        """

        if not isinstance(value, bool):
            raise TypeError(f"<value> should be {bool}, {type(value)} given.")

        self._follow_nameserver_order = value

    def set_follow_nameserver_order(self, value: bool) -> "DNSQueryTool":
        """
        Updates the :code:`follow_nameserver_order` variable.

        :param value:
            The value to set.
        """

        self.follow_nameserver_order = value

        return self

    @property
    def query_record_type(self) -> int:
        """
        Provides the current state of the :code:`_query_record_type` attribute.
        """

        return self._query_record_type

    @query_record_type.setter
    @prepare_query
    @update_lookup_record
    def query_record_type(self, value: Union[str, int]) -> None:
        """
        Sets the DNS record type to query.

        :param value:
            The value to set. It can be the human version (e.g AAAA) or an
            integer as registered in the :code:`value2rdata_type` attribute.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`str` nor
            :py:class:`int`.
        :raise ValueError:
            When the given :code:`value` is unknown or unsupported.
        """

        if not isinstance(value, (str, int)):
            raise TypeError(f"<value> should be {int} or {str}, {type(value)} given.")

        if value in self.rdata_type2value:
            self._query_record_type = self.rdata_type2value[value]
        elif value in self.value2rdata_type:
            self._query_record_type = value
        else:
            raise ValueError(f"<value> ({value!r}) is unknown or unsupported.")

    def set_query_record_type(self, value: Union[str, int]) -> "DNSQueryTool":
        """
        Sets the DNS record type to query.

        :param value:
            The value to set. It can be the human version (e.g AAAA) or an
            integer as registered in the :code:`value2rdata_type` attribute.
        """

        self.query_record_type = value

        return self

    def get_human_query_record_type(self) -> str:
        """
        Provides the currently set record type.
        """

        return self.value2rdata_type[self.query_record_type]

    @property
    def query_timeout(self) -> float:
        """
        Provides the current state of the :code:`_query_timeout` attribute.
        """

        return self._query_timeout

    @query_timeout.setter
    @update_lookup_record
    def query_timeout(self, value: Union[int, float]) -> None:
        """
        Sets the timeout to apply.

        :param value:
            The timeout to apply.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`float`
            nor :py:class.`int`.
        """

        if not isinstance(value, (float, int)):
            raise TypeError(f"<value> should be {float} or {int}, {type(value)} given.")

        self._query_timeout = float(value)

    def set_timeout(self, value: Union[int, float]) -> "DNSQueryTool":
        """
        Sets the timeout to apply.

        :param value:
            The timeout to apply.
        """

        self.query_timeout = value

        return self

    @property
    def trust_server(self) -> Optional[bool]:
        """
        Provides the current state of the :code:`trust_server` attribute.
        """

        return self._trust_server

    @trust_server.setter
    def trust_server(self, value: bool) -> None:
        """
        Sets the value to apply.

        :param value:
            The value to apply.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`bool`.
        """

        if not isinstance(value, bool):
            raise TypeError(f"<value> should be {bool}, {type(value)} given.")

        self._trust_server = value

    def set_trust_server(self, value: bool) -> "DNSQueryTool":
        """
        Sets the value to apply.

        :param value:
            The value to apply.
        """

        self.trust_server = value

        return self

    def guess_and_set_timeout(self) -> "DNSQueryTool":
        """
        Try to guess and set the timeout.
        """

        if PyFunceble.facility.ConfigLoader.is_already_loaded():
            if PyFunceble.storage.CONFIGURATION.lookup.timeout:
                self.query_timeout = PyFunceble.storage.CONFIGURATION.lookup.timeout
            else:
                self.query_timeout = self.STD_TIMEOUT
        else:
            self.query_timeout = self.STD_TIMEOUT

        return self

    @property
    def preferred_protocol(self) -> Optional[str]:
        """
        Provides the current state of the :code:`_preferred_protocol` attribute.
        """

        return self._preferred_protocol

    @preferred_protocol.setter
    def preferred_protocol(self, value: str) -> None:
        """
        Sets the preferred protocol.

        :param value:
            The protocol to use.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`str`.
        :raise ValueError:
            When the given :code:`value` is unknown or unsupported.
        """

        if not isinstance(value, str):
            raise TypeError(f"<value> should be {str}, {type(value)} given.")

        value = value.upper()

        if value not in self.SUPPORTED_PROTOCOL:
            raise ValueError(
                f"<value> {value!r} is unknown or unsupported "
                f"(supported: {self.SUPPORTED_PROTOCOL!r})."
            )

        self._preferred_protocol = self.nameservers.protocol = value

    def set_preferred_protocol(self, value: str) -> "DNSQueryTool":
        """
        Sets the preferred protocol.

        :param value:
            The protocol to use.
        """

        self.preferred_protocol = value

        return self

    @property
    def delay(self) -> float:
        """
        Provides the current state of the :code:`_delay` attribute.
        """

        return self._delay

    @delay.setter
    @update_lookup_record
    def delay(self, value: Union[int, float]) -> None:
        """
        Sets the delay to apply.

        :param value:
            The delay to apply.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`float`
            nor :py:class.`int`.
        :raise ValueError:
            When the given :code:`value` is not a positive.
        """

        if not isinstance(value, (float, int)):
            raise TypeError(f"<value> should be {float} or {int}, {type(value)} given.")

        if value < 0:
            raise ValueError(f"<value> should be positive, {value} given.")

        self._delay = float(value)

    def set_delay(self, value: Union[int, float]) -> "DNSQueryTool":
        """
        Sets the delay to apply.

        :param value:
            The delay to apply.
        """

        self.delay = value

        return self

    def guess_and_set_preferred_protocol(self) -> "DNSQueryTool":
        """
        Try to guess and set the preferred procol.
        """

        if PyFunceble.facility.ConfigLoader.is_already_loaded():
            if isinstance(PyFunceble.storage.CONFIGURATION.dns.protocol, str):
                self.preferred_protocol = PyFunceble.storage.CONFIGURATION.dns.protocol
            else:
                self.preferred_protocol = self.STD_PROTOCOL
        else:
            self.preferred_protocol = self.STD_PROTOCOL

        return self

    def guess_and_set_follow_nameserver_order(self) -> "DNSQueryTool":
        """
        Try to guess and authorize the mix of the nameserver before each
        query.
        """

        if PyFunceble.facility.ConfigLoader.is_already_loaded():
            if isinstance(
                PyFunceble.storage.CONFIGURATION.dns.follow_server_order, bool
            ):
                self.follow_nameserver_order = (
                    PyFunceble.storage.CONFIGURATION.dns.follow_server_order
                )
            else:
                self.follow_nameserver_order = self.STD_FOLLOW_NAMESERVER_ORDER
        else:
            self.follow_nameserver_order = self.STD_FOLLOW_NAMESERVER_ORDER

        return self

    def guess_and_set_trust_server(self) -> "DNSQueryTool":
        """
        Try to guess and set the trust flag.
        """

        if PyFunceble.facility.ConfigLoader.is_already_loaded():
            if isinstance(PyFunceble.storage.CONFIGURATION.dns.trust_server, bool):
                self.trust_server = PyFunceble.storage.CONFIGURATION.dns.trust_server
            else:
                self.trust_server = self.STD_TRUST_SERVER
        else:
            self.trust_server = self.STD_TRUST_SERVER

        return self

    def guess_and_set_delay(self) -> "DNSQueryTool":
        """
        Try to guess and set the delay to apply.
        """

        if PyFunceble.facility.ConfigLoader.is_already_loaded():
            if PyFunceble.storage.CONFIGURATION.dns.delay:
                self.delay = PyFunceble.storage.CONFIGURATION.dns.delay
            else:
                self.delay = self.STD_DELAY
        else:
            self.delay = self.STD_DELAY

    def guess_all_settings(
        self,
    ) -> "DNSQueryTool":  # pragma: no cover ## Method themselves are more important
        """
        Try to guess all settings.
        """

        to_ignore = ["guess_all_settings"]

        for method in dir(self):
            if method in to_ignore or not method.startswith("guess_"):
                continue

            getattr(self, method)()

        return self

    def get_lookup_record(
        self,
    ) -> Optional[DNSQueryToolRecord]:
        """
        Provides the current query record.
        """

        return self.lookup_record

    def _get_result_from_response(
        self, response: dns.message.Message
    ) -> List[str]:  # pragma: no cover ## This just reads upstream result
        """
        Given a response, we return the best possible result.
        """

        result = []

        rrset = response.get_rrset(
            response.answer,
            self.dns_name,
            dns.rdataclass.RdataClass.IN,
            self.query_record_type,
        )

        if rrset:
            result.extend([x.to_text() for x in rrset])

        PyFunceble.facility.Logger.debug("Result from response:\r%r", result)

        return result

    def _mix_order(
        self, data: Union[dict, List[str]]
    ) -> Union[dict, List[str]]:  # pragma: no cover ## Just a shuffle :-)
        """
        Given a dataset, we mix its order.
        """

        dataset = copy.deepcopy(data)

        if not self.follow_nameserver_order:
            if isinstance(dataset, list):
                random.shuffle(dataset)

                return dataset

            if isinstance(dataset, dict):
                temp = list(dataset.items())
                random.shuffle(temp)

                return dict(temp)

        PyFunceble.facility.Logger.debug("Mixed data:\n%r", dataset)
        return dataset

    def _query_protocol(self, protocol: str) -> Optional[List[str]]:
        """
        Given a protocol, we query the nameservers with or through it.

        :param str protocol:
            The protocol to use for the query.

            It can be one of the following:
                - "TCP"
                - "UDP"
                - "HTTPS"
                - "TLS"
        :return:
            The result of the query, or an empty list if the query failed.
        """

        self.lookup_record.used_protocol = protocol.upper()

        if self.lookup_record.used_protocol not in self.SUPPORTED_PROTOCOL:
            raise ValueError(
                f"<protocol> ({self.lookup_record.used_protocol!r}) is unknown or "
                f"unsupported (supported: {self.SUPPORTED_PROTOCOL!r})."
            )

        result = set()

        for nameserver, port in self._mix_order(
            self.nameservers.get_nameserver_ports()
        ).items():
            PyFunceble.facility.Logger.debug(
                "Started to query information of %r from %r (%s)",
                self.subject,
                nameserver,
                self.lookup_record.used_protocol,
            )

            if self.lookup_record.used_protocol == "HTTPS" and port == 53:
                # Default port for nameserver class is 53. So we ensure we
                # overwrite with our own default.
                port = 443

            if self.lookup_record.used_protocol == "TLS" and port == 53:
                # Default port for nameserver class is 53. So we ensure we
                # overwrite with our own default.
                port = 853

            try:
                response = getattr(dns.query, protocol.lower())(
                    self.query_message,
                    nameserver,
                    port=port,
                    timeout=self.query_timeout,
                )

                local_result = self._get_result_from_response(response)

                if local_result:
                    result.update(local_result)

                    self.lookup_record.nameserver = nameserver
                    self.lookup_record.port = port

                    PyFunceble.facility.Logger.debug(
                        "Successfully queried information of %r from %r. (%s)",
                        self.subject,
                        nameserver,
                        self.lookup_record.used_protocol,
                    )

                if self.trust_server:  # pragma: no cover: Per case.
                    break
            except (dns.exception.Timeout, socket.error):
                # Example: Resource temporarily unavailable.
                pass
            except dns.query.UnexpectedSource:
                # Example: got a response from XXX instead of XXX.
                pass
            except dns.query.BadResponse:
                # Example: A DNS query response does not respond to the question
                # asked.
                pass
            except ValueError:
                # Example: Input is malformed.
                break

            PyFunceble.facility.Logger.debug(
                "Unsuccessfully queried information of %r from %r (%s). Sleeping %fs.",
                self.subject,
                nameserver,
                self.lookup_record.used_protocol,
                self.delay,
            )

            time.sleep(self.delay)

        return list(result)

    @ensure_subject_is_given
    @ignore_if_query_message_is_missing
    @update_lookup_record_response
    def tcp(
        self,
    ) -> Optional[List[str]]:
        """
        Request the chosen record through the TCP protocol.
        """

        return self._query_protocol("TCP")

    @ensure_subject_is_given
    @ignore_if_query_message_is_missing
    @update_lookup_record_response
    def udp(
        self,
    ) -> Optional[List[str]]:
        """
        Request the chosen record through the UTP protocol.
        """

        return self._query_protocol("UDP")

    @ensure_subject_is_given
    @ignore_if_query_message_is_missing
    @update_lookup_record_response
    def https(
        self,
    ) -> Optional[List[str]]:
        """
        Request the chosen record through the https protocol.
        """

        return self._query_protocol("HTTPS")

    @ensure_subject_is_given
    @ignore_if_query_message_is_missing
    @update_lookup_record_response
    def tls(
        self,
    ) -> Optional[List[str]]:
        """
        Request the chosen record through the TLS protocol.
        """

        return self._query_protocol("TLS")

    def query(
        self,
    ) -> Optional[List[str]]:
        """
        Process the query based on the preferred protocol.
        """

        return getattr(self, self.preferred_protocol.lower())()

funilrys / PyFunceble / 17046250819

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous