Source code for patrole_tempest_plugin.rbac_rule_validation

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

from oslo_log import versionutils
from oslo_utils import excutils
import six

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

from patrole_tempest_plugin import policy_authority
from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import requirements_authority

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

_SUPPORTED_ERROR_CODES = [403, 404]
_DEFAULT_ERROR_CODE = 403

RBACLOG = logging.getLogger('rbac_reporting')


[docs]def action(service,
           rules=None,
           expected_error_codes=None,
           extra_target_data=None):
    """A decorator for verifying OpenStack policy enforcement.

    A decorator which allows for positive and negative RBAC testing. Given:

    * an OpenStack service,
    * a policy action (``rule``) enforced by that service, and
    * the test roles defined by ``[patrole] rbac_test_roles``

    determines whether the test role has sufficient permissions to perform an
    API call that enforces the ``rule``.

    This decorator should only be applied to an instance or subclass of
    ``tempest.test.BaseTestCase``.

    The result from ``_is_authorized`` is used to determine the *expected*
    test result. The *actual* test result is determined by running the
    Tempest test this decorator applies to.

    Below are the following possibilities from comparing the *expected* and
    *actual* results:

    1) If *expected* is True and the test passes (*actual*), this is a success.
    2) If *expected* is True and the test fails (*actual*), this results in a
       ``RbacUnderPermissionException`` exception failure.
    3) If *expected* is False and the test passes (*actual*), this results in
       an ``RbacOverPermissionException`` exception failure.
    4) If *expected* is False and the test fails (*actual*), this is a success.

    As such, negative and positive testing can be applied using this decorator.

    :param str service: An OpenStack service. Examples: "nova" or "neutron".
    :param list rules: A list of policy actions defined in a policy file or in
        code. The rules are logical-ANDed together to derive the expected
        result. Also accepts list of callables that return a policy action.

        .. note::

            Patrole currently only supports custom JSON policy files.

    :type rules: list[str] or list[callable]
    :param list expected_error_codes: When the ``rules`` list parameter is
        used, then this list indicates the expected error code to use if one
        of the rules does not allow the role being tested. This list must
        coincide with and its elements remain in the same order as the rules
        in the rules list.

        Example::

            rules=["api_action1", "api_action2"]
            expected_error_codes=[404, 403]

        a) If api_action1 fails and api_action2 passes, then the expected
           error code is 404.
        b) if api_action2 fails and api_action1 passes, then the expected
           error code is 403.
        c) if both api_action1 and api_action2 fail, then the expected error
           code is the first error seen (404).

        If it is not passed, then it is defaulted to 403.

        .. warning::

            A 404 should not be provided *unless* the endpoint masks a
            ``Forbidden`` exception as a ``NotFound`` exception.

    :type expected_error_codes: list[int]
    :param dict extra_target_data: Dictionary, keyed with ``oslo.policy``
        generic check names, whose values are string literals that reference
        nested ``tempest.test.BaseTestCase`` attributes. Used by
        ``oslo.policy`` for performing matching against attributes that are
        sent along with the API calls. Example::

            extra_target_data={
                "target.token.user_id":
                "os_alt.auth_provider.credentials.user_id"
            })

    :raises RbacInvalidServiceException: If ``service`` is invalid.
    :raises RbacUnderPermissionException: For item (2) above.
    :raises RbacOverPermissionException: For item (3) above.
    :raises RbacExpectedWrongException: When a 403 is expected but a 404
        is raised instead or vice versa.

    Examples::

        @rbac_rule_validation.action(
            service="nova",
            rules=["os_compute_api:os-agents"])
        def test_list_agents_rbac(self):
            # The call to `override_role` is mandatory.
            with self.override_role():
                self.agents_client.list_agents()
    """

    if extra_target_data is None:
        extra_target_data = {}

    rules, expected_error_codes = _prepare_multi_policy(rules,
                                                        expected_error_codes)

    def decorator(test_func):
        roles = CONF.patrole.rbac_test_roles
        # TODO(vegasq) drop once CONF.patrole.rbac_test_role is removed
        if CONF.patrole.rbac_test_role:
            msg = ('CONF.patrole.rbac_test_role is deprecated in favor of '
                   'CONF.patrole.rbac_test_roles and will be removed in '
                   'future.')
            versionutils.report_deprecated_feature(LOG, msg)
            if not roles:
                roles.append(CONF.patrole.rbac_test_role)

        @functools.wraps(test_func)
        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`.')

            allowed = True
            disallowed_rules = []
            for rule in rules:
                _allowed = _is_authorized(
                    test_obj, service, rule, extra_target_data)
                if not _allowed:
                    disallowed_rules.append(rule)
                allowed = allowed and _allowed

            if disallowed_rules:
                # Choose the first disallowed rule and expect the error
                # code corresponding to it.
                first_error_index = rules.index(disallowed_rules[0])
                exp_error_code = expected_error_codes[first_error_index]
                LOG.debug("%s: Expecting %d to be raised for policy name: %s",
                          test_func.__name__, exp_error_code,
                          disallowed_rules[0])
            else:
                exp_error_code = expected_error_codes[0]

            expected_exception, irregular_msg = _get_exception_type(
                exp_error_code)

            caught_exception = None
            test_status = 'Allowed'

            try:
                test_func(*args, **kwargs)
            except rbac_exceptions.RbacInvalidServiceException:
                with excutils.save_and_reraise_exception():
                    msg = ("%s is not a valid service." % service)
                    # FIXME(felipemonteiro): This test_status is logged too
                    # late. Need a function to log it before re-raising.
                    test_status = ('Error, %s' % (msg))
                    LOG.error(msg)
            except (expected_exception,
                    rbac_exceptions.BasePatroleResponseBodyException) \
                    as actual_exception:
                caught_exception = actual_exception
                test_status = 'Denied'

                if irregular_msg:
                    LOG.warning(irregular_msg,
                                test_func.__name__,
                                ', '.join(rules),
                                service)

                if allowed:
                    msg = ("User with roles %s was not allowed to perform the "
                           "following actions: %s. Expected allowed actions: "
                           "%s. Expected disallowed actions: %s." % (
                               roles, sorted(rules),
                               sorted(set(rules) - set(disallowed_rules)),
                               sorted(disallowed_rules)))
                    LOG.error(msg)
                    raise rbac_exceptions.RbacUnderPermissionException(
                        "%s Exception was: %s" % (msg, actual_exception))
            except Exception as actual_exception:
                caught_exception = actual_exception

                if _check_for_expected_mismatch_exception(expected_exception,
                                                          actual_exception):
                    LOG.error('Expected and actual exceptions do not match. '
                              'Expected: %s. Actual: %s.',
                              expected_exception,
                              actual_exception.__class__)
                    raise rbac_exceptions.RbacExpectedWrongException(
                        expected=expected_exception,
                        actual=actual_exception.__class__,
                        exception=actual_exception)
                else:
                    with excutils.save_and_reraise_exception():
                        exc_info = sys.exc_info()
                        error_details = six.text_type(exc_info[1])
                        msg = ("An unexpected exception has occurred during "
                               "test: %s. Exception was: %s" % (
                                   test_func.__name__, error_details))
                        test_status = 'Error, %s' % (error_details)
                        LOG.error(msg)
            else:
                if not allowed:
                    msg = (
                        "OverPermission: Role %s was allowed to perform the "
                        "following disallowed actions: %s" % (
                            roles, sorted(disallowed_rules)
                        )
                    )
                    LOG.error(msg)
                    raise rbac_exceptions.RbacOverPermissionException(msg)
            finally:
                if CONF.patrole_log.enable_reporting:
                    RBACLOG.info(
                        "[Service]: %s, [Test]: %s, [Rules]: %s, "
                        "[Expected]: %s, [Actual]: %s",
                        service, test_func.__name__, ', '.join(rules),
                        "Allowed" if allowed else "Denied",
                        test_status)

                # Sanity-check that ``override_role`` was called to eliminate
                # false-positives and bad test flows resulting from exceptions
                # getting raised too early, too late or not at all, within
                # the scope of an RBAC test.
                _validate_override_role_called(
                    test_obj,
                    actual_exception=caught_exception)

        return wrapper
    return decorator


def _prepare_multi_policy(rules, exp_error_codes):
    if exp_error_codes:
        if not rules:
            msg = ("The `rules` list must be provided if using the "
                   "`expected_error_codes` list.")
            raise ValueError(msg)
        if len(rules) != len(exp_error_codes):
            msg = ("The `expected_error_codes` list is not the same length "
                   "as the `rules` list.")
            raise ValueError(msg)
        if not isinstance(exp_error_codes, (tuple, list)):
            exp_error_codes = [exp_error_codes]
    else:
        exp_error_codes = []

    if rules is None:
        rules = []
    elif not isinstance(rules, (tuple, list)):
        rules = [rules]

    # Fill in the exp_error_codes if needed. This is needed for the scenarios
    # where no exp_error_codes array is provided, so the error codes must be
    # set to the default error code value and there must be the same number
    # of error codes as rules.
    num_ecs = len(exp_error_codes)
    num_rules = len(rules)
    if (num_ecs < num_rules):
        for i in range(num_rules - num_ecs):
            exp_error_codes.append(_DEFAULT_ERROR_CODE)

    evaluated_rules = [
        r() if callable(r) else r for r in rules
    ]

    return evaluated_rules, exp_error_codes


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

    :param test_obj: An instance or subclass of ``tempest.test.BaseTestCase``.
    :param service: The OpenStack service that enforces ``rule``.
    :param rule: The name of the policy action. Examples include
        "identity:create_user" or "os_compute_api:os-agents".
    :param extra_target_data: Dictionary, keyed with ``oslo.policy`` generic
        check names, whose values are string literals that reference nested
        ``tempest.test.BaseTestCase`` attributes. Used by ``oslo.policy`` for
        performing matching against attributes that are sent along with the API
        calls.

    :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 `auth_provider` attribute in `test_obj`.
    """

    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)

    roles = CONF.patrole.rbac_test_roles
    # TODO(vegasq) drop once CONF.patrole.rbac_test_role is removed
    if CONF.patrole.rbac_test_role:
        if not roles:
            roles.append(CONF.patrole.rbac_test_role)

    # Adding implied roles
    roles = test_obj.get_all_needed_roles(roles)

    # 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 = policy_authority.PolicyAuthority(
            project_id, user_id, service,
            extra_target_data=formatted_target_data)
    is_allowed = authority.allowed(rule, roles)

    if is_allowed:
        LOG.debug("[Policy action]: %s, [Role]: %s is allowed!", rule,
                  roles)
    else:
        LOG.debug("[Policy action]: %s, [Role]: %s is NOT allowed!",
                  rule, roles)

    return is_allowed


def _get_exception_type(expected_error_code=_DEFAULT_ERROR_CODE):
    """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 = lib_exc.Forbidden
    elif expected_error_code == 404:
        expected_exception = lib_exc.NotFound
        irregular_msg = ("NotFound exception was caught for test %s. Expected "
                         "policies which may have caused the error: %s. The "
                         "service %s 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
    ``tempest.test.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 `os_primary` credential"

    :param test_obj: An instance or subclass of ``tempest.test.BaseTestCase``.
    :param extra_target_data: Dictionary, keyed with ``oslo.policy`` generic
        check names, whose values are string literals that reference nested
        ``tempest.test.BaseTestCase`` attributes. Used by ``oslo.policy`` for
        performing matching against attributes that are sent along with the API
        calls.
    :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


def _check_for_expected_mismatch_exception(expected_exception,
                                           actual_exception):
    """Checks that ``expected_exception`` matches ``actual_exception``.

    Since Patrole must handle 403/404 it is important that the expected and
    actual error codes match.

    :param excepted_exception: Expected exception for test.
    :param actual_exception: Actual exception raised by test.
    :returns: True if match, else False.
    :rtype: boolean
    """
    permission_exceptions = (lib_exc.Forbidden, lib_exc.NotFound)
    if isinstance(actual_exception, permission_exceptions):
        if not isinstance(actual_exception, expected_exception.__class__):
            return True
    return False


def _validate_override_role_called(test_obj, actual_exception):
    """Validates that :func:`rbac_utils.RbacUtilsMixin.override_role` is called
    during each Patrole test.

    Useful for validating that the expected exception isn't raised too early
    (before ``override_role`` call) or too late (after ``override_call``) or
    at all (which is a bad test).

    :param test_obj: An instance or subclass of ``tempest.test.BaseTestCase``.
    :param actual_exception: Actual exception raised by test.
    :raises RbacOverrideRoleException: If ``override_role`` isn't called, is
        called too early, or is called too late.
    """
    called = test_obj._validate_override_role_called()
    base_msg = ('This error is unrelated to RBAC and is due to either '
                'an API or override role failure. Exception: %s' %
                actual_exception)

    if not called:
        if actual_exception is not None:
            msg = ('Caught exception (%s) but it was raised before the '
                   '`override_role` context. ' % actual_exception.__class__)
        else:
            msg = 'Test missing required `override_role` call. '
        msg += base_msg
        LOG.error(msg)
        raise rbac_exceptions.RbacOverrideRoleException(msg)
    else:
        exc_caught_in_ctx = test_obj._validate_override_role_caught_exc()
        # This block is only executed if ``override_role`` is called. If
        # an exception is raised and the exception wasn't raised in the
        # ``override_role`` context and if the exception isn't a valid
        # exception type (instance of ``BasePatroleException``), then this is
        # a legitimate error.
        if (not exc_caught_in_ctx and
            actual_exception is not None and
            not isinstance(actual_exception,
                           rbac_exceptions.BasePatroleException)):
            msg = ('Caught exception (%s) but it was raised after the '
                   '`override_role` context. ' % actual_exception.__class__)
            msg += base_msg
            LOG.error(msg)
            raise rbac_exceptions.RbacOverrideRoleException(msg)