diff --git a/.zuul.yaml b/.zuul.yaml index e2c5e011b4..e8476c1722 100644 --- a/.zuul.yaml +++ b/.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 diff --git a/keystone/api/credentials.py b/keystone/api/credentials.py index 5518c244b4..8b7768bb3a 100644 --- a/keystone/api/credentials.py +++ b/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()) diff --git a/keystone/api/trusts.py b/keystone/api/trusts.py index 64988759a3..ab2aa9a7a2 100644 --- a/keystone/api/trusts.py +++ b/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, {} diff --git a/keystone/api/users.py b/keystone/api/users.py index eb7b1366ec..7d529531e6 100644 --- a/keystone/api/users.py +++ b/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): diff --git a/keystone/tests/unit/auth/plugins/test_mapped.py b/keystone/tests/unit/auth/plugins/test_mapped.py index ba7195c746..0a1cec8abf 100644 --- a/keystone/tests/unit/auth/plugins/test_mapped.py +++ b/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( diff --git a/keystone/tests/unit/common/test_rbac_enforcer.py b/keystone/tests/unit/common/test_rbac_enforcer.py index 4393a52de0..b83bbcb724 100644 --- a/keystone/tests/unit/common/test_rbac_enforcer.py +++ b/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 = {} diff --git a/keystone/tests/unit/test_v3_application_credential.py b/keystone/tests/unit/test_v3_application_credential.py index d3b6a954d0..03c03e1bc8 100644 --- a/keystone/tests/unit/test_v3_application_credential.py +++ b/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}, - ) diff --git a/keystone/tests/unit/test_v3_credential.py b/keystone/tests/unit/test_v3_credential.py index 0c3f28e214..b871291934 100644 --- a/keystone/tests/unit/test_v3_credential.py +++ b/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 diff --git a/keystone/tests/unit/test_v3_oauth1.py b/keystone/tests/unit/test_v3_oauth1.py index 58b8b25e73..113e92bcf6 100644 --- a/keystone/tests/unit/test_v3_oauth1.py +++ b/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, )