OpenStack Identity (Keystone)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

trusts.py 13KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320
  1. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  2. # not use this file except in compliance with the License. You may obtain
  3. # a copy of the License at
  4. #
  5. # http://www.apache.org/licenses/LICENSE-2.0
  6. #
  7. # Unless required by applicable law or agreed to in writing, software
  8. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  9. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  10. # License for the specific language governing permissions and limitations
  11. # under the License.
  12. # This file handles all flask-restful resources for /v3/OS-TRUST
  13. # TODO(morgan): Deprecate /v3/OS-TRUST/trusts path in favour of /v3/trusts.
  14. # /v3/OS-TRUST should remain indefinitely.
  15. import flask
  16. import flask_restful
  17. from six.moves import http_client
  18. from keystone.api._shared import json_home_relations
  19. from keystone.common import context
  20. from keystone.common import json_home
  21. from keystone.common import provider_api
  22. from keystone.common import rbac_enforcer
  23. from keystone.common import utils
  24. from keystone.common import validation
  25. from keystone import exception
  26. from keystone.i18n import _
  27. from keystone.server import flask as ks_flask
  28. from keystone.trust import schema
  29. ENFORCER = rbac_enforcer.RBACEnforcer
  30. PROVIDERS = provider_api.ProviderAPIs
  31. _build_resource_relation = json_home_relations.os_trust_resource_rel_func
  32. _build_parameter_relation = json_home_relations.os_trust_parameter_rel_func
  33. TRUST_ID_PARAMETER_RELATION = _build_parameter_relation(
  34. parameter_name='trust_id')
  35. def _trustor_trustee_only(trust):
  36. user_id = flask.request.environ.get(context.REQUEST_CONTEXT_ENV).user_id
  37. if user_id not in [trust.get('trustee_user_id'),
  38. trust.get('trustor_user_id')]:
  39. raise exception.ForbiddenAction(
  40. action=_('Requested user has no relation to this trust'))
  41. def _normalize_trust_expires_at(trust):
  42. # correct isotime
  43. if trust.get('expires_at') is not None:
  44. trust['expires_at'] = utils.isotime(trust['expires_at'],
  45. subsecond=True)
  46. def _normalize_trust_roles(trust):
  47. # fill in role data
  48. trust_full_roles = []
  49. for trust_role in trust.get('roles', []):
  50. trust_role = trust_role['id']
  51. try:
  52. matching_role = PROVIDERS.role_api.get_role(trust_role)
  53. full_role = ks_flask.ResourceBase.wrap_member(
  54. matching_role, collection_name='roles', member_name='role')
  55. trust_full_roles.append(full_role['role'])
  56. except exception.RoleNotFound:
  57. pass
  58. trust['roles'] = trust_full_roles
  59. trust['roles_links'] = {
  60. 'self': ks_flask.base_url(path='/%s/roles' % trust['id']),
  61. 'next': None,
  62. 'previous': None}
  63. class TrustResource(ks_flask.ResourceBase):
  64. collection_key = 'trusts'
  65. member_key = 'trust'
  66. api_prefix = '/OS-TRUST'
  67. json_home_resource_rel_func = _build_resource_relation
  68. json_home_parameter_rel_func = _build_parameter_relation
  69. def _check_unrestricted(self):
  70. token = self.auth_context['token']
  71. if 'application_credential' in token.methods:
  72. if not token.application_credential['unrestricted']:
  73. action = _("Using method 'application_credential' is not "
  74. "allowed for managing trusts.")
  75. raise exception.ForbiddenAction(action=action)
  76. def _find_redelegated_trust(self):
  77. # Check if delegated via trust
  78. redelegated_trust = None
  79. if self.oslo_context.is_delegated_auth:
  80. src_trust_id = self.oslo_context.trust_id
  81. if not src_trust_id:
  82. action = _('Redelegation allowed for delegated by trust only')
  83. raise exception.ForbiddenAction(action=action)
  84. redelegated_trust = PROVIDERS.trust_api.get_trust(src_trust_id)
  85. return redelegated_trust
  86. @staticmethod
  87. def _parse_expiration_date(expiration_date):
  88. if expiration_date is not None:
  89. return utils.parse_expiration_date(expiration_date)
  90. return None
  91. def _require_trustor_has_role_in_project(self, trust):
  92. trustor_roles = self._get_trustor_roles(trust)
  93. for trust_role in trust['roles']:
  94. matching_roles = [x for x in trustor_roles
  95. if x == trust_role['id']]
  96. if not matching_roles:
  97. raise exception.RoleNotFound(role_id=trust_role['id'])
  98. def _get_trustor_roles(self, trust):
  99. original_trust = trust.copy()
  100. while original_trust.get('redelegated_trust_id'):
  101. original_trust = PROVIDERS.trust_api.get_trust(
  102. original_trust['redelegated_trust_id'])
  103. if not ((trust.get('project_id')) in [None, '']):
  104. # Check project exists.
  105. PROVIDERS.resource_api.get_project(trust['project_id'])
  106. # Get a list of roles including any domain specific roles
  107. assignment_list = PROVIDERS.assignment_api.list_role_assignments(
  108. user_id=original_trust['trustor_user_id'],
  109. project_id=original_trust['project_id'],
  110. effective=True, strip_domain_roles=False)
  111. return list({x['role_id'] for x in assignment_list})
  112. else:
  113. return []
  114. def _normalize_role_list(self, trust_roles):
  115. roles = []
  116. for role in trust_roles:
  117. if role.get('id'):
  118. roles.append({'id': role['id']})
  119. else:
  120. roles.append(
  121. PROVIDERS.role_api.get_unique_role_by_name(role['name']))
  122. return roles
  123. def _get_trust(self, trust_id):
  124. ENFORCER.enforce_call(action='identity:get_trust')
  125. trust = PROVIDERS.trust_api.get_trust(trust_id)
  126. _trustor_trustee_only(trust)
  127. _normalize_trust_expires_at(trust)
  128. _normalize_trust_roles(trust)
  129. return self.wrap_member(trust)
  130. def _list_trusts(self):
  131. ENFORCER.enforce_call(action='identity:list_trusts')
  132. trusts = []
  133. trustor_user_id = flask.request.args.get('trustor_user_id')
  134. trustee_user_id = flask.request.args.get('trustee_user_id')
  135. if not flask.request.args:
  136. # NOTE(morgan): Admin can list all trusts.
  137. ENFORCER.enforce_call(action='admin_required')
  138. trusts += PROVIDERS.trust_api.list_trusts()
  139. # TODO(morgan): Convert the trustor/trustee checks into policy
  140. # checkstr we can enforce on. This is duplication of code
  141. # behavior/design as the OS-TRUST controller for ease of review/
  142. # comparison of previous code. Minor optimizations [checks before db
  143. # hits] have been done.
  144. action = _('Cannot list trusts for another user')
  145. if trustor_user_id:
  146. if trustor_user_id != self.oslo_context.user_id:
  147. raise exception.ForbiddenAction(action=action)
  148. if trustee_user_id:
  149. if trustee_user_id != self.oslo_context.user_id:
  150. raise exception.ForbiddenAction(action=action)
  151. trusts += PROVIDERS.trust_api.list_trusts_for_trustor(trustor_user_id)
  152. trusts += PROVIDERS.trust_api.list_trusts_for_trustee(trustee_user_id)
  153. for trust in trusts:
  154. # get_trust returns roles, list_trusts does not
  155. # It seems in some circumstances, roles does not
  156. # exist in the query response, so check first
  157. if 'roles' in trust:
  158. del trust['roles']
  159. if trust.get('expires_at') is not None:
  160. trust['expires_at'] = utils.isotime(trust['expires_at'],
  161. subsecond=True)
  162. return self.wrap_collection(trusts)
  163. def get(self, trust_id=None):
  164. """Dispatch for GET/HEAD or LIST trusts."""
  165. if trust_id is not None:
  166. return self._get_trust(trust_id=trust_id)
  167. else:
  168. return self._list_trusts()
  169. def post(self):
  170. """Create a new trust.
  171. The User creating the trust must be the trustor.
  172. """
  173. ENFORCER.enforce_call(action='identity:create_trust')
  174. trust = self.request_body_json.get('trust', {})
  175. validation.lazy_validate(schema.trust_create, trust)
  176. self._check_unrestricted()
  177. if trust.get('project_id') and not trust.get('roles'):
  178. action = _('At least one role should be specified')
  179. raise exception.ForbiddenAction(action=action)
  180. if self.oslo_context.user_id != trust.get('trustor_user_id'):
  181. action = _("The authenticated user should match the trustor")
  182. raise exception.ForbiddenAction(action=action)
  183. # Ensure the trustee exists
  184. PROVIDERS.identity_api.get_user(trust['trustee_user_id'])
  185. # Normalize roles
  186. trust['roles'] = self._normalize_role_list(trust.get('roles', []))
  187. self._require_trustor_has_role_in_project(trust)
  188. trust['expires_at'] = self._parse_expiration_date(
  189. trust.get('expires_at'))
  190. trust = self._assign_unique_id(trust)
  191. redelegated_trust = self._find_redelegated_trust()
  192. return_trust = PROVIDERS.trust_api.create_trust(
  193. trust_id=trust['id'],
  194. trust=trust,
  195. roles=trust['roles'],
  196. redelegated_trust=redelegated_trust,
  197. initiator=self.audit_initiator)
  198. _normalize_trust_expires_at(return_trust)
  199. _normalize_trust_roles(return_trust)
  200. return self.wrap_member(return_trust), http_client.CREATED
  201. def delete(self, trust_id):
  202. ENFORCER.enforce_call(action='identity:delete_trust')
  203. self._check_unrestricted()
  204. trust = PROVIDERS.trust_api.get_trust(trust_id)
  205. # TODO(morgan): convert this check to an oslo_policy checkstr that
  206. # can be referenced/enforced on.
  207. if (self.oslo_context.user_id != trust.get('trustor_user_id') and
  208. not self.oslo_context.is_admin):
  209. action = _('Only admin or trustor can delete a trust')
  210. raise exception.ForbiddenAction(action=action)
  211. PROVIDERS.trust_api.delete_trust(trust_id,
  212. initiator=self.audit_initiator)
  213. return '', http_client.NO_CONTENT
  214. # NOTE(morgan): Since this Resource is not being used with the automatic
  215. # URL additions and does not have a collection key/member_key, we use
  216. # the flask-restful Resource, not the keystone ResourceBase
  217. class RolesForTrustListResource(flask_restful.Resource):
  218. def get(self, trust_id):
  219. ENFORCER.enforce_call(action='identity:list_roles_for_trust')
  220. # NOTE(morgan): This duplicates a little of the .get_trust from the
  221. # main resource, as it needs some of the same logic. However, due to
  222. # how flask-restful works, this should be fully encapsulated
  223. trust = PROVIDERS.trust_api.get_trust(trust_id)
  224. _trustor_trustee_only(trust)
  225. _normalize_trust_expires_at(trust)
  226. _normalize_trust_roles(trust)
  227. return {'roles': trust['roles'],
  228. 'links': trust['roles_links']}
  229. # NOTE(morgan): Since this Resource is not being used with the automatic
  230. # URL additions and does not have a collection key/member_key, we use
  231. # the flask-restful Resource, not the keystone ResourceBase
  232. class RoleForTrustResource(flask_restful.Resource):
  233. def get(self, trust_id, role_id):
  234. """Get a role that has been assigned to a trust."""
  235. ENFORCER.enforce_call(action='identity:get_role_for_trust')
  236. trust = PROVIDERS.trust_api.get_trust(trust_id)
  237. _trustor_trustee_only(trust)
  238. if not any(role['id'] == role_id for role in trust['roles']):
  239. raise exception.RoleNotFound(role_id=role_id)
  240. role = PROVIDERS.role_api.get_role(role_id)
  241. return ks_flask.ResourceBase.wrap_member(role, collection_name='roles',
  242. member_name='role')
  243. class TrustAPI(ks_flask.APIBase):
  244. _name = 'trusts'
  245. _import_name = __name__
  246. resources = [TrustResource]
  247. resource_mapping = [
  248. ks_flask.construct_resource_map(
  249. resource=RolesForTrustListResource,
  250. url='/trusts/<string:trust_id>/roles',
  251. resource_kwargs={},
  252. rel='trust_roles',
  253. path_vars={
  254. 'trust_id': TRUST_ID_PARAMETER_RELATION},
  255. resource_relation_func=_build_resource_relation),
  256. ks_flask.construct_resource_map(
  257. resource=RoleForTrustResource,
  258. url='/trusts/<string:trust_id>/roles/<string:role_id>',
  259. resource_kwargs={},
  260. rel='trust_role',
  261. path_vars={
  262. 'trust_id': TRUST_ID_PARAMETER_RELATION,
  263. 'role_id': json_home.Parameters.ROLE_ID},
  264. resource_relation_func=_build_resource_relation),
  265. ]
  266. _api_url_prefix = '/OS-TRUST'
  267. APIs = (TrustAPI,)