[thirdparty/pdns.git] / regression-tests.dnsdist / test_Basics.py

#!/usr/bin/env python
import dns
import clientsubnetoption
from dnsdisttests import DNSDistTest

class TestBasics(DNSDistTest):

    _config_template = """
    newServer{address="127.0.0.1:%s"}
    truncateTC(true)
    addAction(QTypeRule(DNSQType.ANY), TCAction())
    addAction(RegexRule("evil[0-9]{4,}\\\\.regex\\\\.tests\\\\.powerdns\\\\.com$"), RCodeAction(DNSRCode.REFUSED))
    mySMN = newSuffixMatchNode()
    mySMN:add(newDNSName("nameAndQtype.tests.powerdns.com."))
    addAction(AndRule{SuffixMatchNodeRule(mySMN), QTypeRule("TXT")}, RCodeAction(DNSRCode.NOTIMP))
    addAction(QNameSuffixRule({"drop.test.powerdns.com.", "drop2.test.powerdns.com."}), DropAction())
    addAction(AndRule({QTypeRule(DNSQType.A),QNameRule("ds9a.nl")}), SpoofAction("1.2.3.4"))
    addAction(newDNSName("dnsname.addaction.powerdns.com."), RCodeAction(DNSRCode.REFUSED))
    addAction({newDNSName("dnsname-table1.addaction.powerdns.com."), newDNSName("dnsname-table2.addaction.powerdns.com.")}, RCodeAction(DNSRCode.REFUSED))
    """

    def testDropped(self):
        """
        Basics: Dropped query

        Send an A query for drop.test.powerdns.com. domain,
        which is dropped by configuration. We expect
        no response.
        """
        for name in ['drop.test.powerdns.com.', 'drop2.test.powerdns.com.']:
            query = dns.message.make_query(name, 'A', 'IN')
            for method in ("sendUDPQuery", "sendTCPQuery"):
                sender = getattr(self, method)
                (_, receivedResponse) = sender(query, response=None, useQueue=False)
                self.assertEqual(receivedResponse, None)

    def testAWithECS(self):
        """
        Basics: A query with an ECS value
        """
        name = 'awithecs.tests.powerdns.com.'
        ecso = clientsubnetoption.ClientSubnetOption('1.2.3.4')
        query = dns.message.make_query(name, 'A', 'IN', use_edns=True, options=[ecso])
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '127.0.0.1')

        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)

    def testSimpleA(self):
        """
        Basics: A query without EDNS
        """
        name = 'simplea.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '127.0.0.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            self.assertTrue(receivedQuery)
            self.assertTrue(receivedResponse)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)

    def testAnyIsTruncated(self):
        """
        Basics: Truncate ANY query

        dnsdist is configured to reply with TC to ANY queries,
        send an ANY query and check the result.
        It should be truncated over UDP, not over TCP.
        """
        name = 'any.tests.powerdns.com.'
        query = dns.message.make_query(name, 'ANY', 'IN')
        # dnsdist sets RA = RD for TC responses
        query.flags &= ~dns.flags.RD
        expectedResponse = dns.message.make_response(query)
        expectedResponse.flags |= dns.flags.TC

        (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
        self.assertEqual(receivedResponse, expectedResponse)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '127.0.0.1')

        response.answer.append(rrset)

        (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
        self.assertTrue(receivedQuery)
        self.assertTrue(receivedResponse)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(receivedResponse, response)

    def testTruncateTC(self):
        """
        Basics: Truncate TC

        dnsdist is configured to truncate TC (default),
        we make the backend send responses
        with TC set and additional content,
        and check that the received response has been fixed.
        """
        name = 'atruncatetc.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '127.0.0.1')

        response.answer.append(rrset)
        response.flags |= dns.flags.TC
        expectedResponse = dns.message.make_response(query)
        expectedResponse.flags |= dns.flags.TC

        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(expectedResponse.flags, receivedResponse.flags)
        self.assertEqual(expectedResponse.question, receivedResponse.question)
        self.assertFalse(response.answer == receivedResponse.answer)
        self.assertEqual(len(receivedResponse.answer), 0)
        self.assertEqual(len(receivedResponse.authority), 0)
        self.assertEqual(len(receivedResponse.additional), 0)
        self.checkMessageNoEDNS(expectedResponse, receivedResponse)

    def testTruncateTCEDNS(self):
        """
        Basics: Truncate TC with EDNS

        dnsdist is configured to truncate TC (default),
        we make the backend send responses
        with TC set and additional content,
        and check that the received response has been fixed.
        Note that the query and initial response had EDNS,
        so the final response should have it too.
        """
        name = 'atruncatetc.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=True)
        # force a different responder payload than the one in the query,
        # so we check that we don't just mirror it
        response = dns.message.make_response(query, our_payload=4242)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '127.0.0.1')

        response.answer.append(rrset)
        response.flags |= dns.flags.TC
        expectedResponse = dns.message.make_response(query, our_payload=4242)
        expectedResponse.flags |= dns.flags.TC

        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response.flags, receivedResponse.flags)
        self.assertEqual(response.question, receivedResponse.question)
        self.assertFalse(response.answer == receivedResponse.answer)
        self.assertEqual(len(receivedResponse.answer), 0)
        self.assertEqual(len(receivedResponse.authority), 0)
        self.assertEqual(len(receivedResponse.additional), 0)
        print(expectedResponse)
        print(receivedResponse)
        self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
        self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
        self.assertEqual(receivedResponse.payload, 4242)

    def testRegexReturnsRefused(self):
        """
        Basics: Refuse query matching regex

        dnsdist is configured to reply 'refused' for query
        matching "evil[0-9]{4,}\\.regex\\.tests\\.powerdns\\.com$".
        We send a query for evil4242.powerdns.com
        and check that the response is "refused".
        """
        name = 'evil4242.regex.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        query.flags &= ~dns.flags.RD
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.REFUSED)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response=None, useQueue=False)
            self.assertEqual(receivedResponse, expectedResponse)

    def testQNameReturnsSpoofed(self):
        """
        Basics: test QNameRule and Spoof

        dnsdist is configured to reply 1.2.3.4 for A query for exactly ds9a.nl
        """
        name = 'ds9a.nl.'
        query = dns.message.make_query(name, 'A', 'IN')
        query.flags &= ~dns.flags.RD
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.NOERROR)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '1.2.3.4')
        expectedResponse.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response=None, useQueue=False)
            self.assertEqual(receivedResponse, expectedResponse)

    def testDomainAndQTypeReturnsNotImplemented(self):
        """
        Basics: NOTIMPL for specific name and qtype

        dnsdist is configured to reply 'not implemented' for query
        matching "nameAndQtype.tests.powerdns.com." AND qtype TXT.
        We send a TXT query for "nameAndQtype.powerdns.com."
        and check that the response is 'not implemented'.
        """
        name = 'nameAndQtype.tests.powerdns.com.'
        query = dns.message.make_query(name, 'TXT', 'IN')
        query.flags &= ~dns.flags.RD
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.NOTIMP)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response=None, useQueue=False)
            self.assertEqual(receivedResponse, expectedResponse)

    def testDomainWithoutQTypeIsNotAffected(self):
        """
        Basics: NOTIMPL qtype canary

        dnsdist is configured to reply 'not implemented' for query
        matching "nameAndQtype.tests.powerdns.com." AND qtype TXT.
        We send a A query for "nameAndQtype.tests.powerdns.com."
        and check that the response is OK.
        """
        name = 'nameAndQtype.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '127.0.0.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            self.assertTrue(receivedQuery)
            self.assertTrue(receivedResponse)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)

    def testOtherDomainANDQTypeIsNotAffected(self):
        """
        Basics: NOTIMPL qname canary

        dnsdist is configured to reply 'not implemented' for query
        matching "nameAndQtype.tests.powerdns.com." AND qtype TXT.
        We send a TXT query for "OtherNameAndQtype.tests.powerdns.com."
        and check that the response is OK.
        """
        name = 'OtherNameAndQtype.tests.powerdns.com.'
        query = dns.message.make_query(name, 'TXT', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'nothing to see here')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            self.assertTrue(receivedQuery)
            self.assertTrue(receivedResponse)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)

    def testWrongResponse(self):
        """
        Basics: Unrelated response from the backend

        The backend send an unrelated answer over UDP, it should
        be discarded by dnsdist. It could happen if we wrap around
        maxOutstanding queries too quickly or have more than maxOutstanding
        queries to a specific backend in the air over UDP,
        but does not really make sense over TCP.
        """
        name = 'query.unrelated.tests.powerdns.com.'
        unrelatedName = 'answer.unrelated.tests.powerdns.com.'
        query = dns.message.make_query(name, 'TXT', 'IN')
        unrelatedQuery = dns.message.make_query(unrelatedName, 'TXT', 'IN')
        unrelatedResponse = dns.message.make_response(unrelatedQuery)
        rrset = dns.rrset.from_text(unrelatedName,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'nothing to see here')
        unrelatedResponse.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, unrelatedResponse)
            self.assertTrue(receivedQuery)
            self.assertEqual(receivedResponse, None)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)

    def testHeaderOnlyRefused(self):
        """
        Basics: Header-only refused response
        """
        name = 'header-only-refused-response.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.set_rcode(dns.rcode.REFUSED)
        response.question = []

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            self.assertTrue(receivedQuery)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(receivedResponse, response)

    def testHeaderOnlyNoErrorResponse(self):
        """
        Basics: Header-only NoError response should be dropped
        """
        name = 'header-only-noerror-response.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.question = []

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            self.assertTrue(receivedQuery)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(receivedResponse, None)

    def testHeaderOnlyNXDResponse(self):
        """
        Basics: Header-only NXD response should be dropped
        """
        name = 'header-only-nxd-response.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.set_rcode(dns.rcode.NXDOMAIN)
        response.question = []

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            self.assertTrue(receivedQuery)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(receivedResponse, None)

    def testAddActionDNSName(self):
        """
        Basics: test if addAction accepts a DNSName
        """
        name = 'dnsname.addaction.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        query.flags &= ~dns.flags.RD
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.REFUSED)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response=None, useQueue=False)
            self.assertEqual(receivedResponse, expectedResponse)

    def testAddActionDNSNames(self):
        """
        Basics: test if addAction accepts a table of DNSNames
        """
        for name in ['dnsname-table{}.addaction.powerdns.com.'.format(i) for i in range(1,2)]:
            query = dns.message.make_query(name, 'A', 'IN')
            query.flags &= ~dns.flags.RD
            expectedResponse = dns.message.make_response(query)
            expectedResponse.set_rcode(dns.rcode.REFUSED)

            for method in ("sendUDPQuery", "sendTCPQuery"):
                sender = getattr(self, method)
                (_, receivedResponse) = sender(query, response=None, useQueue=False)
                self.assertEqual(receivedResponse, expectedResponse)

    def testEmptyQueries(self):
        """
        Basic: NotImp on empty queries
        """
        name = 'notimp-empty-queries.basic.tests.powerdns.com.'
        query = dns.message.Message()

        response = dns.message.make_response(query)
        response.set_rcode(dns.rcode.NOTIMP)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response=None, useQueue=False)
            self.assertEqual(receivedResponse, response)
Commit	Line	Data
ca404e94	1	#!/usr/bin/env python
b1bec9f0 RG	2	import dns
b1bec9f0 RG	3	import clientsubnetoption
ca404e94 RG	4	from dnsdisttests import DNSDistTest
	5
	6	class TestBasics(DNSDistTest):
	7
903853f4 RG	8	_config_template = """
	9	newServer{address="127.0.0.1:%s"}
	10	truncateTC(true)
4a7a0ff8	11	addAction(QTypeRule(DNSQType.ANY), TCAction())
d3ec24f9	12	addAction(RegexRule("evil[0-9]{4,}\\\\.regex\\\\.tests\\\\.powerdns\\\\.com$"), RCodeAction(DNSRCode.REFUSED))
903853f4 RG	13	mySMN = newSuffixMatchNode()
903853f4 RG	14	mySMN:add(newDNSName("nameAndQtype.tests.powerdns.com."))
d3ec24f9	15	addAction(AndRule{SuffixMatchNodeRule(mySMN), QTypeRule("TXT")}, RCodeAction(DNSRCode.NOTIMP))
34d16877	16	addAction(QNameSuffixRule({"drop.test.powerdns.com.", "drop2.test.powerdns.com."}), DropAction())
d3ec24f9 PD	17	addAction(AndRule({QTypeRule(DNSQType.A),QNameRule("ds9a.nl")}), SpoofAction("1.2.3.4"))
	18	addAction(newDNSName("dnsname.addaction.powerdns.com."), RCodeAction(DNSRCode.REFUSED))
	19	addAction({newDNSName("dnsname-table1.addaction.powerdns.com."), newDNSName("dnsname-table2.addaction.powerdns.com.")}, RCodeAction(DNSRCode.REFUSED))
903853f4 RG	20	"""
903853f4 RG	21
b63add03 RG	22	def testDropped(self):
	23	"""
	24	Basics: Dropped query
	25
	26	Send an A query for drop.test.powerdns.com. domain,
	27	which is dropped by configuration. We expect
	28	no response.
	29	"""
1bebf05f RG	30	for name in ['drop.test.powerdns.com.', 'drop2.test.powerdns.com.']:
	31	query = dns.message.make_query(name, 'A', 'IN')
	32	for method in ("sendUDPQuery", "sendTCPQuery"):
	33	sender = getattr(self, method)
	34	(_, receivedResponse) = sender(query, response=None, useQueue=False)
	35	self.assertEqual(receivedResponse, None)
b63add03	36
ca404e94 RG	37	def testAWithECS(self):
ca404e94 RG	38	"""
617dfe22	39	Basics: A query with an ECS value
ca404e94 RG	40	"""
	41	name = 'awithecs.tests.powerdns.com.'
	42	ecso = clientsubnetoption.ClientSubnetOption('1.2.3.4')
	43	query = dns.message.make_query(name, 'A', 'IN', use_edns=True, options=[ecso])
	44	response = dns.message.make_response(query)
	45	rrset = dns.rrset.from_text(name,
feb22f99	46	60,
ca404e94 RG	47	dns.rdataclass.IN,
	48	dns.rdatatype.A,
	49	'127.0.0.1')
	50
	51	response.answer.append(rrset)
	52
6ca2e796 RG	53	for method in ("sendUDPQuery", "sendTCPQuery"):
	54	sender = getattr(self, method)
	55	(receivedQuery, receivedResponse) = sender(query, response)
	56	receivedQuery.id = query.id
4bfebc93 CH	57	self.assertEqual(query, receivedQuery)
4bfebc93 CH	58	self.assertEqual(response, receivedResponse)
ca404e94 RG	59
	60	def testSimpleA(self):
	61	"""
617dfe22	62	Basics: A query without EDNS
ca404e94 RG	63	"""
	64	name = 'simplea.tests.powerdns.com.'
	65	query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
	66	response = dns.message.make_response(query)
e7a1029c	67	rrset = dns.rrset.from_text(name,
ca404e94 RG	68	3600,
	69	dns.rdataclass.IN,
	70	dns.rdatatype.A,
	71	'127.0.0.1')
	72	response.answer.append(rrset)
	73
6ca2e796 RG	74	for method in ("sendUDPQuery", "sendTCPQuery"):
	75	sender = getattr(self, method)
	76	(receivedQuery, receivedResponse) = sender(query, response)
	77	self.assertTrue(receivedQuery)
	78	self.assertTrue(receivedResponse)
	79	receivedQuery.id = query.id
4bfebc93 CH	80	self.assertEqual(query, receivedQuery)
4bfebc93 CH	81	self.assertEqual(response, receivedResponse)
ca404e94	82
ec5f5c6b RG	83	def testAnyIsTruncated(self):
ec5f5c6b RG	84	"""
617dfe22 RG	85	Basics: Truncate ANY query
617dfe22 RG	86
ec5f5c6b RG	87	dnsdist is configured to reply with TC to ANY queries,
ec5f5c6b RG	88	send an ANY query and check the result.
490a29bb	89	It should be truncated over UDP, not over TCP.
ec5f5c6b RG	90	"""
	91	name = 'any.tests.powerdns.com.'
	92	query = dns.message.make_query(name, 'ANY', 'IN')
955b9377 RG	93	# dnsdist sets RA = RD for TC responses
955b9377 RG	94	query.flags &= ~dns.flags.RD
ec5f5c6b RG	95	expectedResponse = dns.message.make_response(query)
	96	expectedResponse.flags \|= dns.flags.TC
	97
0a2087eb	98	(_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False)
4bfebc93	99	self.assertEqual(receivedResponse, expectedResponse)
ec5f5c6b	100
490a29bb	101	response = dns.message.make_response(query)
e7a1029c	102	rrset = dns.rrset.from_text(name,
490a29bb RG	103	3600,
	104	dns.rdataclass.IN,
	105	dns.rdatatype.A,
	106	'127.0.0.1')
	107
	108	response.answer.append(rrset)
	109
	110	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	111	self.assertTrue(receivedQuery)
	112	self.assertTrue(receivedResponse)
	113	receivedQuery.id = query.id
4bfebc93 CH	114	self.assertEqual(query, receivedQuery)
4bfebc93 CH	115	self.assertEqual(receivedResponse, response)
ec5f5c6b RG	116
	117	def testTruncateTC(self):
	118	"""
617dfe22 RG	119	Basics: Truncate TC
617dfe22 RG	120
ec5f5c6b RG	121	dnsdist is configured to truncate TC (default),
	122	we make the backend send responses
	123	with TC set and additional content,
	124	and check that the received response has been fixed.
	125	"""
	126	name = 'atruncatetc.tests.powerdns.com.'
	127	query = dns.message.make_query(name, 'A', 'IN')
	128	response = dns.message.make_response(query)
	129	rrset = dns.rrset.from_text(name,
	130	3600,
	131	dns.rdataclass.IN,
	132	dns.rdatatype.A,
	133	'127.0.0.1')
	134
	135	response.answer.append(rrset)
	136	response.flags \|= dns.flags.TC
e0fd37ec RG	137	expectedResponse = dns.message.make_response(query)
	138	expectedResponse.flags \|= dns.flags.TC
	139
	140	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	141	receivedQuery.id = query.id
4bfebc93 CH	142	self.assertEqual(query, receivedQuery)
	143	self.assertEqual(expectedResponse.flags, receivedResponse.flags)
	144	self.assertEqual(expectedResponse.question, receivedResponse.question)
e0fd37ec	145	self.assertFalse(response.answer == receivedResponse.answer)
4bfebc93 CH	146	self.assertEqual(len(receivedResponse.answer), 0)
	147	self.assertEqual(len(receivedResponse.authority), 0)
	148	self.assertEqual(len(receivedResponse.additional), 0)
e0fd37ec RG	149	self.checkMessageNoEDNS(expectedResponse, receivedResponse)
	150
	151	def testTruncateTCEDNS(self):
	152	"""
	153	Basics: Truncate TC with EDNS
	154
	155	dnsdist is configured to truncate TC (default),
	156	we make the backend send responses
	157	with TC set and additional content,
	158	and check that the received response has been fixed.
	159	Note that the query and initial response had EDNS,
	160	so the final response should have it too.
	161	"""
	162	name = 'atruncatetc.tests.powerdns.com.'
	163	query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, want_dnssec=True)
e0fd37ec RG	164	# force a different responder payload than the one in the query,
e0fd37ec RG	165	# so we check that we don't just mirror it
6e1f856f	166	response = dns.message.make_response(query, our_payload=4242)
e0fd37ec RG	167	rrset = dns.rrset.from_text(name,
	168	3600,
	169	dns.rdataclass.IN,
	170	dns.rdatatype.A,
	171	'127.0.0.1')
	172
	173	response.answer.append(rrset)
	174	response.flags \|= dns.flags.TC
6e1f856f	175	expectedResponse = dns.message.make_response(query, our_payload=4242)
e0fd37ec	176	expectedResponse.flags \|= dns.flags.TC
ec5f5c6b RG	177
	178	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	179	receivedQuery.id = query.id
4bfebc93 CH	180	self.assertEqual(query, receivedQuery)
	181	self.assertEqual(response.flags, receivedResponse.flags)
	182	self.assertEqual(response.question, receivedResponse.question)
ec5f5c6b	183	self.assertFalse(response.answer == receivedResponse.answer)
4bfebc93 CH	184	self.assertEqual(len(receivedResponse.answer), 0)
	185	self.assertEqual(len(receivedResponse.authority), 0)
	186	self.assertEqual(len(receivedResponse.additional), 0)
e0fd37ec RG	187	print(expectedResponse)
	188	print(receivedResponse)
	189	self.checkMessageEDNSWithoutOptions(expectedResponse, receivedResponse)
	190	self.assertFalse(receivedResponse.ednsflags & dns.flags.DO)
4bfebc93	191	self.assertEqual(receivedResponse.payload, 4242)
ec5f5c6b RG	192
	193	def testRegexReturnsRefused(self):
	194	"""
617dfe22 RG	195	Basics: Refuse query matching regex
617dfe22 RG	196
ec5f5c6b RG	197	dnsdist is configured to reply 'refused' for query
	198	matching "evil[0-9]{4,}\\.regex\\.tests\\.powerdns\\.com$".
	199	We send a query for evil4242.powerdns.com
	200	and check that the response is "refused".
	201	"""
	202	name = 'evil4242.regex.tests.powerdns.com.'
	203	query = dns.message.make_query(name, 'A', 'IN')
7af22479	204	query.flags &= ~dns.flags.RD
ec5f5c6b RG	205	expectedResponse = dns.message.make_response(query)
	206	expectedResponse.set_rcode(dns.rcode.REFUSED)
	207
6ca2e796 RG	208	for method in ("sendUDPQuery", "sendTCPQuery"):
	209	sender = getattr(self, method)
	210	(_, receivedResponse) = sender(query, response=None, useQueue=False)
4bfebc93	211	self.assertEqual(receivedResponse, expectedResponse)
ec5f5c6b	212
feb22f99	213	def testQNameReturnsSpoofed(self):
	214	"""
	215	Basics: test QNameRule and Spoof
	216
	217	dnsdist is configured to reply 1.2.3.4 for A query for exactly ds9a.nl
	218	"""
	219	name = 'ds9a.nl.'
	220	query = dns.message.make_query(name, 'A', 'IN')
	221	query.flags &= ~dns.flags.RD
	222	expectedResponse = dns.message.make_response(query)
	223	expectedResponse.set_rcode(dns.rcode.NOERROR)
	224	rrset = dns.rrset.from_text(name,
	225	3600,
	226	dns.rdataclass.IN,
	227	dns.rdatatype.A,
	228	'1.2.3.4')
	229	expectedResponse.answer.append(rrset)
	230
6ca2e796 RG	231	for method in ("sendUDPQuery", "sendTCPQuery"):
	232	sender = getattr(self, method)
	233	(_, receivedResponse) = sender(query, response=None, useQueue=False)
4bfebc93	234	self.assertEqual(receivedResponse, expectedResponse)
feb22f99	235
ec5f5c6b RG	236	def testDomainAndQTypeReturnsNotImplemented(self):
ec5f5c6b RG	237	"""
617dfe22 RG	238	Basics: NOTIMPL for specific name and qtype
617dfe22 RG	239
ec5f5c6b	240	dnsdist is configured to reply 'not implemented' for query
e7a1029c	241	matching "nameAndQtype.tests.powerdns.com." AND qtype TXT.
ec5f5c6b RG	242	We send a TXT query for "nameAndQtype.powerdns.com."
	243	and check that the response is 'not implemented'.
	244	"""
	245	name = 'nameAndQtype.tests.powerdns.com.'
	246	query = dns.message.make_query(name, 'TXT', 'IN')
7af22479	247	query.flags &= ~dns.flags.RD
ec5f5c6b RG	248	expectedResponse = dns.message.make_response(query)
	249	expectedResponse.set_rcode(dns.rcode.NOTIMP)
	250
6ca2e796 RG	251	for method in ("sendUDPQuery", "sendTCPQuery"):
	252	sender = getattr(self, method)
	253	(_, receivedResponse) = sender(query, response=None, useQueue=False)
4bfebc93	254	self.assertEqual(receivedResponse, expectedResponse)
ec5f5c6b RG	255
	256	def testDomainWithoutQTypeIsNotAffected(self):
	257	"""
617dfe22 RG	258	Basics: NOTIMPL qtype canary
617dfe22 RG	259
ec5f5c6b	260	dnsdist is configured to reply 'not implemented' for query
e7a1029c	261	matching "nameAndQtype.tests.powerdns.com." AND qtype TXT.
ec5f5c6b RG	262	We send a A query for "nameAndQtype.tests.powerdns.com."
	263	and check that the response is OK.
	264	"""
	265	name = 'nameAndQtype.tests.powerdns.com.'
	266	query = dns.message.make_query(name, 'A', 'IN')
	267	response = dns.message.make_response(query)
	268	rrset = dns.rrset.from_text(name,
	269	3600,
	270	dns.rdataclass.IN,
	271	dns.rdatatype.A,
	272	'127.0.0.1')
	273	response.answer.append(rrset)
	274
6ca2e796 RG	275	for method in ("sendUDPQuery", "sendTCPQuery"):
	276	sender = getattr(self, method)
	277	(receivedQuery, receivedResponse) = sender(query, response)
	278	self.assertTrue(receivedQuery)
	279	self.assertTrue(receivedResponse)
	280	receivedQuery.id = query.id
4bfebc93 CH	281	self.assertEqual(query, receivedQuery)
4bfebc93 CH	282	self.assertEqual(response, receivedResponse)
ec5f5c6b RG	283
	284	def testOtherDomainANDQTypeIsNotAffected(self):
	285	"""
617dfe22 RG	286	Basics: NOTIMPL qname canary
617dfe22 RG	287
ec5f5c6b	288	dnsdist is configured to reply 'not implemented' for query
e7a1029c	289	matching "nameAndQtype.tests.powerdns.com." AND qtype TXT.
ec5f5c6b RG	290	We send a TXT query for "OtherNameAndQtype.tests.powerdns.com."
	291	and check that the response is OK.
	292	"""
	293	name = 'OtherNameAndQtype.tests.powerdns.com.'
	294	query = dns.message.make_query(name, 'TXT', 'IN')
	295	response = dns.message.make_response(query)
	296	rrset = dns.rrset.from_text(name,
	297	3600,
	298	dns.rdataclass.IN,
	299	dns.rdatatype.TXT,
	300	'nothing to see here')
	301	response.answer.append(rrset)
	302
6ca2e796 RG	303	for method in ("sendUDPQuery", "sendTCPQuery"):
	304	sender = getattr(self, method)
	305	(receivedQuery, receivedResponse) = sender(query, response)
	306	self.assertTrue(receivedQuery)
	307	self.assertTrue(receivedResponse)
	308	receivedQuery.id = query.id
4bfebc93 CH	309	self.assertEqual(query, receivedQuery)
4bfebc93 CH	310	self.assertEqual(response, receivedResponse)
ec5f5c6b	311
7267213a RG	312	def testWrongResponse(self):
	313	"""
	314	Basics: Unrelated response from the backend
	315
	316	The backend send an unrelated answer over UDP, it should
	317	be discarded by dnsdist. It could happen if we wrap around
de4896ba	318	maxOutstanding queries too quickly or have more than maxOutstanding
7267213a RG	319	queries to a specific backend in the air over UDP,
	320	but does not really make sense over TCP.
	321	"""
	322	name = 'query.unrelated.tests.powerdns.com.'
	323	unrelatedName = 'answer.unrelated.tests.powerdns.com.'
	324	query = dns.message.make_query(name, 'TXT', 'IN')
	325	unrelatedQuery = dns.message.make_query(unrelatedName, 'TXT', 'IN')
	326	unrelatedResponse = dns.message.make_response(unrelatedQuery)
	327	rrset = dns.rrset.from_text(unrelatedName,
	328	3600,
	329	dns.rdataclass.IN,
	330	dns.rdatatype.TXT,
	331	'nothing to see here')
	332	unrelatedResponse.answer.append(rrset)
	333
6ca2e796 RG	334	for method in ("sendUDPQuery", "sendTCPQuery"):
	335	sender = getattr(self, method)
	336	(receivedQuery, receivedResponse) = sender(query, unrelatedResponse)
	337	self.assertTrue(receivedQuery)
4bfebc93	338	self.assertEqual(receivedResponse, None)
6ca2e796	339	receivedQuery.id = query.id
4bfebc93	340	self.assertEqual(query, receivedQuery)
f87c4aff	341
c8c3d4e4 RG	342	def testHeaderOnlyRefused(self):
	343	"""
	344	Basics: Header-only refused response
	345	"""
	346	name = 'header-only-refused-response.tests.powerdns.com.'
	347	query = dns.message.make_query(name, 'A', 'IN')
	348	response = dns.message.make_response(query)
	349	response.set_rcode(dns.rcode.REFUSED)
	350	response.question = []
	351
6ca2e796 RG	352	for method in ("sendUDPQuery", "sendTCPQuery"):
	353	sender = getattr(self, method)
	354	(receivedQuery, receivedResponse) = sender(query, response)
	355	self.assertTrue(receivedQuery)
	356	receivedQuery.id = query.id
4bfebc93 CH	357	self.assertEqual(query, receivedQuery)
4bfebc93 CH	358	self.assertEqual(receivedResponse, response)
c8c3d4e4 RG	359
	360	def testHeaderOnlyNoErrorResponse(self):
	361	"""
	362	Basics: Header-only NoError response should be dropped
	363	"""
	364	name = 'header-only-noerror-response.tests.powerdns.com.'
	365	query = dns.message.make_query(name, 'A', 'IN')
	366	response = dns.message.make_response(query)
	367	response.question = []
	368
6ca2e796 RG	369	for method in ("sendUDPQuery", "sendTCPQuery"):
	370	sender = getattr(self, method)
	371	(receivedQuery, receivedResponse) = sender(query, response)
	372	self.assertTrue(receivedQuery)
	373	receivedQuery.id = query.id
4bfebc93 CH	374	self.assertEqual(query, receivedQuery)
4bfebc93 CH	375	self.assertEqual(receivedResponse, None)
c8c3d4e4 RG	376
	377	def testHeaderOnlyNXDResponse(self):
	378	"""
	379	Basics: Header-only NXD response should be dropped
	380	"""
	381	name = 'header-only-nxd-response.tests.powerdns.com.'
	382	query = dns.message.make_query(name, 'A', 'IN')
	383	response = dns.message.make_response(query)
	384	response.set_rcode(dns.rcode.NXDOMAIN)
	385	response.question = []
	386
6ca2e796 RG	387	for method in ("sendUDPQuery", "sendTCPQuery"):
	388	sender = getattr(self, method)
	389	(receivedQuery, receivedResponse) = sender(query, response)
	390	self.assertTrue(receivedQuery)
	391	receivedQuery.id = query.id
4bfebc93 CH	392	self.assertEqual(query, receivedQuery)
4bfebc93 CH	393	self.assertEqual(receivedResponse, None)
c8c3d4e4	394
f850b032 PL	395	def testAddActionDNSName(self):
	396	"""
	397	Basics: test if addAction accepts a DNSName
	398	"""
	399	name = 'dnsname.addaction.powerdns.com.'
	400	query = dns.message.make_query(name, 'A', 'IN')
7af22479	401	query.flags &= ~dns.flags.RD
f850b032 PL	402	expectedResponse = dns.message.make_response(query)
	403	expectedResponse.set_rcode(dns.rcode.REFUSED)
	404
6ca2e796 RG	405	for method in ("sendUDPQuery", "sendTCPQuery"):
	406	sender = getattr(self, method)
	407	(_, receivedResponse) = sender(query, response=None, useQueue=False)
4bfebc93	408	self.assertEqual(receivedResponse, expectedResponse)
f850b032 PL	409
	410	def testAddActionDNSNames(self):
	411	"""
	412	Basics: test if addAction accepts a table of DNSNames
	413	"""
	414	for name in ['dnsname-table{}.addaction.powerdns.com.'.format(i) for i in range(1,2)]:
	415	query = dns.message.make_query(name, 'A', 'IN')
7af22479	416	query.flags &= ~dns.flags.RD
f850b032 PL	417	expectedResponse = dns.message.make_response(query)
	418	expectedResponse.set_rcode(dns.rcode.REFUSED)
	419
6ca2e796 RG	420	for method in ("sendUDPQuery", "sendTCPQuery"):
	421	sender = getattr(self, method)
	422	(_, receivedResponse) = sender(query, response=None, useQueue=False)
4bfebc93	423	self.assertEqual(receivedResponse, expectedResponse)
ca404e94	424
9a872eb7 RG	425	def testEmptyQueries(self):
	426	"""
	427	Basic: NotImp on empty queries
	428	"""
	429	name = 'notimp-empty-queries.basic.tests.powerdns.com.'
	430	query = dns.message.Message()
	431
	432	response = dns.message.make_response(query)
	433	response.set_rcode(dns.rcode.NOTIMP)
	434
	435	for method in ("sendUDPQuery", "sendTCPQuery"):
	436	sender = getattr(self, method)
	437	(_, receivedResponse) = sender(query, response=None, useQueue=False)
4bfebc93	438	self.assertEqual(receivedResponse, response)