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

#!/usr/bin/env python
import threading
import time
import dns
from dnsdisttests import DNSDistTest, pickAvailablePort

class TestAXFR(DNSDistTest):

    # this test suite uses a different responder port
    # because, contrary to the other ones, its
    # TCP responder allows multiple responses and we don't want
    # to mix things up.
    _testServerPort = pickAvailablePort()
    _config_template = """
    newServer{address="127.0.0.1:%s"}
    """

    @classmethod
    def startResponders(cls):
        print("Launching responders..")

        cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
        cls._UDPResponder.daemon = True
        cls._UDPResponder.start()
        cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, False, True, None, None, True])
        cls._TCPResponder.daemon = True
        cls._TCPResponder.start()

    def setUp(self):
        # This function is called before every test
        super(TestAXFR, self).setUp()
        # make sure the TCP connection handlers from the previous test
        # are not going to read our queue
        time.sleep(1)

    def testOneMessageAXFR(self):
        """
        AXFR: One message
        """
        name = 'one.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        response = dns.message.make_response(query)
        soa = dns.rrset.from_text(name,
                                  60,
                                  dns.rdataclass.IN,
                                  dns.rdatatype.SOA,
                                  'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
        response.answer.append(soa)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        response.answer.append(soa)

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

    def testOneNoSOAAXFR(self):
        """
        AXFR: One message, no SOA
        """
        name = 'onenosoa.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        response = dns.message.make_response(query)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))

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

    def testFourMessagesAXFR(self):
        """
        AXFR: Four messages
        """
        name = 'four.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        responses = []
        soa = dns.rrset.from_text(name,
                                  60,
                                  dns.rdataclass.IN,
                                  dns.rdatatype.SOA,
                                  'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
        response = dns.message.make_response(query)
        response.answer.append(soa)
        responses.append(response)

        response = dns.message.make_response(query)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.AAAA,
                                    '2001:DB8::1')
        response.answer.append(rrset)
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'dummy')
        response.answer.append(rrset)
        responses.append(response)

        response = dns.message.make_response(query)
        response.answer.append(soa)
        responses.append(response)

        (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(len(receivedResponses), len(responses))

    def testFourNoFinalSOAAXFR(self):
        """
        AXFR: Four messages, no final SOA
        """
        name = 'fournosoa.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        responses = []
        soa = dns.rrset.from_text(name,
                                  60,
                                  dns.rdataclass.IN,
                                  dns.rdatatype.SOA,
                                  'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
        response = dns.message.make_response(query)
        response.answer.append(soa)
        responses.append(response)

        response = dns.message.make_response(query)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.AAAA,
                                    '2001:DB8::1')
        response.answer.append(rrset)
        responses.append(response)

        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'dummy')
        response.answer.append(rrset)
        responses.append(response)

        (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(len(receivedResponses), len(responses))

    def testThreePlusTrailingAXFR(self):
        """
        AXFR: Three messages including the final SOA, plus a trailing one
        """
        name = 'threeplustrailing.axfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        responses = []
        soa = dns.rrset.from_text(name,
                                  60,
                                  dns.rdataclass.IN,
                                  dns.rdatatype.SOA,
                                  'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')

        # the SOA starts the AXFR
        response = dns.message.make_response(query)
        response.answer.append(soa)
        responses.append(response)

        # one A
        response = dns.message.make_response(query)
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        responses.append(response)

        # one AAAA
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.AAAA,
                                    '2001:DB8::1')
        response.answer.append(rrset)
        responses.append(response)

        # one TXT then the SOA that ends the AXFR
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    "Some text")
        response.answer.append(rrset)
        response.answer.append(soa)
        responses.append(response)

        # and we add a final, dummy TXT message that will
        # be sent by the backend but that dnsdist should not
        # pass along to the client
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'dummy')
        response.answer.append(rrset)
        responses.append(response)

        (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(len(receivedResponses), len(responses) - 1)

    def testThreePlusTrailingIXFR(self):
        """
        IXFR: Three messages including the final SOA, plus a trailing one
        """
        name = 'threeplustrailing.ixfr.tests.powerdns.com.'
        query = dns.message.make_query(name, 'AXFR', 'IN')
        responses = []

        finalSoa = dns.rrset.from_text(name,
                                       60,
                                       dns.rdataclass.IN,
                                       dns.rdatatype.SOA,
                                       'ns.' + name + ' hostmaster.' + name + ' 3 3600 3600 3600 60')

        # the final SOA starts the IXFR, with first an update from 1 to 2 (one removal, two additions)
        response = dns.message.make_response(query)
        response.answer.append(finalSoa)
        # update from 1 to 2
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.SOA,
                                                   'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60'))
        # one removal
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.1'))
        # then additions
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.SOA,
                                                   'ns.' + name + ' hostmaster.' + name + ' 2 3600 3600 3600 60'))
        # new message in the middle of the additions
        responses.append(response)
        response = dns.message.make_response(query)

        response.answer.append(dns.rrset.from_text_list(name,
                                                        60,
                                                        dns.rdataclass.IN,
                                                        dns.rdatatype.A,
                                                        ['192.0.2.2', '192.0.2.3']))
        # done with 1 -> 2
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.SOA,
                                                   'ns.' + name + ' hostmaster.' + name + ' 2 3600 3600 3600 60'))
        # new message
        responses.append(response)
        response = dns.message.make_response(query)

        # then upgrade to 3
        # no removals
        response.answer.append(finalSoa)

        # and one addition
        response.answer.append(dns.rrset.from_text(name,
                                                   60,
                                                   dns.rdataclass.IN,
                                                   dns.rdatatype.A,
                                                   '192.0.2.4'))
        # and the final SOA
        response.answer.append(finalSoa)
        responses.append(response)

        # and we add a final, dummy TXT message that will
        # be sent by the backend but that dnsdist should not
        # pass along to the client
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'dummy')
        response.answer.append(rrset)
        responses.append(response)

        (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(len(receivedResponses), len(responses) - 1)
Commit	Line	Data
548c8b66 RG	1	#!/usr/bin/env python
548c8b66 RG	2	import threading
936dd73c	3	import time
548c8b66	4	import dns
630eb526	5	from dnsdisttests import DNSDistTest, pickAvailablePort
548c8b66 RG	6
	7	class TestAXFR(DNSDistTest):
	8
	9	# this test suite uses a different responder port
	10	# because, contrary to the other ones, its
	11	# TCP responder allows multiple responses and we don't want
	12	# to mix things up.
630eb526	13	_testServerPort = pickAvailablePort()
548c8b66 RG	14	_config_template = """
	15	newServer{address="127.0.0.1:%s"}
	16	"""
645a1ca4	17
548c8b66 RG	18	@classmethod
	19	def startResponders(cls):
	20	print("Launching responders..")
	21
5df86a8a	22	cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
630eb526	23	cls._UDPResponder.daemon = True
548c8b66	24	cls._UDPResponder.start()
645a1ca4	25	cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, False, True, None, None, True])
630eb526	26	cls._TCPResponder.daemon = True
548c8b66 RG	27	cls._TCPResponder.start()
548c8b66 RG	28
936dd73c RG	29	def setUp(self):
	30	# This function is called before every test
	31	super(TestAXFR, self).setUp()
	32	# make sure the TCP connection handlers from the previous test
	33	# are not going to read our queue
	34	time.sleep(1)
	35
548c8b66 RG	36	def testOneMessageAXFR(self):
	37	"""
	38	AXFR: One message
	39	"""
	40	name = 'one.axfr.tests.powerdns.com.'
	41	query = dns.message.make_query(name, 'AXFR', 'IN')
	42	response = dns.message.make_response(query)
	43	soa = dns.rrset.from_text(name,
	44	60,
	45	dns.rdataclass.IN,
	46	dns.rdatatype.SOA,
	47	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
	48	response.answer.append(soa)
	49	response.answer.append(dns.rrset.from_text(name,
	50	60,
	51	dns.rdataclass.IN,
	52	dns.rdatatype.A,
	53	'192.0.2.1'))
	54	response.answer.append(soa)
	55
	56	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	57	receivedQuery.id = query.id
	58	self.assertEqual(query, receivedQuery)
	59	self.assertEqual(response, receivedResponse)
	60
	61	def testOneNoSOAAXFR(self):
	62	"""
	63	AXFR: One message, no SOA
	64	"""
	65	name = 'onenosoa.axfr.tests.powerdns.com.'
	66	query = dns.message.make_query(name, 'AXFR', 'IN')
	67	response = dns.message.make_response(query)
	68	response.answer.append(dns.rrset.from_text(name,
	69	60,
	70	dns.rdataclass.IN,
	71	dns.rdatatype.A,
	72	'192.0.2.1'))
	73
	74	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	75	receivedQuery.id = query.id
	76	self.assertEqual(query, receivedQuery)
	77	self.assertEqual(response, receivedResponse)
	78
	79	def testFourMessagesAXFR(self):
	80	"""
	81	AXFR: Four messages
	82	"""
	83	name = 'four.axfr.tests.powerdns.com.'
	84	query = dns.message.make_query(name, 'AXFR', 'IN')
	85	responses = []
	86	soa = dns.rrset.from_text(name,
	87	60,
	88	dns.rdataclass.IN,
	89	dns.rdatatype.SOA,
	90	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
	91	response = dns.message.make_response(query)
	92	response.answer.append(soa)
	93	responses.append(response)
	94
	95	response = dns.message.make_response(query)
	96	response.answer.append(dns.rrset.from_text(name,
	97	60,
	98	dns.rdataclass.IN,
	99	dns.rdatatype.A,
100	'192.0.2.1'))
101	responses.append(response)
102
103	response = dns.message.make_response(query)
104	rrset = dns.rrset.from_text(name,
105	60,
106	dns.rdataclass.IN,
107	dns.rdatatype.AAAA,
108	'2001:DB8::1')
109	response.answer.append(rrset)
110	responses.append(response)
111
112	response = dns.message.make_response(query)
113	rrset = dns.rrset.from_text(name,
114	60,
115	dns.rdataclass.IN,
116	dns.rdatatype.TXT,
117	'dummy')
118	response.answer.append(rrset)
119	responses.append(response)
120
121	response = dns.message.make_response(query)
122	response.answer.append(soa)
123	responses.append(response)
124
125	(receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
126	receivedQuery.id = query.id
127	self.assertEqual(query, receivedQuery)
128	self.assertEqual(len(receivedResponses), len(responses))
129
130	def testFourNoFinalSOAAXFR(self):
131	"""
132	AXFR: Four messages, no final SOA
133	"""
134	name = 'fournosoa.axfr.tests.powerdns.com.'
135	query = dns.message.make_query(name, 'AXFR', 'IN')
136	responses = []
137	soa = dns.rrset.from_text(name,
138	60,
139	dns.rdataclass.IN,
140	dns.rdatatype.SOA,
141	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
142	response = dns.message.make_response(query)
143	response.answer.append(soa)
144	responses.append(response)
145
146	response = dns.message.make_response(query)
147	response.answer.append(dns.rrset.from_text(name,
148	60,
149	dns.rdataclass.IN,
150	dns.rdatatype.A,
151	'192.0.2.1'))
152	responses.append(response)
153
154	response = dns.message.make_response(query)
155	rrset = dns.rrset.from_text(name,
156	60,
157	dns.rdataclass.IN,
158	dns.rdatatype.AAAA,
159	'2001:DB8::1')
160	response.answer.append(rrset)
161	responses.append(response)
162
163	response = dns.message.make_response(query)
164	rrset = dns.rrset.from_text(name,
165	60,
166	dns.rdataclass.IN,
167	dns.rdatatype.TXT,
168	'dummy')
169	response.answer.append(rrset)
170	responses.append(response)
171
172	(receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
173	receivedQuery.id = query.id
174	self.assertEqual(query, receivedQuery)
175	self.assertEqual(len(receivedResponses), len(responses))
176
4639c3f5 RG	177	def testThreePlusTrailingAXFR(self):
	178	"""
	179	AXFR: Three messages including the final SOA, plus a trailing one
	180	"""
	181	name = 'threeplustrailing.axfr.tests.powerdns.com.'
	182	query = dns.message.make_query(name, 'AXFR', 'IN')
	183	responses = []
	184	soa = dns.rrset.from_text(name,
	185	60,
	186	dns.rdataclass.IN,
	187	dns.rdatatype.SOA,
	188	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
	189
	190	# the SOA starts the AXFR
	191	response = dns.message.make_response(query)
	192	response.answer.append(soa)
	193	responses.append(response)
	194
	195	# one A
	196	response = dns.message.make_response(query)
	197	response.answer.append(dns.rrset.from_text(name,
	198	60,
	199	dns.rdataclass.IN,
	200	dns.rdatatype.A,
	201	'192.0.2.1'))
	202	responses.append(response)
	203
	204	# one AAAA
	205	response = dns.message.make_response(query)
	206	rrset = dns.rrset.from_text(name,
	207	60,
	208	dns.rdataclass.IN,
	209	dns.rdatatype.AAAA,
	210	'2001:DB8::1')
	211	response.answer.append(rrset)
	212	responses.append(response)
	213
	214	# one TXT then the SOA that ends the AXFR
	215	response = dns.message.make_response(query)
	216	rrset = dns.rrset.from_text(name,
	217	60,
	218	dns.rdataclass.IN,
	219	dns.rdatatype.TXT,
	220	"Some text")
	221	response.answer.append(rrset)
	222	response.answer.append(soa)
	223	responses.append(response)
	224
	225	# and we add a final, dummy TXT message that will
	226	# be sent by the backend but that dnsdist should not
	227	# pass along to the client
	228	response = dns.message.make_response(query)
	229	rrset = dns.rrset.from_text(name,
	230	60,
	231	dns.rdataclass.IN,
	232	dns.rdatatype.TXT,
	233	'dummy')
	234	response.answer.append(rrset)
	235	responses.append(response)
	236
	237	(receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
	238	receivedQuery.id = query.id
	239	self.assertEqual(query, receivedQuery)
	240	self.assertEqual(len(receivedResponses), len(responses) - 1)
241
242	def testThreePlusTrailingIXFR(self):
243	"""
244	IXFR: Three messages including the final SOA, plus a trailing one
245	"""
246	name = 'threeplustrailing.ixfr.tests.powerdns.com.'
247	query = dns.message.make_query(name, 'AXFR', 'IN')
248	responses = []
249
250	finalSoa = dns.rrset.from_text(name,
251	60,
252	dns.rdataclass.IN,
253	dns.rdatatype.SOA,
254	'ns.' + name + ' hostmaster.' + name + ' 3 3600 3600 3600 60')
255
936dd73c	256	# the final SOA starts the IXFR, with first an update from 1 to 2 (one removal, two additions)
4639c3f5 RG	257	response = dns.message.make_response(query)
	258	response.answer.append(finalSoa)
	259	# update from 1 to 2
	260	response.answer.append(dns.rrset.from_text(name,
	261	60,
	262	dns.rdataclass.IN,
	263	dns.rdatatype.SOA,
	264	'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60'))
	265	# one removal
	266	response.answer.append(dns.rrset.from_text(name,
	267	60,
	268	dns.rdataclass.IN,
	269	dns.rdatatype.A,
	270	'192.0.2.1'))
	271	# then additions
	272	response.answer.append(dns.rrset.from_text(name,
	273	60,
	274	dns.rdataclass.IN,
	275	dns.rdatatype.SOA,
	276	'ns.' + name + ' hostmaster.' + name + ' 2 3600 3600 3600 60'))
	277	# new message in the middle of the additions
	278	responses.append(response)
	279	response = dns.message.make_response(query)
	280
	281	response.answer.append(dns.rrset.from_text_list(name,
	282	60,
	283	dns.rdataclass.IN,
	284	dns.rdatatype.A,
	285	['192.0.2.2', '192.0.2.3']))
	286	# done with 1 -> 2
	287	response.answer.append(dns.rrset.from_text(name,
	288	60,
	289	dns.rdataclass.IN,
	290	dns.rdatatype.SOA,
	291	'ns.' + name + ' hostmaster.' + name + ' 2 3600 3600 3600 60'))
	292	# new message
	293	responses.append(response)
	294	response = dns.message.make_response(query)
	295
	296	# then upgrade to 3
	297	# no removals
	298	response.answer.append(finalSoa)
	299
	300	# and one addition
	301	response.answer.append(dns.rrset.from_text(name,
	302	60,
	303	dns.rdataclass.IN,
	304	dns.rdatatype.A,
	305	'192.0.2.4'))
	306	# and the final SOA
	307	response.answer.append(finalSoa)
	308	responses.append(response)
	309
	310	# and we add a final, dummy TXT message that will
	311	# be sent by the backend but that dnsdist should not
	312	# pass along to the client
	313	response = dns.message.make_response(query)
	314	rrset = dns.rrset.from_text(name,
	315	60,
	316	dns.rdataclass.IN,
	317	dns.rdatatype.TXT,
	318	'dummy')
	319	response.answer.append(rrset)
	320	responses.append(response)
321
322	(receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
323	receivedQuery.id = query.id
324	self.assertEqual(query, receivedQuery)
325	self.assertEqual(len(receivedResponses), len(responses) - 1)