tyler.durden
/
PyKMIP
mirror of https://github.com/OpenKMIP/PyKMIP.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #423 from OpenKMIP/feat/add-db-file-config 

Add server support for customizing the backend storage file
This commit is contained in:
Peter Hamilton 2018-04-16 14:58:43 -04:00 committed by GitHub
parent f1ccdf9c5a 2e6384a067
commit ea7da73b4f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 119 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
kmip/services/server/config.py
View File

@ -51,7 +51,8 @@ class KmipServerConfig(object):
'policy_path',
'enable_tls_client_auth',
'tls_cipher_suites',
'logging_level'
'logging_level',
'database_path'
]
def set_setting(self, setting, value):
@ -93,8 +94,10 @@ class KmipServerConfig(object):
self._set_enable_tls_client_auth(value)
elif setting == 'tls_cipher_suites':
self._set_tls_cipher_suites(value)
else:
elif setting == 'logging_level':
self._set_logging_level(value)
else:
self._set_database_path(value)
def load_settings(self, path):
"""
@ -179,6 +182,8 @@ class KmipServerConfig(object):
self._set_logging_level(
parser.get('server', 'logging_level')
)
if parser.has_option('server', 'database_path'):
self._set_database_path(parser.get('server', 'database_path'))
def _set_hostname(self, value):
if isinstance(value, six.string_types):
@ -334,3 +339,14 @@ class KmipServerConfig(object):
"The logging level must be a string representing a valid "
"logging level."
)
def _set_database_path(self, value):
if not value:
self.settings['database_path'] = None
elif isinstance(value, six.string_types):
self.settings['database_path'] = value
else:
raise exceptions.ConfigurationError(
"The database path, if specified, must be a valid path to a "
"SQLite database file."
)

11
kmip/services/server/engine.py
View File

@ -74,7 +74,7 @@ class KmipEngine(object):
* Cryptographic usage mask enforcement per object type
"""
def __init__(self, policies=None):
def __init__(self, policies=None, database_path=None):
"""
Create a KmipEngine.
@ -82,13 +82,20 @@ class KmipEngine(object):
policy_path (string): The path to the filesystem directory
containing PyKMIP server operation policy JSON files.
Optional, defaults to None.
database_path (string): The path to the SQLite database file
used to store all server data. Optional, defaults to None.
If none, database path defaults to '/tmp/pykmip.database'.
"""
self._logger = logging.getLogger('kmip.server.engine')
self._cryptography_engine = engine.CryptographyEngine()
self.database_path = 'sqlite:///{}'.format(database_path)
if not database_path:
self.database_path = 'sqlite:////tmp/pykmip.database'
self._data_store = sqlalchemy.create_engine(
'sqlite:////tmp/pykmip.database',
self.database_path,
echo=False,
connect_args={'check_same_thread': False}
)

31
kmip/services/server/server.py
View File

@ -45,6 +45,7 @@ class KmipServer(object):
shutting down all server components upon receiving a termination signal.
"""
# TODO (peter-hamilton) Move to using **kwargs for all server parameters.
def __init__(
self,
hostname=None,
@ -59,7 +60,8 @@ class KmipServer(object):
enable_tls_client_auth=None,
tls_cipher_suites=None,
logging_level=None,
live_policies=False
live_policies=False,
database_path=None
):
"""
Create a KmipServer.
@ -123,6 +125,8 @@ class KmipServer(object):
policy directory should be actively monitored to autoload any
policy changes while the server is running. Optional, defaults
to False.
database_path (string): The path to the server's SQLite database
file. Optional, defaults to None.
"""
self._logger = logging.getLogger('kmip.server')
self._setup_logging(log_path)
@ -139,7 +143,8 @@ class KmipServer(object):
policy_path,
enable_tls_client_auth,
tls_cipher_suites,
logging_level
logging_level,
database_path
)
self.live_policies = live_policies
self.policies = {}
@ -188,7 +193,8 @@ class KmipServer(object):
policy_path=None,
enable_tls_client_auth=None,
tls_cipher_suites=None,
logging_level=None
logging_level=None,
database_path=None
):
if path:
self.config.load_settings(path)
@ -219,6 +225,8 @@ class KmipServer(object):
)
if logging_level:
self.config.set_setting('logging_level', logging_level)
if database_path:
self.config.set_setting('database_path', database_path)
def start(self):
"""
@ -252,7 +260,8 @@ class KmipServer(object):
self.policy_monitor.start()
self._engine = engine.KmipEngine(
policies=self.policies
policies=self.policies,
database_path=self.config.settings.get('database_path')
)
self._logger.info("Starting server socket handler.")
@ -604,6 +613,18 @@ def build_argument_parser():
"DEBUG, INFO). Optional, defaults to None."
)
)
parser.add_option(
"-d",
"--database_path",
action="store",
type="str",
default=None,
dest="database_path",
help=(
"A string representing a path to the server's SQLite database "
"file. Optional, defaults to None."
),
)
return parser
@ -636,6 +657,8 @@ def main(args=None):
kwargs['enable_tls_client_auth'] = False
if opts.logging_level:
kwargs['logging_level'] = opts.logging_level
if opts.database_path:
kwargs['database_path'] = opts.database_path
kwargs['live_policies'] = True

59
kmip/tests/unit/services/server/test_config.py
View File

@ -68,6 +68,7 @@ class TestKmipServerConfig(testtools.TestCase):
c._set_enable_tls_client_auth = mock.MagicMock()
c._set_tls_cipher_suites = mock.MagicMock()
c._set_logging_level = mock.MagicMock()
c._set_database_path = mock.MagicMock()
# Test the right error is generated when setting an unsupported
# setting.
@ -115,6 +116,9 @@ class TestKmipServerConfig(testtools.TestCase):
c.set_setting('logging_level', 'WARNING')
c._set_logging_level.assert_called_once_with('WARNING')
c.set_setting('database_path', '/var/pykmip/pykmip.db')
c._set_database_path.assert_called_once_with('/var/pykmip/pykmip.db')
def test_load_settings(self):
"""
Test that the right calls are made and the right errors generated when
@ -232,6 +236,7 @@ class TestKmipServerConfig(testtools.TestCase):
c._set_enable_tls_client_auth = mock.MagicMock()
c._set_tls_cipher_suites = mock.MagicMock()
c._set_logging_level = mock.MagicMock()
c._set_database_path = mock.MagicMock()
# Test that the right calls are made when correctly parsing settings.
parser = configparser.SafeConfigParser()
@ -250,6 +255,7 @@ class TestKmipServerConfig(testtools.TestCase):
"\n TLS_RSA_WITH_AES_256_CBC_SHA256"
)
parser.set('server', 'logging_level', 'ERROR')
parser.set('server', 'database_path', '/var/pykmip/pykmip.db')
c._parse_settings(parser)
@ -267,6 +273,7 @@ class TestKmipServerConfig(testtools.TestCase):
"\n TLS_RSA_WITH_AES_256_CBC_SHA256"
)
c._set_logging_level.assert_called_once_with('ERROR')
c._set_database_path.assert_called_once_with('/var/pykmip/pykmip.db')
# Test that a ConfigurationError is generated when the expected
# section is missing.
@ -819,3 +826,55 @@ class TestKmipServerConfig(testtools.TestCase):
c._set_logging_level,
*args
)
def test_set_database_path(self):
"""
Test that the database_path configuration property can be set
correctly.
"""
c = config.KmipServerConfig()
c._logger = mock.MagicMock()
self.assertNotIn('database_path', c.settings.keys())
with mock.patch('os.path.exists') as os_mock:
os_mock.return_value = True
c._set_database_path('/test/path/database.db')
self.assertIn('database_path', c.settings.keys())
self.assertEqual(
'/test/path/database.db',
c.settings.get('database_path')
)
def test_set_database_path_default(self):
"""
Test that the database_path configuration property can be set correctly
without specifying a value.
"""
c = config.KmipServerConfig()
c._logger = mock.MagicMock()
self.assertNotIn('database_path', c.settings.keys())
c._set_database_path(None)
self.assertIn('database_path', c.settings.keys())
self.assertEqual(None, c.settings.get('database_path'))
def test_set_database_path_invalid_value(self):
"""
Test that the right error is raised when an invalid value is used to
set the database_path configuration property.
"""
c = config.KmipServerConfig()
c._logger = mock.MagicMock()
self.assertNotIn('database_path', c.settings.keys())
args = (1, )
self.assertRaises(
exceptions.ConfigurationError,
c._set_database_path,
*args
)
self.assertNotEqual(1, c.settings.get('database_path'))

7
kmip/tests/unit/services/server/test_server.py
View File

@ -121,7 +121,8 @@ class TestKmipServer(testtools.TestCase):
'/etc/pykmip/policies',
False,
'TLS_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA',
'DEBUG'
'DEBUG',
'/var/pykmip/pykmip.db'
)
s.config.load_settings.assert_called_with('/etc/pykmip/server.conf')
@ -156,6 +157,10 @@ class TestKmipServer(testtools.TestCase):
]
)
s.config.set_setting.assert_any_call('logging_level', 'DEBUG')
s.config.set_setting.assert_any_call(
'database_path',
'/var/pykmip/pykmip.db'
)
# Test that an attempt is made to instantiate the TLS 1.2 auth suite
s = server.KmipServer(