Commit f89b4428 authored by JINMEI Tatuya's avatar JINMEI Tatuya
Browse files

[master] Merge branch 'trac1458' with fixing conflicts.

parents 195c7eae 24fb9317
......@@ -93,6 +93,22 @@ RRset exists (value independent). At least one RR with a
specified NAME and TYPE (in the zone and class specified by
the Zone Section) must exist.
% LIBDDNS_UPDATE_APPROVED update client %1 for zone %2 approved
Debug message. An update request was approved in terms of the zone's
update ACL.
% LIBDDNS_UPDATE_DENIED update client %1 for zone %2 denied
Informational message. An update request was denied because it was
rejected by the zone's update ACL. When this library is used by
b10-ddns, the server will respond to the request with an RCODE of
REFUSED as described in Section 3.3 of RFC2136.
% LIBDDNS_UPDATE_DROPPED update client %1 for zone %2 dropped
Informational message. An update request was denied because it was
rejected by the zone's update ACL. When this library is used by
b10-ddns, the server will then completely ignore the request; no
response will be sent.
% LIBDDNS_UPDATE_ERROR update client %1 for zone %2: %3
Debug message. An error is found in processing a dynamic update
request. This log message is used for general errors that are not
......@@ -109,14 +125,14 @@ will simply return a response with an RCODE of NOTIMP to the client.
The client's address and the zone name/class are logged.
% LIBDDNS_UPDATE_NOTAUTH update client %1 for zone %2: not authoritative for update zone
Debug message. An update request for a zone for which the receiving
server doesn't have authority. In theory this is an unexpected event,
but there are client implementations that could send update requests
carelessly, so it may not necessarily be so uncommon in practice. If
possible, you may want to check the implementation or configuration of
those clients to suppress the requests. As specified in Section 3.1
of RFC2136, the receiving server will return a response with an RCODE
of NOTAUTH.
Debug message. An update request was received for a zone for which
the receiving server doesn't have authority. In theory this is an
unexpected event, but there are client implementations that could send
update requests carelessly, so it may not necessarily be so uncommon
in practice. If possible, you may want to check the implementation or
configuration of those clients to suppress the requests. As specified
in Section 3.1 of RFC2136, the receiving server will return a response
with an RCODE of NOTAUTH.
% LIBDDNS_UPDATE_PREREQUISITE_FAILED prerequisite failed in update update client %1 for zone %2: result code %3
The handling of the prerequisite section (RFC2136 Section 3.2) found
......
......@@ -28,9 +28,11 @@ class ClientFormatter:
This class is constructed with a Python standard socket address tuple.
If it's 2-element tuple, it's assumed to be an IPv4 socket address
and will be converted to the form of '<addr>:<port>'.
and will be converted to the form of '<addr>:<port>(/key=<tsig-key>)'.
If it's 4-element tuple, it's assumed to be an IPv6 socket address.
and will be converted to the form of '[<addr>]:<por>'.
and will be converted to the form of '[<addr>]:<por>(/key=<tsig-key>)'.
The optional key=<tsig-key> will be added if a TSIG record is given
on construction. tsig-key is the TSIG key name in that case.
This class is designed to delay the conversion until it's explicitly
requested, so the conversion doesn't happen if the corresponding log
......@@ -45,16 +47,23 @@ class ClientFormatter:
Right now this is an open issue.
"""
def __init__(self, addr):
def __init__(self, addr, tsig_record=None):
self.__addr = addr
self.__tsig_record = tsig_record
def __str__(self):
def __format_addr(self):
if len(self.__addr) == 2:
return self.__addr[0] + ':' + str(self.__addr[1])
elif len(self.__addr) == 4:
return '[' + self.__addr[0] + ']:' + str(self.__addr[1])
return None
def __str__(self):
format = self.__format_addr()
if format is not None and self.__tsig_record is not None:
format += '/key=' + self.__tsig_record.get_name().to_text(True)
return format
class ZoneFormatter:
"""A utility class to convert zone name and class to string.
......
......@@ -19,6 +19,7 @@ from isc.log import *
from isc.ddns.logger import logger, ClientFormatter, ZoneFormatter,\
RRsetFormatter
from isc.log_messages.libddns_messages import *
from isc.acl.acl import ACCEPT, REJECT, DROP
import copy
# Result codes for UpdateSession.handle()
......@@ -46,7 +47,8 @@ class UpdateError(Exception):
- msg (string) A string explaining the error.
- zname (isc.dns.Name) The zone name. Can be None when not identified.
- zclass (isc.dns.RRClass) The zone class. Like zname, can be None.
- rcode (isc.dns.RCode) The RCODE to be set in the response message.
- rcode (isc.dns.RCode or None) The RCODE to be set in the response
message; this can be None if the response is not expected to be sent.
- nolog (bool) If True, it indicates there's no more need for logging.
'''
......@@ -67,30 +69,24 @@ class UpdateSession:
class can use the message to send a response to the client.
'''
def __init__(self, req_message, req_data, client_addr, zone_config):
def __init__(self, req_message, client_addr, zone_config):
'''Constructor.
Note: req_data is not really used as of #1512 but is listed since
it's quite likely we need it in a subsequent task soon. We'll
also need to get other parameters such as ACLs, for which, it's less
clear in which form we want to get the information, so it's left
open for now.
Parameters:
- req_message (isc.dns.Message) The request message. This must be
in the PARSE mode.
- req_data (binary) Wire format data of the request message.
It will be used for TSIG verification if necessary.
in the PARSE mode, its Opcode must be UPDATE, and must have been
TSIG validatd if it's TSIG signed.
- client_addr (socket address) The address/port of the update client
in the form of Python socket address object. This is mainly for
logging and access control.
- zone_config (ZoneConfig) A tentative container that encapsulates
the server's zone configuration. See zone_config.py.
(It'll soon need to be passed ACL in some way, too)
- req_data (binary) Wire format data of the request message.
It will be used for TSIG verification if necessary.
'''
self.__message = req_message
self.__tsig = req_message.get_tsig_record()
self.__client_addr = client_addr
self.__zone_config = zone_config
......@@ -98,8 +94,10 @@ class UpdateSession:
'''Return the update message.
After handle() is called, it's generally transformed to the response
to be returned to the client; otherwise it would be identical to
the request message passed on construction.
to be returned to the client. If the request has been dropped,
this method returns None. If this method is called before handle()
the return value would be identical to the request message passed on
construction, although it's of no practical use.
'''
return self.__message
......@@ -115,7 +113,8 @@ class UpdateSession:
UPDATE_DROP Error happened and no response should be sent.
Except the case of UPDATE_DROP, the UpdateSession object will have
created a response that is to be returned to the request client,
which can be retrieved by get_message().
which can be retrieved by get_message(). If it's UPDATE_DROP,
subsequent call to get_message() returns None.
- The name of the updated zone (isc.dns.Name object) in case of
UPDATE_SUCCESS; otherwise None.
- The RR class of the updated zone (isc.dns.RRClass object) in case
......@@ -130,17 +129,22 @@ class UpdateSession:
if prereq_result != Rcode.NOERROR():
self.__make_response(prereq_result)
return UPDATE_ERROR, zname, zclass
# self.__check_update_acl()
self.__check_update_acl(zname, zclass)
# self.__do_update()
# self.__make_response(Rcode.NOERROR())
return UPDATE_SUCCESS, zname, zclass
except UpdateError as e:
if not e.nolog:
logger.debug(logger.DBGLVL_TRACE_BASIC, LIBDDNS_UPDATE_ERROR,
ClientFormatter(self.__client_addr),
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(e.zname, e.zclass), e)
self.__make_response(e.rcode)
return UPDATE_ERROR, None, None
# If RCODE is specified, create a corresponding resonse and return
# ERROR; otherwise clear the message and return DROP.
if e.rcode is not None:
self.__make_response(e.rcode)
return UPDATE_ERROR, None, None
self.__message = None
return UPDATE_DROP, None, None
def __get_update_zone(self):
'''Parse the zone section and find the zone to be updated.
......@@ -173,15 +177,34 @@ class UpdateSession:
# We are a secondary server; since we don't yet support update
# forwarding, we return 'not implemented'.
logger.debug(DBGLVL_TRACE_BASIC, LIBDDNS_UPDATE_FORWARD_FAIL,
ClientFormatter(self.__client_addr),
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
raise UpdateError('forward', zname, zclass, Rcode.NOTIMP(), True)
# zone wasn't found
logger.debug(DBGLVL_TRACE_BASIC, LIBDDNS_UPDATE_NOTAUTH,
ClientFormatter(self.__client_addr),
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
raise UpdateError('notauth', zname, zclass, Rcode.NOTAUTH(), True)
def __check_update_acl(self, zname, zclass):
'''Apply update ACL for the zone to be updated.'''
acl = self.__zone_config.get_update_acl(zname, zclass)
action = acl.execute(isc.acl.dns.RequestContext(
(self.__client_addr[0], self.__client_addr[1]), self.__tsig))
if action == REJECT:
logger.info(LIBDDNS_UPDATE_DENIED,
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
raise UpdateError('rejected', zname, zclass, Rcode.REFUSED(), True)
if action == DROP:
logger.info(LIBDDNS_UPDATE_DROPPED,
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
raise UpdateError('dropped', zname, zclass, None, True)
logger.debug(logger.DBGLVL_TRACE_BASIC, LIBDDNS_UPDATE_APPROVED,
ClientFormatter(self.__client_addr, self.__tsig),
ZoneFormatter(zname, zclass))
def __make_response(self, rcode):
'''Transform the internal message to the update response.
......
......@@ -6,7 +6,7 @@ CLEANFILES = $(builddir)/rwtest.sqlite3.copied
# If necessary (rare cases), explicitly specify paths to dynamic libraries
# required by loadable python modules.
if SET_ENV_LIBRARY_PATH
LIBRARY_PATH_PLACEHOLDER = $(ENV_LIBRARY_PATH)=$(abs_top_builddir)/src/lib/cryptolink/.libs:$(abs_top_builddir)/src/lib/dns/.libs:$(abs_top_builddir)/src/lib/dns/python/.libs:$(abs_top_builddir)/src/lib/cc/.libs:$(abs_top_builddir)/src/lib/config/.libs:$(abs_top_builddir)/src/lib/log/.libs:$(abs_top_builddir)/src/lib/util/.libs:$(abs_top_builddir)/src/lib/exceptions/.libs:$(abs_top_builddir)/src/lib/datasrc/.libs:$$$(ENV_LIBRARY_PATH)
LIBRARY_PATH_PLACEHOLDER = $(ENV_LIBRARY_PATH)=$(abs_top_builddir)/src/lib/cryptolink/.libs:$(abs_top_builddir)/src/lib/dns/.libs:$(abs_top_builddir)/src/lib/dns/python/.libs:$(abs_top_builddir)/src/lib/cc/.libs:$(abs_top_builddir)/src/lib/config/.libs:$(abs_top_builddir)/src/lib/log/.libs:$(abs_top_builddir)/src/lib/util/.libs:$(abs_top_builddir)/src/lib/acl/.libs:$(abs_top_builddir)/src/lib/exceptions/.libs:$(abs_top_builddir)/src/lib/datasrc/.libs:$$$(ENV_LIBRARY_PATH)
endif
# test using command-line arguments, so use check-local target instead of TESTS
......
......@@ -35,8 +35,11 @@ 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=[]):
def create_update_msg(zones=[TEST_ZONE_RECORD], prerequisites=[],
tsig_key=None):
msg = Message(Message.RENDER)
msg.set_qid(5353) # arbitrary chosen
msg.set_opcode(Opcode.UPDATE())
......@@ -47,25 +50,35 @@ def create_update_msg(zones=[TEST_ZONE_RECORD], prerequisites=[]):
msg.add_rrset(SECTION_PREREQUISITE, p)
renderer = MessageRenderer()
msg.to_wire(renderer)
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())
return renderer.get_data(), msg
return msg
class SessionTest(unittest.TestCase):
'''Session tests'''
class SesseionTestBase(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_msgdata, self.__update_msg = create_update_msg()
self.__session = UpdateSession(self.__update_msg,
self.__update_msgdata, TEST_CLIENT4,
ZoneConfig([], TEST_RRCLASS,
self.__datasrc_client))
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))
def check_response(self, msg, expected_rcode):
'''Perform common checks on update resposne message.'''
......@@ -79,9 +92,12 @@ class SessionTest(unittest.TestCase):
self.assertEqual(0, msg.get_rr_count(SECTION_UPDATE))
self.assertEqual(0, msg.get_rr_count(Message.SECTION_ADDITIONAL))
class SessionTest(SesseionTestBase):
'''Basic session tests'''
def test_handle(self):
'''Basic update case'''
result, zname, zclass = self.__session.handle()
result, zname, zclass = self._session.handle()
self.assertEqual(UPDATE_SUCCESS, result)
self.assertEqual(TEST_ZONE_NAME, zname)
self.assertEqual(TEST_RRCLASS, zclass)
......@@ -92,8 +108,8 @@ class SessionTest(unittest.TestCase):
def test_broken_request(self):
# Zone section is empty
msg_data, msg = create_update_msg(zones=[])
session = UpdateSession(msg, msg_data, TEST_CLIENT6, None)
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)
......@@ -101,17 +117,15 @@ class SessionTest(unittest.TestCase):
self.check_response(session.get_message(), Rcode.FORMERR())
# Zone section contains multiple records
msg_data, msg = create_update_msg(zones=[TEST_ZONE_RECORD,
TEST_ZONE_RECORD])
session = UpdateSession(msg, msg_data, TEST_CLIENT4, None)
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_data, msg = create_update_msg(zones=[Question(TEST_ZONE_NAME,
TEST_RRCLASS,
RRType.A())])
session = UpdateSession(msg, msg_data, TEST_CLIENT4, None)
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())
......@@ -119,24 +133,20 @@ class SessionTest(unittest.TestCase):
# specified zone is configured as a secondary. Since this
# implementation doesn't support update forwarding, the result
# should be NOTIMP.
msg_data, msg = create_update_msg(zones=[Question(TEST_ZONE_NAME,
TEST_RRCLASS,
RRType.SOA())])
session = UpdateSession(msg, msg_data, TEST_CLIENT4,
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))
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_data, msg = create_update_msg(zones=[Question(zname, zclass,
RRType.SOA())])
session = UpdateSession(msg, msg_data, TEST_CLIENT4,
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))
TEST_RRCLASS, self._datasrc_client))
self.assertEqual(UPDATE_ERROR, session.handle()[0])
self.check_response(session.get_message(), Rcode.NOTAUTH())
......@@ -151,10 +161,10 @@ class SessionTest(unittest.TestCase):
self.check_notauth(Name('example.org'), RRClass.CH())
def __prereq_helper(self, method, expected, rrset):
'''Calls the given method with self.__datasrc_client
'''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(self.__datasrc_client, rrset))
self.assertEqual(expected, method(self._datasrc_client, rrset))
def __check_prerequisite_exists_combined(self, method, rrclass, expected):
'''shared code for the checks for the very similar (but reversed
......@@ -232,19 +242,19 @@ class SessionTest(unittest.TestCase):
self.__prereq_helper(method, expected, rrset)
def test_check_prerequisite_exists(self):
method = self.__session._UpdateSession__prereq_rrset_exists
method = self._session._UpdateSession__prereq_rrset_exists
self.__check_prerequisite_exists_combined(method,
isc.dns.RRClass.ANY(),
True)
def test_check_prerequisite_does_not_exist(self):
method = self.__session._UpdateSession__prereq_rrset_does_not_exist
method = self._session._UpdateSession__prereq_rrset_does_not_exist
self.__check_prerequisite_exists_combined(method,
isc.dns.RRClass.NONE(),
False)
def test_check_prerequisite_exists_value(self):
method = self.__session._UpdateSession__prereq_rrset_exists_value
method = self._session._UpdateSession__prereq_rrset_exists_value
rrset = isc.dns.RRset(isc.dns.Name("www.example.org"),
isc.dns.RRClass.IN(), isc.dns.RRType.A(),
......@@ -359,13 +369,13 @@ class SessionTest(unittest.TestCase):
self.__prereq_helper(method, expected, rrset)
def test_check_prerequisite_name_in_use(self):
method = self.__session._UpdateSession__prereq_name_in_use
method = self._session._UpdateSession__prereq_name_in_use
self.__check_prerequisite_name_in_use_combined(method,
isc.dns.RRClass.ANY(),
True)
def test_check_prerequisite_name_not_in_use(self):
method = self.__session._UpdateSession__prereq_name_not_in_use
method = self._session._UpdateSession__prereq_name_not_in_use
self.__check_prerequisite_name_in_use_combined(method,
isc.dns.RRClass.NONE(),
False)
......@@ -375,15 +385,15 @@ class SessionTest(unittest.TestCase):
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_data, msg = create_update_msg([TEST_ZONE_RECORD],
prerequisites)
zconfig = ZoneConfig([], TEST_RRCLASS, self.__datasrc_client)
session = UpdateSession(msg, msg_data, TEST_CLIENT4, zconfig)
msg = create_update_msg([TEST_ZONE_RECORD], prerequisites)
zconfig = ZoneConfig([], TEST_RRCLASS, self._datasrc_client,
self._acl_map)
session = UpdateSession(msg, TEST_CLIENT4, zconfig)
# 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(self.__datasrc_client,
session._UpdateSession__check_prerequisites(self._datasrc_client,
TEST_ZONE_NAME,
TEST_RRCLASS).to_text())
# Now see if handle finds the same result
......@@ -485,14 +495,14 @@ class SessionTest(unittest.TestCase):
isc.dns.RRTTL(0))
# Create an UPDATE with all 5 'yes' prereqs
data, update = create_update_msg([TEST_ZONE_RECORD],
[
rrset_exists_yes,
rrset_does_not_exist_yes,
name_in_use_yes,
name_not_in_use_yes,
rrset_exists_value_yes,
])
create_update_msg([TEST_ZONE_RECORD],
[rrset_exists_yes,
rrset_does_not_exist_yes,
name_in_use_yes,
name_not_in_use_yes,
rrset_exists_value_yes,
])
# check 'no' result codes
self.check_prerequisite_result(Rcode.NXRRSET(),
[ rrset_exists_no ])
......@@ -599,6 +609,60 @@ class SessionTest(unittest.TestCase):
"foo"))
self.check_prerequisite_result(Rcode.FORMERR(), [ rrset ])
class SessionACLTest(SesseionTestBase):
'''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()
......
......@@ -14,15 +14,23 @@
# WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
import isc.log
import unittest
from isc.dns import *
from isc.datasrc import DataSourceClient
from isc.ddns.zone_config import *
import isc.acl.dns
from isc.acl.acl import ACCEPT, REJECT, DROP, LoaderError
import unittest
import socket
# Some common test parameters
TEST_ZONE_NAME = Name('example.org')
TEST_SECONDARY_ZONE_NAME = Name('example.com')
TEST_RRCLASS = RRClass.IN()
TEST_TSIG_KEY = TSIGKey("example.com:SFuWd/q99SzF8Yzd1QbB9g==")
TEST_ACL_CONTEXT = isc.acl.dns.RequestContext(
socket.getaddrinfo("192.0.2.1", 1234, 0, socket.SOCK_DGRAM,
socket.IPPROTO_UDP, socket.AI_NUMERICHOST)[0][4])
class FakeDataSourceClient:
'''Faked data source client used in the ZoneConfigTest.
......@@ -93,7 +101,7 @@ class ZoneConfigTest(unittest.TestCase):
# empty secondary list doesn't cause any disruption.
zconfig = ZoneConfig([], TEST_RRCLASS, self.__datasrc_client)
self.assertEqual((ZONE_PRIMARY, self.__datasrc_client),
(self.zconfig.find_zone(TEST_ZONE_NAME, TEST_RRCLASS)))
self.zconfig.find_zone(TEST_ZONE_NAME, TEST_RRCLASS))
# adding some mulitle tuples, including subdomainof the test zone name,
# and the same zone name but a different class
zconfig = ZoneConfig([(TEST_SECONDARY_ZONE_NAME, TEST_RRCLASS),
......@@ -102,14 +110,55 @@ class ZoneConfigTest(unittest.TestCase):
(TEST_ZONE_NAME, RRClass.CH())],
TEST_RRCLASS, self.__datasrc_client)
self.assertEqual((ZONE_PRIMARY, self.__datasrc_client),
(self.zconfig.find_zone(TEST_ZONE_NAME, TEST_RRCLASS)))
self.zconfig.find_zone(TEST_ZONE_NAME, TEST_RRCLASS))
# secondary zone list has a duplicate entry, which is just
# (effecitivey) ignored
zconfig = ZoneConfig([(TEST_SECONDARY_ZONE_NAME, TEST_RRCLASS),
(TEST_SECONDARY_ZONE_NAME, TEST_RRCLASS)],
TEST_RRCLASS, self.__datasrc_client)
self.assertEqual((ZONE_PRIMARY, self.__datasrc_client),
(self.zconfig.find_zone(TEST_ZONE_NAME, TEST_RRCLASS)))
self.zconfig.find_zone(TEST_ZONE_NAME, TEST_RRCLASS))
class ACLConfigTest(unittest.TestCase):
def setUp(self):
self.__datasrc_client = FakeDataSourceClient()
self.__zconfig = ZoneConfig([(TEST_SECONDARY_ZONE_NAME, TEST_RRCLASS)],
TEST_RRCLASS, self.__datasrc_client)
def test_get_update_acl(self):
# By default, no ACL is set, and the default ACL is "reject all"
acl = self.__zconfig.get_update_acl(TEST_ZONE_NAME, TEST_RRCLASS)
self.assertEqual(REJECT, acl.execute(TEST_ACL_CONTEXT))
# Add a map entry that would match the request, and it should now be
# accepted.
acl_map = {(TEST_ZONE_NAME, TEST_RRCLASS):
REQUEST_LOADER.load([{"action": "ACCEPT"}])}
self.__zconfig.set_update_acl_map(acl_map)
acl = self.__zconfig.get_update_acl(TEST_ZONE_NAME, TEST_RRCLASS)
self.assertEqual(ACCEPT, acl.execute(TEST_ACL_CONTEXT))
# 'All reject' ACL will still apply for any other zones
acl = self.__zconfig.get_update_acl(Name('example.com'), TEST_RRCLASS)
self.assertEqual(REJECT, acl.execute(TEST_ACL_CONTEXT))