#
import os
+import sys
import struct
import random
+
+sys.path.insert(0, "bin/python")
+import samba
+samba.ensure_external_module("testtools", "testtools")
+samba.ensure_external_module("subunit", "subunit/python")
+from subunit.run import SubunitTestRunner
+import unittest
+
from samba import socket
import samba.ndr as ndr
-import samba.dcerpc.dns as dns
+from samba import credentials, param
from samba.tests import TestCase
+from samba.dcerpc import dns, dnsp, dnsserver
+from samba.netcmd.dns import TXTRecord, dns_record_match, data_to_dns_record
+import samba.getopt as options
+import optparse
+
+parser = optparse.OptionParser("dns.py <server name> <server ip> [options]")
+sambaopts = options.SambaOptions(parser)
+parser.add_option_group(sambaopts)
FILTER=''.join([(len(repr(chr(x)))==3) and chr(x) or '.' for x in range(256)])
+# use command line creds if available
+credopts = options.CredentialsOptions(parser)
+parser.add_option_group(credopts)
+
+opts, args = parser.parse_args()
+
+lp = sambaopts.get_loadparm()
+creds = credopts.get_credentials(lp)
+
+if len(args) < 2:
+ parser.print_usage()
+ sys.exit(1)
+
+server_name = args[0]
+server_ip = args[1]
+creds.set_krb_forwardable(credentials.NO_KRB_FORWARDABLE)
+
+def make_txt_record(records):
+ rdata_txt = dns.txt_record()
+ s_list = dnsp.string_list()
+ s_list.count = len(records)
+ s_list.str = records
+ rdata_txt.txt = s_list
+ return rdata_txt
class DNSTest(TestCase):
+ def setUp(self):
+ global server, server_ip, lp, creds
+ super(DNSTest, self).setUp()
+ self.server = server_name
+ self.server_ip = server_ip
+ self.lp = lp
+ self.creds = creds
+
def errstr(self, errcode):
"Return a readable error code"
string_codes = [
def get_dns_domain(self):
"Helper to get dns domain"
- return os.getenv('REALM', 'example.com').lower()
+ return self.creds.get_realm().lower()
- def dns_transaction_udp(self, packet, host=os.getenv('SERVER_IP'), dump=False):
+ def dns_transaction_udp(self, packet, host=server_ip,
+ dump=False):
"send a DNS query and read the reply"
s = None
try:
if s is not None:
s.close()
- def dns_transaction_tcp(self, packet, host=os.getenv('SERVER_IP'), dump=False):
+ def dns_transaction_tcp(self, packet, host=server_ip,
+ dump=False):
"send a DNS query and read the reply"
s = None
try:
N+=length
return result
+ def make_txt_update(self, prefix, txt_array):
+ p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
+ updates = []
+
+ name = self.get_dns_domain()
+ u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
+ updates.append(u)
+ self.finish_name_packet(p, updates)
+
+ updates = []
+ r = dns.res_rec()
+ r.name = "%s.%s" % (prefix, self.get_dns_domain())
+ r.rr_type = dns.DNS_QTYPE_TXT
+ r.rr_class = dns.DNS_QCLASS_IN
+ r.ttl = 900
+ r.length = 0xffff
+ rdata = make_txt_record(txt_array)
+ r.rdata = rdata
+ updates.append(r)
+ p.nscount = len(updates)
+ p.nsrecs = updates
+
+ return p
+
+ def check_query_txt(self, prefix, txt_array):
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+ p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
+ questions = []
+
+ q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
+ questions.append(q)
+
+ self.finish_name_packet(p, questions)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.assertEquals(response.ancount, 1)
+ self.assertEquals(response.answers[0].rdata.txt.str, txt_array)
+
+ def assertIsNotNone(self, item):
+ self.assertTrue(item is not None)
+
class TestSimpleQueries(DNSTest):
def test_one_a_query(self):
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
- name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
print "asking for ", q.name
questions.append(q)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rdata,
- os.getenv('SERVER_IP'))
+ self.server_ip)
def test_one_a_query_tcp(self):
"create a query packet containing one query record via TCP"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
- name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
print "asking for ", q.name
questions.append(q)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rdata,
- os.getenv('SERVER_IP'))
+ self.server_ip)
+
+ def test_one_mx_query(self):
+ "create a query packet causing an empty RCODE_OK answer"
+ p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
+ questions = []
+
+ name = "%s.%s" % (self.server, self.get_dns_domain())
+ q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
+ print "asking for ", q.name
+ questions.append(q)
+
+ self.finish_name_packet(p, questions)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
+ self.assertEquals(response.ancount, 0)
+
+ p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
+ questions = []
+
+ name = "invalid-%s.%s" % (self.server, self.get_dns_domain())
+ q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
+ print "asking for ", q.name
+ questions.append(q)
+
+ self.finish_name_packet(p, questions)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
+ self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
+ self.assertEquals(response.ancount, 0)
def test_two_queries(self):
"create a query packet containing two query records"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
- name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
questions.append(q)
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
- name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_ALL, dns.DNS_QCLASS_IN)
print "asking for ", q.name
questions.append(q)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, num_answers)
self.assertEquals(response.answers[0].rdata,
- os.getenv('SERVER_IP'))
+ self.server_ip)
if dc_ipv6 is not None:
self.assertEquals(response.answers[1].rdata, dc_ipv6)
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
- name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_ALL, dns.DNS_QCLASS_NONE)
questions.append(q)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
+ self.assertEquals(response.answers[0].rdata.minimum, 3600)
class TestDNSUpdates(DNSTest):
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
updates = []
- name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ name = "%s.%s" % (self.server, self.get_dns_domain())
u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
updates.append(u)
prereqs = []
r = dns.res_rec()
- r.name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ r.name = "%s.%s" % (self.server, self.get_dns_domain())
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 1
response = self.dns_transaction_udp(p)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
-# I'd love to test this one, but it segfaults. :)
-# def test_update_prereq_with_non_null_length(self):
-# "test update with a non-null length"
-# p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
-# updates = []
-#
-# name = self.get_dns_domain()
-#
-# u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
-# updates.append(u)
-# self.finish_name_packet(p, updates)
-#
-# prereqs = []
-# r = dns.res_rec()
-# r.name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
-# r.rr_type = dns.DNS_QTYPE_TXT
-# r.rr_class = dns.DNS_QCLASS_ANY
-# r.ttl = 0
-# r.length = 1
-# prereqs.append(r)
-#
-# p.ancount = len(prereqs)
-# p.answers = prereqs
-#
-# response = self.dns_transaction_udp(p)
-# self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR)
-
- def test_update_prereq_nonexisting_name(self):
- "test update with a nonexisting name"
+ def test_update_prereq_with_non_null_length(self):
+ "test update with a non-null length"
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
updates = []
prereqs = []
r = dns.res_rec()
- r.name = "idontexist.%s" % self.get_dns_domain()
+ r.name = "%s.%s" % (self.server, self.get_dns_domain())
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_ANY
r.ttl = 0
- r.length = 0
+ r.length = 1
prereqs.append(r)
p.ancount = len(prereqs)
response = self.dns_transaction_udp(p)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET)
- def test_update_add_txt_record(self):
- "test adding records works"
+ def test_update_prereq_nonexisting_name(self):
+ "test update with a nonexisting name"
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
updates = []
updates.append(u)
self.finish_name_packet(p, updates)
- updates = []
+ prereqs = []
r = dns.res_rec()
- r.name = "textrec.%s" % self.get_dns_domain()
+ r.name = "idontexist.%s" % self.get_dns_domain()
r.rr_type = dns.DNS_QTYPE_TXT
- r.rr_class = dns.DNS_QCLASS_IN
- r.ttl = 900
- r.length = 0xffff
- rdata = dns.txt_record()
- rdata.txt = '"This is a test"'
- r.rdata = rdata
- updates.append(r)
- p.nscount = len(updates)
- p.nsrecs = updates
-
- response = self.dns_transaction_udp(p)
- self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
-
- p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
- questions = []
-
- name = "textrec.%s" % self.get_dns_domain()
- q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
- questions.append(q)
-
- self.finish_name_packet(p, questions)
- response = self.dns_transaction_udp(p)
- self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
- self.assertEquals(response.ancount, 1)
- self.assertEquals(response.answers[0].rdata.txt, '"This is a test"')
-
- def test_update_add_two_txt_records(self):
- "test adding two txt records works"
- p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
- updates = []
-
- name = self.get_dns_domain()
-
- u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
- updates.append(u)
- self.finish_name_packet(p, updates)
+ r.rr_class = dns.DNS_QCLASS_ANY
+ r.ttl = 0
+ r.length = 0
+ prereqs.append(r)
- updates = []
- r = dns.res_rec()
- r.name = "textrec2.%s" % self.get_dns_domain()
- r.rr_type = dns.DNS_QTYPE_TXT
- r.rr_class = dns.DNS_QCLASS_IN
- r.ttl = 900
- r.length = 0xffff
- rdata = dns.txt_record()
- rdata.txt = '"This is a test" "and this is a test, too"'
- r.rdata = rdata
- updates.append(r)
- p.nscount = len(updates)
- p.nsrecs = updates
+ p.ancount = len(prereqs)
+ p.answers = prereqs
response = self.dns_transaction_udp(p)
- self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
-
- p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
- questions = []
-
- name = "textrec2.%s" % self.get_dns_domain()
- q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
- questions.append(q)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET)
- self.finish_name_packet(p, questions)
+ def test_update_add_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'textrec', ['"This is a test"']
+ p = self.make_txt_update(prefix, txt)
response = self.dns_transaction_udp(p)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
- self.assertEquals(response.ancount, 1)
- self.assertEquals(response.answers[0].rdata.txt, '"This is a test" "and this is a test, too"')
+ self.check_query_txt(prefix, txt)
def test_delete_record(self):
"Test if deleting records works"
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
- rdata = dns.txt_record()
- rdata.txt = '"This is a test"'
+ rdata = make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 0
r.length = 0xffff
- rdata = dns.txt_record()
- rdata.txt = '"This is a test"'
+ rdata = make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
- rdata = dns.txt_record()
- rdata.txt = '"This is a test"'
+ rdata = make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 0
r.length = 0xffff
- rdata = dns.txt_record()
- rdata.txt = '"This is a test"'
+ rdata = make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
- rdata = dns.txt_record()
- rdata.txt = '"This is a test"'
+ rdata = make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
- r.rdata = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ r.rdata = "%s.%s" % (self.server, self.get_dns_domain())
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 0
r.length = 0xffff
- r.rdata = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ r.rdata = "%s.%s" % (self.server, self.get_dns_domain())
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
self.assertEquals(response.ancount, 2)
self.assertEquals(response.answers[0].rr_type, dns.DNS_QTYPE_CNAME)
self.assertEquals(response.answers[0].rdata, "%s.%s" %
- (os.getenv('SERVER'), self.get_dns_domain()))
+ (self.server, self.get_dns_domain()))
self.assertEquals(response.answers[1].rr_type, dns.DNS_QTYPE_A)
self.assertEquals(response.answers[1].rdata,
- os.getenv('SERVER_IP'))
+ self.server_ip)
class TestInvalidQueries(DNSTest):
s = None
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
- s.connect((os.getenv('SERVER_IP'), 53))
+ s.connect((self.server_ip, 53))
s.send("", 0)
finally:
if s is not None:
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
- name = "%s.%s" % (os.getenv('SERVER'), self.get_dns_domain())
+ name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
print "asking for ", q.name
questions.append(q)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rdata,
- os.getenv('SERVER_IP'))
+ self.server_ip)
+
+ def test_one_a_reply(self):
+ "send a reply instead of a query"
+
+ p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
+ questions = []
+
+ name = "%s.%s" % ('fakefakefake', self.get_dns_domain())
+ q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
+ print "asking for ", q.name
+ questions.append(q)
+
+ self.finish_name_packet(p, questions)
+ p.operation |= dns.DNS_FLAG_REPLY
+ s = None
+ try:
+ send_packet = ndr.ndr_pack(p)
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
+ host=self.server_ip
+ s.connect((host, 53))
+ tcp_packet = struct.pack('!H', len(send_packet))
+ tcp_packet += send_packet
+ s.send(tcp_packet, 0)
+ recv_packet = s.recv(0xffff + 2, 0)
+ self.assertEquals(0, len(recv_packet))
+ finally:
+ if s is not None:
+ s.close()
+
+class TestRPCRoundtrip(DNSTest):
+ def setUp(self):
+ super(TestRPCRoundtrip, self).setUp()
+ self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server_ip),
+ self.lp, self.creds)
+
+ def tearDown(self):
+ super(TestRPCRoundtrip, self).tearDown()
+
+ def test_update_add_txt_rpc_to_dns(self):
+ prefix, txt = 'rpctextrec', ['"This is a test"']
+
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"\\"This is a test\\""')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ def test_update_add_null_padded_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'pad1textrec', ['"This is a test"', '', '']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"\\"This is a test\\"" "" ""'))
+
+ prefix, txt = 'pad2textrec', ['"This is a test"', '', '', 'more text']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"\\"This is a test\\"" "" "" "more text"'))
+
+ prefix, txt = 'pad3textrec', ['', '', '"This is a test"']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"" "" "\\"This is a test\\""'))
+
+ def test_update_add_padding_rpc_to_dns(self):
+ prefix, txt = 'pad1textrec', ['"This is a test"', '', '']
+ prefix = 'rpc' + prefix
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"\\"This is a test\\"" "" ""')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ prefix, txt = 'pad2textrec', ['"This is a test"', '', '', 'more text']
+ prefix = 'rpc' + prefix
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"\\"This is a test\\"" "" "" "more text"')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ prefix, txt = 'pad3textrec', ['', '', '"This is a test"']
+ prefix = 'rpc' + prefix
+ name = "%s.%s" % (prefix, self.get_dns_domain())
-if __name__ == "__main__":
- import unittest
- unittest.main()
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"" "" "\\"This is a test\\""')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ # Test is incomplete due to strlen against txt records
+ def test_update_add_null_char_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'nulltextrec', ['NULL\x00BYTE']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, ['NULL'])
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"NULL"'))
+
+ prefix, txt = 'nulltextrec2', ['NULL\x00BYTE', 'NULL\x00BYTE']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, ['NULL', 'NULL'])
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"NULL" "NULL"'))
+
+ def test_update_add_null_char_rpc_to_dns(self):
+ prefix, txt = 'nulltextrec', ['NULL\x00BYTE']
+ prefix = 'rpc' + prefix
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"NULL"')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, ['NULL'])
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ def test_update_add_hex_char_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'hextextrec', ['HIGH\xFFBYTE']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"HIGH\xFFBYTE"'))
+
+ def test_update_add_hex_rpc_to_dns(self):
+ prefix, txt = 'hextextrec', ['HIGH\xFFBYTE']
+ prefix = 'rpc' + prefix
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"HIGH\xFFBYTE"')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ def test_update_add_slash_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'slashtextrec', ['Th\\=is=is a test']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"Th\\\\=is=is a test"'))
+
+ # This test fails against Windows as it eliminates slashes in RPC
+ # One typical use for a slash is in records like 'var=value' to
+ # escape '=' characters.
+ def test_update_add_slash_rpc_to_dns(self):
+ prefix, txt = 'slashtextrec', ['Th\\=is=is a test']
+ prefix = 'rpc' + prefix
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"Th\\\\=is=is a test"')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ def test_update_add_two_txt_records(self):
+ "test adding two txt records works"
+ prefix, txt = 'textrec2', ['"This is a test"',
+ '"and this is a test, too"']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"\\"This is a test\\""' +
+ ' "\\"and this is a test, too\\""'))
+
+ def test_update_add_two_rpc_to_dns(self):
+ prefix, txt = 'textrec2', ['"This is a test"',
+ '"and this is a test, too"']
+ prefix = 'rpc' + prefix
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT,
+ '"\\"This is a test\\""' +
+ ' "\\"and this is a test, too\\""')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+ def test_update_add_empty_txt_records(self):
+ "test adding two txt records works"
+ prefix, txt = 'emptytextrec', []
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, ''))
+
+ def test_update_add_empty_rpc_to_dns(self):
+ prefix, txt = 'rpcemptytextrec', []
+
+ name = "%s.%s" % (prefix, self.get_dns_domain())
+
+ rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '')
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+ try:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, add_rec_buf, None)
+
+ self.check_query_txt(prefix, txt)
+ finally:
+ self.rpc_conn.DnssrvUpdateRecord2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
+ 0, self.server_ip, self.get_dns_domain(),
+ name, None, add_rec_buf)
+
+runner = SubunitTestRunner()
+rc = 0
+if not runner.run(unittest.makeSuite(DNSTest)).wasSuccessful():
+ rc = 1
+if not runner.run(unittest.makeSuite(TestSimpleQueries)).wasSuccessful():
+ rc = 1
+if not runner.run(unittest.makeSuite(TestDNSUpdates)).wasSuccessful():
+ rc = 1
+if not runner.run(unittest.makeSuite(TestComplexQueries)).wasSuccessful():
+ rc = 1
+if not runner.run(unittest.makeSuite(TestInvalidQueries)).wasSuccessful():
+ rc = 1
+if not runner.run(unittest.makeSuite(TestRPCRoundtrip)).wasSuccessful():
+ rc = 1
+
+sys.exit(rc)
git.samba.org / samba.git / blobdiff