keystone.common.controller

Source code for keystone.common.controller

# Copyright 2013 OpenStack Foundation
#
#    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.

import functools
import uuid

from oslo_log import log
from oslo_log import versionutils
import six

from keystone.common import authorization
from keystone.common import driver_hints
from keystone.common import provider_api
from keystone.common import utils
from keystone.common import wsgi
import keystone.conf
from keystone import exception
from keystone.i18n import _


LOG = log.getLogger(__name__)
CONF = keystone.conf.CONF


[docs]def v2_ec2_deprecated(f): @six.wraps(f) def wrapper(*args, **kwargs): deprecated = versionutils.deprecated( what=f.__name__ + ' of the v2 EC2 APIs', as_of=versionutils.deprecated.MITAKA, in_favor_of=('a similar function in the v3 Credential APIs'), remove_in=+7) return deprecated(f) return wrapper()
[docs]def v2_auth_deprecated(f): @six.wraps(f) def wrapper(*args, **kwargs): deprecated = versionutils.deprecated( what=f.__name__ + ' of the v2 Authentication APIs', as_of=versionutils.deprecated.MITAKA, in_favor_of=('a similar function in the v3 Authentication APIs'), remove_in=+7) return deprecated(f) return wrapper()
[docs]def protected(callback=None): """Wrap API calls with role based access controls (RBAC). This handles both the protection of the API parameters as well as any target entities for single-entity API calls. More complex API calls (for example that deal with several different entities) should pass in a callback function, that will be subsequently called to check protection for these multiple entities. This callback function should gather the appropriate entities needed and then call check_protection() in the V3Controller class. """ def wrapper(f): @functools.wraps(f) def inner(self, request, *args, **kwargs): check_function = authorization.check_protection if callback is not None: check_function = callback protected_wrapper( self, f, check_function, request, None, *args, **kwargs) return f(self, request, *args, **kwargs) return inner return wrapper
[docs]def filterprotected(*filters, **callback): """Wrap API list calls with role based access controls (RBAC). This handles both the protection of the API parameters as well as any filters supplied. More complex API list calls (for example that need to examine the contents of an entity referenced by one of the filters) should pass in a callback function, that will be subsequently called to check protection for these multiple entities. This callback function should gather the appropriate entities needed and then call check_protection() in the V3Controller class. """ def _handle_filters(filters, request): target = dict() if filters: for item in filters: if item in request.params: target[item] = request.params[item] LOG.debug('RBAC: Adding query filter params (%s)', ( ', '.join(['%s=%s' % (item, target[item]) for item in target]))) return target def _filterprotected(f): @functools.wraps(f) def wrapper(self, request, **kwargs): filter_attr = _handle_filters(filters, request) check_function = authorization.check_protection if 'callback' in callback and callback['callback'] is not None: # A callback has been specified to load additional target # data, so pass it the formal url params as well as the # list of filters, so it can augment these and then call # the check_protection() method. check_function = callback['callback'] protected_wrapper( self, f, check_function, request, filter_attr, **kwargs) return f(self, request, filters, **kwargs) return wrapper return _filterprotected
# Unified calls for the decorators above. # TODO(ayoung): Continue the refactoring. Always call check_protection # explicitly, by removing the calls to check protection from the callbacks. # Instead, have a call to the callbacks inserted prior to the call to # `check_protection`.
[docs]def protected_wrapper(self, f, check_function, request, filter_attr, *args, **kwargs): request.assert_authenticated() if request.context.is_admin: LOG.warning('RBAC: Bypassing authorization') return prep_info = {'f_name': f.__name__, 'input_attr': kwargs} if (filter_attr): prep_info['filter_attr'] = filter_attr check_function(self, request, prep_info, *args, **kwargs)
[docs]class V2Controller(provider_api.ProviderAPIMixin, wsgi.Application): """Base controller class for Identity API v2."""
[docs] @staticmethod def v3_to_v2_user(ref): """Convert a user_ref from v3 to v2 compatible. * v2.0 users are not domain aware, and should have domain_id removed * v2.0 users expect the use of tenantId instead of default_project_id * v2.0 users have a username attribute * v2.0 remove password_expires_at If ref is a list type, we will iterate through each element and do the conversion. """ def _format_default_project_id(ref): """Convert default_project_id to tenantId for v2 calls.""" default_project_id = ref.pop('default_project_id', None) if default_project_id is not None: ref['tenantId'] = default_project_id elif 'tenantId' in ref: # NOTE(morganfainberg): To avoid v2.0 confusion if somehow a # tenantId property sneaks its way into the extra blob on the # user, we remove it here. If default_project_id is set, we # would override it in either case. del ref['tenantId'] def _normalize_and_filter_user_properties(ref): _format_default_project_id(ref) ref.pop('password_expires_at', None) ref.pop('domain', None) ref.pop('domain_id', None) if 'username' not in ref and 'name' in ref: ref['username'] = ref['name'] return ref if isinstance(ref, dict): return _normalize_and_filter_user_properties(ref) elif isinstance(ref, list): return [_normalize_and_filter_user_properties(x) for x in ref] else: raise ValueError(_('Expected dict or list: %s') % type(ref))
[docs] @staticmethod def v3_to_v2_project(ref): """Convert a project_ref from v3 to v2. * v2.0 projects are not domain aware, and should have domain_id removed * v2.0 projects are not hierarchy aware, and should have parent_id removed This method should only be applied to project_refs being returned from the v2.0 controller(s). If ref is a list type, we will iterate through each element and do the conversion. """ def _filter_project_properties(ref): ref.pop('domain_id', None) ref.pop('parent_id', None) ref.pop('is_domain', None) return ref if isinstance(ref, dict): return _filter_project_properties(ref) elif isinstance(ref, list): return [_filter_project_properties(x) for x in ref] else: raise ValueError(_('Expected dict or list: %s') % type(ref))
[docs]class V3Controller(provider_api.ProviderAPIMixin, wsgi.Application): """Base controller class for Identity API v3. Child classes should set the ``collection_name`` and ``member_name`` class attributes, representing the collection of entities they are exposing to the API. This is required for supporting self-referential links, pagination, etc. Class parameters: * `_public_parameters` - set of parameters that are exposed to the user. Usually used by cls.filter_params() """ collection_name = 'entities' member_name = 'entity' get_member_from_driver = None
[docs] @classmethod def base_url(cls, context, path=None): endpoint = super(V3Controller, cls).base_url(context, 'public') if not path: path = cls.collection_name return '%s/%s/%s' % (endpoint, 'v3', path.lstrip('/'))
[docs] @classmethod def full_url(cls, context, path=None): url = cls.base_url(context, path) if context['environment'].get('QUERY_STRING'): url = '%s?%s' % (url, context['environment']['QUERY_STRING']) return url
[docs] @classmethod def query_filter_is_true(cls, filter_value): """Determine if bool query param is 'True'. We treat this the same way as we do for policy enforcement: {bool_param}=0 is treated as False Any other value is considered to be equivalent to True, including the absence of a value """ if (isinstance(filter_value, six.string_types) and filter_value == '0'): val = False else: val = True return val
@classmethod def _add_self_referential_link(cls, context, ref): ref.setdefault('links', {}) ref['links']['self'] = cls.base_url(context) + '/' + ref['id']
[docs] @classmethod def wrap_member(cls, context, ref): cls._add_self_referential_link(context, ref) return {cls.member_name: ref}
[docs] @classmethod def wrap_collection(cls, context, refs, hints=None): """Wrap a collection, checking for filtering and pagination. Returns the wrapped collection, which includes: - Executing any filtering not already carried out - Truncate to a set limit if necessary - Adds 'self' links in every member - Adds 'next', 'self' and 'prev' links for the whole collection. :param context: the current context, containing the original url path and query string :param refs: the list of members of the collection :param hints: list hints, containing any relevant filters and limit. Any filters already satisfied by managers will have been removed """ # Check if there are any filters in hints that were not # handled by the drivers. The driver will not have paginated or # limited the output if it found there were filters it was unable to # handle. if hints is not None: refs = cls.filter_by_attributes(refs, hints) list_limited, refs = cls.limit(refs, hints) for ref in refs: cls.wrap_member(context, ref) container = {cls.collection_name: refs} container['links'] = { 'next': None, 'self': cls.full_url(context, path=context['path']), 'previous': None} if list_limited: container['truncated'] = True return container
[docs] @classmethod def limit(cls, refs, hints): """Limit a list of entities. The underlying driver layer may have already truncated the collection for us, but in case it was unable to handle truncation we check here. :param refs: the list of members of the collection :param hints: hints, containing, among other things, the limit requested :returns: boolean indicating whether the list was truncated, as well as the list of (truncated if necessary) entities. """ NOT_LIMITED = False LIMITED = True if hints is None or hints.limit is None: # No truncation was requested return NOT_LIMITED, refs if hints.limit.get('truncated', False): # The driver did truncate the list return LIMITED, refs if len(refs) > hints.limit['limit']: # The driver layer wasn't able to truncate it for us, so we must # do it here return LIMITED, refs[:hints.limit['limit']] return NOT_LIMITED, refs
[docs] @classmethod def filter_by_attributes(cls, refs, hints): """Filter a list of references by filter values.""" def _attr_match(ref_attr, val_attr): """Matche attributes allowing for booleans as strings. We test explicitly for a value that defines it as 'False', which also means that the existence of the attribute with no value implies 'True' """ if type(ref_attr) is bool: return ref_attr == utils.attr_as_boolean(val_attr) else: return ref_attr == val_attr def _inexact_attr_match(filter, ref): """Apply an inexact filter to a result dict. :param filter: the filter in question :param ref: the dict to check :returns: True if there is a match """ comparator = filter['comparator'] key = filter['name'] if key in ref: filter_value = filter['value'] target_value = ref[key] if not filter['case_sensitive']: # We only support inexact filters on strings so # it's OK to use lower() filter_value = filter_value.lower() target_value = target_value.lower() if comparator == 'contains': return (filter_value in target_value) elif comparator == 'startswith': return target_value.startswith(filter_value) elif comparator == 'endswith': return target_value.endswith(filter_value) else: # We silently ignore unsupported filters return True return False for filter in hints.filters: if filter['comparator'] == 'equals': attr = filter['name'] value = filter['value'] refs = [r for r in refs if _attr_match( utils.flatten_dict(r).get(attr), value)] else: # It might be an inexact filter refs = [r for r in refs if _inexact_attr_match( filter, r)] return refs
[docs] @classmethod def build_driver_hints(cls, request, supported_filters): """Build list hints based on the context query string. :param request: the current request :param supported_filters: list of filters supported, so ignore any keys in query_dict that are not in this list. """ hints = driver_hints.Hints() if not request.params: return hints for key, value in request.params.items(): # Check if this is an exact filter if supported_filters is None or key in supported_filters: hints.add_filter(key, value) continue # Check if it is an inexact filter for valid_key in supported_filters: # See if this entry in query_dict matches a known key with an # inexact suffix added. If it doesn't match, then that just # means that there is no inexact filter for that key in this # query. if not key.startswith(valid_key + '__'): continue base_key, comparator = key.split('__', 1) # We map the query-style inexact of, for example: # # {'email__contains', 'myISP'} # # into a list directive add filter call parameters of: # # name = 'email' # value = 'myISP' # comparator = 'contains' # case_sensitive = True case_sensitive = True if comparator.startswith('i'): case_sensitive = False comparator = comparator[1:] hints.add_filter(base_key, value, comparator=comparator, case_sensitive=case_sensitive) # NOTE(henry-nash): If we were to support pagination, we would pull any # pagination directives out of the query_dict here, and add them into # the hints list. return hints
def _require_matching_id(self, value, ref): """Ensure the value matches the reference's ID, if any.""" if 'id' in ref and ref['id'] != value: raise exception.ValidationError('Cannot change ID') def _assign_unique_id(self, ref): """Generate and assigns a unique identifier to a reference.""" ref = ref.copy() ref['id'] = uuid.uuid4().hex return ref def _get_domain_id_for_list_request(self, request): """Get the domain_id for a v3 list call. If we running with multiple domain drivers, then the caller must specify a domain_id either as a filter or as part of the token scope. """ if not CONF.identity.domain_specific_drivers_enabled: # We don't need to specify a domain ID in this case return domain_id = request.params.get('domain_id') if domain_id: return domain_id token_ref = authorization.get_token_ref(request.context_dict) if token_ref.domain_scoped: return token_ref.domain_id elif token_ref.project_scoped: return token_ref.project_domain_id elif token_ref.system_scoped: return else: msg = _('No domain information specified as part of list request') LOG.warning(msg) raise exception.Unauthorized(msg) def _get_domain_id_from_token(self, request): """Get the domain_id for a v3 create call. In the case of a v3 create entity call that does not specify a domain ID, the spec says that we should use the domain scoping from the token being used. """ # return if domain scoped if request.context.domain_id: return request.context.domain_id if request.context.is_admin: raise exception.ValidationError( _('You have tried to create a resource using the admin ' 'token. As this token is not within a domain you must ' 'explicitly include a domain for this resource to ' 'belong to.')) # TODO(henry-nash): We should issue an exception here since if # a v3 call does not explicitly specify the domain_id in the # entity, it should be using a domain scoped token. However, # the current tempest heat tests issue a v3 call without this. # This is raised as bug #1283539. Once this is fixed, we # should remove the line below and replace it with an error. # # Ahead of actually changing the code to raise an exception, we # issue a deprecation warning. versionutils.report_deprecated_feature( LOG, 'Not specifying a domain during a create user, group or ' 'project call, and relying on falling back to the ' 'default domain, is deprecated as of Liberty. There is no ' 'plan to remove this compatibility, however, future API ' 'versions may remove this, so please specify the domain ' 'explicitly or use a domain-scoped token.') return CONF.identity.default_domain_id def _normalize_domain_id(self, request, ref): """Fill in domain_id if not specified in a v3 call.""" if not ref.get('domain_id'): ref['domain_id'] = self._get_domain_id_from_token(request) return ref
[docs] @staticmethod def filter_domain_id(ref): """Override v2 filter to let domain_id out for v3 calls.""" return ref
[docs] def check_protection(self, request, prep_info, target_attr=None): """Provide call protection for complex target attributes. As well as including the standard parameters from the original API call (which is passed in prep_info), this call will add in any additional entities or attributes (passed in target_attr), so that they can be referenced by policy rules. """ authorization.check_protection(self, request, prep_info, target_attr)
[docs] @classmethod def filter_params(cls, ref): """Remove unspecified parameters from the dictionary. This function removes unspecified parameters from the dictionary. This method checks only root-level keys from a ref dictionary. :param ref: a dictionary representing deserialized response to be serialized """ ref_keys = set(ref.keys()) blocked_keys = ref_keys - cls._public_parameters for blocked_param in blocked_keys: del ref[blocked_param] return ref
Creative Commons Attribution 3.0 License

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