mirror of
https://github.com/samba-team/samba.git
synced 2024-12-25 23:21:54 +03:00
CVE-2016-0771: tests/dns: modify tests to check via RPC
This checks that TXT records added over DNS, look the same over RPC. BUG: https://bugzilla.samba.org/show_bug.cgi?id=11128 BUG: https://bugzilla.samba.org/show_bug.cgi?id=11686 Signed-off-by: Garming Sam <garming@catalyst.net.nz> Reviewed-by: Stefan Metzmacher <metze@samba.org> Reviewed-by: Andrew Bartlett <abartlet@samba.org>
This commit is contained in:
Garming Sam
Stefan Metzmacher
parent
2a796e5de7
commit
ffec494826
@ -142,6 +142,44 @@ class DNSTest(TestCase):
|
||||
if s is not None:
|
||||
s.close()
|
||||
|
||||
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)
|
||||
|
||||
class TestSimpleQueries(DNSTest):
|
||||
|
||||
def test_one_a_query(self):
|
||||
@ -439,44 +477,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"']
|
||||
@ -485,74 +485,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"
|
||||
|
||||
@ -1012,7 +944,129 @@ class TestZones(DNSTest):
|
||||
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
|
||||
self.assertEquals(response.ancount, 0)
|
||||
|
||||
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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user