Initial functionality framework.
Includes:
rbac_util - Utility for switching between roles for tests.
rbac_auth - Determines if a given role is valid for a given api call.
rbac_rule_validation - Determines if a allowed proper access and denied improper access (403 error)
rbac_role_converter - Converts policy.json files into a list of api's and the roles that can access them.
One example rbac_base in tests/api/rbac_base
One example test in tests/api/images/test_images_rbac.py
New config settings for rbac_flag, rbac_test_role, and rbac_roles
Implements bp: initial-framework
Co-Authored-By: Sangeet Gupta <sg774j@att.com>
Co-Authored-By: Rick Bartra <rb560u@att.com>
Co-Authored-By: Felipe Monteiro <felipe.monteiro@att.com>
Co-Authored-By: Anthony Bellino <ab2434@att.com>
Co-Authored-By: Avishek Dutta <ad620p@att.com>
Change-Id: Ic97b2558ba33ab47ac8174ae37629d36ceb1c9de
diff --git a/doc/source/installation.rst b/doc/source/installation.rst
index 51b0645..9bfe14e 100644
--- a/doc/source/installation.rst
+++ b/doc/source/installation.rst
@@ -43,6 +43,7 @@
#. [rbac] section updates ::
# The role that you want the RBAC tests to use for RBAC testing
+ # This needs to be edited to run the test as a different role.
rbac_role=_member_
# Tell standard RBAC test cases to run other wise it they are skipped.
rbac_flag=true
diff --git a/patrole_tempest_plugin/config.py b/patrole_tempest_plugin/config.py
index b0f25fd..2b20391 100644
--- a/patrole_tempest_plugin/config.py
+++ b/patrole_tempest_plugin/config.py
@@ -17,3 +17,17 @@
rbac_group = cfg.OptGroup(name='rbac',
title='RBAC testing options')
+
+RbacGroup = [
+ cfg.StrOpt('rbac_test_role',
+ default='admin',
+ help="The current RBAC role against which to run"
+ " Patrole tests."),
+ cfg.BoolOpt('rbac_flag',
+ default=False,
+ help="Enables RBAC tests."),
+ cfg.ListOpt('rbac_roles',
+ default=['admin'],
+ help="List of RBAC roles found in the policy files "
+ "under testing."),
+]
diff --git a/patrole_tempest_plugin/plugin.py b/patrole_tempest_plugin/plugin.py
index 62f4f15..1bc4d04 100644
--- a/patrole_tempest_plugin/plugin.py
+++ b/patrole_tempest_plugin/plugin.py
@@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
-
import os
from tempest import config
diff --git a/patrole_tempest_plugin/rbac_auth.py b/patrole_tempest_plugin/rbac_auth.py
new file mode 100644
index 0000000..88b7032
--- /dev/null
+++ b/patrole_tempest_plugin/rbac_auth.py
@@ -0,0 +1,43 @@
+# Copyright 2017 AT&T Corp
+# 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.
+
+from oslo_log import log as logging
+from tempest.lib import exceptions
+
+from patrole_tempest_plugin import rbac_role_converter
+
+LOG = logging.getLogger(__name__)
+
+
+class RbacAuthority(object):
+ def __init__(self, component=None, service=None):
+ self.converter = rbac_role_converter.RbacPolicyConverter(service)
+ self.roles_dict = self.converter.rules
+
+ def get_permission(self, api, role):
+ if self.roles_dict is None:
+ raise exceptions.InvalidConfiguration("Roles dictionary is empty!")
+ try:
+ _api = self.roles_dict[api]
+ if role in _api:
+ LOG.debug("[API]: %s, [Role]: %s is allowed!", api, role)
+ return True
+ else:
+ LOG.debug("[API]: %s, [Role]: %s is NOT allowed!", api, role)
+ return False
+ except KeyError:
+ raise KeyError("'%s' API is not defined in the policy.json"
+ % api)
+ return False
diff --git a/patrole_tempest_plugin/rbac_exceptions.py b/patrole_tempest_plugin/rbac_exceptions.py
new file mode 100644
index 0000000..94e6bdd
--- /dev/null
+++ b/patrole_tempest_plugin/rbac_exceptions.py
@@ -0,0 +1,28 @@
+# Copyright 2017 AT&T Corp
+# 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.
+
+from tempest.lib import exceptions
+
+
+class RbacActionFailed (exceptions.ClientRestClientException):
+ message = "Rbac action failed"
+
+
+class RbacResourceSetupFailed (exceptions.TempestException):
+ message = "Rbac resource setup failed"
+
+
+class RbacOverPermission (exceptions.TempestException):
+ message = "Action performed that should not be permitted"
diff --git a/patrole_tempest_plugin/rbac_role_converter.py b/patrole_tempest_plugin/rbac_role_converter.py
new file mode 100644
index 0000000..cfb7856
--- /dev/null
+++ b/patrole_tempest_plugin/rbac_role_converter.py
@@ -0,0 +1,153 @@
+# Copyright 2016 AT&T Corp
+# 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 os
+
+from oslo_config import cfg
+from oslo_log import log as logging
+from oslo_policy import _checks
+from oslo_policy import policy
+from tempest import config
+
+from patrole_tempest_plugin.rbac_exceptions import RbacResourceSetupFailed
+
+CONF = config.CONF
+LOG = logging.getLogger(__name__)
+
+RULES_TO_SKIP = []
+TESTED_RULES = []
+PARSED_RULES = []
+
+
+class RbacPolicyConverter(object):
+ """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
+ the roles that constitute it. This class automates that process.
+ """
+
+ def __init__(self, service, path=None):
+ """Initialization of Policy Converter
+
+ Parse policy files to create dictionary mapping
+ policy actions to roles.
+ :param service: type string
+ :param path: type string
+ """
+
+ if path is None:
+ path = '/etc/{0}/policy.json'.format(service)
+
+ if not os.path.isfile(path):
+ raise RbacResourceSetupFailed('Policy file for service: {0}, {1}'
+ ' not found.'.format(service, path))
+
+ self.default_roles = CONF.rbac.rbac_roles
+ self.rules = {}
+
+ self._get_roles_for_each_rule_in_policy_file(path)
+
+ def _get_roles_for_each_rule_in_policy_file(self, path):
+ """Gets the roles for each rule in the policy file at given path."""
+
+ global PARSED_RULES
+ global TESTED_RULES
+ global RULES_TO_SKIP
+
+ rule_to_roles_dict = {}
+ enforcer = self._init_policy_enforcer(path)
+
+ base_rules = set()
+ for rule_name, rule_checker in enforcer.rules.items():
+ if isinstance(rule_checker, _checks.OrCheck):
+ for sub_rule in rule_checker.rules:
+ if hasattr(sub_rule, 'match'):
+ base_rules.add(sub_rule.match)
+ elif isinstance(rule_checker, _checks.RuleCheck):
+ if hasattr(rule_checker, 'match'):
+ base_rules.add(rule_checker.match)
+
+ RULES_TO_SKIP.extend(base_rules)
+ generic_check_dict = self._get_generic_check_dict(enforcer.rules)
+
+ for rule_name, rule_checker in enforcer.rules.items():
+ PARSED_RULES.append(rule_name)
+
+ if rule_name in RULES_TO_SKIP:
+ continue
+ if isinstance(rule_checker, _checks.GenericCheck):
+ continue
+
+ # Determine whether each role is contained within the current rule.
+ for role in self.default_roles:
+ roles = {'roles': [role]}
+ roles.update(generic_check_dict)
+ is_role_in_rule = rule_checker(
+ generic_check_dict, roles, enforcer)
+ if is_role_in_rule:
+ rule_to_roles_dict.setdefault(rule_name, set())
+ rule_to_roles_dict[rule_name].add(role)
+
+ self.rules = rule_to_roles_dict
+
+ def _init_policy_enforcer(self, policy_file):
+ """Initializes oslo policy enforcer"""
+
+ def find_file(path):
+ realpath = os.path.realpath(path)
+ if os.path.isfile(realpath):
+ return realpath
+ else:
+ return None
+
+ CONF = cfg.CONF
+ CONF.find_file = find_file
+
+ enforcer = policy.Enforcer(CONF,
+ policy_file=policy_file,
+ rules=None,
+ default_rule=None,
+ use_conf=True)
+ enforcer.load_rules()
+ return enforcer
+
+ def _get_generic_check_dict(self, enforcer_rules):
+ """Creates permissions dictionary that oslo policy uses
+
+ to determine if a user can perform an action.
+ """
+
+ generic_checks = set()
+ for rule_checker in enforcer_rules.values():
+ entries = set()
+ self._get_generic_check_entries(rule_checker, entries)
+ generic_checks |= entries
+ return {e: '' for e in generic_checks}
+
+ def _get_generic_check_entries(self, rule_checker, entries):
+ if isinstance(rule_checker, _checks.GenericCheck):
+ if hasattr(rule_checker, 'match'):
+ if rule_checker.match.startswith('%(') and\
+ rule_checker.match.endswith(')s'):
+ entries.add(rule_checker.match[2:-2])
+ if hasattr(rule_checker, 'rule'):
+ if isinstance(rule_checker.rule, _checks.GenericCheck) and\
+ hasattr(rule_checker.rule, 'match'):
+ if rule_checker.rule.match.startswith('%(') and\
+ rule_checker.rule.match.endswith(')s'):
+ entries.add(rule_checker.rule.match[2:-2])
+ if hasattr(rule_checker, 'rules'):
+ for rule in rule_checker.rules:
+ self._get_generic_check_entries(rule, entries)
diff --git a/patrole_tempest_plugin/rbac_rule_validation.py b/patrole_tempest_plugin/rbac_rule_validation.py
new file mode 100644
index 0000000..7785eea
--- /dev/null
+++ b/patrole_tempest_plugin/rbac_rule_validation.py
@@ -0,0 +1,60 @@
+# Copyright 2017 AT&T Corp
+# 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
+
+from tempest import config
+from tempest.lib import exceptions
+
+from patrole_tempest_plugin import rbac_auth
+from patrole_tempest_plugin import rbac_exceptions
+
+CONF = config.CONF
+LOG = logging.getLogger(__name__)
+
+
+def action(component, service, rule):
+ def decorator(func):
+ def wrapper(*args, **kwargs):
+ authority = rbac_auth.RbacAuthority(component, service)
+ allowed = authority.get_permission(rule, CONF.rbac.rbac_test_role)
+
+ try:
+ func(*args)
+ except exceptions.Forbidden as e:
+ if allowed:
+ msg = ("Role %s was not allowed to perform %s." %
+ (CONF.rbac.rbac_test_role, rule))
+ LOG.error(msg)
+ raise exceptions.Forbidden(
+ "%s exception was: %s" %
+ (msg, e))
+ except rbac_exceptions.RbacActionFailed as e:
+ if allowed:
+ msg = ("Role %s was not allowed to perform %s." %
+ (CONF.rbac.rbac_test_role, rule))
+ LOG.error(msg)
+ raise exceptions.Forbidden(
+ "%s RbacActionFailed was: %s" %
+ (msg, e))
+ else:
+ if not allowed:
+ LOG.error("Role %s was allowed to perform %s" %
+ (CONF.rbac.rbac_test_role, rule))
+ raise rbac_exceptions.RbacOverPermission(
+ "OverPermission: Role %s was allowed to perform %s" %
+ (CONF.rbac.rbac_test_role, rule))
+ return wrapper
+ return decorator
diff --git a/patrole_tempest_plugin/rbac_utils.py b/patrole_tempest_plugin/rbac_utils.py
new file mode 100644
index 0000000..7792cbd
--- /dev/null
+++ b/patrole_tempest_plugin/rbac_utils.py
@@ -0,0 +1,128 @@
+# Copyright 2017 AT&T Corp
+# 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 json
+import logging
+import six
+import time
+import urllib3
+
+from tempest import config
+
+from patrole_tempest_plugin import rbac_exceptions as rbac_exc
+
+LOG = logging.getLogger(__name__)
+CONF = config.CONF
+http = urllib3.PoolManager()
+
+
+class Singleton(type):
+ _instances = {}
+
+ def __call__(cls, *args, **kwargs):
+ if cls not in cls._instances:
+ cls._instances[cls] = super(Singleton, cls).__call__(*args,
+ **kwargs)
+ return cls._instances[cls]
+
+
+@six.add_metaclass(Singleton)
+class RbacUtils(object):
+ def __init__(self):
+ RbacUtils.dictionary = {}
+
+ @staticmethod
+ def get_roles(caller):
+ admin_role_id = None
+ rbac_role_id = None
+
+ if bool(RbacUtils.dictionary) is False:
+ admin_token = caller.admin_client.token
+ headers = {'X-Auth-Token': admin_token,
+ "Content-Type": "application/json"}
+ url_to_get_role = CONF.identity.uri_v3 + '/roles/'
+ response = http.request('GET', url_to_get_role, headers=headers)
+ if response.status != 200:
+ raise rbac_exc.RbacResourceSetupFailed('Unable to'
+ ' retrieve roles')
+ data = response.data
+ roles = json.loads(data)
+ for item in roles['roles']:
+ if item['name'] == CONF.rbac.rbac_test_role:
+ rbac_role_id = item['id']
+ if item['name'] == 'admin':
+ admin_role_id = item['id']
+
+ RbacUtils.dictionary.update({'admin_role_id': admin_role_id,
+ 'rbac_role_id': rbac_role_id})
+
+ return RbacUtils.dictionary
+
+ @staticmethod
+ def delete_all_roles(self, base_url, headers):
+ # Find the current role
+ response = http.request('GET', base_url, headers=headers)
+ if response.status != 200:
+ raise rbac_exc.RbacResourceSetupFailed('Unable to retrieve'
+ ' user role')
+ data = response.data
+ roles = json.loads(data)
+ for item in roles['roles']:
+ url = base_url + item['id']
+ response = http.request('DELETE', url, headers=headers)
+ self.assertEqual(204, response.status)
+
+ @staticmethod
+ def switch_role(self, switchToRbacRole=None):
+ LOG.debug('Switching role to: %s', switchToRbacRole)
+ if switchToRbacRole is None:
+ return
+
+ roles = rbac_utils.get_roles(self)
+ rbac_role_id = roles.get('rbac_role_id')
+ admin_role_id = roles.get('admin_role_id')
+
+ try:
+ user_id = self.auth_provider.credentials.user_id
+ project_id = self.auth_provider.credentials.tenant_id
+ admin_token = self.admin_client.token
+
+ headers = {'X-Auth-Token': admin_token,
+ "Content-Type": "application/json"}
+ base_url = (CONF.identity.uri_v3 + '/projects/' + project_id +
+ '/users/' + user_id + '/roles/')
+
+ rbac_utils.delete_all_roles(self, base_url, headers)
+
+ if switchToRbacRole:
+ url = base_url + rbac_role_id
+ response = http.request('PUT', url, headers=headers)
+ self.assertEqual(204, response.status)
+ else:
+ url = base_url + admin_role_id
+ response = http.request('PUT', url, headers=headers)
+ self.assertEqual(204, response.status)
+
+ except Exception as exp:
+ LOG.error(exp)
+ raise
+ finally:
+ self.auth_provider.clear_auth()
+ # Sleep to avoid 401 errors caused by rounding
+ # In timing of fernet token creation
+ time.sleep(1)
+ self.auth_provider.set_auth()
+
+rbac_utils = RbacUtils()
diff --git a/patrole_tempest_plugin/tests/api/image/__init__.py b/patrole_tempest_plugin/tests/api/image/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/image/__init__.py
diff --git a/patrole_tempest_plugin/tests/api/image/test_images_rbac.py b/patrole_tempest_plugin/tests/api/image/test_images_rbac.py
new file mode 100644
index 0000000..83f4fe3
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/image/test_images_rbac.py
@@ -0,0 +1,52 @@
+# Copyright 2016 ATT 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
+
+from tempest import config
+from tempest.lib.common.utils import data_utils
+from tempest import test
+
+from patrole_tempest_plugin import rbac_rule_validation
+from patrole_tempest_plugin.rbac_utils import rbac_utils
+from patrole_tempest_plugin.tests.api import rbac_base
+
+CONF = config.CONF
+LOG = logging.getLogger(__name__)
+
+
+class BasicOperationsImagesRbacTest(rbac_base.BaseV2ImageRbacTest):
+
+ @classmethod
+ def setup_clients(cls):
+ super(BasicOperationsImagesRbacTest, cls).setup_clients()
+ cls.client = cls.os.image_client_v2
+
+ def tearDown(self):
+ rbac_utils.switch_role(self, switchToRbacRole=False)
+ super(BasicOperationsImagesRbacTest, self).tearDown()
+
+ @rbac_rule_validation.action(component="Image", service="glance",
+ rule="add_image")
+ @test.idempotent_id('0f148510-63bf-11e6-b348-080027d0d606')
+ def test_create_image(self):
+ uuid = '00000000-1111-2222-3333-444455556666'
+ image_name = data_utils.rand_name('image')
+ rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.create_image(name=image_name,
+ container_format='bare',
+ disk_format='raw',
+ visibility='private',
+ ramdisk_id=uuid)
diff --git a/patrole_tempest_plugin/tests/api/rbac_base.py b/patrole_tempest_plugin/tests/api/rbac_base.py
new file mode 100644
index 0000000..786927f
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/rbac_base.py
@@ -0,0 +1,39 @@
+# Copyright 2017 at&t
+# 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.
+
+# Maybe these should be in lib or recreated?
+from tempest.api.image import base as image_base
+from tempest import config
+
+CONF = config.CONF
+
+
+class BaseV2ImageRbacTest(image_base.BaseV2ImageTest):
+
+ credentials = ['primary', 'admin']
+
+ @classmethod
+ def skip_checks(cls):
+ super(BaseV2ImageRbacTest, cls).skip_checks()
+ if not CONF.rbac.rbac_flag:
+ raise cls.skipException(
+ "%s skipped as RBAC Flag not enabled" % cls.__name__)
+ if 'admin' not in CONF.auth.tempest_roles:
+ raise cls.skipException(
+ "%s skipped because tempest roles is not admin" % cls.__name__)
+
+ @classmethod
+ def setup_clients(cls):
+ super(BaseV2ImageRbacTest, cls).setup_clients()
+ cls.auth_provider = cls.os.auth_provider
+ cls.admin_client = cls.os_adm.image_client_v2
diff --git a/requirements.txt b/requirements.txt
index 1a62d2e..29091ac 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,3 +3,4 @@
# process, which may cause wedges in the gate later.
pbr>=1.8 # Apache-2.0
+urllib3>=1.15.1 # MIT
diff --git a/setup.cfg b/setup.cfg
index e382392..c174fca 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -48,4 +48,8 @@
[build_releasenotes]
all_files = 1
build-dir = releasenotes/build
-source-dir = releasenotes/source
\ No newline at end of file
+source-dir = releasenotes/source
+
+[entry_points]
+tempest.test_plugins =
+ patrole-tempest-plugin = patrole_tempest_plugin.plugin:PatroleTempestPlugin
diff --git a/test-requirements.txt b/test-requirements.txt
index 22838ca..cbb75b1 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -9,3 +9,5 @@
mock>=2.0 # BSD
coverage>=4.0 # Apache-2.0
oslotest>=1.10.0 # Apache-2.0
+oslo.policy>=1.17.0 # Apache-2.0
+tempest>=12.1.0 # Apache-2.0
diff --git a/tests/custom_rbac_policy.json b/tests/custom_rbac_policy.json
new file mode 100644
index 0000000..0e7466a
--- /dev/null
+++ b/tests/custom_rbac_policy.json
@@ -0,0 +1,14 @@
+{
+ "even_rule": "role:two or role:four or role:six or role:eight",
+ "odd_rule": "role:one or role:three or role:five or role:seven or role:nine",
+ "zero_rule": "role:zero",
+ "prime_rule": "role:one or role:two or role:three or role:five or role:seven",
+ "all_rule": "",
+
+ "policy_action_1": "rule:even_rule",
+ "policy_action_2": "rule:odd_rule",
+ "policy_action_3": "rule:zero_rule",
+ "policy_action_4": "rule:prime_rule",
+ "policy_action_5": "rule:all_rule",
+ "policy_action_6": "role:eight",
+}
diff --git a/tests/test_rbac_role_converter.py b/tests/test_rbac_role_converter.py
new file mode 100644
index 0000000..942d7d0
--- /dev/null
+++ b/tests/test_rbac_role_converter.py
@@ -0,0 +1,67 @@
+# Copyright 2017 AT&T Inc.
+#
+# 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 import config
+from tempest.tests import base
+
+from patrole_tempest_plugin import rbac_role_converter
+
+CONF = config.CONF
+
+
+class RbacPolicyTest(base.TestCase):
+
+ def setUp(self):
+ super(RbacPolicyTest, self).setUp()
+
+ current_directory = os.path.dirname(os.path.realpath(__file__))
+ self.custom_policy_file = os.path.join(current_directory,
+ 'custom_rbac_policy.json')
+
+ def test_custom_policy(self):
+ default_roles = ['zero', 'one', 'two', 'three', 'four',
+ 'five', 'six', 'seven', 'eight', 'nine']
+ CONF.set_override('rbac_roles', default_roles, group='rbac',
+ enforce_type=True)
+
+ self.converter = rbac_role_converter.RbacPolicyConverter(
+ "custom",
+ self.custom_policy_file
+ )
+ self.roles_dict = self.converter.rules
+
+ expected = {
+ 'policy_action_1': ['two', 'four', 'six', 'eight'],
+ 'policy_action_2': ['one', 'three', 'five', 'seven', 'nine'],
+ 'policy_action_3': ['zero'],
+ 'policy_action_4': ['one', 'two', 'three', 'five', 'seven'],
+ 'policy_action_5': ['zero', 'one', 'two', 'three', 'four', 'five',
+ 'six', 'seven', 'eight', 'nine'],
+ 'policy_action_6': ['eight'],
+ }
+
+ fake_rule = 'fake_rule'
+
+ self.assertFalse(fake_rule in self.roles_dict.keys())
+
+ for rule in expected.keys():
+ self.assertTrue(rule in self.roles_dict.keys())
+ expected_roles = expected[rule]
+ unexpected_roles = set(default_roles) - set(expected[rule])
+ for role in expected_roles:
+ self.assertTrue(role in self.roles_dict[rule])
+ for role in unexpected_roles:
+ self.assertFalse(role in self.roles_dict[rule])
diff --git a/tests/test_rbac_rule_validation.py b/tests/test_rbac_rule_validation.py
new file mode 100644
index 0000000..a8e2be3
--- /dev/null
+++ b/tests/test_rbac_rule_validation.py
@@ -0,0 +1,89 @@
+# Copyright 2017 AT&T Inc.
+#
+# 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 mock
+
+from patrole_tempest_plugin import rbac_exceptions
+from patrole_tempest_plugin import rbac_rule_validation as rbac_rv
+
+from tempest.lib import exceptions
+
+from tempest.tests import base
+
+
+class RBACRuleValidationTest(base.TestCase):
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_RBAC_rv_happy_path(self, mock_auth):
+ decorator = rbac_rv.action("", "", "")
+ mock_function = mock.Mock()
+ wrapper = decorator(mock_function)
+ wrapper()
+ self.assertTrue(mock_function.called)
+
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_RBAC_rv_forbidden(self, mock_auth):
+ decorator = rbac_rv.action("", "", "")
+ mock_function = mock.Mock()
+ mock_function.side_effect = exceptions.Forbidden
+ wrapper = decorator(mock_function)
+ self.assertRaises(exceptions.Forbidden, wrapper)
+
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_RBAC_rv_rbac_action_failed(self, mock_auth):
+ decorator = rbac_rv.action("", "", "")
+ mock_function = mock.Mock()
+ mock_function.side_effect = rbac_exceptions.RbacActionFailed
+ wrapper = decorator(mock_function)
+ self.assertRaises(exceptions.Forbidden, wrapper)
+
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_RBAC_rv_not_allowed(self, mock_auth):
+ decorator = rbac_rv.action("", "", "")
+
+ mock_function = mock.Mock()
+ wrapper = decorator(mock_function)
+
+ mock_permission = mock.Mock()
+ mock_permission.get_permission.return_value = False
+ mock_auth.return_value = mock_permission
+
+ self.assertRaises(rbac_exceptions.RbacOverPermission, wrapper)
+
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_RBAC_rv_forbidden_not_allowed(self, mock_auth):
+ decorator = rbac_rv.action("", "", "")
+
+ mock_function = mock.Mock()
+ mock_function.side_effect = exceptions.Forbidden
+ wrapper = decorator(mock_function)
+
+ mock_permission = mock.Mock()
+ mock_permission.get_permission.return_value = False
+ mock_auth.return_value = mock_permission
+
+ self.assertIsNone(wrapper())
+
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_RBAC_rv_rbac_action_failed_not_allowed(self, mock_auth):
+ decorator = rbac_rv.action("", "", "")
+
+ mock_function = mock.Mock()
+ mock_function.side_effect = rbac_exceptions.RbacActionFailed
+ wrapper = decorator(mock_function)
+
+ mock_permission = mock.Mock()
+ mock_permission.get_permission.return_value = False
+ mock_auth.return_value = mock_permission
+
+ self.assertIsNone(wrapper())
diff --git a/tests/test_rbac_utils.py b/tests/test_rbac_utils.py
new file mode 100644
index 0000000..657b389
--- /dev/null
+++ b/tests/test_rbac_utils.py
@@ -0,0 +1,198 @@
+# Copyright 2017 AT&T Inc.
+#
+# 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 json
+import mock
+
+from tempest.tests import base
+
+from patrole_tempest_plugin import rbac_exceptions
+from patrole_tempest_plugin import rbac_utils as utils
+
+
+class RBACUtilsTest(base.TestCase):
+ def setUp(self):
+ super(RBACUtilsTest, self).setUp()
+ self.rbac_utils = utils.RbacUtils
+
+ get_response = 200
+ put_response = 204
+ delete_response = 204
+ response_data = json.dumps({"roles": []})
+
+ def _response_side_effect(self, action, *args, **kwargs):
+ response = mock.MagicMock()
+ if action == "GET":
+ response.status = self.get_response
+ response.data = self.response_data
+ if action == "PUT":
+ response.status = self.put_response
+ if action == "DELETE":
+ response.status = self.delete_response
+ return response
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_get_roles(self, http, config):
+ self.rbac_utils.dictionary = {}
+
+ caller = mock.Mock()
+ caller.admin_client.token = "test_token"
+
+ http.request.side_effect = self._response_side_effect
+
+ self.assertEqual({'admin_role_id': None, 'rbac_role_id': None},
+ self.rbac_utils.get_roles(caller))
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_get_roles_member(self, http, config):
+ self.rbac_utils.dictionary = {}
+
+ caller = mock.Mock()
+ caller.admin_client.token = "test_token"
+
+ self.response_data = json.dumps({'roles': [{'name': '_member_',
+ 'id': '_member_id'}]})
+ http.request.side_effect = self._response_side_effect
+
+ config.rbac.rbac_test_role = '_member_'
+
+ self.assertEqual({'admin_role_id': None,
+ 'rbac_role_id': '_member_id'},
+ self.rbac_utils.get_roles(caller))
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_get_roles_admin(self, http, config):
+ self.rbac_utils.dictionary = {}
+
+ caller = mock.Mock()
+ caller.admin_client.token = "test_token"
+
+ self.response_data = json.dumps({'roles': [{'name': 'admin',
+ 'id': 'admin_id'}]})
+
+ http.request.side_effect = self._response_side_effect
+
+ config.rbac.rbac_test_role = 'admin'
+
+ self.assertEqual({'admin_role_id': 'admin_id',
+ 'rbac_role_id': 'admin_id'},
+ self.rbac_utils.get_roles(caller))
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_get_roles_admin_not_role(self, http, config):
+ self.rbac_utils.dictionary = {}
+
+ caller = mock.Mock()
+ caller.admin_client.token = "test_token"
+
+ self.response_data = json.dumps(
+ {'roles': [{'name': 'admin', 'id': 'admin_id'}]}
+ )
+ http.request.side_effect = self._response_side_effect
+
+ self.assertEqual({'admin_role_id': 'admin_id', 'rbac_role_id': None},
+ self.rbac_utils.get_roles(caller))
+
+ def test_RBAC_utils_get_existing_roles(self):
+ self.rbac_utils.dictionary = {'admin_role_id': None,
+ 'rbac_role_id': None}
+
+ self.assertEqual({'admin_role_id': None, 'rbac_role_id': None},
+ self.rbac_utils.get_roles(None))
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_get_roles_response_404(self, http, config):
+ self.rbac_utils.dictionary = {}
+
+ caller = mock.Mock()
+ caller.admin_client.token = "test_token"
+
+ http.request.side_effect = self._response_side_effect
+ self.get_response = 404
+
+ self.assertRaises(rbac_exceptions.RbacResourceSetupFailed,
+ self.rbac_utils.get_roles, caller)
+ self.get_response = 200
+
+ def test_RBAC_utils_switch_roles_none(self):
+ self.assertIsNone(self.rbac_utils.switch_role(None))
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.RbacUtils.get_roles')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_switch_roles_member(self, http,
+ get_roles, config):
+ get_roles.return_value = {'admin_role_id': None,
+ 'rbac_role_id': '_member_id'}
+
+ self.auth_provider = mock.Mock()
+ self.auth_provider.credentials.user_id = "user_id"
+ self.auth_provider.credentials.tenant_id = "tenant_id"
+ self.admin_client = mock.Mock()
+ self.admin_client.token = "admin_token"
+
+ http.request.side_effect = self._response_side_effect
+
+ self.assertIsNone(self.rbac_utils.switch_role(self, "_member_"))
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.RbacUtils.get_roles')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_switch_roles_false(self, http,
+ get_roles, config):
+ get_roles.return_value = {'admin_role_id': None,
+ 'rbac_role_id': '_member_id'}
+
+ self.auth_provider = mock.Mock()
+ self.auth_provider.credentials.user_id = "user_id"
+ self.auth_provider.credentials.tenant_id = "tenant_id"
+ self.admin_client = mock.Mock()
+ self.admin_client.token = "admin_token"
+
+ http.request.side_effect = self._response_side_effect
+
+ self.assertIsNone(self.rbac_utils.switch_role(self, False))
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.CONF')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.RbacUtils.get_roles')
+ @mock.patch('patrole_tempest_plugin.rbac_utils.http')
+ def test_RBAC_utils_switch_roles_get_roles_fails(self, http,
+ get_roles, config):
+ get_roles.return_value = {'admin_role_id': None,
+ 'rbac_role_id': '_member_id'}
+
+ self.auth_provider = mock.Mock()
+ self.auth_provider.credentials.user_id = "user_id"
+ self.auth_provider.credentials.tenant_id = "tenant_id"
+ self.admin_client = mock.Mock()
+ self.admin_client.token = "admin_token"
+
+ self.get_response = 404
+
+ self.assertRaises(rbac_exceptions.RbacResourceSetupFailed,
+ self.rbac_utils.switch_role, self, False)
+
+ self.get_response = 200
+
+ @mock.patch('patrole_tempest_plugin.rbac_utils.RbacUtils.get_roles')
+ def test_RBAC_utils_switch_roles_exception(self, get_roles):
+ get_roles.return_value = {'admin_role_id': None,
+ 'rbac_role_id': '_member_id'}
+ self.assertRaises(AttributeError, self.rbac_utils.switch_role,
+ self, "admin")
diff --git a/tox.ini b/tox.ini
index c888992..847adad 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,6 +1,6 @@
[tox]
-minversion = 2.0
-envlist = py34,py27,pypy,pep8
+minversion = 1.6
+envlist = py35,py27,pypy,pep8
skipsdist = True
[testenv]
@@ -9,8 +9,13 @@
setenv =
VIRTUAL_ENV={envdir}
PYTHONWARNINGS=default::DeprecationWarning
-deps = -r{toxinidir}/test-requirements.txt
-commands = python setup.py test --slowest --testr-args='{posargs}'
+passenv = OS_STDOUT_CAPTURE OS_STDERR_CAPTURE OS_TEST_TIMEOUT OS_TEST_LOCK_PATH OS_TEST_PATH TEMPEST_CONFIG TEMPEST_CONFIG_DIR http_proxy HTTP_PROXY https_proxy HTTPS_PROXY no_proxy NO_PROXY
+whitelist_externals = *
+deps = -r{toxinidir}/requirements.txt
+ -r{toxinidir}/test-requirements.txt
+commands =
+ find . -type f -name "*.pyc" -delete
+ ostestr {posargs}
[testenv:pep8]
commands = flake8 {posargs}