From 529a4ad9c456a4963daa402e15964797b6c4246d Mon Sep 17 00:00:00 2001 From: Stefan Metzmacher Date: Fri, 7 Aug 2015 11:36:47 +0200 Subject: [PATCH 01/18] CVE-2016-0771: s4:librpc: python_dns and python_dcerpc_dnsp doesn't require client bindings BUG: https://bugzilla.samba.org/show_bug.cgi?id=11686 BUG: https://bugzilla.samba.org/show_bug.cgi?id=11128 Signed-off-by: Stefan Metzmacher Reviewed-by: Garming Sam --- librpc/idl/dns.idl | 2 +- librpc/idl/dnsp.idl | 4 ++-- source4/librpc/wscript_build | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/librpc/idl/dns.idl b/librpc/idl/dns.idl index d247e0e..1183bd1 100644 --- a/librpc/idl/dns.idl +++ b/librpc/idl/dns.idl @@ -270,7 +270,7 @@ interface dns /* this is a convenience hook for ndrdump */ - void decode_dns_name_packet( + [nopython] void decode_dns_name_packet( [in] dns_name_packet packet ); } diff --git a/librpc/idl/dnsp.idl b/librpc/idl/dnsp.idl index 4c49001..d705cfc 100644 --- a/librpc/idl/dnsp.idl +++ b/librpc/idl/dnsp.idl @@ -263,11 +263,11 @@ interface dnsp /* these are convenience hooks for ndrdump */ - void decode_DnssrvRpcRecord( + [nopython] void decode_DnssrvRpcRecord( [in] dnsp_DnssrvRpcRecord blob ); - void decode_DnsProperty( + [nopython] void decode_DnsProperty( [in] dnsp_DnsProperty blob ); } diff --git a/source4/librpc/wscript_build b/source4/librpc/wscript_build index 781611e..4da3eaa 100755 --- a/source4/librpc/wscript_build +++ b/source4/librpc/wscript_build @@ -167,7 +167,7 @@ bld.SAMBA_PYTHON('python_echo', bld.SAMBA_PYTHON('python_dns', source='../../librpc/gen_ndr/py_dns.c', - deps='RPC_NDR_DNS pytalloc-util pyrpc_util', + deps='NDR_DNS pytalloc-util pyrpc_util', realname='samba/dcerpc/dns.so' ) @@ -324,7 +324,7 @@ bld.SAMBA_PYTHON('python_dcerpc_drsblobs', bld.SAMBA_PYTHON('python_dcerpc_dnsp', source='../../librpc/gen_ndr/py_dnsp.c', - deps='pytalloc-util pyrpc_util NDR_SECURITY RPC_NDR_DNSP', + deps='pytalloc-util pyrpc_util NDR_SECURITY NDR_DNSP', realname='samba/dcerpc/dnsp.so' ) -- 1.7.0.4 From 6ad4427120efd7e09c346e2d292431e5b0ea0fa1 Mon Sep 17 00:00:00 2001 From: Stefan Metzmacher Date: Fri, 7 Aug 2015 11:36:47 +0200 Subject: [PATCH 02/18] CVE-2016-0771: librpc: add RPC_NDR_DNSSERVER to dcerpc-samba library RPC_NDR_DNSSERVER is the client interface NDR_DNSP contains just marshalling helpers. BUG: https://bugzilla.samba.org/show_bug.cgi?id=11128 BUG: https://bugzilla.samba.org/show_bug.cgi?id=11686 Signed-off-by: Stefan Metzmacher Reviewed-by: Garming Sam --- librpc/wscript_build | 18 ++++-------------- 1 files changed, 4 insertions(+), 14 deletions(-) diff --git a/librpc/wscript_build b/librpc/wscript_build index 1594e72..244da21 100644 --- a/librpc/wscript_build +++ b/librpc/wscript_build @@ -27,7 +27,7 @@ bld.SAMBA_SUBSYSTEM('NDR_NAMED_PIPE_AUTH', bld.SAMBA_SUBSYSTEM('NDR_DNSSERVER', source='gen_ndr/ndr_dnsserver.c ndr/ndr_dnsserver.c', - public_deps='ndr' + public_deps='ndr NDR_DNSP' ) bld.SAMBA_SUBSYSTEM('NDR_DNS', @@ -371,7 +371,7 @@ bld.SAMBA_LIBRARY('ndr-standard', pc_files='ndr_standard.pc', deps='''NDR_SECURITY NDR_LSA NDR_SAMR NDR_NETLOGON NDR_EVENTLOG NDR_DFS NDR_NTSVCS NDR_SVCCTL NDR_INITSHUTDOWN NDR_WKSSVC NDR_SRVSVC NDR_WINREG - NDR_ECHO security NDR_DNS NDR_ATSVC NDR_SPOOLSS NDR_DSSETUP + NDR_ECHO security NDR_DNS NDR_DNSP NDR_ATSVC NDR_SPOOLSS NDR_DSSETUP NDR_SERVER_ID NDR_NOTIFY''', public_deps='ndr', public_headers='gen_ndr/samr.h gen_ndr/ndr_samr.h gen_ndr/lsa.h gen_ndr/netlogon.h gen_ndr/atsvc.h gen_ndr/ndr_atsvc.h gen_ndr/ndr_svcctl.h gen_ndr/svcctl.h', @@ -453,11 +453,6 @@ bld.SAMBA_SUBSYSTEM('RPC_NDR_AUDIOSRV', public_deps='NDR_AUDIOSRV dcerpc-binding' ) -bld.SAMBA_SUBSYSTEM('RPC_NDR_DNS', - source='gen_ndr/ndr_dns_c.c', - public_deps='dcerpc-binding NDR_DNS' - ) - bld.SAMBA_SUBSYSTEM('RPC_NDR_ECHO', source='gen_ndr/ndr_echo_c.c', public_deps='dcerpc-binding NDR_ECHO' @@ -640,11 +635,6 @@ bld.SAMBA_SUBSYSTEM('RPC_NDR_BACKUPKEY', public_deps='dcerpc-binding NDR_BACKUPKEY' ) -bld.SAMBA_SUBSYSTEM('RPC_NDR_DNSP', - source='gen_ndr/ndr_dnsp_c.c', - public_deps='dcerpc-binding NDR_DNSP' - ) - bld.SAMBA_SUBSYSTEM('RPC_NDR_DNSSERVER', source='gen_ndr/ndr_dnsserver_c.c', public_deps='dcerpc-binding ndr-standard' @@ -679,7 +669,7 @@ bld.SAMBA_SUBSYSTEM('RPC_NDR_MDSSVC', bld.SAMBA_LIBRARY('ndr-samba', source=[], deps='''NDR_DRSBLOBS NDR_DRSUAPI NDR_IDMAP NDR_NTLMSSP NDR_SCHANNEL NDR_MGMT - NDR_DNSP NDR_EPMAPPER NDR_XATTR NDR_UNIXINFO NDR_NAMED_PIPE_AUTH NDR_DCOM + NDR_DNSSERVER NDR_EPMAPPER NDR_XATTR NDR_UNIXINFO NDR_NAMED_PIPE_AUTH NDR_DCOM NDR_NTPRINTING NDR_FSRVP NDR_WITNESS NDR_MDSSVC NDR_OPEN_FILES NDR_SMBXSRV''', private_library=True, grouping_library=True @@ -691,7 +681,7 @@ bld.SAMBA_LIBRARY('dcerpc-samba', deps='''RPC_NDR_LSA RPC_NDR_SAMR RPC_NDR_NETLOGON RPC_NDR_EVENTLOG RPC_NDR_DFS RPC_NDR_NTSVCS RPC_NDR_SVCCTL RPC_NDR_INITSHUTDOWN RPC_NDR_WKSSVC RPC_NDR_SRVSVC RPC_NDR_WINREG RPC_NDR_ECHO RPC_NDR_EPMAPPER - RPC_NDR_ATSVC RPC_NDR_SPOOLSS RPC_NDR_DNS''', + RPC_NDR_ATSVC RPC_NDR_SPOOLSS RPC_NDR_DNSSERVER''', public_deps='ndr-standard', private_library=True, grouping_library=True -- 1.7.0.4 From 2149a5bd254a76244bc8e24938a763a91c7cb828 Mon Sep 17 00:00:00 2001 From: Stefan Metzmacher Date: Fri, 7 Aug 2015 11:36:47 +0200 Subject: [PATCH 03/18] CVE-2016-0771: librpc: add ndr_dnsp_string_list_copy() helper function BUG: https://bugzilla.samba.org/show_bug.cgi?id=11128 BUG: https://bugzilla.samba.org/show_bug.cgi?id=11686 Signed-off-by: Stefan Metzmacher Reviewed-by: Garming Sam --- librpc/ndr/ndr_dnsp.c | 24 ++++++++++++++++++++++++ librpc/ndr/ndr_dnsp.h | 4 ++++ 2 files changed, 28 insertions(+), 0 deletions(-) diff --git a/librpc/ndr/ndr_dnsp.c b/librpc/ndr/ndr_dnsp.c index 46141c1..3cb96f9 100644 --- a/librpc/ndr/ndr_dnsp.c +++ b/librpc/ndr/ndr_dnsp.c @@ -222,3 +222,27 @@ enum ndr_err_code ndr_push_dnsp_string_list(struct ndr_push *ndr, int ndr_flags, } return NDR_ERR_SUCCESS; } + +enum ndr_err_code ndr_dnsp_string_list_copy(TALLOC_CTX *mem_ctx, + const struct dnsp_string_list *src, + struct dnsp_string_list *dst) +{ + size_t i; + + dst->count = 0; + dst->str = talloc_zero_array(mem_ctx, const char *, src->count); + if (dst->str == NULL) { + return NDR_ERR_ALLOC; + } + + for (i = 0; i < src->count; i++) { + dst->str[i] = talloc_strdup(dst->str, src->str[i]); + if (dst->str[i] == NULL) { + TALLOC_FREE(dst->str); + return NDR_ERR_ALLOC; + } + } + + dst->count = src->count; + return NDR_ERR_SUCCESS; +} diff --git a/librpc/ndr/ndr_dnsp.h b/librpc/ndr/ndr_dnsp.h index 67f952c..0d56633 100644 --- a/librpc/ndr/ndr_dnsp.h +++ b/librpc/ndr/ndr_dnsp.h @@ -27,3 +27,7 @@ void ndr_print_dnsp_string(struct ndr_print *ndr, const char *name, const char *dns_string); enum ndr_err_code ndr_pull_dnsp_string(struct ndr_pull *ndr, int ndr_flags, const char **string); enum ndr_err_code ndr_push_dnsp_string(struct ndr_push *ndr, int ndr_flags, const char *string); + +enum ndr_err_code ndr_dnsp_string_list_copy(TALLOC_CTX *mem_ctx, + const struct dnsp_string_list *src, + struct dnsp_string_list *dst); -- 1.7.0.4 From b57bdf15cc8506d16c5d39a2256fb1c5d1b9bdb5 Mon Sep 17 00:00:00 2001 From: Stefan Metzmacher Date: Fri, 7 Aug 2015 11:36:47 +0200 Subject: [PATCH 04/18] CVE-2016-0771: s4:dns_server: fix idl for dns_txt_record From RFC 1035: 3.3.14. TXT RDATA format +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+ / TXT-DATA / +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+ where: TXT-DATA One or more s. TXT RRs are used to hold descriptive text. The semantics of the text depends on the domain where it is found. Each record contains an array of strings instead of just one string. BUG: https://bugzilla.samba.org/show_bug.cgi?id=11128 BUG: https://bugzilla.samba.org/show_bug.cgi?id=11686 Signed-off-by: Stefan Metzmacher Reviewed-by: Garming Sam --- librpc/idl/dns.idl | 7 +++---- librpc/ndr/ndr_dns.c | 27 +++++++++++++++++++++++++++ librpc/wscript_build | 2 +- source4/dns_server/dns_query.c | 15 ++++++--------- source4/dns_server/dns_update.c | 31 ++++++------------------------- 5 files changed, 43 insertions(+), 39 deletions(-) diff --git a/librpc/idl/dns.idl b/librpc/idl/dns.idl index 1183bd1..918073c 100644 --- a/librpc/idl/dns.idl +++ b/librpc/idl/dns.idl @@ -8,7 +8,7 @@ encoding if it doesn't work out */ -import "misc.idl"; +import "misc.idl", "dnsp.idl"; [ helper("librpc/ndr/ndr_dns.h"), helpstring("DNS records"), @@ -163,9 +163,8 @@ interface dns dns_string exchange; } dns_mx_record; - typedef [public] struct { - [value(strlen(txt))] uint8 length; - [charset(DOS)] uint8 txt[length]; + typedef [public,nopull] struct { + dnsp_string_list txt; } dns_txt_record; typedef [public] struct { diff --git a/librpc/ndr/ndr_dns.c b/librpc/ndr/ndr_dns.c index 0b9e3b0..065d992 100644 --- a/librpc/ndr/ndr_dns.c +++ b/librpc/ndr/ndr_dns.c @@ -30,6 +30,7 @@ #include "includes.h" #include "librpc/gen_ndr/ndr_dns.h" #include "librpc/gen_ndr/ndr_misc.h" +#include "librpc/gen_ndr/ndr_dnsp.h" #include "system/locale.h" #include "lib/util/util_net.h" @@ -230,6 +231,29 @@ _PUBLIC_ enum ndr_err_code ndr_push_dns_string(struct ndr_push *ndr, return ndr_push_bytes(ndr, (const uint8_t *)"", 1); } +_PUBLIC_ enum ndr_err_code ndr_pull_dns_txt_record(struct ndr_pull *ndr, int ndr_flags, struct dns_txt_record *r) +{ + NDR_PULL_CHECK_FLAGS(ndr, ndr_flags); + if (ndr_flags & NDR_SCALARS) { + enum ndr_err_code ndr_err; + uint32_t data_size = ndr->data_size; + uint32_t record_size = 0; + ndr_err = ndr_token_retrieve(&ndr->array_size_list, r, + &record_size); + if (NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + NDR_PULL_NEED_BYTES(ndr, record_size); + ndr->data_size = ndr->offset + record_size; + } + NDR_CHECK(ndr_pull_align(ndr, 1)); + NDR_CHECK(ndr_pull_dnsp_string_list(ndr, NDR_SCALARS, &r->txt)); + NDR_CHECK(ndr_pull_trailer_align(ndr, 1)); + ndr->data_size = data_size; + } + if (ndr_flags & NDR_BUFFERS) { + } + return NDR_ERR_SUCCESS; +} + _PUBLIC_ enum ndr_err_code ndr_push_dns_res_rec(struct ndr_push *ndr, int ndr_flags, const struct dns_res_rec *r) @@ -302,6 +326,9 @@ _PUBLIC_ enum ndr_err_code ndr_pull_dns_res_rec(struct ndr_pull *ndr, NDR_CHECK(ndr_pull_uint16(ndr, NDR_SCALARS, &r->length)); _saved_offset1 = ndr->offset; if (r->length > 0) { + NDR_CHECK(ndr_token_store(ndr, &ndr->array_size_list, + &r->rdata, + r->length)); NDR_CHECK(ndr_pull_set_switch_value(ndr, &r->rdata, r->rr_type)); NDR_CHECK(ndr_pull_dns_rdata(ndr, NDR_SCALARS, diff --git a/librpc/wscript_build b/librpc/wscript_build index 244da21..d376bec 100644 --- a/librpc/wscript_build +++ b/librpc/wscript_build @@ -32,7 +32,7 @@ bld.SAMBA_SUBSYSTEM('NDR_DNSSERVER', bld.SAMBA_SUBSYSTEM('NDR_DNS', source='gen_ndr/ndr_dns.c ndr/ndr_dns.c', - public_deps='ndr' + public_deps='ndr NDR_DNSP' ) bld.SAMBA_SUBSYSTEM('NDR_DSBACKUP', diff --git a/source4/dns_server/dns_query.c b/source4/dns_server/dns_query.c index 4e3c6cc..9e30b71 100644 --- a/source4/dns_server/dns_query.c +++ b/source4/dns_server/dns_query.c @@ -46,8 +46,7 @@ static WERROR create_response_rr(const struct dns_name_question *question, { struct dns_res_rec *ans = *answers; uint16_t ai = *ancount; - char *tmp; - uint32_t i; + enum ndr_err_code ndr_err; ZERO_STRUCT(ans[ai]); @@ -101,14 +100,12 @@ static WERROR create_response_rr(const struct dns_name_question *question, } break; case DNS_QTYPE_TXT: - tmp = talloc_asprintf(ans, "\"%s\"", rec->data.txt.str[0]); - W_ERROR_HAVE_NO_MEMORY(tmp); - for (i=1; idata.txt.count; i++) { - tmp = talloc_asprintf_append_buffer( - tmp, " \"%s\"", rec->data.txt.str[i]); - W_ERROR_HAVE_NO_MEMORY(tmp); + ndr_err = ndr_dnsp_string_list_copy(ans, + &rec->data.txt, + &ans[ai].rdata.txt_record.txt); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + return WERR_NOMEM; } - ans[ai].rdata.txt_record.txt = tmp; break; default: DEBUG(0, ("Got unhandled type %u query.\n", rec->wType)); diff --git a/source4/dns_server/dns_update.c b/source4/dns_server/dns_update.c index c002b4d..60a4b36 100644 --- a/source4/dns_server/dns_update.c +++ b/source4/dns_server/dns_update.c @@ -299,9 +299,7 @@ static WERROR dns_rr_to_dnsp(TALLOC_CTX *mem_ctx, const struct dns_res_rec *rrec, struct dnsp_DnssrvRpcRecord *r) { - char *tmp; - char *txt_record_txt; - char *saveptr = NULL; + enum ndr_err_code ndr_err; if (rrec->rr_type == DNS_QTYPE_ALL) { return DNS_ERR(FORMAT_ERROR); @@ -354,28 +352,11 @@ static WERROR dns_rr_to_dnsp(TALLOC_CTX *mem_ctx, W_ERROR_HAVE_NO_MEMORY(r->data.mx.nameTarget); break; case DNS_QTYPE_TXT: - r->data.txt.count = 0; - r->data.txt.str = talloc_array(mem_ctx, const char *, - r->data.txt.count); - W_ERROR_HAVE_NO_MEMORY(r->data.txt.str); - - txt_record_txt = talloc_strdup(r->data.txt.str, - rrec->rdata.txt_record.txt); - W_ERROR_HAVE_NO_MEMORY(txt_record_txt); - - tmp = strtok_r(txt_record_txt, "\"", &saveptr); - while (tmp) { - if (strcmp(tmp, " ") == 0) { - tmp = strtok_r(NULL, "\"", &saveptr); - continue; - } - r->data.txt.str = talloc_realloc(mem_ctx, r->data.txt.str, const char *, - r->data.txt.count+1); - r->data.txt.str[r->data.txt.count] = talloc_strdup(r->data.txt.str, tmp); - W_ERROR_HAVE_NO_MEMORY(r->data.txt.str[r->data.txt.count]); - - r->data.txt.count++; - tmp = strtok_r(NULL, "\"", &saveptr); + ndr_err = ndr_dnsp_string_list_copy(mem_ctx, + &rrec->rdata.txt_record.txt, + &r->data.txt); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + return WERR_NOMEM; } break; -- 1.7.0.4 From 6835940066adb0ddd32bae558d1a8b313c45c1af Mon Sep 17 00:00:00 2001 From: Stefan Metzmacher Date: Fri, 7 Aug 2015 11:36:47 +0200 Subject: [PATCH 05/18] CVE-2016-0771: dns.idl: make use of dnsp_hinfo BUG: https://bugzilla.samba.org/show_bug.cgi?id=11128 BUG: https://bugzilla.samba.org/show_bug.cgi?id=11686 Signed-off-by: Stefan Metzmacher Reviewed-by: Garming Sam --- librpc/idl/dns.idl | 9 +-------- 1 files changed, 1 insertions(+), 8 deletions(-) diff --git a/librpc/idl/dns.idl b/librpc/idl/dns.idl index 918073c..5435fcf 100644 --- a/librpc/idl/dns.idl +++ b/librpc/idl/dns.idl @@ -152,13 +152,6 @@ interface dns } dns_soa_record; typedef [public] struct { - [value(strlen(cpu))] uint8 cpu_length; - [charset(DOS)] uint8 cpu[cpu_length]; - [value(strlen(os))] uint8 os_length; - [charset(DOS)] uint8 os[os_length]; - } dns_hinfo_record; - - typedef [public] struct { uint16 preference; dns_string exchange; } dns_mx_record; @@ -231,7 +224,7 @@ interface dns [case(DNS_QTYPE_CNAME)] dns_string cname_record; [case(DNS_QTYPE_SOA)] dns_soa_record soa_record; [case(DNS_QTYPE_PTR)] dns_string ptr_record; - [case(DNS_QTYPE_HINFO)] dns_hinfo_record hinfo_record; + [case(DNS_QTYPE_HINFO)] dnsp_hinfo hinfo_record; [case(DNS_QTYPE_MX)] dns_mx_record mx_record; [case(DNS_QTYPE_TXT)] dns_txt_record txt_record; [case(DNS_QTYPE_RP)] dns_rp_record rp_record; -- 1.7.0.4 From 8cda0b2172889bc125b4bd99b5b9c30fb2f5749e Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Wed, 6 Jan 2016 14:12:35 +1300 Subject: [PATCH 06/18] CVE-2016-0771: tests/dns: Modify dns tests to match new IDL 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 --- python/samba/tests/dns.py | 38 ++++++++++++++++++++------------------ 1 files changed, 20 insertions(+), 18 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 04ac356..6f2bbb5 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -20,12 +20,19 @@ import struct import random 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 dnsp, dnsserver +from samba.dcerpc import dns, dnsp, dnsserver +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 errstr(self, errcode): @@ -411,8 +418,7 @@ class TestDNSUpdates(DNSTest): 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) @@ -432,7 +438,7 @@ class TestDNSUpdates(DNSTest): 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"') + self.assertEquals(response.answers[0].rdata.txt.str[0], '"This is a test"') def test_update_add_two_txt_records(self): "test adding two txt records works" @@ -452,8 +458,8 @@ class TestDNSUpdates(DNSTest): 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"' + rdata = make_txt_record(['"This is a test"', + '"and this is a test, too"']) r.rdata = rdata updates.append(r) p.nscount = len(updates) @@ -473,7 +479,8 @@ class TestDNSUpdates(DNSTest): 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.assertEquals(response.answers[0].rdata.txt.str[0], '"This is a test"') + self.assertEquals(response.answers[0].rdata.txt.str[1], '"and this is a test, too"') def test_delete_record(self): "Test if deleting records works" @@ -497,8 +504,7 @@ class TestDNSUpdates(DNSTest): 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) @@ -534,8 +540,7 @@ class TestDNSUpdates(DNSTest): 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) @@ -577,8 +582,7 @@ class TestDNSUpdates(DNSTest): 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) @@ -614,8 +618,7 @@ class TestDNSUpdates(DNSTest): 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) @@ -652,8 +655,7 @@ class TestDNSUpdates(DNSTest): 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) -- 1.7.0.4 From 99f780f26f5ad49fa4f956ff1a3416e224ba2843 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Thu, 21 Jan 2016 16:58:40 +1300 Subject: [PATCH 07/18] CVE-2016-0771: tests/dns: prepare script for further testing 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 --- python/samba/tests/dns.py | 12 ++++++------ 1 files changed, 6 insertions(+), 6 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 6f2bbb5..c82020b 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -35,6 +35,11 @@ def make_txt_record(records): class DNSTest(TestCase): + def get_loadparm(self): + lp = param.LoadParm() + lp.load(os.getenv("SMB_CONF_PATH")) + return lp + def errstr(self, errcode): "Return a readable error code" string_codes = [ @@ -855,11 +860,6 @@ class TestInvalidQueries(DNSTest): s.close() class TestZones(DNSTest): - def get_loadparm(self): - lp = param.LoadParm() - lp.load(os.getenv("SMB_CONF_PATH")) - return lp - def get_credentials(self, lp): creds = credentials.Credentials() creds.guess(lp) @@ -873,7 +873,7 @@ class TestZones(DNSTest): self.creds = self.get_credentials(self.lp) self.server = os.getenv("SERVER_IP") self.zone = "test.lan" - self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s" % (self.server), + self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server), self.lp, self.creds) def tearDown(self): -- 1.7.0.4 From 69c1ecb13ac908b31435d7edee6b3532ff963b13 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Thu, 21 Jan 2016 15:43:55 +1300 Subject: [PATCH 08/18] CVE-2016-0771: tests/dns: FORMERR can simply timeout against Windows Two requests with identical parameters which are poorly formatted, can non-deterministically return FORMERR or simply fail to give a response. Setting the timeout to a number allows Windows to succeed. 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 --- python/samba/tests/dns.py | 63 +++++++++++++++++++++++++++++++++++++------- 1 files changed, 53 insertions(+), 10 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index c82020b..3e8c2a3 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -24,6 +24,9 @@ from samba import credentials, param from samba.tests import TestCase from samba.dcerpc import dns, dnsp, dnsserver +# This timeout only has relevance when testing against Windows +# Format errors tend to return patchy responses, so a timeout is needed. +timeout = None def make_txt_record(records): rdata_txt = dns.txt_record() @@ -97,7 +100,8 @@ class DNSTest(TestCase): "Helper to get dns domain" return os.getenv('REALM', 'example.com').lower() - def dns_transaction_udp(self, packet, host=os.getenv('SERVER_IP'), dump=False): + def dns_transaction_udp(self, packet, host=os.getenv('SERVER_IP'), + dump=False, timeout=timeout): "send a DNS query and read the reply" s = None try: @@ -105,6 +109,7 @@ class DNSTest(TestCase): if dump: print self.hexdump(send_packet) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0) + s.settimeout(timeout) s.connect((host, 53)) s.send(send_packet, 0) recv_packet = s.recv(2048, 0) @@ -115,7 +120,8 @@ class DNSTest(TestCase): 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=os.getenv('SERVER_IP'), + dump=False, timeout=timeout): "send a DNS query and read the reply" s = None try: @@ -123,6 +129,7 @@ class DNSTest(TestCase): if dump: print self.hexdump(send_packet) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) + s.settimeout(timeout) s.connect((host, 53)) tcp_packet = struct.pack('!H', len(send_packet)) tcp_packet += send_packet @@ -217,8 +224,15 @@ class TestSimpleQueries(DNSTest): questions.append(q) self.finish_name_packet(p, questions) - response = self.dns_transaction_udp(p) - self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR) + try: + response = self.dns_transaction_udp(p) + self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR) + except socket.timeout: + # Windows chooses not to respond to incorrectly formatted queries. + # Although this appears to be non-deterministic even for the same + # request twice, it also appears to be based on a how poorly the + # request is formatted. + pass def test_qtype_all_query(self): "create a QTYPE_ALL query" @@ -256,8 +270,15 @@ class TestSimpleQueries(DNSTest): questions.append(q) self.finish_name_packet(p, questions) - response = self.dns_transaction_udp(p) - self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTIMP) + try: + response = self.dns_transaction_udp(p) + self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTIMP) + except socket.timeout: + # Windows chooses not to respond to incorrectly formatted queries. + # Although this appears to be non-deterministic even for the same + # request twice, it also appears to be based on a how poorly the + # request is formatted. + pass # Only returns an authority section entry in BIND and Win DNS # FIXME: Enable one Samba implements this feature @@ -310,8 +331,15 @@ class TestDNSUpdates(DNSTest): updates.append(u) self.finish_name_packet(p, updates) - response = self.dns_transaction_udp(p) - self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR) + try: + response = self.dns_transaction_udp(p) + self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR) + except socket.timeout: + # Windows chooses not to respond to incorrectly formatted queries. + # Although this appears to be non-deterministic even for the same + # request twice, it also appears to be based on a how poorly the + # request is formatted. + pass def test_update_wrong_qclass(self): "create update with DNS_QCLASS_NONE" @@ -349,8 +377,15 @@ class TestDNSUpdates(DNSTest): p.ancount = len(prereqs) p.answers = prereqs - response = self.dns_transaction_udp(p) - self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR) + try: + response = self.dns_transaction_udp(p) + self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR) + except socket.timeout: + # Windows chooses not to respond to incorrectly formatted queries. + # Although this appears to be non-deterministic even for the same + # request twice, it also appears to be based on a how poorly the + # request is formatted. + pass # I'd love to test this one, but it segfaults. :) # def test_update_prereq_with_non_null_length(self): @@ -833,6 +868,7 @@ class TestInvalidQueries(DNSTest): def test_one_a_reply(self): "send a reply instead of a query" + global timeout p = self.make_name_packet(dns.DNS_OPCODE_QUERY) questions = [] @@ -848,6 +884,7 @@ class TestInvalidQueries(DNSTest): try: send_packet = ndr.ndr_pack(p) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) + s.settimeout(timeout) host=os.getenv('SERVER_IP') s.connect((host, 53)) tcp_packet = struct.pack('!H', len(send_packet)) @@ -855,6 +892,12 @@ class TestInvalidQueries(DNSTest): s.send(tcp_packet, 0) recv_packet = s.recv(0xffff + 2, 0) self.assertEquals(0, len(recv_packet)) + except socket.timeout: + # Windows chooses not to respond to incorrectly formatted queries. + # Although this appears to be non-deterministic even for the same + # request twice, it also appears to be based on a how poorly the + # request is formatted. + pass finally: if s is not None: s.close() -- 1.7.0.4 From fc7e9fe7db8570d6a52683d1ef69e6295493a85b Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Thu, 21 Jan 2016 17:08:18 +1300 Subject: [PATCH 09/18] CVE-2016-0771: tests/dns: Add a comment regarding odd Windows behaviour 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 --- python/samba/tests/dns.py | 1 + 1 files changed, 1 insertions(+), 0 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 3e8c2a3..5f9f3e2 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -963,6 +963,7 @@ class TestZones(DNSTest): self.finish_name_packet(p, questions) response = self.dns_transaction_udp(p) + # Windows returns OK while BIND logically seems to return NXDOMAIN 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) -- 1.7.0.4 From 8ce145f14480f6f4b21ef9a1a08379dcf3193d61 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Tue, 15 Dec 2015 17:22:32 +1300 Subject: [PATCH 10/18] CVE-2016-0771: tests/dns: restore formerly segfaulting test This was on the client side, due the a strlen(NULL) on the previously DOS-encoded TXT field. With a new IDL structure, this segfault no longer exists. Note that both Samba and Windows return NXRRSET instead of FORMERR. 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 --- python/samba/tests/dns.py | 51 ++++++++++++++++++++++----------------------- 1 files changed, 25 insertions(+), 26 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 5f9f3e2..f05ab05 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -387,32 +387,31 @@ class TestDNSUpdates(DNSTest): # request is formatted. pass -# 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_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" -- 1.7.0.4 From 2cd2714270fb3b670136696c683ce66e0ce84c34 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Thu, 21 Jan 2016 10:25:44 +1300 Subject: [PATCH 11/18] CVE-2016-0771: tests/dns: Correct error code for formerly unrun test Both Samba and Windows returned NXRRSET 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 --- python/samba/tests/dns.py | 2 +- 1 files changed, 1 insertions(+), 1 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index f05ab05..d8c488e 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -411,7 +411,7 @@ class TestDNSUpdates(DNSTest): p.answers = prereqs response = self.dns_transaction_udp(p) - self.assert_dns_rcode_equals(response, dns.DNS_RCODE_FORMERR) + self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET) def test_update_prereq_nonexisting_name(self): "test update with a nonexisting name" -- 1.7.0.4 From 02f2f573f455e251c17a3b1f6561acc11e1877c3 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Mon, 18 Jan 2016 12:39:46 +1300 Subject: [PATCH 12/18] CVE-2016-0771: tests/dns: Add some more test cases for TXT records 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 --- python/samba/tests/dns.py | 110 +++++++++++++++++++++++++++++---------------- 1 files changed, 71 insertions(+), 39 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index d8c488e..2033af8 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -439,37 +439,35 @@ class TestDNSUpdates(DNSTest): 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 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 = "textrec.%s" % self.get_dns_domain() + 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(['"This is a test"']) + rdata = make_txt_record(txt_array) 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) + 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 = [] - name = "textrec.%s" % self.get_dns_domain() q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN) questions.append(q) @@ -477,49 +475,83 @@ class TestDNSUpdates(DNSTest): 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[0], '"This is a test"') + self.assertEquals(response.answers[0].rdata.txt.str, txt_array) - def test_update_add_two_txt_records(self): - "test adding two txt records works" - p = self.make_name_packet(dns.DNS_OPCODE_UPDATE) - updates = [] + 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.check_query_txt(prefix, txt) - name = self.get_dns_domain() + 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) - u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN) - updates.append(u) - self.finish_name_packet(p, updates) + 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) - 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 = make_txt_record(['"This is a test"', - '"and this is a test, too"']) - r.rdata = rdata - updates.append(r) - p.nscount = len(updates) - p.nsrecs = updates + 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']) - p = self.make_name_packet(dns.DNS_OPCODE_QUERY) - questions = [] + 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']) - name = "textrec2.%s" % self.get_dns_domain() - q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN) - questions.append(q) + 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.finish_name_packet(p, questions) + 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.assertEquals(response.ancount, 1) - self.assertEquals(response.answers[0].rdata.txt.str[0], '"This is a test"') - self.assertEquals(response.answers[0].rdata.txt.str[1], '"and this is a test, too"') + 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" -- 1.7.0.4 From 13eb70d2396c7fdcb84af7571b56c5f4a5fc8cd3 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Wed, 27 Jan 2016 17:41:44 +1300 Subject: [PATCH 13/18] 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 --- python/samba/tests/dns.py | 266 +++++++++++++++++++++++++++------------------ 1 files changed, 160 insertions(+), 106 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 2033af8..25897a7 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -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 -- 1.7.0.4 From a065a4ac13bfe66a1b58e504869ea0fde53e4dce Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Thu, 28 Jan 2016 12:36:43 +1300 Subject: [PATCH 14/18] CVE-2016-0771: dnsserver: don't force UTF-8 for TXT While using a charset is not entirely logical, it allows testing of non UTF-8 data (like inserting 0xFF into the TXT string). 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 --- librpc/idl/dnsserver.idl | 2 +- 1 files changed, 1 insertions(+), 1 deletions(-) diff --git a/librpc/idl/dnsserver.idl b/librpc/idl/dnsserver.idl index ca9c371..c7742e7 100644 --- a/librpc/idl/dnsserver.idl +++ b/librpc/idl/dnsserver.idl @@ -73,7 +73,7 @@ import "misc.idl", "dnsp.idl"; typedef [public,gensize] struct { [value(strlen(str))] uint8 len; - [charset(UTF8)] uint8 str[len]; + [charset(UNIX)] uint8 str[len]; } DNS_RPC_NAME; -- 1.7.0.4 From a7977680af199f5855f494cf8d8bb0720c46a613 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Thu, 28 Jan 2016 12:54:58 +1300 Subject: [PATCH 15/18] CVE-2016-0771: tests/dns: RPC => DNS roundtrip test Make sure that TXT entries stored via RPC come out the same in DNS. This has one caveat in that adding over RPC in Windows eats slashes, and so fails there. 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 --- python/samba/tests/dns.py | 202 ++++++++++++++++++++++++++++++++++++++++++--- source4/selftest/tests.py | 1 + 2 files changed, 190 insertions(+), 13 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 25897a7..3e655c3 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -23,6 +23,7 @@ import samba.ndr as ndr 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 # This timeout only has relevance when testing against Windows # Format errors tend to return patchy responses, so a timeout is needed. @@ -879,7 +880,7 @@ class TestZones(DNSTest): self.creds = self.get_credentials(self.lp) self.server = os.getenv("SERVER_IP") self.zone = "test.lan" - self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server), + self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server_ip), self.lp, self.creds) def tearDown(self): @@ -899,7 +900,7 @@ class TestZones(DNSTest): zone_create.dwDpFlags = dnsserver.DNS_DP_DOMAIN_DEFAULT self.rpc_conn.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN, 0, - self.server, + self.server_ip, None, 0, 'ZoneCreate', @@ -909,7 +910,7 @@ class TestZones(DNSTest): def delete_zone(self, zone): self.rpc_conn.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN, 0, - self.server, + self.server_ip, zone, 0, 'DeleteZoneFromDs', @@ -957,12 +958,31 @@ class TestRPCRoundtrip(DNSTest): 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.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"', '', ''] @@ -970,7 +990,7 @@ class TestRPCRoundtrip(DNSTest): 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.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\\"" "" ""')) @@ -980,7 +1000,7 @@ class TestRPCRoundtrip(DNSTest): 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.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"')) @@ -990,11 +1010,66 @@ class TestRPCRoundtrip(DNSTest): 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.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()) + + 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" @@ -1003,7 +1078,7 @@ class TestRPCRoundtrip(DNSTest): 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.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"')) @@ -1013,11 +1088,30 @@ class TestRPCRoundtrip(DNSTest): 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.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'] @@ -1025,11 +1119,30 @@ class TestRPCRoundtrip(DNSTest): 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.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'] @@ -1037,11 +1150,33 @@ class TestRPCRoundtrip(DNSTest): 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.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"', @@ -1050,12 +1185,34 @@ class TestRPCRoundtrip(DNSTest): 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.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', [] @@ -1063,11 +1220,30 @@ class TestRPCRoundtrip(DNSTest): 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.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) + if __name__ == "__main__": import unittest unittest.main() diff --git a/source4/selftest/tests.py b/source4/selftest/tests.py index e284c2c..8f4976c 100755 --- a/source4/selftest/tests.py +++ b/source4/selftest/tests.py @@ -291,6 +291,7 @@ for f in sorted(os.listdir(os.path.join(samba4srcdir, "../pidl/tests"))): # DNS tests planpythontestsuite("fl2003dc:local", "samba.tests.dns") + for t in smbtorture4_testsuites("dns_internal."): plansmbtorture4testsuite(t, "ad_dc_ntvfs:local", '//$SERVER/whavever') -- 1.7.0.4 From db7bd372d2bdb541cfb2df3d0e8a95e2ca093288 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Fri, 22 Jan 2016 11:35:03 +1300 Subject: [PATCH 16/18] CVE-2016-0771: tests: rename test getopt to get_opt This avoids any conflicts in this directory with the original toplevel getopt. 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 --- python/samba/tests/get_opt.py | 55 +++++++++++++++++++++++++++++++++++++++++ python/samba/tests/getopt.py | 55 ----------------------------------------- selftest/tests.py | 2 +- 3 files changed, 56 insertions(+), 56 deletions(-) create mode 100644 python/samba/tests/get_opt.py delete mode 100644 python/samba/tests/getopt.py diff --git a/python/samba/tests/get_opt.py b/python/samba/tests/get_opt.py new file mode 100644 index 0000000..14ee0a7 --- /dev/null +++ b/python/samba/tests/get_opt.py @@ -0,0 +1,55 @@ +# Unix SMB/CIFS implementation. +# Copyright (C) Jelmer Vernooij 2011 +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# + +"""Tests for option parsing. + +""" + +import optparse +from samba.getopt import ( + AUTO_USE_KERBEROS, + DONT_USE_KERBEROS, + MUST_USE_KERBEROS, + parse_kerberos_arg, + ) +import samba.tests + +class KerberosOptionTests(samba.tests.TestCase): + + def test_parse_true(self): + self.assertEquals( + MUST_USE_KERBEROS, parse_kerberos_arg("yes", "--kerberos")) + self.assertEquals( + MUST_USE_KERBEROS, parse_kerberos_arg("true", "--kerberos")) + self.assertEquals( + MUST_USE_KERBEROS, parse_kerberos_arg("1", "--kerberos")) + + def test_parse_false(self): + self.assertEquals( + DONT_USE_KERBEROS, parse_kerberos_arg("no", "--kerberos")) + self.assertEquals( + DONT_USE_KERBEROS, parse_kerberos_arg("false", "--kerberos")) + self.assertEquals( + DONT_USE_KERBEROS, parse_kerberos_arg("0", "--kerberos")) + + def test_parse_auto(self): + self.assertEquals( + AUTO_USE_KERBEROS, parse_kerberos_arg("auto", "--kerberos")) + + def test_parse_invalid(self): + self.assertRaises(optparse.OptionValueError, + parse_kerberos_arg, "blah?", "--kerberos") diff --git a/python/samba/tests/getopt.py b/python/samba/tests/getopt.py deleted file mode 100644 index 14ee0a7..0000000 --- a/python/samba/tests/getopt.py +++ /dev/null @@ -1,55 +0,0 @@ -# Unix SMB/CIFS implementation. -# Copyright (C) Jelmer Vernooij 2011 -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -# - -"""Tests for option parsing. - -""" - -import optparse -from samba.getopt import ( - AUTO_USE_KERBEROS, - DONT_USE_KERBEROS, - MUST_USE_KERBEROS, - parse_kerberos_arg, - ) -import samba.tests - -class KerberosOptionTests(samba.tests.TestCase): - - def test_parse_true(self): - self.assertEquals( - MUST_USE_KERBEROS, parse_kerberos_arg("yes", "--kerberos")) - self.assertEquals( - MUST_USE_KERBEROS, parse_kerberos_arg("true", "--kerberos")) - self.assertEquals( - MUST_USE_KERBEROS, parse_kerberos_arg("1", "--kerberos")) - - def test_parse_false(self): - self.assertEquals( - DONT_USE_KERBEROS, parse_kerberos_arg("no", "--kerberos")) - self.assertEquals( - DONT_USE_KERBEROS, parse_kerberos_arg("false", "--kerberos")) - self.assertEquals( - DONT_USE_KERBEROS, parse_kerberos_arg("0", "--kerberos")) - - def test_parse_auto(self): - self.assertEquals( - AUTO_USE_KERBEROS, parse_kerberos_arg("auto", "--kerberos")) - - def test_parse_invalid(self): - self.assertRaises(optparse.OptionValueError, - parse_kerberos_arg, "blah?", "--kerberos") diff --git a/selftest/tests.py b/selftest/tests.py index 872fbaf..2a07f9e 100644 --- a/selftest/tests.py +++ b/selftest/tests.py @@ -48,7 +48,7 @@ planpythontestsuite("none", "api", name="ldb.python", extra_path=['lib/ldb/tests planpythontestsuite("none", "samba.tests.credentials") planpythontestsuite("none", "samba.tests.registry") planpythontestsuite("none", "samba.tests.auth") -planpythontestsuite("none", "samba.tests.getopt") +planpythontestsuite("none", "samba.tests.get_opt") planpythontestsuite("none", "samba.tests.security") planpythontestsuite("none", "samba.tests.dcerpc.misc") planpythontestsuite("none", "samba.tests.dcerpc.integer") -- 1.7.0.4 From 323b3e9dc5db73493b12a7cf43e2dad5086c904a Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Fri, 29 Jan 2016 17:03:56 +1300 Subject: [PATCH 17/18] CVE-2016-0771: tests/dns: change samba.tests.dns from being a unittest This makes it easier to invoke, particularly against Windows. 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 --- python/samba/tests/dns.py | 47 +++++++++++++++++++++++++++++++++++++------- source4/selftest/tests.py | 2 +- 2 files changed, 40 insertions(+), 9 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 3e655c3..729fbc1 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -16,6 +16,7 @@ # import os +import sys import struct import random import socket @@ -24,10 +25,39 @@ 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 +from samba.tests.subunitrun import SubunitOptions, TestProgram +import samba.getopt as options +import optparse + +parser = optparse.OptionParser("dns.py [options]") +sambaopts = options.SambaOptions(parser) +parser.add_option_group(sambaopts) # This timeout only has relevance when testing against Windows # Format errors tend to return patchy responses, so a timeout is needed. -timeout = None +parser.add_option("--timeout", type="int", dest="timeout", + help="Specify timeout for DNS requests") + +# use command line creds if available +credopts = options.CredentialsOptions(parser) +parser.add_option_group(credopts) +subunitopts = SubunitOptions(parser) +parser.add_option_group(subunitopts) + +opts, args = parser.parse_args() + +lp = sambaopts.get_loadparm() +creds = credopts.get_credentials(lp) + +timeout = opts.timeout + +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() @@ -39,10 +69,13 @@ def make_txt_record(records): class DNSTest(TestCase): - def get_loadparm(self): - lp = param.LoadParm() - lp.load(os.getenv("SMB_CONF_PATH")) - return lp + 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" @@ -1244,6 +1277,4 @@ class TestRPCRoundtrip(DNSTest): 0, self.server_ip, self.get_dns_domain(), name, None, add_rec_buf) -if __name__ == "__main__": - import unittest - unittest.main() +TestProgram(module=__name__, opts=subunitopts) diff --git a/source4/selftest/tests.py b/source4/selftest/tests.py index 8f4976c..69330f2 100755 --- a/source4/selftest/tests.py +++ b/source4/selftest/tests.py @@ -290,7 +290,7 @@ for f in sorted(os.listdir(os.path.join(samba4srcdir, "../pidl/tests"))): planperltestsuite("pidl.%s" % f[:-3], os.path.normpath(os.path.join(samba4srcdir, "../pidl/tests", f))) # DNS tests -planpythontestsuite("fl2003dc:local", "samba.tests.dns") +plantestsuite_loadlist("samba.tests.dns", "fl2003dc:local", [python, os.path.join(srcdir(), "python/samba/tests/dns.py"), '$SERVER', '$SERVER_IP', '--machine-pass', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN', '$LOADLIST', '$LISTOPT']) for t in smbtorture4_testsuites("dns_internal."): plansmbtorture4testsuite(t, "ad_dc_ntvfs:local", '//$SERVER/whavever') -- 1.7.0.4 From 11bf712347a38f1f3214272395b2d9232d655604 Mon Sep 17 00:00:00 2001 From: Garming Sam Date: Fri, 29 Jan 2016 17:28:54 +1300 Subject: [PATCH 18/18] CVE-2016-0771: tests/dns: Remove dependencies on env variables Now that it is invoked as a normal script, there should be less of them. 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 --- python/samba/tests/dns.py | 68 ++++++++++++++++----------------------------- 1 files changed, 24 insertions(+), 44 deletions(-) diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py index 729fbc1..e0739d0 100644 --- a/python/samba/tests/dns.py +++ b/python/samba/tests/dns.py @@ -132,9 +132,9 @@ class DNSTest(TestCase): 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'), + def dns_transaction_udp(self, packet, host=server_ip, dump=False, timeout=timeout): "send a DNS query and read the reply" s = None @@ -154,7 +154,7 @@ class DNSTest(TestCase): if s is not None: s.close() - def dns_transaction_tcp(self, packet, host=os.getenv('SERVER_IP'), + def dns_transaction_tcp(self, packet, host=server_ip, dump=False, timeout=timeout): "send a DNS query and read the reply" s = None @@ -221,7 +221,7 @@ class TestSimpleQueries(DNSTest): 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) @@ -232,14 +232,14 @@ class TestSimpleQueries(DNSTest): 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) @@ -250,14 +250,14 @@ class TestSimpleQueries(DNSTest): 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" % (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_MX, dns.DNS_QCLASS_IN) print "asking for ", q.name questions.append(q) @@ -271,7 +271,7 @@ class TestSimpleQueries(DNSTest): p = self.make_name_packet(dns.DNS_OPCODE_QUERY) questions = [] - name = "invalid-%s.%s" % (os.getenv('SERVER'), self.get_dns_domain()) + 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) @@ -287,7 +287,7 @@ class TestSimpleQueries(DNSTest): 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) @@ -311,7 +311,7 @@ class TestSimpleQueries(DNSTest): 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) @@ -328,7 +328,7 @@ class TestSimpleQueries(DNSTest): 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) @@ -337,7 +337,7 @@ class TestSimpleQueries(DNSTest): 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) @@ -394,7 +394,7 @@ 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) @@ -439,7 +439,7 @@ class TestDNSUpdates(DNSTest): 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 @@ -472,7 +472,7 @@ class TestDNSUpdates(DNSTest): 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_ANY r.ttl = 0 @@ -777,7 +777,7 @@ class TestComplexQueries(DNSTest): 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 @@ -803,7 +803,7 @@ class TestComplexQueries(DNSTest): 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 @@ -828,10 +828,10 @@ class TestComplexQueries(DNSTest): 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): @@ -841,7 +841,7 @@ 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: @@ -850,7 +850,7 @@ class TestInvalidQueries(DNSTest): 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) @@ -861,7 +861,7 @@ class TestInvalidQueries(DNSTest): 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" @@ -882,7 +882,7 @@ class TestInvalidQueries(DNSTest): send_packet = ndr.ndr_pack(p) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) s.settimeout(timeout) - host=os.getenv('SERVER_IP') + host=self.server_ip s.connect((host, 53)) tcp_packet = struct.pack('!H', len(send_packet)) tcp_packet += send_packet @@ -900,18 +900,8 @@ class TestInvalidQueries(DNSTest): s.close() class TestZones(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(TestZones, self).setUp() - self.lp = self.get_loadparm() - self.creds = self.get_credentials(self.lp) - self.server = os.getenv("SERVER_IP") self.zone = "test.lan" self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server_ip), self.lp, self.creds) @@ -979,18 +969,8 @@ class TestZones(DNSTest): 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_ip), self.lp, self.creds) -- 1.7.0.4