# 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
[docs]
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'
)
[docs]
@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)
[docs]
@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)
[docs]
@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
[docs]
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'
)
[docs]
@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)
[docs]
@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
[docs]
class UserChangePasswordResource(ks_flask.ResourceBase):
[docs]
@ks_flask.unenforced_api
def get(self, user_id):
# Special case, GET is not allowed.
raise exceptions.MethodNotAllowed(valid_methods=['POST'])
[docs]
@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
[docs]
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'
)
[docs]
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)
[docs]
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'
)
[docs]
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
[docs]
class UserOSEC2CredentialsResourceListCreate(_UserOSEC2CredBaseResource):
[docs]
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)
[docs]
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
[docs]
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
[docs]
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))
[docs]
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']
[docs]
class OAuth1ListAccessTokensResource(_OAuth1ResourceBase):
[docs]
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)
[docs]
class OAuth1AccessTokenCRUDResource(_OAuth1ResourceBase):
[docs]
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)
[docs]
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
[docs]
class OAuth1AccessTokenRoleListResource(ks_flask.ResourceBase):
collection_key = 'roles'
member_key = 'role'
[docs]
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)
[docs]
class OAuth1AccessTokenRoleResource(ks_flask.ResourceBase):
collection_key = 'roles'
member_key = 'role'
[docs]
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)
[docs]
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
[docs]
@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)
[docs]
@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
[docs]
class UserAppCredGetDeleteResource(ks_flask.ResourceBase):
collection_key = 'application_credentials'
member_key = 'application_credential'
[docs]
@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)
[docs]
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
[docs]
class UserAccessRuleListResource(ks_flask.ResourceBase):
collection_key = 'access_rules'
member_key = 'access_rule'
[docs]
@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)
[docs]
class UserAccessRuleGetDeleteResource(ks_flask.ResourceBase):
collection_key = 'access_rules'
member_key = 'access_rule'
[docs]
@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)
[docs]
@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
[docs]
class UserTokensResource(ks_flask.ResourceBase):
collection_key = 'tokens'
member_key = 'token'
[docs]
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
[docs]
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,)
