# 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 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': 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)
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(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
check = rule.check
if getattr(rule, 'deprecated_rule', False):
check = self._handle_deprecated_rule(rule)
rules[rule.name] = 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)
