Fixes policy rules in neutron containing the keyword tenant_id.
The owner policy in neutron's policy.json is "tenant_id:%(tenant_id)s".
Currently, the rbac_role_converter does not populate the target dict,
responsible for providing oslo policy with as much information
about the target or user performing an action as possible, with
the key 'tenant_id'. Consequently, the rules containing the format
outlined above fail.
This patch adds 'tenant_id' as well as 'network:tenant_id' to the
target dict, in order to work with neutron's policy.json.
Change-Id: Iecf30d714ed0e9ebde2bff6013551accdf250cef
Closes-Bug: #1661062
diff --git a/patrole_tempest_plugin/rbac_role_converter.py b/patrole_tempest_plugin/rbac_role_converter.py
index fb0d2fe..bc6e006 100644
--- a/patrole_tempest_plugin/rbac_role_converter.py
+++ b/patrole_tempest_plugin/rbac_role_converter.py
@@ -116,9 +116,8 @@
"name": role
}
],
- "project": {
- "id": self.tenant_id
- }
+ "project_id": self.tenant_id,
+ "tenant_id": self.tenant_id
}
}
return access_token
@@ -134,7 +133,6 @@
"""
access_data = copy.copy(access['token'])
access_data['roles'] = [role['name'] for role in access_data['roles']]
- access_data['project_id'] = access_data['project']['id']
access_data['is_admin'] = is_admin
# TODO(felipemonteiro): Dynamically calculate is_admin_project rather
# than hard-coding it to True. is_admin_project cannot be determined
@@ -148,7 +146,9 @@
o = Object()
o.rules = self.rules
- target = {"project_id": access_data['project_id']}
+ target = {"project_id": access_data['project_id'],
+ "tenant_id": access_data['project_id'],
+ "network:tenant_id": access_data['project_id']}
result = self._try_rule(apply_rule, target, access_data, o)
return result
diff --git a/tests/resources/tenant_rbac_policy.json b/tests/resources/tenant_rbac_policy.json
new file mode 100644
index 0000000..2647e4d
--- /dev/null
+++ b/tests/resources/tenant_rbac_policy.json
@@ -0,0 +1,6 @@
+{
+ "rule1": "tenant_id:%(network:tenant_id)s",
+ "rule2": "tenant_id:%(tenant_id)s",
+ "rule3": "project_id:%(project_id)s",
+ "admin_rule": "role:admin and tenant_id:%(tenant_id)s"
+}
\ No newline at end of file
diff --git a/tests/test_rbac_role_converter.py b/tests/test_rbac_role_converter.py
index f3a97ab..09fa081 100644
--- a/tests/test_rbac_role_converter.py
+++ b/tests/test_rbac_role_converter.py
@@ -39,6 +39,9 @@
self.alt_admin_policy_file = os.path.join(current_directory,
'resources',
'alt_admin_rbac_policy.json')
+ self.tenant_policy_file = os.path.join(current_directory,
+ 'resources',
+ 'tenant_rbac_policy.json')
@mock.patch.object(rbac_role_converter, 'LOG', autospec=True)
def test_custom_policy(self, m_log):
@@ -136,3 +139,44 @@
for rule in disallowed_rules:
allowed = converter.allowed(rule, role)
self.assertFalse(allowed)
+
+ def test_tenant_policy(self):
+ """Test whether rules with format tenant_id:%(tenant_id)s work.
+
+ Test whether Neutron rules that contain project_id, tenant_id, and
+ network:tenant_id pass.
+ """
+ test_tenant_id = mock.sentinel.tenant_id
+ converter = rbac_role_converter.RbacPolicyConverter(
+ test_tenant_id, "test", self.tenant_policy_file)
+
+ # Check whether Member role can perform expected actions.
+ allowed_rules = ['rule1', 'rule2', 'rule3']
+ for rule in allowed_rules:
+ allowed = converter.allowed(rule, 'Member')
+ self.assertTrue(allowed)
+ self.assertFalse(converter.allowed('admin_rule', 'Member'))
+
+ # Check whether admin role can perform expected actions.
+ allowed_rules.append('admin_rule')
+ for rule in allowed_rules:
+ allowed = converter.allowed(rule, 'admin')
+ self.assertTrue(allowed)
+
+ # Check whether _try_rule is called with the correct target dictionary.
+ with mock.patch.object(converter, '_try_rule', autospec=True) \
+ as mock_try_rule:
+ mock_try_rule.return_value = True
+
+ expected_target = {
+ "project_id": test_tenant_id,
+ "tenant_id": test_tenant_id,
+ "network:tenant_id": test_tenant_id
+ }
+
+ for rule in allowed_rules:
+ allowed = converter.allowed(rule, 'Member')
+ self.assertTrue(allowed)
+ mock_try_rule.assert_called_once_with(
+ rule, expected_target, mock.ANY, mock.ANY)
+ mock_try_rule.reset_mock()