mirror of
https://github.com/OpenKMIP/PyKMIP.git
synced 2025-05-07 00:10:08 +02:00
98f5ba39e3
This change adds an authentication plugin framework to be used by the PyKMIP server. This framework will allow the server to query third-party authentication systems for user identity information, improving the access control model for the server. The initial plugin provided queries an instance of the new SLUGS library.
271 lines
12 KiB
Python
271 lines
12 KiB
Python
# Copyright (c) 2018 The Johns Hopkins University/Applied Physics Laboratory
|
|
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
from cryptography import x509
|
|
from cryptography.hazmat import backends
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
|
|
import datetime
|
|
import mock
|
|
import ssl
|
|
import testtools
|
|
|
|
from kmip.core import exceptions
|
|
from kmip.services.server.auth import utils
|
|
|
|
|
|
class TestUtils(testtools.TestCase):
|
|
"""
|
|
Test suite for authentication utilities.
|
|
"""
|
|
|
|
def setUp(self):
|
|
super(TestUtils, self).setUp()
|
|
|
|
self.certificate_bytes = (
|
|
b'\x30\x82\x03\x7c\x30\x82\x02\x64\xa0\x03\x02\x01\x02\x02\x01\x02'
|
|
b'\x30\x0d\x06\x09\x2a\x86\x48\x86\xf7\x0d\x01\x01\x0b\x05\x00\x30'
|
|
b'\x45\x31\x0b\x30\x09\x06\x03\x55\x04\x06\x13\x02\x55\x53\x31\x1f'
|
|
b'\x30\x1d\x06\x03\x55\x04\x0a\x13\x16\x54\x65\x73\x74\x20\x43\x65'
|
|
b'\x72\x74\x69\x66\x69\x63\x61\x74\x65\x73\x20\x32\x30\x31\x31\x31'
|
|
b'\x15\x30\x13\x06\x03\x55\x04\x03\x13\x0c\x54\x72\x75\x73\x74\x20'
|
|
b'\x41\x6e\x63\x68\x6f\x72\x30\x1e\x17\x0d\x31\x30\x30\x31\x30\x31'
|
|
b'\x30\x38\x33\x30\x30\x30\x5a\x17\x0d\x33\x30\x31\x32\x33\x31\x30'
|
|
b'\x38\x33\x30\x30\x30\x5a\x30\x40\x31\x0b\x30\x09\x06\x03\x55\x04'
|
|
b'\x06\x13\x02\x55\x53\x31\x1f\x30\x1d\x06\x03\x55\x04\x0a\x13\x16'
|
|
b'\x54\x65\x73\x74\x20\x43\x65\x72\x74\x69\x66\x69\x63\x61\x74\x65'
|
|
b'\x73\x20\x32\x30\x31\x31\x31\x10\x30\x0e\x06\x03\x55\x04\x03\x13'
|
|
b'\x07\x47\x6f\x6f\x64\x20\x43\x41\x30\x82\x01\x22\x30\x0d\x06\x09'
|
|
b'\x2a\x86\x48\x86\xf7\x0d\x01\x01\x01\x05\x00\x03\x82\x01\x0f\x00'
|
|
b'\x30\x82\x01\x0a\x02\x82\x01\x01\x00\x90\x58\x9a\x47\x62\x8d\xfb'
|
|
b'\x5d\xf6\xfb\xa0\x94\x8f\x7b\xe5\xaf\x7d\x39\x73\x20\x6d\xb5\x59'
|
|
b'\x0e\xcc\xc8\xc6\xc6\xb4\xaf\xe6\xf2\x67\xa3\x0b\x34\x7a\x73\xe7'
|
|
b'\xff\xa4\x98\x44\x1f\xf3\x9c\x0d\x23\x2c\x5e\xaf\x21\xe6\x45\xda'
|
|
b'\x04\x6a\x96\x2b\xeb\xd2\xc0\x3f\xcf\xce\x9e\x4e\x60\x6a\x6d\x5e'
|
|
b'\x61\x8f\x72\xd8\x43\xb4\x0c\x25\xad\xa7\xe4\x18\xe4\xb8\x1a\xa2'
|
|
b'\x09\xf3\xe9\x3d\x5c\x62\xac\xfa\xf4\x14\x5c\x92\xac\x3a\x4e\x3b'
|
|
b'\x46\xec\xc3\xe8\xf6\x6e\xa6\xae\x2c\xd7\xac\x5a\x2d\x5a\x98\x6d'
|
|
b'\x40\xb6\xe9\x47\x18\xd3\xc1\xa9\x9e\x82\xcd\x1c\x96\x52\xfc\x49'
|
|
b'\x97\xc3\x56\x59\xdd\xde\x18\x66\x33\x65\xa4\x8a\x56\x14\xd1\xe7'
|
|
b'\x50\x69\x9d\x88\x62\x97\x50\xf5\xff\xf4\x7d\x1f\x56\x32\x00\x69'
|
|
b'\x0c\x23\x9c\x60\x1b\xa6\x0c\x82\xba\x65\xa0\xcc\x8c\x0f\xa5\x7f'
|
|
b'\x84\x94\x53\x94\xaf\x7c\xfb\x06\x85\x67\x14\xa8\x48\x5f\x37\xbe'
|
|
b'\x56\x64\x06\x49\x6c\x59\xc6\xf5\x83\x50\xdf\x74\x52\x5d\x2d\x2c'
|
|
b'\x4a\x4b\x82\x4d\xce\x57\x15\x01\xe1\x55\x06\xb9\xfd\x79\x38\x93'
|
|
b'\xa9\x82\x8d\x71\x89\xb2\x0d\x3e\x65\xad\xd7\x85\x5d\x6b\x63\x7d'
|
|
b'\xca\xb3\x4a\x96\x82\x46\x64\xda\x8b\x02\x03\x01\x00\x01\xa3\x7c'
|
|
b'\x30\x7a\x30\x1f\x06\x03\x55\x1d\x23\x04\x18\x30\x16\x80\x14\xe4'
|
|
b'\x7d\x5f\xd1\x5c\x95\x86\x08\x2c\x05\xae\xbe\x75\xb6\x65\xa7\xd9'
|
|
b'\x5d\xa8\x66\x30\x1d\x06\x03\x55\x1d\x0e\x04\x16\x04\x14\x58\x01'
|
|
b'\x84\x24\x1b\xbc\x2b\x52\x94\x4a\x3d\xa5\x10\x72\x14\x51\xf5\xaf'
|
|
b'\x3a\xc9\x30\x0e\x06\x03\x55\x1d\x0f\x01\x01\xff\x04\x04\x03\x02'
|
|
b'\x01\x06\x30\x17\x06\x03\x55\x1d\x20\x04\x10\x30\x0e\x30\x0c\x06'
|
|
b'\x0a\x60\x86\x48\x01\x65\x03\x02\x01\x30\x01\x30\x0f\x06\x03\x55'
|
|
b'\x1d\x13\x01\x01\xff\x04\x05\x30\x03\x01\x01\xff\x30\x0d\x06\x09'
|
|
b'\x2a\x86\x48\x86\xf7\x0d\x01\x01\x0b\x05\x00\x03\x82\x01\x01\x00'
|
|
b'\x35\x87\x97\x16\xe6\x75\x35\xcd\xc0\x12\xff\x96\x5c\x21\x42\xac'
|
|
b'\x27\x6b\x32\xbb\x08\x2d\x96\xb1\x70\x41\xaa\x03\x4f\x5a\x3e\xe6'
|
|
b'\xb6\xf4\x3e\x68\xb1\xbc\xff\x9d\x10\x73\x64\xae\x9f\xba\x36\x56'
|
|
b'\x7c\x05\xf4\x3d\x7c\x51\x47\xbc\x1a\x3d\xee\x3d\x46\x07\xfa\x84'
|
|
b'\x88\xd6\xf0\xdd\xc8\xa7\x23\x98\xc6\xca\x45\x4e\x2b\x93\x47\xa8'
|
|
b'\xdd\x41\xcd\x0d\x7c\x2a\x21\x57\x3d\x09\x04\xbd\xb2\x6c\x95\xfb'
|
|
b'\x1d\x47\x0b\x02\xf8\x4d\x3a\xea\xf8\xb5\xcb\x2b\x1f\xea\x56\x28'
|
|
b'\xf4\x62\xa9\x3e\x50\x97\xc0\xb6\xb8\x36\x8e\x76\x0a\x5e\xc0\xae'
|
|
b'\x14\xc0\x50\x42\x75\x82\x1a\xbc\x1a\xd6\x0d\x53\xa6\x14\x69\xfd'
|
|
b'\x19\x98\x1e\x73\x32\x9d\x81\x66\x66\xb5\xed\xcc\x5c\xfe\x53\xd5'
|
|
b'\xc4\x03\xb0\xbe\x80\xfa\xb8\x92\xa0\xc8\xfe\x25\x5f\x21\x3d\x6c'
|
|
b'\xea\x50\x6d\x74\x1e\x74\x96\xb0\xd5\xc2\x5d\xa8\x61\xf0\x2f\x5b'
|
|
b'\xfe\xac\x0b\x6b\x1e\xd9\x09\x5e\x66\x27\x54\x9a\xbc\xe2\x54\xd3'
|
|
b'\xf8\xa0\x47\x97\x20\xda\x24\x53\xa4\xfa\xa7\xff\xc7\x33\x51\x46'
|
|
b'\x41\x8c\x36\x8c\xeb\xe9\x29\xc2\xad\x58\x24\x80\x9d\xe8\x04\x6e'
|
|
b'\x0b\x06\x63\x30\x13\x2a\x39\x8f\x24\xf2\x74\x9e\x91\xc5\xab\x33'
|
|
)
|
|
|
|
private_key = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=2048,
|
|
backend=backends.default_backend()
|
|
)
|
|
subject = issuer = x509.Name([
|
|
x509.NameAttribute(x509.NameOID.COMMON_NAME, u"Jane Doe")
|
|
])
|
|
subject_no_common_name = issuer_no_common_name = x509.Name([
|
|
x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, u"Test, Inc.")
|
|
])
|
|
self.certificate = x509.CertificateBuilder().subject_name(
|
|
subject
|
|
).issuer_name(
|
|
issuer
|
|
).public_key(
|
|
private_key.public_key()
|
|
).serial_number(
|
|
x509.random_serial_number()
|
|
).not_valid_before(
|
|
datetime.datetime.utcnow()
|
|
).not_valid_after(
|
|
datetime.datetime.utcnow() + datetime.timedelta(days=1)
|
|
).add_extension(
|
|
x509.ExtendedKeyUsage([x509.ExtendedKeyUsageOID.CLIENT_AUTH]),
|
|
critical=True
|
|
).sign(private_key, hashes.SHA256(), backends.default_backend())
|
|
|
|
self.certificate_no_name = x509.CertificateBuilder().subject_name(
|
|
subject_no_common_name
|
|
).issuer_name(
|
|
issuer_no_common_name
|
|
).public_key(
|
|
private_key.public_key()
|
|
).serial_number(
|
|
x509.random_serial_number()
|
|
).not_valid_before(
|
|
datetime.datetime.utcnow()
|
|
).not_valid_after(
|
|
datetime.datetime.utcnow() + datetime.timedelta(days=1)
|
|
).sign(private_key, hashes.SHA256(), backends.default_backend())
|
|
|
|
self.certificate_no_extension = x509.CertificateBuilder().subject_name(
|
|
subject
|
|
).issuer_name(
|
|
issuer
|
|
).public_key(
|
|
private_key.public_key()
|
|
).serial_number(
|
|
x509.random_serial_number()
|
|
).not_valid_before(
|
|
datetime.datetime.utcnow()
|
|
).not_valid_after(
|
|
datetime.datetime.utcnow() + datetime.timedelta(days=1)
|
|
).sign(private_key, hashes.SHA256(), backends.default_backend())
|
|
|
|
def tearDown(self):
|
|
super(TestUtils, self).tearDown()
|
|
|
|
def test_get_certificate_from_connection(self):
|
|
"""
|
|
Test that the certificate can be retrieved from a provided connection.
|
|
"""
|
|
mock_connection = mock.MagicMock(ssl.SSLSocket)
|
|
mock_connection.getpeercert.return_value = self.certificate_bytes
|
|
result = utils.get_certificate_from_connection(
|
|
mock_connection
|
|
)
|
|
|
|
self.assertIsInstance(result, x509.Certificate)
|
|
|
|
def test_get_certificate_from_connection_with_load_failure(self):
|
|
"""
|
|
Test that the right value is returned when the certificate cannot be
|
|
retrieved from the provided connection.
|
|
"""
|
|
mock_connection = mock.MagicMock(ssl.SSLSocket)
|
|
mock_connection.getpeercert.return_value = None
|
|
result = utils.get_certificate_from_connection(
|
|
mock_connection
|
|
)
|
|
|
|
self.assertEqual(None, result)
|
|
|
|
def test_get_extended_key_usage_from_certificate(self):
|
|
"""
|
|
Test that the ExtendedKeyUsage extension can be retrieved from a
|
|
certificate.
|
|
"""
|
|
extension = utils.get_extended_key_usage_from_certificate(
|
|
self.certificate
|
|
)
|
|
|
|
self.assertIsInstance(extension, x509.ExtendedKeyUsage)
|
|
self.assertIn(x509.ExtendedKeyUsageOID.CLIENT_AUTH, extension)
|
|
|
|
def test_get_extended_key_usage_from_certificate_with_no_extension(self):
|
|
"""
|
|
Test that the right value is returned when the ExtendedKeyUsage
|
|
extension cannot be retrieved from a certificate.
|
|
"""
|
|
extension = utils.get_extended_key_usage_from_certificate(
|
|
self.certificate_no_extension
|
|
)
|
|
|
|
self.assertEqual(None, extension)
|
|
|
|
def test_get_common_names_from_certificate(self):
|
|
"""
|
|
Test that the common names can be retrieved from a certificate.
|
|
"""
|
|
common_names = utils.get_common_names_from_certificate(
|
|
self.certificate
|
|
)
|
|
|
|
self.assertEqual(["Jane Doe"], common_names)
|
|
|
|
def test_get_common_names_from_certificate_no_common_names(self):
|
|
"""
|
|
Test that the right value is returned when no common names can be
|
|
retrieved from a certificate.
|
|
"""
|
|
common_names = utils.get_common_names_from_certificate(
|
|
self.certificate_no_name
|
|
)
|
|
|
|
self.assertEqual([], common_names)
|
|
|
|
def test_get_client_identity_from_certificate(self):
|
|
"""
|
|
Test that the common names from a certificate can be processed into a
|
|
client identity.
|
|
"""
|
|
result = utils.get_client_identity_from_certificate(self.certificate)
|
|
|
|
self.assertEqual("Jane Doe", result)
|
|
|
|
@mock.patch(
|
|
'kmip.services.server.auth.utils.get_common_names_from_certificate'
|
|
)
|
|
def test_get_client_identity_from_certificate_multiple_names(self,
|
|
mock_get):
|
|
"""
|
|
Test that the a PermissionDenied error is raised if multiple possible
|
|
client identities are discovered.
|
|
"""
|
|
mock_get.return_value = ["John Doe", "Jane Doe"]
|
|
|
|
args = ("test", )
|
|
self.assertRaisesRegexp(
|
|
exceptions.PermissionDenied,
|
|
"Multiple client identities found.",
|
|
utils.get_client_identity_from_certificate,
|
|
*args
|
|
)
|
|
|
|
@mock.patch(
|
|
'kmip.services.server.auth.utils.get_common_names_from_certificate'
|
|
)
|
|
def test_get_client_identity_from_certificate_no_names(self, mock_get):
|
|
"""
|
|
Test that the a PermissionDenied error is raised if no possible client
|
|
identities are discovered.
|
|
"""
|
|
mock_get.return_value = []
|
|
|
|
args = ("test", )
|
|
self.assertRaisesRegexp(
|
|
exceptions.PermissionDenied,
|
|
"The certificate does not define any subject common names. Client "
|
|
"identity unavailable.",
|
|
utils.get_client_identity_from_certificate,
|
|
*args
|
|
)
|