"""
import unittest
-import isc.server_common.tsig_keyring
+from isc.server_common.tsig_keyring import *
+import isc.dns
+from isc.testutils.ccsession_mock import MockModuleCCSession
+
+class Session(MockModuleCCSession):
+ """
+ A class pretending to be the config session.
+ """
+ def __init__(self):
+ MockModuleCCSession.__init__(self)
+ self._name = None
+ self._callback = None
+ self._remove_name = None
+ self._data = None
+
+ def add_remote_config_by_name(self, name, callback):
+ self._name = name
+ self._callback = callback
+
+ def remove_remote_config(self, name):
+ self._remove_name = name
+
+ def get_remote_config_value(self, module, name):
+ if module != 'tsig_keys' or name != 'keys':
+ raise Exception("Asked for bad data element")
+ return self._data
+
+class TSIGKeyRingTest(unittest.TestCase):
+ """
+ Tests for the isc.server_common.tsig_keyring module.
+ """
+ def setUp(self):
+ self.__session = Session()
+ self.__sha1name = isc.dns.Name('hmac-sha1')
+
+ def tearDown(self):
+ deinit_keyring()
+
+ def __do_init(self):
+ init_keyring(self.__session)
+ # Some initialization happened
+ self.assertEqual('tsig_keys', self.__session._name)
+
+ def test_initialization(self):
+ """
+ Test we can initialize and deintialize the keyring. It also
+ tests the interaction with the keyring() function.
+ """
+ # The keyring function raises until initialized
+ self.assertRaises(Unexpected, keyring)
+ self.__do_init()
+ current_keyring = keyring()
+ self.assertTrue(isinstance(current_keyring, isc.dns.TSIGKeyRing))
+ # Another initialization does nothing
+ self.__do_init()
+ self.assertEqual(current_keyring, keyring())
+ # When we deinitialize it, it no longer provides the keyring
+ deinit_keyring()
+ self.assertEqual('tsig_keys', self.__session._remove_name)
+ self.__session._remove_name = None
+ self.assertRaises(Unexpected, keyring)
+ # Another deinitialization doesn't change anything
+ deinit_keyring()
+ self.assertRaises(Unexpected, keyring)
+ self.assertIsNone(self.__session._remove_name)
+
+ def test_load(self):
+ """
+ Test it can load the keys from the configuration and reload them
+ when the data change.
+ """
+ # Initial load
+ self.__session._data = ['key:MTIzNAo=:hmac-sha1']
+ self.__do_init()
+ keys = keyring()
+ self.assertEqual(1, keys.size())
+ (rcode, key) = keys.find(isc.dns.Name('key'), self.__sha1name)
+ self.assertEqual(isc.dns.TSIGKeyRing.SUCCESS, rcode)
+ self.assertEqual(isc.dns.Name('key'), key.get_key_name())
+ # There's a change in the configuration
+ # (The key has a different name)
+ self.__session._data = ['key.example:MTIzNAo=:hmac-sha1']
+ self.__session._callback()
+ orig_keys = keys
+ keys = keyring()
+ self.assertNotEqual(keys, orig_keys)
+ self.assertEqual(1, keys.size())
+ # The old key is not here
+ (rcode, key) = keys.find(isc.dns.Name('key'), self.__sha1name)
+ self.assertEqual(isc.dns.TSIGKeyRing.NOTFOUND, rcode)
+ self.assertIsNone(key)
+ # But the new one is
+ (rcode, key) = keys.find(isc.dns.Name('key.example'), self.__sha1name)
+ self.assertEqual(isc.dns.TSIGKeyRing.SUCCESS, rcode)
+ self.assertEqual(isc.dns.Name('key.example'), key.get_key_name())
+
+ def test_empty_update(self):
+ """
+ Test an update that doesn't carry the correct element doesn't change
+ anything.
+ """
+ self.__session._data = ['key:MTIzNAo=:hmac-sha1']
+ self.__do_init()
+ keys = keyring()
+ self.__session._data = None
+ self.__session._callback()
+ self.assertEqual(keys, keyring())
+
+ def test_update_bad(self):
+ """
+ Test it raises on bad updates and doesn't change anything.
+ """
+ self.__session._data = ['key:MTIzNAo=:hmac-sha1']
+ self.__do_init()
+ keys = keyring()
+ # Bad TSIG string
+ self.__session._data = ['key:this makes no sense:really']
+ self.assertRaises(isc.dns.InvalidParameter, self.__session._callback)
+ self.assertEqual(keys, keyring())
+ # A duplicity
+ self.__session._data = ['key:MTIzNAo=:hmac-sha1', 'key:MTIzNAo=:hmac-sha1']
+ self.assertRaises(AddError, self.__session._callback)
+ self.assertEqual(keys, keyring())
if __name__ == "__main__":
unittest.main()
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
# WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+"""
+This module conveniently keeps a copy of TSIG keyring loaded from the
+tsig_keys module.
+"""
+
+import isc.dns
+
+updater = None
+
+class Unexpected(Exception):
+ """
+ Raised when an unexpected operation is requested by the user of this
+ module. For example if calling keyring() before init_keyring().
+ """
+ pass
+
+class AddError(Exception):
+ """
+ Raised when an key can not be added. This usually means there's a
+ duplicity.
+ """
+ pass
+
+class Updater:
+ """
+ The updater of tsig key ring. Not to be used directly.
+ """
+ def __init__(self, session):
+ """
+ Constructor. Pass the ccsession object so the key ring can be
+ downloaded.
+ """
+ self._session = session
+ self._keyring = isc.dns.TSIGKeyRing()
+ session.add_remote_config_by_name('tsig_keys', self._update)
+ self._update()
+
+ def _update(self):
+ """
+ Update the key ring by the configuration.
+
+ Note that this function is used as a callback, but can raise
+ on bad data. The bad data is expected to be handled by the
+ configuration plugin and not be allowed as far as here.
+ """
+ data = self._session.get_remote_config_value('tsig_keys', 'keys')
+ if data is not None: # There's an update
+ keyring = isc.dns.TSIGKeyRing()
+ for key_data in data:
+ key = isc.dns.TSIGKey(key_data)
+ if keyring.add(key) != isc.dns.TSIGKeyRing.SUCCESS:
+ raise AddError("Can't add key " + str(key))
+ self._keyring = keyring
+
+ def get_keyring(self):
+ """
+ Return the current key ring.
+ """
+ return self._keyring
+
+ def deinit(self):
+ """
+ Unregister from getting updates. The object will not be
+ usable any more after this.
+ """
+ self._session.remove_remote_config('tsig_keys')
+
+def keyring():
+ """
+ Get the current key ring. You need to call init_keyring first.
+ """
+ if updater is None:
+ raise Unexpected("You need to initialize the keyring first by " +
+ "init_keyring()")
+ return updater.get_keyring()
+
+def init_keyring(session):
+ """
+ Initialize the key ring for future use. It does nothing if already
+ initialized.
+ """
+ global updater
+ if updater is None:
+ updater = Updater(session)
+
+def deinit_keyring():
+ """
+ Deinit key ring. Yoeu can no longer access keyring() after this.
+ Does nothing if not initialized.
+ """
+ global updater
+ if updater is not None:
+ updater.deinit()
+ updater = None
projects / thirdparty / kea.git / commitdiff
