# Copyright (C) 2012 Internet Systems Consortium.
#
# Permission to use, copy, modify, and distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM
# DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
# INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT,
# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
# FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
# WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
import os
import shutil
import isc.log
import unittest
from isc.dns import *
from isc.datasrc import DataSourceClient
from isc.ddns.session import *
from isc.ddns.zone_config import *
# Some common test parameters
TESTDATA_PATH = os.environ['TESTDATA_PATH'] + os.sep
READ_ZONE_DB_FILE = TESTDATA_PATH + "rwtest.sqlite3" # original, to be copied
TESTDATA_WRITE_PATH = os.environ['TESTDATA_WRITE_PATH'] + os.sep
WRITE_ZONE_DB_FILE = TESTDATA_WRITE_PATH + "rwtest.sqlite3.copied"
WRITE_ZONE_DB_CONFIG = "{ \"database_file\": \"" + WRITE_ZONE_DB_FILE + "\"}"
TEST_ZONE_NAME = Name('example.org')
UPDATE_RRTYPE = RRType.SOA()
TEST_RRCLASS = RRClass.IN()
TEST_ZONE_RECORD = Question(TEST_ZONE_NAME, TEST_RRCLASS, UPDATE_RRTYPE)
TEST_CLIENT6 = ('2001:db8::1', 53, 0, 0)
TEST_CLIENT4 = ('192.0.2.1', 53)
# TSIG key for tests when needed. The key name is TEST_ZONE_NAME.
TEST_TSIG_KEY = TSIGKey("example.org:SFuWd/q99SzF8Yzd1QbB9g==")
def create_update_msg(zones=[TEST_ZONE_RECORD], prerequisites=[],
updates=[], tsig_key=None):
msg = Message(Message.RENDER)
msg.set_qid(5353) # arbitrary chosen
msg.set_opcode(Opcode.UPDATE())
msg.set_rcode(Rcode.NOERROR())
for z in zones:
msg.add_question(z)
for p in prerequisites:
msg.add_rrset(SECTION_PREREQUISITE, p)
for u in updates:
msg.add_rrset(SECTION_UPDATE, u)
renderer = MessageRenderer()
if tsig_key is not None:
msg.to_wire(renderer, TSIGContext(tsig_key))
else:
msg.to_wire(renderer)
# re-read the created data in the parse mode
msg.clear(Message.PARSE)
msg.from_wire(renderer.get_data(), Message.PRESERVE_ORDER)
return msg
def add_rdata(rrset, rdata):
'''
Helper function for easily adding Rdata fields to RRsets.
This function assumes the given rdata is of type string or bytes,
and corresponds to the given rrset
'''
rrset.add_rdata(isc.dns.Rdata(rrset.get_type(),
rrset.get_class(),
rdata))
def create_rrset(name, rrclass, rrtype, ttl, rdatas = []):
'''
Helper method to easily create RRsets, auto-converts
name, rrclass, rrtype, and ttl (if possibly through their
respective constructors)
rdatas is a list of rr data strings, or bytestrings, which
should match the RRType of the rrset to create
'''
if type(name) != Name:
name = Name(name)
if type(rrclass) != RRClass:
rrclass = RRClass(rrclass)
if type(rrtype) != RRType:
rrtype = RRType(rrtype)
if type(ttl) != RRTTL:
ttl = RRTTL(ttl)
rrset = isc.dns.RRset(name, rrclass, rrtype, ttl)
for rdata in rdatas:
add_rdata(rrset, rdata)
return rrset
class SessionTestBase(unittest.TestCase):
'''Base class for all sesion related tests.
It just initializes common test parameters in its setUp() and defines
some common utility method(s).
'''
def setUp(self):
shutil.copyfile(READ_ZONE_DB_FILE, WRITE_ZONE_DB_FILE)
self._datasrc_client = DataSourceClient("sqlite3",
WRITE_ZONE_DB_CONFIG)
self._update_msg = create_update_msg()
self._acl_map = {(TEST_ZONE_NAME, TEST_RRCLASS):
REQUEST_LOADER.load([{"action": "ACCEPT"}])}
self._session = UpdateSession(self._update_msg, TEST_CLIENT4,
ZoneConfig([], TEST_RRCLASS,
self._datasrc_client,
self._acl_map))
self._session._UpdateSession__get_update_zone()
def check_response(self, msg, expected_rcode):
'''Perform common checks on update resposne message.'''
self.assertTrue(msg.get_header_flag(Message.HEADERFLAG_QR))
# note: we convert opcode to text it'd be more helpful on failure.
self.assertEqual(Opcode.UPDATE().to_text(), msg.get_opcode().to_text())
self.assertEqual(expected_rcode.to_text(), msg.get_rcode().to_text())
# All sections should be cleared
self.assertEqual(0, msg.get_rr_count(SECTION_ZONE))
self.assertEqual(0, msg.get_rr_count(SECTION_PREREQUISITE))
self.assertEqual(0, msg.get_rr_count(SECTION_UPDATE))
self.assertEqual(0, msg.get_rr_count(Message.SECTION_ADDITIONAL))
class SessionTest(SessionTestBase):
'''Basic session tests'''
def test_handle(self):
'''Basic update case'''
result, zname, zclass = self._session.handle()
self.assertEqual(UPDATE_SUCCESS, result)
self.assertEqual(TEST_ZONE_NAME, zname)
self.assertEqual(TEST_RRCLASS, zclass)
# Just checking these are different from the success code.
self.assertNotEqual(UPDATE_ERROR, result)
self.assertNotEqual(UPDATE_DROP, result)
def test_broken_request(self):
# Zone section is empty
msg = create_update_msg(zones=[])
session = UpdateSession(msg, TEST_CLIENT6, None)
result, zname, zclass = session.handle()
self.assertEqual(UPDATE_ERROR, result)
self.assertEqual(None, zname)
self.assertEqual(None, zclass)
self.check_response(session.get_message(), Rcode.FORMERR())
# Zone section contains multiple records
msg = create_update_msg(zones=[TEST_ZONE_RECORD, TEST_ZONE_RECORD])
session = UpdateSession(msg, TEST_CLIENT4, None)
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.FORMERR())
# Zone section's type is not SOA
msg = create_update_msg(zones=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
RRType.A())])
session = UpdateSession(msg, TEST_CLIENT4, None)
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.FORMERR())
def test_update_secondary(self):
# specified zone is configured as a secondary. Since this
# implementation doesn't support update forwarding, the result
# should be NOTIMP.
msg = create_update_msg(zones=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
RRType.SOA())])
session = UpdateSession(msg, TEST_CLIENT4,
ZoneConfig([(TEST_ZONE_NAME, TEST_RRCLASS)],
TEST_RRCLASS, self._datasrc_client))
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.NOTIMP())
def check_notauth(self, zname, zclass=TEST_RRCLASS):
'''Common test sequence for the 'notauth' test'''
msg = create_update_msg(zones=[Question(zname, zclass, RRType.SOA())])
session = UpdateSession(msg, TEST_CLIENT4,
ZoneConfig([(TEST_ZONE_NAME, TEST_RRCLASS)],
TEST_RRCLASS, self._datasrc_client))
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.NOTAUTH())
def test_update_notauth(self):
'''Update attempt for non authoritative zones'''
# zone name doesn't match
self.check_notauth(Name('example.com'))
# zone name is a subdomain of the actual authoritative zone
# (match must be exact)
self.check_notauth(Name('sub.example.org'))
# zone class doesn't match
self.check_notauth(Name('example.org'), RRClass.CH())
def test_update_datasrc_error(self):
# if the data source client raises an exception, it should result in
# a SERVFAIL.
class BadDataSourceClient:
def find_zone(self, name):
raise isc.datasrc.Error('faked exception')
msg = create_update_msg(zones=[Question(TEST_ZONE_NAME, TEST_RRCLASS,
RRType.SOA())])
session = UpdateSession(msg, TEST_CLIENT4,
ZoneConfig([(TEST_ZONE_NAME, TEST_RRCLASS)],
TEST_RRCLASS,
BadDataSourceClient()))
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.SERVFAIL())
def test_foreach_rr_in_rrset(self):
rrset = create_rrset("www.example.org", TEST_RRCLASS,
RRType.A(), 3600, [ "192.0.2.1" ])
l = []
for rr in foreach_rr(rrset):
l.append(str(rr))
self.assertEqual(["www.example.org. 3600 IN A 192.0.2.1\n"], l)
add_rdata(rrset, "192.0.2.2")
add_rdata(rrset, "192.0.2.3")
# but through the generator, there should be several 1-line entries
l = []
for rr in foreach_rr(rrset):
l.append(str(rr))
self.assertEqual(["www.example.org. 3600 IN A 192.0.2.1\n",
"www.example.org. 3600 IN A 192.0.2.2\n",
"www.example.org. 3600 IN A 192.0.2.3\n",
], l)
def test_convert_rrset_class(self):
# Converting an RRSET to a different class should work
# if the rdata types can be converted
rrset = create_rrset("www.example.org", RRClass.NONE(), RRType.A(),
3600, [ b'\xc0\x00\x02\x01', b'\xc0\x00\x02\x02'])
rrset2 = convert_rrset_class(rrset, RRClass.IN())
self.assertEqual("www.example.org. 3600 IN A 192.0.2.1\n" +
"www.example.org. 3600 IN A 192.0.2.2\n",
str(rrset2))
rrset3 = convert_rrset_class(rrset2, RRClass.NONE())
self.assertEqual("www.example.org. 3600 CLASS254 A \\# 4 " +
"c0000201\nwww.example.org. 3600 CLASS254 " +
"A \\# 4 c0000202\n",
str(rrset3))
# depending on what type of bad data is given, a number
# of different exceptions could be raised (TODO: i recall
# there was a ticket about making a better hierarchy for
# dns/parsing related exceptions)
self.assertRaises(InvalidRdataLength, convert_rrset_class,
rrset, RRClass.CH())
add_rdata(rrset, b'\xc0\x00')
self.assertRaises(DNSMessageFORMERR, convert_rrset_class,
rrset, RRClass.IN())
def test_collect_rrsets(self):
'''
Tests the 'rrset collector' method, which collects rrsets
with the same name and type
'''
collected = []
collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
RRType.A(), 0, [ "192.0.2.1" ]))
# Same name and class, different type
collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
RRType.TXT(), 0, [ "one" ]))
collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
RRType.A(), 0, [ "192.0.2.2" ]))
collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
RRType.TXT(), 0, [ "two" ]))
# Same class and type as an existing one, different name
collect_rrsets(collected, create_rrset("b.example.org", RRClass.IN(),
RRType.A(), 0, [ "192.0.2.3" ]))
# Same name and type as an existing one, different class
collect_rrsets(collected, create_rrset("a.example.org", RRClass.CH(),
RRType.TXT(), 0, [ "one" ]))
collect_rrsets(collected, create_rrset("b.example.org", RRClass.IN(),
RRType.A(), 0, [ "192.0.2.4" ]))
collect_rrsets(collected, create_rrset("a.example.org", RRClass.CH(),
RRType.TXT(), 0, [ "two" ]))
strings = [ rrset.to_text() for rrset in collected ]
# note + vs , in this list
expected = ['a.example.org. 0 IN A 192.0.2.1\n' +
'a.example.org. 0 IN A 192.0.2.2\n',
'a.example.org. 0 IN TXT "one"\n' +
'a.example.org. 0 IN TXT "two"\n',
'b.example.org. 0 IN A 192.0.2.3\n' +
'b.example.org. 0 IN A 192.0.2.4\n',
'a.example.org. 0 CH TXT "one"\n' +
'a.example.org. 0 CH TXT "two"\n']
self.assertEqual(expected, strings)
def __prereq_helper(self, method, expected, rrset):
'''Calls the given method with self._datasrc_client
and the given rrset, and compares the return value.
Function does not do much but makes the code look nicer'''
self.assertEqual(expected, method(rrset))
def __check_prerequisite_exists_combined(self, method, rrclass, expected):
'''shared code for the checks for the very similar (but reversed
in behaviour) methods __prereq_rrset_exists and
__prereq_rrset_does_not_exist.
For rrset_exists, rrclass should be ANY, for rrset_does_not_exist,
it should be NONE.
'''
# Basic existence checks
# www.example.org should have an A, but not an MX
rrset = create_rrset("www.example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, expected, rrset)
rrset = create_rrset("www.example.org", rrclass, RRType.MX(), 0)
self.__prereq_helper(method, not expected, rrset)
# example.org should have an MX, but not an A
rrset = create_rrset("example.org", rrclass, RRType.MX(), 0)
self.__prereq_helper(method, expected, rrset)
rrset = create_rrset("example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, not expected, rrset)
# Also check the case where the name does not even exist
rrset = create_rrset("doesnotexist.example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, not expected, rrset)
# Wildcard expansion should not be applied, but literal matches
# should work
rrset = create_rrset("foo.wildcard.example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, not expected, rrset)
rrset = create_rrset("*.wildcard.example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, expected, rrset)
# Likewise, CNAME directly should match, but what it points to should
# not
rrset = create_rrset("cname.example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, not expected, rrset)
rrset = create_rrset("cname.example.org", rrclass, RRType.CNAME(), 0)
self.__prereq_helper(method, expected, rrset)
# And also make sure a delegation (itself) is not treated as existing
# data
rrset = create_rrset("foo.sub.example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, not expected, rrset)
# But the delegation data itself should match
rrset = create_rrset("sub.example.org", rrclass, RRType.NS(), 0)
self.__prereq_helper(method, expected, rrset)
# As should glue
rrset = create_rrset("ns.sub.example.org", rrclass, RRType.A(), 0)
self.__prereq_helper(method, expected, rrset)
def test_check_prerequisite_exists(self):
method = self._session._UpdateSession__prereq_rrset_exists
self.__check_prerequisite_exists_combined(method,
RRClass.ANY(),
True)
def test_check_prerequisite_does_not_exist(self):
method = self._session._UpdateSession__prereq_rrset_does_not_exist
self.__check_prerequisite_exists_combined(method,
RRClass.NONE(),
False)
def test_check_prerequisite_exists_value(self):
method = self._session._UpdateSession__prereq_rrset_exists_value
rrset = create_rrset("www.example.org", RRClass.IN(), RRType.A(), 0)
# empty one should not match
self.__prereq_helper(method, False, rrset)
# When the rdata is added, it should match
add_rdata(rrset, "192.0.2.1")
self.__prereq_helper(method, True, rrset)
# But adding more should not
add_rdata(rrset, "192.0.2.2")
self.__prereq_helper(method, False, rrset)
# Also test one with more than one RR
rrset = create_rrset("example.org", RRClass.IN(), RRType.NS(), 0)
self.__prereq_helper(method, False, rrset)
add_rdata(rrset, "ns1.example.org.")
self.__prereq_helper(method, False, rrset)
add_rdata(rrset, "ns2.example.org")
self.__prereq_helper(method, False, rrset)
add_rdata(rrset, "ns3.example.org.")
self.__prereq_helper(method, True, rrset)
add_rdata(rrset, "ns4.example.org.")
self.__prereq_helper(method, False, rrset)
# Repeat that, but try a different order of Rdata addition
rrset = create_rrset("example.org", RRClass.IN(), RRType.NS(), 0)
self.__prereq_helper(method, False, rrset)
add_rdata(rrset, "ns3.example.org.")
self.__prereq_helper(method, False, rrset)
add_rdata(rrset, "ns2.example.org.")
self.__prereq_helper(method, False, rrset)
add_rdata(rrset, "ns1.example.org.")
self.__prereq_helper(method, True, rrset)
add_rdata(rrset, "ns4.example.org.")
self.__prereq_helper(method, False, rrset)
# and test one where the name does not even exist
rrset = create_rrset("doesnotexist.example.org", RRClass.IN(),
RRType.A(), 0, [ "192.0.2.1" ])
self.__prereq_helper(method, False, rrset)
def __check_prerequisite_name_in_use_combined(self, method, rrclass,
expected):
'''shared code for the checks for the very similar (but reversed
in behaviour) methods __prereq_name_in_use and
__prereq_name_not_in_use
'''
rrset = create_rrset("example.org", rrclass, RRType.ANY(), 0)
self.__prereq_helper(method, expected, rrset)
rrset = create_rrset("www.example.org", rrclass, RRType.ANY(), 0)
self.__prereq_helper(method, expected, rrset)
rrset = create_rrset("doesnotexist.example.org", rrclass,
RRType.ANY(), 0)
self.__prereq_helper(method, not expected, rrset)
rrset = create_rrset("belowdelegation.sub.example.org", rrclass,
RRType.ANY(), 0)
self.__prereq_helper(method, not expected, rrset)
rrset = create_rrset("foo.wildcard.example.org", rrclass,
RRType.ANY(), 0)
self.__prereq_helper(method, not expected, rrset)
# empty nonterminal should not match
rrset = create_rrset("nonterminal.example.org", rrclass,
RRType.ANY(), 0)
self.__prereq_helper(method, not expected, rrset)
rrset = create_rrset("empty.nonterminal.example.org", rrclass,
RRType.ANY(), 0)
self.__prereq_helper(method, expected, rrset)
def test_check_prerequisite_name_in_use(self):
method = self._session._UpdateSession__prereq_name_in_use
self.__check_prerequisite_name_in_use_combined(method,
RRClass.ANY(),
True)
def test_check_prerequisite_name_not_in_use(self):
method = self._session._UpdateSession__prereq_name_not_in_use
self.__check_prerequisite_name_in_use_combined(method,
RRClass.NONE(),
False)
def check_prerequisite_result(self, expected, prerequisites):
'''Helper method for checking the result of a prerequisite check;
creates an update session, and fills it with the list of rrsets
from 'prerequisites'. Then checks if __check_prerequisites()
returns the Rcode specified in 'expected'.'''
msg = create_update_msg([TEST_ZONE_RECORD], prerequisites)
zconfig = ZoneConfig([], TEST_RRCLASS, self._datasrc_client,
self._acl_map)
session = UpdateSession(msg, TEST_CLIENT4, zconfig)
session._UpdateSession__get_update_zone()
# compare the to_text output of the rcodes (nicer error messages)
# This call itself should also be done by handle(),
# but just for better failures, it is first called on its own
self.assertEqual(expected.to_text(),
session._UpdateSession__check_prerequisites().to_text())
# Now see if handle finds the same result
(result, _, _) = session.handle()
self.assertEqual(expected.to_text(),
session._UpdateSession__message.get_rcode().to_text())
# And that the result looks right
if expected == Rcode.NOERROR():
self.assertEqual(UPDATE_SUCCESS, result)
else:
self.assertEqual(UPDATE_ERROR, result)
def check_prescan_result(self, expected, updates, expected_soa = None):
'''Helper method for checking the result of a prerequisite check;
creates an update session, and fills it with the list of rrsets
from 'updates'. Then checks if __do_prescan()
returns the Rcode specified in 'expected'.'''
msg = create_update_msg([TEST_ZONE_RECORD], [], updates)
zconfig = ZoneConfig([], TEST_RRCLASS, self._datasrc_client,
self._acl_map)
session = UpdateSession(msg, TEST_CLIENT4, zconfig)
session._UpdateSession__get_update_zone()
# compare the to_text output of the rcodes (nicer error messages)
# This call itself should also be done by handle(),
# but just for better failures, it is first called on its own
self.assertEqual(expected.to_text(),
session._UpdateSession__do_prescan().to_text())
# If there is an expected soa, check it
self.assertEqual(str(expected_soa),
str(session._UpdateSession__added_soa))
def check_full_handle_result(self, expected, updates):
'''Helper method for checking the result of a full handle;
creates an update session, and fills it with the list of rrsets
from 'updates'. Then checks if __handle()
results in a response with rcode 'expected'.'''
msg = create_update_msg([TEST_ZONE_RECORD], [], updates)
zconfig = ZoneConfig([], TEST_RRCLASS, self._datasrc_client,
self._acl_map)
session = UpdateSession(msg, TEST_CLIENT4, zconfig)
# Now see if handle finds the same result
(result, _, _) = session.handle()
self.assertEqual(expected.to_text(),
session._UpdateSession__message.get_rcode().to_text())
# And that the result looks right
if expected == Rcode.NOERROR():
self.assertEqual(UPDATE_SUCCESS, result)
else:
self.assertEqual(UPDATE_ERROR, result)
def test_check_prerequisites(self):
# This test checks if the actual prerequisite-type-specific
# methods are called.
# It does test all types of prerequisites, but it does not test
# every possible result for those types (those are tested above,
# in the specific prerequisite type tests)
# Let's first define a number of prereq's that should succeed
rrset_exists_yes = create_rrset("example.org", RRClass.ANY(),
RRType.SOA(), 0)
rrset_exists_value_yes = create_rrset("www.example.org", RRClass.IN(),
RRType.A(), 0, [ "192.0.2.1" ])
rrset_does_not_exist_yes = create_rrset("foo.example.org",
RRClass.NONE(), RRType.SOA(),
0)
name_in_use_yes = create_rrset("www.example.org", RRClass.ANY(),
RRType.ANY(), 0)
name_not_in_use_yes = create_rrset("foo.example.org", RRClass.NONE(),
RRType.ANY(), 0)
rrset_exists_value_1 = create_rrset("example.org", RRClass.IN(),
RRType.NS(), 0,
[ "ns1.example.org" ])
rrset_exists_value_2 = create_rrset("example.org", RRClass.IN(),
RRType.NS(), 0,
[ "ns2.example.org" ])
rrset_exists_value_3 = create_rrset("example.org", RRClass.IN(),
RRType.NS(), 0,
[ "ns3.example.org" ])
# and a number that should not
rrset_exists_no = create_rrset("foo.example.org", RRClass.ANY(),
RRType.SOA(), 0)
rrset_exists_value_no = create_rrset("www.example.org", RRClass.IN(),
RRType.A(), 0, [ "192.0.2.2" ])
rrset_does_not_exist_no = create_rrset("example.org", RRClass.NONE(),
RRType.SOA(), 0)
name_in_use_no = create_rrset("foo.example.org", RRClass.ANY(),
RRType.ANY(), 0)
name_not_in_use_no = create_rrset("www.example.org", RRClass.NONE(),
RRType.ANY(), 0)
# check 'no' result codes
self.check_prerequisite_result(Rcode.NXRRSET(),
[ rrset_exists_no ])
self.check_prerequisite_result(Rcode.NXRRSET(),
[ rrset_exists_value_no ])
self.check_prerequisite_result(Rcode.YXRRSET(),
[ rrset_does_not_exist_no ])
self.check_prerequisite_result(Rcode.NXDOMAIN(),
[ name_in_use_no ])
self.check_prerequisite_result(Rcode.YXDOMAIN(),
[ name_not_in_use_no ])
# the 'yes' codes should result in ok
# individually
self.check_prerequisite_result(Rcode.NOERROR(),
[ rrset_exists_yes ] )
self.check_prerequisite_result(Rcode.NOERROR(),
[ rrset_exists_value_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
[ rrset_does_not_exist_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
[ name_in_use_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
[ name_not_in_use_yes ])
self.check_prerequisite_result(Rcode.NOERROR(),
[ rrset_exists_value_1,
rrset_exists_value_2,
rrset_exists_value_3])
# and together
self.check_prerequisite_result(Rcode.NOERROR(),
[ rrset_exists_yes,
rrset_exists_value_yes,
rrset_does_not_exist_yes,
name_in_use_yes,
name_not_in_use_yes,
rrset_exists_value_1,
rrset_exists_value_2,
rrset_exists_value_3])
# try out a permutation, note that one rrset is split up,
# and the order of the RRs should not matter
self.check_prerequisite_result(Rcode.NOERROR(),
[ rrset_exists_value_3,
rrset_exists_yes,
rrset_exists_value_2,
name_in_use_yes,
rrset_exists_value_1])
# Should fail on the first error, even if most of the
# prerequisites are ok
self.check_prerequisite_result(Rcode.NXDOMAIN(),
[ rrset_exists_value_3,
rrset_exists_yes,
rrset_exists_value_2,
name_in_use_yes,
name_in_use_no,
rrset_exists_value_1])
def test_prerequisite_notzone(self):
rrset = create_rrset("some.other.zone.", RRClass.ANY(), RRType.SOA(), 0)
self.check_prerequisite_result(Rcode.NOTZONE(), [ rrset ])
def test_prerequisites_formerr(self):
# test for form errors in the prerequisite section
# Class ANY, non-zero TTL
rrset = create_rrset("example.org", RRClass.ANY(), RRType.SOA(), 1)
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
# Class ANY, but with rdata
rrset = create_rrset("example.org", RRClass.ANY(), RRType.A(), 0,
[ b'\x00\x00\x00\x00' ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
# Class NONE, non-zero TTL
rrset = create_rrset("example.org", RRClass.NONE(), RRType.SOA(), 1)
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
# Class NONE, but with rdata
rrset = create_rrset("example.org", RRClass.NONE(), RRType.A(), 0,
[ b'\x00\x00\x00\x00' ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
# Matching class and type, but non-zero TTL
rrset = create_rrset("www.example.org", RRClass.IN(), RRType.A(), 1,
[ "192.0.2.1" ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
# Completely different class
rrset = create_rrset("example.org", RRClass.CH(), RRType.TXT(), 0,
[ "foo" ])
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
def __prereq_helper(self, method, expected, rrset):
'''Calls the given method with self._datasrc_client
and the given rrset, and compares the return value.
Function does not do much but makes the code look nicer'''
self.assertEqual(expected, method(rrset))
def __initialize_update_rrsets(self):
'''Prepare a number of RRsets to be used in several update tests
The rrsets are stored in self'''
orig_a_rrset = create_rrset("www.example.org", TEST_RRCLASS,
RRType.A(), 3600, [ "192.0.2.1" ])
self.orig_a_rrset = orig_a_rrset
rrset_update_a = create_rrset("www.example.org", TEST_RRCLASS,
RRType.A(), 3600,
[ "192.0.2.2", "192.0.2.3" ])
self.rrset_update_a = rrset_update_a
rrset_update_soa = create_rrset("example.org", TEST_RRCLASS,
RRType.SOA(), 3600,
[ "ns1.example.org. " +
"admin.example.org. " +
"1233 3600 1800 2419200 7200" ])
self.rrset_update_soa = rrset_update_soa
rrset_update_soa_del = create_rrset("example.org", RRClass.NONE(),
RRType.SOA(), 0,
[ "ns1.example.org. " +
"admin.example.org. " +
"1233 3600 1800 2419200 7200" ])
self.rrset_update_soa_del = rrset_update_soa_del
rrset_update_soa2 = create_rrset("example.org", TEST_RRCLASS,
RRType.SOA(), 3600,
[ "ns1.example.org. " +
"admin.example.org. " +
"4000 3600 1800 2419200 7200" ])
self.rrset_update_soa2 = rrset_update_soa2
rrset_update_del_name = create_rrset("www.example.org", RRClass.ANY(),
RRType.ANY(), 0)
self.rrset_update_del_name = rrset_update_del_name
rrset_update_del_name_apex = create_rrset("example.org", RRClass.ANY(),
RRType.ANY(), 0)
self.rrset_update_del_name_apex = rrset_update_del_name_apex
rrset_update_del_rrset = create_rrset("www.example.org", RRClass.ANY(),
RRType.A(), 0)
self.rrset_update_del_rrset = rrset_update_del_rrset
rrset_update_del_mx_apex = create_rrset("example.org", RRClass.ANY(),
RRType.MX(), 0)
self.rrset_update_del_mx_apex = rrset_update_del_mx_apex
rrset_update_del_soa_apex = create_rrset("example.org", RRClass.ANY(),
RRType.SOA(), 0)
self.rrset_update_del_soa_apex = rrset_update_del_soa_apex
rrset_update_del_ns_apex = create_rrset("example.org", RRClass.ANY(),
RRType.NS(), 0)
self.rrset_update_del_ns_apex = rrset_update_del_ns_apex
rrset_update_del_rrset_part = create_rrset("www.example.org",
RRClass.NONE(), RRType.A(),
0,
[ b'\xc0\x00\x02\x02',
b'\xc0\x00\x02\x03' ])
self.rrset_update_del_rrset_part = rrset_update_del_rrset_part
rrset_update_del_rrset_ns = create_rrset("example.org", RRClass.NONE(),
RRType.NS(), 0,
[ b'\x03ns1\x07example\x03org\x00',
b'\x03ns2\x07example\x03org\x00',
b'\x03ns3\x07example\x03org\x00' ])
self.rrset_update_del_rrset_ns = rrset_update_del_rrset_ns
rrset_update_del_rrset_mx = create_rrset("example.org", RRClass.NONE(),
RRType.MX(), 0,
[ b'\x00\x0a\x04mail\x07example\x03org\x00' ])
self.rrset_update_del_rrset_mx = rrset_update_del_rrset_mx
def test_prescan(self):
'''Test whether the prescan succeeds on data that is ok, and whether
if notices the SOA if present'''
# prepare a set of correct update statements
self.__initialize_update_rrsets()
self.check_prescan_result(Rcode.NOERROR(), [ self.rrset_update_a ])
# check if soa is noticed
self.check_prescan_result(Rcode.NOERROR(), [ self.rrset_update_soa ],
self.rrset_update_soa)
# Other types of succesful prechecks
self.check_prescan_result(Rcode.NOERROR(), [ self.rrset_update_soa2 ],
self.rrset_update_soa2)
self.check_prescan_result(Rcode.NOERROR(),
[ self.rrset_update_del_name ])
self.check_prescan_result(Rcode.NOERROR(),
[ self.rrset_update_del_name_apex ])
self.check_prescan_result(Rcode.NOERROR(),
[ self.rrset_update_del_rrset ])
self.check_prescan_result(Rcode.NOERROR(),
[ self.rrset_update_del_mx_apex ])
self.check_prescan_result(Rcode.NOERROR(),
[ self.rrset_update_del_rrset_part ])
# and check a few permutations of the above
# all of them (with one of the soas)
self.check_prescan_result(Rcode.NOERROR(),
[
self.rrset_update_a,
self.rrset_update_soa,
self.rrset_update_del_name,
self.rrset_update_del_name_apex,
self.rrset_update_del_rrset,
self.rrset_update_del_mx_apex,
self.rrset_update_del_rrset_part
],
self.rrset_update_soa)
# Two soas. Should we reject or simply use the last?
# (RFC is not really explicit on this, but between the lines I read
# use the last)
self.check_prescan_result(Rcode.NOERROR(),
[ self.rrset_update_soa,
self.rrset_update_soa2 ],
self.rrset_update_soa2)
self.check_prescan_result(Rcode.NOERROR(),
[ self.rrset_update_soa2,
self.rrset_update_soa ],
self.rrset_update_soa)
self.check_prescan_result(Rcode.NOERROR(),
[
self.rrset_update_del_mx_apex,
self.rrset_update_del_name,
self.rrset_update_del_name_apex,
self.rrset_update_del_rrset_part,
self.rrset_update_a,
self.rrset_update_del_rrset,
self.rrset_update_soa
],
self.rrset_update_soa)
def test_prescan_failures(self):
'''Test whether prescan fails on bad data'''
# out of zone data
rrset = create_rrset("different.zone", RRClass.ANY(), RRType.TXT(), 0)
self.check_prescan_result(Rcode.NOTZONE(), [ rrset ])
# forbidden type, zone class
rrset = create_rrset(TEST_ZONE_NAME, TEST_RRCLASS, RRType.ANY(), 0,
[ b'\x00' ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
# non-zero TTL, class ANY
rrset = create_rrset(TEST_ZONE_NAME, RRClass.ANY(), RRType.TXT(), 1)
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
# non-zero Rdata, class ANY
rrset = create_rrset(TEST_ZONE_NAME, RRClass.ANY(), RRType.TXT(), 0,
[ "foo" ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
# forbidden type, class ANY
rrset = create_rrset(TEST_ZONE_NAME, RRClass.ANY(), RRType.AXFR(), 0,
[ b'\x00' ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
# non-zero TTL, class NONE
rrset = create_rrset(TEST_ZONE_NAME, RRClass.NONE(), RRType.TXT(), 1)
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
# forbidden type, class NONE
rrset = create_rrset(TEST_ZONE_NAME, RRClass.NONE(), RRType.AXFR(), 0,
[ b'\x00' ])
self.check_prescan_result(Rcode.FORMERR(), [ rrset ])
def __check_inzone_data(self, expected_result, name, rrtype,
expected_rrset = None):
'''Does a find on TEST_ZONE for the given rrset's name and type,
then checks if the result matches the expected result.
If so, and if expected_rrset is given, they are compared as
well.'''
_, finder = self._datasrc_client.find_zone(TEST_ZONE_NAME)
result, found_rrset, _ = finder.find(name, rrtype,
finder.NO_WILDCARD |
finder.FIND_GLUE_OK)
self.assertEqual(expected_result, result)
# Sigh. Need rrsets.compare() again.
# To be sure, compare name, class, type, and ttl
if expected_rrset is not None:
self.assertEqual(expected_rrset.get_name(), found_rrset.get_name())
self.assertEqual(expected_rrset.get_class(), found_rrset.get_class())
self.assertEqual(expected_rrset.get_type(), found_rrset.get_type())
self.assertEqual(expected_rrset.get_ttl().to_text(),
found_rrset.get_ttl().to_text())
expected_rdata =\
[ rdata.to_text() for rdata in expected_rrset.get_rdata() ]
found_rdata =\
[ rdata.to_text() for rdata in found_rrset.get_rdata() ]
expected_rdata.sort()
found_rdata.sort()
self.assertEqual(expected_rdata, found_rdata)
def test_update_add_delete_rrset(self):
'''
Tests a sequence of related add and delete updates. Some other
cases are tested by later tests.
'''
self.__initialize_update_rrsets()
# initially, the www should only contain one rr
# (set to self.orig_a_rrset)
# during this test, we will extend it at some point
extended_a_rrset = create_rrset("www.example.org", TEST_RRCLASS,
RRType.A(), 3600,
[ "192.0.2.1",
"192.0.2.2",
"192.0.2.3" ])
# Sanity check, make sure original data is really there before updates
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A(),
self.orig_a_rrset)
# Add two rrs
self.check_full_handle_result(Rcode.NOERROR(), [ self.rrset_update_a ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A(),
extended_a_rrset)
# Adding the same RRsets should not make a difference.
self.check_full_handle_result(Rcode.NOERROR(), [ self.rrset_update_a ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A(),
extended_a_rrset)
# Now delete those two, and we should end up with the original RRset
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_rrset_part ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A(),
self.orig_a_rrset)
# 'Deleting' them again should make no difference
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_rrset_part ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A(),
self.orig_a_rrset)
# But deleting the entire rrset, independent of its contents, should
# work
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("www.example.org"),
RRType.A())
# Check that if we update the SOA, it is updated to our value
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_soa2 ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.SOA(),
self.rrset_update_soa2)
def test_glue_deletions(self):
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("sub.example.org."),
RRType.NS())
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("ns.sub.example.org."),
RRType.A())
# See that we can delete glue
rrset_delete_glue = create_rrset("ns.sub.example.org.",
RRClass.ANY(),
RRType.A(),
0)
self.check_full_handle_result(Rcode.NOERROR(),
[ rrset_delete_glue ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("sub.example.org."),
RRType.NS())
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("ns.sub.example.org."),
RRType.A())
# Check that we don't accidentally delete a delegation if we
# try to delete non-existent glue
rrset_delete_nonexistent_glue = create_rrset("foo.sub.example.org.",
RRClass.ANY(),
RRType.A(),
0)
self.check_full_handle_result(Rcode.NOERROR(),
[ rrset_delete_nonexistent_glue ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("sub.example.org."),
RRType.NS())
def test_update_add_new_data(self):
'''
This tests adds data where none is present
'''
# Add data at a completely new name
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("new.example.org"),
RRType.A())
rrset = create_rrset("new.example.org", TEST_RRCLASS, RRType.A(),
3600, [ "192.0.2.1", "192.0.2.2" ])
self.check_full_handle_result(Rcode.NOERROR(), [ rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("new.example.org"),
RRType.A(),
rrset)
# Also try a name where data is present, but none of this
# specific type
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXRRSET,
isc.dns.Name("new.example.org"),
RRType.TXT())
rrset = create_rrset("new.example.org", TEST_RRCLASS, RRType.TXT(),
3600, [ "foo" ])
self.check_full_handle_result(Rcode.NOERROR(), [ rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("new.example.org"),
RRType.TXT(),
rrset)
def test_update_add_new_data_interspersed(self):
'''
This tests adds data where none is present, similar to
test_update_add_new_data, but this time the second RRset
is put into the record between the two RRs of the first
RRset.
'''
# Add data at a completely new name
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("new_a.example.org"),
RRType.A())
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("new_txt.example.org"),
RRType.TXT())
rrset1 = create_rrset("new_a.example.org", TEST_RRCLASS, RRType.A(),
3600, [ "192.0.2.1" ])
rrset2 = create_rrset("new_txt.example.org", TEST_RRCLASS, RRType.TXT(),
3600, [ "foo" ])
rrset3 = create_rrset("new_a.example.org", TEST_RRCLASS, RRType.A(),
3600, [ "192.0.2.2" ])
self.check_full_handle_result(Rcode.NOERROR(),
[ rrset1, rrset2, rrset3 ])
# The update should have merged rrset1 and rrset3
rrset_merged = create_rrset("new_a.example.org", TEST_RRCLASS,
RRType.A(), 3600,
[ "192.0.2.1", "192.0.2.2" ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("new_a.example.org"),
RRType.A(),
rrset_merged)
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("new_txt.example.org"),
RRType.TXT(),
rrset2)
def test_update_delete_name(self):
self.__initialize_update_rrsets()
# First check it is there
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A())
# Delete the entire name
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_name ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("www.example.org"),
RRType.A())
# Should still be gone after pointless second delete
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_name ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXDOMAIN,
isc.dns.Name("www.example.org"),
RRType.A())
def test_update_apex_special_cases(self):
'''
Tests a few special cases when deleting data from the apex
'''
self.__initialize_update_rrsets()
# the original SOA
orig_soa_rrset = create_rrset("example.org", TEST_RRCLASS,
RRType.SOA(), 3600,
[ "ns1.example.org. " +
"admin.example.org. " +
"1234 3600 1800 2419200 7200" ])
# We will delete some of the NS records
orig_ns_rrset = create_rrset("example.org", TEST_RRCLASS,
RRType.NS(), 3600,
[ "ns1.example.org.",
"ns2.example.org.",
"ns3.example.org." ])
# Sanity check, make sure original data is really there before updates
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.NS(),
orig_ns_rrset)
# We will delete the MX record later in this test, so let's make
# sure that it exists (we do not care about its value)
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.MX())
# Check that we cannot delete the SOA record by direction deletion
# both by name+type and by full rrset
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_soa_apex,
self.rrset_update_soa_del ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.SOA(),
orig_soa_rrset)
# If we delete everything at the apex, the SOA and NS rrsets should be
# untouched
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_name_apex ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.SOA(),
orig_soa_rrset)
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.NS(),
orig_ns_rrset)
# but the MX should be gone
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXRRSET,
isc.dns.Name("example.org"),
RRType.MX())
# Deleting the NS rrset by name and type only, it should also be left
# untouched
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_ns_apex ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.NS(),
orig_ns_rrset)
def DISABLED_test_update_apex_special_case_ns_rrset(self):
# If we delete the NS at the apex specifically, it should still
# keep one record
self.__initialize_update_rrsets()
# When we are done, we should have a reduced NS rrset
short_ns_rrset = create_rrset("example.org", TEST_RRCLASS,
RRType.NS(), 3600,
[ "ns3.example.org." ])
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_rrset_ns ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.NS(),
short_ns_rrset)
def test_update_delete_normal_rrset_at_apex(self):
'''
Tests a number of 'normal rrset' deletes at the apex
'''
# MX should simply be deleted
self.__initialize_update_rrsets()
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("example.org"),
RRType.MX())
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_rrset_mx ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.NXRRSET,
isc.dns.Name("example.org"),
RRType.MX())
def test_update_cname_special_cases(self):
self.__initialize_update_rrsets()
# Sanity check
orig_cname_rrset = create_rrset("cname.example.org", TEST_RRCLASS,
RRType.CNAME(), 3600,
[ "www.example.org." ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.CNAME,
isc.dns.Name("cname.example.org"),
RRType.A(),
orig_cname_rrset)
# If we try to add data where a cname is preset
rrset = create_rrset("cname.example.org", TEST_RRCLASS, RRType.A(),
3600, [ "192.0.2.1" ])
self.check_full_handle_result(Rcode.NOERROR(), [ rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.CNAME,
isc.dns.Name("cname.example.org"),
RRType.A(),
orig_cname_rrset)
# But updating the cname itself should work
new_cname_rrset = create_rrset("cname.example.org", TEST_RRCLASS,
RRType.CNAME(), 3600,
[ "mail.example.org." ])
self.check_full_handle_result(Rcode.NOERROR(), [ new_cname_rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.CNAME,
isc.dns.Name("cname.example.org"),
RRType.A(),
new_cname_rrset)
self.__initialize_update_rrsets()
# Likewise, adding a cname where other data is
# present should do nothing either
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A(),
self.orig_a_rrset)
new_cname_rrset = create_rrset("www.example.org", TEST_RRCLASS,
RRType.CNAME(), 3600,
[ "mail.example.org." ])
self.check_full_handle_result(Rcode.NOERROR(), [ new_cname_rrset ])
self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
isc.dns.Name("www.example.org"),
RRType.A(),
self.orig_a_rrset)
def test_update_bad_class(self):
rrset = create_rrset("example.org.", RRClass.CH(), RRType.TXT(), 0,
[ "foo" ])
self.check_full_handle_result(Rcode.FORMERR(), [ rrset ])
def test_uncaught_exception(self):
def my_exc():
raise Exception("foo")
self._session._UpdateSession__update_soa = my_exc
self.assertEqual(Rcode.SERVFAIL().to_text(),
self._session._UpdateSession__do_update().to_text())
class SessionACLTest(SessionTestBase):
'''ACL related tests for update session.'''
def test_update_acl_check(self):
'''Test for various ACL checks.
Note that accepted cases are covered in the basic tests.
'''
# create a separate session, with default (empty) ACL map.
session = UpdateSession(self._update_msg,
TEST_CLIENT4, ZoneConfig([], TEST_RRCLASS,
self._datasrc_client))
# then the request should be rejected.
self.assertEqual((UPDATE_ERROR, None, None), session.handle())
# recreate the request message, and test with an ACL that would result
# in 'DROP'. get_message() should return None.
msg = create_update_msg()
acl_map = {(TEST_ZONE_NAME, TEST_RRCLASS):
REQUEST_LOADER.load([{"action": "DROP", "from":
TEST_CLIENT4[0]}])}
session = UpdateSession(msg, TEST_CLIENT4,
ZoneConfig([], TEST_RRCLASS,
self._datasrc_client, acl_map))
self.assertEqual((UPDATE_DROP, None, None), session.handle())
self.assertEqual(None, session.get_message())
def test_update_tsigacl_check(self):
'''Test for various ACL checks using TSIG.'''
# This ACL will accept requests from TEST_CLIENT4 (any port) *and*
# has TSIG signed by TEST_ZONE_NAME; all others will be rejected.
acl_map = {(TEST_ZONE_NAME, TEST_RRCLASS):
REQUEST_LOADER.load([{"action": "ACCEPT",
"from": TEST_CLIENT4[0],
"key": TEST_ZONE_NAME.to_text()}])}
# If the message doesn't contain TSIG, it doesn't match the ACCEPT
# ACL entry, and the request should be rejected.
session = UpdateSession(self._update_msg,
TEST_CLIENT4, ZoneConfig([], TEST_RRCLASS,
self._datasrc_client,
acl_map))
self.assertEqual((UPDATE_ERROR, None, None), session.handle())
self.check_response(session.get_message(), Rcode.REFUSED())
# If the message contains TSIG, it should match the ACCEPT
# ACL entry, and the request should be granted.
session = UpdateSession(create_update_msg(tsig_key=TEST_TSIG_KEY),
TEST_CLIENT4, ZoneConfig([], TEST_RRCLASS,
self._datasrc_client,
acl_map))
self.assertEqual((UPDATE_SUCCESS, TEST_ZONE_NAME, TEST_RRCLASS),
session.handle())
if __name__ == "__main__":
isc.log.init("bind10")
isc.log.resetUnitTestRootLogger()
unittest.main()