# 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/OS-TRUST
# TODO(morgan): Deprecate /v3/OS-TRUST/trusts path in favour of /v3/trusts.
# /v3/OS-TRUST should remain indefinitely.
import http.client
import flask
import flask_restful
from oslo_log import log
from oslo_policy import _checks as op_checks
from keystone.api._shared import delegation
from keystone.api._shared import json_home_relations
from keystone.api import validation
from keystone.common import authorization
from keystone.common import context
from keystone.common import json_home
from keystone.common import provider_api
from keystone.common import rbac_enforcer
from keystone.common.rbac_enforcer import policy
from keystone.common import utils
import keystone.conf
from keystone import exception
from keystone.i18n import _
from keystone.server import flask as ks_flask
from keystone.trust import schema
LOG = log.getLogger(__name__)
CONF = keystone.conf.CONF
ENFORCER = rbac_enforcer.RBACEnforcer
PROVIDERS = provider_api.ProviderAPIs
_build_resource_relation = json_home_relations.os_trust_resource_rel_func
_build_parameter_relation = json_home_relations.os_trust_parameter_rel_func
TRUST_ID_PARAMETER_RELATION = _build_parameter_relation(
parameter_name='trust_id'
)
def _check_delegated_token():
"""Block delegated tokens from all trust operations.
Delegated tokens (application credentials, OAuth1 access tokens, EC2
credentials) are narrower-scope grants than a normal user session.
Allowing them to read or manage trusts would permit a compromised
delegation to enumerate or manipulate the trust delegation chain -- or,
for creation, mint a new trust delegating roles the trustor holds but
the delegation itself was never scoped to, since trust role validation
checks the trustor's full role assignments, not the requesting token's
own scoped roles -- expanding its effective scope well beyond what it
was issued for.
Trust-scoped tokens are deliberately NOT blocked here: managing trusts
with a trust-scoped token is the redelegation feature working as
designed (a trustee creating a further, narrower trust from one they
were delegated), and trustee self-service reads of their own trusts.
A trust-scoped token whose underlying auth method is itself delegated
(e.g. an EC2 credential's blob embeds a trust_id -- see
keystone.api.credentials._assign_unique_id) is still caught below,
since token.methods reflects that underlying method regardless of the
trust scoping.
application_credential tokens have a documented, opt-in escape hatch
(allow_insecure_application_credential_trust_escalation, LP#2150089)
for workflows such as Heat that need it. OAuth1 and EC2 credentials
have no such use case and are blocked unconditionally, regardless of
that option.
"""
token = flask.request.environ.get(authorization.AUTH_CONTEXT_ENV, {}).get(
'token'
)
if not token:
return
if not delegation.is_delegated_method(token):
return
if (
'application_credential' in token.methods
and CONF.security_compliance.allow_insecure_application_credential_trust_escalation
):
return
raise exception.ForbiddenAction(
action=_('Delegated tokens cannot manage trusts.')
)
def _build_trust_target_enforcement():
target = {}
# NOTE(cmurphy) unlike other APIs, in the event the trust doesn't exist or
# has 0 remaining uses, we actually do expect it to return a 404 and not a
# 403, so don't catch NotFound here (lp#1840288)
target['trust'] = PROVIDERS.trust_api.get_trust(
flask.request.view_args.get('trust_id')
)
return target
def _trustor_trustee_only(trust):
user_id = flask.request.environ.get(context.REQUEST_CONTEXT_ENV).user_id
if user_id not in [
trust.get('trustee_user_id'),
trust.get('trustor_user_id'),
]:
raise exception.ForbiddenAction(
action=_('Requested user has no relation to this trust')
)
def _normalize_trust_expires_at(trust):
# correct isotime
if trust.get('expires_at') is not None:
trust['expires_at'] = utils.isotime(
trust['expires_at'], subsecond=True
)
def _normalize_trust_roles(trust):
# fill in role data
trust_full_roles = []
for trust_role in trust.get('roles', []):
trust_role = trust_role['id']
try:
matching_role = PROVIDERS.role_api.get_role(trust_role)
full_role = ks_flask.ResourceBase.wrap_member(
matching_role, collection_name='roles', member_name='role'
)
trust_full_roles.append(full_role['role'])
except exception.RoleNotFound:
pass
trust['roles'] = trust_full_roles
trust['roles_links'] = {
'self': ks_flask.base_url(path='/{}/roles'.format(trust['id'])),
'next': None,
'previous': None,
}
[docs]
class TrustResourceBase(ks_flask.ResourceBase):
def _check_unrestricted(self):
_check_delegated_token()
[docs]
class TrustsResource(TrustResourceBase):
collection_key = 'trusts'
member_key = 'trust'
api_prefix = '/OS-TRUST'
json_home_resource_rel_func = _build_resource_relation
json_home_parameter_rel_func = _build_parameter_relation
def _find_redelegated_trust(self):
# Check if delegated via trust
redelegated_trust = None
if self.oslo_context.is_delegated_auth:
src_trust_id = self.oslo_context.trust_id
if not src_trust_id:
action = _('Redelegation allowed for delegated by trust only')
raise exception.ForbiddenAction(action=action)
redelegated_trust = PROVIDERS.trust_api.get_trust(src_trust_id)
return redelegated_trust
@staticmethod
def _parse_expiration_date(expiration_date):
if expiration_date is not None:
return utils.parse_expiration_date(expiration_date)
return None
def _require_trustor_has_role_in_project(self, trust):
trustor_roles = self._get_trustor_roles(trust)
for trust_role in trust['roles']:
matching_roles = [
x for x in trustor_roles if x == trust_role['id']
]
if not matching_roles:
raise exception.RoleNotFound(role_id=trust_role['id'])
def _get_trustor_roles(self, trust):
original_trust = trust.copy()
while original_trust.get('redelegated_trust_id'):
original_trust = PROVIDERS.trust_api.get_trust(
original_trust['redelegated_trust_id']
)
if (trust.get('project_id')) not in [None, '']:
# Check project exists.
PROVIDERS.resource_api.get_project(trust['project_id'])
# Get a list of roles including any domain specific roles
assignment_list = PROVIDERS.assignment_api.list_role_assignments(
user_id=original_trust['trustor_user_id'],
project_id=original_trust['project_id'],
effective=True,
strip_domain_roles=False,
)
return list({x['role_id'] for x in assignment_list})
else:
return []
def _normalize_role_list(self, trust_roles):
roles = []
for role in trust_roles:
if role.get('id'):
roles.append({'id': role['id']})
else:
roles.append(
PROVIDERS.role_api.get_unique_role_by_name(role['name'])
)
return roles
[docs]
@validation.request_query_schema(schema.trust_index_request_query)
@validation.response_body_schema(schema.trust_index_response_body)
def get(self):
"""Dispatch for LIST trusts.
GET /v3/OS-TRUST/trusts
"""
trustor_user_id = flask.request.args.get('trustor_user_id')
trustee_user_id = flask.request.args.get('trustee_user_id')
if trustor_user_id:
target = {'trust': {'trustor_user_id': trustor_user_id}}
ENFORCER.enforce_call(
action='identity:list_trusts_for_trustor', target_attr=target
)
elif trustee_user_id:
target = {'trust': {'trustee_user_id': trustee_user_id}}
ENFORCER.enforce_call(
action='identity:list_trusts_for_trustee', target_attr=target
)
else:
ENFORCER.enforce_call(action='identity:list_trusts')
_check_delegated_token()
trusts = []
# NOTE(cmurphy) As of Train, the default policies enforce the
# identity:list_trusts rule and there are new policies in-code to
# enforce identity:list_trusts_for_trustor and
# identity:list_trusts_for_trustee. However, in case the
# identity:list_trusts rule has been locally overridden by the default
# that would have been produced by the sample config, we need to
# enforce it again and warn that the behavior is changing.
rules = policy._ENFORCER._enforcer.rules.get('identity:list_trusts')
# rule check_str is ""
if isinstance(rules, op_checks.TrueCheck):
LOG.warning(
"The policy check string for rule \"identity:list_trusts\" "
"has been overridden to \"always true\". In the next release, "
"this will cause the \"identity:list_trusts\" action to be "
"fully permissive as hardcoded enforcement will be removed. "
"To correct this issue, either stop overriding the "
"\"identity:list_trusts\" rule in config to accept the "
"defaults, or explicitly set a rule that is not empty."
)
if not flask.request.args:
# NOTE(morgan): Admin can list all trusts.
ENFORCER.enforce_call(action='admin_required')
if not flask.request.args:
trusts += PROVIDERS.trust_api.list_trusts()
elif trustor_user_id:
trusts += PROVIDERS.trust_api.list_trusts_for_trustor(
trustor_user_id
)
elif trustee_user_id:
trusts += PROVIDERS.trust_api.list_trusts_for_trustee(
trustee_user_id
)
for trust in trusts:
# get_trust returns roles, list_trusts does not
# It seems in some circumstances, roles does not
# exist in the query response, so check first
if 'roles' in trust:
del trust['roles']
if trust.get('expires_at') is not None:
trust['expires_at'] = utils.isotime(
trust['expires_at'], subsecond=True
)
return self.wrap_collection(trusts)
[docs]
@validation.request_body_schema(schema.trust_create_request_body)
@validation.response_body_schema(schema.trust_response_body)
def post(self):
"""Create a new trust.
The User creating the trust must be the trustor.
POST /v3/OS-TRUST/trusts
"""
trust = self.request_body_json.get('trust', {})
ENFORCER.enforce_call(
action='identity:create_trust', target_attr={'trust': trust}
)
self._check_unrestricted()
if trust.get('project_id') and not trust.get('roles'):
action = _('At least one role should be specified')
raise exception.ForbiddenAction(action=action)
if self.oslo_context.user_id != trust.get('trustor_user_id'):
action = _("The authenticated user should match the trustor")
raise exception.ForbiddenAction(action=action)
# Ensure the trustee exists
PROVIDERS.identity_api.get_user(trust['trustee_user_id'])
# Normalize roles
trust['roles'] = self._normalize_role_list(trust.get('roles', []))
self._require_trustor_has_role_in_project(trust)
trust['expires_at'] = self._parse_expiration_date(
trust.get('expires_at')
)
trust = self._assign_unique_id(trust)
redelegated_trust = self._find_redelegated_trust()
return_trust = PROVIDERS.trust_api.create_trust(
trust_id=trust['id'],
trust=trust,
roles=trust['roles'],
redelegated_trust=redelegated_trust,
initiator=self.audit_initiator,
)
_normalize_trust_expires_at(return_trust)
_normalize_trust_roles(return_trust)
return self.wrap_member(return_trust), http.client.CREATED
[docs]
class TrustResource(TrustResourceBase):
collection_key = 'trusts'
member_key = 'trust'
api_prefix = '/OS-TRUST'
json_home_resource_rel_func = _build_resource_relation
json_home_parameter_rel_func = _build_parameter_relation
[docs]
@validation.request_query_schema(schema.trust_request_query)
@validation.response_body_schema(schema.trust_response_body)
def get(self, trust_id):
"""Get trust.
GET /v3/OS-TRUST/trusts/{trust_id}
"""
ENFORCER.enforce_call(
action='identity:get_trust',
build_target=_build_trust_target_enforcement,
)
_check_delegated_token()
# NOTE(cmurphy) look up trust before doing is_admin authorization - to
# maintain the API contract, we expect a missing trust to raise a 404
# before we get to enforcement (lp#1840288)
trust = PROVIDERS.trust_api.get_trust(trust_id)
if self.oslo_context.is_admin:
# policies are not loaded for the is_admin context, so need to
# block access here
raise exception.ForbiddenAction(
action=_('Requested user has no relation to this trust')
)
# NOTE(cmurphy) As of Train, the default policies enforce the
# identity:get_trust rule. However, in case the
# identity:get_trust rule has been locally overridden by the
# default that would have been produced by the sample config, we need
# to enforce it again and warn that the behavior is changing.
rules = policy._ENFORCER._enforcer.rules.get('identity:get_trust')
# rule check_str is ""
if isinstance(rules, op_checks.TrueCheck):
LOG.warning(
"The policy check string for rule \"identity:get_trust\" "
"has been overridden to \"always true\". In the next release, "
"this will cause the"
"\"identity:get_trust\" action to "
"be fully permissive as hardcoded enforcement will be "
"removed. To correct this issue, either stop overriding the "
"\"identity:get_trust\" rule in config to accept the "
"defaults, or explicitly set a rule that is not empty."
)
_trustor_trustee_only(trust)
_normalize_trust_expires_at(trust)
_normalize_trust_roles(trust)
return self.wrap_member(trust)
[docs]
def delete(self, trust_id):
"""Delete trust.
DELETE /v3/OS-TRUST/trusts/{trust_id}
"""
ENFORCER.enforce_call(
action='identity:delete_trust',
build_target=_build_trust_target_enforcement,
)
self._check_unrestricted()
# NOTE(cmurphy) As of Train, the default policies enforce the
# identity:delete_trust rule. However, in case the
# identity:delete_trust rule has been locally overridden by the
# default that would have been produced by the sample config, we need
# to enforce it again and warn that the behavior is changing.
rules = policy._ENFORCER._enforcer.rules.get('identity:delete_trust')
# rule check_str is ""
if isinstance(rules, op_checks.TrueCheck):
LOG.warning(
"The policy check string for rule \"identity:delete_trust\" "
"has been overridden to \"always true\". In the next release, "
"this will cause the"
"\"identity:delete_trust\" action to "
"be fully permissive as hardcoded enforcement will be "
"removed. To correct this issue, either stop overriding the "
"\"identity:delete_trust\" rule in config to accept the "
"defaults, or explicitly set a rule that is not empty."
)
trust = PROVIDERS.trust_api.get_trust(trust_id)
if (
self.oslo_context.user_id != trust.get('trustor_user_id')
and not self.oslo_context.is_admin
):
action = _('Only admin or trustor can delete a trust')
raise exception.ForbiddenAction(action=action)
PROVIDERS.trust_api.delete_trust(
trust_id, initiator=self.audit_initiator
)
return '', http.client.NO_CONTENT
# NOTE(morgan): Since this Resource is not being used with the automatic
# URL additions and does not have a collection key/member_key, we use
# the flask-restful Resource, not the keystone ResourceBase
[docs]
class RolesForTrustListResource(flask_restful.Resource):
@property
def oslo_context(self):
return flask.request.environ.get(context.REQUEST_CONTEXT_ENV, None)
[docs]
def get(self, trust_id):
ENFORCER.enforce_call(
action='identity:list_roles_for_trust',
build_target=_build_trust_target_enforcement,
)
# NOTE(morgan): This duplicates a little of the .get_trust from the
# main resource, as it needs some of the same logic. However, due to
# how flask-restful works, this should be fully encapsulated
if self.oslo_context.is_admin:
# policies are not loaded for the is_admin context, so need to
# block access here
raise exception.ForbiddenAction(
action=_('Requested user has no relation to this trust')
)
_check_delegated_token()
trust = PROVIDERS.trust_api.get_trust(trust_id)
# NOTE(cmurphy) As of Train, the default policies enforce the
# identity:list_roles_for_trust rule. However, in case the
# identity:list_roles_for_trust rule has been locally overridden by the
# default that would have been produced by the sample config, we need
# to enforce it again and warn that the behavior is changing.
rules = policy._ENFORCER._enforcer.rules.get(
'identity:list_roles_for_trust'
)
# rule check_str is ""
if isinstance(rules, op_checks.TrueCheck):
LOG.warning(
"The policy check string for rule "
"\"identity:list_roles_for_trust\" has been overridden to "
"\"always true\". In the next release, this will cause the "
"\"identity:list_roles_for_trust\" action to be fully "
"permissive as hardcoded enforcement will be removed. To "
"correct this issue, either stop overriding the "
"\"identity:get_trust\" rule in config to accept the "
"defaults, or explicitly set a rule that is not empty."
)
_trustor_trustee_only(trust)
_normalize_trust_expires_at(trust)
_normalize_trust_roles(trust)
return {'roles': trust['roles'], 'links': trust['roles_links']}
# NOTE(morgan): Since this Resource is not being used with the automatic
# URL additions and does not have a collection key/member_key, we use
# the flask-restful Resource, not the keystone ResourceBase
[docs]
class RoleForTrustResource(flask_restful.Resource):
@property
def oslo_context(self):
return flask.request.environ.get(context.REQUEST_CONTEXT_ENV, None)
[docs]
def get(self, trust_id, role_id):
"""Get a role that has been assigned to a trust."""
ENFORCER.enforce_call(
action='identity:get_role_for_trust',
build_target=_build_trust_target_enforcement,
)
if self.oslo_context.is_admin:
# policies are not loaded for the is_admin context, so need to
# block access here
raise exception.ForbiddenAction(
action=_('Requested user has no relation to this trust')
)
_check_delegated_token()
trust = PROVIDERS.trust_api.get_trust(trust_id)
# NOTE(cmurphy) As of Train, the default policies enforce the
# identity:get_role_for_trust rule. However, in case the
# identity:get_role_for_trust rule has been locally overridden by the
# default that would have been produced by the sample config, we need
# to enforce it again and warn that the behavior is changing.
rules = policy._ENFORCER._enforcer.rules.get(
'identity:get_role_for_trust'
)
# rule check_str is ""
if isinstance(rules, op_checks.TrueCheck):
LOG.warning(
"The policy check string for rule "
"\"identity:get_role_for_trust\" has been overridden to "
"\"always true\". In the next release, this will cause the "
"\"identity:get_role_for_trust\" action to be fully "
"permissive as hardcoded enforcement will be removed. To "
"correct this issue, either stop overriding the "
"\"identity:get_role_for_trust\" rule in config to accept the "
"defaults, or explicitly set a rule that is not empty."
)
_trustor_trustee_only(trust)
if not any(role['id'] == role_id for role in trust['roles']):
raise exception.RoleNotFound(role_id=role_id)
role = PROVIDERS.role_api.get_role(role_id)
return ks_flask.ResourceBase.wrap_member(
role, collection_name='roles', member_name='role'
)
[docs]
class TrustAPI(ks_flask.APIBase):
_name = 'trusts'
_import_name = __name__
resource_mapping = [
ks_flask.construct_resource_map(
resource=TrustsResource,
url='/trusts',
resource_kwargs={},
rel="trusts",
path_vars=None,
resource_relation_func=_build_resource_relation,
),
ks_flask.construct_resource_map(
resource=TrustResource,
url='/trusts/<string:trust_id>',
resource_kwargs={},
rel="trust",
path_vars={'trust_id': TRUST_ID_PARAMETER_RELATION},
resource_relation_func=_build_resource_relation,
),
ks_flask.construct_resource_map(
resource=RolesForTrustListResource,
url='/trusts/<string:trust_id>/roles',
resource_kwargs={},
rel='trust_roles',
path_vars={'trust_id': TRUST_ID_PARAMETER_RELATION},
resource_relation_func=_build_resource_relation,
),
ks_flask.construct_resource_map(
resource=RoleForTrustResource,
url='/trusts/<string:trust_id>/roles/<string:role_id>',
resource_kwargs={},
rel='trust_role',
path_vars={
'trust_id': TRUST_ID_PARAMETER_RELATION,
'role_id': json_home.Parameters.ROLE_ID,
},
resource_relation_func=_build_resource_relation,
),
]
_api_url_prefix = '/OS-TRUST'
APIs = (TrustAPI,)
