Commit 18d0a74b authored by JINMEI Tatuya's avatar JINMEI Tatuya
Browse files

[trac1104] supported TSIG ACL for xfrout.

parent ae1cf18d
......@@ -172,6 +172,55 @@ class TestXfroutSession(unittest.TestCase):
self.assertEqual(rcode.to_text(), "NOTAUTH")
self.assertTrue(self.xfrsess._tsig_ctx is not None)
# ACL using TSIG: successful case
self.xfrsess._acl = isc.acl.dns.REQUEST_LOADER.load([
{"key": "example.com", "action": "ACCEPT"}, {"action": "REJECT"}
])
self.xfrsess._tsig_key_ring.add(TSIG_KEY)
[rcode, msg] = self.xfrsess._parse_query_message(request_data)
self.assertEqual(rcode.to_text(), "NOERROR")
# ACL using TSIG: key name doesn't match; should be rejected
self.xfrsess._acl = isc.acl.dns.REQUEST_LOADER.load([
{"key": "example.org", "action": "ACCEPT"}, {"action": "REJECT"}
])
self.xfrsess._tsig_key_ring.add(TSIG_KEY)
[rcode, msg] = self.xfrsess._parse_query_message(request_data)
self.assertEqual(rcode.to_text(), "REFUSED")
# ACL using TSIG: no TSIG; should be rejected
self.xfrsess._acl = isc.acl.dns.REQUEST_LOADER.load([
{"key": "example.org", "action": "ACCEPT"}, {"action": "REJECT"}
])
self.xfrsess._tsig_key_ring.add(TSIG_KEY)
[rcode, msg] = self.xfrsess._parse_query_message(self.mdata)
self.assertEqual(rcode.to_text(), "REFUSED")
#
# ACL using IP + TSIG: both should match
#
self.xfrsess._acl = isc.acl.dns.REQUEST_LOADER.load([
{"ALL": [{"key": "example.com"}, {"from": "192.0.2.1"}],
"action": "ACCEPT"},
{"action": "REJECT"}
])
# both matches
self.xfrsess._remote = ('192.0.2.1', 12345)
[rcode, msg] = self.xfrsess._parse_query_message(request_data)
self.assertEqual(rcode.to_text(), "NOERROR")
# TSIG matches, but address doesn't
self.xfrsess._remote = ('192.0.2.2', 12345)
[rcode, msg] = self.xfrsess._parse_query_message(request_data)
self.assertEqual(rcode.to_text(), "REFUSED")
# Address matches, but TSIG doesn't (not included)
self.xfrsess._remote = ('192.0.2.1', 12345)
[rcode, msg] = self.xfrsess._parse_query_message(self.mdata)
self.assertEqual(rcode.to_text(), "REFUSED")
# Neither address nor TSIG matches
self.xfrsess._remote = ('192.0.2.2', 12345)
[rcode, msg] = self.xfrsess._parse_query_message(self.mdata)
self.assertEqual(rcode.to_text(), "REFUSED")
def test_get_query_zone_name(self):
msg = self.getmsg()
self.assertEqual(self.xfrsess._get_query_zone_name(msg), "example.com.")
......
......@@ -147,7 +147,8 @@ class XfroutSession():
if rcode == Rcode.NOERROR():
# ACL checks
acl_result = self._acl.execute(
isc.acl.dns.RequestContext(self._remote))
isc.acl.dns.RequestContext(self._remote,
msg.get_tsig_record()))
if acl_result == DROP:
logger.info(XFROUT_QUERY_DROPPED,
self._get_query_zone_name(msg),
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment