patrole_tempest_plugin/rbac_rule_validation.py - packaging/sources/patrole - Gitiles

 # 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 logging
 import sys
 import testtools

 import six

 from tempest import config
 from tempest.lib import exceptions
 from tempest import test

 from patrole_tempest_plugin import rbac_exceptions
 from patrole_tempest_plugin import rbac_policy_parser
 from patrole_tempest_plugin import rbac_utils
 from patrole_tempest_plugin import requirements_authority

 CONF = config.CONF
 LOG = logging.getLogger(__name__)

 _SUPPORTED_ERROR_CODES = [403, 404]


 def action(service, rule='', admin_only=False, expected_error_code=403,
            extra_target_data=None):
     """A decorator which does a policy check and matches it against test run.

     A decorator which allows for positive and negative RBAC testing. Given
     an OpenStack service and a policy action enforced by that service, an
     oslo.policy lookup is performed by calling `authority.get_permission`.
     Alternatively, the RBAC tests can run against a YAML file that defines
     policy requirements.

     The following cases are possible:

     * If `allowed` is True and the test passes, this is a success.
     * If `allowed` is True and the test fails, this is a failure.
     * If `allowed` is False and the test passes, this is a failure.
     * If `allowed` is False and the test fails, this is a success.

     :param service: A OpenStack service: for example, "nova" or "neutron".
     :param rule: A policy action defined in a policy.json file (or in code).
     :param admin_only: Skips over oslo.policy check because the policy action
         defined by `rule` is not enforced by the service's  policy enforcement
         logic. For example, Keystone v2 performs an admin check for most of its
         endpoints. If True, `rule` is effectively ignored.
     :param expected_error_code: Overrides default value of 403 (Forbidden)
         with endpoint-specific error code. Currently only supports 403 and 404.
         Support for 404 is needed because some services, like Neutron,
         intentionally throw a 404 for security reasons.

     :raises NotFound: if `service` is invalid or
                       if Tempest credentials cannot be found.
     :raises Forbidden: for bullet (2) above.
     :raises RbacOverPermission: for bullet (3) above.
     """

     if extra_target_data is None:
         extra_target_data = {}

     def decorator(func):
         role = CONF.patrole.rbac_test_role

         def wrapper(*args, **kwargs):
             if args and isinstance(args[0], test.BaseTestCase):
                 test_obj = args[0]
             else:
                 raise rbac_exceptions.RbacResourceSetupFailed(
                     '`rbac_rule_validation` decorator can only be applied to '
                     'an instance of `tempest.test.BaseTestCase`.')

             if admin_only:
                 LOG.info("As admin_only is True, only admin role should be "
                          "allowed to perform the API. Skipping oslo.policy "
                          "check for policy action {0}.".format(rule))
                 allowed = rbac_utils.is_admin()
             else:
                 allowed = _is_authorized(test_obj, service, rule,
                                          extra_target_data)

             expected_exception, irregular_msg = _get_exception_type(
                 expected_error_code)

             try:
                 func(*args, **kwargs)
             except rbac_exceptions.RbacInvalidService as e:
                 msg = ("%s is not a valid service." % service)
                 LOG.error(msg)
                 raise exceptions.NotFound(
                     "%s RbacInvalidService was: %s" % (msg, e))
             except (expected_exception,
                     rbac_exceptions.RbacConflictingPolicies,
                     rbac_exceptions.RbacMalformedResponse) as e:
                 if irregular_msg:
                     LOG.warning(irregular_msg.format(rule, service))
                 if allowed:
                     msg = ("Role %s was not allowed to perform %s." %
                            (role, rule))
                     LOG.error(msg)
                     raise exceptions.Forbidden(
                         "%s Exception was: %s" % (msg, e))
             except Exception as e:
                 exc_info = sys.exc_info()
                 error_details = exc_info[1].__str__()
                 msg = ("%s An unexpected exception has occurred: Expected "
                        "exception was %s, which was not thrown."
                        % (error_details, expected_exception.__name__))
                 LOG.error(msg)
                 six.reraise(exc_info[0], exc_info[0](msg), exc_info[2])
             else:
                 if not allowed:
                     LOG.error("Role %s was allowed to perform %s",
                               role, rule)
                     raise rbac_exceptions.RbacOverPermission(
                         "OverPermission: Role %s was allowed to perform %s" %
                         (role, rule))
             finally:
                 test_obj.rbac_utils.switch_role(test_obj,
                                                 toggle_rbac_role=False)

         _wrapper = testtools.testcase.attr(role)(wrapper)
         return _wrapper
     return decorator


 def _is_authorized(test_obj, service, rule_name, extra_target_data):
     """Validates whether current RBAC role has permission to do policy action.

     :param test_obj: type BaseTestCase (tempest base test class)
     :param service: the OpenStack service that enforces ``rule_name``
     :param rule_name: the name of the policy action
     :param extra_target_data: dictionary with unresolved string literals that
         reference nested BaseTestCase attributes
     :returns: True if the current RBAC role can perform the policy action else
         False

     :raises RbacResourceSetupFailed: if project_id or user_id are missing from
         the Tempest test object's `auth_provider`
     :raises RbacParsingException: if ``CONF.patrole.strict_policy_check`` is
         enabled and the ``rule_name`` does not exist in the system
     :raises skipException: if ``CONF.patrole.strict_policy_check`` is
         disabled and the ``rule_name`` does not exist in the system
     """
     try:
         project_id = test_obj.os_primary.credentials.project_id
         user_id = test_obj.os_primary.credentials.user_id
     except AttributeError as e:
         msg = ("{0}: project_id or user_id not found in os_primary.credentials"
                .format(e))
         LOG.error(msg)
         raise rbac_exceptions.RbacResourceSetupFailed(msg)

     try:
         role = CONF.patrole.rbac_test_role
         # Test RBAC against custom requirements. Otherwise use oslo.policy
         if CONF.patrole.test_custom_requirements:
             authority = requirements_authority.RequirementsAuthority(
                 CONF.patrole.custom_requirements_file, service)
         else:
             formatted_target_data = _format_extra_target_data(
                 test_obj, extra_target_data)
             authority = rbac_policy_parser.RbacPolicyParser(
                 project_id, user_id, service,
                 extra_target_data=formatted_target_data)
         is_allowed = authority.allowed(rule_name, role)

         if is_allowed:
             LOG.debug("[Action]: %s, [Role]: %s is allowed!", rule_name,
                       role)
         else:
             LOG.debug("[Action]: %s, [Role]: %s is NOT allowed!",
                       rule_name, role)
         return is_allowed
     except rbac_exceptions.RbacParsingException as e:
         if CONF.patrole.strict_policy_check:
             raise e
         else:
             raise testtools.TestCase.skipException(str(e))
     return False


 def _get_exception_type(expected_error_code=403):
     """Dynamically calculate the expected exception to be caught.

     Dynamically calculate the expected exception to be caught by the test case.
     Only `Forbidden` and `NotFound` exceptions are permitted. `NotFound` is
     supported because Neutron, for security reasons, masks `Forbidden`
     exceptions as `NotFound` exceptions.

     :param expected_error_code: the integer representation of the expected
         exception to be caught. Must be contained in `_SUPPORTED_ERROR_CODES`.
     :returns: tuple of the exception type corresponding to
         `expected_error_code` and a message explaining that a non-Forbidden
         exception was expected, if applicable.
     """
     expected_exception = None
     irregular_msg = None

     if not isinstance(expected_error_code, six.integer_types) \
         or expected_error_code not in _SUPPORTED_ERROR_CODES:
         msg = ("Please pass an expected error code. Currently "
                "supported codes: {0}".format(_SUPPORTED_ERROR_CODES))
         LOG.error(msg)
         raise rbac_exceptions.RbacInvalidErrorCode(msg)

     if expected_error_code == 403:
         expected_exception = exceptions.Forbidden
     elif expected_error_code == 404:
         expected_exception = exceptions.NotFound
         irregular_msg = ("NotFound exception was caught for policy action "
                          "{0}. The service {1} throws a 404 instead of a 403, "
                          "which is irregular.")

     return expected_exception, irregular_msg


 def _format_extra_target_data(test_obj, extra_target_data):
     """Formats the "extra_target_data" dictionary with correct test data.

     Before being formatted, "extra_target_data" is a dictionary that maps a
     policy string like "trust.trustor_user_id" to a nested list of BaseTestCase
     attributes. For example, the attribute list in:

         "trust.trustor_user_id": "os.auth_provider.credentials.user_id"

     is parsed by iteratively calling `getattr` until the value of "user_id"
     is resolved. The resulting dictionary returns:

         "trust.trustor_user_id": "the user_id of the `primary` credential"

     :param test_obj: type BaseTestCase (tempest base test class)
     :param extra_target_data: dictionary with unresolved string literals that
         reference nested BaseTestCase attributes
     :returns: dictionary containing additional object data needed by
         oslo.policy to validate generic checks
     """
     attr_value = test_obj
     formatted_target_data = {}

     for user_attribute, attr_string in extra_target_data.items():
         attrs = attr_string.split('.')
         for attr in attrs:
             attr_value = getattr(attr_value, attr)
         formatted_target_data[user_attribute] = attr_value

     return formatted_target_data
	# 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 logging
	import sys
	import testtools

	import six

	from tempest import config
	from tempest.lib import exceptions
	from tempest import test

	from patrole_tempest_plugin import rbac_exceptions
	from patrole_tempest_plugin import rbac_policy_parser
	from patrole_tempest_plugin import rbac_utils
	from patrole_tempest_plugin import requirements_authority

	CONF = config.CONF
	LOG = logging.getLogger(__name__)

	_SUPPORTED_ERROR_CODES = [403, 404]


	def action(service, rule='', admin_only=False, expected_error_code=403,
	extra_target_data=None):
	"""A decorator which does a policy check and matches it against test run.

	A decorator which allows for positive and negative RBAC testing. Given
	an OpenStack service and a policy action enforced by that service, an
	oslo.policy lookup is performed by calling `authority.get_permission`.
	Alternatively, the RBAC tests can run against a YAML file that defines
	policy requirements.

	The following cases are possible:

	* If `allowed` is True and the test passes, this is a success.
	* If `allowed` is True and the test fails, this is a failure.
	* If `allowed` is False and the test passes, this is a failure.
	* If `allowed` is False and the test fails, this is a success.

	:param service: A OpenStack service: for example, "nova" or "neutron".
	:param rule: A policy action defined in a policy.json file (or in code).
	:param admin_only: Skips over oslo.policy check because the policy action
	defined by `rule` is not enforced by the service's policy enforcement
	logic. For example, Keystone v2 performs an admin check for most of its
	endpoints. If True, `rule` is effectively ignored.
	:param expected_error_code: Overrides default value of 403 (Forbidden)
	with endpoint-specific error code. Currently only supports 403 and 404.
	Support for 404 is needed because some services, like Neutron,
	intentionally throw a 404 for security reasons.

	:raises NotFound: if `service` is invalid or
	if Tempest credentials cannot be found.
	:raises Forbidden: for bullet (2) above.
	:raises RbacOverPermission: for bullet (3) above.
	"""

	if extra_target_data is None:
	extra_target_data = {}

	def decorator(func):
	role = CONF.patrole.rbac_test_role

	def wrapper(args, *kwargs):
	if args and isinstance(args[0], test.BaseTestCase):
	test_obj = args[0]
	else:
	raise rbac_exceptions.RbacResourceSetupFailed(
	'`rbac_rule_validation` decorator can only be applied to '
	'an instance of `tempest.test.BaseTestCase`.')

	if admin_only:
	LOG.info("As admin_only is True, only admin role should be "
	"allowed to perform the API. Skipping oslo.policy "
	"check for policy action {0}.".format(rule))
	allowed = rbac_utils.is_admin()
	else:
	allowed = _is_authorized(test_obj, service, rule,
	extra_target_data)

	expected_exception, irregular_msg = _get_exception_type(
	expected_error_code)

	try:
	func(args, *kwargs)
	except rbac_exceptions.RbacInvalidService as e:
	msg = ("%s is not a valid service." % service)
	LOG.error(msg)
	raise exceptions.NotFound(
	"%s RbacInvalidService was: %s" % (msg, e))
	except (expected_exception,
	rbac_exceptions.RbacConflictingPolicies,
	rbac_exceptions.RbacMalformedResponse) as e:
	if irregular_msg:
	LOG.warning(irregular_msg.format(rule, service))
	if allowed:
	msg = ("Role %s was not allowed to perform %s." %
	(role, rule))
	LOG.error(msg)
	raise exceptions.Forbidden(
	"%s Exception was: %s" % (msg, e))
	except Exception as e:
	exc_info = sys.exc_info()
	error_details = exc_info[1].__str__()
	msg = ("%s An unexpected exception has occurred: Expected "
	"exception was %s, which was not thrown."
	% (error_details, expected_exception.__name__))
	LOG.error(msg)
	six.reraise(exc_info[0], exc_info[0](msg), exc_info[2])
	else:
	if not allowed:
	LOG.error("Role %s was allowed to perform %s",
	role, rule)
	raise rbac_exceptions.RbacOverPermission(
	"OverPermission: Role %s was allowed to perform %s" %
	(role, rule))
	finally:
	test_obj.rbac_utils.switch_role(test_obj,
	toggle_rbac_role=False)

	_wrapper = testtools.testcase.attr(role)(wrapper)
	return _wrapper
	return decorator


	def _is_authorized(test_obj, service, rule_name, extra_target_data):
	"""Validates whether current RBAC role has permission to do policy action.

	:param test_obj: type BaseTestCase (tempest base test class)
	:param service: the OpenStack service that enforces ``rule_name``
	:param rule_name: the name of the policy action
	:param extra_target_data: dictionary with unresolved string literals that
	reference nested BaseTestCase attributes
	:returns: True if the current RBAC role can perform the policy action else
	False

	:raises RbacResourceSetupFailed: if project_id or user_id are missing from
	the Tempest test object's `auth_provider`
	:raises RbacParsingException: if ``CONF.patrole.strict_policy_check`` is
	enabled and the ``rule_name`` does not exist in the system
	:raises skipException: if ``CONF.patrole.strict_policy_check`` is
	disabled and the ``rule_name`` does not exist in the system
	"""
	try:
	project_id = test_obj.os_primary.credentials.project_id
	user_id = test_obj.os_primary.credentials.user_id
	except AttributeError as e:
	msg = ("{0}: project_id or user_id not found in os_primary.credentials"
	.format(e))
	LOG.error(msg)
	raise rbac_exceptions.RbacResourceSetupFailed(msg)

	try:
	role = CONF.patrole.rbac_test_role
	# Test RBAC against custom requirements. Otherwise use oslo.policy
	if CONF.patrole.test_custom_requirements:
	authority = requirements_authority.RequirementsAuthority(
	CONF.patrole.custom_requirements_file, service)
	else:
	formatted_target_data = _format_extra_target_data(
	test_obj, extra_target_data)
	authority = rbac_policy_parser.RbacPolicyParser(
	project_id, user_id, service,
	extra_target_data=formatted_target_data)
	is_allowed = authority.allowed(rule_name, role)

	if is_allowed:
	LOG.debug("[Action]: %s, [Role]: %s is allowed!", rule_name,
	role)
	else:
	LOG.debug("[Action]: %s, [Role]: %s is NOT allowed!",
	rule_name, role)
	return is_allowed
	except rbac_exceptions.RbacParsingException as e:
	if CONF.patrole.strict_policy_check:
	raise e
	else:
	raise testtools.TestCase.skipException(str(e))
	return False


	def _get_exception_type(expected_error_code=403):
	"""Dynamically calculate the expected exception to be caught.

	Dynamically calculate the expected exception to be caught by the test case.
	Only `Forbidden` and `NotFound` exceptions are permitted. `NotFound` is
	supported because Neutron, for security reasons, masks `Forbidden`
	exceptions as `NotFound` exceptions.

	:param expected_error_code: the integer representation of the expected
	exception to be caught. Must be contained in `_SUPPORTED_ERROR_CODES`.
	:returns: tuple of the exception type corresponding to
	`expected_error_code` and a message explaining that a non-Forbidden
	exception was expected, if applicable.
	"""
	expected_exception = None
	irregular_msg = None

	if not isinstance(expected_error_code, six.integer_types) \
	or expected_error_code not in _SUPPORTED_ERROR_CODES:
	msg = ("Please pass an expected error code. Currently "
	"supported codes: {0}".format(_SUPPORTED_ERROR_CODES))
	LOG.error(msg)
	raise rbac_exceptions.RbacInvalidErrorCode(msg)

	if expected_error_code == 403:
	expected_exception = exceptions.Forbidden
	elif expected_error_code == 404:
	expected_exception = exceptions.NotFound
	irregular_msg = ("NotFound exception was caught for policy action "
	"{0}. The service {1} throws a 404 instead of a 403, "
	"which is irregular.")

	return expected_exception, irregular_msg


	def _format_extra_target_data(test_obj, extra_target_data):
	"""Formats the "extra_target_data" dictionary with correct test data.

	Before being formatted, "extra_target_data" is a dictionary that maps a
	policy string like "trust.trustor_user_id" to a nested list of BaseTestCase
	attributes. For example, the attribute list in:

	"trust.trustor_user_id": "os.auth_provider.credentials.user_id"

	is parsed by iteratively calling `getattr` until the value of "user_id"
	is resolved. The resulting dictionary returns:

	"trust.trustor_user_id": "the user_id of the `primary` credential"

	:param test_obj: type BaseTestCase (tempest base test class)
	:param extra_target_data: dictionary with unresolved string literals that
	reference nested BaseTestCase attributes
	:returns: dictionary containing additional object data needed by
	oslo.policy to validate generic checks
	"""
	attr_value = test_obj
	formatted_target_data = {}

	for user_attribute, attr_string in extra_target_data.items():
	attrs = attr_string.split('.')
	for attr in attrs:
	attr_value = getattr(attr_value, attr)
	formatted_target_data[user_attribute] = attr_value

	return formatted_target_data