# keystone.api.users

# Source code for keystone.api.users

    #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    #    not use this file except in compliance with the License. You may obtain
    #    a copy of the License at
    #
    #         http://www.apache.org/licenses/LICENSE-2.0
    #
    #    Unless required by applicable law or agreed to in writing, software
    #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
    #    License for the specific language governing permissions and limitations
    #    under the License.

    # This file handles all flask-restful resources for /v3/users

    import base64
    import http.client
    import secrets
    import uuid

    import flask
    from oslo_serialization import jsonutils
    from werkzeug import exceptions

    from keystone.api._shared import json_home_relations
    from keystone.api import validation
    from keystone.application_credential import schema as app_cred_schema
    from keystone.common import json_home
    from keystone.common import provider_api
    from keystone.common import rbac_enforcer
    from keystone.common import utils
    from keystone.common import validation as ks_validation
    import keystone.conf
    from keystone import exception as ks_exception
    from keystone.i18n import _
    from keystone.identity import schema
    from keystone import notifications
    from keystone.server import flask as ks_flask

    CRED_TYPE_EC2 = 'ec2'
    CONF = keystone.conf.CONF
    ENFORCER = rbac_enforcer.RBACEnforcer
    PROVIDERS = provider_api.ProviderAPIs

    ACCESS_TOKEN_ID_PARAMETER_RELATION = (
        json_home_relations.os_oauth1_parameter_rel_func(
            parameter_name='access_token_id'
        )
    )

    def _convert_v3_to_ec2_credential(credential):
        # Prior to bug #1259584 fix, blob was stored unserialized
        # but it should be stored as a json string for compatibility
        # with the v3 credentials API.  Fall back to the old behavior
        # for backwards compatibility with existing DB contents
        try:
            blob = jsonutils.loads(credential['blob'])
        except TypeError:
            blob = credential['blob']
        return {
            'user_id': credential.get('user_id'),
            'tenant_id': credential.get('project_id'),
            'access': blob.get('access'),
            'secret': blob.get('secret'),
            'trust_id': blob.get('trust_id'),
        }

    def _format_token_entity(entity):
        formatted_entity = entity.copy()
        access_token_id = formatted_entity['id']
        user_id = formatted_entity.get('authorizing_user_id', '')
        if 'role_ids' in entity:
            formatted_entity.pop('role_ids')
        if 'access_secret' in entity:
            formatted_entity.pop('access_secret')

        url = f'/users/{user_id}/OS-OAUTH1/access_tokens/{access_token_id}/roles'

        formatted_entity.setdefault('links', {})
        formatted_entity['links']['roles'] = ks_flask.base_url(url)

        return formatted_entity

    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 application "
                    "credentials."
                )
                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 _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.
        """
        if not _is_delegated_token(oslo_context, token):
            return
        if project_id != oslo_context.project_id:
            raise ks_exception.ForbiddenAction(
                action=_(
                    'EC2 credential project does not match the '
                    'project scope of the delegated token'
                )
            )

    def _block_delegated_token(oslo_context, token):
        """Raise Forbidden if the token is any form of delegation."""
        if oslo_context.is_delegated_auth:
            raise ks_exception.Forbidden(
                _(
                    'Cannot manage OAuth access tokens with a token '
                    'issued via delegation.'
                )
            )
        if 'application_credential' in token.methods:
            raise ks_exception.Forbidden(
                _(
                    'Cannot manage OAuth access tokens with a token '
                    'issued via delegation.'
                )
            )

    def _block_delegated_token_app_creds(oslo_context, token):
        """Raise Forbidden if the token is a trust or OAuth1 delegation.

        Trust-scoped and OAuth1 access token-scoped tokens must not be used to
        create, list, read, or delete application credentials or access rules.
        Creating an application credential via such a token produces a persistent
        credential that outlives the delegation's expiry or scope, providing a
        backdoor that breaks the accountability model: the trust-scoped token
        carries the full delegation chain enabling audit, but a derived application
        credential does not.

        Application credential tokens are intentionally excluded from this check.
        The unrestricted/restricted distinction for application credentials is a
        documented feature handled separately by
        _check_unrestricted_application_credential.
        """
        trust_id = getattr(oslo_context, 'trust_id', None)
        access_token_id = getattr(token, 'access_token_id', None)
        if trust_id or access_token_id:
            raise ks_exception.Forbidden(
                _(
                    'Cannot manage application credentials with a token '
                    'issued via delegation.'
                )
            )

    def _build_user_target_enforcement():
        target = {}
        try:
            target['user'] = PROVIDERS.identity_api.get_user(
                flask.request.view_args.get('user_id')
            )
            if flask.request.view_args.get('group_id'):
                target['group'] = PROVIDERS.identity_api.get_group(
                    flask.request.view_args.get('group_id')
                )
        except ks_exception.NotFound:  # nosec
            # Defer existence in the event the user doesn't exist, we'll
            # check this later anyway.
            pass

        return target

    def _build_enforcer_target_data_owner_and_user_id_match():
        ref = {}
        if flask.request.view_args:
            credential_id = flask.request.view_args.get('credential_id')
            if credential_id is not None:
                hashed_id = utils.hash_access_key(credential_id)
                ref['credential'] = PROVIDERS.credential_api.get_credential(
                    hashed_id
                )
        return ref

    def _update_request_user_id_attribute():
        # This method handles a special case in policy enforcement. The application
        # credential API is underneath the user path (e.g.,
        # /v3/users/{user_id}/application_credentials/{application_credential_id}).
        # The RBAC enforcer thinks the user to evaluate for application credential
        # ownership comes from the path, but it should come from the actual
        # application credential reference. By ensuring we pull the user ID from
        # the application credential, we close a loop hole where users could
        # effectively bypass authorization to view or delete any application
        # credential in the system, assuming the attacker knows the application
        # credential ID of another user. So long as the attacker matches the user
        # ID in the request path to the user in the token of the request, they can
        # pass the `rule:owner` policy check. This method protects against that by
        # ensuring we use the application credential user ID and not something
        # determined from the client.
        try:
            app_cred = (
                PROVIDERS.application_credential_api.get_application_credential(
                    flask.request.view_args.get('application_credential_id')
                )
            )
            flask.request.view_args['user_id'] = app_cred['user_id']

            # This target isn't really used in the default policy for application
            # credentials, but we return it since we're using this method as a hook
            # to update the flask request variables, which are used later in the
            # keystone RBAC enforcer to populate the policy_dict, which ultimately
            # turns into target attributes.
            return {'user_id': app_cred['user_id']}
        except ks_exception.NotFound:  # nosec
            # Defer existance in the event the application credential doesn't
            # exist, we'll check this later anyway.
            pass

    def _format_role_entity(role_id):
        role = PROVIDERS.role_api.get_role(role_id)
        formatted_entity = role.copy()
        if 'description' in role:
            formatted_entity.pop('description')
        if 'enabled' in role:
            formatted_entity.pop('enabled')
        return formatted_entity

    class UserResource(ks_flask.ResourceBase):
        collection_key = 'users'
        member_key = 'user'
        get_member_from_driver = PROVIDERS.deferred_provider_lookup(
            api='identity_api', method='get_user'
        )

        @validation.request_query_schema(None)
        @validation.request_body_schema(None)
        @validation.response_body_schema(schema.user_get_response_body)
        def get(self, user_id: str):
            """Get a user resource.

            GET/HEAD /v3/users/{user_id}
            """
            ENFORCER.enforce_call(
                action='identity:get_user',
                build_target=_build_user_target_enforcement,
            )
            ref = PROVIDERS.identity_api.get_user(user_id)
            return self.wrap_member(ref)

        @validation.request_query_schema(None)
        @validation.request_body_schema(schema.user_update_request)
        @validation.response_body_schema(schema.user_update_request)
        def patch(self, user_id: str):
            """Update a user.

            PATCH /v3/users/{user_id}
            """
            ENFORCER.enforce_call(
                action='identity:update_user',
                build_target=_build_user_target_enforcement,
            )
            PROVIDERS.identity_api.get_user(user_id)
            user_data = self.request_body_json.get('user', {})
            self._require_matching_id(user_data)
            ref = PROVIDERS.identity_api.update_user(
                user_id, user_data, initiator=self.audit_initiator
            )
            return self.wrap_member(ref)

        @validation.request_query_schema(None)
        @validation.request_body_schema(None)
        @validation.response_body_schema(None)
        def delete(self, user_id: str):
            """Delete a user.

            DELETE /v3/users/{user_id}
            """
            ENFORCER.enforce_call(
                action='identity:delete_user',
                build_target=_build_user_target_enforcement,
            )
            PROVIDERS.identity_api.delete_user(
                user_id, initiator=self.audit_initiator
            )
            return None, http.client.NO_CONTENT

    class UsersResource(ks_flask.ResourceBase):
        collection_key = 'users'
        member_key = 'user'
        get_member_from_driver = PROVIDERS.deferred_provider_lookup(
            api='identity_api', method='get_user'
        )

        @validation.request_query_schema(schema.user_index_request_query)
        @validation.request_body_schema(None)
        @validation.response_body_schema(schema.user_index_response_body)
        def get(self):
            """List users.

            GET/HEAD /v3/users
            """
            filters = (
                'domain_id',
                'enabled',
                'idp_id',
                'name',
                'protocol_id',
                'unique_id',
                'password_expires_at',
            )
            target = None
            if self.oslo_context.domain_id:
                target = {'domain_id': self.oslo_context.domain_id}
            hints = self.build_driver_hints(filters)
            ENFORCER.enforce_call(
                action='identity:list_users', filters=filters, target_attr=target
            )
            domain = self._get_domain_id_for_list_request()
            if domain is None and self.oslo_context.domain_id:
                domain = self.oslo_context.domain_id
            refs = PROVIDERS.identity_api.list_users(
                domain_scope=domain, hints=hints
            )

            # If the user making the request used a domain-scoped token, let's make
            # sure we filter out users that are not in that domain. Otherwise, we'd
            # be exposing users in other domains. This if statement is needed in
            # case _get_domain_id_for_list_request() short-circuits due to
            # configuration and protects against information from other domains
            # leaking to people who shouldn't see it.
            if self.oslo_context.domain_id:
                domain_id = self.oslo_context.domain_id
                users = [user for user in refs if user['domain_id'] == domain_id]
            else:
                users = refs

            return self.wrap_collection(users, hints=hints)

        @validation.request_query_schema(None)
        @validation.request_body_schema(schema.user_create_request)
        @validation.response_body_schema(schema.user_create_request)
        def post(self):
            """Create a user.

            POST /v3/users
            """
            user_data = self.request_body_json.get('user', {})
            target = {'user': user_data}
            ENFORCER.enforce_call(
                action='identity:create_user', target_attr=target
            )
            user_data = self._normalize_dict(user_data)
            user_data = self._normalize_domain_id(user_data)
            ref = PROVIDERS.identity_api.create_user(
                user_data, initiator=self.audit_initiator
            )
            return self.wrap_member(ref), http.client.CREATED

    class UserChangePasswordResource(ks_flask.ResourceBase):

        @ks_flask.unenforced_api
        def get(self, user_id):
            # Special case, GET is not allowed.
            raise exceptions.MethodNotAllowed(valid_methods=['POST'])

        @ks_flask.unenforced_api
        def post(self, user_id):
            user_data = self.request_body_json.get('user', {})
            ks_validation.lazy_validate(schema.password_change, user_data)

            try:
                PROVIDERS.identity_api.change_password(
                    user_id=user_id,
                    original_password=user_data['original_password'],
                    new_password=user_data['password'],
                    initiator=self.audit_initiator,
                )
            except AssertionError as e:
                raise ks_exception.Unauthorized(
                    _('Error when changing user password: %s') % e
                )
            return None, http.client.NO_CONTENT

    class UserProjectsResource(ks_flask.ResourceBase):
        collection_key = 'projects'
        member_key = 'project'
        get_member_from_driver = PROVIDERS.deferred_provider_lookup(
            api='resource_api', method='get_project'
        )

        def get(self, user_id):
            filters = ('domain_id', 'enabled', 'name')
            ENFORCER.enforce_call(
                action='identity:list_user_projects',
                filters=filters,
                build_target=_build_user_target_enforcement,
            )
            hints = self.build_driver_hints(filters)
            refs = PROVIDERS.assignment_api.list_projects_for_user(user_id)
            return self.wrap_collection(refs, hints=hints)

    class UserGroupsResource(ks_flask.ResourceBase):
        collection_key = 'groups'
        member_key = 'group'
        get_member_from_driver = PROVIDERS.deferred_provider_lookup(
            api='identity_api', method='get_group'
        )

        def get(self, user_id):
            """Get groups for a user.

            GET/HEAD /v3/users/{user_id}/groups
            """
            filters = ('name',)
            hints = self.build_driver_hints(filters)
            ENFORCER.enforce_call(
                action='identity:list_groups_for_user',
                build_target=_build_user_target_enforcement,
                filters=filters,
            )
            refs = PROVIDERS.identity_api.list_groups_for_user(
                user_id=user_id, hints=hints
            )
            if self.oslo_context.domain_id:
                filtered_refs = []
                for ref in refs:
                    if ref['domain_id'] == self.oslo_context.domain_id:
                        filtered_refs.append(ref)
                refs = filtered_refs
            return self.wrap_collection(refs, hints=hints)

    class _UserOSEC2CredBaseResource(ks_flask.ResourceBase):
        collection_key = 'credentials'
        member_key = 'credential'

        @classmethod
        def _add_self_referential_link(cls, ref, collection_name=None):
            # NOTE(morgan): This should be refactored to have an EC2 Cred API with
            # a sane prefix instead of overloading the "_add_self_referential_link"
            # method. This was chosen as it more closely mirrors the pre-flask
            # code (for transition).
            path = '/users/%(user_id)s/credentials/OS-EC2/%(credential_id)s'

            url = ks_flask.base_url(path) % {
                'user_id': ref['user_id'],
                'credential_id': ref['access'],
            }
            ref.setdefault('links', {})
            ref['links']['self'] = url

    class UserOSEC2CredentialsResourceListCreate(_UserOSEC2CredBaseResource):

        def get(self, user_id):
            """List EC2 Credentials for user.

            GET/HEAD /v3/users/{user_id}/credentials/OS-EC2
            """
            ENFORCER.enforce_call(action='identity:ec2_list_credentials')
            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))
            return self.wrap_collection(collection_refs)

        def post(self, user_id):
            """Create EC2 Credential for user.

            POST /v3/users/{user_id}/credentials/OS-EC2
            """
            target = {}
            target['credential'] = {'user_id': user_id}
            ENFORCER.enforce_call(
                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)
            blob = {
                'access': uuid.uuid4().hex,
                'secret': uuid.uuid4().hex,
                'trust_id': self.oslo_context.trust_id,
            }
            credential_id = utils.hash_access_key(blob['access'])
            cred_data = {
                'user_id': user_id,
                'project_id': tenant_id,
                'blob': jsonutils.dumps(blob),
                'id': credential_id,
                'type': CRED_TYPE_EC2,
            }
            PROVIDERS.credential_api.create_credential(credential_id, cred_data)
            ref = _convert_v3_to_ec2_credential(cred_data)
            return self.wrap_member(ref), http.client.CREATED

    class UserOSEC2CredentialsResourceGetDelete(_UserOSEC2CredBaseResource):
        @staticmethod
        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.')
                )
            return cred

        def get(self, user_id, credential_id):
            """Get a specific EC2 credential.

            GET/HEAD /users/{user_id}/credentials/OS-EC2/{credential_id}
            """
            func = _build_enforcer_target_data_owner_and_user_id_match
            ENFORCER.enforce_call(
                action='identity:ec2_get_credential', build_target=func
            )
            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'),
            )
            return self.wrap_member(_convert_v3_to_ec2_credential(cred))

        def delete(self, user_id, credential_id):
            """Delete a specific EC2 credential.

            DELETE /users/{user_id}/credentials/OS-EC2/{credential_id}
            """
            func = _build_enforcer_target_data_owner_and_user_id_match
            ENFORCER.enforce_call(
                action='identity:ec2_delete_credential', build_target=func
            )
            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'),
            )
            PROVIDERS.credential_api.delete_credential(ec2_cred_id)
            return None, http.client.NO_CONTENT

    class _OAuth1ResourceBase(ks_flask.ResourceBase):
        collection_key = 'access_tokens'
        member_key = 'access_token'

        @classmethod
        def _add_self_referential_link(cls, ref, collection_name=None):
            # NOTE(morgan): This should be refactored to have an OAuth1 API with
            # a sane prefix instead of overloading the "_add_self_referential_link"
            # method. This was chosen as it more closely mirrors the pre-flask
            # code (for transition).
            ref.setdefault('links', {})
            path = '/users/{user_id}/OS-OAUTH1/access_tokens'.format(
                user_id=ref.get('authorizing_user_id', '')
            )
            ref['links']['self'] = ks_flask.base_url(path) + '/' + ref['id']

    class OAuth1ListAccessTokensResource(_OAuth1ResourceBase):

        def get(self, user_id):
            """List OAuth1 Access Tokens for user.

            GET /v3/users/{user_id}/OS-OAUTH1/access_tokens
            """
            ENFORCER.enforce_call(action='identity:list_access_tokens')
            _block_delegated_token(self.oslo_context, self.auth_context['token'])
            refs = PROVIDERS.oauth_api.list_access_tokens(user_id)
            formatted_refs = [_format_token_entity(x) for x in refs]
            return self.wrap_collection(formatted_refs)

    class OAuth1AccessTokenCRUDResource(_OAuth1ResourceBase):

        def get(self, user_id, access_token_id):
            """Get specific access token.

            GET/HEAD /v3/users/{user_id}/OS-OAUTH1/access_tokens/{access_token_id}
            """
            ENFORCER.enforce_call(action='identity:get_access_token')
            _block_delegated_token(self.oslo_context, self.auth_context['token'])
            access_token = PROVIDERS.oauth_api.get_access_token(access_token_id)
            if access_token['authorizing_user_id'] != user_id:
                raise ks_exception.NotFound()
            access_token = _format_token_entity(access_token)
            return self.wrap_member(access_token)

        def delete(self, user_id, access_token_id):
            """Delete specific access token.

            DELETE /v3/users/{user_id}/OS-OAUTH1/access_tokens/{access_token_id}
            """
            ENFORCER.enforce_call(
                action='identity:ec2_delete_credential',
                build_target=_build_enforcer_target_data_owner_and_user_id_match,
            )
            _block_delegated_token(self.oslo_context, self.auth_context['token'])
            access_token = PROVIDERS.oauth_api.get_access_token(access_token_id)
            reason = (
                'Invalidating the token cache because an access token for '
                'consumer {consumer_id} has been deleted. Authorization for '
                'users with OAuth tokens will be recalculated and enforced '
                'accordingly the next time they authenticate or validate a '
                'token.'.format(consumer_id=access_token['consumer_id'])
            )
            notifications.invalidate_token_cache_notification(reason)
            PROVIDERS.oauth_api.delete_access_token(
                user_id, access_token_id, initiator=self.audit_initiator
            )
            return None, http.client.NO_CONTENT

    class OAuth1AccessTokenRoleListResource(ks_flask.ResourceBase):
        collection_key = 'roles'
        member_key = 'role'

        def get(self, user_id, access_token_id):
            """List roles for a user access token.

            GET/HEAD /v3/users/{user_id}/OS-OAUTH1/access_tokens/
                     {access_token_id}/roles
            """
            ENFORCER.enforce_call(action='identity:list_access_token_roles')
            access_token = PROVIDERS.oauth_api.get_access_token(access_token_id)
            if access_token['authorizing_user_id'] != user_id:
                raise ks_exception.NotFound()
            authed_role_ids = access_token['role_ids']
            authed_role_ids = jsonutils.loads(authed_role_ids)
            refs = [_format_role_entity(x) for x in authed_role_ids]
            return self.wrap_collection(refs)

    class OAuth1AccessTokenRoleResource(ks_flask.ResourceBase):
        collection_key = 'roles'
        member_key = 'role'

        def get(self, user_id, access_token_id, role_id):
            """Get role for access token.

            GET/HEAD /v3/users/{user_id}/OS-OAUTH1/access_tokens/
                     {access_token_id}/roles/{role_id}
            """
            ENFORCER.enforce_call(action='identity:get_access_token_role')
            access_token = PROVIDERS.oauth_api.get_access_token(access_token_id)
            if access_token['authorizing_user_id'] != user_id:
                raise ks_exception.Unauthorized(_('User IDs do not match'))
            authed_role_ids = access_token['role_ids']
            authed_role_ids = jsonutils.loads(authed_role_ids)
            for authed_role_id in authed_role_ids:
                if authed_role_id == role_id:
                    role = _format_role_entity(role_id)
                    return self.wrap_member(role)
            raise ks_exception.RoleNotFound(role_id=role_id)

    class UserAppCredListCreateResource(ks_flask.ResourceBase):
        collection_key = 'application_credentials'
        member_key = 'application_credential'
        _public_parameters = frozenset(
            [
                'id',
                'name',
                'description',
                'expires_at',
                'project_id',
                'roles',
                # secret is only exposed after create, it is not stored
                'secret',
                'links',
                'unrestricted',
                'access_rules',
            ]
        )

        @staticmethod
        def _generate_secret():
            length = 64
            secret = secrets.token_bytes(length)
            secret = base64.urlsafe_b64encode(secret)
            secret = secret.rstrip(b'=')
            secret = secret.decode('utf-8')
            return secret

        @staticmethod
        def _normalize_role_list(app_cred_roles):
            roles = []
            for role in app_cred_roles:
                if role.get('id'):
                    roles.append(role)
                else:
                    roles.append(
                        PROVIDERS.role_api.get_unique_role_by_name(role['name'])
                    )
            return roles

        def _get_roles(self, app_cred_data, token):
            if app_cred_data.get('roles'):
                roles = self._normalize_role_list(app_cred_data['roles'])
                # When "roles" passed into the application credentials creation
                # we need to ensure also all implied roles are included similarly
                # to how it behaves when no roles are passed and current user roles
                # are being used.
                # So loop over all roles implied by the current role and add it
                # explicitly if not already there
                for role in roles:
                    for implied_role in PROVIDERS.role_api.list_implied_roles(
                        role['id']
                    ):
                        imp_role_obj = PROVIDERS.role_api.get_role(
                            implied_role['implied_role_id']
                        )
                        if imp_role_obj['id'] not in [x['id'] for x in roles]:
                            roles.append(imp_role_obj)
                # NOTE(cmurphy): The user is not allowed to add a role that is not
                # in their token. This is to prevent trustees or application
                # credential users from escallating their privileges to include
                # additional roles that the trustor or application credential
                # creator has assigned on the project.
                token_roles = [r['id'] for r in token.roles]
                for role in roles:
                    if role['id'] not in token_roles:
                        detail = _(
                            'Cannot create an application credential with '
                            'unassigned role'
                        )
                        raise ks_exception.ApplicationCredentialValidationError(
                            detail=detail
                        )
            else:
                roles = token.roles
            return roles

        @validation.request_query_schema(
            app_cred_schema.application_credential_index_request_query
        )
        @validation.response_body_schema(
            app_cred_schema.application_credential_index_response_body
        )
        def get(self, user_id):
            """List application credentials for user.

            GET/HEAD /v3/users/{user_id}/application_credentials
            """
            filters = ('name',)
            ENFORCER.enforce_call(
                action='identity:list_application_credentials', filters=filters
            )
            token = self.auth_context['token']
            _block_delegated_token_app_creds(self.oslo_context, token)
            app_cred_api = PROVIDERS.application_credential_api
            hints = self.build_driver_hints(filters)
            refs = app_cred_api.list_application_credentials(user_id, hints=hints)
            return self.wrap_collection(refs, hints=hints)

        @validation.request_body_schema(
            app_cred_schema.application_credential_create_request_body
        )
        @validation.response_body_schema(
            app_cred_schema.application_credential_create_response_body
        )
        def post(self, user_id):
            """Create application credential.

            POST /v3/users/{user_id}/application_credentials
            """
            ENFORCER.enforce_call(action='identity:create_application_credential')
            app_cred_data = self.request_body_json.get(
                'application_credential', {}
            )
            token = self.auth_context['token']
            _block_delegated_token_app_creds(self.oslo_context, token)
            _check_unrestricted_application_credential(token)
            if self.oslo_context.user_id != user_id:
                action = _(
                    'Cannot create an application credential for another user.'
                )
                raise ks_exception.ForbiddenAction(action=action)
            project_id = self.oslo_context.project_id
            app_cred_data = self._assign_unique_id(app_cred_data)
            if not app_cred_data.get('secret'):
                app_cred_data['secret'] = self._generate_secret()
            app_cred_data['user_id'] = user_id
            app_cred_data['project_id'] = project_id
            app_cred_data['roles'] = self._get_roles(app_cred_data, token)
            if app_cred_data.get('expires_at'):
                app_cred_data['expires_at'] = utils.parse_expiration_date(
                    app_cred_data['expires_at']
                )
            if app_cred_data.get('access_rules'):
                for access_rule in app_cred_data['access_rules']:
                    # If user provides an access rule by ID, it will be looked up
                    # by ID. If user provides an access rule that is identical to
                    # an existing one, the ID generated here will be ignored and
                    # the pre-existing access rule will be used.
                    if 'id' not in access_rule:
                        # Generate directly, rather than using _assign_unique_id,
                        # so that there is no deep copy made
                        access_rule['id'] = uuid.uuid4().hex
            app_cred_data = self._normalize_dict(app_cred_data)
            app_cred_api = PROVIDERS.application_credential_api

            try:
                ref = app_cred_api.create_application_credential(
                    app_cred_data, initiator=self.audit_initiator
                )
            except ks_exception.RoleAssignmentNotFound as e:
                # Raise a Bad Request, not a Not Found, in accordance with the
                # API-SIG recommendations:
                # https://specs.openstack.org/openstack/api-wg/guidelines/http.html#failure-code-clarifications
                raise ks_exception.ApplicationCredentialValidationError(
                    detail=str(e)
                )
            return self.wrap_member(ref), http.client.CREATED

    class UserAppCredGetDeleteResource(ks_flask.ResourceBase):
        collection_key = 'application_credentials'
        member_key = 'application_credential'

        @validation.request_body_schema(
            app_cred_schema.application_credential_request_body
        )
        @validation.response_body_schema(
            app_cred_schema.application_credential_response_body
        )
        def get(self, user_id, application_credential_id):
            """Get application credential resource.

            GET/HEAD /v3/users/{user_id}/application_credentials/
                     {application_credential_id}
            """
            target = _update_request_user_id_attribute()
            ENFORCER.enforce_call(
                action='identity:get_application_credential', target_attr=target
            )
            token = self.auth_context['token']
            _block_delegated_token_app_creds(self.oslo_context, token)
            ref = PROVIDERS.application_credential_api.get_application_credential(
                application_credential_id
            )
            return self.wrap_member(ref)

        def delete(self, user_id, application_credential_id):
            """Delete application credential resource.

            DELETE /v3/users/{user_id}/application_credentials/
                   {application_credential_id}
            """
            target = _update_request_user_id_attribute()
            ENFORCER.enforce_call(
                action='identity:delete_application_credential', target_attr=target
            )
            token = self.auth_context['token']
            _block_delegated_token_app_creds(self.oslo_context, token)
            _check_unrestricted_application_credential(token)
            PROVIDERS.application_credential_api.delete_application_credential(
                application_credential_id, initiator=self.audit_initiator
            )
            return None, http.client.NO_CONTENT

    class UserAccessRuleListResource(ks_flask.ResourceBase):
        collection_key = 'access_rules'
        member_key = 'access_rule'

        @validation.request_query_schema(
            app_cred_schema.access_rule_index_request_query
        )
        @validation.response_body_schema(
            app_cred_schema.access_rule_index_response_body
        )
        def get(self, user_id):
            """List access rules for user.

            GET/HEAD /v3/users/{user_id}/access_rules
            """
            filters = ('service', 'path', 'method')
            ENFORCER.enforce_call(
                action='identity:list_access_rules',
                filters=filters,
                build_target=_build_user_target_enforcement,
            )
            token = self.auth_context['token']
            _block_delegated_token_app_creds(self.oslo_context, token)
            app_cred_api = PROVIDERS.application_credential_api
            hints = self.build_driver_hints(filters)
            refs = app_cred_api.list_access_rules_for_user(user_id, hints=hints)
            hints = self.build_driver_hints(filters)
            return self.wrap_collection(refs, hints=hints)

    class UserAccessRuleGetDeleteResource(ks_flask.ResourceBase):
        collection_key = 'access_rules'
        member_key = 'access_rule'

        @validation.request_query_schema(
            app_cred_schema.access_rule_show_request_query
        )
        @validation.response_body_schema(
            app_cred_schema.access_rule_show_response_body
        )
        def get(self, user_id, access_rule_id):
            """Get access rule resource.

            GET/HEAD /v3/users/{user_id}/access_rules/{access_rule_id}
            """
            ENFORCER.enforce_call(
                action='identity:get_access_rule',
                build_target=_build_user_target_enforcement,
            )
            token = self.auth_context['token']
            _block_delegated_token_app_creds(self.oslo_context, token)
            ref = PROVIDERS.application_credential_api.get_access_rule(
                access_rule_id, user_id=user_id
            )
            return self.wrap_member(ref)

        @validation.request_body_schema(None)
        @validation.response_body_schema(None)
        def delete(self, user_id, access_rule_id):
            """Delete access rule resource.

            DELETE /v3/users/{user_id}/access_rules/{access_rule_id}
            """
            ENFORCER.enforce_call(
                action='identity:delete_access_rule',
                build_target=_build_user_target_enforcement,
            )
            token = self.auth_context['token']
            _block_delegated_token_app_creds(self.oslo_context, token)
            PROVIDERS.application_credential_api.delete_access_rule(
                access_rule_id, initiator=self.audit_initiator, user_id=user_id
            )
            return None, http.client.NO_CONTENT

    class UserTokensResource(ks_flask.ResourceBase):
        collection_key = 'tokens'
        member_key = 'token'

        def delete(self, user_id):
            """Revoke all tokens for a user.

            DELETE /v3/users/{user_id}/tokens

            Revokes every token issued for the given user by creating a
            revocation event scoped to user_id. This is the self-service
            remedy for users who cannot change their password (LDAP,
            federated, OIDC) and need to invalidate all active sessions.

            Users may revoke their own tokens. Admins may revoke tokens for
            any user.
            """
            ENFORCER.enforce_call(
                action='identity:revoke_tokens_for_user',
                build_target=_build_user_target_enforcement,
            )
            PROVIDERS.identity_api.get_user(user_id)
            PROVIDERS.revoke_api.revoke_by_user(user_id)
            notifications.invalidate_token_cache_notification(
                f'Invalidating token cache because all tokens for user '
                f'{user_id} were explicitly revoked.'
            )
            return None, http.client.NO_CONTENT

    class UserAPI(ks_flask.APIBase):
        _name = 'users'
        _import_name = __name__
        resource_mapping = [
            ks_flask.construct_resource_map(
                resource=UsersResource,
                url='/users',
                resource_kwargs={},
                rel='users',
                path_vars={},
            ),
            ks_flask.construct_resource_map(
                resource=UserResource,
                url='/users/<string:user_id>',
                resource_kwargs={},
                rel='user',
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=UserChangePasswordResource,
                url='/users/<string:user_id>/password',
                resource_kwargs={},
                rel='user_change_password',
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=UserGroupsResource,
                url='/users/<string:user_id>/groups',
                resource_kwargs={},
                rel='user_groups',
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=UserProjectsResource,
                url='/users/<string:user_id>/projects',
                resource_kwargs={},
                rel='user_projects',
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=UserOSEC2CredentialsResourceListCreate,
                url='/users/<string:user_id>/credentials/OS-EC2',
                resource_kwargs={},
                rel='user_credentials',
                resource_relation_func=(
                    json_home_relations.os_ec2_resource_rel_func
                ),
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=UserOSEC2CredentialsResourceGetDelete,
                url=(
                    '/users/<string:user_id>/credentials/OS-EC2/'
                    '<string:credential_id>'
                ),
                resource_kwargs={},
                rel='user_credential',
                resource_relation_func=(
                    json_home_relations.os_ec2_resource_rel_func
                ),
                path_vars={
                    'credential_id': json_home.build_v3_parameter_relation(
                        'credential_id'
                    ),
                    'user_id': json_home.Parameters.USER_ID,
                },
            ),
            ks_flask.construct_resource_map(
                resource=OAuth1ListAccessTokensResource,
                url='/users/<string:user_id>/OS-OAUTH1/access_tokens',
                resource_kwargs={},
                rel='user_access_tokens',
                resource_relation_func=(
                    json_home_relations.os_oauth1_resource_rel_func
                ),
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=OAuth1AccessTokenCRUDResource,
                url=(
                    '/users/<string:user_id>/OS-OAUTH1/'
                    'access_tokens/<string:access_token_id>'
                ),
                resource_kwargs={},
                rel='user_access_token',
                resource_relation_func=(
                    json_home_relations.os_oauth1_resource_rel_func
                ),
                path_vars={
                    'access_token_id': ACCESS_TOKEN_ID_PARAMETER_RELATION,
                    'user_id': json_home.Parameters.USER_ID,
                },
            ),
            ks_flask.construct_resource_map(
                resource=OAuth1AccessTokenRoleListResource,
                url=(
                    '/users/<string:user_id>/OS-OAUTH1/access_tokens/'
                    '<string:access_token_id>/roles'
                ),
                resource_kwargs={},
                rel='user_access_token_roles',
                resource_relation_func=(
                    json_home_relations.os_oauth1_resource_rel_func
                ),
                path_vars={
                    'access_token_id': ACCESS_TOKEN_ID_PARAMETER_RELATION,
                    'user_id': json_home.Parameters.USER_ID,
                },
            ),
            ks_flask.construct_resource_map(
                resource=OAuth1AccessTokenRoleResource,
                url=(
                    '/users/<string:user_id>/OS-OAUTH1/access_tokens/'
                    '<string:access_token_id>/roles/<string:role_id>'
                ),
                resource_kwargs={},
                rel='user_access_token_role',
                resource_relation_func=(
                    json_home_relations.os_oauth1_resource_rel_func
                ),
                path_vars={
                    'access_token_id': ACCESS_TOKEN_ID_PARAMETER_RELATION,
                    'role_id': json_home.Parameters.ROLE_ID,
                    'user_id': json_home.Parameters.USER_ID,
                },
            ),
            ks_flask.construct_resource_map(
                resource=UserAppCredListCreateResource,
                url='/users/<string:user_id>/application_credentials',
                resource_kwargs={},
                rel='application_credentials',
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=UserAppCredGetDeleteResource,
                url=(
                    '/users/<string:user_id>/application_credentials/'
                    '<string:application_credential_id>'
                ),
                resource_kwargs={},
                rel='application_credential',
                path_vars={
                    'user_id': json_home.Parameters.USER_ID,
                    'application_credential_id': json_home.Parameters.APPLICATION_CRED_ID,
                },
            ),
            ks_flask.construct_resource_map(
                resource=UserAccessRuleListResource,
                url='/users/<string:user_id>/access_rules',
                resource_kwargs={},
                rel='access_rules',
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
            ks_flask.construct_resource_map(
                resource=UserAccessRuleGetDeleteResource,
                url=(
                    '/users/<string:user_id>/access_rules/<string:access_rule_id>'
                ),
                resource_kwargs={},
                rel='access_rule',
                path_vars={
                    'user_id': json_home.Parameters.USER_ID,
                    'access_rule_id': json_home.Parameters.ACCESS_RULE_ID,
                },
            ),
            ks_flask.construct_resource_map(
                resource=UserTokensResource,
                url='/users/<string:user_id>/tokens',
                resource_kwargs={},
                rel='user_tokens',
                path_vars={'user_id': json_home.Parameters.USER_ID},
            ),
        ]

    APIs = (UserAPI,)

this page last updated: 2026-08-17 12:45:24

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug