Fix BOLA in access rule GET/DELETE and ownership check in CREATE

GET /v3/users/{user_id}/access_rules/{access_rule_id} and the
corresponding DELETE pass the RBAC policy check when the requester's
user_id matches the path user_id, but the backend then fetches or
deletes the target rule globally by access_rule_id with no owner
filter. An authenticated user who knows another user's access_rule_id
can read that rule's details or delete it by substituting their own
user_id in the path.

Fix by threading user_id from the URL path through the manager down to
the SQL query, so the lookup is scoped to external_id AND user_id. A
mismatched owner produces AccessRuleNotFound (HTTP 404) rather than
returning or deleting the rule.

Also add an ownership check in create_application_credential: when an
existing rule is found by external_id, verify it belongs to the
creating user before linking it to the new application credential.

Four regression tests are added covering the BOLA GET, BOLA DELETE,
and cross-user reuse scenarios across all project-scoped personas.

Closes-Bug: #2148825

Change-Id: I69954f0549cc9aa731b952e3b8cf546a7da1ce8e
Signed-off-by: Grzegorz Grasza <xek@redhat.com>
Assisted-by: Claude Sonnet 4.6 <noreply@anthropic.com>

keystone/api/users.py

@@ -850,7 +850,7 @@ class UserAccessRuleGetDeleteResource(ks_flask.ResourceBase):
             build_target=_build_user_target_enforcement,
         )
         ref = PROVIDERS.application_credential_api.get_access_rule(
-            access_rule_id
+            access_rule_id, user_id=user_id
         )
         return self.wrap_member(ref)
 
@@ -866,7 +866,7 @@ class UserAccessRuleGetDeleteResource(ks_flask.ResourceBase):
             build_target=_build_user_target_enforcement,
         )
         PROVIDERS.application_credential_api.delete_access_rule(
-            access_rule_id, initiator=self.audit_initiator
+            access_rule_id, initiator=self.audit_initiator, user_id=user_id
         )
         return None, http.client.NO_CONTENT
 

keystone/application_credential/backends/base.py

@@ -94,10 +94,13 @@ class ApplicationCredentialDriverBase(metaclass=abc.ABCMeta):
         raise exception.NotImplemented()  # pragma: no cover
 
     @abc.abstractmethod
-    def get_access_rule(self, access_rule_id):
+    def get_access_rule(self, access_rule_id, user_id=None):
         """Get an access rule by its ID.
 
         :param str access_rule_id: Access Rule ID
+        :param str user_id: If provided, the rule must belong to this user.
+            Returns AccessRuleNotFound if the rule exists but belongs to a
+            different user, preventing cross-user BOLA/IDOR.
         """
         raise exception.NotImplemented()  # pragma: no cover
 
@@ -113,10 +116,13 @@ class ApplicationCredentialDriverBase(metaclass=abc.ABCMeta):
         raise exception.NotImplemented()  # pragma: no cover
 
     @abc.abstractmethod
-    def delete_access_rule(self, access_rule_id):
+    def delete_access_rule(self, access_rule_id, user_id=None):
         """Delete one access rule.
 
         :param str access_rule_id: Access Rule ID
+        :param str user_id: If provided, the rule must belong to this user.
+            Returns AccessRuleNotFound if the rule exists but belongs to a
+            different user, preventing cross-user BOLA/IDOR.
         """
         raise exception.NotImplemented()  # pragma: no cover
 

keystone/application_credential/backends/sql.py

@@ -173,6 +173,12 @@ class ApplicationCredential(base.ApplicationCredentialDriverBase):
                         .filter_by(external_id=access_rule['id'])
                         .first()
                     )
+                    if access_rule_ref is not None and (
+                        access_rule_ref.user_id != app_cred['user_id']
+                    ):
+                        raise exception.AccessRuleNotFound(
+                            access_rule_id=access_rule['id']
+                        )
                     if not access_rule_ref:
                         query = session.query(AccessRuleModel)
                         access_rule_ref = query.filter_by(
@@ -270,11 +276,13 @@ class ApplicationCredential(base.ApplicationCredentialDriverBase):
             query = query.filter_by(project_id=project_id)
             query.delete()
 
-    def get_access_rule(self, access_rule_id):
+    def get_access_rule(self, access_rule_id, user_id=None):
         with sql.session_for_read() as session:
             query = session.query(AccessRuleModel).filter_by(
                 external_id=access_rule_id
             )
+            if user_id is not None:
+                query = query.filter_by(user_id=user_id)
             ref = query.first()
             if not ref:
                 raise exception.AccessRuleNotFound(
@@ -289,11 +297,15 @@ class ApplicationCredential(base.ApplicationCredentialDriverBase):
             refs = sql.filter_limit_query(AccessRuleModel, query, hints)
             return [self._access_rule_to_dict(ref) for ref in refs]
 
-    def delete_access_rule(self, access_rule_id):
+    def delete_access_rule(self, access_rule_id, user_id=None):
         try:
             with sql.session_for_write() as session:
-                query = session.query(AccessRuleModel)
-                ref = query.filter_by(external_id=access_rule_id).first()
+                query = session.query(AccessRuleModel).filter_by(
+                    external_id=access_rule_id
+                )
+                if user_id is not None:
+                    query = query.filter_by(user_id=user_id)
+                ref = query.first()
                 if not ref:
                     raise exception.AccessRuleNotFound(
                         access_rule_id=access_rule_id

keystone/application_credential/core.py

@@ -179,14 +179,15 @@ class Manager(manager.Manager):
         return [self._process_app_cred(app_cred) for app_cred in app_cred_list]
 
     @MEMOIZE
-    def get_access_rule(self, access_rule_id):
+    def get_access_rule(self, access_rule_id, user_id=None):
         """Get access rule details.
 
         :param str access_rule_id: Access Rule ID
+        :param str user_id: If provided, the rule must belong to this user.
 
         :returns: an access rule
         """
-        return self.driver.get_access_rule(access_rule_id)
+        return self.driver.get_access_rule(access_rule_id, user_id=user_id)
 
     def list_access_rules_for_user(self, user_id, hints=None):
         """List access rules for user.
@@ -258,17 +259,18 @@ class Manager(manager.Manager):
         for app_cred in app_creds:
             self.get_application_credential.invalidate(self, app_cred['id'])
 
-    def delete_access_rule(self, access_rule_id, initiator=None):
+    def delete_access_rule(self, access_rule_id, initiator=None, user_id=None):
         """Delete an access rule.
 
         :param str: access_rule_id: Access Rule ID
         :param initiator: CADF initiator
+        :param str user_id: If provided, the rule must belong to this user.
 
         :raises keystone.exception.AccessRuleNotFound: If the access rule
             doesn't exist.
         """
-        self.driver.delete_access_rule(access_rule_id)
-        self.get_access_rule.invalidate(self, access_rule_id)
+        self.driver.delete_access_rule(access_rule_id, user_id=user_id)
+        self.get_access_rule.invalidate(self, access_rule_id, user_id)
         notifications.Audit.deleted(
             self._ACCESS_RULE, access_rule_id, initiator
         )
@@ -285,7 +287,7 @@ class Manager(manager.Manager):
         )
         self.driver.delete_access_rules_for_user(user_id)
         for rule in access_rules:
-            self.get_access_rule.invalidate(self, rule['id'])
+            self.get_access_rule.invalidate(self, rule['id'], user_id)
             notifications.Audit.deleted(
                 self._ACCESS_RULE, rule['id'], initiator
             )

keystone/tests/protection/v3/test_access_rules.py

@@ -152,6 +152,152 @@ class _ProjectUsersTests:
                 expected_status_code=http.client.FORBIDDEN,
             )
 
+    def test_user_cannot_get_other_users_rule_via_own_user_id(self):
+        """BOLA regression: own user_id in path with victim's rule ID.
+
+        The policy check passes (requester matches path user_id) but the
+        rule belongs to a different user and must not be returned.
+        """
+        victim = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
+        victim = PROVIDERS.identity_api.create_user(victim)
+        project = unit.new_project_ref(
+            domain_id=CONF.identity.default_domain_id
+        )
+        project = PROVIDERS.resource_api.create_project(project['id'], project)
+        PROVIDERS.assignment_api.create_grant(
+            self.bootstrapper.member_role_id,
+            user_id=victim['id'],
+            project_id=project['id'],
+        )
+        victim_rule_id = uuid.uuid4().hex
+        app_cred = {
+            'id': uuid.uuid4().hex,
+            'name': uuid.uuid4().hex,
+            'user_id': victim['id'],
+            'project_id': project['id'],
+            'secret': uuid.uuid4().hex,
+            'access_rules': [
+                {
+                    'id': victim_rule_id,
+                    'service': uuid.uuid4().hex,
+                    'path': uuid.uuid4().hex,
+                    'method': uuid.uuid4().hex[16:],
+                }
+            ],
+        }
+        PROVIDERS.application_credential_api.create_application_credential(
+            app_cred
+        )
+        with self.test_client() as c:
+            # Attacker uses their OWN user_id in the path but the victim's
+            # access_rule_id — the policy check passes, ownership must not.
+            path = f'/v3/users/{self.user_id}/access_rules/{victim_rule_id}'
+            c.get(
+                path,
+                headers=self.headers,
+                expected_status_code=http.client.NOT_FOUND,
+            )
+
+    def test_user_cannot_delete_other_users_rule_via_own_user_id(self):
+        """BOLA regression: own user_id in path with victim's rule ID.
+
+        The policy check passes (requester matches path user_id) but the
+        rule belongs to a different user and must not be deleted.
+        """
+        victim = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
+        victim = PROVIDERS.identity_api.create_user(victim)
+        project = unit.new_project_ref(
+            domain_id=CONF.identity.default_domain_id
+        )
+        project = PROVIDERS.resource_api.create_project(project['id'], project)
+        PROVIDERS.assignment_api.create_grant(
+            self.bootstrapper.member_role_id,
+            user_id=victim['id'],
+            project_id=project['id'],
+        )
+        victim_rule_id = uuid.uuid4().hex
+        victim_app_cred = {
+            'id': uuid.uuid4().hex,
+            'name': uuid.uuid4().hex,
+            'user_id': victim['id'],
+            'project_id': project['id'],
+            'secret': uuid.uuid4().hex,
+            'access_rules': [
+                {
+                    'id': victim_rule_id,
+                    'service': uuid.uuid4().hex,
+                    'path': uuid.uuid4().hex,
+                    'method': uuid.uuid4().hex[16:],
+                }
+            ],
+        }
+        PROVIDERS.application_credential_api.create_application_credential(
+            victim_app_cred
+        )
+        PROVIDERS.application_credential_api.delete_application_credential(
+            victim_app_cred['id']
+        )
+        with self.test_client() as c:
+            path = f'/v3/users/{self.user_id}/access_rules/{victim_rule_id}'
+            c.delete(
+                path,
+                headers=self.headers,
+                expected_status_code=http.client.NOT_FOUND,
+            )
+
+    def test_user_cannot_reuse_other_users_access_rule_id(self):
+        """Cross-user reuse regression: referencing another user's rule by ID.
+
+        When creating an application credential with an existing access_rule
+        ID, the rule must belong to the creating user.
+        """
+        victim = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
+        victim = PROVIDERS.identity_api.create_user(victim)
+        victim_project = unit.new_project_ref(
+            domain_id=CONF.identity.default_domain_id
+        )
+        victim_project = PROVIDERS.resource_api.create_project(
+            victim_project['id'], victim_project
+        )
+        PROVIDERS.assignment_api.create_grant(
+            self.bootstrapper.member_role_id,
+            user_id=victim['id'],
+            project_id=victim_project['id'],
+        )
+        victim_rule_id = uuid.uuid4().hex
+        victim_app_cred = {
+            'id': uuid.uuid4().hex,
+            'name': uuid.uuid4().hex,
+            'user_id': victim['id'],
+            'project_id': victim_project['id'],
+            'secret': uuid.uuid4().hex,
+            'access_rules': [
+                {
+                    'id': victim_rule_id,
+                    'service': uuid.uuid4().hex,
+                    'path': uuid.uuid4().hex,
+                    'method': uuid.uuid4().hex[16:],
+                }
+            ],
+        }
+        PROVIDERS.application_credential_api.create_application_credential(
+            victim_app_cred
+        )
+        with self.test_client() as c:
+            path = f'/v3/users/{self.user_id}/application_credentials'
+            body = {
+                'application_credential': {
+                    'name': uuid.uuid4().hex,
+                    'access_rules': [{'id': victim_rule_id}],
+                }
+            }
+            c.post(
+                path,
+                json=body,
+                headers=self.headers,
+                expected_status_code=http.client.NOT_FOUND,
+            )
+
     def test_user_cannot_get_own_non_existent_access_rule_not_found(self):
         with self.test_client() as c:
             c.get(