diff options
Diffstat (limited to 'python')
| -rw-r--r-- | python/samba/tests/dns.py | 271 |
1 files changed, 165 insertions, 106 deletions
diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index e153a2b1b94..784485e9b38 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -146,6 +146,47 @@ class DNSTest(TestCase): 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): @@ -415,44 +456,6 @@ class TestDNSUpdates(DNSTest): response = self.dns_transaction_udp(p) self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET) - 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 test_update_add_txt_record(self): "test adding records works" prefix, txt = 'textrec', ['"This is a test"'] @@ -461,74 +464,6 @@ class TestDNSUpdates(DNSTest): self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK) self.check_query_txt(prefix, txt) - 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) - - 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) - - 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) - - # 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']) - - 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']) - - 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) - - 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) - - 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) - - 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) - def test_delete_record(self): "Test if deleting records works" @@ -901,6 +836,130 @@ class TestInvalidQueries(DNSTest): if s is not None: s.close() +class TestRPCRoundtrip(DNSTest): + def get_credentials(self, lp): + creds = credentials.Credentials() + creds.guess(lp) + creds.set_machine_account(lp) + creds.set_krb_forwardable(credentials.NO_KRB_FORWARDABLE) + return creds + + def setUp(self): + super(TestRPCRoundtrip, self).setUp() + self.lp = self.get_loadparm() + self.creds = self.get_credentials(self.lp) + self.server = os.getenv("SERVER_IP") + self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server), + self.lp, self.creds) + + def tearDown(self): + super(TestRPCRoundtrip, self).tearDown() + + 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, + 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, + 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, + self.get_dns_domain(), + "%s.%s" % (prefix, self.get_dns_domain()), + dnsp.DNS_TYPE_TXT, '"" "" "\\"This is a test\\""')) + + # 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, + 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, + self.get_dns_domain(), + "%s.%s" % (prefix, self.get_dns_domain()), + dnsp.DNS_TYPE_TXT, '"NULL" "NULL"')) + + 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, + self.get_dns_domain(), + "%s.%s" % (prefix, self.get_dns_domain()), + dnsp.DNS_TYPE_TXT, '"HIGH\xFFBYTE"')) + + 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, + self.get_dns_domain(), + "%s.%s" % (prefix, self.get_dns_domain()), + dnsp.DNS_TYPE_TXT, '"Th\\\\=is=is a test"')) + + 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, + 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_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, + self.get_dns_domain(), + "%s.%s" % (prefix, self.get_dns_domain()), + dnsp.DNS_TYPE_TXT, '')) + if __name__ == "__main__": import unittest unittest.main() |