openstack/keystone
Code Issues Proposed changes
29af9bf5e5
keystone/keystone/tests/unit/token/test_fernet_provider.py
Dolph Mathews 41fdcf78b0 Test for fernet rotation recovery after disk full 
This expanded test illustrates that the key rotation mechanism can
recover from repeated rotation attempts while the disk was full.

Change-Id: I042a15929c5e61aeaf357d88f224cdc026b5a5ed
2017-03-07 15:17:45 +00:00

689 lines
27 KiB
Python
Raw Blame History

# 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
import datetime
import hashlib
import mock
import os
import uuid
from oslo_utils import timeutils
import six
from keystone import auth
from keystone.common import fernet_utils
from keystone.common import utils
import keystone.conf
from keystone import exception
from keystone.federation import constants as federation_constants
from keystone.tests import unit
from keystone.tests.unit import ksfixtures
from keystone.tests.unit.ksfixtures import database
from keystone.token.providers import common
from keystone.token.providers import fernet
from keystone.token.providers.fernet import token_formatters
CONF = keystone.conf.CONF
class TestFernetTokenProvider(unit.TestCase):
def setUp(self):
super(TestFernetTokenProvider, self).setUp()
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'fernet_tokens',
CONF.fernet_tokens.max_active_keys
)
)
self.provider = fernet.Provider()
def test_supports_bind_authentication_returns_false(self):
self.assertFalse(self.provider._supports_bind_authentication)
def test_needs_persistence_returns_false(self):
self.assertFalse(self.provider.needs_persistence())
def test_invalid_v3_token_raises_token_not_found(self):
token_id = uuid.uuid4().hex
e = self.assertRaises(
exception.TokenNotFound,
self.provider.validate_token,
token_id)
self.assertIn(token_id, u'%s' % e)
def test_invalid_v2_token_raises_token_not_found(self):
token_id = uuid.uuid4().hex
e = self.assertRaises(
exception.TokenNotFound,
self.provider.validate_token,
token_id)
self.assertIn(token_id, u'%s' % e)
class TestValidate(unit.TestCase):
def setUp(self):
super(TestValidate, self).setUp()
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'fernet_tokens',
CONF.fernet_tokens.max_active_keys
)
)
self.useFixture(database.Database())
self.load_backends()
def config_overrides(self):
super(TestValidate, self).config_overrides()
self.config_fixture.config(group='token', provider='fernet')
def test_validate_v3_token_simple(self):
# Check the fields in the token result when use validate_v3_token
# with a simple token.
domain_ref = unit.new_domain_ref()
domain_ref = self.resource_api.create_domain(domain_ref['id'],
domain_ref)
user_ref = unit.new_user_ref(domain_ref['id'])
user_ref = self.identity_api.create_user(user_ref)
method_names = ['password']
token_id, token_data_ = self.token_provider_api.issue_token(
user_ref['id'], method_names)
token_data = self.token_provider_api.validate_token(token_id)
token = token_data['token']
self.assertIsInstance(token['audit_ids'], list)
self.assertIsInstance(token['expires_at'], str)
self.assertIsInstance(token['issued_at'], str)
self.assertEqual(method_names, token['methods'])
exp_user_info = {
'id': user_ref['id'],
'name': user_ref['name'],
'domain': {
'id': domain_ref['id'],
'name': domain_ref['name'],
},
'password_expires_at': user_ref['password_expires_at']
}
self.assertEqual(exp_user_info, token['user'])
def test_validate_v3_token_federated_info(self):
# Check the user fields in the token result when use validate_v3_token
# when the token has federated info.
domain_ref = unit.new_domain_ref()
domain_ref = self.resource_api.create_domain(domain_ref['id'],
domain_ref)
user_ref = unit.new_user_ref(domain_ref['id'])
user_ref = self.identity_api.create_user(user_ref)
method_names = ['mapped']
group_ids = [uuid.uuid4().hex, ]
identity_provider = uuid.uuid4().hex
protocol = uuid.uuid4().hex
auth_context_params = {
'user_id': user_ref['id'],
'user_name': user_ref['name'],
'group_ids': group_ids,
federation_constants.IDENTITY_PROVIDER: identity_provider,
federation_constants.PROTOCOL: protocol,
}
auth_context = auth.core.AuthContext(**auth_context_params)
token_id, token_data_ = self.token_provider_api.issue_token(
user_ref['id'], method_names, auth_context=auth_context)
token_data = self.token_provider_api.validate_token(token_id)
token = token_data['token']
exp_user_info = {
'id': user_ref['id'],
'name': user_ref['name'],
'domain': {'id': CONF.federation.federated_domain_name,
'name': CONF.federation.federated_domain_name, },
federation_constants.FEDERATION: {
'groups': [{'id': group_id} for group_id in group_ids],
'identity_provider': {'id': identity_provider, },
'protocol': {'id': protocol, },
},
}
self.assertDictEqual(exp_user_info, token['user'])
def test_validate_v3_token_trust(self):
# Check the trust fields in the token result when use validate_v3_token
# when the token has trust info.
domain_ref = unit.new_domain_ref()
domain_ref = self.resource_api.create_domain(domain_ref['id'],
domain_ref)
user_ref = unit.new_user_ref(domain_ref['id'])
user_ref = self.identity_api.create_user(user_ref)
trustor_user_ref = unit.new_user_ref(domain_ref['id'])
trustor_user_ref = self.identity_api.create_user(trustor_user_ref)
project_ref = unit.new_project_ref(domain_id=domain_ref['id'])
project_ref = self.resource_api.create_project(project_ref['id'],
project_ref)
role_ref = unit.new_role_ref()
role_ref = self.role_api.create_role(role_ref['id'], role_ref)
self.assignment_api.create_grant(
role_ref['id'], user_id=user_ref['id'],
project_id=project_ref['id'])
self.assignment_api.create_grant(
role_ref['id'], user_id=trustor_user_ref['id'],
project_id=project_ref['id'])
trustor_user_id = trustor_user_ref['id']
trustee_user_id = user_ref['id']
trust_ref = unit.new_trust_ref(
trustor_user_id, trustee_user_id, project_id=project_ref['id'],
role_ids=[role_ref['id'], ])
trust_ref = self.trust_api.create_trust(trust_ref['id'], trust_ref,
trust_ref['roles'])
method_names = ['password']
token_id, token_data_ = self.token_provider_api.issue_token(
user_ref['id'], method_names, project_id=project_ref['id'],
trust=trust_ref)
token_data = self.token_provider_api.validate_token(token_id)
token = token_data['token']
exp_trust_info = {
'id': trust_ref['id'],
'impersonation': False,
'trustee_user': {'id': user_ref['id'], },
'trustor_user': {'id': trustor_user_ref['id'], },
}
self.assertEqual(exp_trust_info, token['OS-TRUST:trust'])
def test_validate_v3_token_validation_error_exc(self):
# When the token format isn't recognized, TokenNotFound is raised.
# A uuid string isn't a valid Fernet token.
token_id = uuid.uuid4().hex
self.assertRaises(exception.TokenNotFound,
self.token_provider_api.validate_token, token_id)
class TestTokenFormatter(unit.TestCase):
def setUp(self):
super(TestTokenFormatter, self).setUp()
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'fernet_tokens',
CONF.fernet_tokens.max_active_keys
)
)
def test_restore_padding(self):
# 'a' will result in '==' padding, 'aa' will result in '=' padding, and
# 'aaa' will result in no padding.
binary_to_test = [b'a', b'aa', b'aaa']
for binary in binary_to_test:
# base64.urlsafe_b64encode takes six.binary_type and returns
# six.binary_type.
encoded_string = base64.urlsafe_b64encode(binary)
encoded_string = encoded_string.decode('utf-8')
# encoded_string is now six.text_type.
encoded_str_without_padding = encoded_string.rstrip('=')
self.assertFalse(encoded_str_without_padding.endswith('='))
encoded_str_with_padding_restored = (
token_formatters.TokenFormatter.restore_padding(
encoded_str_without_padding)
)
self.assertEqual(encoded_string, encoded_str_with_padding_restored)
class TestPayloads(unit.TestCase):
def assertTimestampsEqual(self, expected, actual):
# The timestamp that we get back when parsing the payload may not
# exactly match the timestamp that was put in the payload due to
# conversion to and from a float.
exp_time = timeutils.parse_isotime(expected)
actual_time = timeutils.parse_isotime(actual)
# the granularity of timestamp string is microseconds and it's only the
# last digit in the representation that's different, so use a delta
# just above nanoseconds.
return self.assertCloseEnoughForGovernmentWork(exp_time, actual_time,
delta=1e-05)
def test_strings_can_be_converted_to_bytes(self):
s = common.random_urlsafe_str()
self.assertIsInstance(s, six.text_type)
b = token_formatters.BasePayload.random_urlsafe_str_to_bytes(s)
self.assertIsInstance(b, six.binary_type)
def test_uuid_hex_to_byte_conversions(self):
payload_cls = token_formatters.BasePayload
expected_hex_uuid = uuid.uuid4().hex
uuid_obj = uuid.UUID(expected_hex_uuid)
expected_uuid_in_bytes = uuid_obj.bytes
actual_uuid_in_bytes = payload_cls.convert_uuid_hex_to_bytes(
expected_hex_uuid)
self.assertEqual(expected_uuid_in_bytes, actual_uuid_in_bytes)
actual_hex_uuid = payload_cls.convert_uuid_bytes_to_hex(
expected_uuid_in_bytes)
self.assertEqual(expected_hex_uuid, actual_hex_uuid)
def test_time_string_to_float_conversions(self):
payload_cls = token_formatters.BasePayload
original_time_str = utils.isotime(subsecond=True)
time_obj = timeutils.parse_isotime(original_time_str)
expected_time_float = (
(timeutils.normalize_time(time_obj) -
datetime.datetime.utcfromtimestamp(0)).total_seconds())
# NOTE(lbragstad): The token expiration time for Fernet tokens is
# passed in the payload of the token. This is different from the token
# creation time, which is handled by Fernet and doesn't support
# subsecond precision because it is a timestamp integer.
self.assertIsInstance(expected_time_float, float)
actual_time_float = payload_cls._convert_time_string_to_float(
original_time_str)
self.assertIsInstance(actual_time_float, float)
self.assertEqual(expected_time_float, actual_time_float)
# Generate expected_time_str using the same time float. Using
# original_time_str from utils.isotime will occasionally fail due to
# floating point rounding differences.
time_object = datetime.datetime.utcfromtimestamp(actual_time_float)
expected_time_str = utils.isotime(time_object, subsecond=True)
actual_time_str = payload_cls._convert_float_to_time_string(
actual_time_float)
self.assertEqual(expected_time_str, actual_time_str)
def _test_payload(self, payload_class, exp_user_id=None, exp_methods=None,
exp_project_id=None, exp_domain_id=None,
exp_trust_id=None, exp_federated_info=None,
exp_access_token_id=None):
exp_user_id = exp_user_id or uuid.uuid4().hex
exp_methods = exp_methods or ['password']
exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
exp_audit_ids = [common.random_urlsafe_str()]
payload = payload_class.assemble(
exp_user_id, exp_methods, exp_project_id, exp_domain_id,
exp_expires_at, exp_audit_ids, exp_trust_id, exp_federated_info,
exp_access_token_id)
(user_id, methods, project_id,
domain_id, expires_at, audit_ids,
trust_id, federated_info,
access_token_id) = payload_class.disassemble(payload)
self.assertEqual(exp_user_id, user_id)
self.assertEqual(exp_methods, methods)
self.assertTimestampsEqual(exp_expires_at, expires_at)
self.assertEqual(exp_audit_ids, audit_ids)
self.assertEqual(exp_project_id, project_id)
self.assertEqual(exp_domain_id, domain_id)
self.assertEqual(exp_trust_id, trust_id)
self.assertEqual(exp_access_token_id, access_token_id)
if exp_federated_info:
self.assertDictEqual(exp_federated_info, federated_info)
else:
self.assertIsNone(federated_info)
def test_unscoped_payload(self):
self._test_payload(token_formatters.UnscopedPayload)
def test_project_scoped_payload(self):
self._test_payload(token_formatters.ProjectScopedPayload,
exp_project_id=uuid.uuid4().hex)
def test_domain_scoped_payload(self):
self._test_payload(token_formatters.DomainScopedPayload,
exp_domain_id=uuid.uuid4().hex)
def test_domain_scoped_payload_with_default_domain(self):
self._test_payload(token_formatters.DomainScopedPayload,
exp_domain_id=CONF.identity.default_domain_id)
def test_trust_scoped_payload(self):
self._test_payload(token_formatters.TrustScopedPayload,
exp_project_id=uuid.uuid4().hex,
exp_trust_id=uuid.uuid4().hex)
def test_unscoped_payload_with_non_uuid_user_id(self):
self._test_payload(token_formatters.UnscopedPayload,
exp_user_id='someNonUuidUserId')
def test_unscoped_payload_with_16_char_non_uuid_user_id(self):
self._test_payload(token_formatters.UnscopedPayload,
exp_user_id='0123456789abcdef')
def test_project_scoped_payload_with_non_uuid_ids(self):
self._test_payload(token_formatters.ProjectScopedPayload,
exp_user_id='someNonUuidUserId',
exp_project_id='someNonUuidProjectId')
def test_project_scoped_payload_with_16_char_non_uuid_ids(self):
self._test_payload(token_formatters.ProjectScopedPayload,
exp_user_id='0123456789abcdef',
exp_project_id='0123456789abcdef')
def test_domain_scoped_payload_with_non_uuid_user_id(self):
self._test_payload(token_formatters.DomainScopedPayload,
exp_user_id='nonUuidUserId',
exp_domain_id=uuid.uuid4().hex)
def test_domain_scoped_payload_with_16_char_non_uuid_user_id(self):
self._test_payload(token_formatters.DomainScopedPayload,
exp_user_id='0123456789abcdef',
exp_domain_id=uuid.uuid4().hex)
def test_trust_scoped_payload_with_non_uuid_ids(self):
self._test_payload(token_formatters.TrustScopedPayload,
exp_user_id='someNonUuidUserId',
exp_project_id='someNonUuidProjectId',
exp_trust_id=uuid.uuid4().hex)
def test_trust_scoped_payload_with_16_char_non_uuid_ids(self):
self._test_payload(token_formatters.TrustScopedPayload,
exp_user_id='0123456789abcdef',
exp_project_id='0123456789abcdef',
exp_trust_id=uuid.uuid4().hex)
def _test_federated_payload_with_ids(self, exp_user_id, exp_group_id):
exp_federated_info = {'group_ids': [{'id': exp_group_id}],
'idp_id': uuid.uuid4().hex,
'protocol_id': uuid.uuid4().hex}
self._test_payload(token_formatters.FederatedUnscopedPayload,
exp_user_id=exp_user_id,
exp_federated_info=exp_federated_info)
def test_federated_payload_with_non_uuid_ids(self):
self._test_federated_payload_with_ids('someNonUuidUserId',
'someNonUuidGroupId')
def test_federated_payload_with_16_char_non_uuid_ids(self):
self._test_federated_payload_with_ids('0123456789abcdef',
'0123456789abcdef')
def test_federated_project_scoped_payload(self):
exp_federated_info = {'group_ids': [{'id': 'someNonUuidGroupId'}],
'idp_id': uuid.uuid4().hex,
'protocol_id': uuid.uuid4().hex}
self._test_payload(token_formatters.FederatedProjectScopedPayload,
exp_user_id='someNonUuidUserId',
exp_methods=['token'],
exp_project_id=uuid.uuid4().hex,
exp_federated_info=exp_federated_info)
def test_federated_domain_scoped_payload(self):
exp_federated_info = {'group_ids': [{'id': 'someNonUuidGroupId'}],
'idp_id': uuid.uuid4().hex,
'protocol_id': uuid.uuid4().hex}
self._test_payload(token_formatters.FederatedDomainScopedPayload,
exp_user_id='someNonUuidUserId',
exp_methods=['token'],
exp_domain_id=uuid.uuid4().hex,
exp_federated_info=exp_federated_info)
def test_oauth_scoped_payload(self):
self._test_payload(token_formatters.OauthScopedPayload,
exp_project_id=uuid.uuid4().hex,
exp_access_token_id=uuid.uuid4().hex)
class TestFernetKeyRotation(unit.TestCase):
def setUp(self):
super(TestFernetKeyRotation, self).setUp()
# A collection of all previously-seen signatures of the key
# repository's contents.
self.key_repo_signatures = set()
@property
def keys(self):
"""Key files converted to numbers."""
return sorted(
int(x) for x in os.listdir(CONF.fernet_tokens.key_repository))
@property
def key_repository_size(self):
"""The number of keys in the key repository."""
return len(self.keys)
@property
def key_repository_signature(self):
"""Create a "thumbprint" of the current key repository.
Because key files are renamed, this produces a hash of the contents of
the key files, ignoring their filenames.
The resulting signature can be used, for example, to ensure that you
have a unique set of keys after you perform a key rotation (taking a
static set of keys, and simply shuffling them, would fail such a test).
"""
# Load the keys into a list, keys is list of six.text_type.
key_utils = fernet_utils.FernetUtils(
CONF.fernet_tokens.key_repository,
CONF.fernet_tokens.max_active_keys,
'fernet_tokens'
)
keys = key_utils.load_keys()
# Sort the list of keys by the keys themselves (they were previously
# sorted by filename).
keys.sort()
# Create the thumbprint using all keys in the repository.
signature = hashlib.sha1()
for key in keys:
# Need to convert key to six.binary_type for update.
signature.update(key.encode('utf-8'))
return signature.hexdigest()
def assertRepositoryState(self, expected_size):
"""Validate the state of the key repository."""
self.assertEqual(expected_size, self.key_repository_size)
self.assertUniqueRepositoryState()
def assertUniqueRepositoryState(self):
"""Ensure that the current key repo state has not been seen before."""
# This is assigned to a variable because it takes some work to
# calculate.
signature = self.key_repository_signature
# Ensure the signature is not in the set of previously seen signatures.
self.assertNotIn(signature, self.key_repo_signatures)
# Add the signature to the set of repository signatures to validate
# that we don't see it again later.
self.key_repo_signatures.add(signature)
def test_rotation(self):
# Initializing a key repository results in this many keys. We don't
# support max_active_keys being set any lower.
min_active_keys = 2
# Simulate every rotation strategy up to "rotating once a week while
# maintaining a year's worth of keys."
for max_active_keys in range(min_active_keys, 52 + 1):
self.config_fixture.config(group='fernet_tokens',
max_active_keys=max_active_keys)
# Ensure that resetting the key repository always results in 2
# active keys.
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'fernet_tokens',
CONF.fernet_tokens.max_active_keys
)
)
# Validate the initial repository state.
self.assertRepositoryState(expected_size=min_active_keys)
# The repository should be initialized with a staged key (0) and a
# primary key (1). The next key is just auto-incremented.
exp_keys = [0, 1]
next_key_number = exp_keys[-1] + 1 # keep track of next key
self.assertEqual(exp_keys, self.keys)
# Rotate the keys just enough times to fully populate the key
# repository.
key_utils = fernet_utils.FernetUtils(
CONF.fernet_tokens.key_repository,
CONF.fernet_tokens.max_active_keys,
'fernet_tokens'
)
for rotation in range(max_active_keys - min_active_keys):
key_utils.rotate_keys()
self.assertRepositoryState(expected_size=rotation + 3)
exp_keys.append(next_key_number)
next_key_number += 1
self.assertEqual(exp_keys, self.keys)
# We should have a fully populated key repository now.
self.assertEqual(max_active_keys, self.key_repository_size)
# Rotate an additional number of times to ensure that we maintain
# the desired number of active keys.
key_utils = fernet_utils.FernetUtils(
CONF.fernet_tokens.key_repository,
CONF.fernet_tokens.max_active_keys,
'fernet_tokens'
)
for rotation in range(10):
key_utils.rotate_keys()
self.assertRepositoryState(expected_size=max_active_keys)
exp_keys.pop(1)
exp_keys.append(next_key_number)
next_key_number += 1
self.assertEqual(exp_keys, self.keys)
def test_rotation_disk_write_fail(self):
# Init the key repository
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'fernet_tokens',
CONF.fernet_tokens.max_active_keys
)
)
# Make sure that the init key repository contains 2 keys
self.assertRepositoryState(expected_size=2)
key_utils = fernet_utils.FernetUtils(
CONF.fernet_tokens.key_repository,
CONF.fernet_tokens.max_active_keys,
'fernet_tokens'
)
# Simulate the disk full situation
mock_open = mock.mock_open()
file_handle = mock_open()
file_handle.flush.side_effect = IOError('disk full')
with mock.patch('keystone.common.fernet_utils.open', mock_open):
self.assertRaises(IOError, key_utils.rotate_keys)
# Assert that the key repository is unchanged
self.assertEqual(self.key_repository_size, 2)
with mock.patch('keystone.common.fernet_utils.open', mock_open):
self.assertRaises(IOError, key_utils.rotate_keys)
# Assert that the key repository is still unchanged, even after
# repeated rotation attempts.
self.assertEqual(self.key_repository_size, 2)
# Rotate the keys normally, without any mocking, to show that the
# system can recover.
key_utils.rotate_keys()
# Assert that the key repository is now expanded.
self.assertEqual(self.key_repository_size, 3)
def test_non_numeric_files(self):
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'fernet_tokens',
CONF.fernet_tokens.max_active_keys
)
)
evil_file = os.path.join(CONF.fernet_tokens.key_repository, '99.bak')
with open(evil_file, 'w'):
pass
key_utils = fernet_utils.FernetUtils(
CONF.fernet_tokens.key_repository,
CONF.fernet_tokens.max_active_keys,
'fernet_tokens'
)
key_utils.rotate_keys()
self.assertTrue(os.path.isfile(evil_file))
keys = 0
for x in os.listdir(CONF.fernet_tokens.key_repository):
if x == '99.bak':
continue
keys += 1
self.assertEqual(3, keys)
class TestLoadKeys(unit.TestCase):
def assertValidFernetKey(self, keys):
self.assertGreater(len(keys[0]), 0)
self.assertIsInstance(keys[0], str)
def test_non_numeric_files(self):
self.useFixture(
ksfixtures.KeyRepository(
self.config_fixture,
'fernet_tokens',
CONF.fernet_tokens.max_active_keys
)
)
evil_file = os.path.join(CONF.fernet_tokens.key_repository, '~1')
with open(evil_file, 'w'):
pass
key_utils = fernet_utils.FernetUtils(
CONF.fernet_tokens.key_repository,
CONF.fernet_tokens.max_active_keys,
'fernet_tokens'
)
keys = key_utils.load_keys()
self.assertEqual(2, len(keys))
self.assertValidFernetKey(keys)
View Git Blame Copy Permalink