Add service validation to Patrole framework
Verify that the service being passed in the rbac_rule_validation
decorator is a valid service. The Tempest Identity v3 services_client
is being used to make a call to Keystone to list the services that
are available.
If an invalid service is passed in the decorator, then an exception
is thrown.
Change-Id: I3de3fccf18456bb8382864eeabcbfe64e2cffebb
Implements: blueprint add-service-validation
diff --git a/patrole_tempest_plugin/rbac_exceptions.py b/patrole_tempest_plugin/rbac_exceptions.py
index c6165ff..ee42e19 100644
--- a/patrole_tempest_plugin/rbac_exceptions.py
+++ b/patrole_tempest_plugin/rbac_exceptions.py
@@ -26,3 +26,7 @@
class RbacOverPermission (exceptions.TempestException):
message = "Action performed that should not be permitted"
+
+
+class RbacInvalidService (exceptions.TempestException):
+ message = "Attempted to test an invalid service"
diff --git a/patrole_tempest_plugin/rbac_policy_parser.py b/patrole_tempest_plugin/rbac_policy_parser.py
index 045a9f8..9926613 100644
--- a/patrole_tempest_plugin/rbac_policy_parser.py
+++ b/patrole_tempest_plugin/rbac_policy_parser.py
@@ -20,6 +20,8 @@
from oslo_policy import generator
from oslo_policy import policy
+from tempest.common import credentials_factory as credentials
+
from patrole_tempest_plugin import rbac_exceptions
LOG = logging.getLogger(__name__)
@@ -57,7 +59,17 @@
:param service: type string
:param path: type string
"""
- service = service.lower().strip()
+ # First check if the service is valid
+ service = service.lower().strip() if service else None
+ self.admin_mgr = credentials.AdminManager()
+ services = self.admin_mgr.identity_services_v3_client.\
+ list_services()['services']
+ service_names = [s['name'] for s in services]
+ if not service or not any(service in name for name in service_names):
+ LOG.debug(str(service) + " is NOT a valid service.")
+ raise rbac_exceptions.RbacInvalidService
+
+ # Use default path if no path provided
if path is None:
self.path = os.path.join('/etc', service, 'policy.json')
else:
@@ -65,7 +77,7 @@
policy_data = "{}"
- # First check whether policy file exists.
+ # Check whether policy file exists.
if os.path.isfile(self.path):
policy_data = open(self.path, 'r').read()
# Otherwise use oslo_policy to fetch the rules for provided service.
diff --git a/patrole_tempest_plugin/rbac_rule_validation.py b/patrole_tempest_plugin/rbac_rule_validation.py
index 36784b7..284d8f0 100644
--- a/patrole_tempest_plugin/rbac_rule_validation.py
+++ b/patrole_tempest_plugin/rbac_rule_validation.py
@@ -41,6 +41,12 @@
try:
func(*args)
+ 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 exceptions.Forbidden as e:
if allowed:
msg = ("Role %s was not allowed to perform %s." %
diff --git a/patrole_tempest_plugin/tests/unit/test_rbac_policy_parser.py b/patrole_tempest_plugin/tests/unit/test_rbac_policy_parser.py
index 35aaa82..edc63b8 100644
--- a/patrole_tempest_plugin/tests/unit/test_rbac_policy_parser.py
+++ b/patrole_tempest_plugin/tests/unit/test_rbac_policy_parser.py
@@ -19,6 +19,7 @@
from tempest import config
from tempest.tests import base
+from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_policy_parser
CONF = config.CONF
@@ -28,6 +29,8 @@
def setUp(self):
super(RbacPolicyTest, self).setUp()
+ self.mock_admin_mgr = mock.patch.object(
+ rbac_policy_parser, 'credentials').start()
current_directory = os.path.dirname(os.path.realpath(__file__))
self.custom_policy_file = os.path.join(current_directory,
@@ -42,6 +45,35 @@
self.tenant_policy_file = os.path.join(current_directory,
'resources',
'tenant_rbac_policy.json')
+ services = {
+ 'services': [
+ {'name': 'cinder', 'links': 'link', 'enabled': True,
+ 'type': 'volume', 'id': 'id',
+ 'description': 'description'},
+ {'name': 'glance', 'links': 'link', 'enabled': True,
+ 'type': 'image', 'id': 'id',
+ 'description': 'description'},
+ {'name': 'nova', 'links': 'link', 'enabled': True,
+ 'type': 'compute', 'id': 'id',
+ 'description': 'description'},
+ {'name': 'keystone', 'links': 'link', 'enabled': True,
+ 'type': 'identity', 'id': 'id',
+ 'description': 'description'},
+ {'name': 'heat', 'links': 'link', 'enabled': True,
+ 'type': 'orchestration', 'id': 'id',
+ 'description': 'description'},
+ {'name': 'neutron', 'links': 'link', 'enabled': True,
+ 'type': 'networking', 'id': 'id',
+ 'description': 'description'},
+ {'name': 'test', 'links': 'link', 'enabled': True,
+ 'type': 'unit_test', 'id': 'id',
+ 'description': 'description'}
+ ]
+ }
+
+ self.mock_admin_mgr.AdminManager.return_value.\
+ identity_services_v3_client.list_services.return_value = \
+ services
@mock.patch.object(rbac_policy_parser, 'LOG', autospec=True)
def test_custom_policy(self, m_log):
@@ -203,3 +235,33 @@
mock_try_rule.assert_called_once_with(
rule, expected_target, expected_access_data, mock.ANY)
mock_try_rule.reset_mock()
+
+ @mock.patch.object(rbac_policy_parser, 'LOG', autospec=True)
+ def test_invalid_service_raises_exception(self, m_log):
+ test_tenant_id = mock.sentinel.tenant_id
+ test_user_id = mock.sentinel.user_id
+ service = 'invalid_service'
+
+ self.assertRaises(rbac_exceptions.RbacInvalidService,
+ rbac_policy_parser.RbacPolicyParser,
+ test_tenant_id,
+ test_user_id,
+ service)
+
+ m_log.debug.assert_called_once_with(
+ "{0} is NOT a valid service.".format(str(service)))
+
+ @mock.patch.object(rbac_policy_parser, 'LOG', autospec=True)
+ def test_service_is_none_raises_exception(self, m_log):
+ test_tenant_id = mock.sentinel.tenant_id
+ test_user_id = mock.sentinel.user_id
+ service = None
+
+ self.assertRaises(rbac_exceptions.RbacInvalidService,
+ rbac_policy_parser.RbacPolicyParser,
+ test_tenant_id,
+ test_user_id,
+ service)
+
+ m_log.debug.assert_called_once_with(
+ "{0} is NOT a valid service.".format(str(service)))