Synchronise 2024.1 with upstream

devstack/tools/oidc/setup_keycloak_client.py

@@ -1,3 +1,4 @@
+import http
 import os
 import requests
 
@@ -34,6 +35,17 @@ def _admin_auth(self, realm):
         self.session.headers.update(headers)
         return r
 
+    def _get_client_uuid(self, realm, client_id):
+        resp = self.session.get(
+            self.construct_url(realm, 'clients'),
+            params={'clientId': client_id},
+        )
+        resp.raise_for_status()
+        for client in resp.json():
+            if client.get('clientId') == client_id:
+                return client['id']
+        return None
+
     def create_client(self, realm, client_id, client_secret, redirect_uris):
         self._admin_auth(realm)
         data = {
@@ -43,7 +55,49 @@ def create_client(self, realm, client_id, client_secret, redirect_uris):
             'implicitFlowEnabled': True,
             'directAccessGrantsEnabled': True,
         }
-        return self.session.post(self.construct_url(realm, 'clients'), json=data)
+        resp = self.session.post(
+            self.construct_url(realm, 'clients'), json=data
+        )
+        if resp.status_code == http.HTTPStatus.CONFLICT:
+            # Client already exists from a previous run; reuse it.
+            client_uuid = self._get_client_uuid(realm, client_id)
+            if client_uuid is None:
+                resp.raise_for_status()
+        else:
+            resp.raise_for_status()
+            # Keycloak returns 201 with the new client's UUID in the
+            # Location header: .../admin/realms/<realm>/clients/<uuid>
+            client_uuid = resp.headers['Location'].rsplit('/', 1)[-1]
+        # Since Keycloak 26.6.2 (CVE-2026-37979) the OAuth2 token
+        # introspection endpoint requires the introspecting client to
+        # be present in the access token's "aud" claim. Apache's
+        # mod_auth_openidc uses this same client to introspect bearer
+        # tokens it receives from federated users, so we add an
+        # audience protocol mapper that lists the client itself in
+        # "aud". Without this, introspection returns active=false and
+        # mod_auth_openidc rejects the token with HTTP 401.
+        self._add_audience_mapper(realm, client_uuid, client_id)
+        return resp
+
+    def _add_audience_mapper(self, realm, client_uuid, audience_client_id):
+        mapper = {
+            'name': f'{audience_client_id}-audience',
+            'protocol': 'openid-connect',
+            'protocolMapper': 'oidc-audience-mapper',
+            'config': {
+                'included.client.audience': audience_client_id,
+                'access.token.claim': 'true',
+                'id.token.claim': 'false',
+                'introspection.token.claim': 'true',
+            },
+        }
+        path = f'clients/{client_uuid}/protocol-mappers/models'
+        resp = self.session.post(self.construct_url(realm, path), json=mapper)
+        if resp.status_code == http.HTTPStatus.CONFLICT:
+            # Mapper already present from a previous run; nothing to do.
+            return resp
+        resp.raise_for_status()
+        return resp
 
 
 def main():

keystone/api/credentials.py

@@ -216,6 +216,9 @@ def post(self):
         if credential.get('type', '').lower() == 'ec2':
             _check_unrestricted_application_credential(token)
         validation.lazy_validate(schema.credential_create, credential)
+        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)

keystone/api/os_oauth1.py

@@ -292,6 +292,17 @@ def put(self, request_token_id):
             raise exception.Forbidden(
                 _('Cannot authorize a request token with a token issued via '
                   'delegation.'))
+        auth_context = flask.request.environ.get(
+            authorization.AUTH_CONTEXT_ENV, {}
+        )
+        token = auth_context.get('token')
+        if token and 'application_credential' in token.methods:
+            raise exception.Forbidden(
+                _(
+                    'Cannot authorize a request token with a token issued via '
+                    'delegation.'
+                )
+            )
 
         req_token = PROVIDERS.oauth_api.get_request_token(request_token_id)
 

keystone/tests/unit/test_v3_credential.py

@@ -1418,6 +1418,53 @@ def test_ec2_delete_credential(self):
                           PROVIDERS.credential_api.get_credential,
                           cred_from_credential_api[0]['id'])
 
+    def _get_app_cred_token(self, unrestricted=False):
+        """Create an application credential and return a token for it."""
+        ref = unit.new_application_credential_ref(roles=[{'id': self.role_id}])
+        del ref['id']
+        if unrestricted:
+            ref['unrestricted'] = True
+        r = self.post(
+            f'/users/{self.user_id}/application_credentials',
+            body={'application_credential': ref},
+        )
+        app_cred = r.result['application_credential']
+        auth_data = self.build_authentication_request(
+            app_cred_id=app_cred['id'], secret=app_cred['secret']
+        )
+        r = self.v3_create_token(auth_data)
+        return r.headers.get('X-Subject-Token')
+
+    def test_ec2_create_credential_with_restricted_app_cred(self):
+        """Test that a restricted app cred cannot create EC2 credentials.
+
+        A restricted application credential must not be allowed to create
+        EC2 credentials, as this would bypass the role restriction and
+        grant full user access to S3.
+        """
+        token_id = self._get_app_cred_token(unrestricted=False)
+        uri = self._get_ec2_cred_uri()
+        self.post(
+            uri,
+            body={'tenant_id': self.project_id},
+            token=token_id,
+            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."""
+        token_id = self._get_app_cred_token(unrestricted=True)
+        uri = self._get_ec2_cred_uri()
+        r = self.post(
+            uri,
+            body={'tenant_id': self.project_id},
+            token=token_id,
+            expected_status=http.client.CREATED,
+        )
+        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."""
         trustee = unit.new_user_ref(domain_id=self.domain_id)
@@ -1443,7 +1490,7 @@ def _get_trust_token(self):
         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."""  # noqa: E501
+        """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
@@ -1470,7 +1517,7 @@ def test_ec2_create_credential_trust_same_project_allowed(self):
         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."""  # noqa: E501
+        """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
@@ -1494,7 +1541,7 @@ def test_ec2_get_credential_trust_cross_project_blocked(self):
         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."""  # noqa: E501
+        """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

keystone/tests/unit/test_v3_oauth1.py

@@ -763,6 +763,30 @@ def test_trust_token_cannot_list_request_tokens(self):
         self.get(url, token=trust_token,
                  expected_status=http.client.FORBIDDEN)
 
+    def _create_app_cred_get_token(self):
+        ref = unit.new_application_credential_ref(roles=[{'id': self.role_id}])
+        del ref['id']
+        r = self.post(
+            f'/users/{self.user_id}/application_credentials',
+            body={'application_credential': ref},
+        )
+        app_cred = r.result['application_credential']
+        auth_data = self.build_authentication_request(
+            app_cred_id=app_cred['id'], secret=app_cred['secret']
+        )
+        return self.get_requested_token(auth_data)
+
+    def test_app_cred_token_cannot_authorize_request_token(self):
+        app_cred_token = self._create_app_cred_get_token()
+        url = self._approve_request_token_url()
+        body = {'roles': [{'id': self.role_id}]}
+        self.put(
+            url,
+            body=body,
+            token=app_cred_token,
+            expected_status=http.client.FORBIDDEN,
+        )
+
 
 class FernetAuthTokenTests(AuthTokenTests, OAuthFlowTests):