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

Update the server session to use the auth plugin framework 

This change updates how the server session handles message
processing, adding support for the new authentication plugin
framework. Session unit tests have been updated to account for
this change.
This commit is contained in:
Peter Hamilton 2018-03-25 00:54:43 -04:00
parent f06014e7c6
commit 1a093f141e
5 changed files with 640 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
kmip/core/exceptions.py
View File

@ -38,6 +38,25 @@ class KmipError(Exception):
self.status = status
self.reason = reason
def __eq__(self, other):
if isinstance(other, KmipError):
if str(self) != str(other):
return False
elif self.status != other.status:
return False
elif self.reason != other.reason:
return False
else:
return True
else:
return NotImplemented
def __ne__(self, other):
if isinstance(other, KmipError):
return self == other
else:
return NotImplemented
class CryptographicFailure(KmipError):
"""

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

@ -335,9 +335,6 @@ class KmipEngine(object):
else:
auth_credentials = None
# TODO (peter-hamilton) This is a shim until SLUGS integration is done.
credential = [credential, None]
self._verify_credential(auth_credentials, credential)
# Process the batch error continuation option

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

@ -398,10 +398,12 @@ class KmipServer(object):
s = session.KmipSession(
self._engine,
connection,
address,
name=session_name,
enable_tls_client_auth=self.config.settings.get(
'enable_tls_client_auth'
)
),
auth_settings=self.config.settings.get('auth_plugins')
)
s.daemon = True
s.start()

221
kmip/services/server/session.py
View File

@ -17,9 +17,9 @@ import logging
import socket
import struct
import threading
import time
from cryptography import x509
from cryptography.hazmat import backends
from kmip.core import enums
from kmip.core import exceptions
@ -27,6 +27,8 @@ from kmip.core.messages import contents
from kmip.core.messages import messages
from kmip.core import utils
from kmip.services.server import auth
class KmipSession(threading.Thread):
"""
@ -36,8 +38,10 @@ class KmipSession(threading.Thread):
def __init__(self,
engine,
connection,
address,
name=None,
enable_tls_client_auth=True):
enable_tls_client_auth=True,
auth_settings=None):
"""
Create a KmipSession.
@ -46,12 +50,19 @@ class KmipSession(threading.Thread):
that handles message processing. Required.
connection (socket): A client socket.socket TLS connection
representing a new KMIP connection. Required.
address (tuple): The address tuple produced with the session
connection. Contains the IP address and port number of the
remote connection endpoint. Required.
name (str): The name of the KmipSession. Optional, defaults to
None.
enable_tls_client_auth (bool): A flag that enables a strict check
for the client auth flag in the extended key usage extension
in client certificates when establishing the client/server TLS
connection. Optional, defaults to True.
auth_settings (list): A list of tuples, each containing (1) the
name of the 'auth:' settings block from the server config file,
and (2) a dictionary of configuration settings for a specific
authentication plugin. Optional, defaults to None.
"""
super(KmipSession, self).__init__(
group=None,
@ -67,9 +78,12 @@ class KmipSession(threading.Thread):
self._engine = engine
self._connection = connection
self._address = address
self._enable_tls_client_auth = enable_tls_client_auth
self._auth_settings = [] if auth_settings is None else auth_settings
self._session_time = time.time()
self._max_buffer_size = 4096
self._max_request_size = 1048576
self._max_response_size = 1048576
@ -96,61 +110,6 @@ class KmipSession(threading.Thread):
self._connection.close()
self._logger.info("Stopping session: {0}".format(self.name))
def _get_client_identity(self):
certificate_data = self._connection.getpeercert(binary_form=True)
try:
cert = x509.load_der_x509_certificate(
certificate_data,
backends.default_backend()
)
except Exception:
# This should never get raised "in theory," as the ssl socket
# should fail to connect non-TLS connections before the session
# gets created. This is a failsafe in case that protection fails.
raise exceptions.PermissionDenied(
"Failure loading the client certificate from the session "
"connection. Could not retrieve client identity."
)
if self._enable_tls_client_auth:
try:
extended_key_usage = cert.extensions.get_extension_for_oid(
x509.oid.ExtensionOID.EXTENDED_KEY_USAGE
).value
except x509.ExtensionNotFound:
raise exceptions.PermissionDenied(
"The extended key usage extension is missing from the "
"client certificate. Session client identity unavailable."
)
if x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH not in \
extended_key_usage:
raise exceptions.PermissionDenied(
"The extended key usage extension is not marked for "
"client authentication in the client certificate. Session "
"client identity unavailable."
)
client_identities = cert.subject.get_attributes_for_oid(
x509.oid.NameOID.COMMON_NAME
)
if len(client_identities) > 0:
if len(client_identities) > 1:
self._logger.warning(
"Multiple client identities found. Using the first "
"one processed."
)
client_identity = client_identities[0].value
self._logger.info(
"Session client identity: {0}".format(client_identity)
)
return client_identity
else:
raise exceptions.PermissionDenied(
"The client certificate does not define a subject common "
"name. Session client identity unavailable."
)
def _handle_message_loop(self):
request_data = self._receive_request()
request = messages.RequestMessage()
@ -170,8 +129,41 @@ class KmipSession(threading.Thread):
self._connection.cipher()
)
)
client_identity = self._get_client_identity()
certificate = auth.get_certificate_from_connection(
self._connection
)
if certificate is None:
raise exceptions.PermissionDenied(
"The client certificate could not be loaded from the "
"session connection."
)
if self._enable_tls_client_auth:
extension = auth.get_extended_key_usage_from_certificate(
certificate
)
if extension is None:
raise exceptions.PermissionDenied(
"The extended key usage extension is missing from "
"the client certificate."
)
if x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH not in extension:
raise exceptions.PermissionDenied(
"The extended key usage extension is not marked for "
"client authentication in the client certificate."
)
request.read(request_data)
except exceptions.PermissionDenied as e:
self._logger.warning("Failure verifying the client certificate.")
self._logger.exception(e)
response = self._engine.build_error_response(
contents.ProtocolVersion(1, 0),
enums.ResultReason.AUTHENTICATION_NOT_SUCCESSFUL,
"Error verifying the client certificate. "
"See server logs for more information."
)
except Exception as e:
self._logger.warning("Failure parsing request message.")
self._logger.exception(e)
@ -183,29 +175,44 @@ class KmipSession(threading.Thread):
)
else:
try:
response, max_response_size = self._engine.process_request(
request,
client_identity
client_identity = self.authenticate(certificate, request)
self._logger.info(
"Session client identity: {}".format(client_identity[0])
)
if max_response_size:
max_size = max_response_size
except exceptions.KmipError as e:
except Exception:
self._logger.warning("Authentication failed.")
response = self._engine.build_error_response(
request.request_header.protocol_version,
e.reason,
str(e)
)
except Exception as e:
self._logger.warning(
"An unexpected error occurred while processing request."
)
self._logger.exception(e)
response = self._engine.build_error_response(
request.request_header.protocol_version,
enums.ResultReason.GENERAL_FAILURE,
"An unexpected error occurred while processing request. "
enums.ResultReason.AUTHENTICATION_NOT_SUCCESSFUL,
"An error occurred during client authentication. "
"See server logs for more information."
)
else:
try:
response, max_response_size = self._engine.process_request(
request,
client_identity
)
if max_response_size:
max_size = max_response_size
except exceptions.KmipError as e:
response = self._engine.build_error_response(
request.request_header.protocol_version,
e.reason,
str(e)
)
except Exception as e:
self._logger.warning(
"An unexpected error occurred while processing "
"request."
)
self._logger.exception(e)
response = self._engine.build_error_response(
request.request_header.protocol_version,
enums.ResultReason.GENERAL_FAILURE,
"An unexpected error occurred while processing "
"request. See server logs for more information."
)
response_data = utils.BytearrayStream()
response.write(response_data)
@ -229,6 +236,68 @@ class KmipSession(threading.Thread):
self._send_response(response_data.buffer)
def authenticate(self, certificate, request):
credentials = []
if request.request_header.authentication is not None:
credentials = request.request_header.authentication.credentials
plugin_enabled = False
for auth_settings in self._auth_settings:
plugin_name, plugin_config = auth_settings
if plugin_name.startswith("auth:slugs"):
if plugin_config.get("enabled") == "True":
plugin_enabled = True
plugin = auth.SLUGSConnector(plugin_config.get("url"))
self._logger.debug(
"Authenticating with plugin: {}".format(plugin_name)
)
try:
client_identity = plugin.authenticate(
certificate,
(self._address, self._session_time),
credentials
)
except Exception as e:
self._logger.warning(
"Authentication failed."
)
self._logger.exception(e)
else:
self._logger.debug(
"Authentication succeeded for client identity: "
"{}".format(client_identity[0])
)
return client_identity
else:
self._logger.warning(
"Authentication plugin '{}' is not "
"supported.".format(plugin_name)
)
if not plugin_enabled:
self._logger.debug(
"No authentication plugins are enabled. The client identity "
"will be extracted from the client certificate."
)
try:
client_identity = auth.get_client_identity_from_certificate(
certificate
)
except Exception as e:
self._logger.warning("Client identity extraction failed.")
self._logger.exception(e)
else:
self._logger.debug(
"Extraction succeeded for client identity: {}".format(
client_identity
)
)
return tuple([client_identity, None])
raise exceptions.PermissionDenied("Authentication failed.")
def _receive_request(self):
header = self._receive_bytes(8)
message_size = struct.unpack('!I', header[4:])[0]

640
kmip/tests/unit/services/server/test_session.py
View File

@ -16,7 +16,6 @@
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
@ -28,6 +27,7 @@ import time
from kmip.core import enums
from kmip.core import exceptions
from kmip.core import objects
from kmip.core import utils
from kmip.core.messages import contents
@ -122,20 +122,20 @@ class TestKmipSession(testtools.TestCase):
"""
Test that a KmipSession can be created without errors.
"""
session.KmipSession(None, None, 'name')
session.KmipSession(None, None, None, 'name')
def test_init_without_name(self):
"""
Test that a KmipSession without 'name' can be created without errors.
"""
session.KmipSession(None, None, None)
session.KmipSession(None, None, None, None)
def test_run(self):
"""
Test that the message handling loop is handled properly on normal
execution.
"""
kmip_session = session.KmipSession(None, None, 'name')
kmip_session = session.KmipSession(None, None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._handle_message_loop = mock.MagicMock(
side_effect=[
@ -160,7 +160,7 @@ class TestKmipSession(testtools.TestCase):
Test that the correct logging and error handling occurs when the
thread encounters an error with the message handling loop.
"""
kmip_session = session.KmipSession(None, None, 'name')
kmip_session = session.KmipSession(None, None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
@ -186,147 +186,9 @@ class TestKmipSession(testtools.TestCase):
kmip_session._connection.close.assert_called_once_with()
kmip_session._logger.info.assert_called_with("Stopping session: name")
def test_get_client_identity(self):
"""
Test that a client identity is obtained from a valid client
certificate.
"""
client_certificate = build_certificate([u'Test Identity'])
der_encoding = client_certificate.public_bytes(
serialization.Encoding.DER
)
kmip_session = session.KmipSession(None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._connection.getpeercert.return_value = der_encoding
identity = kmip_session._get_client_identity()
self.assertEqual(u'Test Identity', identity)
kmip_session._logger.info.assert_called_once_with(
"Session client identity: Test Identity"
)
kmip_session._logger.warning.assert_not_called()
def test_get_client_identity_with_no_certificate(self):
"""
Test that the right error is generated when no certificate is
available to provide the client identity.
"""
kmip_session = session.KmipSession(None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._connection.getpeercert.return_value = None
self.assertRaisesRegexp(
exceptions.PermissionDenied,
"Failure loading the client certificate from the session "
"connection. Could not retrieve client identity.",
kmip_session._get_client_identity
)
def test_get_client_identity_with_no_extended_key_usage(self):
"""
Test that the right error is generated when the client certificate
is missing its extended key usage extension.
"""
client_certificate = build_certificate([u'Test Identity'], False)
der_encoding = client_certificate.public_bytes(
serialization.Encoding.DER
)
kmip_session = session.KmipSession(None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._connection.getpeercert.return_value = der_encoding
self.assertRaisesRegexp(
exceptions.PermissionDenied,
"The extended key usage extension is missing from the client "
"certificate. Session client identity unavailable.",
kmip_session._get_client_identity
)
def test_get_client_identity_with_no_common_name(self):
"""
Test that the right error is generated when the client certificate
does not define a subject common name.
"""
client_certificate = build_certificate([])
der_encoding = client_certificate.public_bytes(
serialization.Encoding.DER
)
kmip_session = session.KmipSession(None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._connection.getpeercert.return_value = der_encoding
self.assertRaisesRegexp(
exceptions.PermissionDenied,
"The client certificate does not define a subject common "
"name. Session client identity unavailable.",
kmip_session._get_client_identity
)
def test_get_client_identity_with_multiple_common_names(self):
"""
Test that the right client identity is returned when the client
certificate has multiple subject common names.
"""
client_certificate = build_certificate([
u'Test Identity 1',
u'Test Identity 2'
])
der_encoding = client_certificate.public_bytes(
serialization.Encoding.DER
)
kmip_session = session.KmipSession(None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._connection.getpeercert.return_value = der_encoding
identity = kmip_session._get_client_identity()
self.assertEqual(u'Test Identity 1', identity)
kmip_session._logger.info.assert_called_once_with(
"Session client identity: Test Identity 1"
)
kmip_session._logger.warning.assert_called_once_with(
"Multiple client identities found. Using the first one processed."
)
def test_get_client_identity_with_incorrect_extended_key_usage(self):
"""
Test that the right error is generated when the client certificate
does not have client authentication set in its extended key usage
extension.
"""
client_certificate = build_certificate(
[u'Test Identity'],
bad_extension=True
)
der_encoding = client_certificate.public_bytes(
serialization.Encoding.DER
)
kmip_session = session.KmipSession(None, None, 'name')
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._connection.getpeercert.return_value = der_encoding
self.assertRaisesRegexp(
exceptions.PermissionDenied,
"The extended key usage extension is not marked for client "
"authentication in the client certificate. Session client "
"identity unavailable.",
kmip_session._get_client_identity
)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage')
def test_handle_message_loop(self, request_mock):
def test_handle_message_loop(self, request_mock, cert_mock):
"""
Test that the correct logging and error handling occurs during the
message handling loop.
@ -354,12 +216,22 @@ class TestKmipSession(testtools.TestCase):
batch_items=batch_items
)
cert_mock.return_value = 'test_certificate'
kmip_engine = engine.KmipEngine()
kmip_engine._logger = mock.MagicMock()
kmip_session = session.KmipSession(kmip_engine, None, 'name')
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=False
)
kmip_session._engine = mock.MagicMock()
kmip_session._get_client_identity = mock.MagicMock()
kmip_session._get_client_identity.return_value = 'test'
kmip_session.authenticate = mock.MagicMock()
kmip_session.authenticate.return_value = (
'test',
['group A', 'group B']
)
kmip_session._engine.process_request = mock.MagicMock(
return_value=(message, kmip_session._max_response_size)
)
@ -376,11 +248,16 @@ class TestKmipSession(testtools.TestCase):
)
kmip_session._receive_request = mock.MagicMock(return_value=data)
kmip_session._send_response = mock.MagicMock()
kmip_session.authenticate = mock.MagicMock(
return_value=("John Doe", ["Group A"])
)
kmip_session._handle_message_loop()
kmip_session._receive_request.assert_called_once_with()
kmip_session._logger.info.assert_not_called()
kmip_session._logger.info.assert_any_call(
"Session client identity: John Doe"
)
kmip_session._logger.debug.assert_any_call(
"Possible session ciphers: 2"
)
@ -399,19 +276,30 @@ class TestKmipSession(testtools.TestCase):
kmip_session._logger.exception.assert_not_called()
self.assertTrue(kmip_session._send_response.called)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage.read',
mock.MagicMock(side_effect=Exception()))
def test_handle_message_loop_with_parse_failure(self):
def test_handle_message_loop_with_parse_failure(self, cert_mock):
"""
Test that the correct logging and error handling occurs during the
message handling loop.
"""
data = utils.BytearrayStream(())
cert_mock.return_value = 'test_certificate'
kmip_engine = engine.KmipEngine()
kmip_session = session.KmipSession(kmip_engine, None, 'name')
kmip_session._get_client_identity = mock.MagicMock()
kmip_session._get_client_identity.return_value = 'test'
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=False
)
kmip_session.authenticate = mock.MagicMock()
kmip_session.authenticate.return_value = (
'test',
['group A', 'group B']
)
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._receive_request = mock.MagicMock(return_value=data)
@ -427,18 +315,31 @@ class TestKmipSession(testtools.TestCase):
kmip_session._logger.error.assert_not_called()
self.assertTrue(kmip_session._send_response.called)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage')
def test_handle_message_loop_with_response_too_long(self, request_mock):
def test_handle_message_loop_with_response_too_long(self,
request_mock,
cert_mock):
"""
Test that the correct logging and error handling occurs during the
message handling loop.
"""
data = utils.BytearrayStream(())
cert_mock.return_value = 'test_certificate'
kmip_engine = engine.KmipEngine()
kmip_session = session.KmipSession(kmip_engine, None, 'name')
kmip_session._get_client_identity = mock.MagicMock()
kmip_session._get_client_identity.return_value = 'test'
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=False
)
kmip_session.authenticate = mock.MagicMock()
kmip_session.authenticate.return_value = (
'test',
['group A', 'group B']
)
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._receive_request = mock.MagicMock(return_value=data)
@ -448,24 +349,36 @@ class TestKmipSession(testtools.TestCase):
kmip_session._handle_message_loop()
kmip_session._receive_request.assert_called_once_with()
# kmip_session._logger.info.assert_not_called()
self.assertTrue(kmip_session._logger.warning.called)
kmip_session._logger.exception.assert_not_called()
self.assertTrue(kmip_session._send_response.called)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage')
def test_handle_message_loop_with_unexpected_error(self, request_mock):
def test_handle_message_loop_with_unexpected_error(self,
request_mock,
cert_mock):
"""
Test that the correct logging and error handling occurs when an
unexpected error is generated while processing a request.
"""
data = utils.BytearrayStream(())
cert_mock.return_value = 'test_certificate'
kmip_engine = engine.KmipEngine()
kmip_engine._logger = mock.MagicMock()
kmip_session = session.KmipSession(kmip_engine, None, 'name')
kmip_session._get_client_identity = mock.MagicMock()
kmip_session._get_client_identity.return_value = 'test'
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=False
)
kmip_session.authenticate = mock.MagicMock()
kmip_session.authenticate.return_value = (
'test',
['group A', 'group B']
)
kmip_session._engine = mock.MagicMock()
test_exception = Exception("Unexpected error.")
kmip_session._engine.process_request = mock.MagicMock(
@ -479,13 +392,406 @@ class TestKmipSession(testtools.TestCase):
kmip_session._handle_message_loop()
kmip_session._receive_request.assert_called_once_with()
# kmip_session._logger.info.assert_not_called()
kmip_session._logger.warning.assert_called_once_with(
"An unexpected error occurred while processing request."
)
kmip_session._logger.exception.assert_called_once_with(test_exception)
self.assertTrue(kmip_session._send_response.called)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage')
def test_handle_message_loop_with_authentication_failure(self,
request_mock,
cert_mock):
"""
Test that the correct logging and error handling occurs when an
authentication error is generated while processing a request.
"""
data = utils.BytearrayStream(())
cert_mock.return_value = 'test_certificate'
kmip_engine = engine.KmipEngine()
kmip_engine._logger = mock.MagicMock()
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=False
)
kmip_session.authenticate = mock.MagicMock()
kmip_session.authenticate.side_effect = exceptions.PermissionDenied(
"Authentication failed."
)
kmip_session._engine = mock.MagicMock()
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._receive_request = mock.MagicMock(return_value=data)
kmip_session._send_response = mock.MagicMock()
fake_version = contents.ProtocolVersion(1, 2)
fake_credential = objects.Credential(
credential_type=enums.CredentialType.USERNAME_AND_PASSWORD,
credential_value=objects.UsernamePasswordCredential(
username="John Doe",
password="secret"
)
)
fake_header = messages.RequestHeader(
protocol_version=fake_version,
authentication=contents.Authentication(
credentials=[fake_credential]
)
)
fake_request = messages.RequestMessage()
fake_request.request_header = fake_header
fake_request.read = mock.MagicMock()
request_mock.return_value = fake_request
kmip_session._handle_message_loop()
kmip_session._receive_request.assert_called_once_with()
fake_request.read.assert_called_once_with(data)
kmip_session.authenticate.assert_called_once_with(
"test_certificate",
fake_request
)
kmip_session._logger.warning.assert_called_once_with(
"Authentication failed."
)
kmip_session._engine.build_error_response.assert_called_once_with(
fake_version,
enums.ResultReason.AUTHENTICATION_NOT_SUCCESSFUL,
"An error occurred during client authentication. "
"See server logs for more information."
)
kmip_session._logger.exception.assert_not_called()
self.assertTrue(kmip_session._send_response.called)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage')
def test_handle_message_loop_no_certificate(self,
request_mock,
cert_mock):
"""
Test that the correct logging and error handling occurs when no
certificate is encountered while processing a request.
"""
data = utils.BytearrayStream(())
cert_mock.return_value = None
kmip_engine = engine.KmipEngine()
kmip_engine._logger = mock.MagicMock()
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=True
)
kmip_session.authenticate = mock.MagicMock()
kmip_session._engine = mock.MagicMock()
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._receive_request = mock.MagicMock(return_value=data)
kmip_session._send_response = mock.MagicMock()
kmip_session._handle_message_loop()
kmip_session._receive_request.assert_called_once_with()
kmip_session._logger.warning(
"Failure verifying the client certificate."
)
kmip_session._logger.exception.assert_called_once_with(
exceptions.PermissionDenied(
"The client certificate could not be loaded from the session "
"connection."
)
)
kmip_session._engine.build_error_response.assert_called_once_with(
contents.ProtocolVersion(1, 0),
enums.ResultReason.AUTHENTICATION_NOT_SUCCESSFUL,
"Error verifying the client certificate. "
"See server logs for more information."
)
self.assertTrue(kmip_session._send_response.called)
@mock.patch(
'kmip.services.server.auth.get_extended_key_usage_from_certificate'
)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage')
def test_handle_message_loop_no_certificate_extension(self,
request_mock,
cert_mock,
ext_mock):
"""
Test that the correct logging and error handling occurs when an
invalid certificate is encountered while processing a request.
"""
data = utils.BytearrayStream(())
cert_mock.return_value = 'test_certificate'
ext_mock.return_value = None
kmip_engine = engine.KmipEngine()
kmip_engine._logger = mock.MagicMock()
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=True
)
kmip_session.authenticate = mock.MagicMock()
kmip_session._engine = mock.MagicMock()
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._receive_request = mock.MagicMock(return_value=data)
kmip_session._send_response = mock.MagicMock()
kmip_session._handle_message_loop()
kmip_session._receive_request.assert_called_once_with()
kmip_session._logger.warning(
"Failure verifying the client certificate."
)
kmip_session._logger.exception.assert_called_once_with(
exceptions.PermissionDenied(
"The extended key usage extension is missing from the client "
"certificate."
)
)
kmip_session._engine.build_error_response.assert_called_once_with(
contents.ProtocolVersion(1, 0),
enums.ResultReason.AUTHENTICATION_NOT_SUCCESSFUL,
"Error verifying the client certificate. "
"See server logs for more information."
)
self.assertTrue(kmip_session._send_response.called)
@mock.patch(
'kmip.services.server.auth.get_extended_key_usage_from_certificate'
)
@mock.patch('kmip.services.server.auth.get_certificate_from_connection')
@mock.patch('kmip.core.messages.messages.RequestMessage')
def test_handle_message_loop_invalid_certificate_extension(self,
request_mock,
cert_mock,
ext_mock):
"""
Test that the correct logging and error handling occurs when an
invalid certificate is encountered while processing a request.
"""
data = utils.BytearrayStream(())
cert_mock.return_value = 'test_certificate'
ext_mock.return_value = []
kmip_engine = engine.KmipEngine()
kmip_engine._logger = mock.MagicMock()
kmip_session = session.KmipSession(
kmip_engine,
None,
None,
name='name',
enable_tls_client_auth=True
)
kmip_session.authenticate = mock.MagicMock()
kmip_session._engine = mock.MagicMock()
kmip_session._logger = mock.MagicMock()
kmip_session._connection = mock.MagicMock()
kmip_session._receive_request = mock.MagicMock(return_value=data)
kmip_session._send_response = mock.MagicMock()
kmip_session._handle_message_loop()
kmip_session._receive_request.assert_called_once_with()
kmip_session._logger.warning(
"Failure verifying the client certificate."
)
kmip_session._logger.exception.assert_called_once_with(
exceptions.PermissionDenied(
"The extended key usage extension is not marked for client "
"authentication in the client certificate."
)
)
kmip_session._engine.build_error_response.assert_called_once_with(
contents.ProtocolVersion(1, 0),
enums.ResultReason.AUTHENTICATION_NOT_SUCCESSFUL,
"Error verifying the client certificate. "
"See server logs for more information."
)
self.assertTrue(kmip_session._send_response.called)
@mock.patch(
"kmip.services.server.auth.get_client_identity_from_certificate"
)
def test_authenticate(self, mock_get):
"""
Test that the session correctly uses the authentication plugin
framework to authenticate new connections.
"""
mock_get.return_value = "John Doe"
kmip_session = session.KmipSession(
None,
None,
None,
name='TestSession'
)
kmip_session._logger = mock.MagicMock()
fake_request = messages.RequestMessage(
request_header=messages.RequestHeader()
)
session_identity = kmip_session.authenticate(
"fake_certificate",
fake_request
)
kmip_session._logger.debug.assert_any_call(
"No authentication plugins are enabled. The client identity will "
"be extracted from the client certificate."
)
mock_get.assert_any_call("fake_certificate")
kmip_session._logger.debug.assert_any_call(
"Extraction succeeded for client identity: John Doe"
)
self.assertEqual(("John Doe", None), session_identity)
@mock.patch("kmip.services.server.auth.SLUGSConnector")
def test_authenticate_against_slugs(self, mock_connector):
"""
Test that the session correctly handles authentication with SLUGS.
"""
mock_instance = mock.MagicMock()
mock_instance.authenticate.return_value = ("John Doe", ["Group A"])
mock_connector.return_value = mock_instance
kmip_session = session.KmipSession(
None,
None,
("127.0.0.1", 48026),
name='TestSession',
auth_settings=[(
"auth:slugs",
{"enabled": "True", "url": "test_url"}
)]
)
kmip_session._logger = mock.MagicMock()
fake_credential = objects.Credential(
credential_type=enums.CredentialType.USERNAME_AND_PASSWORD,
credential_value=objects.UsernamePasswordCredential(
username="John Doe",
password="secret"
)
)
fake_request = messages.RequestMessage(
request_header=messages.RequestHeader(
authentication=contents.Authentication(
credentials=[fake_credential]
)
)
)
result = kmip_session.authenticate(
"fake_certificate",
fake_request
)
mock_connector.assert_any_call("test_url")
kmip_session._logger.debug.assert_any_call(
"Authenticating with plugin: auth:slugs"
)
mock_instance.authenticate.assert_any_call(
"fake_certificate",
(("127.0.0.1", 48026), kmip_session._session_time),
fake_request.request_header.authentication.credentials
)
kmip_session._logger.debug(
"Authentication succeeded for client identity: John Doe"
)
self.assertEqual(2, len(result))
self.assertEqual("John Doe", result[0])
self.assertEqual(["Group A"], result[1])
@mock.patch("kmip.services.server.auth.SLUGSConnector")
def test_authenticate_against_slugs_with_failure(self, mock_connector):
"""
Test that the session correctly handles a SLUGS authentication error.
"""
mock_instance = mock.MagicMock()
test_exception = exceptions.PermissionDenied(
"Unrecognized user ID: John Doe"
)
mock_instance.authenticate.side_effect = test_exception
mock_connector.return_value = mock_instance
kmip_session = session.KmipSession(
None,
None,
("127.0.0.1", 48026),
name='TestSession',
auth_settings=[(
"auth:slugs",
{"enabled": "True", "url": "test_url"}
)]
)
kmip_session._logger = mock.MagicMock()
fake_credential = objects.Credential(
credential_type=enums.CredentialType.USERNAME_AND_PASSWORD,
credential_value=objects.UsernamePasswordCredential(
username="John Doe",
password="secret"
)
)
fake_request = messages.RequestMessage(
request_header=messages.RequestHeader(
authentication=contents.Authentication(
credentials=[fake_credential]
)
)
)
args = ("fake_certificate", fake_request)
self.assertRaisesRegexp(
exceptions.PermissionDenied,
"Authentication failed.",
kmip_session.authenticate,
*args
)
mock_connector.assert_any_call("test_url")
kmip_session._logger.debug.assert_any_call(
"Authenticating with plugin: auth:slugs"
)
kmip_session._logger.warning.assert_any_call("Authentication failed.")
kmip_session._logger.exception.assert_any_call(test_exception)
def test_authenticate_against_unrecognized_plugin(self):
"""
Test that the session correctly handles an unrecognized plugin
configuration.
"""
kmip_session = session.KmipSession(
None,
None,
None,
name='TestSession',
auth_settings=[("auth:unrecognized", {})]
)
kmip_session._logger = mock.MagicMock()
fake_request = messages.RequestMessage(
request_header=messages.RequestHeader()
)
args = ("fake_certificate", fake_request)
self.assertRaisesRegexp(
exceptions.PermissionDenied,
"Authentication failed.",
kmip_session.authenticate,
*args
)
kmip_session._logger.warning.assert_any_call(
"Authentication plugin 'auth:unrecognized' is not supported."
)
def test_receive_request(self):
"""
Test that the session can correctly receive and parse a message
@ -494,7 +800,7 @@ class TestKmipSession(testtools.TestCase):
content = b'\x00\x00\x00\x00\x00\x00\x00\x00'
expected = utils.BytearrayStream((content))
kmip_session = session.KmipSession(None, None, 'name')
kmip_session = session.KmipSession(None, None, None, 'name')
kmip_session._receive_bytes = mock.MagicMock(
side_effect=[content, b'']
)
@ -512,7 +818,7 @@ class TestKmipSession(testtools.TestCase):
"""
content = b'\x00\x00\x00\x00\x00\x00\x00\x00'
kmip_session = session.KmipSession(None, None, 'name')
kmip_session = session.KmipSession(None, None, None, 'name')
kmip_session._connection = mock.MagicMock()
kmip_session._connection.recv = mock.MagicMock(
side_effect=[content, content]
@ -542,7 +848,7 @@ class TestKmipSession(testtools.TestCase):
"""
content = b'\x00\x00\x00\x00\x00\x00\x00\x00'
kmip_session = session.KmipSession(None, None, 'name')
kmip_session = session.KmipSession(None, None, None, 'name')
kmip_session._connection = mock.MagicMock()
kmip_session._connection.recv = mock.MagicMock(
side_effect=[content, content, None]
@ -563,7 +869,7 @@ class TestKmipSession(testtools.TestCase):
))
buffer_empty = utils.BytearrayStream()
kmip_session = session.KmipSession(None, None, 'name')
kmip_session = session.KmipSession(None, None, None, 'name')
kmip_session._connection = mock.MagicMock()
kmip_session._send_response(buffer_empty.buffer)