Merge "Add developer test writing guide for Patrole tests"
diff --git a/devstack/plugin.sh b/devstack/plugin.sh
index 01be7d6..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,6 +41,14 @@
# 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
+
+ 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/patrole_tempest_plugin/config.py b/patrole_tempest_plugin/config.py
index 4ad7f08..56a786b 100644
--- a/patrole_tempest_plugin/config.py
+++ b/patrole_tempest_plugin/config.py
@@ -31,7 +31,7 @@
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. All
the paths that are found to contain the service's policy file will be used and
-all policy files will be merged.
+all policy files will be merged. Allowed ``json`` or ``yaml`` formats.
"""),
cfg.BoolOpt('test_custom_requirements',
default=False,
@@ -162,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 27786ae..2a49b6c 100644
--- a/patrole_tempest_plugin/policy_authority.py
+++ b/patrole_tempest_plugin/policy_authority.py
@@ -16,7 +16,6 @@
import collections
import copy
import glob
-import json
import os
from oslo_log import log as logging
@@ -112,7 +111,7 @@
if CONF.patrole.custom_policy_files:
self.discover_policy_files()
- self.rules = policy.Rules.load(self._get_policy_data(), 'default')
+ self.rules = self.get_rules()
self.project_id = project_id
self.user_id = user_id
self.extra_target_data = extra_target_data
@@ -170,81 +169,60 @@
is_admin=is_admin_context)
return is_allowed
- def _get_policy_data(self):
- 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.
for path in self.policy_files[self.service]:
try:
with open(path, 'r') as fp:
- for k, v in json.load(fp).items():
- if k not in file_policy_data:
- file_policy_data[k] = v
- else:
- # If the policy name and rule are the same, no
- # ambiguity, so no reason to warn.
- if v != file_policy_data[k]:
- LOG.warning(
- "The same policy name: %s was found in "
- "multiple policies files for service %s. "
- "This can lead to policy rule ambiguity. "
- "Using rule: %s", k, self.service,
- file_policy_data[k])
- 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)
+ 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=[self.service],
- on_load_failure_callback=None,
invoke_on_load=True,
warn_on_missing_entrypoint=False)
if mgr:
policy_generator = {plc.name: plc.obj for plc in mgr}
- if policy_generator and self.service in policy_generator:
+ if self.service in policy_generator:
for rule in policy_generator[self.service]:
- mgr_policy_data[rule.name] = str(rule.check)
+ 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 = (
+ 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(self.service,
- [loc % self.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 files for {0} service are invalid.'.format(
- self.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.
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/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/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 b2af1c6..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
@@ -303,15 +320,12 @@
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='tenant_rbac_policy')
@@ -324,10 +338,10 @@
authority = policy_authority.PolicyAuthority(
test_tenant_id, test_user_id, "tenant_rbac_policy")
- policy_data = authority._get_policy_data()
- 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",
@@ -336,23 +350,22 @@
"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='tenant_rbac_policy')
@@ -365,24 +378,24 @@
authority = policy_authority.PolicyAuthority(
test_tenant_id, test_user_id, 'tenant_rbac_policy')
- policy_data = authority._get_policy_data()
- 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,
@@ -395,51 +408,21 @@
[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 files for {0} service are 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 files for {0} service were 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']))
+ '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):
@@ -451,11 +434,11 @@
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)
@@ -494,7 +477,8 @@
expected_error = (
'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,
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/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.