regression-tests.dnsdist/test_AXFR.py

   1 #!/usr/bin/env python
   2 import threading
   3 import dns
   4 from dnsdisttests import DNSDistTest
   5
   6 class TestAXFR(DNSDistTest):
   7
   8     # this test suite uses a different responder port
   9     # because, contrary to the other ones, its
  10     # TCP responder allows multiple responses and we don't want
  11     # to mix things up.
  12     _testServerPort = 5370
  13     _config_template = """
  14     newServer{address="127.0.0.1:%s"}
  15     """
  16     @classmethod
  17     def startResponders(cls):
  18         print("Launching responders..")
  19
  20         cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
  21         cls._UDPResponder.setDaemon(True)
  22         cls._UDPResponder.start()
  23         cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, False, True])
  24         cls._TCPResponder.setDaemon(True)
  25         cls._TCPResponder.start()
  26
  27     def testOneMessageAXFR(self):
  28         """
  29         AXFR: One message
  30         """
  31         name = 'one.axfr.tests.powerdns.com.'
  32         query = dns.message.make_query(name, 'AXFR', 'IN')
  33         response = dns.message.make_response(query)
  34         soa = dns.rrset.from_text(name,
  35                                   60,
  36                                   dns.rdataclass.IN,
  37                                   dns.rdatatype.SOA,
  38                                   'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
  39         response.answer.append(soa)
  40         response.answer.append(dns.rrset.from_text(name,
  41                                                    60,
  42                                                    dns.rdataclass.IN,
  43                                                    dns.rdatatype.A,
  44                                                    '192.0.2.1'))
  45         response.answer.append(soa)
  46
  47         (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
  48         receivedQuery.id = query.id
  49         self.assertEqual(query, receivedQuery)
  50         self.assertEqual(response, receivedResponse)
  51
  52     def testOneNoSOAAXFR(self):
  53         """
  54         AXFR: One message, no SOA
  55         """
  56         name = 'onenosoa.axfr.tests.powerdns.com.'
  57         query = dns.message.make_query(name, 'AXFR', 'IN')
  58         response = dns.message.make_response(query)
  59         response.answer.append(dns.rrset.from_text(name,
  60                                                    60,
  61                                                    dns.rdataclass.IN,
  62                                                    dns.rdatatype.A,
  63                                                    '192.0.2.1'))
  64
  65         (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
  66         receivedQuery.id = query.id
  67         self.assertEqual(query, receivedQuery)
  68         self.assertEqual(response, receivedResponse)
  69
  70     def testFourMessagesAXFR(self):
  71         """
  72         AXFR: Four messages
  73         """
  74         name = 'four.axfr.tests.powerdns.com.'
  75         query = dns.message.make_query(name, 'AXFR', 'IN')
  76         responses = []
  77         soa = dns.rrset.from_text(name,
  78                                   60,
  79                                   dns.rdataclass.IN,
  80                                   dns.rdatatype.SOA,
  81                                   'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
  82         response = dns.message.make_response(query)
  83         response.answer.append(soa)
  84         responses.append(response)
  85
  86         response = dns.message.make_response(query)
  87         response.answer.append(dns.rrset.from_text(name,
  88                                                    60,
  89                                                    dns.rdataclass.IN,
  90                                                    dns.rdatatype.A,
  91                                                    '192.0.2.1'))
  92         responses.append(response)
  93
  94         response = dns.message.make_response(query)
  95         rrset = dns.rrset.from_text(name,
  96                                     60,
  97                                     dns.rdataclass.IN,
  98                                     dns.rdatatype.AAAA,
  99                                     '2001:DB8::1')
 100         response.answer.append(rrset)
 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.TXT,
 108                                     'dummy')
 109         response.answer.append(rrset)
 110         responses.append(response)
 111
 112         response = dns.message.make_response(query)
 113         response.answer.append(soa)
 114         responses.append(response)
 115
 116         (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
 117         receivedQuery.id = query.id
 118         self.assertEqual(query, receivedQuery)
 119         self.assertEqual(len(receivedResponses), len(responses))
 120
 121     def testFourNoFinalSOAAXFR(self):
 122         """
 123         AXFR: Four messages, no final SOA
 124         """
 125         name = 'fournosoa.axfr.tests.powerdns.com.'
 126         query = dns.message.make_query(name, 'AXFR', 'IN')
 127         responses = []
 128         soa = dns.rrset.from_text(name,
 129                                   60,
 130                                   dns.rdataclass.IN,
 131                                   dns.rdatatype.SOA,
 132                                   'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
 133         response = dns.message.make_response(query)
 134         response.answer.append(soa)
 135         responses.append(response)
 136
 137         response = dns.message.make_response(query)
 138         response.answer.append(dns.rrset.from_text(name,
 139                                                    60,
 140                                                    dns.rdataclass.IN,
 141                                                    dns.rdatatype.A,
 142                                                    '192.0.2.1'))
 143         responses.append(response)
 144
 145         response = dns.message.make_response(query)
 146         rrset = dns.rrset.from_text(name,
 147                                     60,
 148                                     dns.rdataclass.IN,
 149                                     dns.rdatatype.AAAA,
 150                                     '2001:DB8::1')
 151         response.answer.append(rrset)
 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.TXT,
 159                                     'dummy')
 160         response.answer.append(rrset)
 161         responses.append(response)
 162
 163         (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
 164         receivedQuery.id = query.id
 165         self.assertEqual(query, receivedQuery)
 166         self.assertEqual(len(receivedResponses), len(responses))
 167
 168     def testFourNoFirstSOAAXFR(self):
 169         """
 170         AXFR: Four messages, no SOA in the first one
 171         """
 172         name = 'fournosoainfirst.axfr.tests.powerdns.com.'
 173         query = dns.message.make_query(name, 'AXFR', 'IN')
 174         responses = []
 175         soa = dns.rrset.from_text(name,
 176                                   60,
 177                                   dns.rdataclass.IN,
 178                                   dns.rdatatype.SOA,
 179                                   'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
 180         response = dns.message.make_response(query)
 181         response.answer.append(dns.rrset.from_text(name,
 182                                                    60,
 183                                                    dns.rdataclass.IN,
 184                                                    dns.rdatatype.A,
 185                                                    '192.0.2.1'))
 186         responses.append(response)
 187
 188         response = dns.message.make_response(query)
 189         rrset = dns.rrset.from_text(name,
 190                                     60,
 191                                     dns.rdataclass.IN,
 192                                     dns.rdatatype.AAAA,
 193                                     '2001:DB8::1')
 194         response.answer.append(soa)
 195         response.answer.append(rrset)
 196         responses.append(response)
 197
 198         response = dns.message.make_response(query)
 199         rrset = dns.rrset.from_text('dummy.' + name,
 200                                     60,
 201                                     dns.rdataclass.IN,
 202                                     dns.rdatatype.AAAA,
 203                                     '2001:DB8::1')
 204         response.answer.append(rrset)
 205         responses.append(response)
 206
 207         response = dns.message.make_response(query)
 208         rrset = dns.rrset.from_text(name,
 209                                     60,
 210                                     dns.rdataclass.IN,
 211                                     dns.rdatatype.TXT,
 212                                     'dummy')
 213         response.answer.append(rrset)
 214         response.answer.append(soa)
 215         responses.append(response)
 216
 217         (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
 218         receivedQuery.id = query.id
 219         self.assertEqual(query, receivedQuery)
 220         self.assertEqual(len(receivedResponses), 1)
 221
 222     def testFourLastSOAInSecondAXFR(self):
 223         """
 224         AXFR: Four messages, SOA in the first one and the second one
 225         """
 226         name = 'foursecondsoainsecond.axfr.tests.powerdns.com.'
 227         query = dns.message.make_query(name, 'AXFR', 'IN')
 228         responses = []
 229         soa = dns.rrset.from_text(name,
 230                                   60,
 231                                   dns.rdataclass.IN,
 232                                   dns.rdatatype.SOA,
 233                                   'ns.' + name + ' hostmaster.' + name + ' 1 3600 3600 3600 60')
 234
 235         response = dns.message.make_response(query)
 236         response.answer.append(soa)
 237         response.answer.append(dns.rrset.from_text(name,
 238                                                    60,
 239                                                    dns.rdataclass.IN,
 240                                                    dns.rdatatype.A,
 241                                                    '192.0.2.1'))
 242         responses.append(response)
 243
 244         response = dns.message.make_response(query)
 245         response.answer.append(soa)
 246         rrset = dns.rrset.from_text(name,
 247                                     60,
 248                                     dns.rdataclass.IN,
 249                                     dns.rdatatype.AAAA,
 250                                     '2001:DB8::1')
 251         response.answer.append(rrset)
 252         responses.append(response)
 253
 254         response = dns.message.make_response(query)
 255         rrset = dns.rrset.from_text('dummy.' + name,
 256                                     60,
 257                                     dns.rdataclass.IN,
 258                                     dns.rdatatype.AAAA,
 259                                     '2001:DB8::1')
 260         response.answer.append(rrset)
 261         responses.append(response)
 262
 263         response = dns.message.make_response(query)
 264         rrset = dns.rrset.from_text(name,
 265                                     60,
 266                                     dns.rdataclass.IN,
 267                                     dns.rdatatype.TXT,
 268                                     'dummy')
 269         response.answer.append(rrset)
 270         responses.append(response)
 271
 272         (receivedQuery, receivedResponses) = self.sendTCPQueryWithMultipleResponses(query, responses)
 273         receivedQuery.id = query.id
 274         self.assertEqual(query, receivedQuery)
 275         self.assertEqual(len(receivedResponses), 2)