# -*- coding: utf-8 -*- # Copyright (c) 2015 OpenStack Foundation # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or # implied. # See the License for the specific language governing permissions and # limitations under the License. import base64 from unittest import mock import unittest import sys sys.modules['castellan'] = mock.Mock() sys.modules['castellan.common'] = mock.Mock() sys.modules['castellan.common.credentials'] = mock.Mock() from keystoneauth1.exceptions.connection import ConnectFailure from keystoneauth1.exceptions.http import Unauthorized from keystoneclient.exceptions import DiscoveryFailure from swift.common.middleware.crypto import kms_keymaster from swift.common.swob import Request from test.unit.common.middleware.helpers import FakeSwift, FakeAppThatExcepts TEST_KMS_INVALID_KEY_ID = 'invalid-kms-key-id' TEST_KMS_NONEXISTENT_KEY_ID = '11111111-1111-1111-1111-ffffffffffff' TEST_KMS_OPAQUE_KEY_ID = '22222222-2222-2222-2222-aaaaaaaaaaaa' TEST_KMS_SHORT_KEY_ID = '22222222-2222-2222-2222-bbbbbbbbbbbb' TEST_KMS_DES_KEY_ID = '22222222-2222-2222-2222-cccccccccccc' TEST_KMS_NONE_KEY_ID = '22222222-2222-2222-2222-dddddddddddd' TEST_KMS_INVALID_API_VERSION = 'vBadVersion' TEST_KMS_INVALID_USER_DOMAIN_NAME = "baduserdomainname" TEST_KMS_CONNECT_FAILURE_URL = 'http://endpoint_url_connect_error:45621' TEST_KMS_NON_BARBICAN_URL = 'http://endpoint_url_nonbarbican:45621' TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF = { 'keymaster_config_path': 'PATH_TO_KEYMASTER_CONFIG_FILE', } TEST_KMS_KEYMASTER_CONF = { 'auth_endpoint': 'kmsauthurlv3', 'password': 'kmspass', 'username': 'kmsuser', 'user_domain_id': None, 'user_domain_name': 'default', 'project_id': None, 'project_name': 'kmsproject', 'project_domain_id': None, 'project_domain_name': 'default', 'key_id': 'valid_kms_key_id-abcdefg-123456' } def capture_start_response(): calls = [] def start_response(*args): calls.append(args) return start_response, calls def mock_castellan_api_side_effect(*args, **kwargs): return MockBarbicanKeyManager(args[0]) def mock_options_set_defaults_side_effect(*args, **kwargs): ''' Add options from kwargs into args dict. ''' args[0].update(kwargs) def mock_config_opts_side_effect(*args, **kwargs): return dict() def mock_keystone_password_side_effect(auth_url, username, password, project_name, user_domain_name, project_domain_name, user_id, user_domain_id, trust_id, domain_id, domain_name, project_id, project_domain_id, reauthenticate): return MockPassword(auth_url, username, password, project_name, user_domain_name, project_domain_name, user_id, user_domain_id, trust_id, domain_id, domain_name, project_id, project_domain_id, reauthenticate) ERR_MESSAGE_SECRET_INCORRECTLY_SPECIFIED = 'Secret incorrectly specified.' ERR_MESSAGE_KEY_UUID_NOT_FOUND = 'Key not found, uuid: ' class MockBarbicanKeyManager(object): def __init__(self, conf): self.conf = conf def get(self, ctxt, key_id): # If authentication fails, raise an exception here. if (TEST_KMS_KEYMASTER_CONF['username'] != ctxt.username or TEST_KMS_KEYMASTER_CONF['password'] != ctxt.password or TEST_KMS_KEYMASTER_CONF['user_domain_name'] != ctxt.user_domain_name): raise Unauthorized( message='The request you have made requires authentication.', http_status=401) elif self.conf['auth_endpoint'] == TEST_KMS_CONNECT_FAILURE_URL: raise ConnectFailure('Unable to establish connection') elif self.conf['auth_endpoint'] == TEST_KMS_NON_BARBICAN_URL: raise DiscoveryFailure( 'Could not determine a suitable URL for the plugin') elif (self.conf['auth_endpoint'] != TEST_KMS_KEYMASTER_CONF['auth_endpoint']): raise Unauthorized( message='Cannot authorize API client.') elif (key_id == TEST_KMS_NONEXISTENT_KEY_ID): message = ERR_MESSAGE_KEY_UUID_NOT_FOUND + key_id ''' Raising a ManagedObjectNotFoundError would require importing it from castellan.common.exception. To avoid this import, raising a general Exception. ''' raise Exception(message) elif key_id == TEST_KMS_INVALID_KEY_ID: raise ValueError(ERR_MESSAGE_SECRET_INCORRECTLY_SPECIFIED) elif key_id == TEST_KMS_NONE_KEY_ID: return None if 'unicode' in key_id: key_str = key_id[0] * 32 else: key_str = (str(key_id[0]) * 32).encode('utf8') return MockBarbicanKey(key_str, key_id) class MockBarbicanKey(object): def __init__(self, key_material, key_id): self.key_material = key_material self.bit_length = len(key_material) * 8 if key_id == TEST_KMS_OPAQUE_KEY_ID: self.format = 'Opaque' else: self.format = 'RAW' self.algorithm = "aes" if key_id == TEST_KMS_DES_KEY_ID: self.format = 'des' if key_id == TEST_KMS_SHORT_KEY_ID: self.bit_length = 128 self.key_material[:128] def get_encoded(self): return self.key_material def format(self): return self.format class MockPassword(object): def __init__(self, auth_url, username, password, project_name, user_domain_name, project_domain_name, user_id, user_domain_id, trust_id, domain_id, domain_name, project_id, project_domain_id, reauthenticate): self.auth_url = auth_url self.password = password self.username = username self.user_domain_name = user_domain_name self.project_name = project_name self.project_domain_name = project_domain_name self.user_id = user_id, self.user_domain_id = user_domain_id, self.trust_id = trust_id, self.domain_id = domain_id, self.domain_name = domain_name, self.project_id = project_id, self.project_domain_id = project_domain_id, self.reauthenticate = reauthenticate class TestKmsKeymaster(unittest.TestCase): """ Unit tests for storing the encryption root secret in a Barbican external key management system accessed using Castellan. """ def setUp(self): super(TestKmsKeymaster, self).setUp() self.swift = FakeSwift() """ Tests using the v3 Identity API, where all calls to Barbican are mocked. """ @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch.object(kms_keymaster.KmsKeyMaster, '_get_root_secret') def test_filter_v3(self, mock_get_root_secret_from_kms, mock_readconf): mock_get_root_secret_from_kms.return_value = ( base64.b64encode(b'x' * 32)) mock_readconf.return_value = TEST_KMS_KEYMASTER_CONF factory = kms_keymaster.filter_factory(TEST_KMS_KEYMASTER_CONF) self.assertTrue(callable(factory)) self.assertTrue(callable(factory(self.swift))) @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch.object(kms_keymaster.KmsKeyMaster, '_get_root_secret') def test_app_exception_v3(self, mock_get_root_secret_from_kms, mock_readconf): mock_get_root_secret_from_kms.return_value = ( base64.b64encode(b'x' * 32)) mock_readconf.return_value = TEST_KMS_KEYMASTER_CONF app = kms_keymaster.KmsKeyMaster( FakeAppThatExcepts(), TEST_KMS_KEYMASTER_CONF) req = Request.blank('/', environ={'REQUEST_METHOD': 'PUT'}) start_response, _ = capture_start_response() self.assertRaises(Exception, app, req.environ, start_response) @mock.patch.object(kms_keymaster.KmsKeyMaster, '_get_root_secret') def test_get_root_secret( self, mock_get_root_secret_from_kms): # Successful call with coarse _get_root_secret_from_kms() mock. mock_get_root_secret_from_kms.return_value = ( base64.b64encode(b'x' * 32)) # Provide valid Barbican configuration parameters in proxy-server # config. self.app = kms_keymaster.KmsKeyMaster(self.swift, TEST_KMS_KEYMASTER_CONF) # Verify that _get_root_secret_from_kms() was called with the # correct parameters. mock_get_root_secret_from_kms.assert_called_with( TEST_KMS_KEYMASTER_CONF ) @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch.object(kms_keymaster.KmsKeyMaster, '_get_root_secret') def test_get_root_secret_from_external_file( self, mock_get_root_secret_from_kms, mock_readconf): # Return valid Barbican configuration parameters. mock_readconf.return_value = TEST_KMS_KEYMASTER_CONF # Successful call with coarse _get_root_secret_from_kms() mock. mock_get_root_secret_from_kms.return_value = ( base64.b64encode(b'x' * 32)) # Point to external config in proxy-server config. self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) # Verify that _get_root_secret_from_kms() was called with the # correct parameters. mock_get_root_secret_from_kms.assert_called_with( TEST_KMS_KEYMASTER_CONF ) self.assertEqual(mock_readconf.mock_calls, [ mock.call('PATH_TO_KEYMASTER_CONFIG_FILE', 'kms_keymaster')]) @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Successful call with finer grained mocks. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return valid Barbican configuration parameters. ''' mock_readconf.return_value = TEST_KMS_KEYMASTER_CONF ''' Verify that no exceptions are raised by the mocked functions. ''' try: self.app = kms_keymaster.KmsKeyMaster(self.swift, TEST_KMS_KEYMASTER_CONF) except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_invalid_key_id( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Invalid key ID. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['key_id'] = TEST_KMS_INVALID_KEY_ID mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though key id invalid') except ValueError as e: self.assertEqual(e.args[0], ERR_MESSAGE_SECRET_INCORRECTLY_SPECIFIED) except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_nonexistent_key_id( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Nonexistent key. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['key_id'] = TEST_KMS_NONEXISTENT_KEY_ID mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though key id invalid') except Exception as e: expected_message = ('Key not found, uuid: ' + TEST_KMS_NONEXISTENT_KEY_ID) self.assertEqual(e.args[0], expected_message) @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_invalid_key_format( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Nonexistent key. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['key_id'] = TEST_KMS_OPAQUE_KEY_ID mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though key format invalid') except ValueError: pass except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_config_file_and_params( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Both external config file and config parameters specified. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['keymaster_config_path'] = ( 'PATH_TO_KEYMASTER_CONFIG_FILE' ) mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster(self.swift, kms_conf) raise Exception('Success even though config invalid') except Exception as e: expected_message = ('keymaster_config_path is set, but there are ' 'other config options specified:') self.assertTrue(e.args[0].startswith(expected_message), "Error message does not start with '%s'" % expected_message) @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_invalid_username( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Invalid username. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['username'] = 'invaliduser' mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though username invalid') except Unauthorized as e: self.assertEqual(e.http_status, 401) except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_invalid_password( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Invalid password. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['password'] = 'invalidpassword' mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though password invalid') except Unauthorized as e: self.assertEqual(e.http_status, 401) except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_connect_failure_auth_url( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Connect failure kms auth_url. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['auth_endpoint'] = TEST_KMS_CONNECT_FAILURE_URL mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though auth_url invalid') except ConnectFailure: pass except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_bad_auth_url( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Bad kms auth_url. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['auth_endpoint'] = TEST_KMS_NON_BARBICAN_URL mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though auth_url invalid') except DiscoveryFailure: pass except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_bad_user_domain_name( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Bad user domain name with mocks. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['user_domain_name'] = ( TEST_KMS_INVALID_USER_DOMAIN_NAME) mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though api_version invalid') except Unauthorized as e: self.assertEqual(e.http_status, 401) except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_invalid_key_algorithm( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Nonexistent key. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['key_id'] = TEST_KMS_DES_KEY_ID mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though key format invalid') except ValueError: pass except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_invalid_key_length( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Nonexistent key. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['key_id'] = TEST_KMS_SHORT_KEY_ID mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though key format invalid') except ValueError: pass except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword') @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_mocked_castellan_keymanager_none_key( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config, mock_keystone_password): # Nonexistent key. mock_keystone_password.side_effect = ( mock_keystone_password_side_effect) ''' Set side_effect functions. ''' mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) ''' Return invalid Barbican configuration parameters. ''' kms_conf = dict(TEST_KMS_KEYMASTER_CONF) kms_conf['key_id'] = TEST_KMS_NONE_KEY_ID mock_readconf.return_value = kms_conf ''' Verify that an exception is raised by the mocked function. ''' try: self.app = kms_keymaster.KmsKeyMaster( self.swift, TEST_PROXYSERVER_CONF_EXTERNAL_KEYMASTER_CONF) raise Exception('Success even though None key returned') except ValueError: pass except Exception: print("Unexpected error: %s" % sys.exc_info()[0]) raise @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword', MockPassword) @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_get_root_secret_multiple_keys( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config,): config = dict(TEST_KMS_KEYMASTER_CONF) config.update({ 'key_id_foo': 'foo-valid_kms_key_id-123456', 'key_id_bar': 'bar-valid_kms_key_id-123456', 'key_id_baz': 'zz-valid_unicode_kms_key_id-123456', 'key_id_non_ascii': u'\N{SNOWMAN}_unicode_key_id', 'active_root_secret_id': 'foo'}) # Set side_effect functions. mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) # Return valid Barbican configuration parameters. mock_readconf.return_value = config self.app = kms_keymaster.KmsKeyMaster(self.swift, config) expected_secrets = { None: b'vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv', 'foo': b'ffffffffffffffffffffffffffffffff', 'bar': b'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', 'baz': b'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz', 'non_ascii': b'\xe2\x98\x83' * 32} self.assertDictEqual(self.app._root_secrets, expected_secrets) self.assertEqual(self.app.active_secret_id, 'foo') @mock.patch('swift.common.middleware.crypto.kms_keymaster.' 'keystone_password.KeystonePassword', MockPassword) @mock.patch('swift.common.middleware.crypto.kms_keymaster.cfg') @mock.patch('swift.common.middleware.crypto.kms_keymaster.options') @mock.patch('swift.common.middleware.crypto.keymaster.readconf') @mock.patch('swift.common.middleware.crypto.kms_keymaster.key_manager') def test_get_root_secret_legacy_key_id( self, mock_castellan_key_manager, mock_readconf, mock_castellan_options, mock_oslo_config): # Set side_effect functions. mock_castellan_key_manager.API.side_effect = ( mock_castellan_api_side_effect) mock_castellan_options.set_defaults.side_effect = ( mock_options_set_defaults_side_effect) mock_oslo_config.ConfigOpts.side_effect = ( mock_config_opts_side_effect) # Return valid Barbican configuration parameters. mock_readconf.return_value = TEST_KMS_KEYMASTER_CONF self.app = kms_keymaster.KmsKeyMaster(self.swift, TEST_KMS_KEYMASTER_CONF) expected_secrets = {None: b'vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv'} self.assertDictEqual(self.app._root_secrets, expected_secrets) self.assertIsNone(self.app.active_secret_id) if __name__ == '__main__': unittest.main()