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
 128         result = ''
 129         while src:
 130             s, src = src[:length], src[length:]
 131             hexa = ' '.join(["%02X" % ord(x) for x in s])
 132             s = s.translate(FILTER)
 133             result += "%04X   %-*s   %s\n" % (N, length*3, hexa, s)
 134             N += length
 135         return result
 136
 137
 138 class TestSimpleQueries(DNSTest):
 139
 140     def test_one_a_query(self):
 141         "create a query packet containing one query record"
 142         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 143         questions = []
 144
 145         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 146         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 147         print "asking for ", q.name
 148         questions.append(q)
 149
 150         self.finish_name_packet(p, questions)
 151         response = self.dns_transaction_udp(p)
 152         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 153         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 154         self.assertEquals(response.ancount, 1)
 155         self.assertEquals(response.answers[0].rdata,
 156                           os.getenv('SERVER_IP'))
 157
 158     def test_one_a_query_tcp(self):
 159         "create a query packet containing one query record via TCP"
 160         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 161         questions = []
 162
 163         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 164         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 165         print "asking for ", q.name
 166         questions.append(q)
 167
 168         self.finish_name_packet(p, questions)
 169         response = self.dns_transaction_tcp(p)
 170         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 171         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 172         self.assertEquals(response.ancount, 1)
 173         self.assertEquals(response.answers[0].rdata,
 174                           os.getenv('SERVER_IP'))
 175
 176     def test_one_mx_query(self):
 177         "create a query packet causing an empty RCODE_OK answer"
 178         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 179         questions = []
 180
 181         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 182         q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
 183         print "asking for ", q.name
 184         questions.append(q)
 185
 186         self.finish_name_packet(p, questions)
 187         response = self.dns_transaction_udp(p)
 188         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 189         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 190         self.assertEquals(response.ancount, 0)
 191
 192         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 193         questions = []
 194
 195         name = "invalid-%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 196         q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
 197         print "asking for ", q.name
 198         questions.append(q)
 199
 200         self.finish_name_packet(p, questions)
 201         response = self.dns_transaction_udp(p)
 202         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
 203         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 204         self.assertEquals(response.ancount, 0)
 205
 206     def test_two_queries(self):
 207         "create a query packet containing two query records"
 208         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 209         questions = []
 210
 211         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 212         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 213         questions.append(q)
 214
 215         name = "%s.%s" % ('bogusname', self.get_dns_domain())
 216         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 217         questions.append(q)
 218
 219         self.finish_name_packet(p, questions)
 220         response = self.dns_transaction_udp(p)
 221         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 222
 223     def test_qtype_all_query(self):
 224         "create a QTYPE_ALL query"
 225         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 226         questions = []
 227
 228         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 229         q = self.make_name_question(name, dns.DNS_QTYPE_ALL, dns.DNS_QCLASS_IN)
 230         print "asking for ", q.name
 231         questions.append(q)
 232
 233         self.finish_name_packet(p, questions)
 234         response = self.dns_transaction_udp(p)
 235
 236         num_answers = 1
 237         dc_ipv6 = os.getenv('SERVER_IPV6')
 238         if dc_ipv6 is not None:
 239             num_answers += 1
 240
 241         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 242         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 243         self.assertEquals(response.ancount, num_answers)
 244         self.assertEquals(response.answers[0].rdata,
 245                           os.getenv('SERVER_IP'))
 246         if dc_ipv6 is not None:
 247             self.assertEquals(response.answers[1].rdata, dc_ipv6)
 248
 249     def test_qclass_none_query(self):
 250         "create a QCLASS_NONE query"
 251         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 252         questions = []
 253
 254         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 255         q = self.make_name_question(name, dns.DNS_QTYPE_ALL, dns.DNS_QCLASS_NONE)
 256         questions.append(q)
 257
 258         self.finish_name_packet(p, questions)
 259         response = self.dns_transaction_udp(p)
 260         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTIMP)
 261
 262 # Only returns an authority section entry in BIND and Win DNS
 263 # FIXME: Enable one Samba implements this feature
 264 #    def test_soa_hostname_query(self):
 265 #        "create a SOA query for a hostname"
 266 #        p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 267 #        questions = []
 268 #
 269 #        name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 270 #        q = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 271 #        questions.append(q)
 272 #
 273 #        self.finish_name_packet(p, questions)
 274 #        response = self.dns_transaction_udp(p)
 275 #        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 276 #        self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 277 #        # We don't get SOA records for single hosts
 278 #        self.assertEquals(response.ancount, 0)
 279
 280     def test_soa_domain_query(self):
 281         "create a SOA query for a domain"
 282         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 283         questions = []
 284
 285         name = self.get_dns_domain()
 286         q = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 287         questions.append(q)
 288
 289         self.finish_name_packet(p, questions)
 290         response = self.dns_transaction_udp(p)
 291         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 292         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 293         self.assertEquals(response.ancount, 1)
 294         self.assertEquals(response.answers[0].rdata.minimum, 3600)
 295
 296
 297 class TestDNSUpdates(DNSTest):
 298
 299     def test_two_updates(self):
 300         "create two update requests"
 301         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 302         updates = []
 303
 304         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 305         u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 306         updates.append(u)
 307
 308         name = self.get_dns_domain()
 309         u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 310         updates.append(u)
 311
 312         self.finish_name_packet(p, updates)
 313         response = self.dns_transaction_udp(p)
 314         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 315
 316     def test_update_wrong_qclass(self):
 317         "create update with DNS_QCLASS_NONE"
 318         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 319         updates = []
 320
 321         name = self.get_dns_domain()
 322         u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_NONE)
 323         updates.append(u)
 324
 325         self.finish_name_packet(p, updates)
 326         response = self.dns_transaction_udp(p)
 327         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTIMP)
 328
 329     def test_update_prereq_with_non_null_ttl(self):
 330         "test update with a non-null TTL"
 331         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 332         updates = []
 333
 334         name = self.get_dns_domain()
 335
 336         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 337         updates.append(u)
 338         self.finish_name_packet(p, updates)
 339
 340         prereqs = []
 341         r = dns.res_rec()
 342         r.name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 343         r.rr_type = dns.DNS_QTYPE_TXT
 344         r.rr_class = dns.DNS_QCLASS_NONE
 345         r.ttl = 1
 346         r.length = 0
 347         prereqs.append(r)
 348
 349         p.ancount = len(prereqs)
 350         p.answers = prereqs
 351
 352         response = self.dns_transaction_udp(p)
 353         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 354
 355 # I'd love to test this one, but it segfaults. :)
 356 #    def test_update_prereq_with_non_null_length(self):
 357 #        "test update with a non-null length"
 358 #        p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 359 #        updates = []
 360 #
 361 #        name = self.get_dns_domain()
 362 #
 363 #        u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 364 #        updates.append(u)
 365 #        self.finish_name_packet(p, updates)
 366 #
 367 #        prereqs = []
 368 #        r = dns.res_rec()
 369 #        r.name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 370 #        r.rr_type = dns.DNS_QTYPE_TXT
 371 #        r.rr_class = dns.DNS_QCLASS_ANY
 372 #        r.ttl = 0
 373 #        r.length = 1
 374 #        prereqs.append(r)
 375 #
 376 #        p.ancount = len(prereqs)
 377 #        p.answers = prereqs
 378 #
 379 #        response = self.dns_transaction_udp(p)
 380 #        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
 381
 382     def test_update_prereq_nonexisting_name(self):
 383         "test update with a nonexisting name"
 384         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 385         updates = []
 386
 387         name = self.get_dns_domain()
 388
 389         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 390         updates.append(u)
 391         self.finish_name_packet(p, updates)
 392
 393         prereqs = []
 394         r = dns.res_rec()
 395         r.name = "idontexist.%s" % self.get_dns_domain()
 396         r.rr_type = dns.DNS_QTYPE_TXT
 397         r.rr_class = dns.DNS_QCLASS_ANY
 398         r.ttl = 0
 399         r.length = 0
 400         prereqs.append(r)
 401
 402         p.ancount = len(prereqs)
 403         p.answers = prereqs
 404
 405         response = self.dns_transaction_udp(p)
 406         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET)
 407
 408     def test_update_add_txt_record(self):
 409         "test adding records works"
 410         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 411         updates = []
 412
 413         name = self.get_dns_domain()
 414
 415         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 416         updates.append(u)
 417         self.finish_name_packet(p, updates)
 418
 419         updates = []
 420         r = dns.res_rec()
 421         r.name = "textrec.%s" % self.get_dns_domain()
 422         r.rr_type = dns.DNS_QTYPE_TXT
 423         r.rr_class = dns.DNS_QCLASS_IN
 424         r.ttl = 900
 425         r.length = 0xffff
 426         rdata = dns.txt_record()
 427         rdata.txt = '"This is a test"'
 428         r.rdata = rdata
 429         updates.append(r)
 430         p.nscount = len(updates)
 431         p.nsrecs = updates
 432
 433         response = self.dns_transaction_udp(p)
 434         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 435
 436         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 437         questions = []
 438
 439         name = "textrec.%s" % self.get_dns_domain()
 440         q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 441         questions.append(q)
 442
 443         self.finish_name_packet(p, questions)
 444         response = self.dns_transaction_udp(p)
 445         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 446         self.assertEquals(response.ancount, 1)
 447         self.assertEquals(response.answers[0].rdata.txt, '"This is a test"')
 448
 449     def test_update_add_two_txt_records(self):
 450         "test adding two txt records works"
 451         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 452         updates = []
 453
 454         name = self.get_dns_domain()
 455
 456         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 457         updates.append(u)
 458         self.finish_name_packet(p, updates)
 459
 460         updates = []
 461         r = dns.res_rec()
 462         r.name = "textrec2.%s" % self.get_dns_domain()
 463         r.rr_type = dns.DNS_QTYPE_TXT
 464         r.rr_class = dns.DNS_QCLASS_IN
 465         r.ttl = 900
 466         r.length = 0xffff
 467         rdata = dns.txt_record()
 468         rdata.txt = '"This is a test" "and this is a test, too"'
 469         r.rdata = rdata
 470         updates.append(r)
 471         p.nscount = len(updates)
 472         p.nsrecs = updates
 473
 474         response = self.dns_transaction_udp(p)
 475         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 476
 477         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 478         questions = []
 479
 480         name = "textrec2.%s" % self.get_dns_domain()
 481         q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 482         questions.append(q)
 483
 484         self.finish_name_packet(p, questions)
 485         response = self.dns_transaction_udp(p)
 486         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 487         self.assertEquals(response.ancount, 1)
 488         self.assertEquals(response.answers[0].rdata.txt, '"This is a test" "and this is a test, too"')
 489
 490     def test_delete_record(self):
 491         "Test if deleting records works"
 492
 493         NAME = "deleterec.%s" % self.get_dns_domain()
 494
 495         # First, create a record to make sure we have a record to delete.
 496         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 497         updates = []
 498
 499         name = self.get_dns_domain()
 500
 501         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 502         updates.append(u)
 503         self.finish_name_packet(p, updates)
 504
 505         updates = []
 506         r = dns.res_rec()
 507         r.name = NAME
 508         r.rr_type = dns.DNS_QTYPE_TXT
 509         r.rr_class = dns.DNS_QCLASS_IN
 510         r.ttl = 900
 511         r.length = 0xffff
 512         rdata = dns.txt_record()
 513         rdata.txt = '"This is a test"'
 514         r.rdata = rdata
 515         updates.append(r)
 516         p.nscount = len(updates)
 517         p.nsrecs = updates
 518
 519         response = self.dns_transaction_udp(p)
 520         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 521
 522         # Now check the record is around
 523         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 524         questions = []
 525         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 526         questions.append(q)
 527
 528         self.finish_name_packet(p, questions)
 529         response = self.dns_transaction_udp(p)
 530         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 531
 532         # Now delete the record
 533         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 534         updates = []
 535
 536         name = self.get_dns_domain()
 537
 538         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 539         updates.append(u)
 540         self.finish_name_packet(p, updates)
 541
 542         updates = []
 543         r = dns.res_rec()
 544         r.name = NAME
 545         r.rr_type = dns.DNS_QTYPE_TXT
 546         r.rr_class = dns.DNS_QCLASS_NONE
 547         r.ttl = 0
 548         r.length = 0xffff
 549         rdata = dns.txt_record()
 550         rdata.txt = '"This is a test"'
 551         r.rdata = rdata
 552         updates.append(r)
 553         p.nscount = len(updates)
 554         p.nsrecs = updates
 555
 556         response = self.dns_transaction_udp(p)
 557         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 558
 559         # And finally check it's gone
 560         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 561         questions = []
 562
 563         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 564         questions.append(q)
 565
 566         self.finish_name_packet(p, questions)
 567         response = self.dns_transaction_udp(p)
 568         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
 569
 570     def test_readd_record(self):
 571         "Test if adding, deleting and then readding a records works"
 572
 573         NAME = "readdrec.%s" % self.get_dns_domain()
 574
 575         # Create the record
 576         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 577         updates = []
 578
 579         name = self.get_dns_domain()
 580
 581         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 582         updates.append(u)
 583         self.finish_name_packet(p, updates)
 584
 585         updates = []
 586         r = dns.res_rec()
 587         r.name = NAME
 588         r.rr_type = dns.DNS_QTYPE_TXT
 589         r.rr_class = dns.DNS_QCLASS_IN
 590         r.ttl = 900
 591         r.length = 0xffff
 592         rdata = dns.txt_record()
 593         rdata.txt = '"This is a test"'
 594         r.rdata = rdata
 595         updates.append(r)
 596         p.nscount = len(updates)
 597         p.nsrecs = updates
 598
 599         response = self.dns_transaction_udp(p)
 600         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 601
 602         # Now check the record is around
 603         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 604         questions = []
 605         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 606         questions.append(q)
 607
 608         self.finish_name_packet(p, questions)
 609         response = self.dns_transaction_udp(p)
 610         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 611
 612         # Now delete the record
 613         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 614         updates = []
 615
 616         name = self.get_dns_domain()
 617
 618         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 619         updates.append(u)
 620         self.finish_name_packet(p, updates)
 621
 622         updates = []
 623         r = dns.res_rec()
 624         r.name = NAME
 625         r.rr_type = dns.DNS_QTYPE_TXT
 626         r.rr_class = dns.DNS_QCLASS_NONE
 627         r.ttl = 0
 628         r.length = 0xffff
 629         rdata = dns.txt_record()
 630         rdata.txt = '"This is a test"'
 631         r.rdata = rdata
 632         updates.append(r)
 633         p.nscount = len(updates)
 634         p.nsrecs = updates
 635
 636         response = self.dns_transaction_udp(p)
 637         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 638
 639         # check it's gone
 640         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 641         questions = []
 642
 643         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 644         questions.append(q)
 645
 646         self.finish_name_packet(p, questions)
 647         response = self.dns_transaction_udp(p)
 648         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
 649
 650         # recreate the record
 651         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 652         updates = []
 653
 654         name = self.get_dns_domain()
 655
 656         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 657         updates.append(u)
 658         self.finish_name_packet(p, updates)
 659
 660         updates = []
 661         r = dns.res_rec()
 662         r.name = NAME
 663         r.rr_type = dns.DNS_QTYPE_TXT
 664         r.rr_class = dns.DNS_QCLASS_IN
 665         r.ttl = 900
 666         r.length = 0xffff
 667         rdata = dns.txt_record()
 668         rdata.txt = '"This is a test"'
 669         r.rdata = rdata
 670         updates.append(r)
 671         p.nscount = len(updates)
 672         p.nsrecs = updates
 673
 674         response = self.dns_transaction_udp(p)
 675         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 676
 677         # Now check the record is around
 678         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 679         questions = []
 680         q = self.make_name_question(NAME, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
 681         questions.append(q)
 682
 683         self.finish_name_packet(p, questions)
 684         response = self.dns_transaction_udp(p)
 685         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 686
 687     def test_update_add_mx_record(self):
 688         "test adding MX records works"
 689         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 690         updates = []
 691
 692         name = self.get_dns_domain()
 693
 694         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 695         updates.append(u)
 696         self.finish_name_packet(p, updates)
 697
 698         updates = []
 699         r = dns.res_rec()
 700         r.name = "%s" % self.get_dns_domain()
 701         r.rr_type = dns.DNS_QTYPE_MX
 702         r.rr_class = dns.DNS_QCLASS_IN
 703         r.ttl = 900
 704         r.length = 0xffff
 705         rdata = dns.mx_record()
 706         rdata.preference = 10
 707         rdata.exchange = 'mail.%s' % self.get_dns_domain()
 708         r.rdata = rdata
 709         updates.append(r)
 710         p.nscount = len(updates)
 711         p.nsrecs = updates
 712
 713         response = self.dns_transaction_udp(p)
 714         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 715
 716         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 717         questions = []
 718
 719         name = "%s" % self.get_dns_domain()
 720         q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
 721         questions.append(q)
 722
 723         self.finish_name_packet(p, questions)
 724         response = self.dns_transaction_udp(p)
 725         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 726         self.assertEqual(response.ancount, 1)
 727         ans = response.answers[0]
 728         self.assertEqual(ans.rr_type, dns.DNS_QTYPE_MX)
 729         self.assertEqual(ans.rdata.preference, 10)
 730         self.assertEqual(ans.rdata.exchange, 'mail.%s' % self.get_dns_domain())
 731
 732
 733 class TestComplexQueries(DNSTest):
 734
 735     def setUp(self):
 736         super(TestComplexQueries, self).setUp()
 737         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 738         updates = []
 739
 740         name = self.get_dns_domain()
 741
 742         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 743         updates.append(u)
 744         self.finish_name_packet(p, updates)
 745
 746         updates = []
 747         r = dns.res_rec()
 748         r.name = "cname_test.%s" % self.get_dns_domain()
 749         r.rr_type = dns.DNS_QTYPE_CNAME
 750         r.rr_class = dns.DNS_QCLASS_IN
 751         r.ttl = 900
 752         r.length = 0xffff
 753         r.rdata = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 754         updates.append(r)
 755         p.nscount = len(updates)
 756         p.nsrecs = updates
 757
 758         response = self.dns_transaction_udp(p)
 759         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 760
 761     def tearDown(self):
 762         super(TestComplexQueries, self).tearDown()
 763         p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
 764         updates = []
 765
 766         name = self.get_dns_domain()
 767
 768         u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
 769         updates.append(u)
 770         self.finish_name_packet(p, updates)
 771
 772         updates = []
 773         r = dns.res_rec()
 774         r.name = "cname_test.%s" % self.get_dns_domain()
 775         r.rr_type = dns.DNS_QTYPE_CNAME
 776         r.rr_class = dns.DNS_QCLASS_NONE
 777         r.ttl = 0
 778         r.length = 0xffff
 779         r.rdata = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 780         updates.append(r)
 781         p.nscount = len(updates)
 782         p.nsrecs = updates
 783
 784         response = self.dns_transaction_udp(p)
 785         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 786
 787     def test_one_a_query(self):
 788         "create a query packet containing one query record"
 789         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 790         questions = []
 791
 792         name = "cname_test.%s" % self.get_dns_domain()
 793         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 794         print "asking for ", q.name
 795         questions.append(q)
 796
 797         self.finish_name_packet(p, questions)
 798         response = self.dns_transaction_udp(p)
 799         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 800         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 801         self.assertEquals(response.ancount, 2)
 802         self.assertEquals(response.answers[0].rr_type, dns.DNS_QTYPE_CNAME)
 803         self.assertEquals(response.answers[0].rdata, "%s.%s" %
 804                           (os.getenv('SERVER'), self.get_dns_domain()))
 805         self.assertEquals(response.answers[1].rr_type, dns.DNS_QTYPE_A)
 806         self.assertEquals(response.answers[1].rdata,
 807                           os.getenv('SERVER_IP'))
 808
 809 class TestInvalidQueries(DNSTest):
 810
 811     def test_one_a_query(self):
 812         "send 0 bytes follows by create a query packet containing one query record"
 813
 814         s = None
 815         try:
 816             s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
 817             s.connect((os.getenv('SERVER_IP'), 53))
 818             s.send("", 0)
 819         finally:
 820             if s is not None:
 821                 s.close()
 822
 823         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 824         questions = []
 825
 826         name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
 827         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 828         print "asking for ", q.name
 829         questions.append(q)
 830
 831         self.finish_name_packet(p, questions)
 832         response = self.dns_transaction_udp(p)
 833         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 834         self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
 835         self.assertEquals(response.ancount, 1)
 836         self.assertEquals(response.answers[0].rdata,
 837                           os.getenv('SERVER_IP'))
 838
 839     def test_one_a_reply(self):
 840         "send a reply instead of a query"
 841
 842         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
 843         questions = []
 844
 845         name = "%s.%s" % ('fakefakefake', self.get_dns_domain())
 846         q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
 847         print "asking for ", q.name
 848         questions.append(q)
 849
 850         self.finish_name_packet(p, questions)
 851         p.operation |= dns.DNS_FLAG_REPLY
 852         s = None
 853         try:
 854             send_packet = ndr.ndr_pack(p)
 855             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
 856             host=os.getenv('SERVER_IP')
 857             s.connect((host, 53))
 858             tcp_packet = struct.pack('!H', len(send_packet))
 859             tcp_packet += send_packet
 860             s.send(tcp_packet, 0)
 861             recv_packet = s.recv(0xffff + 2, 0)
 862             self.assertEquals(0, len(recv_packet))
 863         finally:
 864             if s is not None:
 865                 s.close()
 866
 867
 868 if __name__ == "__main__":
 869     import unittest
 870     unittest.main()