Add support for testing custom RBAC requirements

Add support of running Patrole against a custom requirements YAML that
defines RBAC requirements. The YAML file lists all the APIs and the roles
that should have access to the APIs. The purpose of running Patrole against
a requirements YAML is to verify that the RBAC policy is in accordance to
deployment specific requirements. Running Patrole against a requirements
YAML is completely optional and can be enabled through the rbac section of
the tempest.conf.

Change-Id: I8ba89ab5e134b15e97ac20a7aacbfd70896e192f
Implements: blueprint support-custom-yaml
Co-Authored-By: Sangeet Gupta <sg774j@att.com>
Co-Authored-By: David Purcell <d.purcell222@gmail.com>

patrole_tempest_plugin/config.py

@@ -31,6 +31,8 @@ RbacGroup = [
                 help="If true, throws RbacParsingException for"
                      " policies which don't exist. If false, "
                      "throws skipException."),
+    # TODO(rb560u): There needs to be support for reading these JSON files from
+    # other hosts. It may be possible to leverage the v3 identity policy API
     cfg.StrOpt('cinder_policy_file',
                default='/etc/cinder/policy.json',
                help="Location of the neutron policy file."),
@@ -45,5 +47,56 @@ RbacGroup = [
                help="Location of the neutron policy file."),
     cfg.StrOpt('nova_policy_file',
                default='/etc/nova/policy.json',
-               help="Location of the nova policy file.")
+               help="Location of the nova policy file."),
+    cfg.BoolOpt('test_custom_requirements',
+                default=False,
+                help="""
+This option determines whether Patrole should run against a
+`custom_requirements_file` which defines RBAC requirements. The
+purpose of setting this flag to True is to verify that RBAC policy
+is in accordance to requirements. The idea is that the
+`custom_requirements_file` perfectly defines what the RBAC requirements are.
+
+Here are the possible outcomes when running the Patrole tests against
+a `custom_requirements_file`:
+
+YAML definition: allowed
+test run: allowed
+test result: pass
+
+YAML definition: allowed
+test run: not allowed
+test result: fail (under-permission)
+
+YAML definition: not allowed
+test run: allowed
+test result: fail (over-permission)
+"""),
+    cfg.StrOpt('custom_requirements_file',
+               help="""
+File path of the yaml file that defines your RBAC requirements. This
+file must be located on the same host that Patrole runs on. The yaml
+file should be written as follows:
+
+```
+<service>:
+  <api_action>:
+    - <allowed_role>
+    - <allowed_role>
+    - <allowed_role>
+  <api_action>:
+    - <allowed_role>
+    - <allowed_role>
+<service>
+  <api_action>:
+    - <allowed_role>
+```
+Where:
+service = the service that is being tested (cinder, nova, etc)
+api_action = the policy action that is being tested. Examples:
+             - volume:create
+             - os_compute_api:servers:start
+             - add_image
+allowed_role = the Keystone role that is allowed to perform the API
+""")
 ]

patrole_tempest_plugin/rbac_policy_parser.py

@@ -25,12 +25,13 @@ import stevedore
 from tempest.common import credentials_factory as credentials
 
 from patrole_tempest_plugin import rbac_exceptions
+from patrole_tempest_plugin.rbac_utils import RbacAuthority
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
 
 
-class RbacPolicyParser(object):
+class RbacPolicyParser(RbacAuthority):
     """A class for parsing policy rules into lists of allowed roles.
 
     RBAC testing requires that each rule in a policy file be broken up into

patrole_tempest_plugin/rbac_rule_validation.py

@@ -25,6 +25,7 @@ from tempest import test
 
 from patrole_tempest_plugin import rbac_exceptions
 from patrole_tempest_plugin import rbac_policy_parser
+from patrole_tempest_plugin import requirements_authority
 
 CONF = config.CONF
 LOG = logging.getLogger(__name__)
@@ -39,6 +40,9 @@ def action(service, rule='', admin_only=False, expected_error_code=403,
     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.
@@ -141,12 +145,17 @@ def _is_authorized(test_obj, service, rule_name, extra_target_data):
 
     try:
         role = CONF.rbac.rbac_test_role
-        formatted_target_data = _format_extra_target_data(
-            test_obj, extra_target_data)
-        policy_parser = rbac_policy_parser.RbacPolicyParser(
-            project_id, user_id, service,
-            extra_target_data=formatted_target_data)
-        is_allowed = policy_parser.allowed(rule_name, role)
+        # Test RBAC against custom requirements. Otherwise use oslo.policy
+        if CONF.rbac.test_custom_requirements:
+            authority = requirements_authority.RequirementsAuthority(
+                CONF.rbac.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,

patrole_tempest_plugin/rbac_utils.py

@@ -13,6 +13,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import abc
+import six
 import sys
 import time
 
@@ -170,3 +172,13 @@ class RbacUtils(object):
         :returns: True if ``rbac_test_role`` is the admin role.
         """
         return CONF.rbac.rbac_test_role == CONF.identity.admin_role
+
+
+@six.add_metaclass(abc.ABCMeta)
+class RbacAuthority(object):
+    # TODO(rb560u): Add documentation explaining what this class is for
+
+    @abc.abstractmethod
+    def allowed(self, rule_name, role):
+        """Determine whether the role should be able to perform the API"""
+        return

patrole_tempest_plugin/requirements_authority.py

@@ -0,0 +1,72 @@
+# 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 yaml
+
+from oslo_log import log as logging
+
+from tempest.lib import exceptions
+
+from patrole_tempest_plugin.rbac_utils import RbacAuthority
+
+LOG = logging.getLogger(__name__)
+
+
+class RequirementsParser(object):
+    _inner = None
+
+    class Inner(object):
+        _rbac_map = None
+
+        def __init__(self, filepath):
+            with open(filepath) as f:
+                RequirementsParser.Inner._rbac_map = \
+                    list(yaml.safe_load_all(f))
+
+    def __init__(self, filepath):
+        if RequirementsParser._inner is None:
+            RequirementsParser._inner = RequirementsParser.Inner(filepath)
+
+    @staticmethod
+    def parse(component):
+        try:
+            for section in RequirementsParser.Inner._rbac_map:
+                if component in section:
+                    return section[component]
+        except yaml.parser.ParserError:
+            LOG.error("Error while parsing the requirements YAML file. Did "
+                      "you pass a valid component name from the test case?")
+        return None
+
+
+class RequirementsAuthority(RbacAuthority):
+    def __init__(self, filepath=None, component=None):
+        if filepath is not None and component is not None:
+            self.roles_dict = RequirementsParser(filepath).parse(component)
+        else:
+            self.roles_dict = None
+
+    def allowed(self, rule_name, role):
+        if self.roles_dict is None:
+            raise exceptions.InvalidConfiguration(
+                "Roles dictionary parsed from requirements YAML file is "
+                "empty. Ensure the requirements YAML file is correctly "
+                "formatted.")
+        try:
+            _api = self.roles_dict[rule_name]
+            return role in _api
+        except KeyError:
+            raise KeyError("'%s' API is not defined in the requirements YAML "
+                           "file" % rule_name)
+        return False

patrole_tempest_plugin/tests/unit/resources/rbac_roles.yaml

@@ -0,0 +1,6 @@
+Test:
+  test:create:
+    - test_member
+    - _member_
+  test:create2:
+    - test_member

patrole_tempest_plugin/tests/unit/test_requirements_authority.py

@@ -0,0 +1,85 @@
+# Copyright 2017 AT&T Corporation.
+#
+#    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 os
+
+from tempest.lib import exceptions
+from tempest.tests import base
+
+from patrole_tempest_plugin import requirements_authority as req_auth
+
+
+class RequirementsAuthorityTest(base.TestCase):
+    def setUp(self):
+        super(RequirementsAuthorityTest, self).setUp()
+        self.rbac_auth = req_auth.RequirementsAuthority()
+        self.current_directory = os.path.dirname(os.path.realpath(__file__))
+        self.yaml_test_file = os.path.join(self.current_directory,
+                                           'resources',
+                                           'rbac_roles.yaml')
+        self.expected_result = {'test:create': ['test_member', '_member_'],
+                                'test:create2': ['test_member']}
+
+    def test_requirements_auth_init(self):
+        rbac_auth = req_auth.RequirementsAuthority(self.yaml_test_file, 'Test')
+        self.assertEqual(self.expected_result, rbac_auth.roles_dict)
+
+    def test_auth_allowed_empty_roles(self):
+        self.rbac_auth.roles_dict = None
+        self.assertRaises(exceptions.InvalidConfiguration,
+                          self.rbac_auth.allowed, "", "")
+
+    def test_auth_allowed_role_in_api(self):
+        self.rbac_auth.roles_dict = {'api': ['_member_']}
+        self.assertTrue(self.rbac_auth.allowed("api", "_member_"))
+
+    def test_auth_allowed_role_not_in_api(self):
+        self.rbac_auth.roles_dict = {'api': ['_member_']}
+        self.assertFalse(self.rbac_auth.allowed("api", "support_member"))
+
+    def test_parser_get_allowed_except_keyerror(self):
+        self.rbac_auth.roles_dict = {}
+        self.assertRaises(KeyError, self.rbac_auth.allowed,
+                          "api", "support_member")
+
+    def test_parser_init(self):
+        req_auth.RequirementsParser(self.yaml_test_file)
+        self.assertEqual([{'Test': self.expected_result}],
+                         req_auth.RequirementsParser.Inner._rbac_map)
+
+    def test_parser_role_in_api(self):
+        req_auth.RequirementsParser.Inner._rbac_map = \
+            [{'Test': self.expected_result}]
+        self.rbac_auth.roles_dict = req_auth.RequirementsParser.parse("Test")
+
+        self.assertEqual(self.expected_result, self.rbac_auth.roles_dict)
+        self.assertTrue(self.rbac_auth.allowed("test:create2", "test_member"))
+
+    def test_parser_role_not_in_api(self):
+        req_auth.RequirementsParser.Inner._rbac_map = \
+            [{'Test': self.expected_result}]
+        self.rbac_auth.roles_dict = req_auth.RequirementsParser.parse("Test")
+
+        self.assertEqual(self.expected_result, self.rbac_auth.roles_dict)
+        self.assertFalse(self.rbac_auth.allowed("test:create2", "_member_"))
+
+    def test_parser_except_invalid_configuration(self):
+        req_auth.RequirementsParser.Inner._rbac_map = \
+            [{'Test': self.expected_result}]
+        self.rbac_auth.roles_dict = \
+            req_auth.RequirementsParser.parse("Failure")
+
+        self.assertIsNone(self.rbac_auth.roles_dict)
+        self.assertRaises(exceptions.InvalidConfiguration,
+                          self.rbac_auth.allowed, "", "")

releasenotes/notes/support_requirements_yaml-a90e0188a19421ba.yaml

@@ -0,0 +1,12 @@
+---
+features:
+  - |
+    Add support of running Patrole against a custom requirements YAML that
+    defines RBAC requirements. The YAML file lists all the APIs and the roles
+    that should have access to the APIs. The purpose of running Patrole against
+    a requirements YAML is to verify that the RBAC policy is in accordance to
+    deployment specific requirements. Running Patrole against a requirements
+    YAML is completely optional and can be enabled by setting the
+    ``[rbac] test_custom_requirements`` option to True in Tempest's
+    configuration file. The requirements YAML must be located on the same host
+    that Patrole runs on.