credentials: reject delegated tokens, restrict PATCH to blob only

Delegated tokens (application_credential, OAuth1, trust, and
ec2credential) could read, create, modify, or delete credentials
via /v3/credentials and the OS-EC2 compat endpoint. The existing
guards only compared project_id and never treated ec2credential
as delegated.

Replace those guards with a primary-auth-only gate: any token
whose methods aren't entirely primary auth methods, or that
carries a trust_id, is rejected from GET/POST/PATCH/DELETE on
those endpoints. This also removes trust/OAuth1 EC2 credential
creation, never a deliberate feature (bug 1242597 /
OSSA-2013-032).

PATCH now only accepts `blob`; `type`, `user_id`, and
`project_id` are immutable after creation. Delete and recreate
the credential to change them. This also closes bug 2158931: the
old guard only validated the stored (pre-image) project_id on
PATCH, never the new project_id in the request body, so a
delegated token scoped to project A could rewrite project_id to
any project B on a credential it already owned in A. Banning
project_id from PATCH entirely removes that write path for every
caller, not just delegated tokens.

Narrow the existing, deprecated
allow_insecure_admin_trust_cross_project_credentials_access
option (LP#2150089) to ec2-type credentials only, matching its
documented use case: an admin-role trust used to list/sync EC2
credentials.

An empty token.methods list is treated as delegated.

Uses keystone.api._shared.delegation from the LP#2153453 change.

Closes-Bug: #2159643
Closes-Bug: #2158931
Depends-On: https://review.opendev.org/c/openstack/tempest/+/1002295
Depends-On: https://review.opendev.org/c/openstack/keystone-tempest-plugin/+/1002296
Change-Id: I0d0c0e1a1d0728655e559fbe12ed3fd117d17d33
Assisted-by: Cursor Grok 4.6 <noreply@cursor.com>
Signed-off-by: Grzegorz Grasza <xek@redhat.com>

api-ref/source/v3/credentials.inc

@@ -220,6 +220,10 @@ Update credential
 
 Updates a credential.
 
+Only ``blob`` may be updated. ``project_id``, ``type``, and ``user_id`` are
+immutable after creation; to change any of those, delete the credential and
+create a new one.
+
 Relationship: ``https://docs.openstack.org/api/openstack-identity/3/rel/credential``
 
 Request
@@ -232,10 +236,7 @@ Parameters
 
    - credential_id: credential_id_path
    - credential: credential
-   - project_id: project_id
-   - type: credential_type_not_required
    - blob: credential_blob_not_required
-   - user_id: credential_user_id_not_required
 
 Example
 ~~~~~~~

api-ref/source/v3/samples/admin/credential-update-request.json

@@ -1,8 +1,5 @@
 {
     "credential": {
-        "blob": "{\"access\":\"181920\",\"secret\":\"secretKey\"}",
-        "project_id": "731fc6f265cd486d900f16e84c5cb594",
-        "type": "ec2",
-        "user_id": "bb5476fd12884539b41d5a88f838d773"
+        "blob": "{\"access\":\"181920\",\"secret\":\"secretKey\"}"
     }
 }

keystone/api/credentials.py

@@ -18,6 +18,7 @@ import http.client
 import flask
 from oslo_serialization import jsonutils
 
+from keystone.api._shared import delegation
 from keystone.api import validation
 from keystone.common import json_home
 from keystone.common import provider_api
@@ -33,57 +34,37 @@ PROVIDERS = provider_api.ProviderAPIs
 ENFORCER = rbac_enforcer.RBACEnforcer
 
 
-def _check_unrestricted_application_credential(token):
-    if 'application_credential' in token.methods:
-        if not token.application_credential['unrestricted']:
-            action = _(
-                "Using method 'application_credential' is not "
-                "allowed for managing additional credentials."
-            )
-            raise exception.ForbiddenAction(action=action)
+def _require_primary_auth(token, oslo_context, credential_type):
+    """Reject delegated tokens from touching another credential at all.
 
+    Trust delegation is a scope, not a method -- a trust-scoped token's
+    methods list reflects whatever primary method was used to obtain it,
+    so it's checked separately via oslo_context.trust_id. This also drops
+    trust-scoped EC2 credential creation, never a deliberate feature (bug
+    1242597 / OSSA-2013-032).
 
-def _check_credential_project_scope(token, oslo_context, credential):
-    """Enforce project boundary for delegated tokens.
-
-    Non-delegated tokens (password, totp, etc.) are not restricted here --
-    an admin with a regular token can legitimately manage credentials across
-    projects. Delegated tokens (trusts, application credentials, OAuth1) are
-    always bound to a single project at delegation time; only credentials
-    whose project_id exactly matches the token's project scope are in bounds.
-
-    Credentials with project_id=None (e.g. TOTP/MFA bindings) are treated as
-    out-of-scope for any delegated token: they are user-level secrets with no
-    project anchor, and a delegated token should never be able to enumerate,
-    read, or mutate them -- doing so would allow a stolen delegation token to
-    exfiltrate or destroy a user's MFA binding.
+    allow_insecure_admin_trust_cross_project_credentials_access exempts
+    admin-role delegated tokens, but only for ec2-type credentials,
+    matching its one use case (LP#2150089 comment #57). See LP#2159643.
     """
     trust_id = getattr(oslo_context, 'trust_id', None)
-    app_cred_id = getattr(token, 'application_credential_id', None)
-    access_token_id = getattr(token, 'access_token_id', None)
-
-    if not (trust_id or app_cred_id or access_token_id):
+    if not (trust_id or delegation.is_delegated_method(token)):
         return
 
-    token_project_id = oslo_context.project_id
-    cred_project_id = credential.get('project_id')
-
-    insecure_cross_project = CONF.security_compliance.allow_insecure_admin_trust_cross_project_credentials_access
-    if cred_project_id != token_project_id:
-        if insecure_cross_project:
-            # When insecure cross-project access is enabled, still restrict to
-            # admin-role delegated tokens only. See LP#2150089.
-            try:
-                ENFORCER.enforce_call(action='admin_required')
-                return
-            except exception.ForbiddenAction:
-                pass
-        raise exception.ForbiddenAction(
-            action=_(
-                'Credential project does not match the '
-                'project scope of the delegated token'
-            )
-        )
+    insecure_cross_project = getattr(
+        CONF.security_compliance,
+        'allow_insecure_admin_trust_cross_project_credentials_access',
+    )
+    if insecure_cross_project and (credential_type or '').lower() == 'ec2':
+        # Still restrict to admin-role delegated tokens only. See LP#2150089.
+        try:
+            ENFORCER.enforce_call(action='admin_required')
+            return
+        except exception.ForbiddenAction:
+            pass
+    raise exception.ForbiddenAction(
+        action=_('Delegated tokens cannot access credentials directly')
+    )
 
 
 def _build_target_enforcement():
@@ -220,7 +201,9 @@ class CredentialsResource(ks_flask.ResourceBase):
                     action='identity:get_credential',
                     target_attr={'credential': ref},
                 )
-                _check_credential_project_scope(token, self.oslo_context, ref)
+                _require_primary_auth(
+                    token, self.oslo_context, ref.get('type')
+                )
                 filtered_refs.append(ref)
             except (exception.Forbidden, exception.ForbiddenAction):
                 pass
@@ -242,12 +225,10 @@ class CredentialsResource(ks_flask.ResourceBase):
             action='identity:create_credential', target_attr=target
         )
         token = self.auth_context['token']
-        if credential.get('type', '').lower() == 'ec2':
-            _check_unrestricted_application_credential(token)
         trust_id = getattr(self.oslo_context, 'trust_id', None)
         app_cred_id = getattr(token, 'application_credential_id', None)
         access_token_id = getattr(token, 'access_token_id', None)
-        _check_credential_project_scope(token, self.oslo_context, credential)
+        _require_primary_auth(token, self.oslo_context, credential.get('type'))
         ref = self._assign_unique_id(
             self._normalize_dict(credential),
             trust_id=trust_id,
@@ -293,8 +274,10 @@ class CredentialResource(ks_flask.ResourceBase):
             build_target=_build_target_enforcement,
         )
         credential = PROVIDERS.credential_api.get_credential(credential_id)
-        _check_credential_project_scope(
-            self.auth_context['token'], self.oslo_context, credential
+        _require_primary_auth(
+            self.auth_context['token'],
+            self.oslo_context,
+            credential.get('type'),
         )
         return self.wrap_member(_blob_to_json(credential))
 
@@ -310,8 +293,8 @@ class CredentialResource(ks_flask.ResourceBase):
             build_target=_build_target_enforcement,
         )
         current = PROVIDERS.credential_api.get_credential(credential_id)
-        _check_credential_project_scope(
-            self.auth_context['token'], self.oslo_context, current
+        _require_primary_auth(
+            self.auth_context['token'], self.oslo_context, current.get('type')
         )
         credential = self.request_body_json.get('credential', {})
         self._validate_blob_update_keys(current.copy(), credential.copy())
@@ -336,8 +319,10 @@ class CredentialResource(ks_flask.ResourceBase):
             build_target=_build_target_enforcement,
         )
         credential = PROVIDERS.credential_api.get_credential(credential_id)
-        _check_credential_project_scope(
-            self.auth_context['token'], self.oslo_context, credential
+        _require_primary_auth(
+            self.auth_context['token'],
+            self.oslo_context,
+            credential.get('type'),
         )
 
         return (

keystone/api/users.py

@@ -95,28 +95,17 @@ def _check_unrestricted_application_credential(token):
             raise ks_exception.ForbiddenAction(action=action)
 
 
-def _is_delegated_token(oslo_context, token):
-    """Return True if the token is any form of delegation."""
-    trust_id = getattr(oslo_context, 'trust_id', None)
-    app_cred_id = getattr(token, 'application_credential_id', None)
-    access_token_id = getattr(token, 'access_token_id', None)
-    return bool(trust_id or app_cred_id or access_token_id)
+def _require_primary_auth_for_ec2(oslo_context, token):
+    """Reject delegated tokens from the OS-EC2 compat endpoints entirely.
 
-
-def _check_delegation_for_ec2(oslo_context, token, project_id):
-    """For delegated tokens raise unless project_id exactly matches scope.
-
-    Credentials with project_id=None (user-scoped secrets such as TOTP) are
-    treated as out-of-scope: a delegated token must not read or modify them.
+    Matches keystone.api.credentials._require_primary_auth, without its
+    ec2-only escape hatch: no legitimate use case needs delegated access to
+    these endpoints. See LP#2159643.
     """
-    if not _is_delegated_token(oslo_context, token):
-        return
-    if project_id != oslo_context.project_id:
+    trust_id = getattr(oslo_context, 'trust_id', None)
+    if trust_id or delegation.is_delegated_method(token):
         raise ks_exception.ForbiddenAction(
-            action=_(
-                'EC2 credential project does not match the '
-                'project scope of the delegated token'
-            )
+            action=_('Delegated tokens cannot access credentials directly')
         )
 
 
@@ -470,20 +459,15 @@ class UserOSEC2CredentialsResourceListCreate(_UserOSEC2CredBaseResource):
         GET/HEAD /v3/users/{user_id}/credentials/OS-EC2
         """
         ENFORCER.enforce_call(action='identity:ec2_list_credentials')
+        token = self.auth_context['token']
+        _require_primary_auth_for_ec2(self.oslo_context, token)
         PROVIDERS.identity_api.get_user(user_id)
         credential_refs = PROVIDERS.credential_api.list_credentials_for_user(
             user_id, type=CRED_TYPE_EC2
         )
-        token = self.auth_context['token']
-        collection_refs = []
-        for cred in credential_refs:
-            try:
-                _check_delegation_for_ec2(
-                    self.oslo_context, token, cred.get('project_id')
-                )
-            except (ks_exception.Forbidden, ks_exception.ForbiddenAction):
-                continue
-            collection_refs.append(_convert_v3_to_ec2_credential(cred))
+        collection_refs = [
+            _convert_v3_to_ec2_credential(cred) for cred in credential_refs
+        ]
         return self.wrap_collection(collection_refs)
 
     def post(self, user_id):
@@ -497,11 +481,10 @@ class UserOSEC2CredentialsResourceListCreate(_UserOSEC2CredBaseResource):
             action='identity:ec2_create_credential', target_attr=target
         )
         token = self.auth_context['token']
-        _check_unrestricted_application_credential(token)
         PROVIDERS.identity_api.get_user(user_id)
         tenant_id = self.request_body_json.get('tenant_id')
         PROVIDERS.resource_api.get_project(tenant_id)
-        _check_delegation_for_ec2(self.oslo_context, token, tenant_id)
+        _require_primary_auth_for_ec2(self.oslo_context, token)
         blob = {
             'access': uuid.uuid4().hex,
             'secret': uuid.uuid4().hex,
@@ -542,10 +525,8 @@ class UserOSEC2CredentialsResourceGetDelete(_UserOSEC2CredBaseResource):
         PROVIDERS.identity_api.get_user(user_id)
         ec2_cred_id = utils.hash_access_key(credential_id)
         cred = self._get_raw_cred(ec2_cred_id)
-        _check_delegation_for_ec2(
-            self.oslo_context,
-            self.auth_context['token'],
-            cred.get('project_id'),
+        _require_primary_auth_for_ec2(
+            self.oslo_context, self.auth_context['token']
         )
         return self.wrap_member(_convert_v3_to_ec2_credential(cred))
 
@@ -560,11 +541,9 @@ class UserOSEC2CredentialsResourceGetDelete(_UserOSEC2CredBaseResource):
         )
         PROVIDERS.identity_api.get_user(user_id)
         ec2_cred_id = utils.hash_access_key(credential_id)
-        cred = self._get_raw_cred(ec2_cred_id)
-        _check_delegation_for_ec2(
-            self.oslo_context,
-            self.auth_context['token'],
-            cred.get('project_id'),
+        self._get_raw_cred(ec2_cred_id)
+        _require_primary_auth_for_ec2(
+            self.oslo_context, self.auth_context['token']
         )
         PROVIDERS.credential_api.delete_credential(ec2_cred_id)
         return None, http.client.NO_CONTENT

keystone/conf/security_compliance.py

@@ -238,18 +238,17 @@ service account credentials instead, then remove this option.
     deprecated_since='2026.1',
     help=utils.fmt(
         """
-INSECURE: When enabled, admin-role delegated tokens (trusts, application
-credentials, OAuth1 access tokens) are allowed to access credentials outside
-their project scope. By default (False), delegated tokens can only access
-credentials whose project_id matches the token's project scope, preventing
-cross-project lateral movement via a compromised delegation token.
+INSECURE: By default (False), delegated tokens (trusts, application
+credentials, OAuth1 access tokens, EC2 credentials) cannot access
+/v3/credentials or the OS-EC2 compat endpoints at all -- see LP#2159643.
+When enabled, admin-role delegated tokens are exempted for ec2-type
+credentials only, including across projects; every other type stays
+blocked regardless of role.
 
-Enable this only if you have automated workflows (e.g. Mistral cron triggers)
-that use admin-role trusts to access credentials across multiple projects and
-cannot be migrated to use non-delegated service account credentials. Enabling
-this option weakens the isolation guarantee provided by the delegation boundary
-fix for LP#2150089. This option is deprecated and will be removed in a future
-release.
+Enable this only for automated workflows (e.g. Mistral cron triggers) that
+use admin-role trusts to list/sync EC2 credentials across projects and
+cannot migrate to non-delegated service account credentials. Weakens the
+isolation from LP#2150089 and LP#2159643. Deprecated for removal.
 """
     ),
 )

keystone/credential/core.py

@@ -143,23 +143,19 @@ class Manager(manager.Manager):
         notifications.Audit.created(self._CRED, credential_id, initiator)
         return ref
 
-    def _validate_credential_update(self, credential_id, credential):
-        # ec2 credentials require a "project_id" to be functional. Before we
-        # update, check the case where a non-ec2 credential changes its type
-        # to be "ec2", but has no associated "project_id", either in the
-        # request or already set in the database
-        if credential.get('type', '').lower() == 'ec2' and not credential.get(
-            'project_id'
-        ):
-            existing_cred = self.get_credential(credential_id)
-            if not existing_cred['project_id']:
-                raise exception.ValidationError(
-                    attribute='project_id', target='credential'
-                )
-
     def update_credential(self, credential_id, credential):
-        """Update an existing credential."""
-        self._validate_credential_update(credential_id, credential)
+        """Update an existing credential.
+
+        Only `blob` is accepted here: the API layer's PATCH schema enforces
+        this (see LP#2159643 -- retyping, rescoping, or reassigning a
+        credential in place was a repeated source of security issues), so
+        `type`, `user_id`, and `project_id` are never present in
+        `credential`. There is currently no other caller of this method
+        that passes those fields either (keystone-manage credential_migrate
+        only ever passes `blob`). Do not reintroduce type/project_id/
+        user_id-specific validation here without also reintroducing a way
+        for a caller to actually change them.
+        """
         if 'blob' in credential:
             credential_copy = self._encrypt_credential(credential)
         else:

keystone/credential/schema.py

@@ -127,15 +127,19 @@ create_request_body: dict[str, Any] = {
 }
 
 # Request body of the `PATCH /credentials/{credential_id}` operation
+#
+# Only `blob` may be updated in place. `id`, `type`, `user_id`, and
+# `project_id` are immutable after creation. To change any of those,
+# delete the credential and create a new one.
 update_request_body: dict[str, Any] = {
     "type": "object",
     "description": "A credential object.",
     "properties": {
         "credential": {
             "type": "object",
-            "properties": _credential_properties,
-            "additionalProperties": True,
-            "minProperties": 1,
+            "properties": {"blob": _credential_properties["blob"]},
+            "additionalProperties": False,
+            "required": ["blob"],
         }
     },
     "required": ["credential"],

keystone/tests/unit/test_v3_credential.py

@@ -23,7 +23,9 @@ from keystoneclient.contrib.ec2 import utils as ec2_utils
 from oslo_db import exception as oslo_db_exception
 from testtools import matchers
 
+from keystone.api import credentials as credentials_api
 from keystone.api import ec2tokens
+from keystone.api import users as users_api
 from keystone.common import provider_api
 from keystone.common import utils
 from keystone.credential.providers import fernet as credential_fernet
@@ -276,52 +278,23 @@ class CredentialTestCase(CredentialBaseTestCase):
 
     def test_update_credential(self):
         """Call ``PATCH /credentials/{credential_id}``."""
-        ref = unit.new_credential_ref(
-            user_id=self.user['id'], project_id=self.project_id
-        )
-        del ref['id']
+        update_ref = {'blob': uuid.uuid4().hex}
         r = self.patch(
             '/credentials/{credential_id}'.format(
                 credential_id=self.credential['id']
             ),
-            body={'credential': ref},
-        )
-        self.assertValidCredentialResponse(r, ref)
-
-    def test_update_credential_to_ec2_type(self):
-        """Call ``PATCH /credentials/{credential_id}``."""
-        # Create a credential without providing a project_id
-        ref = unit.new_credential_ref(user_id=self.user['id'])
-        r = self.post('/credentials', body={'credential': ref})
-        self.assertValidCredentialResponse(r, ref)
-        credential_id = r.result.get('credential')['id']
-
-        # Updating the credential to ec2 requires a project_id
-        update_ref = {'type': 'ec2', 'project_id': self.project_id}
-        self.patch(
-            f'/credentials/{credential_id}', body={'credential': update_ref}
-        )
-
-    def test_update_credential_to_ec2_missing_project_id(self):
-        """Call ``PATCH /credentials/{credential_id}``."""
-        # Create a credential without providing a project_id
-        ref = unit.new_credential_ref(user_id=self.user['id'])
-        r = self.post('/credentials', body={'credential': ref})
-        self.assertValidCredentialResponse(r, ref)
-        credential_id = r.result.get('credential')['id']
-
-        # Updating such credential to ec2 type without providing a project_id
-        # will fail
-        update_ref = {'type': 'ec2'}
-        self.patch(
-            f'/credentials/{credential_id}',
             body={'credential': update_ref},
-            expected_status=http.client.BAD_REQUEST,
         )
+        expected = dict(self.credential, **update_ref)
+        self.assertValidCredentialResponse(r, expected)
 
-    def test_update_credential_to_ec2_with_previously_set_project_id(self):
-        """Call ``PATCH /credentials/{credential_id}``."""
-        # Create a credential providing a project_id
+    def test_update_credential_rejects_non_blob_fields(self):
+        """PATCH only accepts `blob` (LP#2159643).
+
+        `type`, `project_id`, and `user_id` are all immutable after
+        creation. The only way to change any of those is to delete the
+        credential and create a new one.
+        """
         ref = unit.new_credential_ref(
             user_id=self.user['id'], project_id=self.project_id
         )
@@ -329,12 +302,20 @@ class CredentialTestCase(CredentialBaseTestCase):
         self.assertValidCredentialResponse(r, ref)
         credential_id = r.result.get('credential')['id']
 
-        # Since the created credential above already has a project_id, the
-        # update request will not fail
-        update_ref = {'type': 'ec2'}
-        self.patch(
-            f'/credentials/{credential_id}', body={'credential': update_ref}
-        )
+        for update_ref in (
+            {'type': 'ec2', 'project_id': self.project_id},
+            {'project_id': self.project_id},
+            {'user_id': uuid.uuid4().hex},
+        ):
+            self.patch(
+                f'/credentials/{credential_id}',
+                body={'credential': update_ref},
+                expected_status=http.client.BAD_REQUEST,
+            )
+        stored = PROVIDERS.credential_api.get_credential(credential_id)
+        self.assertEqual(ref['type'], stored['type'])
+        self.assertEqual(self.project_id, stored['project_id'])
+        self.assertEqual(self.user['id'], stored['user_id'])
 
     def test_update_credential_non_owner(self):
         """Call ``PATCH /credentials/{credential_id}``."""
@@ -363,11 +344,12 @@ class CredentialTestCase(CredentialBaseTestCase):
         self.assertValidCredentialResponse(r, ref)
         credential_id = r.result.get('credential')['id']
 
-        # Cannot change the credential to be owned by another user
+        # Cannot change the credential to be owned by another user -- PATCH
+        # no longer accepts user_id/project_id at all. See LP#2159643.
         update_ref = {'user_id': self.user_id, 'project_id': self.project_id}
         self.patch(
             f'/credentials/{credential_id}',
-            expected_status=403,
+            expected_status=http.client.BAD_REQUEST,
             auth=auth,
             body={'credential': update_ref},
         )
@@ -480,6 +462,21 @@ class CredentialTestCase(CredentialBaseTestCase):
             expected_status=http.client.BAD_REQUEST,
         )
 
+    def test_cannot_change_credential_type_via_patch(self):
+        """PATCH rejects a type change; only blob is mutable."""
+        ref = unit.new_credential_ref(
+            user_id=self.user_id, project_id=self.project_id
+        )
+        r = self.post('/credentials', body={'credential': ref})
+        credential_id = r.result['credential']['id']
+        self.patch(
+            f'/credentials/{credential_id}',
+            body={'credential': {'type': 'ec2'}},
+            expected_status=http.client.BAD_REQUEST,
+        )
+        stored = PROVIDERS.credential_api.get_credential(credential_id)
+        self.assertEqual(ref['type'], stored['type'])
+
     def test_delete_credential(self):
         """Call ``DELETE /credentials/{credential_id}``."""
         self.delete(
@@ -658,12 +655,19 @@ class TestCredentialTrustScoped(CredentialBaseTestCase):
         super().config_overrides()
         self.config_fixture.config(group='trust')
 
-    def test_trust_scoped_ec2_credential(self):
-        """Test creating trust scoped ec2 credential.
+    def test_trust_scoped_token_cannot_create_ec2_credential(self):
+        """A trust-scoped token cannot create any credential (LP#2159643).
 
-        Call ``POST /credentials``.
+        Trust-scoped tokens creating EC2 credentials was never a
+        deliberately-designed feature: it dates back to bug 1242597 /
+        OSSA-2013-032, where trust tokens creating EC2 credentials already
+        existed generically and the ec2tokens exchange did not preserve
+        trust scoping, allowing privilege escalation to the trustor's
+        full role set. That bug's fix embedded trust_id in the blob
+        rather than removing the underlying capability. Delegated tokens
+        (trust included) are now rejected from /v3/credentials outright,
+        which removes this path rather than carving it out.
         """
-        # Create the trust
         ref = unit.new_trust_ref(
             trustor_user_id=self.user_id,
             trustee_user_id=self.trustee_user_id,
@@ -676,7 +680,6 @@ class TestCredentialTrustScoped(CredentialBaseTestCase):
         r = self.post('/OS-TRUST/trusts', body={'trust': ref})
         trust = self.assertValidTrustResponse(r)
 
-        # Get a trust scoped token
         auth_data = self.build_authentication_request(
             user_id=self.trustee_user['id'],
             password=self.trustee_user['password'],
@@ -684,53 +687,16 @@ class TestCredentialTrustScoped(CredentialBaseTestCase):
         )
         r = self.v3_create_token(auth_data)
         self.assertValidProjectScopedTokenResponse(r, self.user)
-        trust_id = r.result['token']['OS-TRUST:trust']['id']
         token_id = r.headers.get('X-Subject-Token')
 
-        # Create the credential with the trust scoped token
-        blob, ref = unit.new_ec2_credential(
+        _, ref = unit.new_ec2_credential(
             user_id=self.user_id, project_id=self.project_id
         )
-        r = self.post('/credentials', body={'credential': ref}, token=token_id)
-
-        # We expect the response blob to contain the trust_id
-        ret_ref = ref.copy()
-        ret_blob = blob.copy()
-        ret_blob['trust_id'] = trust_id
-        ret_ref['blob'] = json.dumps(ret_blob)
-        self.assertValidCredentialResponse(r, ref=ret_ref)
-
-        # Assert credential id is same as hash of access key id for
-        # ec2 credentials
-        access = blob['access'].encode('utf-8')
-        self.assertEqual(
-            hashlib.sha256(access).hexdigest(), r.result['credential']['id']
-        )
-
-        # Create a role assignment to ensure that it is ignored and only the
-        # trust-delegated roles are used
-        role = unit.new_role_ref(name='reader')
-        role_id = role['id']
-        PROVIDERS.role_api.create_role(role_id, role)
-        PROVIDERS.assignment_api.add_role_to_user_and_project(
-            self.user_id, self.project_id, role_id
-        )
-
-        ret_blob = json.loads(r.result['credential']['blob'])
-        ec2token = self._test_get_token(
-            access=ret_blob['access'], secret=ret_blob['secret']
-        )
-        ec2_roles = [role['id'] for role in ec2token['roles']]
-        self.assertIn(self.role_id, ec2_roles)
-        self.assertNotIn(role_id, ec2_roles)
-
-        # Create second ec2 credential with the same access key id and check
-        # for conflict.
         self.post(
             '/credentials',
             body={'credential': ref},
             token=token_id,
-            expected_status=http.client.CONFLICT,
+            expected_status=http.client.FORBIDDEN,
         )
 
     def _get_trust_token(self):
@@ -752,6 +718,38 @@ class TestCredentialTrustScoped(CredentialBaseTestCase):
         r = self.v3_create_token(auth_data)
         return r.headers.get('X-Subject-Token')
 
+    def test_trust_token_cannot_rescope_credential_project(self):
+        """A credential's project_id can never be moved via PATCH.
+
+        Originally, the project-boundary check only validated the stored
+        (pre-image) project_id, not the requested one, so a delegated
+        token could PATCH an in-scope credential to move it to any other
+        project. PATCH no longer accepts project_id at all, for any
+        token, which closes this more generally. See LP#2159643.
+        """
+        other_project = unit.new_project_ref(domain_id=self.domain_id)
+        other_project = PROVIDERS.resource_api.create_project(
+            other_project['id'], other_project
+        )
+        PROVIDERS.assignment_api.add_role_to_user_and_project(
+            self.user_id, other_project['id'], self.role_id
+        )
+        ref = unit.new_credential_ref(
+            user_id=self.user_id, project_id=self.project_id
+        )
+        r = self.post('/credentials', body={'credential': ref})
+        cred_id = r.result['credential']['id']
+
+        trust_token = self._get_trust_token()
+        self.patch(
+            f'/credentials/{cred_id}',
+            body={'credential': {'project_id': other_project['id']}},
+            token=trust_token,
+            expected_status=http.client.BAD_REQUEST,
+        )
+        stored = PROVIDERS.credential_api.get_credential(cred_id)
+        self.assertEqual(self.project_id, stored['project_id'])
+
     def test_trust_token_cannot_list_totp_credentials(self):
         """Trust-scoped token must not see TOTP/MFA credentials (project_id=None).
 
@@ -790,6 +788,89 @@ class TestCredentialTrustScoped(CredentialBaseTestCase):
             expected_status=http.client.FORBIDDEN,
         )
 
+    def test_escape_hatch_allows_admin_trust_token_to_read_ec2_credential(
+        self,
+    ):
+        """The escape hatch's one use case still works: admin trust + ec2."""
+        self.config_fixture.config(
+            group='security_compliance',
+            allow_insecure_admin_trust_cross_project_credentials_access=True,
+        )
+        blob, ref = unit.new_ec2_credential(
+            user_id=self.user_id, project_id=self.project_id
+        )
+        r = self.post('/credentials', body={'credential': ref})
+        ec2_id = r.result['credential']['id']
+
+        # _get_trust_token() grants self.role_id, which load_sample_data()
+        # creates as the "admin" role.
+        trust_token = self._get_trust_token()
+        r = self.get(f'/credentials/{ec2_id}', token=trust_token)
+        self.assertEqual(
+            blob['access'],
+            json.loads(r.result['credential']['blob'])['access'],
+        )
+
+    def test_escape_hatch_still_blocks_non_ec2_credential_regardless_of_role(
+        self,
+    ):
+        """The escape hatch never exempts non-ec2 credentials."""
+        self.config_fixture.config(
+            group='security_compliance',
+            allow_insecure_admin_trust_cross_project_credentials_access=True,
+        )
+        ref = unit.new_credential_ref(user_id=self.user_id)
+        r = self.post('/credentials', body={'credential': ref})
+        credential_id = r.result['credential']['id']
+
+        trust_token = self._get_trust_token()
+        self.get(
+            f'/credentials/{credential_id}',
+            token=trust_token,
+            expected_status=http.client.FORBIDDEN,
+        )
+
+    def test_escape_hatch_still_blocks_non_admin_trust_token(self):
+        """The insecure escape hatch does not exempt non-admin-role tokens."""
+        self.config_fixture.config(
+            group='security_compliance',
+            allow_insecure_admin_trust_cross_project_credentials_access=True,
+        )
+        _, ref = unit.new_ec2_credential(
+            user_id=self.user_id, project_id=self.project_id
+        )
+        r = self.post('/credentials', body={'credential': ref})
+        ec2_id = r.result['credential']['id']
+
+        member_role = unit.new_role_ref(name='member')
+        PROVIDERS.role_api.create_role(member_role['id'], member_role)
+        PROVIDERS.assignment_api.add_role_to_user_and_project(
+            self.user_id, self.project_id, member_role['id']
+        )
+        trust_ref = unit.new_trust_ref(
+            trustor_user_id=self.user_id,
+            trustee_user_id=self.trustee_user_id,
+            project_id=self.project_id,
+            impersonation=True,
+            role_ids=[member_role['id']],
+        )
+        del trust_ref['id']
+        r = self.post('/OS-TRUST/trusts', body={'trust': trust_ref})
+        trust = r.result['trust']
+        auth_data = self.build_authentication_request(
+            user_id=self.trustee_user['id'],
+            password=self.trustee_user['password'],
+            trust_id=trust['id'],
+        )
+        r = self.v3_create_token(auth_data)
+        trust_token = r.headers.get('X-Subject-Token')
+
+        self.get(
+            f'/credentials/{ec2_id}',
+            token=trust_token,
+            expected_status=http.client.FORBIDDEN,
+        )
+
     def test_trust_token_cannot_update_totp_credential(self):
         """Trust-scoped token must not be able to update a TOTP credential blob."""
         totp_ref = {
@@ -804,7 +885,7 @@ class TestCredentialTrustScoped(CredentialBaseTestCase):
         self.patch(
             f'/credentials/{totp_id}',
             token=trust_token,
-            body={'credential': totp_ref},
+            body={'credential': {'blob': totp_ref['blob']}},
             expected_status=http.client.FORBIDDEN,
         )
 
@@ -893,13 +974,12 @@ class TestCredentialAppCreds(CredentialBaseTestCase):
             )
         )
 
-    def test_app_cred_ec2_credential(self):
-        """Test creating ec2 credential from an application credential.
+    def test_app_cred_cannot_create_ec2_credential(self):
+        """An application credential cannot create any credential (LP#2159643).
 
-        Call ``POST /credentials``.
+        Delegated tokens (application credentials included) are now
+        rejected from /v3/credentials outright.
         """
-        # Create an unrestricted app cred (restricted app creds are
-        # blocked from creating EC2 credentials)
         ref = unit.new_application_credential_ref(roles=[{'id': self.role_id}])
         del ref['id']
         ref['unrestricted'] = True
@@ -909,57 +989,20 @@ class TestCredentialAppCreds(CredentialBaseTestCase):
         )
         app_cred = r.result['application_credential']
 
-        # Get an application credential token
         auth_data = self.build_authentication_request(
             app_cred_id=app_cred['id'], secret=app_cred['secret']
         )
         r = self.v3_create_token(auth_data)
         token_id = r.headers.get('X-Subject-Token')
 
-        # Create the credential with the app cred token
-        blob, ref = unit.new_ec2_credential(
+        _, ref = unit.new_ec2_credential(
             user_id=self.user_id, project_id=self.project_id
         )
-        r = self.post('/credentials', body={'credential': ref}, token=token_id)
-
-        # We expect the response blob to contain the app_cred_id
-        ret_ref = ref.copy()
-        ret_blob = blob.copy()
-        ret_blob['app_cred_id'] = app_cred['id']
-        ret_ref['blob'] = json.dumps(ret_blob)
-        self.assertValidCredentialResponse(r, ref=ret_ref)
-
-        # Assert credential id is same as hash of access key id for
-        # ec2 credentials
-        access = blob['access'].encode('utf-8')
-        self.assertEqual(
-            hashlib.sha256(access).hexdigest(), r.result['credential']['id']
-        )
-
-        # Create a role assignment to ensure that it is ignored and only the
-        # roles in the app cred are used
-        role = unit.new_role_ref(name='reader')
-        role_id = role['id']
-        PROVIDERS.role_api.create_role(role_id, role)
-        PROVIDERS.assignment_api.add_role_to_user_and_project(
-            self.user_id, self.project_id, role_id
-        )
-
-        ret_blob = json.loads(r.result['credential']['blob'])
-        ec2token = self._test_get_token(
-            access=ret_blob['access'], secret=ret_blob['secret']
-        )
-        ec2_roles = [role['id'] for role in ec2token['roles']]
-        self.assertIn(self.role_id, ec2_roles)
-        self.assertNotIn(role_id, ec2_roles)
-
-        # Create second ec2 credential with the same access key id and check
-        # for conflict.
         self.post(
             '/credentials',
             body={'credential': ref},
             token=token_id,
-            expected_status=http.client.CONFLICT,
+            expected_status=http.client.FORBIDDEN,
         )
 
     def _get_app_cred_token(self, unrestricted=False):
@@ -979,6 +1022,64 @@ class TestCredentialAppCreds(CredentialBaseTestCase):
         r = self.v3_create_token(auth_data)
         return r.headers.get('X-Subject-Token')
 
+    def test_app_cred_token_cannot_patch_credential(self):
+        """An application-credential token cannot PATCH /v3/credentials."""
+        token_id = self._get_app_cred_token(unrestricted=True)
+        ref = unit.new_credential_ref(
+            user_id=self.user_id, project_id=self.project_id
+        )
+        r = self.post('/credentials', body={'credential': ref})
+        credential_id = r.result['credential']['id']
+
+        self.patch(
+            f'/credentials/{credential_id}',
+            body={'credential': {'type': 'ec2'}},
+            token=token_id,
+            expected_status=http.client.BAD_REQUEST,
+        )
+        stored = PROVIDERS.credential_api.get_credential(credential_id)
+        self.assertEqual('cert', stored['type'])
+
+        self.patch(
+            f'/credentials/{credential_id}',
+            body={'credential': {'blob': 'rotated-blob'}},
+            token=token_id,
+            expected_status=http.client.FORBIDDEN,
+        )
+        stored = PROVIDERS.credential_api.get_credential(credential_id)
+        self.assertEqual('cert', stored['type'])
+        self.assertEqual(self.project_id, stored['project_id'])
+
+    def test_app_cred_token_cannot_move_credential_cross_project(self):
+        """PATCH must not let a caller rewrite project_id either (LP#2158931).
+
+        The pre-image project check alone wasn't enough: it validated
+        the credential's *current* project, never the *new* project_id
+        in the request body, so a delegated token scoped to project A
+        could rewrite a credential it already owned there to point at
+        an unrelated project B.
+        """
+        app_cred_token = self._get_app_cred_token(unrestricted=True)
+        ref = unit.new_credential_ref(
+            user_id=self.user_id, project_id=self.project_id
+        )
+        r = self.post('/credentials', body={'credential': ref})
+        credential_id = r.result['credential']['id']
+
+        other_project = unit.new_project_ref(domain_id=self.domain_id)
+        PROVIDERS.resource_api.create_project(
+            other_project['id'], other_project
+        )
+
+        self.patch(
+            f'/credentials/{credential_id}',
+            body={'credential': {'project_id': other_project['id']}},
+            token=app_cred_token,
+            expected_status=http.client.BAD_REQUEST,
+        )
+        stored = PROVIDERS.credential_api.get_credential(credential_id)
+        self.assertEqual(self.project_id, stored['project_id'])
+
     def test_restricted_app_cred_cannot_create_ec2_credential(self):
         """Test that a restricted app cred cannot create EC2 credentials.
 
@@ -1143,7 +1244,7 @@ class TestCredentialAppCreds(CredentialBaseTestCase):
         self.patch(
             f'/credentials/{totp_id}',
             token=app_cred_token,
-            body={'credential': totp_ref},
+            body={'credential': {'blob': totp_ref['blob']}},
             expected_status=http.client.FORBIDDEN,
         )
 
@@ -1283,49 +1384,52 @@ class TestCredentialAccessToken(CredentialBaseTestCase):
         content = self.post(url, headers=headers, body=body)
         return access_key, content.headers['X-Subject-Token']
 
-    def test_access_token_ec2_credential(self):
-        """Test creating ec2 credential from an oauth access token.
+    def test_access_token_cannot_create_ec2_credential(self):
+        """An OAuth1 access token cannot create any credential (LP#2159643).
 
-        Call ``POST /credentials``.
+        Same history as the trust-scoped case: this was never a
+        deliberate feature, just a side effect of access-token-derived
+        tokens being otherwise broadly capable. Delegated tokens
+        (OAuth1 access tokens included) are now rejected from
+        /v3/credentials outright.
         """
-        access_key, token_id = self._get_access_token()
+        _, token_id = self._get_access_token()
 
-        # Create the credential with the access token
-        blob, ref = unit.new_ec2_credential(
+        _, ref = unit.new_ec2_credential(
             user_id=self.user_id, project_id=self.project_id
         )
-        r = self.post('/credentials', body={'credential': ref}, token=token_id)
-
-        # We expect the response blob to contain the access_token_id
-        ret_ref = ref.copy()
-        ret_blob = blob.copy()
-        ret_blob['access_token_id'] = access_key.decode('utf-8')
-        ret_ref['blob'] = json.dumps(ret_blob)
-        self.assertValidCredentialResponse(r, ref=ret_ref)
-
-        # Assert credential id is same as hash of access key id for
-        # ec2 credentials
-        access = blob['access'].encode('utf-8')
-        self.assertEqual(
-            hashlib.sha256(access).hexdigest(), r.result['credential']['id']
+        self.post(
+            '/credentials',
+            body={'credential': ref},
+            token=token_id,
+            expected_status=http.client.FORBIDDEN,
         )
 
-        # Create a role assignment to ensure that it is ignored and only the
-        # roles in the access token are used
-        role = unit.new_role_ref(name='reader')
-        role_id = role['id']
-        PROVIDERS.role_api.create_role(role_id, role)
-        PROVIDERS.assignment_api.add_role_to_user_and_project(
-            self.user_id, self.project_id, role_id
+    def test_access_token_cannot_get_credential_via_os_ec2(self):
+        """An OAuth1 token cannot read an EC2 cred via OS-EC2 (LP#2159643)."""
+        uri = f'/users/{self.user_id}/credentials/OS-EC2'
+        ec2_cred = self.post(uri, body={'tenant_id': self.project_id}).result[
+            'credential'
+        ]
+        _, token_id = self._get_access_token()
+        self.get(
+            '/'.join([uri, ec2_cred['access']]),
+            token=token_id,
+            expected_status=http.client.FORBIDDEN,
         )
 
-        ret_blob = json.loads(r.result['credential']['blob'])
-        ec2token = self._test_get_token(
-            access=ret_blob['access'], secret=ret_blob['secret']
+    def test_access_token_cannot_delete_credential_via_os_ec2(self):
+        """An OAuth1 token cannot delete an EC2 cred via OS-EC2 (2159643)."""
+        uri = f'/users/{self.user_id}/credentials/OS-EC2'
+        ec2_cred = self.post(uri, body={'tenant_id': self.project_id}).result[
+            'credential'
+        ]
+        _, token_id = self._get_access_token()
+        cred_uri = '/'.join([uri, ec2_cred['access']])
+        self.delete(
+            cred_uri, token=token_id, expected_status=http.client.FORBIDDEN
         )
-        ec2_roles = [role['id'] for role in ec2token['roles']]
-        self.assertIn(self.role_id, ec2_roles)
-        self.assertNotIn(role_id, ec2_roles)
+        self.get(cred_uri, expected_status=http.client.OK)
 
     def test_ec2_auth_access_token_cross_project_blocked(self):
         """OAuth1 access-token-backed EC2 credential must not auth cross-project.
@@ -1432,6 +1536,72 @@ class TestCredentialEc2(CredentialBaseTestCase):
         self.assertDictEqual(ec2_cred, r.result['credential'])
         self.assertThat(ec2_cred['links']['self'], matchers.EndsWith(uri))
 
+    def _get_ec2_token_via_own_credential(self):
+        """Create an EC2 credential and exchange it for a token."""
+        ec2_cred = self._get_ec2_cred()
+        blob = {'access': ec2_cred['access'], 'secret': ec2_cred['secret']}
+        r = self.post(
+            '/ec2tokens',
+            body={'ec2Credentials': self._get_ec2_sig_ref(blob)},
+            expected_status=http.client.OK,
+        )
+        return ec2_cred, r.headers.get('X-Subject-Token')
+
+    def test_ec2_token_cannot_list_credentials_via_os_ec2(self):
+        """An ec2credential token must not list EC2 creds via OS-EC2.
+
+        The OS-EC2 compat endpoint has its own, separate delegation check
+        from /v3/credentials, and it never recognized ec2credential tokens
+        as delegated either -- so an ec2credential token could list every
+        EC2 credential (secrets included) belonging to the user.
+        """
+        _, ec2_token = self._get_ec2_token_via_own_credential()
+        self.get(
+            self._get_ec2_cred_uri(),
+            token=ec2_token,
+            expected_status=http.client.FORBIDDEN,
+        )
+
+    def test_ec2_token_cannot_get_credential_via_os_ec2(self):
+        """An ec2credential token must not read an EC2 cred via OS-EC2."""
+        ec2_cred, ec2_token = self._get_ec2_token_via_own_credential()
+        uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
+        self.get(uri, token=ec2_token, expected_status=http.client.FORBIDDEN)
+
+    def test_ec2_token_cannot_delete_credential_via_os_ec2(self):
+        """An ec2credential token must not delete an EC2 cred via OS-EC2."""
+        ec2_cred, ec2_token = self._get_ec2_token_via_own_credential()
+        uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
+        self.delete(
+            uri, token=ec2_token, expected_status=http.client.FORBIDDEN
+        )
+        # the credential must still exist afterwards
+        self.get(uri, expected_status=http.client.OK)
+
+    def test_ec2_token_cannot_patch_credential_via_credentials_api(self):
+        """An ec2credential token must not PATCH /v3/credentials."""
+        ref = unit.new_credential_ref(
+            user_id=self.user_id, project_id=self.project_id
+        )
+        r = self.post('/credentials', body={'credential': ref})
+        credential_id = r.result['credential']['id']
+        original_blob = r.result['credential']['blob']
+
+        _, ec2_token = self._get_ec2_token_via_own_credential()
+        self.patch(
+            f'/credentials/{credential_id}',
+            body={'credential': {'blob': 'rotated-blob'}},
+            token=ec2_token,
+            expected_status=http.client.FORBIDDEN,
+        )
+        self.get(
+            f'/credentials/{credential_id}',
+            token=ec2_token,
+            expected_status=http.client.FORBIDDEN,
+        )
+        stored = PROVIDERS.credential_api.get_credential(credential_id)
+        self.assertEqual(original_blob, stored['blob'])
+
     def test_ec2_cannot_get_non_ec2_credential(self):
         access_key = uuid.uuid4().hex
         cred_id = utils.hash_access_key(access_key)
@@ -1518,19 +1688,19 @@ class TestCredentialEc2(CredentialBaseTestCase):
             expected_status=http.client.FORBIDDEN,
         )
 
-    def test_ec2_create_credential_with_unrestricted_app_cred(self):
-        """Test that an unrestricted app cred can create EC2 credentials."""
+    def test_unrestricted_app_cred_cannot_create_ec2_credential(self):
+        """An unrestricted app cred cannot create EC2 creds either (LP#2159643).
+
+        "unrestricted" only ever governed app-cred management, not this.
+        """
         token_id = self._get_app_cred_token(unrestricted=True)
         uri = self._get_ec2_cred_uri()
-        r = self.post(
+        self.post(
             uri,
             body={'tenant_id': self.project_id},
             token=token_id,
-            expected_status=http.client.CREATED,
+            expected_status=http.client.FORBIDDEN,
         )
-        ec2_cred = r.result['credential']
-        self.assertEqual(self.user_id, ec2_cred['user_id'])
-        self.assertEqual(self.project_id, ec2_cred['tenant_id'])
 
     def _get_trust_token(self):
         """Create a trust and return a trust-scoped token for the trustee."""
@@ -1556,78 +1726,181 @@ class TestCredentialEc2(CredentialBaseTestCase):
         r = self.v3_create_token(auth_data)
         return r.headers.get('X-Subject-Token')
 
-    def test_ec2_create_credential_trust_cross_project_blocked(self):
-        """Trust-scoped token cannot create EC2 cred for a different project."""
-        other_project = unit.new_project_ref(domain_id=self.domain_id)
-        other_project = PROVIDERS.resource_api.create_project(
-            other_project['id'], other_project
-        )
+    def test_trust_scoped_token_cannot_create_ec2_credential(self):
+        """A trust-scoped token cannot create an EC2 cred via OS-EC2 (LP#2159643).
+
+        Previously only checked project scope, so same-project access
+        was allowed. Now rejected outright, regardless of project.
+        """
         trust_token = self._get_trust_token()
-        uri = f'/users/{self.user_id}/credentials/OS-EC2'
+        uri = self._get_ec2_cred_uri()
         self.post(
             uri,
-            body={'tenant_id': other_project['id']},
+            body={'tenant_id': self.project_id},
             token=trust_token,
             expected_status=http.client.FORBIDDEN,
         )
 
-    def test_ec2_create_credential_trust_same_project_allowed(self):
-        """Trust-scoped token can create EC2 cred for the trust project."""
-        trust_token = self._get_trust_token()
-        uri = self._get_ec2_cred_uri()
-        r = self.post(
-            uri,
-            body={'tenant_id': self.project_id},
-            token=trust_token,
-            expected_status=http.client.CREATED,
-        )
-        self.assertEqual(self.project_id, r.result['credential']['tenant_id'])
-
-    def test_ec2_get_credential_trust_cross_project_blocked(self):
-        """Trust-scoped token cannot get an EC2 cred from a different project."""
-        other_project = unit.new_project_ref(domain_id=self.domain_id)
-        other_project = PROVIDERS.resource_api.create_project(
-            other_project['id'], other_project
-        )
-        PROVIDERS.assignment_api.add_role_to_user_and_project(
-            self.user_id, other_project['id'], self.role_id
-        )
+    def test_trust_scoped_token_cannot_get_credential_via_os_ec2(self):
+        """A trust-scoped token cannot read an EC2 cred (LP#2159643)."""
         ec2_cred = self._get_ec2_cred()
-        # Change the credential's project to the other project directly
-        PROVIDERS.credential_api.update_credential(
-            next(
-                c['id']
-                for c in PROVIDERS.credential_api.list_credentials_for_user(
-                    self.user_id, type=CRED_TYPE_EC2
-                )
-            ),
-            {'project_id': other_project['id']},
-        )
         trust_token = self._get_trust_token()
         uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
         self.get(uri, token=trust_token, expected_status=http.client.FORBIDDEN)
 
-    def test_ec2_delete_credential_trust_cross_project_blocked(self):
-        """Trust-scoped token cannot delete EC2 cred from a different project."""
-        other_project = unit.new_project_ref(domain_id=self.domain_id)
-        other_project = PROVIDERS.resource_api.create_project(
-            other_project['id'], other_project
-        )
-        PROVIDERS.assignment_api.add_role_to_user_and_project(
-            self.user_id, other_project['id'], self.role_id
-        )
+    def test_trust_scoped_token_cannot_delete_credential_via_os_ec2(self):
+        """A trust-scoped token cannot delete an EC2 cred (LP#2159643)."""
         ec2_cred = self._get_ec2_cred()
-        PROVIDERS.credential_api.update_credential(
-            next(
-                c['id']
-                for c in PROVIDERS.credential_api.list_credentials_for_user(
-                    self.user_id, type=CRED_TYPE_EC2
-                )
-            ),
-            {'project_id': other_project['id']},
-        )
         trust_token = self._get_trust_token()
         uri = '/'.join([self._get_ec2_cred_uri(), ec2_cred['access']])
         self.delete(
             uri, token=trust_token, expected_status=http.client.FORBIDDEN
         )
+        # the credential must still exist afterwards
+        self.get(uri, expected_status=http.client.OK)
+
+
+class TestPrimaryAuthGuards(unit.BaseTestCase):
+    """Unit-level tests for the delegated-token guards (LP#2159643).
+
+    Calls the guards directly with stub tokens, independent of HTTP/
+    middleware. Matters for ec2credential specifically: an unrelated,
+    already-landed fix bans ec2credential-method tokens at the auth
+    middleware layer before these guards ever run, so an HTTP-level test
+    would pass regardless of whether the guards recognize it as delegated.
+    """
+
+    def _token(self, methods):
+        token = mock.Mock()
+        token.methods = methods
+        return token
+
+    def _oslo_context(self, trust_id=None):
+        ctx = mock.Mock()
+        ctx.trust_id = trust_id
+        return ctx
+
+    def test_require_primary_auth_rejects_ec2credential_token(self):
+        self.assertRaises(
+            exception.ForbiddenAction,
+            credentials_api._require_primary_auth,
+            self._token(['ec2credential']),
+            self._oslo_context(),
+            'cert',
+        )
+
+    def test_require_primary_auth_allows_password_token(self):
+        credentials_api._require_primary_auth(
+            self._token(['password']), self._oslo_context(), 'cert'
+        )
+
+    def test_require_primary_auth_allows_custom_method_via_config(self):
+        """A custom auth plugin is not mistaken for a delegated credential.
+
+        Once listed in [auth] additional_primary_auth_methods.
+        """
+        credentials_api.CONF.set_override(
+            'additional_primary_auth_methods', ['sso'], group='auth'
+        )
+        self.addCleanup(
+            credentials_api.CONF.clear_override,
+            'additional_primary_auth_methods',
+            group='auth',
+        )
+        credentials_api._require_primary_auth(
+            self._token(['sso']), self._oslo_context(), 'cert'
+        )
+
+    def test_require_primary_auth_rejects_unlisted_custom_method(self):
+        self.assertRaises(
+            exception.ForbiddenAction,
+            credentials_api._require_primary_auth,
+            self._token(['sso']),
+            self._oslo_context(),
+            'cert',
+        )
+
+    def test_require_primary_auth_rejects_empty_methods(self):
+        """An empty method list must be treated as delegated, not allowed."""
+        self.assertRaises(
+            exception.ForbiddenAction,
+            credentials_api._require_primary_auth,
+            self._token([]),
+            self._oslo_context(),
+            'cert',
+        )
+
+    def test_require_primary_auth_rejects_empty_methods_with_trust_scope(self):
+        """Same, for a token that is additionally trust-scoped."""
+        self.assertRaises(
+            exception.ForbiddenAction,
+            credentials_api._require_primary_auth,
+            self._token([]),
+            self._oslo_context(trust_id=uuid.uuid4().hex),
+            'cert',
+        )
+
+    def test_require_primary_auth_rejects_trust_scoped_token(self):
+        self.assertRaises(
+            exception.ForbiddenAction,
+            credentials_api._require_primary_auth,
+            self._token(['password']),
+            self._oslo_context(trust_id=uuid.uuid4().hex),
+            'cert',
+        )
+
+    def test_require_primary_auth_escape_hatch_requires_ec2_type(self):
+        """The escape hatch is scoped to ec2-type credentials only."""
+        credentials_api.CONF.set_override(
+            'allow_insecure_admin_trust_cross_project_credentials_access',
+            True,
+            group='security_compliance',
+        )
+        self.addCleanup(
+            credentials_api.CONF.clear_override,
+            'allow_insecure_admin_trust_cross_project_credentials_access',
+            group='security_compliance',
+        )
+        with mock.patch.object(
+            credentials_api.ENFORCER, 'enforce_call', return_value=None
+        ):
+            credentials_api._require_primary_auth(
+                self._token(['ec2credential']), self._oslo_context(), 'ec2'
+            )
+            self.assertRaises(
+                exception.ForbiddenAction,
+                credentials_api._require_primary_auth,
+                self._token(['ec2credential']),
+                self._oslo_context(),
+                'cert',
+            )
+
+    def test_require_primary_auth_for_ec2_rejects_ec2credential_token(self):
+        self.assertRaises(
+            exception.ForbiddenAction,
+            users_api._require_primary_auth_for_ec2,
+            self._oslo_context(),
+            self._token(['ec2credential']),
+        )
+
+    def test_require_primary_auth_for_ec2_allows_password_token(self):
+        users_api._require_primary_auth_for_ec2(
+            self._oslo_context(), self._token(['password'])
+        )
+
+    def test_require_primary_auth_for_ec2_rejects_trust_scoped_token(self):
+        self.assertRaises(
+            exception.ForbiddenAction,
+            users_api._require_primary_auth_for_ec2,
+            self._oslo_context(trust_id=uuid.uuid4().hex),
+            self._token(['password']),
+        )
+
+    def test_require_primary_auth_for_ec2_rejects_empty_methods(self):
+        """Same fernet round-trip gap as _require_primary_auth."""
+        self.assertRaises(
+            exception.ForbiddenAction,
+            users_api._require_primary_auth_for_ec2,
+            self._oslo_context(),
+            self._token([]),
+        )

keystone/tests/unit/test_validation.py

@@ -1188,14 +1188,7 @@ class CredentialValidationTestCase(unit.BaseTestCase):
 
     def test_validate_credential_update_succeeds(self):
         """Test that a credential request is properly validated."""
-        request_to_validate = {
-            'credential': {
-                'blob': 'some string',
-                'project_id': uuid.uuid4().hex,
-                'type': 'ec2',
-                'user_id': uuid.uuid4().hex,
-            }
-        }
+        request_to_validate = {'credential': {'blob': 'some string'}}
         self.update_credential_validator.validate(request_to_validate)
 
     def test_validate_credential_update_without_parameters_fails(self):
@@ -1207,18 +1200,35 @@ class CredentialValidationTestCase(unit.BaseTestCase):
             request_to_validate,
         )
 
-    def test_validate_credential_update_with_extra_parameters_succeeds(self):
-        """Validate credential update with extra parameters."""
+    def test_validate_credential_update_rejects_non_blob_fields(self):
+        """PATCH only accepts `blob` (LP#2159643).
+
+        `type`, `project_id`, and `user_id` are immutable after creation.
+        """
         request_to_validate = {
             'credential': {
                 'blob': 'some string',
-                'extra': False,
                 'project_id': uuid.uuid4().hex,
                 'type': 'ec2',
                 'user_id': uuid.uuid4().hex,
             }
         }
-        self.update_credential_validator.validate(request_to_validate)
+        self.assertRaises(
+            exception.SchemaValidationError,
+            self.update_credential_validator.validate,
+            request_to_validate,
+        )
+
+    def test_validate_credential_update_with_extra_parameters_fails(self):
+        """Validate credential update rejects unknown extra parameters."""
+        request_to_validate = {
+            'credential': {'blob': 'some string', 'extra': False}
+        }
+        self.assertRaises(
+            exception.SchemaValidationError,
+            self.update_credential_validator.validate,
+            request_to_validate,
+        )
 
 
 class RegionValidationTestCase(unit.BaseTestCase):

releasenotes/notes/bug-2159643-45b677fc1b848df4.yaml

@@ -0,0 +1,55 @@
+---
+security:
+  - |
+    [`bug 2159643 <https://launchpad.net/bugs/2159643>`_]
+    Delegated tokens (trusts, application credentials, OAuth1 access
+    tokens, and EC2 credentials, which were not previously recognized
+    as delegated by this check) can no longer access
+    ``/v3/credentials`` or the ``/v3/users/{user_id}/credentials/OS-EC2``
+    compat endpoints at all, superseding the project-scope comparison
+    introduced by `bug 2150089
+    <https://launchpad.net/bugs/2150089>`_. Trust-scoped and OAuth1
+    access-token-scoped tokens creating EC2 credentials is also removed;
+    this was never a deliberate feature (see `bug 1242597
+    <https://launchpad.net/bugs/1242597>`_ / OSSA-2013-032).
+
+    ``PATCH /v3/credentials/{credential_id}`` now only accepts ``blob``.
+    ``type``, ``user_id``, and ``project_id`` are immutable after
+    creation. Delete and re-create the credential to change any of
+    those fields. This also fixes `bug 2158931
+    <https://launchpad.net/bugs/2158931>`_: the previous guard only
+    validated a PATCH request's stored (pre-image) ``project_id``,
+    never the new ``project_id`` in the request body, so a delegated
+    token scoped to one project could rewrite a credential it already
+    owned there to point at a different project.
+
+    This closes the API-level creation path only. It does not
+    retroactively affect ``type=ec2`` credentials that were already
+    bound to a trust or OAuth1 access token before this fix (for
+    example, created directly through the provider API or
+    ``keystone-manage`` rather than through this guarded endpoint). To
+    fully mitigate the risk this change addresses, operators should
+    also audit their existing EC2 credentials for a ``trust_id`` or
+    ``access_token_id`` in their stored ``blob`` and delete and
+    re-create any that have one, so that no EC2 credential remains
+    tied to a delegation that can independently expire, be revoked, or
+    otherwise change state out from under it.
+upgrade:
+  - |
+    [`bug 2159643 <https://launchpad.net/bugs/2159643>`_]
+    ``PATCH /v3/credentials/{credential_id}`` requests that include
+    ``type``, ``user_id``, or ``project_id`` now fail with ``400 Bad
+    Request``. Update any client or automation that PATCHes credentials
+    to send ``blob`` only; delete and re-create the credential for any
+    other field change.
+
+    The existing, opt-in, deprecated
+    ``[security_compliance] allow_insecure_admin_trust_cross_project_credentials_access``
+    option (see `bug 2150089 <https://launchpad.net/bugs/2150089>`_) is
+    now scoped to EC2 credentials only, matching its one documented use
+    case (an admin-role trust driving automated tooling that lists or
+    syncs EC2 credentials with a third-party system). Enabling the
+    option no longer grants access to any credential type other than
+    EC2. Operators relying on this option should migrate the underlying
+    automated workflow away from delegated tokens; it will be removed
+    in a future release.