sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2024-12-27 03:21:53 +03:00
Code
93c2a95223
samba-mirror/python/samba/tests/dns.py
Douglas Bagnall 018c0138c2 python/tests/dns*: remove unused imports 
Signed-off-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Noel Power <npower@samba.org>
2018-10-25 21:45:54 +02:00

1913 lines
71 KiB
Python
Raw Blame History

# Unix SMB/CIFS implementation.
# Copyright (C) Kai Blin <kai@samba.org> 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 <http://www.gnu.org/licenses/>.
#
from __future__ import print_function
from samba import dsdb
from samba.ndr import ndr_unpack, ndr_pack
from samba.samdb import SamDB
from samba.auth import system_session
import ldb
from ldb import ERR_OPERATIONS_ERROR
import os
import sys
import struct
import socket
import samba.ndr as ndr
from samba import credentials
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
from samba import werror, WERRORError
from samba.tests.dns_base import DNSTest
import samba.getopt as options
import optparse
parser = optparse.OptionParser("dns.py <server name> <server ip> [options]")
sambaopts = options.SambaOptions(parser)
parser.add_option_group(sambaopts)
# This timeout only has relevance when testing against Windows
# Format errors tend to return patchy responses, so a timeout is needed.
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)
class TestSimpleQueries(DNSTest):
def setUp(self):
super(TestSimpleQueries, self).setUp()
global server, server_ip, lp, creds, timeout
self.server = server_name
self.server_ip = server_ip
self.lp = lp
self.creds = creds
self.timeout = timeout
def test_one_a_query(self):
"create a query packet containing one query record"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rdata,
self.server_ip)
def test_one_SOA_query(self):
"create a query packet containing one query record for the SOA"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s" % (self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(
response.answers[0].rdata.mname.upper(),
("%s.%s" % (self.server, self.get_dns_domain())).upper())
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" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_tcp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rdata,
self.server_ip)
def test_one_mx_query(self):
"create a query packet causing an empty RCODE_OK answer"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 0)
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "invalid-%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 0)
def test_two_queries(self):
"create a query packet containing two query records"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
questions.append(q)
name = "%s.%s" % ('bogusname', self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
try:
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
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"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_ALL, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
num_answers = 1
dc_ipv6 = os.getenv('SERVER_IPV6')
if dc_ipv6 is not None:
num_answers += 1
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, num_answers)
self.assertEquals(response.answers[0].rdata,
self.server_ip)
if dc_ipv6 is not None:
self.assertEquals(response.answers[1].rdata, dc_ipv6)
def test_qclass_none_query(self):
"create a QCLASS_NONE query"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(
name,
dns.DNS_QTYPE_ALL,
dns.DNS_QCLASS_NONE)
questions.append(q)
self.finish_name_packet(p, questions)
try:
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
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
def test_soa_hostname_query(self):
"create a SOA query for a hostname"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
# We don't get SOA records for single hosts
self.assertEquals(response.ancount, 0)
# But we do respond with an authority section
self.assertEqual(response.nscount, 1)
def test_soa_domain_query(self):
"create a SOA query for a domain"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = self.get_dns_domain()
q = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rdata.minimum, 3600)
class TestDNSUpdates(DNSTest):
def setUp(self):
super(TestDNSUpdates, self).setUp()
global server, server_ip, lp, creds, timeout
self.server = server_name
self.server_ip = server_ip
self.lp = lp
self.creds = creds
self.timeout = timeout
def test_two_updates(self):
"create two update requests"
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
updates = []
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)
name = self.get_dns_domain()
u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
updates.append(u)
self.finish_name_packet(p, updates)
try:
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
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"
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
updates = []
name = self.get_dns_domain()
u = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_NONE)
updates.append(u)
self.finish_name_packet(p, updates)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTIMP)
def test_update_prereq_with_non_null_ttl(self):
"test update with a non-null TTL"
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" % (self.server, self.get_dns_domain())
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 1
r.length = 0
prereqs.append(r)
p.ancount = len(prereqs)
p.answers = prereqs
try:
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
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_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" % (self.server, self.get_dns_domain())
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_ANY
r.ttl = 0
r.length = 1
prereqs.append(r)
p.ancount = len(prereqs)
p.answers = prereqs
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET)
def test_update_prereq_nonexisting_name(self):
"test update with a nonexisting name"
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 = "idontexist.%s" % self.get_dns_domain()
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_ANY
r.ttl = 0
r.length = 0
prereqs.append(r)
p.ancount = len(prereqs)
p.answers = prereqs
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXRRSET)
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, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
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"
NAME = "deleterec.%s" % self.get_dns_domain()
# First, create a record to make sure we have a record to delete.
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 = NAME
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
rdata = self.make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
# Now check the record is around
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, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
# Now delete the record
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 = NAME
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 0
r.length = 0xffff
rdata = self.make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
# And finally check it's gone
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, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
def test_readd_record(self):
"Test if adding, deleting and then readding a records works"
NAME = "readdrec.%s" % self.get_dns_domain()
# Create the record
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 = NAME
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
rdata = self.make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
# Now check the record is around
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, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
# Now delete the record
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 = NAME
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 0
r.length = 0xffff
rdata = self.make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
# check it's gone
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, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NXDOMAIN)
# recreate the record
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 = NAME
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
rdata = self.make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
# Now check the record is around
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, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
def test_update_add_mx_record(self):
"test adding MX records works"
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
updates = []
name = self.get_dns_domain()
u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
updates.append(u)
self.finish_name_packet(p, updates)
updates = []
r = dns.res_rec()
r.name = "%s" % self.get_dns_domain()
r.rr_type = dns.DNS_QTYPE_MX
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
rdata = dns.mx_record()
rdata.preference = 10
rdata.exchange = 'mail.%s' % self.get_dns_domain()
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s" % self.get_dns_domain()
q = self.make_name_question(name, dns.DNS_QTYPE_MX, dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assertEqual(response.ancount, 1)
ans = response.answers[0]
self.assertEqual(ans.rr_type, dns.DNS_QTYPE_MX)
self.assertEqual(ans.rdata.preference, 10)
self.assertEqual(ans.rdata.exchange, 'mail.%s' % self.get_dns_domain())
class TestComplexQueries(DNSTest):
def make_dns_update(self, key, value, qtype):
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
name = self.get_dns_domain()
u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
self.finish_name_packet(p, [u])
r = dns.res_rec()
r.name = key
r.rr_type = qtype
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 900
r.length = 0xffff
r.rdata = value
p.nscount = 1
p.nsrecs = [r]
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
def setUp(self):
super(TestComplexQueries, self).setUp()
global server, server_ip, lp, creds, timeout
self.server = server_name
self.server_ip = server_ip
self.lp = lp
self.creds = creds
self.timeout = timeout
def test_one_a_query(self):
"create a query packet containing one query record"
try:
# Create the record
name = "cname_test.%s" % self.get_dns_domain()
rdata = "%s.%s" % (self.server, self.get_dns_domain())
self.make_dns_update(name, rdata, dns.DNS_QTYPE_CNAME)
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
# Check the record
name = "cname_test.%s" % self.get_dns_domain()
q = self.make_name_question(name,
dns.DNS_QTYPE_A,
dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=self.server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 2)
self.assertEquals(response.answers[0].rr_type, dns.DNS_QTYPE_CNAME)
self.assertEquals(response.answers[0].rdata, "%s.%s" %
(self.server, self.get_dns_domain()))
self.assertEquals(response.answers[1].rr_type, dns.DNS_QTYPE_A)
self.assertEquals(response.answers[1].rdata,
self.server_ip)
finally:
# Delete the record
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 = "cname_test.%s" % self.get_dns_domain()
r.rr_type = dns.DNS_QTYPE_CNAME
r.rr_class = dns.DNS_QCLASS_NONE
r.ttl = 0
r.length = 0xffff
r.rdata = "%s.%s" % (self.server, self.get_dns_domain())
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
(response, response_packet) =\
self.dns_transaction_udp(p, host=self.server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
def test_cname_two_chain(self):
name0 = "cnamechain0.%s" % self.get_dns_domain()
name1 = "cnamechain1.%s" % self.get_dns_domain()
name2 = "cnamechain2.%s" % self.get_dns_domain()
self.make_dns_update(name1, name2, dns.DNS_QTYPE_CNAME)
self.make_dns_update(name2, name0, dns.DNS_QTYPE_CNAME)
self.make_dns_update(name0, server_ip, dns.DNS_QTYPE_A)
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
q = self.make_name_question(name1, dns.DNS_QTYPE_A,
dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 3)
self.assertEquals(response.answers[0].rr_type, dns.DNS_QTYPE_CNAME)
self.assertEquals(response.answers[0].name, name1)
self.assertEquals(response.answers[0].rdata, name2)
self.assertEquals(response.answers[1].rr_type, dns.DNS_QTYPE_CNAME)
self.assertEquals(response.answers[1].name, name2)
self.assertEquals(response.answers[1].rdata, name0)
self.assertEquals(response.answers[2].rr_type, dns.DNS_QTYPE_A)
self.assertEquals(response.answers[2].rdata,
self.server_ip)
def test_invalid_empty_cname(self):
name0 = "cnamedotprefix0.%s" % self.get_dns_domain()
try:
self.make_dns_update(name0, "", dns.DNS_QTYPE_CNAME)
except AssertionError:
pass
else:
self.fail("Successfully added empty CNAME, which is invalid.")
def test_cname_two_chain_not_matching_qtype(self):
name0 = "cnamechain0.%s" % self.get_dns_domain()
name1 = "cnamechain1.%s" % self.get_dns_domain()
name2 = "cnamechain2.%s" % self.get_dns_domain()
self.make_dns_update(name1, name2, dns.DNS_QTYPE_CNAME)
self.make_dns_update(name2, name0, dns.DNS_QTYPE_CNAME)
self.make_dns_update(name0, server_ip, dns.DNS_QTYPE_A)
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
q = self.make_name_question(name1, dns.DNS_QTYPE_TXT,
dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
# CNAME should return all intermediate results!
# Only the A records exists, not the TXT.
self.assertEquals(response.ancount, 2)
self.assertEquals(response.answers[0].rr_type, dns.DNS_QTYPE_CNAME)
self.assertEquals(response.answers[0].name, name1)
self.assertEquals(response.answers[0].rdata, name2)
self.assertEquals(response.answers[1].rr_type, dns.DNS_QTYPE_CNAME)
self.assertEquals(response.answers[1].name, name2)
self.assertEquals(response.answers[1].rdata, name0)
class TestInvalidQueries(DNSTest):
def setUp(self):
super(TestInvalidQueries, self).setUp()
global server, server_ip, lp, creds, timeout
self.server = server_name
self.server_ip = server_ip
self.lp = lp
self.creds = creds
self.timeout = timeout
def test_one_a_query(self):
"""send 0 bytes follows by create a query packet
containing one query record"""
s = None
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
s.connect((self.server_ip, 53))
s.send("", 0)
finally:
if s is not None:
s.close()
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
name = "%s.%s" % (self.server, self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=self.server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rdata,
self.server_ip)
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 = []
name = "%s.%s" % ('fakefakefake', self.get_dns_domain())
q = self.make_name_question(name, dns.DNS_QTYPE_A, dns.DNS_QCLASS_IN)
print("asking for ", q.name)
questions.append(q)
self.finish_name_packet(p, questions)
p.operation |= dns.DNS_FLAG_REPLY
s = None
try:
send_packet = ndr.ndr_pack(p)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
s.settimeout(timeout)
host = self.server_ip
s.connect((host, 53))
tcp_packet = struct.pack('!H', len(send_packet))
tcp_packet += send_packet
s.send(tcp_packet, 0)
recv_packet = s.recv(0xffff + 2, 0)
self.assertEquals(0, len(recv_packet))
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()
class TestZones(DNSTest):
def setUp(self):
super(TestZones, self).setUp()
global server, server_ip, lp, creds, timeout
self.server = server_name
self.server_ip = server_ip
self.lp = lp
self.creds = creds
self.timeout = timeout
self.zone = "test.lan"
self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" %
(self.server_ip),
self.lp, self.creds)
self.samdb = SamDB(url="ldap://" + self.server_ip,
lp=self.get_loadparm(),
session_info=system_session(),
credentials=self.creds)
self.zone_dn = "DC=" + self.zone +\
",CN=MicrosoftDNS,DC=DomainDNSZones," +\
str(self.samdb.get_default_basedn())
def tearDown(self):
super(TestZones, self).tearDown()
try:
self.delete_zone(self.zone)
except RuntimeError as e:
(num, string) = e.args
if num != werror.WERR_DNS_ERROR_ZONE_DOES_NOT_EXIST:
raise
def create_zone(self, zone, aging_enabled=False):
zone_create = dnsserver.DNS_RPC_ZONE_CREATE_INFO_LONGHORN()
zone_create.pszZoneName = zone
zone_create.dwZoneType = dnsp.DNS_ZONE_TYPE_PRIMARY
zone_create.fAging = int(aging_enabled)
zone_create.dwDpFlags = dnsserver.DNS_DP_DOMAIN_DEFAULT
zone_create.fDsIntegrated = 1
zone_create.fLoadExisting = 1
zone_create.fAllowUpdate = dnsp.DNS_ZONE_UPDATE_UNSECURE
try:
client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
self.rpc_conn.DnssrvOperation2(client_version,
0,
self.server_ip,
None,
0,
'ZoneCreate',
dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
zone_create)
except WERRORError as e:
self.fail(str(e))
def set_params(self, **kwargs):
zone = kwargs.pop('zone', None)
for key, val in kwargs.items():
name_param = dnsserver.DNS_RPC_NAME_AND_PARAM()
name_param.dwParam = val
name_param.pszNodeName = key
client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
nap_type = dnsserver.DNSSRV_TYPEID_NAME_AND_PARAM
try:
self.rpc_conn.DnssrvOperation2(client_version,
0,
self.server,
zone,
0,
'ResetDwordProperty',
nap_type,
name_param)
except WERRORError as e:
self.fail(str(e))
def ldap_modify_dnsrecs(self, name, func):
dn = 'DC={0},{1}'.format(name, self.zone_dn)
dns_recs = self.ldap_get_dns_records(name)
for rec in dns_recs:
func(rec)
update_dict = {'dn': dn, 'dnsRecord': [ndr_pack(r) for r in dns_recs]}
self.samdb.modify(ldb.Message.from_dict(self.samdb,
update_dict,
ldb.FLAG_MOD_REPLACE))
def dns_update_record(self, prefix, txt):
p = self.make_txt_update(prefix, txt, self.zone)
(code, response) = self.dns_transaction_udp(p, host=self.server_ip)
self.assert_dns_rcode_equals(code, dns.DNS_RCODE_OK)
recs = self.ldap_get_dns_records(prefix)
recs = [r for r in recs if r.data.str == txt]
self.assertEqual(len(recs), 1)
return recs[0]
def dns_tombstone(self, prefix, txt, zone):
name = prefix + "." + zone
to = dnsp.DnssrvRpcRecord()
to.dwTimeStamp = 1000
to.wType = dnsp.DNS_TYPE_TOMBSTONE
self.samdb.dns_replace(name, [to])
def ldap_get_records(self, name):
# The use of SCOPE_SUBTREE here avoids raising an exception in the
# 0 results case for a test below.
expr = "(&(objectClass=dnsNode)(name={0}))".format(name)
return self.samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
def ldap_get_dns_records(self, name):
records = self.ldap_get_records(name)
return [ndr_unpack(dnsp.DnssrvRpcRecord, r)
for r in records[0].get('dnsRecord')]
def ldap_get_zone_settings(self):
records = self.samdb.search(base=self.zone_dn, scope=ldb.SCOPE_BASE,
expression="(&(objectClass=dnsZone)" +
"(name={0}))".format(self.zone),
attrs=["dNSProperty"])
self.assertEqual(len(records), 1)
props = [ndr_unpack(dnsp.DnsProperty, r)
for r in records[0].get('dNSProperty')]
# We have no choice but to repeat these here.
zone_prop_ids = {0x00: "EMPTY",
0x01: "TYPE",
0x02: "ALLOW_UPDATE",
0x08: "SECURE_TIME",
0x10: "NOREFRESH_INTERVAL",
0x11: "SCAVENGING_SERVERS",
0x12: "AGING_ENABLED_TIME",
0x20: "REFRESH_INTERVAL",
0x40: "AGING_STATE",
0x80: "DELETED_FROM_HOSTNAME",
0x81: "MASTER_SERVERS",
0x82: "AUTO_NS_SERVERS",
0x83: "DCPROMO_CONVERT",
0x90: "SCAVENGING_SERVERS_DA",
0x91: "MASTER_SERVERS_DA",
0x92: "NS_SERVERS_DA",
0x100: "NODE_DBFLAGS"}
return {zone_prop_ids[p.id].lower(): p.data for p in props}
def set_aging(self, enable=False):
self.create_zone(self.zone, aging_enabled=enable)
self.set_params(NoRefreshInterval=1, RefreshInterval=1,
Aging=int(bool(enable)), zone=self.zone,
AllowUpdate=dnsp.DNS_ZONE_UPDATE_UNSECURE)
def test_set_aging(self, enable=True, name='agingtest', txt=['test txt']):
self.set_aging(enable=True)
settings = self.ldap_get_zone_settings()
self.assertTrue(settings['aging_state'] is not None)
self.assertTrue(settings['aging_state'])
rec = self.dns_update_record('agingtest', ['test txt'])
self.assertNotEqual(rec.dwTimeStamp, 0)
def test_set_aging_disabled(self):
self.set_aging(enable=False)
settings = self.ldap_get_zone_settings()
self.assertTrue(settings['aging_state'] is not None)
self.assertFalse(settings['aging_state'])
rec = self.dns_update_record('agingtest', ['test txt'])
self.assertNotEqual(rec.dwTimeStamp, 0)
def test_aging_update(self, enable=True):
name, txt = 'agingtest', ['test txt']
self.set_aging(enable=True)
before_mod = self.dns_update_record(name, txt)
if not enable:
self.set_params(zone=self.zone, Aging=0)
dec = 2
def mod_ts(rec):
self.assertTrue(rec.dwTimeStamp > 0)
rec.dwTimeStamp -= dec
self.ldap_modify_dnsrecs(name, mod_ts)
after_mod = self.ldap_get_dns_records(name)
self.assertEqual(len(after_mod), 1)
after_mod = after_mod[0]
self.assertEqual(after_mod.dwTimeStamp,
before_mod.dwTimeStamp - dec)
after_update = self.dns_update_record(name, txt)
after_should_equal = before_mod if enable else after_mod
self.assertEqual(after_should_equal.dwTimeStamp,
after_update.dwTimeStamp)
def test_aging_update_disabled(self):
self.test_aging_update(enable=False)
def test_aging_refresh(self):
name, txt = 'agingtest', ['test txt']
self.create_zone(self.zone, aging_enabled=True)
interval = 10
self.set_params(NoRefreshInterval=interval, RefreshInterval=interval,
Aging=1, zone=self.zone,
AllowUpdate=dnsp.DNS_ZONE_UPDATE_UNSECURE)
before_mod = self.dns_update_record(name, txt)
def mod_ts(rec):
self.assertTrue(rec.dwTimeStamp > 0)
rec.dwTimeStamp -= interval / 2
self.ldap_modify_dnsrecs(name, mod_ts)
update_during_norefresh = self.dns_update_record(name, txt)
def mod_ts(rec):
self.assertTrue(rec.dwTimeStamp > 0)
rec.dwTimeStamp -= interval + interval / 2
self.ldap_modify_dnsrecs(name, mod_ts)
update_during_refresh = self.dns_update_record(name, txt)
self.assertEqual(update_during_norefresh.dwTimeStamp,
before_mod.dwTimeStamp - interval / 2)
self.assertEqual(update_during_refresh.dwTimeStamp,
before_mod.dwTimeStamp)
def test_rpc_add_no_timestamp(self):
name, txt = 'agingtest', ['test txt']
self.set_aging(enable=True)
rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
rec_buf.rec = TXTRecord(txt)
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.zone,
name,
rec_buf,
None)
recs = self.ldap_get_dns_records(name)
self.assertEqual(len(recs), 1)
self.assertEqual(recs[0].dwTimeStamp, 0)
def test_static_record_dynamic_update(self):
name, txt = 'agingtest', ['test txt']
txt2 = ['test txt2']
self.set_aging(enable=True)
rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
rec_buf.rec = TXTRecord(txt)
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.zone,
name,
rec_buf,
None)
rec2 = self.dns_update_record(name, txt2)
self.assertEqual(rec2.dwTimeStamp, 0)
def test_dynamic_record_static_update(self):
name, txt = 'agingtest', ['test txt']
txt2 = ['test txt2']
txt3 = ['test txt3']
self.set_aging(enable=True)
self.dns_update_record(name, txt)
rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
rec_buf.rec = TXTRecord(txt2)
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.zone,
name,
rec_buf,
None)
self.dns_update_record(name, txt3)
recs = self.ldap_get_dns_records(name)
# Put in dict because ldap recs might be out of order
recs = {str(r.data.str): r for r in recs}
self.assertNotEqual(recs[str(txt)].dwTimeStamp, 0)
self.assertEqual(recs[str(txt2)].dwTimeStamp, 0)
self.assertEqual(recs[str(txt3)].dwTimeStamp, 0)
def test_dns_tombstone_custom_match_rule(self):
lp = self.get_loadparm()
self.samdb = SamDB(url=lp.samdb_url(), lp=lp,
session_info=system_session(),
credentials=self.creds)
name, txt = 'agingtest', ['test txt']
name2, txt2 = 'agingtest2', ['test txt2']
name3, txt3 = 'agingtest3', ['test txt3']
name4, txt4 = 'agingtest4', ['test txt4']
name5, txt5 = 'agingtest5', ['test txt5']
self.create_zone(self.zone, aging_enabled=True)
interval = 10
self.set_params(NoRefreshInterval=interval, RefreshInterval=interval,
Aging=1, zone=self.zone,
AllowUpdate=dnsp.DNS_ZONE_UPDATE_UNSECURE)
self.dns_update_record(name, txt)
self.dns_update_record(name2, txt)
self.dns_update_record(name2, txt2)
self.dns_update_record(name3, txt)
self.dns_update_record(name3, txt2)
last_update = self.dns_update_record(name3, txt3)
# Modify txt1 of the first 2 names
def mod_ts(rec):
if rec.data.str == txt:
rec.dwTimeStamp -= 2
self.ldap_modify_dnsrecs(name, mod_ts)
self.ldap_modify_dnsrecs(name2, mod_ts)
# create a static dns record.
rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
rec_buf.rec = TXTRecord(txt4)
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.zone,
name4,
rec_buf,
None)
# Create a tomb stoned record.
self.dns_update_record(name5, txt5)
self.dns_tombstone(name5, txt5, self.zone)
self.ldap_get_dns_records(name3)
expr = "(dnsRecord:1.3.6.1.4.1.7165.4.5.3:={0})"
expr = expr.format(int(last_update.dwTimeStamp) - 1)
try:
res = self.samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
except ldb.LdbError as e:
self.fail(str(e))
updated_names = {str(r.get('name')) for r in res}
self.assertEqual(updated_names, set([name, name2]))
def test_dns_tombstone_custom_match_rule_no_records(self):
lp = self.get_loadparm()
self.samdb = SamDB(url=lp.samdb_url(), lp=lp,
session_info=system_session(),
credentials=self.creds)
self.create_zone(self.zone, aging_enabled=True)
interval = 10
self.set_params(NoRefreshInterval=interval, RefreshInterval=interval,
Aging=1, zone=self.zone,
AllowUpdate=dnsp.DNS_ZONE_UPDATE_UNSECURE)
expr = "(dnsRecord:1.3.6.1.4.1.7165.4.5.3:={0})"
expr = expr.format(1)
try:
res = self.samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
except ldb.LdbError as e:
self.fail(str(e))
self.assertEqual(0, len(res))
def test_dns_tombstone_custom_match_rule_fail(self):
self.create_zone(self.zone, aging_enabled=True)
samdb = SamDB(url=lp.samdb_url(),
lp=lp,
session_info=system_session(),
credentials=self.creds)
# Property name in not dnsRecord
expr = "(dnsProperty:1.3.6.1.4.1.7165.4.5.3:=1)"
res = samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
self.assertEquals(len(res), 0)
# No value for tombstone time
try:
expr = "(dnsRecord:1.3.6.1.4.1.7165.4.5.3:=)"
res = samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
self.assertEquals(len(res), 0)
self.fail("Exception: ldb.ldbError not generated")
except ldb.LdbError as e:
(num, msg) = e.args
self.assertEquals(num, ERR_OPERATIONS_ERROR)
# Tombstone time = -
try:
expr = "(dnsRecord:1.3.6.1.4.1.7165.4.5.3:=-)"
res = samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
self.assertEquals(len(res), 0)
self.fail("Exception: ldb.ldbError not generated")
except ldb.LdbError as e:
(num, _) = e.args
self.assertEquals(num, ERR_OPERATIONS_ERROR)
# Tombstone time longer than 64 characters
try:
expr = "(dnsRecord:1.3.6.1.4.1.7165.4.5.3:={0})"
expr = expr.format("1" * 65)
res = samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
self.assertEquals(len(res), 0)
self.fail("Exception: ldb.ldbError not generated")
except ldb.LdbError as e:
(num, _) = e.args
self.assertEquals(num, ERR_OPERATIONS_ERROR)
# Non numeric Tombstone time
try:
expr = "(dnsRecord:1.3.6.1.4.1.7165.4.5.3:=expired)"
res = samdb.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
self.assertEquals(len(res), 0)
self.fail("Exception: ldb.ldbError not generated")
except ldb.LdbError as e:
(num, _) = e.args
self.assertEquals(num, ERR_OPERATIONS_ERROR)
# Non system session
try:
db = SamDB(url="ldap://" + self.server_ip,
lp=self.get_loadparm(),
credentials=self.creds)
expr = "(dnsRecord:1.3.6.1.4.1.7165.4.5.3:=2)"
res = db.search(base=self.zone_dn, scope=ldb.SCOPE_SUBTREE,
expression=expr, attrs=["*"])
self.assertEquals(len(res), 0)
self.fail("Exception: ldb.ldbError not generated")
except ldb.LdbError as e:
(num, _) = e.args
self.assertEquals(num, ERR_OPERATIONS_ERROR)
def test_basic_scavenging(self):
lp = self.get_loadparm()
self.samdb = SamDB(url=lp.samdb_url(), lp=lp,
session_info=system_session(),
credentials=self.creds)
self.create_zone(self.zone, aging_enabled=True)
interval = 1
self.set_params(NoRefreshInterval=interval, RefreshInterval=interval,
zone=self.zone, Aging=1,
AllowUpdate=dnsp.DNS_ZONE_UPDATE_UNSECURE)
name, txt = 'agingtest', ['test txt']
name2, txt2 = 'agingtest2', ['test txt2']
name3, txt3 = 'agingtest3', ['test txt3']
self.dns_update_record(name, txt)
self.dns_update_record(name2, txt)
self.dns_update_record(name2, txt2)
self.dns_update_record(name3, txt)
self.dns_update_record(name3, txt2)
last_add = self.dns_update_record(name3, txt3)
def mod_ts(rec):
self.assertTrue(rec.dwTimeStamp > 0)
if rec.data.str == txt:
rec.dwTimeStamp -= interval * 5
self.ldap_modify_dnsrecs(name, mod_ts)
self.ldap_modify_dnsrecs(name2, mod_ts)
self.ldap_modify_dnsrecs(name3, mod_ts)
self.assertTrue(callable(getattr(dsdb, '_scavenge_dns_records', None)))
dsdb._scavenge_dns_records(self.samdb)
recs = self.ldap_get_dns_records(name)
self.assertEqual(len(recs), 1)
self.assertEqual(recs[0].wType, dnsp.DNS_TYPE_TOMBSTONE)
recs = self.ldap_get_dns_records(name2)
self.assertEqual(len(recs), 1)
self.assertEqual(recs[0].wType, dnsp.DNS_TYPE_TXT)
self.assertEqual(recs[0].data.str, txt2)
recs = self.ldap_get_dns_records(name3)
self.assertEqual(len(recs), 2)
txts = {str(r.data.str) for r in recs}
self.assertEqual(txts, {str(txt2), str(txt3)})
self.assertEqual(recs[0].wType, dnsp.DNS_TYPE_TXT)
self.assertEqual(recs[1].wType, dnsp.DNS_TYPE_TXT)
for make_it_work in [False, True]:
inc = -1 if make_it_work else 1
def mod_ts(rec):
rec.data = (last_add.dwTimeStamp - 24 * 14) + inc
self.ldap_modify_dnsrecs(name, mod_ts)
dsdb._dns_delete_tombstones(self.samdb)
recs = self.ldap_get_records(name)
if make_it_work:
self.assertEqual(len(recs), 0)
else:
self.assertEqual(len(recs), 1)
def delete_zone(self, zone):
self.rpc_conn.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
zone,
0,
'DeleteZoneFromDs',
dnsserver.DNSSRV_TYPEID_NULL,
None)
def test_soa_query(self):
zone = "test.lan"
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
q = self.make_name_question(zone, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
# 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)
self.create_zone(zone)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assert_dns_opcode_equals(response, dns.DNS_OPCODE_QUERY)
self.assertEquals(response.ancount, 1)
self.assertEquals(response.answers[0].rr_type, dns.DNS_QTYPE_SOA)
self.delete_zone(zone)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
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)
class TestRPCRoundtrip(DNSTest):
def setUp(self):
super(TestRPCRoundtrip, self).setUp()
global server, server_ip, lp, creds
self.server = server_name
self.server_ip = server_ip
self.lp = lp
self.creds = creds
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)
except WERRORError as e:
self.fail(str(e))
try:
self.check_query_txt(prefix, txt)
finally:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
None,
add_rec_buf)
def test_update_add_null_padded_txt_record(self):
"test adding records works"
prefix, txt = 'pad1textrec', ['"This is a test"', '', '']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, txt)
self.assertIsNotNone(
dns_record_match(self.rpc_conn,
self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT,
'"\\"This is a test\\"" "" ""'))
prefix, txt = 'pad2textrec', ['"This is a test"', '', '', 'more text']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, txt)
self.assertIsNotNone(
dns_record_match(
self.rpc_conn,
self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT,
'"\\"This is a test\\"" "" "" "more text"'))
prefix, txt = 'pad3textrec', ['', '', '"This is a test"']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, txt)
self.assertIsNotNone(
dns_record_match(
self.rpc_conn,
self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT,
'"" "" "\\"This is a test\\""'))
def test_update_add_padding_rpc_to_dns(self):
prefix, txt = 'pad1textrec', ['"This is a test"', '', '']
prefix = 'rpc' + prefix
name = "%s.%s" % (prefix, self.get_dns_domain())
rec = data_to_dns_record(dnsp.DNS_TYPE_TXT,
'"\\"This is a test\\"" "" ""')
add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
add_rec_buf.rec = rec
try:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
add_rec_buf,
None)
except WERRORError as e:
self.fail(str(e))
try:
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)
except WERRORError as e:
self.fail(str(e))
try:
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)
except WERRORError as e:
self.fail(str(e))
try:
self.check_query_txt(prefix, txt)
finally:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
None,
add_rec_buf)
# Test is incomplete due to strlen against txt records
def test_update_add_null_char_txt_record(self):
"test adding records works"
prefix, txt = 'nulltextrec', ['NULL\x00BYTE']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, ['NULL'])
self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT, '"NULL"'))
prefix, txt = 'nulltextrec2', ['NULL\x00BYTE', 'NULL\x00BYTE']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, ['NULL', 'NULL'])
self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT, '"NULL" "NULL"'))
def test_update_add_null_char_rpc_to_dns(self):
prefix = 'rpcnulltextrec'
name = "%s.%s" % (prefix, self.get_dns_domain())
rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"NULL\x00BYTE"')
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)
except WERRORError as e:
self.fail(str(e))
try:
self.check_query_txt(prefix, ['NULL'])
finally:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
None,
add_rec_buf)
def test_update_add_hex_char_txt_record(self):
"test adding records works"
prefix, txt = 'hextextrec', ['HIGH\xFFBYTE']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, txt)
self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT, '"HIGH\xFFBYTE"'))
def test_update_add_hex_rpc_to_dns(self):
prefix, txt = 'hextextrec', ['HIGH\xFFBYTE']
prefix = 'rpc' + prefix
name = "%s.%s" % (prefix, self.get_dns_domain())
rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"HIGH\xFFBYTE"')
add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
add_rec_buf.rec = rec
try:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
add_rec_buf,
None)
except WERRORError as e:
self.fail(str(e))
try:
self.check_query_txt(prefix, txt)
finally:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
None,
add_rec_buf)
def test_update_add_slash_txt_record(self):
"test adding records works"
prefix, txt = 'slashtextrec', ['Th\\=is=is a test']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, txt)
self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT, '"Th\\\\=is=is a test"'))
# This test fails against Windows as it eliminates slashes in RPC
# One typical use for a slash is in records like 'var=value' to
# escape '=' characters.
def test_update_add_slash_rpc_to_dns(self):
prefix, txt = 'slashtextrec', ['Th\\=is=is a test']
prefix = 'rpc' + prefix
name = "%s.%s" % (prefix, self.get_dns_domain())
rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '"Th\\\\=is=is a test"')
add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
add_rec_buf.rec = rec
try:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
add_rec_buf,
None)
except WERRORError as e:
self.fail(str(e))
try:
self.check_query_txt(prefix, txt)
finally:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
None,
add_rec_buf)
def test_update_add_two_txt_records(self):
"test adding two txt records works"
prefix, txt = 'textrec2', ['"This is a test"',
'"and this is a test, too"']
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, txt)
self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT, '"\\"This is a test\\""' +
' "\\"and this is a test, too\\""'))
def test_update_add_two_rpc_to_dns(self):
prefix, txt = 'textrec2', ['"This is a test"',
'"and this is a test, too"']
prefix = 'rpc' + prefix
name = "%s.%s" % (prefix, self.get_dns_domain())
rec = data_to_dns_record(dnsp.DNS_TYPE_TXT,
'"\\"This is a test\\""' +
' "\\"and this is a test, too\\""')
add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
add_rec_buf.rec = rec
try:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
add_rec_buf,
None)
except WERRORError as e:
self.fail(str(e))
try:
self.check_query_txt(prefix, txt)
finally:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
None,
add_rec_buf)
def test_update_add_empty_txt_records(self):
"test adding two txt records works"
prefix, txt = 'emptytextrec', []
p = self.make_txt_update(prefix, txt)
(response, response_packet) =\
self.dns_transaction_udp(p, host=server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.check_query_txt(prefix, txt)
self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server_ip,
self.get_dns_domain(),
"%s.%s" % (prefix, self.get_dns_domain()),
dnsp.DNS_TYPE_TXT, ''))
def test_update_add_empty_rpc_to_dns(self):
prefix, txt = 'rpcemptytextrec', []
name = "%s.%s" % (prefix, self.get_dns_domain())
rec = data_to_dns_record(dnsp.DNS_TYPE_TXT, '')
add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
add_rec_buf.rec = rec
try:
self.rpc_conn.DnssrvUpdateRecord2(
dnsserver.DNS_CLIENT_VERSION_LONGHORN,
0,
self.server_ip,
self.get_dns_domain(),
name,
add_rec_buf,
None)
except WERRORError as e:
self.fail(str(e))
try:
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)
TestProgram(module=__name__, opts=subunitopts)
View Git Blame Copy Permalink