Synchronise 2024.1 with upstream

.zuul.yaml

@@ -186,6 +186,8 @@
               - ^keystone/tests/unit/.*$
               - ^releasenotes/.*$
               - ^setup.cfg$
+        - openstack-tox-pep8:
+            voting: false
         - tempest-full-py3:
             irrelevant-files: *tempest-irrelevant-files
         - grenade:
@@ -206,6 +208,8 @@
             irrelevant-files: *tempest-irrelevant-files
         - grenade:
             irrelevant-files: *tempest-irrelevant-files
+        - openstack-tox-pep8:
+            voting: false
         - tempest-ipv6-only:
             irrelevant-files: *tempest-irrelevant-files
         # FIXME(dmendiza): temporarily disabling protection job

keystone/api/credentials.py

@@ -68,13 +68,13 @@ def _check_credential_project_scope(token, oslo_context, credential):
     cred_project_id = credential.get('project_id')
 
     if cred_project_id != token_project_id:
-        if CONF.security_compliance.allow_insecure_admin_trust_cross_project_credentials_access:  # noqa: E501
+        if CONF.security_compliance.allow_insecure_admin_trust_cross_project_credentials_access:  # noqa
             # 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:  # nosec
+            except exception.ForbiddenAction:
                 pass
         raise exception.ForbiddenAction(
             action=_(
@@ -212,9 +212,6 @@ def post(self):
         ENFORCER.enforce_call(
             action='identity:create_credential', target_attr=target
         )
-        token = self.auth_context['token']
-        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':
@@ -223,19 +220,6 @@ def post(self):
         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)
-        if (
-            app_cred_id is not None
-            and credential.get('type', '').lower() == 'ec2'
-        ):
-            ac_api = PROVIDERS.application_credential_api
-            app_cred = ac_api.get_application_credential(app_cred_id)
-            if credential.get('project_id') != app_cred['project_id']:
-                action = _(
-                    'EC2 credential project_id must match the '
-                    'project of the application credential used '
-                    'to authenticate'
-                )
-                raise exception.ForbiddenAction(action=action)
         ref = self._assign_unique_id(
             self._normalize_dict(credential),
             trust_id=trust_id, app_cred_id=app_cred_id,
@@ -267,6 +251,7 @@ def patch(self, credential_id):
         _check_credential_project_scope(
             self.auth_context['token'], self.oslo_context, current
         )
+
         credential = self.request_body_json.get('credential', {})
         validation.lazy_validate(schema.credential_update, credential)
         self._validate_blob_update_keys(current.copy(), credential.copy())

keystone/api/trusts.py

@@ -58,7 +58,7 @@ def _check_application_credential():
     expanding its effective scope beyond the single project it was issued for.
     This applies regardless of the 'unrestricted' flag.
     """
-    if CONF.security_compliance.allow_insecure_application_credential_trust_escalation:  # noqa: E501
+    if CONF.security_compliance.allow_insecure_application_credential_trust_escalation:  # noqa
         return
     auth_context = flask.request.environ.get(
         authorization.AUTH_CONTEXT_ENV, {}

keystone/api/users.py

@@ -495,7 +495,8 @@ def _get_raw_cred(credential_id):
         cred = PROVIDERS.credential_api.get_credential(credential_id)
         if not cred or cred['type'] != CRED_TYPE_EC2:
             raise ks_exception.Unauthorized(
-                message=_('EC2 access key not found.'))
+                message=_('EC2 access key not found.')
+            )
         return cred
 
     def get(self, user_id, credential_id):

keystone/tests/unit/auth/plugins/test_mapped.py

@@ -210,7 +210,7 @@ def test_handle_scoped_token_preserves_expires_at(
         bypass vulnerability: handle_scoped_token must include expires_at in
         the returned response_data so that issue_token() does not fall back to
         default_expire_time().
-        """  # noqa: E501
+        """
         original_expiry = '2026-04-26T08:59:30.000000Z'
         token = self._make_federated_token_mock(original_expiry)
 
@@ -244,7 +244,7 @@ def test_handle_scoped_token_returns_federation_metadata(
         mock_validate_groups,
         mock_notify,
     ):
-        """Rescoped federated token still returns all required federation data."""  # noqa: E501
+        """Rescoped federated token still returns all required federation data."""
         token = self._make_federated_token_mock('2026-04-26T08:59:30.000000Z')
 
         result = mapped.handle_scoped_token(

keystone/tests/unit/common/test_rbac_enforcer.py

@@ -776,7 +776,7 @@ def test_query_filter_cannot_overwrite_view_args(self):
         overwrite the URL-path-sourced value used in %(user_id)s policy
         substitutions, bypassing ownership checks such as
         ADMIN_OR_SYSTEM_READER_OR_OWNER on /v3/users/{user_id}/... endpoints.
-        """  # noqa: E501
+        """
         real_arg_id = uuid.uuid4().hex
         injected_arg_id = uuid.uuid4().hex
         seen = {}

keystone/tests/unit/test_v3_application_credential.py

@@ -12,7 +12,6 @@
 
 import datetime
 from testtools import matchers
-import unittest
 import uuid
 
 import http.client
@@ -616,14 +615,11 @@ def test_update_application_credential(self):
             # need to be rolled into the base MEMBER_PATH_FMT
             member_path = '/v3%s' % MEMBER_PATH_FMT % {
                 'user_id': self.user_id,
-                'app_cred_id': app_cred_id,
-            }
-            c.patch(
-                member_path,
-                json=app_cred_body,
-                expected_status_code=http.client.METHOD_NOT_ALLOWED,
-                headers={'X-Auth-Token': token},
-            )
+                'app_cred_id': app_cred_id}
+            c.patch(member_path,
+                    json=app_cred_body,
+                    expected_status_code=http.client.METHOD_NOT_ALLOWED,
+                    headers={'X-Auth-Token': token})
 
     def _get_trust_token(self, c, pw_token):
         """Return a trust-scoped token for self.user_id on self.project_id."""
@@ -741,7 +737,7 @@ def test_delegation_guard_trust_list_access_rules(self):
             )
 
     def test_delegation_guard_trust_get_access_rule(self):
-        """Trust-scoped token cannot read a specific access rule (LP#2150089)."""  # noqa: E501
+        """Trust-scoped token cannot read a specific access rule (LP#2150089)."""
         access_rules = [
             {'path': '/v3/projects', 'method': 'GET', 'service': 'identity'}
         ]
@@ -795,115 +791,3 @@ def test_delegation_guard_trust_delete_access_rule(self):
                 headers={'X-Auth-Token': trust_token},
                 expected_status_code=http.client.FORBIDDEN,
             )
-
-    def test_list_access_rules(self):
-        access_rules: list[dict[str, str]] = [
-            {"service": "foo", "method": "GET", "path": "/bar"}
-        ]
-        with self.test_client() as c:
-            roles = [{'id': self.role_id}]
-            app_cred_body = self._app_cred_body(
-                roles=roles, access_rules=access_rules
-            )
-            token = self.get_scoped_token()
-            c.post(
-                f"/v3/users/{self.user_id}/application_credentials",
-                json=app_cred_body,
-                expected_status_code=http.client.CREATED,
-                headers={"X-Auth-Token": token},
-            )
-            # Invoke GET access_rules and trigger internal validation
-            r = c.get(
-                f"/v3/users/{self.user_id}/access_rules",
-                expected_status_code=http.client.OK,
-                headers={"X-Auth-Token": token},
-            )
-            ar = r.json["access_rules"]
-            self.assertEqual(access_rules[0]["method"], ar[0]["method"])
-
-    # TODO(stephenfin): This will pass once we increase strictness of the query
-    # string validation
-    @unittest.expectedFailure
-    def test_list_access_rules_invalid_qs(self):
-        with self.test_client() as c:
-            token = self.get_scoped_token()
-            # Invoke GET access_rules with unsupported query parameters and
-            # trigger internal validation
-            c.get(
-                f"/v3/users/{self.user_id}/access_rules?user_id=foo",
-                expected_status_code=http.client.BAD_REQUEST,
-                headers={"X-Auth-Token": token},
-            )
-
-    def test_show_access_rule(self):
-        access_rules: list[dict[str, str]] = [
-            {"service": "foo", "method": "GET", "path": "/bar"}
-        ]
-        with self.test_client() as c:
-            roles = [{'id': self.role_id}]
-            app_cred_body = self._app_cred_body(
-                roles=roles, access_rules=access_rules
-            )
-            token = self.get_scoped_token()
-            resp = c.post(
-                f"/v3/users/{self.user_id}/application_credentials",
-                json=app_cred_body,
-                expected_status_code=http.client.CREATED,
-                headers={"X-Auth-Token": token},
-            )
-            access_rule_id = resp.json["application_credential"][
-                "access_rules"
-            ][0]["id"]
-            # Invoke GET access_rules/{id} and trigger internal validation
-            c.get(
-                f"/v3/users/{self.user_id}/access_rules/{access_rule_id}",
-                expected_status_code=http.client.OK,
-                headers={"X-Auth-Token": token},
-            )
-
-    # TODO(stephenfin): This will pass once we increase strictness of the query
-    # string validation
-    @unittest.expectedFailure
-    def test_show_access_rule_invalid_qs(self):
-        with self.test_client() as c:
-            token = self.get_scoped_token()
-            # Invoke GET access_rules/{id} with unsupported query parameters
-            # and trigger internal validation
-            c.get(
-                f"/v3/users/{self.user_id}/access_rules/{access_rule_id}"  # noqa: E501,F821
-                "?foo=bar",
-                expected_status_code=http.client.BAD_REQUEST,
-                headers={"X-Auth-Token": token},
-            )
-
-    def test_delete_access_rule(self):
-        access_rules: list[dict[str, str]] = [
-            {"service": "foo", "method": "GET", "path": "/bar"}
-        ]
-        with self.test_client() as c:
-            roles = [{'id': self.role_id}]
-            app_cred_body = self._app_cred_body(
-                roles=roles, access_rules=access_rules
-            )
-            token = self.get_scoped_token()
-            resp = c.post(
-                f"/v3/users/{self.user_id}/application_credentials",
-                json=app_cred_body,
-                expected_status_code=http.client.CREATED,
-                headers={"X-Auth-Token": token},
-            )
-            app_cred: dict = resp.json["application_credential"]
-            access_rule_id = app_cred["access_rules"][0]["id"]
-            c.delete(
-                f"/v3/users/{self.user_id}/application_credentials"
-                f"/{app_cred['id']}",
-                json=app_cred_body,
-                expected_status_code=http.client.NO_CONTENT,
-                headers={"X-Auth-Token": token},
-            )
-            # Invoke GET access_rules/{id} and trigger internal validation
-            c.delete(
-                f"/v3/users/{self.user_id}/access_rules/{access_rule_id}",
-                expected_status_code=http.client.NO_CONTENT,
-                headers={"X-Auth-Token": token},
-            )

keystone/tests/unit/test_v3_credential.py

@@ -718,7 +718,7 @@ def test_trust_token_cannot_list_totp_credentials(self):
         TOTP credentials have no project anchor. Before this fix the
         project boundary check skipped null-project credentials, allowing a
         delegation token to enumerate and exfiltrate MFA secrets.
-        """  # noqa: E501
+        """
         totp_ref = {
             'user_id': self.user_id,
             'type': 'totp',
@@ -751,7 +751,7 @@ def test_trust_token_cannot_read_totp_credential(self):
         )
 
     def test_trust_token_cannot_update_totp_credential(self):
-        """Trust-scoped token must not be able to update a TOTP credential blob."""  # noqa: E501
+        """Trust-scoped token must not be able to update a TOTP credential blob."""
         totp_ref = {
             'user_id': self.user_id,
             'type': 'totp',
@@ -834,8 +834,7 @@ def test_ec2_auth_trust_cross_project_scoped_to_trust(self):
             expected_status=http.client.OK,
         )
         # The resulting token is scoped to the trust's project, not to
-        # other_project -- the trust mechanism prevents cross-project
-        # escalation.
+        # other_project -- the trust mechanism prevents cross-project escalation.
         token_project = r.result['token']['project']['id']
         self.assertEqual(self.project_id, token_project)
         self.assertNotEqual(other_project['id'], token_project)
@@ -914,8 +913,7 @@ def test_app_cred_ec2_credential(self):
             '/credentials',
             body={'credential': ref},
             token=token_id,
-            expected_status=http.client.CONFLICT,
-        )
+            expected_status=http.client.CONFLICT)
 
     def _get_app_cred_token(self, unrestricted=False):
         """Create an application credential and return its token."""
@@ -1272,11 +1270,11 @@ def test_ec2_auth_access_token_cross_project_blocked(self):
         Auth-time check: if a cross-project EC2 credential backed by an OAuth1
         access token exists, POST /ec2tokens must reject it when the
         credential's project_id differs from the access token's project_id.
-        """  # noqa: E501
+        """
         access_key, _ = self._get_access_token()
 
         # Retrieve the stored access token to get its project_id
-        access_token = PROVIDERS.oauth_api.get_access_token(  # noqa F841
+        access_token = PROVIDERS.oauth_api.get_access_token(
             access_key.decode('utf-8')
             if isinstance(access_key, bytes)
             else access_key

keystone/tests/unit/test_v3_oauth1.py

@@ -459,12 +459,12 @@ def test_list_access_tokens_with_app_cred_blocked(self):
         )
 
     def test_get_access_token_with_app_cred_blocked(self):
-        """Application credential token must not get a specific access token."""  # noqa: E501
+        """Application credential token must not get a specific access token."""
         self.test_oauth_flow()
         token = self._get_app_cred_token()
         access_token_key = self.access_token.key.decode()
         self.get(
-            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',  # noqa: E501
+            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',
             token=token,
             expected_status=http.client.FORBIDDEN,
         )
@@ -475,7 +475,7 @@ def test_delete_access_token_with_app_cred_blocked(self):
         token = self._get_app_cred_token()
         access_token_key = self.access_token.key.decode()
         self.delete(
-            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',  # noqa: E501
+            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',
             token=token,
             expected_status=http.client.FORBIDDEN,
         )
@@ -506,7 +506,7 @@ def test_get_access_token_with_trust_token_blocked(self):
         trust_token = r.headers['X-Subject-Token']
         access_token_key = self.access_token.key.decode()
         self.get(
-            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',  # noqa: E501
+            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',
             token=trust_token,
             expected_status=http.client.FORBIDDEN,
         )
@@ -537,7 +537,7 @@ def test_delete_access_token_with_trust_token_blocked(self):
         trust_token = r.headers['X-Subject-Token']
         access_token_key = self.access_token.key.decode()
         self.delete(
-            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',  # noqa: E501
+            f'/users/{self.user_id}/OS-OAUTH1/access_tokens/{access_token_key}',
             token=trust_token,
             expected_status=http.client.FORBIDDEN,
         )