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
#
#         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__)


[docs]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

[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'):
            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

[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': deprecated_rule.name,
                'old_check_str': deprecated_rule.check_str,
                'release': default.deprecated_since,
                'name': default.name,
                'check_str': default.check_str,
                'reason': default.deprecated_reason
            }
        )
        LOG.warn(deprecated_msg)
        check_str = '(%s) or (%s)' % (default.check_str,
                                      deprecated_rule.check_str)
        return policy.RuleDefault(default.name, check_str)

[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(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:
                        if CONF.patrole.validate_deprecated_rules:
                            # NOTE (sergey.vilgelm):
                            # The `DocumentedRuleDefault` object has no
                            # `deprecated_rule` attribute in Pike
                            if getattr(rule, 'deprecated_rule', False):
                                rule = self._handle_deprecated_rule(rule)
                        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:
            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:
        # https://git.openstack.org/cgit/openstack/keystone/tree/keystone/token/providers/common.py?id=37ce5417418f8acbd27f3dacb70c605b0fe48301#n150
        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)