Merge "Add tests to cover neutron-agents"
diff --git a/.zuul.yaml b/.zuul.yaml
index 66339eb..fb110f0 100644
--- a/.zuul.yaml
+++ b/.zuul.yaml
@@ -87,6 +87,11 @@
RBAC_TEST_ROLE: member
- job:
+ name: patrole-member-rocky
+ parent: patrole-member
+ override-checkout: stable/rocky
+
+- job:
name: patrole-member-queens
parent: patrole-member
override-checkout: stable/queens
@@ -151,6 +156,7 @@
tempest: true
neutron: true
neutron-segments: true
+ neutron-qos: true
- job:
name: patrole-plugin-member
@@ -184,6 +190,7 @@
jobs:
- patrole-admin
- patrole-member
+ - patrole-member-rocky
- patrole-member-queens
- patrole-member-pike
- patrole-py35-member
@@ -195,3 +202,8 @@
jobs:
- patrole-admin
- patrole-member
+ periodic-stable:
+ jobs:
+ - patrole-member-rocky
+ - patrole-member-queens
+ - patrole-member-pike
diff --git a/devstack/plugin.sh b/devstack/plugin.sh
index 4826d21..a6259f4 100644
--- a/devstack/plugin.sh
+++ b/devstack/plugin.sh
@@ -29,6 +29,9 @@
# These policies were removed in Stein but are available in Pike.
iniset $TEMPEST_CONFIG policy-feature-enabled removed_nova_policies_stein False
+
+ # TODO(cl566n): Policies used by Patrole testing. Remove these once stable/pike becomes EOL.
+ iniset $TEMPEST_CONFIG policy-feature-enabled added_cinder_policies_stein False
fi
if [[ ${DEVSTACK_SERIES} == 'queens' ]]; then
@@ -38,9 +41,16 @@
# These policies were removed in Stein but are available in Queens.
iniset $TEMPEST_CONFIG policy-feature-enabled removed_nova_policies_stein False
+
+ # TODO(cl566n): Policies used by Patrole testing. Remove these once stable/queens becomes EOL.
+ iniset $TEMPEST_CONFIG policy-feature-enabled added_cinder_policies_stein False
fi
- iniset $TEMPEST_CONFIG patrole enable_rbac True
+ if [[ ${DEVSTACK_SERIES} == 'rocky' ]]; then
+ # TODO(cl566n): Policies used by Patrole testing. Remove these once stable/rocky becomes EOL.
+ iniset $TEMPEST_CONFIG policy-feature-enabled added_cinder_policies_stein False
+ fi
+
iniset $TEMPEST_CONFIG patrole rbac_test_role $RBAC_TEST_ROLE
}
diff --git a/doc/source/framework/overview.rst b/doc/source/framework/overview.rst
index 4902f7b..113d461 100644
--- a/doc/source/framework/overview.rst
+++ b/doc/source/framework/overview.rst
@@ -1,6 +1,8 @@
RBAC Testing Validation
=======================
+.. _framework-overview:
+
--------
Overview
--------
diff --git a/doc/source/framework/policy_authority.rst b/doc/source/framework/policy_authority.rst
index 822c7b6..37b698c 100644
--- a/doc/source/framework/policy_authority.rst
+++ b/doc/source/framework/policy_authority.rst
@@ -60,3 +60,4 @@
.. automodule:: patrole_tempest_plugin.policy_authority
:members:
:undoc-members:
+ :special-members:
diff --git a/doc/source/framework/rbac_authority.rst b/doc/source/framework/rbac_authority.rst
index 84c372b..40f2a8d 100644
--- a/doc/source/framework/rbac_authority.rst
+++ b/doc/source/framework/rbac_authority.rst
@@ -35,3 +35,4 @@
.. automodule:: patrole_tempest_plugin.rbac_authority
:members:
:undoc-members:
+ :special-members:
diff --git a/doc/source/framework/rbac_utils.rst b/doc/source/framework/rbac_utils.rst
index 7143928..b13a4a3 100644
--- a/doc/source/framework/rbac_utils.rst
+++ b/doc/source/framework/rbac_utils.rst
@@ -23,156 +23,14 @@
and test execution, respectively. This is especially true when considering
custom policy rule definitions, which can be arbitrarily complex.
-.. _role-overriding:
-
-Role Overriding
-^^^^^^^^^^^^^^^
-
-Role overriding is the way Patrole is able to create resources and delete
-resources -- including those that require admin credentials -- while still
-being able to exercise the same set of Tempest credentials to perform the API
-action that authorizes the policy under test, by manipulating the role of
-the Tempest credentials.
-
-Patrole implicitly splits up each test into 3 stages: set up, test execution,
-and teardown.
-
-The role workflow is as follows:
-
-#. Setup: Admin role is used automatically. The primary credentials are
- overridden with the admin role.
-#. Test execution: ``[patrole] rbac_test_role`` is used manually via the
- call to ``with rbac_utils.override_role(self)``. Everything that
- is executed within this contextmanager uses the primary
- credentials overridden with the ``[patrole] rbac_test_role``.
-#. Teardown: Admin role is used automatically. The primary credentials have
- been overridden with the admin role.
-
-.. _Tempest credentials: https://docs.openstack.org/tempest/latest/library/credential_providers.html
-.. _dynamic credentials: https://docs.openstack.org/tempest/latest/configuration.html#dynamic-credentials
-
-Test Setup
-----------
-
-Automatic role override in background.
-
-Resources can be set up inside the ``resource_setup`` class method that Tempest
-provides. These resources are typically reserved for "expensive" resources
-in terms of memory or storage requirements, like volumes and VMs. These
-resources are **always** created via the admin role; Patrole automatically
-handles this.
-
-Like Tempest, however, Patrole must also create resources inside tests
-themselves. At the beginning of each test, the primary credentials have already
-been overridden with the admin role. One can create whatever test-level
-resources one needs, without having to worry about permissions.
-
-Test Execution
---------------
-
-Manual role override required.
-
-"Test execution" here means calling the API endpoint that enforces the policy
-action expected by the ``rbac_rule_validation`` decorator. Test execution
-should be performed *only after* calling
-``with rbac_utils.override_role(self)``.
-
-Immediately after that call, the API endpoint that enforces the policy should
-be called.
-
-Examples
-^^^^^^^^
-
-Always use the contextmanager before calling the API that enforces the
-expected policy action.
-
-Example::
-
- @rbac_rule_validation.action(
- service="nova",
- rule="os_compute_api:os-aggregates:show")
- def test_show_aggregate_rbac(self):
- # Do test setup before the ``override_role`` call.
- aggregate_id = self._create_aggregate()
- # Call the ``override_role`` method so that the primary credentials
- # have the test role needed for test execution.
- with self.rbac_utils.override_role(self):
- self.aggregates_client.show_aggregate(aggregate_id)
-
-When using a waiter, do the wait outside the contextmanager. "Waiting" always
-entails executing a ``GET`` request to the server, until the state of the
-returned resource matches a desired state. These ``GET`` requests enforce
-a different policy than the one expected. This is undesirable because
-Patrole should only test policies in isolation from one another.
-
-Otherwise, the test result will be tainted, because instead of only the
-expected policy getting enforced with the ``os_primary`` role, at least
-two policies get enforced.
-
-Example using waiter::
-
- @rbac_rule_validation.action(
- service="nova",
- rule="os_compute_api:os-admin-password")
- def test_change_server_password(self):
- original_password = self.servers_client.show_password(
- self.server['id'])
- self.addCleanup(self.servers_client.change_password, self.server['id'],
- adminPass=original_password)
-
- with self.rbac_utils.override_role(self):
- self.servers_client.change_password(
- self.server['id'], adminPass=data_utils.rand_password())
- # Call the waiter outside the ``override_role`` contextmanager, so that
- # it is executed with admin role.
- waiters.wait_for_server_status(
- self.servers_client, self.server['id'], 'ACTIVE')
-
-Below is an example of a method that enforces multiple policies getting
-called inside the contextmanager. The ``_complex_setup_method`` below
-performs the correct API that enforces the expected policy -- in this
-case ``self.resources_client.create_resource`` -- but then proceeds to
-use a waiter.
-
-Incorrect::
-
- def _complex_setup_method(self):
- resource = self.resources_client.create_resource(
- **kwargs)['resource']
- self.addCleanup(test_utils.call_and_ignore_notfound_exc,
- self._delete_resource, resource)
- waiters.wait_for_resource_status(
- self.resources_client, resource['id'], 'available')
- return resource
-
- @rbac_rule_validation.action(
- service="example-service",
- rule="example-rule")
- def test_change_server_password(self):
- # Never call a helper function inside the contextmanager that calls a
- # bunch of APIs. Only call the API that enforces the policy action
- # contained in the decorator above.
- with self.rbac_utils.override_role(self):
- self._complex_setup_method()
-
-To fix this test, see the "Example using waiter" section above. It is
-recommended to re-implement the logic in a helper method inside a test such
-that only the relevant API is called inside the contextmanager, with
-everything extraneous outside.
-
-Test Cleanup
-------------
-
-Automatic role override in background.
-
-After the test -- no matter whether it ended successfully or in failure --
-the credentials are overridden with the admin role by the Patrole framework,
-*before* ``tearDown`` or ``tearDownClass`` are called. This means that
-resources are always cleaned up using the admin role.
-
Implementation
--------------
.. automodule:: patrole_tempest_plugin.rbac_utils
:members:
:private-members:
+ :special-members:
+
+.. _Tempest credentials: https://docs.openstack.org/tempest/latest/library/credential_providers.html
+.. _dynamic credentials: https://docs.openstack.org/tempest/latest/configuration.html#dynamic-credentials
+
diff --git a/doc/source/framework/rbac_validation.rst b/doc/source/framework/rbac_validation.rst
index 186dfe2..6cd1534 100644
--- a/doc/source/framework/rbac_validation.rst
+++ b/doc/source/framework/rbac_validation.rst
@@ -17,3 +17,4 @@
.. automodule:: patrole_tempest_plugin.rbac_rule_validation
:members:
:private-members:
+ :special-members:
diff --git a/doc/source/framework/requirements_authority.rst b/doc/source/framework/requirements_authority.rst
index 6c4fcc0..628f0c0 100644
--- a/doc/source/framework/requirements_authority.rst
+++ b/doc/source/framework/requirements_authority.rst
@@ -103,3 +103,4 @@
.. automodule:: patrole_tempest_plugin.requirements_authority
:members:
:undoc-members:
+ :special-members:
diff --git a/doc/source/index.rst b/doc/source/index.rst
index 2dbf63b..c03aac6 100644
--- a/doc/source/index.rst
+++ b/doc/source/index.rst
@@ -54,6 +54,7 @@
HACKING
REVIEWING
+ test_writing_guide
Framework
---------
diff --git a/doc/source/test_writing_guide.rst b/doc/source/test_writing_guide.rst
new file mode 100644
index 0000000..d25f60a
--- /dev/null
+++ b/doc/source/test_writing_guide.rst
@@ -0,0 +1,166 @@
+Patrole Test Writing Overview
+=============================
+
+Introduction
+------------
+
+Patrole tests are broken up into 3 stages:
+
+#. :ref:`rbac-test-setup`
+#. :ref:`rbac-test-execution`
+#. :ref:`rbac-test-cleanup`
+
+See the :ref:`framework overview documentation <framework-overview>` for a
+high-level explanation of the entire testing work flow and framework
+implementation. The guide that follows is concerned with helping developers
+know how to write Patrole tests.
+
+.. _role-overriding:
+
+Role Overriding
+---------------
+
+Role overriding is the way Patrole is able to create resources and delete
+resources -- including those that require admin credentials -- while still
+being able to exercise the same set of Tempest credentials to perform the API
+action that authorizes the policy under test, by manipulating the role of
+the Tempest credentials.
+
+Patrole implicitly splits up each test into 3 stages: set up, test execution,
+and teardown.
+
+The role workflow is as follows:
+
+#. Setup: Admin role is used automatically. The primary credentials are
+ overridden with the admin role.
+#. Test execution: ``[patrole] rbac_test_role`` is used manually via the
+ call to ``with rbac_utils.override_role(self)``. Everything that
+ is executed within this contextmanager uses the primary
+ credentials overridden with the ``[patrole] rbac_test_role``.
+#. Teardown: Admin role is used automatically. The primary credentials have
+ been overridden with the admin role.
+
+.. _rbac-test-setup:
+
+Test Setup
+----------
+
+Automatic role override in background.
+
+Resources can be set up inside the ``resource_setup`` class method that Tempest
+provides. These resources are typically reserved for "expensive" resources
+in terms of memory or storage requirements, like volumes and VMs. These
+resources are **always** created via the admin role; Patrole automatically
+handles this.
+
+Like Tempest, however, Patrole must also create resources inside tests
+themselves. At the beginning of each test, the primary credentials have already
+been overridden with the admin role. One can create whatever test-level
+resources one needs, without having to worry about permissions.
+
+.. _rbac-test-execution:
+
+Test Execution
+--------------
+
+Manual role override required.
+
+"Test execution" here means calling the API endpoint that enforces the policy
+action expected by the ``rbac_rule_validation`` decorator. Test execution
+should be performed *only after* calling
+``with rbac_utils.override_role(self)``.
+
+Immediately after that call, the API endpoint that enforces the policy should
+be called.
+
+Examples
+^^^^^^^^
+
+Always use the contextmanager before calling the API that enforces the
+expected policy action.
+
+Example::
+
+ @rbac_rule_validation.action(
+ service="nova",
+ rule="os_compute_api:os-aggregates:show")
+ def test_show_aggregate_rbac(self):
+ # Do test setup before the ``override_role`` call.
+ aggregate_id = self._create_aggregate()
+ # Call the ``override_role`` method so that the primary credentials
+ # have the test role needed for test execution.
+ with self.rbac_utils.override_role(self):
+ self.aggregates_client.show_aggregate(aggregate_id)
+
+When using a waiter, do the wait outside the contextmanager. "Waiting" always
+entails executing a ``GET`` request to the server, until the state of the
+returned resource matches a desired state. These ``GET`` requests enforce
+a different policy than the one expected. This is undesirable because
+Patrole should only test policies in isolation from one another.
+
+Otherwise, the test result will be tainted, because instead of only the
+expected policy getting enforced with the ``os_primary`` role, at least
+two policies get enforced.
+
+Example using waiter::
+
+ @rbac_rule_validation.action(
+ service="nova",
+ rule="os_compute_api:os-admin-password")
+ def test_change_server_password(self):
+ original_password = self.servers_client.show_password(
+ self.server['id'])
+ self.addCleanup(self.servers_client.change_password, self.server['id'],
+ adminPass=original_password)
+
+ with self.rbac_utils.override_role(self):
+ self.servers_client.change_password(
+ self.server['id'], adminPass=data_utils.rand_password())
+ # Call the waiter outside the ``override_role`` contextmanager, so that
+ # it is executed with admin role.
+ waiters.wait_for_server_status(
+ self.servers_client, self.server['id'], 'ACTIVE')
+
+Below is an example of a method that enforces multiple policies getting
+called inside the contextmanager. The ``_complex_setup_method`` below
+performs the correct API that enforces the expected policy -- in this
+case ``self.resources_client.create_resource`` -- but then proceeds to
+use a waiter.
+
+Incorrect::
+
+ def _complex_setup_method(self):
+ resource = self.resources_client.create_resource(
+ **kwargs)['resource']
+ self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+ self._delete_resource, resource)
+ waiters.wait_for_resource_status(
+ self.resources_client, resource['id'], 'available')
+ return resource
+
+ @rbac_rule_validation.action(
+ service="example-service",
+ rule="example-rule")
+ def test_change_server_password(self):
+ # Never call a helper function inside the contextmanager that calls a
+ # bunch of APIs. Only call the API that enforces the policy action
+ # contained in the decorator above.
+ with self.rbac_utils.override_role(self):
+ self._complex_setup_method()
+
+To fix this test, see the "Example using waiter" section above. It is
+recommended to re-implement the logic in a helper method inside a test such
+that only the relevant API is called inside the contextmanager, with
+everything extraneous outside.
+
+.. _rbac-test-cleanup:
+
+Test Cleanup
+------------
+
+Automatic role override in background.
+
+After the test -- no matter whether it ended successfully or in failure --
+the credentials are overridden with the admin role by the Patrole framework,
+*before* ``tearDown`` or ``tearDownClass`` are called. This means that
+resources are always cleaned up using the admin role.
diff --git a/patrole_tempest_plugin/config.py b/patrole_tempest_plugin/config.py
index 47b76d4..56a786b 100644
--- a/patrole_tempest_plugin/config.py
+++ b/patrole_tempest_plugin/config.py
@@ -24,23 +24,14 @@
default='admin',
help="""The current RBAC role against which to run
Patrole tests."""),
- cfg.BoolOpt('enable_rbac',
- default=True,
- deprecated_for_removal=True,
- deprecated_reason="""This is a legacy option that was
-meaningful when Patrole existed downstream as a suite of tests inside Tempest.
-Installing the Patrole plugin necessarily means that RBAC tests should be run.
-This option is paradoxical with the Tempest plugin architecture.
-""",
- deprecated_since='R',
- help="Enables Patrole RBAC tests."),
cfg.ListOpt('custom_policy_files',
default=['/etc/%s/policy.json'],
help="""List of the paths to search for policy files. Each
policy path assumes that the service name is included in the path once. Also
assumes Patrole is on the same host as the policy files. The paths should be
-ordered by precedence, with high-priority paths before low-priority paths. The
-first path that is found to contain the service's policy file will be used.
+ordered by precedence, with high-priority paths before low-priority paths. All
+the paths that are found to contain the service's policy file will be used and
+all policy files will be merged. Allowed ``json`` or ``yaml`` formats.
"""),
cfg.BoolOpt('test_custom_requirements',
default=False,
@@ -171,6 +162,11 @@
help="""Are the Nova API extension policies available in the
cloud (e.g. os_compute_api:os-extended-availability-zone)? These policies were
removed in Stein because Nova API extension concept was removed in Pike."""),
+ cfg.BoolOpt('added_cinder_policies_stein',
+ default=True,
+ help="""Are the Cinder API extension policies available in the
+cloud (e.g. [create|update|get|delete]_encryption_policy)? These policies are
+added in Stein.""")
]
diff --git a/patrole_tempest_plugin/policy_authority.py b/patrole_tempest_plugin/policy_authority.py
index 3339a5d..2a49b6c 100644
--- a/patrole_tempest_plugin/policy_authority.py
+++ b/patrole_tempest_plugin/policy_authority.py
@@ -13,8 +13,9 @@
# License for the specific language governing permissions and limitations
# under the License.
+import collections
import copy
-import json
+import glob
import os
from oslo_log import log as logging
@@ -103,17 +104,14 @@
if extra_target_data is None:
extra_target_data = {}
- self.validate_service(service)
+ self.service = self.validate_service(service)
# Prioritize dynamically searching for policy files over relying on
# deprecated service-specific policy file locations.
- self.path = None
if CONF.patrole.custom_policy_files:
self.discover_policy_files()
- self.path = self.policy_files.get(service)
- self.rules = policy.Rules.load(self._get_policy_data(service),
- 'default')
+ self.rules = self.get_rules()
self.project_id = project_id
self.user_id = user_id
self.extra_target_data = extra_target_data
@@ -139,19 +137,22 @@
raise rbac_exceptions.RbacInvalidServiceException(
"%s is NOT a valid service." % service)
+ return service
+
@classmethod
def discover_policy_files(cls):
"""Dynamically discover the policy file for each service in
- ``cls.available_services``. Pick the first candidate path found
+ ``cls.available_services``. Pick all candidate paths found
out of the potential paths in ``[patrole] custom_policy_files``.
"""
if not hasattr(cls, 'policy_files'):
- cls.policy_files = {}
+ cls.policy_files = collections.defaultdict(list)
for service in cls.available_services:
for candidate_path in CONF.patrole.custom_policy_files:
- if os.path.isfile(candidate_path % service):
- cls.policy_files.setdefault(service,
- candidate_path % service)
+ path = candidate_path % service
+ for filename in glob.iglob(path):
+ if os.path.isfile(filename):
+ cls.policy_files[service].append(filename)
def allowed(self, rule_name, role):
"""Checks if a given rule in a policy is allowed with given role.
@@ -168,71 +169,60 @@
is_admin=is_admin_context)
return is_allowed
- def _get_policy_data(self, service):
- file_policy_data = {}
- mgr_policy_data = {}
- policy_data = {}
-
+ def get_rules(self):
+ rules = policy.Rules()
# Check whether policy file exists and attempt to read it.
- if self.path and os.path.isfile(self.path):
+ for path in self.policy_files[self.service]:
try:
- with open(self.path, 'r') as policy_file:
- file_policy_data = policy_file.read()
- file_policy_data = json.loads(file_policy_data)
- except (IOError, ValueError) as e:
- msg = "Failed to read policy file for service. "
- if isinstance(e, IOError):
- msg += "Please check that policy path exists."
- else:
- msg += "JSON may be improperly formatted."
- LOG.debug(msg)
- file_policy_data = {}
+ with open(path, 'r') as fp:
+ for k, v in policy.Rules.load(fp.read()).items():
+ if k not in rules:
+ rules[k] = v
+ # If the policy name and rule are the same, no
+ # ambiguity, so no reason to warn.
+ elif str(v) != str(rules[k]):
+ msg = ("The same policy name: %s was found in "
+ "multiple policies files for service %s. "
+ "This can lead to policy rule ambiguity. "
+ "Using rule: %s; Rule from file: %s")
+ LOG.warning(msg, k, self.service, rules[k], v)
+ except (ValueError, IOError):
+ LOG.warning("Failed to read policy file '%s' for service %s.",
+ path, self.service)
# Check whether policy actions are defined in code. Nova and Keystone,
# for example, define their default policy actions in code.
mgr = stevedore.named.NamedExtensionManager(
'oslo.policy.policies',
- names=[service],
- on_load_failure_callback=None,
+ names=[self.service],
invoke_on_load=True,
warn_on_missing_entrypoint=False)
if mgr:
- policy_generator = {policy.name: policy.obj for policy in mgr}
- if policy_generator and service in policy_generator:
- for rule in policy_generator[service]:
- mgr_policy_data[rule.name] = str(rule.check)
+ policy_generator = {plc.name: plc.obj for plc in mgr}
+ if self.service in policy_generator:
+ for rule in policy_generator[self.service]:
+ if rule.name not in rules:
+ rules[rule.name] = rule.check
+ elif str(rule.check) != str(rules[rule.name]):
+ msg = ("The same policy name: %s was found in the "
+ "policies files and in the code for service "
+ "%s. This can lead to policy rule ambiguity. "
+ "Using rule: %s; Rule from code: %s")
+ LOG.warning(msg, rule.name, self.service,
+ rules[rule.name], rule.check)
- # If data from both file and code exist, combine both together.
- if file_policy_data and mgr_policy_data:
- # Add the policy actions from code first.
- for action, rule in mgr_policy_data.items():
- policy_data[action] = rule
- # Overwrite with any custom policy actions defined in policy.json.
- for action, rule in file_policy_data.items():
- policy_data[action] = rule
- elif file_policy_data:
- policy_data = file_policy_data
- elif mgr_policy_data:
- policy_data = mgr_policy_data
- else:
- error_message = (
- 'Policy file for {0} service was not found among the '
+ if not rules:
+ msg = (
+ 'Policy files for {0} service were not found among the '
'registered in-code policies or in any of the possible policy '
- 'files: {1}.'.format(service,
- [loc % service for loc in
- CONF.patrole.custom_policy_files])
- )
- raise rbac_exceptions.RbacParsingException(error_message)
+ 'files: {1}.'.format(
+ self.service,
+ [loc % self.service
+ for loc in CONF.patrole.custom_policy_files]))
+ raise rbac_exceptions.RbacParsingException(msg)
- try:
- policy_data = json.dumps(policy_data)
- except (TypeError, ValueError):
- error_message = 'Policy file for {0} service is invalid.'.format(
- service)
- raise rbac_exceptions.RbacParsingException(error_message)
-
- return policy_data
+ return rules
def _is_admin_context(self, role):
"""Checks whether a role has admin context.
@@ -296,9 +286,11 @@
def _try_rule(self, apply_rule, target, access_data, o):
if apply_rule not in self.rules:
- message = ("Policy action \"{0}\" not found in policy file: {1} or"
- " among registered policy in code defaults for service."
- ).format(apply_rule, self.path)
+ message = ('Policy action "{0}" not found in policy files: '
+ '{1} or among registered policy in code defaults for '
+ '{2} service.').format(apply_rule,
+ self.policy_files[self.service],
+ self.service)
LOG.debug(message)
raise rbac_exceptions.RbacParsingException(message)
else:
diff --git a/patrole_tempest_plugin/rbac_rule_validation.py b/patrole_tempest_plugin/rbac_rule_validation.py
index a7927fc..d3b057c 100644
--- a/patrole_tempest_plugin/rbac_rule_validation.py
+++ b/patrole_tempest_plugin/rbac_rule_validation.py
@@ -38,8 +38,11 @@
RBACLOG = logging.getLogger('rbac_reporting')
-def action(service, rule='', rules=None,
- expected_error_code=_DEFAULT_ERROR_CODE, expected_error_codes=None,
+def action(service,
+ rule='',
+ rules=None,
+ expected_error_code=_DEFAULT_ERROR_CODE,
+ expected_error_codes=None,
extra_target_data=None):
"""A decorator for verifying OpenStack policy enforcement.
@@ -72,16 +75,18 @@
As such, negative and positive testing can be applied using this decorator.
:param str service: An OpenStack service. Examples: "nova" or "neutron".
- :param str rule: (DEPRECATED) A policy action defined in a policy.json file
- or in code.
- :param list rules: A list of policy actions defined in a policy.json file
+ :param rule: (DEPRECATED) A policy action defined in a policy.json file
+ or in code. Also accepts a callable that returns a policy action.
+ :type rule: str or callable
+ :param rules: A list of policy actions defined in a policy.json file
or in code. The rules are logical-ANDed together to derive the expected
- result.
+ 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 int expected_error_code: (DEPRECATED) 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
@@ -316,7 +321,11 @@
for i in range(num_rules - num_ecs):
exp_error_codes.append(_DEFAULT_ERROR_CODE)
- return rules, exp_error_codes
+ 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):
diff --git a/patrole_tempest_plugin/rbac_utils.py b/patrole_tempest_plugin/rbac_utils.py
index c928f40..366e033 100644
--- a/patrole_tempest_plugin/rbac_utils.py
+++ b/patrole_tempest_plugin/rbac_utils.py
@@ -18,7 +18,6 @@
import time
from oslo_log import log as logging
-from oslo_log import versionutils
from oslo_utils import excutils
from tempest import clients
@@ -244,18 +243,6 @@
return [cls.os_primary.auth_provider]
@classmethod
- def skip_rbac_checks(cls):
- if not CONF.patrole.enable_rbac:
- deprecation_msg = ("The `[patrole].enable_rbac` option is "
- "deprecated and will be removed in the S "
- "release. Patrole tests will always be enabled "
- "following installation of the Patrole Tempest "
- "plugin. Use a regex to skip tests")
- versionutils.report_deprecated_feature(LOG, deprecation_msg)
- raise cls.skipException(
- 'Patrole testing not enabled so skipping %s.' % cls.__name__)
-
- @classmethod
def setup_rbac_utils(cls):
cls.rbac_utils = RbacUtils(cls)
diff --git a/patrole_tempest_plugin/tests/api/README.rst b/patrole_tempest_plugin/tests/api/README.rst
new file mode 120000
index 0000000..e2853ec
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/README.rst
@@ -0,0 +1 @@
+../../../doc/source/field_guide/rbac.rst
\ No newline at end of file
diff --git a/patrole_tempest_plugin/tests/api/compute/rbac_base.py b/patrole_tempest_plugin/tests/api/compute/rbac_base.py
index 99137cc..ab4551e 100644
--- a/patrole_tempest_plugin/tests/api/compute/rbac_base.py
+++ b/patrole_tempest_plugin/tests/api/compute/rbac_base.py
@@ -22,11 +22,6 @@
compute_base.BaseV2ComputeTest):
@classmethod
- def skip_checks(cls):
- super(BaseV2ComputeRbacTest, cls).skip_checks()
- cls.skip_rbac_checks()
-
- @classmethod
def setup_clients(cls):
super(BaseV2ComputeRbacTest, cls).setup_clients()
cls.setup_rbac_utils()
diff --git a/patrole_tempest_plugin/tests/api/identity/rbac_base.py b/patrole_tempest_plugin/tests/api/identity/rbac_base.py
index d34caaa..44f5962 100644
--- a/patrole_tempest_plugin/tests/api/identity/rbac_base.py
+++ b/patrole_tempest_plugin/tests/api/identity/rbac_base.py
@@ -28,11 +28,6 @@
base.BaseIdentityTest):
@classmethod
- def skip_checks(cls):
- super(BaseIdentityRbacTest, cls).skip_checks()
- cls.skip_rbac_checks()
-
- @classmethod
def setup_clients(cls):
super(BaseIdentityRbacTest, cls).setup_clients()
cls.setup_rbac_utils()
diff --git a/patrole_tempest_plugin/tests/api/image/rbac_base.py b/patrole_tempest_plugin/tests/api/image/rbac_base.py
index 2220335..becd564 100644
--- a/patrole_tempest_plugin/tests/api/image/rbac_base.py
+++ b/patrole_tempest_plugin/tests/api/image/rbac_base.py
@@ -20,11 +20,6 @@
image_base.BaseV2ImageTest):
@classmethod
- def skip_checks(cls):
- super(BaseV2ImageRbacTest, cls).skip_checks()
- cls.skip_rbac_checks()
-
- @classmethod
def setup_clients(cls):
super(BaseV2ImageRbacTest, cls).setup_clients()
cls.setup_rbac_utils()
diff --git a/patrole_tempest_plugin/tests/api/network/README.rst b/patrole_tempest_plugin/tests/api/network/README.rst
new file mode 100644
index 0000000..20d6196
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/network/README.rst
@@ -0,0 +1,49 @@
+.. _network-rbac-tests:
+
+Network RBAC Tests
+==================
+
+What are these tests?
+---------------------
+
+These tests are RBAC tests for Neutron and its associated plugins. They are
+broken up into the following categories:
+
+* :ref:`neutron-rbac-tests`
+* :ref:`neutron-plugin-rbac-tests`
+
+.. _neutron-rbac-tests:
+
+Neutron tests
+^^^^^^^^^^^^^
+
+Neutron RBAC tests inherit from the base class ``BaseNetworkRbacTest``. They
+test many of the Neutron policies found in the service's `policy.json file`_.
+These tests are gated in many `Zuul jobs`_ (master, n-1, n-2) against many
+roles (member, admin).
+
+.. _neutron-plugin-rbac-tests:
+
+Neutron plugin tests
+^^^^^^^^^^^^^^^^^^^^
+
+The Neutron RBAC plugin tests focus on testing RBAC for various Neutron
+extensions and plugins, or, stated differently:
+
+* tests that rely on `neutron-tempest-plugin`_
+* external Neutron plugins
+
+These tests inherit from the base class ``BaseNetworkPluginRbacTest``. If an
+extension or plugin is not enabled in the cloud, the corresponding tests are
+gracefully skipped.
+
+.. note::
+
+ Patrole should import as few dependencies from ``neutron_tempest_plugin`` as
+ possible (such as ``neutron_tempest_plugin.api.clients`` for the service
+ clients) because the module is not a `stable interface`_.
+
+.. _policy.json file: https://github.com/openstack/neutron/blob/master/etc/policy.json
+.. _Zuul jobs: https://github.com/openstack/patrole/blob/master/.zuul.yaml
+.. _neutron-tempest-plugin: https://github.com/openstack/neutron-tempest-plugin
+.. _stable interface: https://github.com/openstack/neutron-tempest-plugin/tree/master/neutron_tempest_plugin#warning
diff --git a/patrole_tempest_plugin/tests/api/network/rbac_base.py b/patrole_tempest_plugin/tests/api/network/rbac_base.py
index 39ad311..6102347 100644
--- a/patrole_tempest_plugin/tests/api/network/rbac_base.py
+++ b/patrole_tempest_plugin/tests/api/network/rbac_base.py
@@ -22,11 +22,6 @@
network_base.BaseNetworkTest):
@classmethod
- def skip_checks(cls):
- super(BaseNetworkRbacTest, cls).skip_checks()
- cls.skip_rbac_checks()
-
- @classmethod
def setup_clients(cls):
super(BaseNetworkRbacTest, cls).setup_clients()
cls.setup_rbac_utils()
diff --git a/patrole_tempest_plugin/tests/api/network/test_address_scope_rbac.py b/patrole_tempest_plugin/tests/api/network/test_address_scope_rbac.py
new file mode 100644
index 0000000..cf73669
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/network/test_address_scope_rbac.py
@@ -0,0 +1,139 @@
+# Copyright 2018 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.
+
+
+from tempest.common import utils
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
+from tempest.lib import decorators
+
+from patrole_tempest_plugin import rbac_rule_validation
+from patrole_tempest_plugin.tests.api.network import rbac_base as base
+
+
+class AddressScopeRbacTest(base.BaseNetworkPluginRbacTest):
+
+ @classmethod
+ def skip_checks(cls):
+ super(AddressScopeRbacTest, cls).skip_checks()
+ if not utils.is_extension_enabled('address-scope', 'network'):
+ msg = "address-scope extension not enabled."
+ raise cls.skipException(msg)
+
+ @classmethod
+ def resource_setup(cls):
+ super(AddressScopeRbacTest, cls).resource_setup()
+ cls.network = cls.create_network()
+
+ def _create_address_scope(self, name=None, **kwargs):
+ name = name or data_utils.rand_name(self.__class__.__name__)
+ address_scope = self.ntp_client.create_address_scope(name=name,
+ ip_version=6,
+ **kwargs)
+ address_scope = address_scope['address_scope']
+ self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+ self.ntp_client.delete_address_scope,
+ address_scope['id'])
+ return address_scope
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["create_address_scope"],
+ expected_error_codes=[403])
+ @decorators.idempotent_id('8cb2d6b5-23c2-4648-997b-7a6ae55be3ad')
+ def test_create_address_scope(self):
+
+ """Create Address Scope
+
+ RBAC test for the neutron create_address_scope policy
+ """
+ with self.rbac_utils.override_role(self):
+ self._create_address_scope()
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["create_address_scope",
+ "create_address_scope:shared"],
+ expected_error_codes=[403, 403])
+ @decorators.idempotent_id('0c3f55c0-6ebe-4251-afca-62c5cb4632ca')
+ def test_create_address_scope_shared(self):
+
+ """Create Shared Address Scope
+
+ RBAC test for the neutron create_address_scope:shared policy
+ """
+ with self.rbac_utils.override_role(self):
+ self._create_address_scope(shared=True)
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_address_scope"],
+ expected_error_codes=[404])
+ @decorators.idempotent_id('a53f741b-46f6-412f-936f-ac920d449da8')
+ def test_get_address_scope(self):
+
+ """Get Address Scope
+
+ RBAC test for the neutron get_address_scope policy
+ """
+ address_scope = self._create_address_scope()
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.show_address_scope(address_scope['id'])
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_address_scope",
+ "update_address_scope"],
+ expected_error_codes=[404, 403])
+ @decorators.idempotent_id('3ce4d606-e067-4ef5-840f-96c680226e73')
+ def test_update_address_scope(self):
+
+ """Update Address Scope
+
+ RBAC test for neutron update_address_scope policy
+ """
+ address_scope = self._create_address_scope()
+ name = data_utils.rand_name(self.__class__.__name__)
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.update_address_scope(address_scope['id'],
+ name=name)
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_address_scope",
+ "update_address_scope",
+ "update_address_scope:shared"],
+ expected_error_codes=[404, 403, 403])
+ @decorators.idempotent_id('77d3a9d2-721a-4d9f-9654-6b52f113df85')
+ def test_update_address_scope_shared(self):
+
+ """Update Shared Address Scope
+
+ RBAC test for neutron update_address_scope:shared policy
+ """
+ address_scope = self._create_address_scope(shared=True)
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.update_address_scope(address_scope['id'],
+ shared=False)
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_address_scope",
+ "delete_address_scope"],
+ expected_error_codes=[404, 403])
+ @decorators.idempotent_id('277d8e47-e498-4452-b969-a91f747296ba')
+ def test_delete_address_scope(self):
+
+ """Delete Address Scope
+
+ RBAC test for neutron delete_address_scope policy
+ """
+ address_scope = self._create_address_scope()
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.delete_address_scope(address_scope['id'])
diff --git a/patrole_tempest_plugin/tests/api/network/test_dscp_marking_rule_rbac.py b/patrole_tempest_plugin/tests/api/network/test_dscp_marking_rule_rbac.py
new file mode 100644
index 0000000..b9f8365
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/network/test_dscp_marking_rule_rbac.py
@@ -0,0 +1,106 @@
+# Copyright 2018 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.
+
+from tempest.common import utils
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
+from tempest.lib import decorators
+
+from patrole_tempest_plugin import rbac_rule_validation
+from patrole_tempest_plugin.tests.api.network import rbac_base as base
+
+
+class DscpMarkingRulePluginRbacTest(base.BaseNetworkPluginRbacTest):
+
+ @classmethod
+ def skip_checks(cls):
+ super(DscpMarkingRulePluginRbacTest, cls).skip_checks()
+ if not utils.is_extension_enabled('qos', 'network'):
+ msg = "qos extension not enabled."
+ raise cls.skipException(msg)
+
+ @classmethod
+ def resource_setup(cls):
+ super(DscpMarkingRulePluginRbacTest, cls).resource_setup()
+ name = data_utils.rand_name(cls.__class__.__name__ + '-qos')
+ cls.policy_id = cls.ntp_client.create_qos_policy(
+ name=name)["policy"]["id"]
+ cls.addClassResourceCleanup(
+ cls.ntp_client.delete_qos_policy, cls.policy_id)
+
+ def create_policy_dscp_marking_rule(cls):
+ rule = cls.ntp_client.create_dscp_marking_rule(cls.policy_id, 10)
+ rule_id = rule['dscp_marking_rule']['id']
+ cls.addCleanup(
+ test_utils.call_and_ignore_notfound_exc,
+ cls.ntp_client.delete_dscp_marking_rule, cls.policy_id, rule_id)
+ return rule_id
+
+ @decorators.idempotent_id('2717AB75-E4CF-4CA4-AF04-5BEC0C808AA5')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["create_policy_dscp_marking_rule"])
+ def test_create_policy_dscp_marking_rule(self):
+ """Create policy_dscp_marking_rule.
+
+ RBAC test for the neutron "create_policy_dscp_marking_rule" policy
+ """
+
+ with self.rbac_utils.override_role(self):
+ self.create_policy_dscp_marking_rule()
+
+ @decorators.idempotent_id('3D68F50E-B948-4B25-8A72-F6F4890BBC6F')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy_dscp_marking_rule"],
+ expected_error_codes=[404])
+ def test_show_policy_dscp_marking_rule(self):
+ """Show policy_dscp_marking_rule.
+
+ RBAC test for the neutron "get_policy_dscp_marking_rule" policy
+ """
+ rule_id = self.create_policy_dscp_marking_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.show_dscp_marking_rule(self.policy_id, rule_id)
+
+ @decorators.idempotent_id('33830794-8731-45C3-BC97-17718555DD7C')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy_dscp_marking_rule",
+ "update_policy_dscp_marking_rule"],
+ expected_error_codes=[404, 403])
+ def test_update_policy_dscp_marking_rule(self):
+ """Update policy_dscp_marking_rule.
+
+ RBAC test for the neutron "update_policy_dscp_marking_rule" policy
+ """
+ rule_id = self.create_policy_dscp_marking_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.update_dscp_marking_rule(
+ self.policy_id, rule_id, dscp_mark=16)
+
+ @decorators.idempotent_id('7BF564DD-3648-4D12-8A8B-6D5E576D1843')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy_dscp_marking_rule",
+ "delete_policy_dscp_marking_rule"],
+ expected_error_codes=[404, 403])
+ def test_delete_policy_dscp_marking_rule(self):
+ """Delete policy_dscp_marking_rule.
+
+ RBAC test for the neutron "delete_policy_dscp_marking_rule" policy
+ """
+ rule_id = self.create_policy_dscp_marking_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.delete_dscp_marking_rule(self.policy_id, rule_id)
diff --git a/patrole_tempest_plugin/tests/api/network/test_policy_bandwidth_limit_rule_rbac.py b/patrole_tempest_plugin/tests/api/network/test_policy_bandwidth_limit_rule_rbac.py
new file mode 100644
index 0000000..8f9635d
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/network/test_policy_bandwidth_limit_rule_rbac.py
@@ -0,0 +1,109 @@
+# 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.
+
+from tempest.common import utils
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
+from tempest.lib import decorators
+
+from patrole_tempest_plugin import rbac_rule_validation
+from patrole_tempest_plugin.tests.api.network import rbac_base as base
+
+
+class PolicyBandwidthLimitRulePluginRbacTest(base.BaseNetworkPluginRbacTest):
+
+ @classmethod
+ def skip_checks(cls):
+ super(PolicyBandwidthLimitRulePluginRbacTest, cls).skip_checks()
+ if not utils.is_extension_enabled('qos', 'network'):
+ msg = "qos extension not enabled."
+ raise cls.skipException(msg)
+
+ @classmethod
+ def resource_setup(cls):
+ super(PolicyBandwidthLimitRulePluginRbacTest, cls).resource_setup()
+ name = data_utils.rand_name(cls.__class__.__name__ + '-qos-policy')
+ cls.policy_id = cls.ntp_client.create_qos_policy(
+ name=name)["policy"]["id"]
+ cls.addClassResourceCleanup(cls.ntp_client.delete_qos_policy,
+ cls.policy_id)
+
+ def _create_bandwidth_limit_rule(self):
+ rule = self.ntp_client.create_bandwidth_limit_rule(
+ self.policy_id, max_kbps=1000, max_burst_kbps=1000,
+ direction="egress")
+ rule_id = rule['bandwidth_limit_rule']['id']
+ self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+ self.ntp_client.delete_bandwidth_limit_rule,
+ self.policy_id, rule_id)
+ return rule_id
+
+ @decorators.idempotent_id('E0FDCB39-E16D-4AF5-9165-3FEFD116E69D')
+ @rbac_rule_validation.action(
+ service="neutron", rules=["create_policy_bandwidth_limit_rule"])
+ def test_create_policy_bandwidth_limit_rule(self):
+ """Create bandwidth_limit_rule.
+
+ RBAC test for the neutron "create_policy_bandwidth_limit_rule" policy
+ """
+
+ with self.rbac_utils.override_role(self):
+ self._create_bandwidth_limit_rule()
+
+ @decorators.idempotent_id('A092BD50-364F-4F55-B81A-37DAD6E77B95')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy_bandwidth_limit_rule"],
+ expected_error_codes=[404])
+ def test_show_policy_bandwidth_limit_rule(self):
+ """Show bandwidth_limit_rule.
+
+ RBAC test for the neutron "get_policy_bandwidth_limit_rule" policy
+ """
+ rule_id = self._create_bandwidth_limit_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.show_bandwidth_limit_rule(self.policy_id, rule_id)
+
+ @decorators.idempotent_id('CAA27599-082B-44B9-AF09-8C9B8E777ED7')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy_bandwidth_limit_rule",
+ "update_policy_bandwidth_limit_rule"],
+ expected_error_codes=[404, 403])
+ def test_update_policy_bandwidth_limit_rule(self):
+ """Update bandwidth_limit_rule.
+
+ RBAC test for the neutron "update_policy_bandwidth_limit_rule" policy
+ """
+ rule_id = self._create_bandwidth_limit_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.update_bandwidth_limit_rule(
+ self.policy_id, rule_id, max_kbps=2000)
+
+ @decorators.idempotent_id('BF6D9ED7-4B04-423D-857D-455DB0705852')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy_bandwidth_limit_rule",
+ "delete_policy_bandwidth_limit_rule"],
+ expected_error_codes=[404, 403])
+ def test_delete_policy_bandwidth_limit_rule(self):
+ """Delete bandwidth_limit_rule.
+
+ RBAC test for the neutron "delete_policy_bandwidth_limit_rule" policy
+ """
+ rule_id = self._create_bandwidth_limit_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.delete_bandwidth_limit_rule(self.policy_id,
+ rule_id)
diff --git a/patrole_tempest_plugin/tests/api/network/test_policy_minimum_bandwidth_rule_rbac.py b/patrole_tempest_plugin/tests/api/network/test_policy_minimum_bandwidth_rule_rbac.py
new file mode 100644
index 0000000..4f85cb6
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/network/test_policy_minimum_bandwidth_rule_rbac.py
@@ -0,0 +1,112 @@
+# 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.
+
+from tempest.common import utils
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
+from tempest.lib import decorators
+
+from patrole_tempest_plugin import rbac_rule_validation
+from patrole_tempest_plugin.tests.api.network import rbac_base as base
+
+
+class PolicyMinimumBandwidthRulePluginRbacTest(base.BaseNetworkPluginRbacTest):
+
+ @classmethod
+ def skip_checks(cls):
+ super(PolicyMinimumBandwidthRulePluginRbacTest, cls).skip_checks()
+ if not utils.is_extension_enabled('qos', 'network'):
+ msg = "qos extension not enabled."
+ raise cls.skipException(msg)
+
+ @classmethod
+ def resource_setup(cls):
+ super(PolicyMinimumBandwidthRulePluginRbacTest, cls).resource_setup()
+ name = data_utils.rand_name(cls.__class__.__name__ + '-qos')
+ cls.policy_id = cls.ntp_client.create_qos_policy(
+ name=name)["policy"]["id"]
+ cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
+ cls.ntp_client.delete_qos_policy,
+ cls.policy_id)
+
+ def create_minimum_bandwidth_rule(self):
+ rule = self.ntp_client.create_minimum_bandwidth_rule(
+ self.policy_id, direction="egress", min_kbps=1000)
+ rule_id = rule['minimum_bandwidth_rule']['id']
+ self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+ self.ntp_client.delete_minimum_bandwidth_rule,
+ self.policy_id, rule_id)
+ return rule_id
+
+ @decorators.idempotent_id('25B5EF3A-DF2A-4C80-A498-3BE14A321D97')
+ @rbac_rule_validation.action(
+ service="neutron", rules=["create_policy_minimum_bandwidth_rule"])
+ def test_create_policy_minimum_bandwidth_rule(self):
+ """Create policy_minimum_bandwidth_rule.
+
+ RBAC test for the neutron "create_policy_minimum_bandwidth_rule" policy
+ """
+
+ with self.rbac_utils.override_role(self):
+ self.create_minimum_bandwidth_rule()
+
+ @decorators.idempotent_id('01DD902C-47C5-45D2-9A0E-7AF05981DF21')
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy_minimum_bandwidth_rule"],
+ expected_error_codes=[404])
+ def test_show_policy_minimum_bandwidth_rule(self):
+ """Show policy_minimum_bandwidth_rule.
+
+ RBAC test for the neutron "get_policy_minimum_bandwidth_rule" policy
+ """
+ rule_id = self.create_minimum_bandwidth_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.show_minimum_bandwidth_rule(
+ self.policy_id, rule_id)
+
+ @decorators.idempotent_id('50AFE69B-455C-413A-BDC6-26B42DC8D55D')
+ @rbac_rule_validation.action(
+ service="neutron",
+ rules=["get_policy_minimum_bandwidth_rule",
+ "update_policy_minimum_bandwidth_rule"],
+ expected_error_codes=[404, 403])
+ def test_update_policy_minimum_bandwidth_rule(self):
+ """Update policy_minimum_bandwidth_rule.
+
+ RBAC test for the neutron "update_policy_minimum_bandwidth_rule" policy
+ """
+ rule_id = self.create_minimum_bandwidth_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.update_minimum_bandwidth_rule(
+ self.policy_id, rule_id, min_kbps=2000)
+
+ @decorators.idempotent_id('2112E325-C3B2-4071-8A93-B218F275A83B')
+ @rbac_rule_validation.action(
+ service="neutron",
+ rules=["get_policy_minimum_bandwidth_rule",
+ "delete_policy_minimum_bandwidth_rule"],
+ expected_error_codes=[404, 403])
+ def test_delete_policy_minimum_bandwidth_rule(self):
+ """Delete policy_minimum_bandwidth_rule.
+
+ RBAC test for the neutron "delete_policy_minimum_bandwidth_rule" policy
+ """
+ rule_id = self.create_minimum_bandwidth_rule()
+
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.delete_minimum_bandwidth_rule(
+ self.policy_id, rule_id)
diff --git a/patrole_tempest_plugin/tests/api/network/test_qos_rbac.py b/patrole_tempest_plugin/tests/api/network/test_qos_rbac.py
new file mode 100644
index 0000000..20f9e61
--- /dev/null
+++ b/patrole_tempest_plugin/tests/api/network/test_qos_rbac.py
@@ -0,0 +1,100 @@
+# Copyright 2018 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.
+
+from tempest.common import utils
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
+from tempest.lib import decorators
+
+from patrole_tempest_plugin import rbac_rule_validation
+from patrole_tempest_plugin.tests.api.network import rbac_base as base
+
+
+class QosRbacTest(base.BaseNetworkPluginRbacTest):
+
+ @classmethod
+ def skip_checks(cls):
+ super(QosRbacTest, cls).skip_checks()
+ if not utils.is_extension_enabled('qos', 'network'):
+ msg = "qos extension not enabled."
+ raise cls.skipException(msg)
+
+ @classmethod
+ def resource_setup(cls):
+ super(QosRbacTest, cls).resource_setup()
+ cls.network = cls.create_network()
+
+ def create_policy(self, name=None):
+ name = name or data_utils.rand_name(self.__class__.__name__)
+ policy = self.ntp_client.create_qos_policy(name)['policy']
+ self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+ self.ntp_client.delete_qos_policy, policy['id'])
+ return policy
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["create_policy"],
+ expected_error_codes=[403])
+ @decorators.idempotent_id('2ade2e48-7f82-4650-a69c-933d8d594636')
+ def test_create_policy(self):
+
+ """Create Policy Test
+
+ RBAC test for the neutron create_policy policy
+ """
+ with self.rbac_utils.override_role(self):
+ self.create_policy()
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy"],
+ expected_error_codes=[404])
+ @decorators.idempotent_id('d004a8de-b226-4eb4-9fdc-8202a7f64c56')
+ def test_get_policy(self):
+
+ """Show Policy Test
+
+ RBAC test for the neutron get_policy policy
+ """
+ policy = self.create_policy()
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.show_qos_policy(policy['id'])
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy", "update_policy"],
+ expected_error_codes=[404, 403])
+ @decorators.idempotent_id('fb74d56f-1dfc-490b-a9e1-454af583eefb')
+ def test_update_policy(self):
+
+ """Update Policy Test
+
+ RBAC test for the neutron update_policy policy
+ """
+ policy = self.create_policy()
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.update_qos_policy(policy['id'],
+ description='updated')
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["get_policy", "delete_policy"],
+ expected_error_codes=[404, 403])
+ @decorators.idempotent_id('ef4c23a6-4095-47a6-958e-1df585f7d8db')
+ def test_delete_policy(self):
+
+ """Delete Policy Test
+
+ RBAC test for the neutron delete_policy policy
+ """
+ policy = self.create_policy()
+ with self.rbac_utils.override_role(self):
+ self.ntp_client.delete_qos_policy(policy['id'])
diff --git a/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py b/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py
index 7d02271..62735d7 100644
--- a/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py
@@ -65,7 +65,33 @@
@rbac_rule_validation.action(service="neutron",
rules=["create_subnetpool",
- "create_subnetpool:shared"])
+ "create_subnetpool:is_default"],
+ expected_error_codes=[403, 403])
+ @decorators.idempotent_id('1b5509fd-2c32-44a8-a786-1b6ca162dbd2')
+ def test_create_subnetpool_default(self):
+ """Create default subnetpool.
+
+ RBAC test for the neutron create_subnetpool:is_default policy
+ """
+ # Most likely we already have default subnetpools for ipv4 and ipv6,
+ # so we temporary mark them as is_default=False, to let this test pass.
+ def_pools = self.subnetpools_client.list_subnetpools(is_default=True)
+ for default_pool in def_pools["subnetpools"]:
+ self.subnetpools_client.update_subnetpool(default_pool["id"],
+ is_default=False)
+
+ self.addCleanup(self.subnetpools_client.update_subnetpool,
+ default_pool["id"], is_default=True)
+
+ with self.rbac_utils.override_role(self):
+ # It apparently only enforces the policy for is_default=True.
+ # It does nothing for is_default=False
+ self._create_subnetpool(is_default=True)
+
+ @rbac_rule_validation.action(service="neutron",
+ rules=["create_subnetpool",
+ "create_subnetpool:shared"],
+ expected_error_codes=[403, 403])
@decorators.idempotent_id('cf730989-0d47-40bc-b39a-99e7de484723')
def test_create_subnetpool_shared(self):
"""Create subnetpool shared.
diff --git a/patrole_tempest_plugin/tests/api/volume/rbac_base.py b/patrole_tempest_plugin/tests/api/volume/rbac_base.py
index 8fa3b67..14b3151 100644
--- a/patrole_tempest_plugin/tests/api/volume/rbac_base.py
+++ b/patrole_tempest_plugin/tests/api/volume/rbac_base.py
@@ -29,11 +29,6 @@
_api_version = 3
@classmethod
- def skip_checks(cls):
- super(BaseVolumeRbacTest, cls).skip_checks()
- cls.skip_rbac_checks()
-
- @classmethod
def setup_clients(cls):
super(BaseVolumeRbacTest, cls).setup_clients()
cls.setup_rbac_utils()
diff --git a/patrole_tempest_plugin/tests/api/volume/test_encryption_types_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_encryption_types_rbac.py
index f10e41b..2ee80eb 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_encryption_types_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_encryption_types_rbac.py
@@ -13,12 +13,36 @@
# License for the specific language governing permissions and limitations
# under the License.
+import functools
+
from tempest.common import utils
+from tempest import config
from tempest.lib import decorators
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.volume import rbac_base
+CONF = config.CONF
+
+
+def _get_volume_type_encryption_policy(action):
+ feature_flag = CONF.policy_feature_enabled.added_cinder_policies_stein
+
+ if feature_flag:
+ return "volume_extension:volume_type_encryption:%s" % action
+
+ return "volume_extension:volume_type_encryption"
+
+
+_CREATE_VOLUME_TYPE_ENCRYPTION = functools.partial(
+ _get_volume_type_encryption_policy, "create")
+_SHOW_VOLUME_TYPE_ENCRYPTION = functools.partial(
+ _get_volume_type_encryption_policy, "get")
+_UPDATE_VOLUME_TYPE_ENCRYPTION = functools.partial(
+ _get_volume_type_encryption_policy, "update")
+_DELETE_VOLUME_TYPE_ENCRYPTION = functools.partial(
+ _get_volume_type_encryption_policy, "delete")
+
class EncryptionTypesV3RbacTest(rbac_base.BaseVolumeRbacTest):
@@ -45,7 +69,7 @@
@decorators.idempotent_id('ffd94ce5-c24b-4b6c-84c9-c5aad8c3010c')
@rbac_rule_validation.action(
service="cinder",
- rule="volume_extension:volume_type_encryption")
+ rule=_CREATE_VOLUME_TYPE_ENCRYPTION)
def test_create_volume_type_encryption(self):
vol_type_id = self.create_volume_type()['id']
with self.rbac_utils.override_role(self):
@@ -57,7 +81,7 @@
@decorators.idempotent_id('6599e72e-acef-4c0d-a9b2-463fca30d1da')
@rbac_rule_validation.action(
service="cinder",
- rule="volume_extension:volume_type_encryption")
+ rule=_DELETE_VOLUME_TYPE_ENCRYPTION)
def test_delete_volume_type_encryption(self):
vol_type_id = self._create_volume_type_encryption()
with self.rbac_utils.override_role(self):
@@ -66,7 +90,7 @@
@decorators.idempotent_id('42da9fec-32fd-4dca-9242-8a53b2fed25a')
@rbac_rule_validation.action(
service="cinder",
- rule="volume_extension:volume_type_encryption")
+ rule=_UPDATE_VOLUME_TYPE_ENCRYPTION)
def test_update_volume_type_encryption(self):
vol_type_id = self._create_volume_type_encryption()
with self.rbac_utils.override_role(self):
@@ -77,7 +101,7 @@
@decorators.idempotent_id('1381a3dc-248f-4282-b231-c9399018c804')
@rbac_rule_validation.action(
service="cinder",
- rule="volume_extension:volume_type_encryption")
+ rule=_SHOW_VOLUME_TYPE_ENCRYPTION)
def test_show_volume_type_encryption(self):
vol_type_id = self._create_volume_type_encryption()
with self.rbac_utils.override_role(self):
@@ -86,7 +110,7 @@
@decorators.idempotent_id('d4ed3cf8-52b2-4fa2-910d-e405361f0881')
@rbac_rule_validation.action(
service="cinder",
- rule="volume_extension:volume_type_encryption")
+ rule=_SHOW_VOLUME_TYPE_ENCRYPTION)
def test_show_encryption_specs_item(self):
vol_type_id = self._create_volume_type_encryption()
with self.rbac_utils.override_role(self):
diff --git a/patrole_tempest_plugin/tests/unit/resources/custom_rbac_policy.yaml b/patrole_tempest_plugin/tests/unit/resources/custom_rbac_policy.yaml
new file mode 100644
index 0000000..444bd2e
--- /dev/null
+++ b/patrole_tempest_plugin/tests/unit/resources/custom_rbac_policy.yaml
@@ -0,0 +1,13 @@
+---
+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/patrole_tempest_plugin/tests/unit/test_policy_authority.py b/patrole_tempest_plugin/tests/unit/test_policy_authority.py
index d396a29..624c0c5 100644
--- a/patrole_tempest_plugin/tests/unit/test_policy_authority.py
+++ b/patrole_tempest_plugin/tests/unit/test_policy_authority.py
@@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
-import json
import mock
import os
@@ -61,11 +60,14 @@
self.tenant_policy_file = os.path.join(current_directory,
'resources',
'tenant_rbac_policy.json')
- self.conf_policy_path = os.path.join(
+ self.conf_policy_path_json = os.path.join(
current_directory, 'resources', '%s.json')
+ self.conf_policy_path_yaml = os.path.join(
+ current_directory, 'resources', '%s.yaml')
+
self.useFixture(fixtures.ConfPatcher(
- custom_policy_files=[self.conf_policy_path], group='patrole'))
+ custom_policy_files=[self.conf_policy_path_json], group='patrole'))
self.useFixture(fixtures.ConfPatcher(
api_v3=True, api_v2=False, group='identity-feature-enabled'))
@@ -74,13 +76,18 @@
if attr in dir(policy_authority.PolicyAuthority):
delattr(policy_authority.PolicyAuthority, attr)
- def _get_fake_policy_rule(self, name, rule):
- fake_rule = mock.Mock(check=rule, __name__='foo')
- fake_rule.name = name
- return fake_rule
+ @staticmethod
+ def _get_fake_policies(rules):
+ fake_rules = []
+ rules = policy_authority.policy.Rules.from_dict(rules)
+ for name, check in rules.items():
+ fake_rule = mock.Mock(check=check, __name__='foo')
+ fake_rule.name = name
+ fake_rules.append(fake_rule)
+ return fake_rules
@mock.patch.object(policy_authority, 'LOG', autospec=True)
- def test_custom_policy(self, m_log):
+ def _test_custom_policy(self, *args):
default_roles = ['zero', 'one', 'two', 'three', 'four',
'five', 'six', 'seven', 'eight', 'nine']
@@ -105,6 +112,16 @@
for role in set(default_roles) - set(role_list):
self.assertFalse(authority.allowed(rule, role))
+ def test_custom_policy_json(self):
+ # The CONF.patrole.custom_policy_files has a path to JSON file by
+ # default, so we don't need to use ConfPatcher here.
+ self._test_custom_policy()
+
+ def test_custom_policy_yaml(self):
+ self.useFixture(fixtures.ConfPatcher(
+ custom_policy_files=[self.conf_policy_path_yaml], group='patrole'))
+ self._test_custom_policy()
+
def test_admin_policy_file_with_admin_role(self):
test_tenant_id = mock.sentinel.tenant_id
test_user_id = mock.sentinel.user_id
@@ -270,9 +287,9 @@
fake_rule = 'fake_rule'
expected_message = (
- "Policy action \"{0}\" not found in policy file: {1} or among "
- "registered policy in code defaults for service.").format(
- fake_rule, self.custom_policy_file)
+ 'Policy action "{0}" not found in policy files: {1} or among '
+ 'registered policy in code defaults for {2} service.').format(
+ fake_rule, [self.custom_policy_file], "custom_rbac_policy")
e = self.assertRaises(rbac_exceptions.RbacParsingException,
authority.allowed, fake_rule, None)
@@ -292,9 +309,10 @@
mock.sentinel.error)})
expected_message = (
- "Policy action \"{0}\" not found in policy file: {1} or among "
- "registered policy in code defaults for service.").format(
- mock.sentinel.rule, self.custom_policy_file)
+ 'Policy action "{0}" not found in policy files: {1} or among '
+ 'registered policy in code defaults for {2} service.').format(
+ mock.sentinel.rule, [self.custom_policy_file],
+ "custom_rbac_policy")
e = self.assertRaises(rbac_exceptions.RbacParsingException,
authority.allowed, mock.sentinel.rule, None)
@@ -302,18 +320,15 @@
m_log.debug.assert_called_once_with(expected_message)
@mock.patch.object(policy_authority, 'stevedore', autospec=True)
- def test_get_policy_data_from_file_and_from_code(self, mock_stevedore):
- fake_policy_rules = [
- self._get_fake_policy_rule('code_policy_action_1',
- 'rule:code_rule_1'),
- self._get_fake_policy_rule('code_policy_action_2',
- 'rule:code_rule_2'),
- self._get_fake_policy_rule('code_policy_action_3',
- 'rule:code_rule_3'),
- ]
+ def test_get_rules_from_file_and_from_code(self, mock_stevedore):
+ fake_policy_rules = self._get_fake_policies({
+ 'code_policy_action_1': 'rule:code_rule_1',
+ 'code_policy_action_2': 'rule:code_rule_2',
+ 'code_policy_action_3': 'rule:code_rule_3',
+ })
mock_manager = mock.Mock(obj=fake_policy_rules, __name__='foo')
- mock_manager.configure_mock(name='fake_service')
+ mock_manager.configure_mock(name='tenant_rbac_policy')
mock_stevedore.named.NamedExtensionManager.return_value = [
mock_manager
]
@@ -323,10 +338,10 @@
authority = policy_authority.PolicyAuthority(
test_tenant_id, test_user_id, "tenant_rbac_policy")
- policy_data = authority._get_policy_data('fake_service')
- self.assertIsInstance(policy_data, str)
+ rules = authority.get_rules()
+ self.assertIsInstance(rules, policy_authority.policy.Rules)
- actual_policy_data = json.loads(policy_data)
+ actual_policy_data = {k: str(v) for k, v in rules.items()}
expected_policy_data = {
"code_policy_action_1": "rule:code_rule_1",
"code_policy_action_2": "rule:code_rule_2",
@@ -335,26 +350,25 @@
"rule2": "tenant_id:%(tenant_id)s",
"rule3": "project_id:%(project_id)s",
"rule4": "user_id:%(user_id)s",
- "admin_tenant_rule": "role:admin and tenant_id:%(tenant_id)s",
- "admin_user_rule": "role:admin and user_id:%(user_id)s"
+ "admin_tenant_rule": "(role:admin and tenant_id:%(tenant_id)s)",
+ "admin_user_rule": "(role:admin and user_id:%(user_id)s)"
}
self.assertEqual(expected_policy_data, actual_policy_data)
@mock.patch.object(policy_authority, 'stevedore', autospec=True)
- def test_get_policy_data_from_file_and_from_code_with_overwrite(
+ def test_get_rules_from_file_and_from_code_with_overwrite(
self, mock_stevedore):
# The custom policy file should overwrite default rules rule1 and rule2
# that are defined in code.
- fake_policy_rules = [
- self._get_fake_policy_rule('rule1', 'rule:code_rule_1'),
- self._get_fake_policy_rule('rule2', 'rule:code_rule_2'),
- self._get_fake_policy_rule('code_policy_action_3',
- 'rule:code_rule_3'),
- ]
+ fake_policy_rules = self._get_fake_policies({
+ 'rule1': 'rule:code_rule_1',
+ 'rule2': 'rule:code_rule_2',
+ 'code_policy_action_3': 'rule:code_rule_3',
+ })
mock_manager = mock.Mock(obj=fake_policy_rules, __name__='foo')
- mock_manager.configure_mock(name='fake_service')
+ mock_manager.configure_mock(name='tenant_rbac_policy')
mock_stevedore.named.NamedExtensionManager.return_value = [
mock_manager
]
@@ -364,81 +378,51 @@
authority = policy_authority.PolicyAuthority(
test_tenant_id, test_user_id, 'tenant_rbac_policy')
- policy_data = authority._get_policy_data('fake_service')
- self.assertIsInstance(policy_data, str)
+ rules = authority.get_rules()
+ self.assertIsInstance(rules, policy_authority.policy.Rules)
- actual_policy_data = json.loads(policy_data)
+ actual_policy_data = {k: str(v) for k, v in rules.items()}
expected_policy_data = {
"code_policy_action_3": "rule:code_rule_3",
"rule1": "tenant_id:%(network:tenant_id)s",
"rule2": "tenant_id:%(tenant_id)s",
"rule3": "project_id:%(project_id)s",
"rule4": "user_id:%(user_id)s",
- "admin_tenant_rule": "role:admin and tenant_id:%(tenant_id)s",
- "admin_user_rule": "role:admin and user_id:%(user_id)s"
+ "admin_tenant_rule": "(role:admin and tenant_id:%(tenant_id)s)",
+ "admin_user_rule": "(role:admin and user_id:%(user_id)s)"
}
self.assertEqual(expected_policy_data, actual_policy_data)
@mock.patch.object(policy_authority, 'stevedore', autospec=True)
- def test_get_policy_data_cannot_find_policy(self, mock_stevedore):
+ def test_get_rules_cannot_find_policy(self, mock_stevedore):
mock_stevedore.named.NamedExtensionManager.return_value = None
e = self.assertRaises(rbac_exceptions.RbacParsingException,
policy_authority.PolicyAuthority,
None, None, 'test_service')
expected_error = (
- 'Policy file for {0} service was not found among the registered '
+ 'Policy files for {0} service were not found among the registered '
'in-code policies or in any of the possible policy files: {1}.'
.format('test_service',
[CONF.patrole.custom_policy_files[0] % 'test_service']))
self.assertIn(expected_error, str(e))
- @mock.patch.object(policy_authority, 'json', autospec=True)
+ @mock.patch.object(policy_authority.policy, 'parse_file_contents',
+ autospec=True)
@mock.patch.object(policy_authority, 'stevedore', autospec=True)
- def test_get_policy_data_without_valid_policy(self, mock_stevedore,
- mock_json):
- test_policy_action = mock.Mock(check='rule:bar', __name__='foo')
- test_policy_action.configure_mock(name='foo')
-
- test_policy = mock.Mock(obj=[test_policy_action], __name__='foo')
- test_policy.configure_mock(name='test_service')
-
- mock_stevedore.named.NamedExtensionManager\
- .return_value = [test_policy]
-
- mock_json.dumps.side_effect = ValueError
-
- e = self.assertRaises(rbac_exceptions.RbacParsingException,
- policy_authority.PolicyAuthority,
- None, None, 'test_service')
-
- expected_error = "Policy file for {0} service is invalid."\
- .format("test_service")
- self.assertIn(expected_error, str(e))
-
- mock_stevedore.named.NamedExtensionManager.assert_called_once_with(
- 'oslo.policy.policies',
- names=['test_service'],
- on_load_failure_callback=None,
- invoke_on_load=True,
- warn_on_missing_entrypoint=False)
-
- @mock.patch.object(policy_authority, 'json', autospec=True)
- @mock.patch.object(policy_authority, 'stevedore', autospec=True)
- def test_get_policy_data_from_file_not_json(self, mock_stevedore,
- mock_json):
+ def test_get_rules_without_valid_policy(self, mock_stevedore,
+ mock_parse_file_contents):
mock_stevedore.named.NamedExtensionManager.return_value = None
- mock_json.loads.side_effect = ValueError
+ mock_parse_file_contents.side_effect = ValueError
e = self.assertRaises(rbac_exceptions.RbacParsingException,
policy_authority.PolicyAuthority,
None, None, 'tenant_rbac_policy')
expected_error = (
- 'Policy file for {0} service was not found among the registered '
- 'in-code policies or in any of the possible policy files: {1}.'
- .format('tenant_rbac_policy', [CONF.patrole.custom_policy_files[0]
- % 'tenant_rbac_policy']))
+ 'Policy files for {0} service were not found among the registered '
+ 'in-code policies or in any of the possible policy files:'
+ .format('tenant_rbac_policy'))
self.assertIn(expected_error, str(e))
def test_discover_policy_files(self):
@@ -450,56 +434,62 @@
dir(policy_authority.PolicyAuthority))
self.assertIn('policy_files', dir(policy_parser))
self.assertIn('tenant_rbac_policy', policy_parser.policy_files)
- self.assertEqual(self.conf_policy_path % 'tenant_rbac_policy',
+ self.assertEqual([self.conf_policy_path_json % 'tenant_rbac_policy'],
policy_parser.policy_files['tenant_rbac_policy'])
@mock.patch.object(policy_authority, 'policy', autospec=True)
- @mock.patch.object(policy_authority.PolicyAuthority, '_get_policy_data',
+ @mock.patch.object(policy_authority.PolicyAuthority, 'get_rules',
autospec=True)
@mock.patch.object(policy_authority, 'clients', autospec=True)
@mock.patch.object(policy_authority, 'os', autospec=True)
- def test_discover_policy_files_with_many_invalid_one_valid(self, m_os,
- m_creds, *args):
+ @mock.patch.object(policy_authority, 'glob', autospec=True)
+ def test_discover_policy_files_with_many_invalid_one_valid(self, m_glob,
+ m_os, m_creds,
+ *args):
+ service = 'test_service'
+ custom_policy_files = ['foo/%s', 'bar/%s', 'baz/%s']
+ m_glob.iglob.side_effect = [iter([path % service])
+ for path in custom_policy_files]
# Only the 3rd path is valid.
- m_os.path.isfile.side_effect = [False, False, True, False]
+ m_os.path.isfile.side_effect = [False, False, True]
# Ensure the outer for loop runs only once in `discover_policy_files`.
m_creds.Manager().identity_services_v3_client.\
list_services.return_value = {
- 'services': [{'name': 'test_service'}]}
+ 'services': [{'name': service}]}
# The expected policy will be 'baz/test_service'.
self.useFixture(fixtures.ConfPatcher(
- custom_policy_files=['foo/%s', 'bar/%s', 'baz/%s'],
+ custom_policy_files=custom_policy_files,
group='patrole'))
policy_parser = policy_authority.PolicyAuthority(
- None, None, 'test_service')
+ None, None, service)
# Ensure that "policy_files" is set at class and instance levels.
- self.assertIn('policy_files',
- dir(policy_authority.PolicyAuthority))
- self.assertIn('policy_files', dir(policy_parser))
- self.assertIn('test_service', policy_parser.policy_files)
- self.assertEqual('baz/test_service',
- policy_parser.policy_files['test_service'])
+ self.assertTrue(hasattr(policy_authority.PolicyAuthority,
+ 'policy_files'))
+ self.assertTrue(hasattr(policy_parser, 'policy_files'))
+ self.assertEqual(['baz/%s' % service],
+ policy_parser.policy_files[service])
def test_discover_policy_files_with_no_valid_files(self):
expected_error = (
- 'Policy file for {0} service was not found among the registered '
+ 'Policy files for {0} service were not found among the registered '
'in-code policies or in any of the possible policy files: {1}.'
- .format('test_service', [self.conf_policy_path % 'test_service']))
+ .format('test_service',
+ [self.conf_policy_path_json % 'test_service']))
e = self.assertRaises(rbac_exceptions.RbacParsingException,
policy_authority.PolicyAuthority,
None, None, 'test_service')
self.assertIn(expected_error, str(e))
- self.assertIn('policy_files',
- dir(policy_authority.PolicyAuthority))
- self.assertNotIn(
- 'test_service',
- policy_authority.PolicyAuthority.policy_files.keys())
+ self.assertTrue(hasattr(policy_authority.PolicyAuthority,
+ 'policy_files'))
+ self.assertEqual(
+ [],
+ policy_authority.PolicyAuthority.policy_files['test_service'])
def _test_validate_service(self, v2_services, v3_services,
expected_failure=False, expected_services=None):
diff --git a/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py b/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py
index fe36f2c..1772047 100644
--- a/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py
+++ b/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py
@@ -14,6 +14,7 @@
from __future__ import absolute_import
+import functools
import mock
from oslo_config import cfg
@@ -80,7 +81,6 @@
pass
test_policy(self.mock_test_args)
- mock_log.warning.assert_not_called()
mock_log.error.assert_not_called()
@mock.patch.object(rbac_rv, 'LOG', autospec=True)
@@ -99,7 +99,6 @@
raise exceptions.Forbidden()
test_policy(self.mock_test_args)
- mock_log.warning.assert_not_called()
mock_log.error.assert_not_called()
@mock.patch.object(rbac_rv, 'LOG', autospec=True)
@@ -130,7 +129,8 @@
@mock.patch.object(rbac_rv, 'policy_authority', autospec=True)
def test_rule_validation_rbac_malformed_response_positive(
self, mock_authority, mock_log):
- """Test RbacMalformedResponse error is thrown without permission passes.
+ """Test RbacMalformedResponse error is thrown without permission
+ passes.
Positive test case: if RbacMalformedResponse is thrown and the user is
not allowed to perform the action, then this is a success.
@@ -143,7 +143,6 @@
raise rbac_exceptions.RbacMalformedResponse()
mock_log.error.assert_not_called()
- mock_log.warning.assert_not_called()
@mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch.object(rbac_rv, 'policy_authority', autospec=True)
@@ -171,7 +170,8 @@
@mock.patch.object(rbac_rv, 'policy_authority', autospec=True)
def test_rule_validation_rbac_conflicting_policies_positive(
self, mock_authority, mock_log):
- """Test RbacConflictingPolicies error is thrown without permission passes.
+ """Test RbacConflictingPolicies error is thrown without permission
+ passes.
Positive test case: if RbacConflictingPolicies is thrown and the user
is not allowed to perform the action, then this is a success.
@@ -184,7 +184,6 @@
raise rbac_exceptions.RbacConflictingPolicies()
mock_log.error.assert_not_called()
- mock_log.warning.assert_not_called()
@mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch.object(rbac_rv, 'policy_authority', autospec=True)
@@ -448,6 +447,66 @@
"Allowed",
"Allowed")
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
+ @mock.patch.object(rbac_rv, 'policy_authority', autospec=True)
+ def test_rule_validation_with_callable_rule(self, mock_authority,
+ mock_log):
+ """Test that a callable as the rule is evaluated correctly."""
+ mock_authority.PolicyAuthority.return_value.allowed.return_value = True
+
+ @rbac_rv.action(mock.sentinel.service,
+ rule=lambda: mock.sentinel.action)
+ def test_policy(*args):
+ pass
+
+ test_policy(self.mock_test_args)
+
+ policy_authority = mock_authority.PolicyAuthority.return_value
+ policy_authority.allowed.assert_called_with(
+ mock.sentinel.action,
+ CONF.patrole.rbac_test_role)
+
+ mock_log.error.assert_not_called()
+
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
+ @mock.patch.object(rbac_rv, 'policy_authority', autospec=True)
+ def test_rule_validation_with_conditional_callable_rule(
+ self, mock_authority, mock_log):
+ """Test that a complex callable with conditional logic as the rule is
+ evaluated correctly.
+ """
+ mock_authority.PolicyAuthority.return_value.allowed.return_value = True
+
+ def partial_func(x):
+ return "foo" if x == "bar" else "qux"
+ foo_callable = functools.partial(partial_func, "bar")
+ bar_callable = functools.partial(partial_func, "baz")
+
+ @rbac_rv.action(mock.sentinel.service,
+ rule=foo_callable)
+ def test_foo_policy(*args):
+ pass
+
+ @rbac_rv.action(mock.sentinel.service,
+ rule=bar_callable)
+ def test_bar_policy(*args):
+ pass
+
+ test_foo_policy(self.mock_test_args)
+ policy_authority = mock_authority.PolicyAuthority.return_value
+ policy_authority.allowed.assert_called_with(
+ "foo",
+ CONF.patrole.rbac_test_role)
+ policy_authority.allowed.reset_mock()
+
+ test_bar_policy(self.mock_test_args)
+ policy_authority = mock_authority.PolicyAuthority.return_value
+ policy_authority.allowed.assert_called_with(
+ "qux",
+ CONF.patrole.rbac_test_role)
+
+ mock_log.error.assert_not_called()
+
class RBACRuleValidationNegativeTest(BaseRBACRuleValidationTest):
diff --git a/patrole_tempest_plugin/tests/unit/test_rbac_utils.py b/patrole_tempest_plugin/tests/unit/test_rbac_utils.py
index c5264aa..5132079 100644
--- a/patrole_tempest_plugin/tests/unit/test_rbac_utils.py
+++ b/patrole_tempest_plugin/tests/unit/test_rbac_utils.py
@@ -208,11 +208,6 @@
class FakeRbacTest(rbac_utils.RbacUtilsMixin, test.BaseTestCase):
@classmethod
- def skip_checks(cls):
- super(FakeRbacTest, cls).skip_checks()
- cls.skip_rbac_checks()
-
- @classmethod
def setup_clients(cls):
super(FakeRbacTest, cls).setup_clients()
cls.setup_rbac_utils()
@@ -237,21 +232,3 @@
self.assertTrue(hasattr(child_test, 'rbac_utils'))
self.assertIsInstance(child_test.rbac_utils, rbac_utils.RbacUtils)
-
- def test_skip_rbac_checks(self):
- """Validate that the child class is skipped if `[patrole] enable_rbac`
- is False and that the child class's name is in the skip message.
- """
- self.useFixture(patrole_fixtures.ConfPatcher(enable_rbac=False,
- group='patrole'))
-
- class ChildRbacTest(self.parent_class):
- pass
-
- child_test = ChildRbacTest()
-
- with testtools.ExpectedException(
- testtools.TestCase.skipException,
- value_re=('Patrole testing not enabled so skipping %s.'
- % ChildRbacTest.__name__)):
- child_test.setUpClass()
diff --git a/releasenotes/notes/multiple-policy-files-9aa7f7583283739e.yaml b/releasenotes/notes/multiple-policy-files-9aa7f7583283739e.yaml
new file mode 100644
index 0000000..a3555e6
--- /dev/null
+++ b/releasenotes/notes/multiple-policy-files-9aa7f7583283739e.yaml
@@ -0,0 +1,17 @@
+---
+features:
+ - |
+ In order to implement the tests for plugins which do not maintain the
+ ``policy.json`` with full list of the policy rules and provide policy file
+ with only their own policy rules, the Patrole should be able to load and
+ merge multiple policy files for any of the services.
+
+ - Discovery all policy files for each of the services.
+ The updated ``discover_policy_files`` function picks all candidate paths
+ found out of the potential paths in the ``[patrole].custom_policy_files``
+ config option. Using ``glob.glob()`` function makes it possible to use
+ the patterns like '\*.json' to discover the policy files.
+
+ - Loading and merging a data from multiple policy files.
+ Patrole loads a data from each of the discovered policy files for a
+ service and merge the data from all files.
diff --git a/releasenotes/notes/remove-deprecated-enable-rbac-config-option-a5e46ce1053b7dea.yaml b/releasenotes/notes/remove-deprecated-enable-rbac-config-option-a5e46ce1053b7dea.yaml
new file mode 100644
index 0000000..53b1710
--- /dev/null
+++ b/releasenotes/notes/remove-deprecated-enable-rbac-config-option-a5e46ce1053b7dea.yaml
@@ -0,0 +1,5 @@
+---
+upgrade:
+ - |
+ Remove deprecated ``[patrole].enable_rbac`` configuration option. To skip
+ Patrole tests going forward, use an appropriate regex.
diff --git a/releasenotes/notes/volume-type-encryption-policy-granularity-141ac283b9c0778e.yaml b/releasenotes/notes/volume-type-encryption-policy-granularity-141ac283b9c0778e.yaml
new file mode 100644
index 0000000..4aeb107
--- /dev/null
+++ b/releasenotes/notes/volume-type-encryption-policy-granularity-141ac283b9c0778e.yaml
@@ -0,0 +1,19 @@
+---
+features:
+ - |
+ Added new Cinder feature flag (``CONF.policy_feature_enabled.added_cinder_policies_stein``)
+ for the following newly introduced granular Cinder policies:
+
+ - ``volume_extension:volume_type_encryption:create``
+ - ``volume_extension:volume_type_encryption:get``
+ - ``volume_extension:volume_type_encryption:update``
+ - ``volume_extension:volume_type_encryption:delete``
+
+ The corresponding Patrole test cases are modified to support
+ the granularity. The test cases also support backward
+ compatibility with the old single rule:
+ ``volume_extension:volume_type_encryption``
+
+ The ``rules`` parameter in ``rbac_rule_validation.action``
+ decorator now also accepts a list of callables; each callable
+ should return a policy action (str).
diff --git a/releasenotes/notes/yaml-policy-file-support-278d3edf64f98d69.yaml b/releasenotes/notes/yaml-policy-file-support-278d3edf64f98d69.yaml
new file mode 100644
index 0000000..e333377
--- /dev/null
+++ b/releasenotes/notes/yaml-policy-file-support-278d3edf64f98d69.yaml
@@ -0,0 +1,7 @@
+---
+features:
+- |
+ Patrole now supports parsing custom YAML policy files, the new policy file
+ extension since Ocata. The function ``_get_policy_data`` has been renamed to
+ ``get_rules`` and been changed to re-use ``oslo_policy.policy.Rules.load``
+ function.