Source code for patrole_tempest_plugin.policy_authority

# 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
#    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 pkg_resources
import stevedore
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__)

[docs]class PolicyAuthority(RbacAuthority): """A class that uses ``oslo.policy`` for validating RBAC.""" os_admin = None 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
[docs] @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') and cls.os_admin: services_client = (cls.os_admin.identity_services_v3_client if CONF.identity_feature_enabled.api_v3 else cls.os_admin.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
[docs] @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)
[docs] 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 _handle_deprecated_rule(self, default): deprecated_rule = default.deprecated_rule deprecated_msg = ( 'Policy "%(old_name)s":"%(old_check_str)s" was deprecated in ' '%(release)s in favor of "%(name)s":"%(check_str)s". Reason: ' '%(reason)s. Either ensure your deployment is ready for the new ' 'default or copy/paste the deprecated policy into your policy ' 'file and maintain it manually.' % { 'old_name':, 'old_check_str': deprecated_rule.check_str, 'release': default.deprecated_since, 'name':, 'check_str': default.check_str, 'reason': default.deprecated_reason } ) LOG.warn(deprecated_msg) oslo_policy_version = pkg_resources.parse_version( pkg_resources.get_distribution("oslo.policy").version) # NOTE(gmann): oslo policy 3.7.0 onwards does not allow to modify # the Rule object check attribute. required_version = pkg_resources.parse_version('3.7.0') if oslo_policy_version >= required_version: return policy.OrCheck([default.check, deprecated_rule.check]) else: default.check = policy.OrCheck( [policy._parser.parse_rule(cs) for cs in [default.check_str, deprecated_rule.check_str]]) return default.check
[docs] 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( 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.obj for plc in mgr} if self.service in policy_generator: for rule in policy_generator[self.service]: if not in rules: if CONF.patrole.validate_deprecated_rules: # NOTE (sergey.vilgelm): # The `DocumentedRuleDefault` object has no # `deprecated_rule` attribute in Pike check = rule.check if getattr(rule, 'deprecated_rule', False): check = self._handle_deprecated_rule(rule) rules[] = check elif str(rule.check) != str(rules[]): 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,, self.service, rules[], 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: 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': r} for r 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: # 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)