# Copyright 2017 AT&T Corporation. # All Rights Reserved. # # 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 collections import copy import glob import os from oslo_log import log as logging from oslo_policy import policy import stevedore from tempest import clients from tempest.common import credentials_factory as credentials from tempest import config from patrole_tempest_plugin.rbac_authority import RbacAuthority from patrole_tempest_plugin import rbac_exceptions CONF = config.CONF LOG = logging.getLogger(__name__) class PolicyAuthority(RbacAuthority): """A class that uses ``oslo.policy`` for validating RBAC.""" def __init__(self, project_id, user_id, service, extra_target_data=None): """Initialization of Policy Authority class. Validates whether a test role can perform a policy action by querying ``oslo.policy`` with necessary test data. If a policy file does not exist, checks whether the policy file is registered as a namespace under "oslo.policy.policies". Nova, for example, doesn't use a policy file by default; its policies are implemented in code and registered as "nova" under "oslo.policy.policies". If the policy file is not found in either code or in a policy file, then an exception is raised. Additionally, if a custom policy file exists along with the default policy in code implementation, the custom policy is prioritized. :param uuid project_id: project_id of object performing API call :param uuid user_id: user_id of object performing API call :param string service: service of the policy file :param dict extra_target_data: dictionary containing additional object data needed by oslo.policy to validate generic checks Example: .. code-block:: python # Below is the default policy implementation in code, defined in # a service like Nova. test_policies = [ policy.DocumentedRuleDefault( 'service:test_rule', base.RULE_ADMIN_OR_OWNER, "This is a description for a test policy", [ { 'method': 'POST', 'path': '/path/to/test/resource' } ]), 'service:another_test_rule', base.RULE_ADMIN_OR_OWNER, "This is a description for another test policy", [ { 'method': 'GET', 'path': '/path/to/test/resource' } ]), ] .. code-block:: yaml # Below is the custom override of the default policy in a YAML # policy file. Note that the default rule is "rule:admin_or_owner" # and the custom rule is "rule:admin_api". The `PolicyAuthority` # class will use the "rule:admin_api" definition for this policy # action. "service:test_rule" : "rule:admin_api" # Note below that no override is provided for # "service:another_test_rule", which means that the default policy # rule is used: "rule:admin_or_owner". """ if extra_target_data is None: extra_target_data = {} self.service = self.validate_service(service) # Prioritize dynamically searching for policy files over relying on # deprecated service-specific policy file locations. if CONF.patrole.custom_policy_files: self.discover_policy_files() self.rules = self.get_rules() self.project_id = project_id self.user_id = user_id self.extra_target_data = extra_target_data @classmethod def validate_service(cls, service): """Validate whether the service passed to ``__init__`` exists.""" service = service.lower().strip() if service else None # Cache the list of available services in memory to avoid needlessly # doing an API call every time. if not hasattr(cls, 'available_services'): admin_mgr = clients.Manager( credentials.get_configured_admin_credentials()) services_client = (admin_mgr.identity_services_v3_client if CONF.identity_feature_enabled.api_v3 else admin_mgr.identity_services_client) services = services_client.list_services()['services'] cls.available_services = [s['name'] for s in services] if not service or service not in cls.available_services: LOG.debug("%s is NOT a valid service.", service) raise rbac_exceptions.RbacInvalidServiceException( "%s is NOT a valid service." % service) return service @classmethod def discover_policy_files(cls): """Dynamically discover the policy file for each service in ``cls.available_services``. Pick all candidate paths found out of the potential paths in ``[patrole] custom_policy_files``. """ if not hasattr(cls, 'policy_files'): cls.policy_files = collections.defaultdict(list) for service in cls.available_services: for candidate_path in CONF.patrole.custom_policy_files: path = candidate_path % service for filename in glob.iglob(path): if os.path.isfile(filename): cls.policy_files[service].append(filename) def allowed(self, rule_name, roles): """Checks if a given rule in a policy is allowed with given role. :param string rule_name: Policy name to pass to``oslo.policy``. :param List[string] roles: List of roles to validate for authorization. :raises RbacParsingException: If ``rule_name`` does not exist in the cloud (in policy file or among registered in-code policy defaults). """ is_admin_context = self._is_admin_context(roles) is_allowed = self._allowed( access=self._get_access_token(roles), apply_rule=rule_name, is_admin=is_admin_context) return is_allowed def get_rules(self): rules = policy.Rules() # Check whether policy file exists and attempt to read it. for path in self.policy_files[self.service]: try: with open(path, 'r') as fp: for k, v in policy.Rules.load(fp.read()).items(): if k not in rules: rules[k] = v # If the policy name and rule are the same, no # ambiguity, so no reason to warn. elif str(v) != str(rules[k]): msg = ("The same policy name: %s was found in " "multiple policies files for service %s. " "This can lead to policy rule ambiguity. " "Using rule: %s; Rule from file: %s") LOG.warning(msg, k, self.service, rules[k], v) except (ValueError, IOError): LOG.warning("Failed to read policy file '%s' for service %s.", path, self.service) # Check whether policy actions are defined in code. Nova and Keystone, # for example, define their default policy actions in code. mgr = stevedore.named.NamedExtensionManager( 'oslo.policy.policies', names=[self.service], invoke_on_load=True, warn_on_missing_entrypoint=False) if mgr: policy_generator = {plc.name: plc.obj for plc in mgr} if self.service in policy_generator: for rule in policy_generator[self.service]: if rule.name not in rules: rules[rule.name] = rule.check elif str(rule.check) != str(rules[rule.name]): msg = ("The same policy name: %s was found in the " "policies files and in the code for service " "%s. This can lead to policy rule ambiguity. " "Using rule: %s; Rule from code: %s") LOG.warning(msg, rule.name, self.service, rules[rule.name], rule.check) if not rules: msg = ( 'Policy files for {0} service were not found among the ' 'registered in-code policies or in any of the possible policy ' 'files: {1}.'.format( self.service, [loc % self.service for loc in CONF.patrole.custom_policy_files])) raise rbac_exceptions.RbacParsingException(msg) return rules def _is_admin_context(self, roles): """Checks whether a role has admin context. If context_is_admin is contained in the policy file, then checks whether the given role is contained in context_is_admin. If it is not in the policy file, then default to context_is_admin: admin. """ if 'context_is_admin' in self.rules.keys(): return self._allowed( access=self._get_access_token(roles), apply_rule='context_is_admin') return CONF.identity.admin_role in roles def _get_access_token(self, roles): access_token = { "token": { "roles": [ { "name": role } for role in roles ], "project_id": self.project_id, "tenant_id": self.project_id, "user_id": self.user_id } } return access_token def _allowed(self, access, apply_rule, is_admin=False): """Checks if a given rule in a policy is allowed with given ``access``. :param dict access: Dictionary from ``_get_access_token``. :param string apply_rule: Rule to be checked using ``oslo.policy``. :param bool is_admin: Whether admin context is used. """ access_data = copy.copy(access['token']) access_data['roles'] = [role['name'] for role in access_data['roles']] access_data['is_admin'] = is_admin # TODO(felipemonteiro): Dynamically calculate is_admin_project rather # than hard-coding it to True. is_admin_project cannot be determined # from the role, but rather from project and domain names. For more # information, see: # https://github.com/openstack/keystone/blob/37ce5417418f8acbd27f3dacb70c605b0fe48301/keystone/token/providers/common.py#L150 access_data['is_admin_project'] = True class Object(object): pass o = Object() o.rules = self.rules target = {"project_id": access_data['project_id'], "tenant_id": access_data['project_id'], "network:tenant_id": access_data['project_id'], "user_id": access_data['user_id']} if self.extra_target_data: target.update(self.extra_target_data) result = self._try_rule(apply_rule, target, access_data, o) return result def _try_rule(self, apply_rule, target, access_data, o): if apply_rule not in self.rules: message = ('Policy action "{0}" not found in policy files: ' '{1} or among registered policy in code defaults for ' '{2} service.').format(apply_rule, self.policy_files[self.service], self.service) LOG.debug(message) raise rbac_exceptions.RbacParsingException(message) else: rule = self.rules[apply_rule] return rule(target, access_data, o)