dnsbench

#! /usr/bin/python
#
"""
DNS Benchmarking Script

This script is designed to benchmark multiple DNS servers by performing DNS
queries on randomly selected domains from a whitelist. The script fetches a
list of domains from a specified URL, tests each DNS server by querying a
random domain from the list, and calculates the average response time for each
server.

The script uses the following components:
  - **fetch_whitelist(url, timeout=10):** Fetches a list of domains from the
    specified URL with an optional timeout. The list is used for DNS query
    testing.

  - **get_random_domain(domains_list):** Selects and returns a random domain
    from the provided list, ensuring varied queries to avoid caching effects.

  - **run_dns_test(nameserver, domain):** Performs a DNS query using the
    specified nameserver for the given domain and returns the query time in
    ms.  Handles exceptions for timeouts, NXDOMAIN, and other DNS errors.

  - **test_nameserver(nameserver, domains_list, num_tests=10):** Tests a DNS
    nameserver by performing a specified number of DNS queries on randomly
    selected domains.  The function calculates and returns the average query
    time across all tests.
  - **main():** The entry point of the script. It fetches the whitelist, tests
    all configured DNS servers in parallel, and prints the average query time
    for each server.

  Usage:
  - The script can be run directly from the command line.
  - The output consists of the DNS server name and average query time in ms.

  Dependencies:
  - `requests` for fetching the domain whitelist from a URL.
  - `dnspython` for performing DNS queries.

  Example:
  dnsbench.py | sort -t: -k2,2n
"""

import random
import concurrent.futures
import statistics
import dns.resolver
import requests

# Constants
WHITELIST_URL_BASE = "https://raw.githubusercontent.com/anudeepND"
WHITELIST_URL = f"{WHITELIST_URL_BASE}/whitelist/master/domains/whitelist.txt"
NAMESERVERS = [
    ("10.6.0.10", "AdguardHome 1"),
    ("10.6.0.11", "AdguardHome 2"),
    ("10.6.0.12", "AdguardHome 3"),
    ("10.6.0.20", "Pihole 1     "),
    ("10.6.0.21", "Pihole 2     "),
    ("10.6.0.62", "Windows DNS 1"),
    ("10.6.0.63", "Windows DNS 2"),
    ("10.6.0.1", "Unbound 1     ")]


# Fetch the whitelist domains
def fetch_whitelist(url, timeout=10):
    """
    Fetches a list of domains from a specified URL and returns them as a list
    of strings.

    The function sends an HTTP GET request to the provided URL, retrieves the
    content, and splits it into individual lines. Each line represents a domain
    name that can be used for DNS query testing.

    Parameters:
        url (str): The URL from which to fetch the domain whitelist.

        Returns:
        list: A list of domain names from the URL, each as a separate string.

        Raises:
        requests.exceptions.RequestException: If there is an issue with the
        HTTP request, such as network problems or a non-successful HTTP status
        code.
    """
    response = requests.get(url, timeout=timeout)
    response.raise_for_status()
    return response.text.splitlines()


# Function to get a random domain from the whitelist
def get_random_domain(domains_list):
    """
    Selects and returns a random domain from the provided list of domains.

    This function is used to randomly choose a domain name for DNS query
    testing. The randomness helps in avoiding caching effects and ensures that
    different domains are tested across different DNS servers.

    Parameters:
    domains_list (list): A list of domain names to choose from.

    Returns:
    str: A randomly selected domain name from the list.
    """
    return random.choice(domains_list)


# Function to run a DNS query and return the query time
def run_dns_test(nameserver, domain):
    """
    Performs a DNS query for a given domain using a specified DNS nameserver
    and returns the time taken to receive a response.

    The function configures a DNS resolver to use the specified nameserver,
    then executes a DNS query for the given domain. If the query is successful,
    the time taken for the response is returned in milliseconds. If the query
    fails due to a timeout, non-existent domain, or other DNS errors, the
    function returns None.

    Parameters:
        nameserver (str): The IP address of the DNS server to query.
        domain (str): The domain name to query.

    Returns:
        float or None: The time taken on the query in ms, or None if failed.
    """
    resolver = dns.resolver.Resolver()
    resolver.nameservers = [nameserver]
    try:
        # Time the DNS query
        response = resolver.resolve(domain, 'A', lifetime=2.0)
        # If successful, return the response time in milliseconds
        return response.response.time * 1000
    except (dns.resolver.Timeout,
            dns.resolver.NXDOMAIN,
            dns.resolver.NoNameservers,
            dns.resolver.YXDOMAIN) as e:
        raise Exception(
            f"The DNS response does not contain an answer to the question: "
            f"{domain}. IN A"
        ) from e


# Function to test a nameserver and return the average response time
def test_nameserver(nameserver, domains_list, num_tests=10):
    """
    Tests a given DNS nameserver by performing DNS queries on a specified
    number of random domains. The function queries a random domain from the
    provided list, measures the query time, and repeats the process for a
    specified number of tests. If a DNS query fails, the function retries with
    a different random domain. Finally, the function calculates and returns the
    average query time across all tests.

    Parameters:
        nameserver (tuple): A tuple containing the IP and name of the server.
        domains_list (list): A list of domains to select from for DNS queries.
        num_tests (int): The number of tests to perform on the nameserver.

    Returns:
        float: The average DNS query time in milliseconds for the nameserver.
    """
    times = []
    for _ in range(num_tests):
        while True:
            domain = get_random_domain(domains_list)
            try:
                query_time = run_dns_test(nameserver[0], domain)
                if query_time is not None:
                    times.append(query_time)
                    break  # Successful query, move to the next test
            except Exception:
                continue  # Retry with another domain

    avg_time = statistics.mean(times) if times else 0
    print(f"{nameserver[1]} ({nameserver[0]}) | Average: {avg_time:.2f} ms\n",
          end="")


def main():
    """
    Fetches a list of domains from a whitelist, tests multiple DNS servers by
    querying random domains from the list, and prints the average query time
    for each server. The DNS queries are performed in parallel for efficiency.
    """
    domains_list = fetch_whitelist(WHITELIST_URL)

    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = {executor.submit(test_nameserver, ns, domains_list): ns
                   for ns in NAMESERVERS}

        for future in concurrent.futures.as_completed(futures):
            ns = futures[future]
            try:
                future.result()
            except Exception as e:
                print(f"Exception occurred during testing of {ns[1]}: {e}")


if __name__ == "__main__":
    main()