path: root/tests/integration/deckard/pydnstest/testserver.py

import argparse
import itertools
import logging
import os
import random
import signal
import selectors
import socket
import sys
import threading
import time

import dns.message
import dns.rdatatype

from pydnstest import scenario, mock_client
from networking import InterfaceManager


class TestServer:
    """ This simulates UDP DNS server returning scripted or mirror DNS responses. """

    RETRIES_ON_BIND = 3

    def __init__(self, test_scenario, addr_family,
                 deckard_address=None, if_manager=None):
        """ Initialize server instance. """
        self.thread = None
        self.srv_socks = []
        self.client_socks = []
        self.connections = []
        self.active = False
        self.active_lock = threading.Lock()
        self.condition = threading.Condition()
        self.scenario = test_scenario
        self.scenario.deckard_address = deckard_address
        self.addr_map = []
        self.start_iface = 2
        self.cur_iface = self.start_iface
        self.addr_family = addr_family
        self.undefined_answers = 0
        self.if_manager = if_manager

    def __del__(self):
        """ Cleanup after deletion. """
        with self.active_lock:
            active = self.active
        if active:
            self.stop()

    def start(self):
        """ Synchronous start """
        with self.active_lock:
            if self.active:
                raise Exception('TestServer already started')
        with self.active_lock:
            self.active = True

        self._bind_sockets()

    def stop(self):
        """ Stop socket server operation. """
        with self.active_lock:
            self.active = False
        if self.thread:
            self.thread.join()
        for conn in self.connections:
            conn.close()
        for srv_sock in self.srv_socks:
            srv_sock.close()
        for client_sock in self.client_socks:
            client_sock.close()
        self.client_socks = []
        self.srv_socks = []
        self.connections = []
        self.scenario = None

    def address(self):
        """ Returns opened sockets list """
        addrlist = []
        for s in self.srv_socks:
            addrlist.append(s.getsockname())
        return addrlist

    def handle_query(self, client):
        """
        Receive query from client socket and send an answer.

        Returns:
            True if client socket should be closed by caller
            False if client socket should be kept open
        """
        log = logging.getLogger('pydnstest.testserver.handle_query')
        server_addr = client.getsockname()[0]
        query, client_addr = mock_client.recvfrom_msg(client)
        if query is None:
            return False
        log.debug('server %s received query from %s: %s', server_addr, client_addr, query)

        message = self.scenario.reply(query, server_addr)
        if not message:
            log.debug('ignoring')
            return True
        elif isinstance(message, scenario.DNSReplyServfail):
            self.undefined_answers += 1
            self.scenario.current_step.log.error(
                'server %s has no response for question %s, answering with SERVFAIL',
                server_addr,
                '; '.join([str(rr) for rr in query.question]))
        else:
            log.debug('response: %s', message)

        wire = message.to_wire()
        wire_len = len(wire)
        if query.payload > 0 and wire_len > query.payload:
            log.warning('Reply length (%d B) is greater than the client\'s '
                        'reported EDNS payload (%d B). The test may fail '
                        'due to insufficient buffer size.',
                        wire_len, query.payload)

        mock_client.sendto_msg(client, wire, client_addr)
        return True

    def query_io(self):
        """ Main server process """
        self.undefined_answers = 0
        with self.active_lock:
            if not self.active:
                raise Exception("[query_io] Test server not active")
        while True:
            with self.condition:
                self.condition.notify()
            with self.active_lock:
                if not self.active:
                    break
            objects = self.srv_socks + self.connections
            sel = selectors.DefaultSelector()
            for obj in objects:
                sel.register(obj, selectors.EVENT_READ)
            items = sel.select(0.1)
            for key, event in items:
                sock = key.fileobj
                if event & selectors.EVENT_READ:
                    if sock in self.srv_socks:
                        if sock.proto == socket.IPPROTO_TCP:
                            conn, _ = sock.accept()
                            self.connections.append(conn)
                        else:
                            self.handle_query(sock)
                    elif sock in self.connections:
                        if not self.handle_query(sock):
                            sock.close()
                            self.connections.remove(sock)
                    else:
                        raise Exception(
                            "[query_io] Socket IO internal error {}, exit"
                            .format(sock.getsockname()))
                else:
                    raise Exception("[query_io] Socket IO error {}, exit"
                                    .format(sock.getsockname()))

    def start_srv(self, address, family, proto=socket.IPPROTO_UDP):
        """ Starts listening thread if necessary """
        assert address
        assert address[0]  # host
        assert address[1]  # port
        assert family
        assert proto
        if family == socket.AF_INET6:
            if not socket.has_ipv6:
                raise NotImplementedError("[start_srv] IPv6 is not supported by socket {0}"
                                          .format(socket))
        elif family != socket.AF_INET:
            raise NotImplementedError("[start_srv] unsupported protocol family {0}".format(family))

        if proto == socket.IPPROTO_TCP:
            socktype = socket.SOCK_STREAM
        elif proto == socket.IPPROTO_UDP:
            socktype = socket.SOCK_DGRAM
        else:
            raise NotImplementedError("[start_srv] unsupported protocol {0}".format(proto))

        if self.thread is None:
            self.thread = threading.Thread(target=self.query_io)
            self.thread.start()
            with self.condition:
                self.condition.wait()

        for srv_sock in self.srv_socks:
            if (srv_sock.family == family
                    and srv_sock.getsockname()[:2] == address
                    and srv_sock.proto == proto):
                return

        sock = socket.socket(family, socktype, proto)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        # Add address to interface when running from Deckard
        if self.if_manager is not None:
            if address[0] not in self.if_manager.added_addresses:
                self.if_manager.add_address(address[0])

        # A lot of addresses are added to the interface while runnning from Deckard in
        # the small amount of time which caused ocassional hiccups while binding to them
        # right afterwards in testing. Therefore, we retry a few times.
        final_ex = None
        for i in range(self.RETRIES_ON_BIND):
            try:
                sock.bind(address)
                break
            except OSError as ex:
                # Exponential backoff
                time.sleep((2 ** i) + random.random())
                final_ex = ex
                continue
        else:
            print(final_ex, address)
            raise final_ex

        if proto == socket.IPPROTO_TCP:
            sock.listen(5)
        self.srv_socks.append(sock)

    def _bind_sockets(self):
        """
        Bind test server to port 53 on all addresses referenced by test scenario.
        """
        # Bind to test servers
        for r in self.scenario.ranges:
            for addr in r.addresses:
                family = socket.AF_INET6 if ':' in addr else socket.AF_INET
                self.start_srv((addr, 53), family)
                self.start_srv((addr, 53), family, proto=socket.IPPROTO_TCP)

        # Bind addresses in ad-hoc REPLYs
        for s in self.scenario.steps:
            if s.type == 'REPLY':
                reply = s.data[0].message
                for rr in itertools.chain(reply.answer,
                                          reply.additional,
                                          reply.question,
                                          reply.authority):
                    for rd in rr:
                        if rd.rdtype == dns.rdatatype.A:
                            self.start_srv((rd.address, 53), socket.AF_INET)
                            self.start_srv((rd.address, 53), socket.AF_INET,
                                           proto=socket.IPPROTO_TCP)
                        elif rd.rdtype == dns.rdatatype.AAAA:
                            self.start_srv((rd.address, 53), socket.AF_INET6)
                            self.start_srv((rd.address, 53), socket.AF_INET6,
                                           proto=socket.IPPROTO_TCP)

    def play(self, subject_addr):
        self.scenario.play({'': (subject_addr, 53)})


def empty_test_case():
    """
    Return (scenario, config) pair which answers to any query on 127.0.0.10.
    """
    # Mirror server
    empty_test_path = os.path.dirname(os.path.realpath(__file__)) + "/empty.rpl"
    test_config = {'ROOT_ADDR': '127.0.0.10',
                   '_SOCKET_FAMILY': socket.AF_INET}
    return scenario.parse_file(empty_test_path)[0], test_config


def standalone_self_test():
    """
    Self-test code

    Usage:
    unshare -rn $PYTHON -m pydnstest.testserver --help
    """
    logging.basicConfig(level=logging.DEBUG)
    argparser = argparse.ArgumentParser()
    argparser.add_argument('--scenario', help='absolute path to test scenario',
                           required=False)
    argparser.add_argument('--step', help='step # in the scenario (default: first)',
                           required=False, type=int)
    args = argparser.parse_args()
    if args.scenario:
        test_scenario, test_config_text = scenario.parse_file(args.scenario)
        test_config = scenario.parse_config(test_config_text, True, os.getcwd())
    else:
        test_scenario, test_config = empty_test_case()

    if args.step:
        for step in test_scenario.steps:
            if step.id == args.step:
                test_scenario.current_step = step
        if not test_scenario.current_step:
            raise ValueError('step ID %s not found in scenario' % args.step)
    else:
        test_scenario.current_step = test_scenario.steps[0]

    if_manager = InterfaceManager(interface="testserver")
    server = TestServer(test_scenario, test_config['_SOCKET_FAMILY'], if_manager=if_manager)
    server.start()

    logging.info("[==========] Mirror server running at %s", server.address())

    def kill(signum, frame):  # pylint: disable=unused-argument
        logging.info("[==========] Shutdown.")
        server.stop()
        sys.exit(128 + signum)

    signal.signal(signal.SIGINT, kill)
    signal.signal(signal.SIGTERM, kill)

    while True:
        time.sleep(0.5)


if __name__ == '__main__':
    # this is done to avoid creating global variables
    standalone_self_test()