python/samba/tests/dns.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Kai Blin  <kai@samba.org> 2011
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 import os
  19 import struct
  20 import random
  21 import socket
  22 import samba.ndr as ndr
  23 import samba.dcerpc.dns as dns
  24 from samba.tests import TestCase
  25
  26 FILTER=''.join([(len(repr(chr(x)))==3) and chr(x) or '.' for x in range(256)])
  27
  28
  29 class DNSTest(TestCase):
  30
  31     def errstr(self, errcode):
  32         "Return a readable error code"
  33         string_codes = [
  34             "OK",
  35             "FORMERR",
  36             "SERVFAIL",
  37             "NXDOMAIN",
  38             "NOTIMP",
  39             "REFUSED",
  40             "YXDOMAIN",
  41             "YXRRSET",
  42             "NXRRSET",
  43             "NOTAUTH",
  44             "NOTZONE",
  45         ]
  46
  47         return string_codes[errcode]
  48
  49
  50     def assert_dns_rcode_equals(self, packet, rcode):
  51         "Helper function to check return code"
  52         p_errcode = packet.operation & 0x000F
  53         self.assertEquals(p_errcode, rcode, "Expected RCODE %s, got %s" %
  54                             (self.errstr(rcode), self.errstr(p_errcode)))
  55
  56     def assert_dns_opcode_equals(self, packet, opcode):
  57         "Helper function to check opcode"
  58         p_opcode = packet.operation & 0x7800
  59         self.assertEquals(p_opcode, opcode, "Expected OPCODE %s, got %s" %
  60                             (opcode, p_opcode))
  61
  62     def make_name_packet(self, opcode, qid=None):
  63         "Helper creating a dns.name_packet"
  64         p = dns.name_packet()
  65         if qid is None:
  66             p.id = random.randint(0x0, 0xffff)
  67         p.operation = opcode
  68         p.questions = []
  69         return p
  70
  71     def finish_name_packet(self, packet, questions):
  72         "Helper to finalize a dns.name_packet"
  73         packet.qdcount = len(questions)
  74         packet.questions = questions
  75
  76     def make_name_question(self, name, qtype, qclass):
  77         "Helper creating a dns.name_question"
  78         q = dns.name_question()
  79         q.name = name
  80         q.question_type = qtype
  81         q.question_class = qclass
  82         return q
  83
  84     def get_dns_domain(self):
  85         "Helper to get dns domain"
  86         return os.getenv('REALM', 'example.com').lower()
  87
  88     def dns_transaction_udp(self, packet, host=os.getenv('SERVER_IP'), dump=False):
  89         "send a DNS query and read the reply"
  90         s = None
  91         try:
  92             send_packet = ndr.ndr_pack(packet)
  93             if dump:
  94                 print self.hexdump(send_packet)
  95             s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
  96             s.connect((host, 53))
  97             s.send(send_packet, 0)
  98             recv_packet = s.recv(2048, 0)
  99             if dump:
 100                 print self.hexdump(recv_packet)
 101             return ndr.ndr_unpack(dns.name_packet, recv_packet)
 102         finally:
 103             if s is not None:
 104                 s.close()
 105
 106     def dns_transaction_tcp(self, packet, host=os.getenv('SERVER_IP'), dump=False):
 107         "send a DNS query and read the reply"
 108         s = None
 109         try:
 110             send_packet = ndr.ndr_pack(packet)
 111             if dump:
 112                 print self.hexdump(send_packet)
 113             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
 114             s.connect((host, 53))
 115             tcp_packet = struct.pack('!H', len(send_packet))
 116             tcp_packet += send_packet
 117             s.send(tcp_packet, 0)
 118             recv_packet = s.recv(0xffff + 2, 0)
 119             if dump:
 120                 print self.hexdump(recv_packet)
 121             return ndr.ndr_unpack(dns.name_packet, recv_packet[2:])
 122         finally:
 123                 if s is not None:
 124                     s.close()
 125
 126     def hexdump(self, src, length=8):
 127         N=0; result=''
 128         while src:
 129            s,src = src[:length],src[length:]
 130            hexa = ' '.join(["%02X"%ord(x) for x in s])
 131            s = s.translate(FILTER)
 132            result += "%04X   %-*s   %s\n" % (N, length*3, hexa, s)
 133            N+=length
 134         return result
 135
 136 class TestSimpleQueries(DNSTest):
 137
 138     def test_one_a_query(self):
 139         "create a query packet containing one query record"
 140         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 141         questions = []
 142
 143         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 144         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 145         print "asking for ", q.name
 146         questions.append(q)
 147
 148         self.finish_name_packet(p, questions)
 149         response = self.dns_transaction_udp(p)
 150         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 151         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 152         self.assertEquals(response.ancount, 1)
 153         self.assertEquals(response.answers[0].rdata,
 154                           os.getenv('SERVER_IP'))
 155
 156     def test_one_a_query_tcp(self):
 157         "create a query packet containing one query record via TCP"
 158         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 159         questions = []
 160
 161         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 162         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 163         print "asking for ", q.name
 164         questions.append(q)
 165
 166         self.finish_name_packet(p, questions)
 167         response = self.dns_transaction_tcp(p)
 168         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 169         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 170         self.assertEquals(response.ancount, 1)
 171         self.assertEquals(response.answers[0].rdata,
 172                           os.getenv('SERVER_IP'))
 173
 174     def test_one_mx_query(self):
 175         "create a query packet causing an empty RCODE_OK answer"
 176         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 177         questions = []
 178
 179         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 180         q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
 181         print "asking for ", q.name
 182         questions.append(q)
 183
 184         self.finish_name_packet(p, questions)
 185         response = self.dns_transaction_udp(p)
 186         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 187         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 188         self.assertEquals(response.ancount, 0)
 189
 190         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 191         questions = []
 192
 193         name = "invalid-%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 194         q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
 195         print "asking for ", q.name
 196         questions.append(q)
 197
 198         self.finish_name_packet(p, questions)
 199         response = self.dns_transaction_udp(p)
 200         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
 201         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 202         self.assertEquals(response.ancount, 0)
 203
 204     def test_two_queries(self):
 205         "create a query packet containing two query records"
 206         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 207         questions = []
 208
 209         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 210         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 211         questions.append(q)
 212
 213         name = "%s.%s" % ('bogusname', self.get_dns_domain())
 214         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 215         questions.append(q)
 216
 217         self.finish_name_packet(p, questions)
 218         response = self.dns_transaction_udp(p)
 219         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 220
 221     def test_qtype_all_query(self):
 222         "create a QTYPE_ALL query"
 223         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 224         questions = []
 225
 226         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 227         q = self.make_name_question(name, dns.DNS_QTYPE_ALL, dns.DNS_QCLASS_IN)
 228         print "asking for ", q.name
 229         questions.append(q)
 230
 231         self.finish_name_packet(p, questions)
 232         response = self.dns_transaction_udp(p)
 233
 234         num_answers = 1
 235         dc_ipv6 = os.getenv('SERVER_IPV6')
 236         if dc_ipv6 is not None:
 237             num_answers += 1
 238
 239         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 240         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 241         self.assertEquals(response.ancount, num_answers)
 242         self.assertEquals(response.answers[0].rdata,
 243                           os.getenv('SERVER_IP'))
 244         if dc_ipv6 is not None:
 245             self.assertEquals(response.answers[1].rdata, dc_ipv6)
 246
 247     def test_qclass_none_query(self):
 248         "create a QCLASS_NONE query"
 249         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 250         questions = []
 251
 252         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 253         q = self.make_name_question(name, dns.DNS_QTYPE_ALL, dns.DNS_QCLASS_NONE)
 254         questions.append(q)
 255
 256         self.finish_name_packet(p, questions)
 257         response = self.dns_transaction_udp(p)
 258         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTIMP)
 259
 260 # Only returns an authority section entry in BIND and Win DNS
 261 # FIXME: Enable one Samba implements this feature
 262 #    def test_soa_hostname_query(self):
 263 #        "create a SOA query for a hostname"
 264 #        p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 265 #        questions = []
 266 #
 267 #        name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 268 #        q = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 269 #        questions.append(q)
 270 #
 271 #        self.finish_name_packet(p, questions)
 272 #        response = self.dns_transaction_udp(p)
 273 #        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 274 #        self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 275 #        # We don't get SOA records for single hosts
 276 #        self.assertEquals(response.ancount, 0)
 277
 278     def test_soa_domain_query(self):
 279         "create a SOA query for a domain"
 280         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 281         questions = []
 282
 283         name = self.get_dns_domain()
 284         q = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 285         questions.append(q)
 286
 287         self.finish_name_packet(p, questions)
 288         response = self.dns_transaction_udp(p)
 289         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 290         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 291         self.assertEquals(response.ancount, 1)
 292
 293
 294 class TestDNSUpdates(DNSTest):
 295
 296     def test_two_updates(self):
 297         "create two update requests"
 298         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 299         updates = []
 300
 301         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 302         u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 303         updates.append(u)
 304
 305         name = self.get_dns_domain()
 306         u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 307         updates.append(u)
 308
 309         self.finish_name_packet(p, updates)
 310         response = self.dns_transaction_udp(p)
 311         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 312
 313     def test_update_wrong_qclass(self):
 314         "create update with DNS_QCLASS_NONE"
 315         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 316         updates = []
 317
 318         name = self.get_dns_domain()
 319         u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_NONE)
 320         updates.append(u)
 321
 322         self.finish_name_packet(p, updates)
 323         response = self.dns_transaction_udp(p)
 324         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTIMP)
 325
 326     def test_update_prereq_with_non_null_ttl(self):
 327         "test update with a non-null TTL"
 328         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 329         updates = []
 330
 331         name = self.get_dns_domain()
 332
 333         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 334         updates.append(u)
 335         self.finish_name_packet(p, updates)
 336
 337         prereqs = []
 338         r = dns.res_rec()
 339         r.name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 340         r.rr_type = dns.DNS_QTYPE_TXT
 341         r.rr_class = dns.DNS_QCLASS_NONE
 342         r.ttl = 1
 343         r.length = 0
 344         prereqs.append(r)
 345
 346         p.ancount = len(prereqs)
 347         p.answers = prereqs
 348
 349         response = self.dns_transaction_udp(p)
 350         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 351
 352 # I'd love to test this one, but it segfaults. :)
 353 #    def test_update_prereq_with_non_null_length(self):
 354 #        "test update with a non-null length"
 355 #        p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 356 #        updates = []
 357 #
 358 #        name = self.get_dns_domain()
 359 #
 360 #        u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 361 #        updates.append(u)
 362 #        self.finish_name_packet(p, updates)
 363 #
 364 #        prereqs = []
 365 #        r = dns.res_rec()
 366 #        r.name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 367 #        r.rr_type = dns.DNS_QTYPE_TXT
 368 #        r.rr_class = dns.DNS_QCLASS_ANY
 369 #        r.ttl = 0
 370 #        r.length = 1
 371 #        prereqs.append(r)
 372 #
 373 #        p.ancount = len(prereqs)
 374 #        p.answers = prereqs
 375 #
 376 #        response = self.dns_transaction_udp(p)
 377 #        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 378
 379     def test_update_prereq_nonexisting_name(self):
 380         "test update with a nonexisting name"
 381         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 382         updates = []
 383
 384         name = self.get_dns_domain()
 385
 386         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 387         updates.append(u)
 388         self.finish_name_packet(p, updates)
 389
 390         prereqs = []
 391         r = dns.res_rec()
 392         r.name = "idontexist.%s" % self.get_dns_domain()
 393         r.rr_type = dns.DNS_QTYPE_TXT
 394         r.rr_class = dns.DNS_QCLASS_ANY
 395         r.ttl = 0
 396         r.length = 0
 397         prereqs.append(r)
 398
 399         p.ancount = len(prereqs)
 400         p.answers = prereqs
 401
 402         response = self.dns_transaction_udp(p)
 403         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET)
 404
 405     def test_update_add_txt_record(self):
 406         "test adding records works"
 407         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 408         updates = []
 409
 410         name = self.get_dns_domain()
 411
 412         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 413         updates.append(u)
 414         self.finish_name_packet(p, updates)
 415
 416         updates = []
 417         r = dns.res_rec()
 418         r.name = "textrec.%s" % self.get_dns_domain()
 419         r.rr_type = dns.DNS_QTYPE_TXT
 420         r.rr_class = dns.DNS_QCLASS_IN
 421         r.ttl = 900
 422         r.length = 0xffff
 423         rdata = dns.txt_record()
 424         rdata.txt = '"This is a test"'
 425         r.rdata = rdata
 426         updates.append(r)
 427         p.nscount = len(updates)
 428         p.nsrecs = updates
 429
 430         response = self.dns_transaction_udp(p)
 431         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 432
 433         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 434         questions = []
 435
 436         name = "textrec.%s" % self.get_dns_domain()
 437         q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 438         questions.append(q)
 439
 440         self.finish_name_packet(p, questions)
 441         response = self.dns_transaction_udp(p)
 442         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 443         self.assertEquals(response.ancount, 1)
 444         self.assertEquals(response.answers[0].rdata.txt, '"This is a test"')
 445
 446     def test_update_add_two_txt_records(self):
 447         "test adding two txt records works"
 448         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 449         updates = []
 450
 451         name = self.get_dns_domain()
 452
 453         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 454         updates.append(u)
 455         self.finish_name_packet(p, updates)
 456
 457         updates = []
 458         r = dns.res_rec()
 459         r.name = "textrec2.%s" % self.get_dns_domain()
 460         r.rr_type = dns.DNS_QTYPE_TXT
 461         r.rr_class = dns.DNS_QCLASS_IN
 462         r.ttl = 900
 463         r.length = 0xffff
 464         rdata = dns.txt_record()
 465         rdata.txt = '"This is a test" "and this is a test, too"'
 466         r.rdata = rdata
 467         updates.append(r)
 468         p.nscount = len(updates)
 469         p.nsrecs = updates
 470
 471         response = self.dns_transaction_udp(p)
 472         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 473
 474         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 475         questions = []
 476
 477         name = "textrec2.%s" % self.get_dns_domain()
 478         q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 479         questions.append(q)
 480
 481         self.finish_name_packet(p, questions)
 482         response = self.dns_transaction_udp(p)
 483         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 484         self.assertEquals(response.ancount, 1)
 485         self.assertEquals(response.answers[0].rdata.txt, '"This is a test" "and this is a test, too"')
 486
 487     def test_delete_record(self):
 488         "Test if deleting records works"
 489
 490         NAME = "deleterec.%s" % self.get_dns_domain()
 491
 492         # First, create a record to make sure we have a record to delete.
 493         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 494         updates = []
 495
 496         name = self.get_dns_domain()
 497
 498         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 499         updates.append(u)
 500         self.finish_name_packet(p, updates)
 501
 502         updates = []
 503         r = dns.res_rec()
 504         r.name = NAME
 505         r.rr_type = dns.DNS_QTYPE_TXT
 506         r.rr_class = dns.DNS_QCLASS_IN
 507         r.ttl = 900
 508         r.length = 0xffff
 509         rdata = dns.txt_record()
 510         rdata.txt = '"This is a test"'
 511         r.rdata = rdata
 512         updates.append(r)
 513         p.nscount = len(updates)
 514         p.nsrecs = updates
 515
 516         response = self.dns_transaction_udp(p)
 517         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 518
 519         # Now check the record is around
 520         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 521         questions = []
 522         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 523         questions.append(q)
 524
 525         self.finish_name_packet(p, questions)
 526         response = self.dns_transaction_udp(p)
 527         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 528
 529         # Now delete the record
 530         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 531         updates = []
 532
 533         name = self.get_dns_domain()
 534
 535         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 536         updates.append(u)
 537         self.finish_name_packet(p, updates)
 538
 539         updates = []
 540         r = dns.res_rec()
 541         r.name = NAME
 542         r.rr_type = dns.DNS_QTYPE_TXT
 543         r.rr_class = dns.DNS_QCLASS_NONE
 544         r.ttl = 0
 545         r.length = 0xffff
 546         rdata = dns.txt_record()
 547         rdata.txt = '"This is a test"'
 548         r.rdata = rdata
 549         updates.append(r)
 550         p.nscount = len(updates)
 551         p.nsrecs = updates
 552
 553         response = self.dns_transaction_udp(p)
 554         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 555
 556         # And finally check it's gone
 557         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 558         questions = []
 559
 560         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 561         questions.append(q)
 562
 563         self.finish_name_packet(p, questions)
 564         response = self.dns_transaction_udp(p)
 565         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
 566
 567     def test_readd_record(self):
 568         "Test if adding, deleting and then readding a records works"
 569
 570         NAME = "readdrec.%s" % self.get_dns_domain()
 571
 572         # Create the record
 573         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 574         updates = []
 575
 576         name = self.get_dns_domain()
 577
 578         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 579         updates.append(u)
 580         self.finish_name_packet(p, updates)
 581
 582         updates = []
 583         r = dns.res_rec()
 584         r.name = NAME
 585         r.rr_type = dns.DNS_QTYPE_TXT
 586         r.rr_class = dns.DNS_QCLASS_IN
 587         r.ttl = 900
 588         r.length = 0xffff
 589         rdata = dns.txt_record()
 590         rdata.txt = '"This is a test"'
 591         r.rdata = rdata
 592         updates.append(r)
 593         p.nscount = len(updates)
 594         p.nsrecs = updates
 595
 596         response = self.dns_transaction_udp(p)
 597         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 598
 599         # Now check the record is around
 600         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 601         questions = []
 602         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 603         questions.append(q)
 604
 605         self.finish_name_packet(p, questions)
 606         response = self.dns_transaction_udp(p)
 607         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 608
 609         # Now delete the record
 610         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 611         updates = []
 612
 613         name = self.get_dns_domain()
 614
 615         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 616         updates.append(u)
 617         self.finish_name_packet(p, updates)
 618
 619         updates = []
 620         r = dns.res_rec()
 621         r.name = NAME
 622         r.rr_type = dns.DNS_QTYPE_TXT
 623         r.rr_class = dns.DNS_QCLASS_NONE
 624         r.ttl = 0
 625         r.length = 0xffff
 626         rdata = dns.txt_record()
 627         rdata.txt = '"This is a test"'
 628         r.rdata = rdata
 629         updates.append(r)
 630         p.nscount = len(updates)
 631         p.nsrecs = updates
 632
 633         response = self.dns_transaction_udp(p)
 634         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 635
 636         # check it's gone
 637         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 638         questions = []
 639
 640         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 641         questions.append(q)
 642
 643         self.finish_name_packet(p, questions)
 644         response = self.dns_transaction_udp(p)
 645         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
 646
 647         # recreate the record
 648         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 649         updates = []
 650
 651         name = self.get_dns_domain()
 652
 653         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 654         updates.append(u)
 655         self.finish_name_packet(p, updates)
 656
 657         updates = []
 658         r = dns.res_rec()
 659         r.name = NAME
 660         r.rr_type = dns.DNS_QTYPE_TXT
 661         r.rr_class = dns.DNS_QCLASS_IN
 662         r.ttl = 900
 663         r.length = 0xffff
 664         rdata = dns.txt_record()
 665         rdata.txt = '"This is a test"'
 666         r.rdata = rdata
 667         updates.append(r)
 668         p.nscount = len(updates)
 669         p.nsrecs = updates
 670
 671         response = self.dns_transaction_udp(p)
 672         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 673
 674         # Now check the record is around
 675         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 676         questions = []
 677         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 678         questions.append(q)
 679
 680         self.finish_name_packet(p, questions)
 681         response = self.dns_transaction_udp(p)
 682         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 683
 684     def test_update_add_mx_record(self):
 685         "test adding MX records works"
 686         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 687         updates = []
 688
 689         name = self.get_dns_domain()
 690
 691         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 692         updates.append(u)
 693         self.finish_name_packet(p, updates)
 694
 695         updates = []
 696         r = dns.res_rec()
 697         r.name = "%s" % self.get_dns_domain()
 698         r.rr_type = dns.DNS_QTYPE_MX
 699         r.rr_class = dns.DNS_QCLASS_IN
 700         r.ttl = 900
 701         r.length = 0xffff
 702         rdata = dns.mx_record()
 703         rdata.preference = 10
 704         rdata.exchange = 'mail.%s' % self.get_dns_domain()
 705         r.rdata = rdata
 706         updates.append(r)
 707         p.nscount = len(updates)
 708         p.nsrecs = updates
 709
 710         response = self.dns_transaction_udp(p)
 711         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 712
 713         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 714         questions = []
 715
 716         name = "%s" % self.get_dns_domain()
 717         q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
 718         questions.append(q)
 719
 720         self.finish_name_packet(p, questions)
 721         response = self.dns_transaction_udp(p)
 722         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 723         self.assertEqual(response.ancount, 1)
 724         ans = response.answers[0]
 725         self.assertEqual(ans.rr_type, dns.DNS_QTYPE_MX)
 726         self.assertEqual(ans.rdata.preference, 10)
 727         self.assertEqual(ans.rdata.exchange, 'mail.%s' % self.get_dns_domain())
 728
 729
 730 class TestComplexQueries(DNSTest):
 731
 732     def setUp(self):
 733         super(TestComplexQueries, self).setUp()
 734         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 735         updates = []
 736
 737         name = self.get_dns_domain()
 738
 739         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 740         updates.append(u)
 741         self.finish_name_packet(p, updates)
 742
 743         updates = []
 744         r = dns.res_rec()
 745         r.name = "cname_test.%s" % self.get_dns_domain()
 746         r.rr_type = dns.DNS_QTYPE_CNAME
 747         r.rr_class = dns.DNS_QCLASS_IN
 748         r.ttl = 900
 749         r.length = 0xffff
 750         r.rdata = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 751         updates.append(r)
 752         p.nscount = len(updates)
 753         p.nsrecs = updates
 754
 755         response = self.dns_transaction_udp(p)
 756         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 757
 758     def tearDown(self):
 759         super(TestComplexQueries, self).tearDown()
 760         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 761         updates = []
 762
 763         name = self.get_dns_domain()
 764
 765         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 766         updates.append(u)
 767         self.finish_name_packet(p, updates)
 768
 769         updates = []
 770         r = dns.res_rec()
 771         r.name = "cname_test.%s" % self.get_dns_domain()
 772         r.rr_type = dns.DNS_QTYPE_CNAME
 773         r.rr_class = dns.DNS_QCLASS_NONE
 774         r.ttl = 0
 775         r.length = 0xffff
 776         r.rdata = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 777         updates.append(r)
 778         p.nscount = len(updates)
 779         p.nsrecs = updates
 780
 781         response = self.dns_transaction_udp(p)
 782         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 783
 784     def test_one_a_query(self):
 785         "create a query packet containing one query record"
 786         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 787         questions = []
 788
 789         name = "cname_test.%s" % self.get_dns_domain()
 790         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 791         print "asking for ", q.name
 792         questions.append(q)
 793
 794         self.finish_name_packet(p, questions)
 795         response = self.dns_transaction_udp(p)
 796         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 797         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 798         self.assertEquals(response.ancount, 2)
 799         self.assertEquals(response.answers[0].rr_type, dns.DNS_QTYPE_CNAME)
 800         self.assertEquals(response.answers[0].rdata, "%s.%s" %
 801                           (os.getenv('SERVER'), self.get_dns_domain()))
 802         self.assertEquals(response.answers[1].rr_type, dns.DNS_QTYPE_A)
 803         self.assertEquals(response.answers[1].rdata,
 804                           os.getenv('SERVER_IP'))
 805
 806 class TestInvalidQueries(DNSTest):
 807
 808     def test_one_a_query(self):
 809         "send 0 bytes follows by create a query packet containing one query record"
 810
 811         s = None
 812         try:
 813             s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
 814             s.connect((os.getenv('SERVER_IP'), 53))
 815             s.send("", 0)
 816         finally:
 817             if s is not None:
 818                 s.close()
 819
 820         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 821         questions = []
 822
 823         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 824         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 825         print "asking for ", q.name
 826         questions.append(q)
 827
 828         self.finish_name_packet(p, questions)
 829         response = self.dns_transaction_udp(p)
 830         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 831         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 832         self.assertEquals(response.ancount, 1)
 833         self.assertEquals(response.answers[0].rdata,
 834                           os.getenv('SERVER_IP'))
 835
 836 if __name__ == "__main__":
 837     import unittest
 838     unittest.main()