Ensure any relevant tokens are revoked when a role is deleted

Add a controller class method to delete tokens for a role, along the
lines of those that exist for deleting tokens for user and project. Ensure
this is called for both the V2 and V3 delete_role APIs.

Fixes bug 1153645

Change-Id: I3c8d70eeb387a18c30df489142ea3aefc2224ae3

keystone/assignment/core.py

@@ -315,6 +315,17 @@ def delete_role(self, role_id):
         self.driver.delete_role(role_id)
         self.get_role.invalidate(self, role_id)
 
+    def list_role_assignments_for_role(self, role_id=None):
+        # NOTE(henry-nash): Currently the efficiency of the key driver
+        # implementation (SQL) of list_role_assignments is severely hampered by
+        # the existence of the multiple grant tables - hence there is little
+        # advantage in pushing the logic of this method down into the driver.
+        # Once the single assignment table is implemented, then this situation
+        # will be different, and this method should have its own driver
+        # implementation.
+        return [r for r in self.driver.list_role_assignments()
+                if r['role_id'] == role_id]
+
 
 class Driver(object):
 

keystone/common/controller.py

@@ -218,6 +218,70 @@ def _delete_tokens_for_project(self, project_id):
         for user_id in user_ids:
             self._delete_tokens_for_user(user_id, project_id=project_id)
 
+    def _delete_tokens_for_role(self, role_id):
+        assignments = self.assignment_api.list_role_assignments_for_role(
+            role_id=role_id)
+
+        # Iterate over the assignments for this role and build the list of
+        # user or user+project IDs for the tokens we need to delete
+        user_ids = set()
+        user_and_project_ids = list()
+        for assignment in assignments:
+            # If we have a project assignment, then record both the user and
+            # project IDs so we can target the right token to delete. If it is
+            # a domain assignment, we might as well kill all the tokens for
+            # the user, since in the vast majority of cases all the tokens
+            # for a user will be within one domain anyway, so not worth
+            # trying to delete tokens for each project in the domain.
+            if 'user_id' in assignment:
+                if 'project_id' in assignment:
+                    user_and_project_ids.append(
+                        (assignment['user_id'], assignment['project_id']))
+                elif 'domain_id' in assignment:
+                    user_ids.add(assignment['user_id'])
+            elif 'group_id' in assignment:
+                # Add in any users for this group, being tolerant of any
+                # cross-driver database integrity errors.
+                try:
+                    users = self.identity_api.list_users_in_group(
+                        assignment['group_id'])
+                except exception.GroupNotFound:
+                    # Ignore it, but log a debug message
+                    if 'project_id' in assignment:
+                        target = _('Project (%s)') % assignment['project_id']
+                    elif 'domain_id' in assignment:
+                        target = _('Domain (%s)') % assignment['domain_id']
+                    else:
+                        target = _('Unknown Target')
+                    msg = (_('Group (%(group)s), referenced in assignment '
+                             'for %(target)s, not found - ignoring.') % {
+                                 'group': assignment['group_id'],
+                                 'target': target})
+                    LOG.debug(msg)
+                    continue
+
+                if 'project_id' in assignment:
+                    for user in users:
+                        user_and_project_ids.append(
+                            (user['id'], assignment['project_id']))
+                elif 'domain_id' in assignment:
+                    for user in users:
+                        user_ids.add(user['id'])
+
+        # Now process the built up lists.  Before issuing calls to delete any
+        # tokens, let's try and minimize the number of calls by pruning out
+        # any user+project deletions where a general token deletion for that
+        # same user is also planned.
+        user_and_project_ids_to_action = []
+        for user_and_project_id in user_and_project_ids:
+            if user_and_project_id[0] not in user_ids:
+                user_and_project_ids_to_action.append(user_and_project_id)
+
+        for user_id in user_ids:
+            self._delete_tokens_for_user(user_id)
+        for user_id, project_id in user_and_project_ids_to_action:
+            self._delete_tokens_for_user(user_id, project_id)
+
     def _require_attribute(self, ref, attr):
         """Ensures the reference contains the specified attribute."""
         if ref.get(attr) is None or ref.get(attr) == '':

keystone/identity/controllers.py

@@ -336,6 +336,10 @@ def create_role(self, context, role):
 
     def delete_role(self, context, role_id):
         self.assert_admin(context)
+        # The driver will delete any assignments for this role.
+        # We must first, however, revoke any tokens for users that have an
+        # assignment with this role.
+        self._delete_tokens_for_role(role_id)
         self.identity_api.delete_role(role_id)
 
     def get_roles(self, context):
@@ -857,7 +861,11 @@ def update_role(self, context, role_id, role):
 
     @controller.protected()
     def delete_role(self, context, role_id):
-        return self.identity_api.delete_role(role_id)
+        # The driver will delete any assignments for this role.
+        # We must first, however, revoke any tokens for users that have an
+        # assignment with this role.
+        self._delete_tokens_for_role(role_id)
+        self.identity_api.delete_role(role_id)
 
     def _require_domain_xor_project(self, domain_id, project_id):
         if (domain_id and project_id) or (not domain_id and not project_id):
@@ -1263,7 +1271,7 @@ def list_role_assignments(self, context, filters):
         # to pass the filters into the driver call, so that the list size is
         # kept a minimum.
 
-        refs = self.identity_api.list_role_assignments()
+        refs = self.assignment_api.list_role_assignments()
         formatted_refs = (
             [self._format_entity(x) for x in refs
              if self._filter_inherited(x)])

keystone/identity/core.py

@@ -555,9 +555,6 @@ def add_user_to_project(self, tenant_id, user_id):
     def remove_user_from_project(self, tenant_id, user_id):
         return self.assignment_api.remove_user_from_project(tenant_id, user_id)
 
-    def list_role_assignments(self):
-        return self.assignment_api.list_role_assignments()
-
 
 class Driver(object):
     """Interface description for an Identity driver."""

keystone/tests/test_auth.py

@@ -19,6 +19,7 @@
 from keystone import auth
 from keystone import config
 from keystone import exception
+from keystone import identity
 from keystone.openstack.common import timeutils
 from keystone import tests
 from keystone.tests import default_fixtures
@@ -28,6 +29,7 @@
 
 CONF = config.CONF
 TIME_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'
+DEFAULT_DOMAIN_ID = CONF.identity.default_domain_id
 
 
 def _build_user_auth(token=None, user_id=None, username=None,
@@ -401,6 +403,40 @@ def test_token_auth_with_binding(self):
         bind = scoped_token['access']['token']['bind']
         self.assertEqual(bind['kerberos'], 'FOO')
 
+    def test_deleting_role_revokes_token(self):
+        role_controller = identity.controllers.Role()
+        project1 = {'id': 'Project1', 'name': uuid.uuid4().hex,
+                    'domain_id': DEFAULT_DOMAIN_ID}
+        self.assignment_api.create_project(project1['id'], project1)
+        role_one = {'id': 'role_one', 'name': uuid.uuid4().hex}
+        self.assignment_api.create_role(role_one['id'], role_one)
+        self.identity_api.add_role_to_user_and_project(
+            self.user_foo['id'], project1['id'], role_one['id'])
+        no_context = {}
+
+        # Get a scoped token for the tenant
+        body_dict = _build_user_auth(
+            username=self.user_foo['name'],
+            password=self.user_foo['password'],
+            tenant_name=project1['name'])
+        token = self.controller.authenticate(no_context, body_dict)
+        # Ensure it is valid
+        token_id = token['access']['token']['id']
+        self.controller.validate_token(
+            dict(is_admin=True, query_string={}),
+            token_id=token_id)
+
+        # Delete the role, which should invalidate the token
+        role_controller.delete_role(
+            dict(is_admin=True, query_string={}), role_one['id'])
+
+        # Check the token is now invalid
+        self.assertRaises(
+            exception.Unauthorized,
+            self.controller.validate_token,
+            dict(is_admin=True, query_string={}),
+            token_id=token_id)
+
 
 class AuthWithPasswordCredentials(AuthTest):
     def setUp(self):

keystone/tests/test_backend.py

@@ -438,6 +438,8 @@ def test_list_role_assignments_unfiltered(self):
         - Create a grant of each type (user/group on project/domain)
         - Check the number of assignments has gone up by 4 and that
           the entries we added are in the list returned
+        - Check that if we list assignments by role_id, then we get back
+          assignments that only contain that role.
 
         """
         new_domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
@@ -455,8 +457,11 @@ def test_list_role_assignments_unfiltered(self):
                        'domain_id': new_domain['id']}
         self.assignment_api.create_project(new_project['id'], new_project)
 
-        # First check how many role grant already exist
-        existing_assignments = len(self.identity_api.list_role_assignments())
+        # First check how many role grants already exist
+        existing_assignments = len(self.assignment_api.list_role_assignments())
+        existing_assignments_for_role = len(
+            self.assignment_api.list_role_assignments_for_role(
+                role_id='admin'))
 
         # Now create the grants (roles are defined in default_fixtures)
         self.identity_api.create_grant(user_id=new_user['id'],
@@ -472,8 +477,8 @@ def test_list_role_assignments_unfiltered(self):
                                        project_id=new_project['id'],
                                        role_id='admin')
 
-        # Read back the list of assignments - check it is gone up by 4
-        assignment_list = self.identity_api.list_role_assignments()
+        # Read back the full list of assignments - check it is gone up by 4
+        assignment_list = self.assignment_api.list_role_assignments()
         self.assertEquals(len(assignment_list), existing_assignments + 4)
 
         # Now check that each of our four new entries are in the list
@@ -494,6 +499,28 @@ def test_list_role_assignments_unfiltered(self):
              'role_id': 'admin'},
             assignment_list)
 
+        # Read back the list of assignments for just the admin role, checking
+        # this only goes up by two.
+        assignment_list = self.assignment_api.list_role_assignments_for_role(
+            role_id='admin')
+        self.assertEquals(len(assignment_list),
+                          existing_assignments_for_role + 2)
+
+        # Now check that each of our two new entries are in the list
+        self.assertIn(
+            {'group_id': new_group['id'], 'domain_id': new_domain['id'],
+             'role_id': 'admin'},
+            assignment_list)
+        self.assertIn(
+            {'group_id': new_group['id'], 'project_id': new_project['id'],
+             'role_id': 'admin'},
+            assignment_list)
+
+    def test_list_role_assignments_bad_role(self):
+        assignment_list = self.assignment_api.list_role_assignments_for_role(
+            role_id=uuid.uuid4().hex)
+        self.assertEqual(assignment_list, [])
+
     def test_add_duplicate_role_grant(self):
         roles_ref = self.identity_api.get_roles_for_user_and_project(
             self.user_foo['id'], self.tenant_bar['id'])

keystone/tests/test_backend_ldap.py

@@ -208,7 +208,10 @@ def test_get_roles_for_user_and_domain(self):
         self.skipTest('N/A: LDAP does not support multiple domains')
 
     def test_list_role_assignments_unfiltered(self):
-        self.skipTest('Blocked by bug 1195019')
+        self.skipTest('Blocked by bug 1221805')
+
+    def test_list_role_assignments_bad_role(self):
+        self.skipTest('Blocked by bug 1221805')
 
     def test_multi_group_grants_on_project_domain(self):
         self.skipTest('Blocked by bug 1101287')