summaryrefslogtreecommitdiffstats
path: root/keystone-moon/keystone/tests/unit/token
diff options
context:
space:
mode:
Diffstat (limited to 'keystone-moon/keystone/tests/unit/token')
-rw-r--r--keystone-moon/keystone/tests/unit/token/test_fernet_provider.py276
1 files changed, 225 insertions, 51 deletions
diff --git a/keystone-moon/keystone/tests/unit/token/test_fernet_provider.py b/keystone-moon/keystone/tests/unit/token/test_fernet_provider.py
index 5f74b430..bfb590db 100644
--- a/keystone-moon/keystone/tests/unit/token/test_fernet_provider.py
+++ b/keystone-moon/keystone/tests/unit/token/test_fernet_provider.py
@@ -16,13 +16,17 @@ import hashlib
import os
import uuid
+import msgpack
from oslo_utils import timeutils
+from six.moves import urllib
from keystone.common import config
from keystone.common import utils
+from keystone.contrib.federation import constants as federation_constants
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import ksfixtures
+from keystone.tests.unit.ksfixtures import database
from keystone.token import provider
from keystone.token.providers import fernet
from keystone.token.providers.fernet import token_formatters
@@ -57,7 +61,156 @@ class TestFernetTokenProvider(unit.TestCase):
uuid.uuid4().hex)
+class TestValidate(unit.TestCase):
+ def setUp(self):
+ super(TestValidate, self).setUp()
+ self.useFixture(ksfixtures.KeyRepository(self.config_fixture))
+ 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_v3_token(
+ user_ref['id'], method_names)
+
+ token_data = self.token_provider_api.validate_v3_token(token_id)
+ token = token_data['token']
+ self.assertIsInstance(token['audit_ids'], list)
+ self.assertIsInstance(token['expires_at'], str)
+ self.assertEqual({}, token['extras'])
+ 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'],
+ },
+ }
+ 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 = {
+ 'user_id': user_ref['id'],
+ 'group_ids': group_ids,
+ federation_constants.IDENTITY_PROVIDER: identity_provider,
+ federation_constants.PROTOCOL: protocol,
+ }
+ token_id, token_data_ = self.token_provider_api.issue_v3_token(
+ user_ref['id'], method_names, auth_context=auth_context)
+
+ token_data = self.token_provider_api.validate_v3_token(token_id)
+ token = token_data['token']
+ exp_user_info = {
+ 'id': user_ref['id'],
+ 'name': user_ref['id'],
+ '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.assertEqual(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_v3_token(
+ user_ref['id'], method_names, project_id=project_ref['id'],
+ trust=trust_ref)
+
+ token_data = self.token_provider_api.validate_v3_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_v3_token, token_id)
+
+
class TestTokenFormatter(unit.TestCase):
+ def setUp(self):
+ super(TestTokenFormatter, self).setUp()
+ self.useFixture(ksfixtures.KeyRepository(self.config_fixture))
+
def test_restore_padding(self):
# 'a' will result in '==' padding, 'aa' will result in '=' padding, and
# 'aaa' will result in no padding.
@@ -73,6 +226,39 @@ class TestTokenFormatter(unit.TestCase):
)
self.assertEqual(encoded_string, encoded_str_with_padding_restored)
+ def test_legacy_padding_validation(self):
+ first_value = uuid.uuid4().hex
+ second_value = uuid.uuid4().hex
+ payload = (first_value, second_value)
+ msgpack_payload = msgpack.packb(payload)
+
+ # NOTE(lbragstad): This method perserves the way that keystone used to
+ # percent encode the tokens, prior to bug #1491926.
+ def legacy_pack(payload):
+ tf = token_formatters.TokenFormatter()
+ encrypted_payload = tf.crypto.encrypt(payload)
+
+ # the encrypted_payload is returned with padding appended
+ self.assertTrue(encrypted_payload.endswith('='))
+
+ # using urllib.parse.quote will percent encode the padding, like
+ # keystone did in Kilo.
+ percent_encoded_payload = urllib.parse.quote(encrypted_payload)
+
+ # ensure that the padding was actaully percent encoded
+ self.assertTrue(percent_encoded_payload.endswith('%3D'))
+ return percent_encoded_payload
+
+ token_with_legacy_padding = legacy_pack(msgpack_payload)
+ tf = token_formatters.TokenFormatter()
+
+ # demonstrate the we can validate a payload that has been percent
+ # encoded with the Fernet logic that existed in Kilo
+ serialized_payload = tf.unpack(token_with_legacy_padding)
+ returned_payload = msgpack.unpackb(serialized_payload)
+ self.assertEqual(first_value, returned_payload[0])
+ self.assertEqual(second_value, returned_payload[1])
+
class TestPayloads(unit.TestCase):
def test_uuid_hex_to_byte_conversions(self):
@@ -204,8 +390,7 @@ class TestPayloads(unit.TestCase):
self.assertEqual(exp_audit_ids, audit_ids)
self.assertEqual(exp_trust_id, trust_id)
- def test_unscoped_payload_with_non_uuid_user_id(self):
- exp_user_id = 'someNonUuidUserId'
+ def _test_unscoped_payload_with_user_id(self, exp_user_id):
exp_methods = ['password']
exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
exp_audit_ids = [provider.random_urlsafe_str()]
@@ -221,30 +406,15 @@ class TestPayloads(unit.TestCase):
self.assertEqual(exp_expires_at, expires_at)
self.assertEqual(exp_audit_ids, audit_ids)
- def test_project_scoped_payload_with_non_uuid_user_id(self):
- exp_user_id = 'someNonUuidUserId'
- exp_methods = ['password']
- exp_project_id = uuid.uuid4().hex
- exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
- exp_audit_ids = [provider.random_urlsafe_str()]
-
- payload = token_formatters.ProjectScopedPayload.assemble(
- exp_user_id, exp_methods, exp_project_id, exp_expires_at,
- exp_audit_ids)
-
- (user_id, methods, project_id, expires_at, audit_ids) = (
- token_formatters.ProjectScopedPayload.disassemble(payload))
+ def test_unscoped_payload_with_non_uuid_user_id(self):
+ self._test_unscoped_payload_with_user_id('someNonUuidUserId')
- self.assertEqual(exp_user_id, user_id)
- self.assertEqual(exp_methods, methods)
- self.assertEqual(exp_project_id, project_id)
- self.assertEqual(exp_expires_at, expires_at)
- self.assertEqual(exp_audit_ids, audit_ids)
+ def test_unscoped_payload_with_16_char_non_uuid_user_id(self):
+ self._test_unscoped_payload_with_user_id('0123456789abcdef')
- def test_project_scoped_payload_with_non_uuid_project_id(self):
- exp_user_id = uuid.uuid4().hex
+ def _test_project_scoped_payload_with_ids(self, exp_user_id,
+ exp_project_id):
exp_methods = ['password']
- exp_project_id = 'someNonUuidProjectId'
exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
exp_audit_ids = [provider.random_urlsafe_str()]
@@ -261,8 +431,15 @@ class TestPayloads(unit.TestCase):
self.assertEqual(exp_expires_at, expires_at)
self.assertEqual(exp_audit_ids, audit_ids)
- def test_domain_scoped_payload_with_non_uuid_user_id(self):
- exp_user_id = 'someNonUuidUserId'
+ def test_project_scoped_payload_with_non_uuid_user_id(self):
+ self._test_project_scoped_payload_with_ids('someNonUuidUserId',
+ 'someNonUuidProjectId')
+
+ def test_project_scoped_payload_with_16_char_non_uuid_user_id(self):
+ self._test_project_scoped_payload_with_ids('0123456789abcdef',
+ '0123456789abcdef')
+
+ def _test_domain_scoped_payload_with_user_id(self, exp_user_id):
exp_methods = ['password']
exp_domain_id = uuid.uuid4().hex
exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
@@ -281,32 +458,14 @@ class TestPayloads(unit.TestCase):
self.assertEqual(exp_expires_at, expires_at)
self.assertEqual(exp_audit_ids, audit_ids)
- def test_trust_scoped_payload_with_non_uuid_user_id(self):
- exp_user_id = 'someNonUuidUserId'
- exp_methods = ['password']
- exp_project_id = uuid.uuid4().hex
- exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
- exp_audit_ids = [provider.random_urlsafe_str()]
- exp_trust_id = uuid.uuid4().hex
-
- payload = token_formatters.TrustScopedPayload.assemble(
- exp_user_id, exp_methods, exp_project_id, exp_expires_at,
- exp_audit_ids, exp_trust_id)
-
- (user_id, methods, project_id, expires_at, audit_ids, trust_id) = (
- token_formatters.TrustScopedPayload.disassemble(payload))
+ def test_domain_scoped_payload_with_non_uuid_user_id(self):
+ self._test_domain_scoped_payload_with_user_id('nonUuidUserId')
- self.assertEqual(exp_user_id, user_id)
- self.assertEqual(exp_methods, methods)
- self.assertEqual(exp_project_id, project_id)
- self.assertEqual(exp_expires_at, expires_at)
- self.assertEqual(exp_audit_ids, audit_ids)
- self.assertEqual(exp_trust_id, trust_id)
+ def test_domain_scoped_payload_with_16_char_non_uuid_user_id(self):
+ self._test_domain_scoped_payload_with_user_id('0123456789abcdef')
- def test_trust_scoped_payload_with_non_uuid_project_id(self):
- exp_user_id = uuid.uuid4().hex
+ def _test_trust_scoped_payload_with_ids(self, exp_user_id, exp_project_id):
exp_methods = ['password']
- exp_project_id = 'someNonUuidProjectId'
exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
exp_audit_ids = [provider.random_urlsafe_str()]
exp_trust_id = uuid.uuid4().hex
@@ -325,12 +484,19 @@ class TestPayloads(unit.TestCase):
self.assertEqual(exp_audit_ids, audit_ids)
self.assertEqual(exp_trust_id, trust_id)
- def test_federated_payload_with_non_uuid_ids(self):
- exp_user_id = 'someNonUuidUserId'
+ def test_trust_scoped_payload_with_non_uuid_user_id(self):
+ self._test_trust_scoped_payload_with_ids('someNonUuidUserId',
+ 'someNonUuidProjectId')
+
+ def test_trust_scoped_payload_with_16_char_non_uuid_user_id(self):
+ self._test_trust_scoped_payload_with_ids('0123456789abcdef',
+ '0123456789abcdef')
+
+ def _test_federated_payload_with_ids(self, exp_user_id, exp_group_id):
exp_methods = ['password']
exp_expires_at = utils.isotime(timeutils.utcnow(), subsecond=True)
exp_audit_ids = [provider.random_urlsafe_str()]
- exp_federated_info = {'group_ids': [{'id': 'someNonUuidGroupId'}],
+ exp_federated_info = {'group_ids': [{'id': exp_group_id}],
'idp_id': uuid.uuid4().hex,
'protocol_id': uuid.uuid4().hex}
@@ -352,6 +518,14 @@ class TestPayloads(unit.TestCase):
self.assertEqual(exp_federated_info['protocol_id'],
federated_info['protocol_id'])
+ 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_user_id = 'someNonUuidUserId'
exp_methods = ['token']