# 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 datetime
import uuid
import mock
from oslo_utils import timeutils
from six.moves import range
from testtools import matchers
from keystone.common import utils
from keystone.contrib.revoke import model
from keystone import exception
from keystone.tests import unit
from keystone.tests.unit import test_backend_sql
from keystone.token import provider
def _new_id():
return uuid.uuid4().hex
def _future_time():
expire_delta = datetime.timedelta(seconds=1000)
future_time = timeutils.utcnow() + expire_delta
return future_time
def _past_time():
expire_delta = datetime.timedelta(days=-1000)
past_time = timeutils.utcnow() + expire_delta
return past_time
def _sample_blank_token():
issued_delta = datetime.timedelta(minutes=-2)
issued_at = timeutils.utcnow() + issued_delta
token_data = model.blank_token_data(issued_at)
return token_data
def _matches(event, token_values):
"""See if the token matches the revocation event.
Used as a secondary check on the logic to Check
By Tree Below: This is abrute force approach to checking.
Compare each attribute from the event with the corresponding
value from the token. If the event does not have a value for
the attribute, a match is still possible. If the event has a
value for the attribute, and it does not match the token, no match
is possible, so skip the remaining checks.
:param event one revocation event to match
:param token_values dictionary with set of values taken from the
token
:returns if the token matches the revocation event, indicating the
token has been revoked
"""
# The token has three attributes that can match the user_id
if event.user_id is not None:
for attribute_name in ['user_id', 'trustor_id', 'trustee_id']:
if event.user_id == token_values[attribute_name]:
break
else:
return False
# The token has two attributes that can match the domain_id
if event.domain_id is not None:
for attribute_name in ['identity_domain_id', 'assignment_domain_id']:
if event.domain_id == token_values[attribute_name]:
break
else:
return False
if event.domain_scope_id is not None:
if event.domain_scope_id != token_values['assignment_domain_id']:
return False
# If any one check does not match, the while token does
# not match the event. The numerous return False indicate
# that the token is still valid and short-circuits the
# rest of the logic.
attribute_names = ['project_id',
'expires_at', 'trust_id', 'consumer_id',
'access_token_id', 'audit_id', 'audit_chain_id']
for attribute_name in attribute_names:
if getattr(event, attribute_name) is not None:
if (getattr(event, attribute_name) !=
token_values[attribute_name]):
return False
if event.role_id is not None:
roles = token_values['roles']
for role in roles:
if event.role_id == role:
break
else:
return False
if token_values['issued_at'] > event.issued_before:
return False
return True
class RevokeTests(object):
def test_list(self):
self.revoke_api.revoke_by_user(user_id=1)
self.assertEqual(1, len(self.revoke_api.list_events()))
self.revoke_api.revoke_by_user(user_id=2)
self.assertEqual(2, len(self.revoke_api.list_events()))
def test_list_since(self):
self.revoke_api.revoke_by_user(user_id=1)
self.revoke_api.revoke_by_user(user_id=2)
past = timeutils.utcnow() - datetime.timedelta(seconds=1000)
self.assertEqual(2, len(self.revoke_api.list_events(past)))
future = timeutils.utcnow() + datetime.timedelta(seconds=1000)
self.assertEqual(0, len(self.revoke_api.list_events(future)))
def test_past_expiry_are_removed(self):
user_id = 1
self.revoke_api.revoke_by_expiration(user_id, _future_time())
self.assertEqual(1, len(self.revoke_api.list_events()))
event = model.RevokeEvent()
event.revoked_at = _past_time()
self.revoke_api.revoke(event)
self.assertEqual(1, len(self.revoke_api.list_events()))
@mock.patch.object(timeutils, 'utcnow')
def test_expired_events_removed_validate_token_success(self, mock_utcnow):
def _sample_token_values():
token = _sample_blank_token()
token['expires_at'] = utils.isotime(_future_time(),
subsecond=True)
return token
now = datetime.datetime.utcnow()
now_plus_2h = now + datetime.timedelta(hours=2)
mock_utcnow.return_value = now
# Build a token and validate it. This will seed the cache for the
# future 'synchronize' call.
token_values = _sample_token_values()
user_id = _new_id()
self.revoke_api.revoke_by_user(user_id)
token_values['user_id'] = user_id
self.assertRaises(exception.TokenNotFound,
self.revoke_api.check_token,
token_values)
# Move our clock forward by 2h, build a new token and validate it.
# 'synchronize' should now be exercised and remove old expired events
mock_utcnow.return_value = now_plus_2h
self.revoke_api.revoke_by_expiration(_new_id(), now_plus_2h)
# should no longer throw an exception
self.revoke_api.check_token(token_values)
def test_revoke_by_expiration_project_and_domain_fails(self):
user_id = _new_id()
expires_at = utils.isotime(_future_time(), subsecond=True)
domain_id = _new_id()
project_id = _new_id()
self.assertThat(
lambda: self.revoke_api.revoke_by_expiration(
user_id, expires_at, domain_id=domain_id,
project_id=project_id),
matchers.raises(exception.UnexpectedError))
class SqlRevokeTests(test_backend_sql.SqlTests, RevokeTests):
def config_overrides(self):
super(SqlRevokeTests, self).config_overrides()
self.config_fixture.config(group='revoke', driver='sql')
self.config_fixture.config(
group='token',
provider='pki',
revoke_by_id=False)
class KvsRevokeTests(unit.TestCase, RevokeTests):
def config_overrides(self):
super(KvsRevokeTests, self).config_overrides()
self.config_fixture.config(group='revoke', driver='kvs')
self.config_fixture.config(
group='token',
provider='pki',
revoke_by_id=False)
def setUp(self):
super(KvsRevokeTests, self).setUp()
self.load_backends()
class RevokeTreeTests(unit.TestCase):
def setUp(self):
super(RevokeTreeTests, self).setUp()
self.events = []
self.tree = model.RevokeTree()
self._sample_data()
def _sample_data(self):
user_ids = []
project_ids = []
role_ids = []
for i in range(0, 3):
user_ids.append(_new_id())
project_ids.append(_new_id())
role_ids.append(_new_id())
project_tokens = []
i = len(project_tokens)
project_tokens.append(_sample_blank_token())
project_tokens[i]['user_id'] = user_ids[0]
project_tokens[i]['project_id'] = project_ids[0]
project_tokens[i]['roles'] = [role_ids[1]]
i = len(project_tokens)
project_tokens.append(_sample_blank_token())
project_tokens[i]['user_id'] = user_ids[1]
project_tokens[i]['project_id'] = project_ids[0]
project_tokens[i]['roles'] = [role_ids[0]]
i = len(project_tokens)
project_tokens.append(_sample_blank_token())
project_tokens[i]['user_id'] = user_ids[0]
project_tokens[i]['project_id'] = project_ids[1]
project_tokens[i]['roles'] = [role_ids[0]]
token_to_revoke = _sample_blank_token()
token_to_revoke['user_id'] = user_ids[0]
token_to_revoke['project_id'] = project_ids[0]
token_to_revoke['roles'] = [role_ids[0]]
self.project_tokens = project_tokens
self.user_ids = user_ids
self.project_ids = project_ids
self.role_ids = role_ids
self.token_to_revoke = token_to_revoke
def _assertTokenRevoked(self, token_data):
self.assertTrue(any([_matches(e, token_data) for e in self.events]))
return self.assertTrue(self.tree.is_revoked(token_data),
'Token should be revoked')
def _assertTokenNotRevoked(self, token_data):
self.assertFalse(any([_matches(e, token_data) for e in self.events]))
return self.assertFalse(self.tree.is_revoked(token_data),
'Token should not be revoked')
def _revoke_by_user(self, user_id):
return self.tree.add_event(
model.RevokeEvent(user_id=user_id))
def _revoke_by_audit_id(self, audit_id):
event = self.tree.add_event(
model.RevokeEvent(audit_id=audit_id))
self.events.append(event)
return event
def _revoke_by_audit_chain_id(self, audit_chain_id, project_id=None,
domain_id=None):
event = self.tree.add_event(
model.RevokeEvent(audit_chain_id=audit_chain_id,
project_id=project_id,
domain_id=domain_id)
)
self.events.append(event)
return event
def _revoke_by_expiration(self, user_id, expires_at, project_id=None,
domain_id=None):
event = self.tree.add_event(
model.RevokeEvent(user_id=user_id,
expires_at=expires_at,
project_id=project_id,
domain_id=domain_id))
self.events.append(event)
return event
def _revoke_by_grant(self, role_id, user_id=None,
domain_id=None, project_id=None):
event = self.tree.add_event(
model.RevokeEvent(user_id=user_id,
role_id=role_id,
domain_id=domain_id,
project_id=project_id))
self.events.append(event)
return event
def _revoke_by_user_and_project(self, user_id, project_id):
event = self.tree.add_event(
model.RevokeEvent(project_id=project_id,
user_id=user_id))
self.events.append(event)
return event
def _revoke_by_project_role_assignment(self, project_id, role_id):
event = self.tree.add_event(
model.RevokeEvent(project_id=project_id,
role_id=role_id))
self.events.append(event)
return event
def _revoke_by_domain_role_assignment(self, domain_id, role_id):
event = self.tree.add_event(
model.RevokeEvent(domain_id=domain_id,
role_id=role_id))
self.events.append(event)
return event
def _revoke_by_domain(self, domain_id):
event = self.tree.add_event(model.RevokeEvent(domain_id=domain_id))
self.events.append(event)
def _user_field_test(self, field_name):
user_id = _new_id()
event = self._revoke_by_user(user_id)
self.events.append(event)
token_data_u1 = _sample_blank_token()
token_data_u1[field_name] = user_id
self._assertTokenRevoked(token_data_u1)
token_data_u2 = _sample_blank_token()
token_data_u2[field_name] = _new_id()
self._assertTokenNotRevoked(token_data_u2)
self.tree.remove_event(event)
self.events.remove(event)
self._assertTokenNotRevoked(token_data_u1)
def test_revoke_by_user(self):
self._user_field_test('user_id')
def test_revoke_by_user_matches_trustee(self):
self._user_field_test('trustee_id')
def test_revoke_by_user_matches_trustor(self):
self._user_field_test('trustor_id')
def test_by_user_expiration(self):
future_time = _future_time()
user_id = 1
event = self._revoke_by_expiration(user_id, future_time)
token_data_1 = _sample_blank_token()
token_data_1['user_id'] = user_id
token_data_1['expires_at'] = future_time.replace(microsecond=0)
self._assertTokenRevoked(token_data_1)
token_data_2 = _sample_blank_token()
token_data_2['user_id'] = user_id
expire_delta = datetime.timedelta(seconds=2000)
future_time = timeutils.utcnow() + expire_delta
token_data_2['expires_at'] = future_time
self._assertTokenNotRevoked(token_data_2)
self.remove_event(event)
self._assertTokenNotRevoked(token_data_1)
def test_revoke_by_audit_id(self):
audit_id = provider.audit_info(parent_audit_id=None)[0]
token_data_1 = _sample_blank_token()
# Audit ID and Audit Chain ID are populated with the same value
# if the token is an original token
token_data_1['audit_id'] = audit_id
token_data_1['audit_chain_id'] = audit_id
event = self._revoke_by_audit_id(audit_id)
self._assertTokenRevoked(token_data_1)
audit_id_2 = provider.audit_info(parent_audit_id=audit_id)[0]
token_data_2 = _sample_blank_token()
token_data_2['audit_id'] = audit_id_2
token_data_2['audit_chain_id'] = audit_id
self._assertTokenNotRevoked(token_data_2)
self.remove_event(event)
self._assertTokenNotRevoked(token_data_1)
def test_revoke_by_audit_chain_id(self):
audit_id = provider.audit_info(parent_audit_id=None)[0]
token_data_1 = _sample_blank_token()
# Audit ID and Audit Chain ID are populated with the same value
# if the token is an original token
token_data_1['audit_id'] = audit_id
token_data_1['audit_chain_id'] = audit_id
event = self._revoke_by_audit_chain_id(audit_id)
self._assertTokenRevoked(token_data_1)
audit_id_2 = provider.audit_info(parent_audit_id=audit_id)[0]
token_data_2 = _sample_blank_token()
token_data_2['audit_id'] = audit_id_2
token_data_2['audit_chain_id'] = audit_id
self._assertTokenRevoked(token_data_2)
self.remove_event(event)
self._assertTokenNotRevoked(token_data_1)
self._assertTokenNotRevoked(token_data_2)
def test_by_user_project(self):
# When a user has a project-scoped token and the project-scoped token
# is revoked then the token is revoked.
user_id = _new_id()
project_id = _new_id()
future_time = _future_time()
token_data = _sample_blank_token()
token_data['user_id'] = user_id
token_data['project_id'] = project_id
token_data['expires_at'] = future_time.replace(microsecond=0)
self._revoke_by_expiration(user_id, future_time, project_id=project_id)
self._assertTokenRevoked(token_data)
def test_by_user_domain(self):
# When a user has a domain-scoped token and the domain-scoped token
# is revoked then the token is revoked.
user_id = _new_id()
domain_id = _new_id()
future_time = _future_time()
token_data = _sample_blank_token()
token_data['user_id'] = user_id
token_data['assignment_domain_id'] = domain_id
token_data['expires_at'] = future_time.replace(microsecond=0)
self._revoke_by_expiration(user_id, future_time, domain_id=domain_id)
self._assertTokenRevoked(token_data)
def remove_event(self, event):
self.events.remove(event)
self.tree.remove_event(event)
def test_by_project_grant(self):
token_to_revoke = self.token_to_revoke
tokens = self.project_tokens
self._assertTokenNotRevoked(token_to_revoke)
for token in tokens:
self._assertTokenNotRevoked(token)
event = self._revoke_by_grant(role_id=self.role_ids[0],
user_id=self.user_ids[0],
project_id=self.project_ids[0])
self._assertTokenRevoked(token_to_revoke)
for token in tokens:
self._assertTokenNotRevoked(token)
self.remove_event(event)
self._assertTokenNotRevoked(token_to_revoke)
for token in tokens:
self._assertTokenNotRevoked(token)
token_to_revoke['roles'] = [self.role_ids[0],
self.role_ids[1],
self.role_ids[2]]
event = self._revoke_by_grant(role_id=self.role_ids[0],
user_id=self.user_ids[0],
project_id=self.project_ids[0])
self._assertTokenRevoked(token_to_revoke)
self.remove_event(event)
self._assertTokenNotRevoked(token_to_revoke)
event = self._revoke_by_grant(role_id=self.role_ids[1],
user_id=self.user_ids[0],
project_id=self.project_ids[0])
self._assertTokenRevoked(token_to_revoke)
self.remove_event(event)
self._assertTokenNotRevoked(token_to_revoke)
self._revoke_by_grant(role_id=self.role_ids[0],
user_id=self.user_ids[0],
project_id=self.project_ids[0])
self._revoke_by_grant(role_id=self.role_ids[1],
user_id=self.user_ids[0],
project_id=self.project_ids[0])
self._revoke_by_grant(role_id=self.role_ids[2],
user_id=self.user_ids[0],
project_id=self.project_ids[0])
self._assertTokenRevoked(token_to_revoke)
def test_by_project_and_user_and_role(self):
user_id1 = _new_id()
user_id2 = _new_id()
project_id = _new_id()
self.events.append(self._revoke_by_user(user_id1))
self.events.append(
self._revoke_by_user_and_project(user_id2, project_id))
token_data = _sample_blank_token()
token_data['user_id'] = user_id2
token_data['project_id'] = project_id
self._assertTokenRevoked(token_data)
def test_by_domain_user(self):
# If revoke a domain, then a token for a user in the domain is revoked
user_id = _new_id()
domain_id = _new_id()
token_data = _sample_blank_token()
token_data['user_id'] = user_id
token_data['identity_domain_id'] = domain_id
self._revoke_by_domain(domain_id)
self._assertTokenRevoked(token_data)
def test_by_domain_project(self):
# If revoke a domain, then a token scoped to a project in the domain
# is revoked.
user_id = _new_id()
user_domain_id = _new_id()
project_id = _new_id()
project_domain_id = _new_id()
token_data = _sample_blank_token()
token_data['user_id'] = user_id
token_data['identity_domain_id'] = user_domain_id
token_data['project_id'] = project_id
token_data['assignment_domain_id'] = project_domain_id
self._revoke_by_domain(project_domain_id)
self._assertTokenRevoked(token_data)
def test_by_domain_domain(self):
# If revoke a domain, then a token scoped to the domain is revoked.
user_id = _new_id()
user_domain_id = _new_id()
domain_id = _new_id()
token_data = _sample_blank_token()
token_data['user_id'] = user_id
token_data['identity_domain_id'] = user_domain_id
token_data['assignment_domain_id'] = domain_id
self._revoke_by_domain(domain_id)
self._assertTokenRevoked(token_data)
def _assertEmpty(self, collection):
return self.assertEqual(0, len(collection), "collection not empty")
def _assertEventsMatchIteration(self, turn):
self.assertEqual(1, len(self.tree.revoke_map))
self.assertEqual(turn + 1, len(self.tree.revoke_map
['trust_id=*']
['consumer_id=*']
['access_token_id=*']
['audit_id=*']
['audit_chain_id=*']))
# two different functions add domain_ids, +1 for None
self.assertEqual(2 * turn + 1, len(self.tree.revoke_map
['trust_id=*']
['consumer_id=*']
['access_token_id=*']
['audit_id=*']
['audit_chain_id=*']
['expires_at=*']))
# two different functions add project_ids, +1 for None
self.assertEqual(2 * turn + 1, len(self.tree.revoke_map
['trust_id=*']
['consumer_id=*']
['access_token_id=*']
['audit_id=*']
['audit_chain_id=*']
['expires_at=*']
['domain_id=*']))
# 10 users added
self.assertEqual(turn, len(self.tree.revoke_map
['trust_id=*']
['consumer_id=*']
['access_token_id=*']
['audit_id=*']
['audit_chain_id=*']
['expires_at=*']
['domain_id=*']
['project_id=*']))
def test_cleanup(self):
events = self.events
self._assertEmpty(self.tree.revoke_map)
expiry_base_time = _future_time()
for i in range(0, 10):
events.append(
self._revoke_by_user(_new_id()))
args = (_new_id(),
expiry_base_time + datetime.timedelta(seconds=i))
events.append(
self._revoke_by_expiration(*args))
self.assertEqual(i + 2, len(self.tree.revoke_map
['trust_id=*']
['consumer_id=*']
['access_token_id=*']
['audit_id=*']
['audit_chain_id=*']),
'adding %s to %s' % (args,
self.tree.revoke_map))
events.append(
self._revoke_by_project_role_assignment(_new_id(), _new_id()))
events.append(
self._revoke_by_domain_role_assignment(_new_id(), _new_id()))
events.append(
self._revoke_by_domain_role_assignment(_new_id(), _new_id()))
events.append(
self._revoke_by_user_and_project(_new_id(), _new_id()))
self._assertEventsMatchIteration(i + 1)
for event in self.events:
self.tree.remove_event(event)
self._assertEmpty(self.tree.revoke_map)