residual grants after delete action (bug1125637)

remove all applicable grants when role is deleted
(sql/kvs solution only)

Fixes: bug #1125637

Change-Id: I3a958c6d56739e37a95f6c713fab154827e9ceca
(cherry picked from commit e16742b)

keystone/identity/backends/kvs.py

@@ -402,21 +402,33 @@ def update_role(self, role_id, role):
         return role
 
     def delete_role(self, role_id):
-        try:
-            self.db.delete('role-%s' % role_id)
-            metadata_keys = filter(lambda x: x.startswith("metadata-"),
-                                   self.db.keys())
-            for key in metadata_keys:
-                tenant_id = key.split('-')[1]
-                user_id = key.split('-')[2]
-                try:
-                    self.remove_role_from_user_and_project(user_id,
-                                                           tenant_id,
-                                                           role_id)
-                except exception.RoleNotFound:
-                    pass
-        except exception.NotFound:
-            raise exception.RoleNotFound(role_id=role_id)
+        self.get_role(role_id)
+        metadata_keys = filter(lambda x: x.startswith("metadata-"),
+                               self.db.keys())
+        for key in metadata_keys:
+            meta_id1 = key.split('-')[1]
+            meta_id2 = key.split('-')[2]
+            try:
+                self.delete_grant(role_id, project_id=meta_id1,
+                                  user_id=meta_id2)
+            except exception.NotFound:
+                pass
+            try:
+                self.delete_grant(role_id, project_id=meta_id1,
+                                  group_id=meta_id2)
+            except exception.NotFound:
+                pass
+            try:
+                self.delete_grant(role_id, domain_id=meta_id1,
+                                  user_id=meta_id2)
+            except exception.NotFound:
+                pass
+            try:
+                self.delete_grant(role_id, domain_id=meta_id1,
+                                  group_id=meta_id2)
+            except exception.NotFound:
+                pass
+        self.db.delete('role-%s' % role_id)
         role_list = set(self.db.get('role_list', []))
         role_list.remove(role_id)
         self.db.set('role_list', list(role_list))

keystone/identity/backends/sql.py

@@ -962,14 +962,29 @@ def delete_role(self, role_id):
 
         with session.begin():
             for metadata_ref in session.query(UserProjectGrant):
-                metadata = metadata_ref.to_dict()
                 try:
-                    self.remove_role_from_user_and_project(
-                        metadata['user_id'], metadata['project_id'], role_id)
+                    self.delete_grant(role_id, user_id=metadata_ref.user_id,
+                                      project_id=metadata_ref.project_id)
+                except exception.RoleNotFound:
+                    pass
+            for metadata_ref in session.query(UserDomainGrant):
+                try:
+                    self.delete_grant(role_id, user_id=metadata_ref.user_id,
+                                      domain_id=metadata_ref.domain_id)
+                except exception.RoleNotFound:
+                    pass
+            for metadata_ref in session.query(GroupProjectGrant):
+                try:
+                    self.delete_grant(role_id, group_id=metadata_ref.group_id,
+                                      project_id=metadata_ref.project_id)
+                except exception.RoleNotFound:
+                    pass
+            for metadata_ref in session.query(GroupDomainGrant):
+                try:
+                    self.delete_grant(role_id, group_id=metadata_ref.group_id,
+                                      domain_id=metadata_ref.domain_id)
                 except exception.RoleNotFound:
                     pass
-
-            # FIXME(dolph): user-domain metadata needs to be updated
 
             if not session.query(Role).filter_by(id=role_id).delete():
                 raise exception.RoleNotFound(role_id=role_id)

tests/test_backend.py

@@ -1136,7 +1136,6 @@ def test_multi_role_grant_by_user_group_on_project_domain(self):
         self.assertIn(role_list[7], roles_ref)
 
     def test_delete_role_with_user_and_group_grants(self):
-        raise nose.exc.SkipTest('Blocked by bug 1097472')
         role1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
         self.identity_api.create_role(role1['id'], role1)
         domain1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
@@ -1180,22 +1179,22 @@ def test_delete_role_with_user_and_group_grants(self):
             domain_id=domain1['id'])
         self.assertEquals(len(roles_ref), 1)
         self.identity_api.delete_role(role1['id'])
-        self.assertRaises(exception.RoleNotFound,
-                          self.identity_api.list_grants,
-                          user_id=user1['id'],
-                          project_id=project1['id'])
-        self.assertRaises(exception.RoleNotFound,
-                          self.identity_api.list_grants,
-                          group_id=group1['id'],
-                          project_id=project1['id'])
-        self.assertRaises(exception.RoleNotFound,
-                          self.identity_api.list_grants,
-                          user_id=user1['id'],
-                          domain_id=domain1['id'])
-        self.assertRaises(exception.RoleNotFound,
-                          self.identity_api.list_grants,
-                          group_id=group1['id'],
-                          domain_id=domain1['id'])
+        roles_ref = self.identity_api.list_grants(
+            user_id=user1['id'],
+            project_id=project1['id'])
+        self.assertEquals(len(roles_ref), 0)
+        roles_ref = self.identity_api.list_grants(
+            group_id=group1['id'],
+            project_id=project1['id'])
+        self.assertEquals(len(roles_ref), 0)
+        roles_ref = self.identity_api.list_grants(
+            user_id=user1['id'],
+            domain_id=domain1['id'])
+        self.assertEquals(len(roles_ref), 0)
+        roles_ref = self.identity_api.list_grants(
+            group_id=group1['id'],
+            domain_id=domain1['id'])
+        self.assertEquals(len(roles_ref), 0)
 
     def test_delete_user_with_group_project_domain_links(self):
         role1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}