openstack/patrole
Code Issues Proposed changes

Merge "Adds meaningful exceptions for missing attributes"

Browse Source
This commit is contained in:
Jenkins 2017-08-12 03:07:06 +00:00 committed by Gerrit Code Review
commit da52cc3fac
12 changed files with 115 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
patrole_tempest_plugin/rbac_exceptions.py
View File

@ -16,8 +16,25 @@
from tempest.lib import exceptions
class RbacActionFailed(exceptions.ClientRestClientException):
message = "Rbac action failed"
class RbacConflictingPolicies(exceptions.TempestException):
message = ("Conflicting policies preventing this action from being "
"performed.")
class RbacMalformedResponse(exceptions.TempestException):
message = ("The response body is missing the expected %(attribute)s due "
"to policy enforcement failure.")
def __init__(self, empty=False, extra_attr=False, **kwargs):
if empty:
self.message = ("The response body is empty due to policy "
"enforcement failure.")
kwargs = {}
if extra_attr:
self.message = ("The response body contained an unexpected "
"attribute due to policy enforcement failure.")
kwargs = {}
super(RbacMalformedResponse, self).__init__(**kwargs)
class RbacResourceSetupFailed(exceptions.TempestException):

4
patrole_tempest_plugin/rbac_rule_validation.py
View File

@ -101,7 +101,9 @@ def action(service, rule='', admin_only=False, expected_error_code=403,
LOG.error(msg)
raise exceptions.NotFound(
"%s RbacInvalidService was: %s" % (msg, e))
except (expected_exception, rbac_exceptions.RbacActionFailed) as e:
except (expected_exception,
rbac_exceptions.RbacConflictingPolicies,
rbac_exceptions.RbacMalformedResponse) as e:
if irregular_msg:
LOG.warning(irregular_msg.format(rule, service))
if allowed:

5
patrole_tempest_plugin/tests/api/compute/test_server_actions_rbac.py
View File

@ -340,6 +340,5 @@ class ServerActionsV216RbacTest(rbac_base.BaseV2ComputeRbacTest):
server = self.servers_client.show_server(self.server_id)['server']
if 'host_status' not in server:
LOG.info("host_status attribute not returned when role doesn't "
"have permission to access it.")
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(
attribute='host_status')

17
patrole_tempest_plugin/tests/api/compute/test_server_misc_policy_actions_rbac.py
View File

@ -137,8 +137,8 @@ class MiscPolicyActionsRbacTest(rbac_base.BaseV2ComputeRbacTest):
body = self.servers_client.list_servers(detail=True)['servers']
# If the first server contains "config_drive", then all the others do.
if 'config_drive' not in body[0]:
raise rbac_exceptions.RbacActionFailed(
'"config_drive" attribute not found in response body.')
raise rbac_exceptions.RbacMalformedResponse(
attribute='config_drive')
@test.requires_ext(extension='os-config-drive', service='compute')
@decorators.idempotent_id('55c62ef7-b72b-4970-acc6-05b0a4316e5d')
@ -150,8 +150,8 @@ class MiscPolicyActionsRbacTest(rbac_base.BaseV2ComputeRbacTest):
self.rbac_utils.switch_role(self, toggle_rbac_role=True)
body = self.servers_client.show_server(self.server['id'])['server']
if 'config_drive' not in body:
raise rbac_exceptions.RbacActionFailed(
'"config_drive" attribute not found in response body.')
raise rbac_exceptions.RbacMalformedResponse(
attribute="config_drive")
@test.requires_ext(extension='os-deferred-delete', service='compute')
@decorators.idempotent_id('189bfed4-1e6d-475c-bb8c-d57e60895391')
@ -192,14 +192,13 @@ class MiscPolicyActionsRbacTest(rbac_base.BaseV2ComputeRbacTest):
self.server['id'], request_id)['instanceAction']
if 'events' not in instance_action:
raise rbac_exceptions.RbacActionFailed(
'"%s" attribute not found in response body.' % 'events')
raise rbac_exceptions.RbacMalformedResponse(
attribute='events')
# Microversion 2.51+ returns 'events' always, but not 'traceback'. If
# 'traceback' is also present then policy enforcement passed.
if 'traceback' not in instance_action['events'][0]:
raise rbac_exceptions.RbacActionFailed(
'"%s" attribute not found in response body.'
% 'events.traceback')
raise rbac_exceptions.RbacMalformedResponse(
attribute='events.traceback')
@rbac_rule_validation.action(
service="nova",

2
patrole_tempest_plugin/tests/api/compute/test_server_rbac.py
View File

@ -184,7 +184,7 @@ class ComputeServersRbacTest(base.BaseV2ComputeRbacTest):
# Some other policy may have blocked it.
LOG.info("ServerFault exception caught. Some other policy "
"blocked updating of server")
raise rbac_exceptions.RbacActionFailed(e)
raise rbac_exceptions.RbacConflictingPolicies(e)
class SecurtiyGroupsRbacTest(base.BaseV2ComputeRbacTest):

10
patrole_tempest_plugin/tests/api/identity/v2/test_projects_rbac.py
View File

@ -122,10 +122,8 @@ class IdentityProjectV2AdminRbacTest(rbac_base.BaseIdentityV2AdminRbacTest):
tenant_ids = [t['id'] for t in tenants]
if admin_tenant_id not in tenant_ids:
raise rbac_exceptions.RbacActionFailed(
"The admin tenant id was not returned by the call to "
"``list_tenants``.")
raise rbac_exceptions.RbacMalformedResponse(
attribute="admin tenant id")
if non_admin_tenant_id in tenant_ids:
raise rbac_exceptions.RbacActionFailed(
"The non-admin tenant id was returned by the call to "
"``list_tenants``.")
raise rbac_exceptions.RbacMalformedResponse(
extra_attr=True)

5
patrole_tempest_plugin/tests/api/network/test_networks_multiprovider_rbac.py
View File

@ -95,6 +95,5 @@ class NetworksMultiProviderRbacTest(base.BaseNetworkRbacTest):
if len(response_network) == 0:
LOG.info("NotFound or Forbidden exception are not thrown when "
"role doesn't have access to the endpoint. Instead, "
"the response will have an empty network body. "
"This is irregular and should be fixed.")
raise rbac_exceptions.RbacActionFailed
"the response will have an empty network body.")
raise rbac_exceptions.RbacMalformedResponse(True)

6
patrole_tempest_plugin/tests/api/network/test_networks_rbac.py
View File

@ -238,7 +238,7 @@ class NetworksRbacTest(base.BaseNetworkRbacTest):
self.network['id'], **kwargs)['network']
if len(retrieved_network) == 0:
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(True)
@test.requires_ext(extension='provider', service='network')
@rbac_rule_validation.action(service="neutron",
@ -257,7 +257,7 @@ class NetworksRbacTest(base.BaseNetworkRbacTest):
self.network['id'], **kwargs)['network']
if len(retrieved_network) == 0:
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(empty=True)
@test.requires_ext(extension='provider', service='network')
@rbac_rule_validation.action(service="neutron",
@ -276,7 +276,7 @@ class NetworksRbacTest(base.BaseNetworkRbacTest):
self.network['id'], **kwargs)['network']
if len(retrieved_network) == 0:
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(empty=True)
key = retrieved_network.get('provider:segmentation_id', "NotFound")
self.assertNotEqual(key, "NotFound")

12
patrole_tempest_plugin/tests/api/network/test_ports_rbac.py
View File

@ -170,7 +170,8 @@ class PortsRbacTest(base.BaseNetworkRbacTest):
# Rather than throwing a 403, the field is not present, so raise exc.
if fields[0] not in retrieved_port:
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(
attribute='binding:vif_type')
@test.requires_ext(extension='binding', service='network')
@rbac_rule_validation.action(service="neutron",
@ -188,7 +189,8 @@ class PortsRbacTest(base.BaseNetworkRbacTest):
# Rather than throwing a 403, the field is not present, so raise exc.
if fields[0] not in retrieved_port:
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(
attribute='binding:vif_details')
@test.requires_ext(extension='binding', service='network')
@rbac_rule_validation.action(service="neutron",
@ -208,7 +210,8 @@ class PortsRbacTest(base.BaseNetworkRbacTest):
# Rather than throwing a 403, the field is not present, so raise exc.
if fields[0] not in retrieved_port:
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(
attribute='binding:host_id')
@test.requires_ext(extension='binding', service='network')
@rbac_rule_validation.action(service="neutron",
@ -229,7 +232,8 @@ class PortsRbacTest(base.BaseNetworkRbacTest):
# Rather than throwing a 403, the field is not present, so raise exc.
if fields[0] not in retrieved_port:
raise rbac_exceptions.RbacActionFailed
raise rbac_exceptions.RbacMalformedResponse(
attribute='binding:profile')
@rbac_rule_validation.action(service="neutron",
rule="update_port")

4
patrole_tempest_plugin/tests/api/network/test_routers_rbac.py
View File

@ -173,8 +173,8 @@ class RouterRbacTest(base.BaseNetworkRbacTest):
# Rather than throwing a 403, the field is not present, so raise exc.
if 'distributed' not in retrieved_fields:
raise rbac_exceptions.RbacActionFailed(
'"distributed" parameter not present in response body')
raise rbac_exceptions.RbacMalformedResponse(
attribute='distributed')
@rbac_rule_validation.action(
service="neutron", rule="update_router")

5
patrole_tempest_plugin/tests/api/volume/test_groups_rbac.py
View File

@ -128,9 +128,8 @@ class GroupTypesV3RbacTest(rbac_base.BaseVolumeRbacTest):
group_type = self.create_group_type(ignore_notfound=True)
if 'group_specs' not in group_type:
raise rbac_exceptions.RbacActionFailed(
'Policy %s does not return %s in response body.' %
('group:access_group_types_specs', 'group_specs'))
raise rbac_exceptions.RbacMalformedResponse(
attribute='group_specs')
@decorators.idempotent_id('f77f8156-4fc9-4f02-be15-8930f748e10c')
@rbac_rule_validation.action(

72
patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py
View File

@ -108,16 +108,17 @@ class RBACRuleValidationTest(base.TestCase):
@mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch.object(rbac_rv, 'rbac_policy_parser', autospec=True)
def test_rule_validation_rbac_action_failed_positive(self, mock_policy,
mock_log):
"""Test RbacActionFailed error is thrown without permission passes.
def test_rule_validation_rbac_malformed_response_positive(self,
mock_policy,
mock_log):
"""Test RbacMalformedResponse error is thrown without permission passes.
Positive test case: if RbacActionFailed is thrown and the user is not
allowed to perform the action, then this is a success.
Positive test case: if RbacMalformedResponse is thrown and the user is
not allowed to perform the action, then this is a success.
"""
decorator = rbac_rv.action(mock.sentinel.service, mock.sentinel.action)
mock_function = mock.Mock()
mock_function.side_effect = rbac_exceptions.RbacActionFailed
mock_function.side_effect = rbac_exceptions.RbacMalformedResponse
wrapper = decorator(mock_function)
mock_policy.RbacPolicyParser.return_value.allowed.return_value = False
@ -130,16 +131,65 @@ class RBACRuleValidationTest(base.TestCase):
@mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch.object(rbac_rv, 'rbac_policy_parser', autospec=True)
def test_rule_validation_rbac_action_failed_negative(self, mock_policy,
mock_log):
"""Test RbacActionFailed error is thrown with permission fails.
def test_rule_validation_rbac_malformed_response_negative(self,
mock_policy,
mock_log):
"""Test RbacMalformedResponse error is thrown with permission fails.
Negative test case: if RbacActionFailed is thrown and the user is
Negative test case: if RbacMalformedResponse is thrown and the user is
allowed to perform the action, then this is an expected failure.
"""
decorator = rbac_rv.action(mock.sentinel.service, mock.sentinel.action)
mock_function = mock.Mock()
mock_function.side_effect = rbac_exceptions.RbacActionFailed
mock_function.side_effect = rbac_exceptions.RbacMalformedResponse
wrapper = decorator(mock_function)
mock_policy.RbacPolicyParser.return_value.allowed.return_value = True
e = self.assertRaises(exceptions.Forbidden, wrapper, self.mock_args)
self.assertIn(
"Role Member was not allowed to perform sentinel.action.",
e.__str__())
mock_log.error.assert_called_once_with("Role Member was not allowed to"
" perform sentinel.action.")
@mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch.object(rbac_rv, 'rbac_policy_parser', autospec=True)
def test_rule_validation_rbac_conflicting_policies_positive(self,
mock_policy,
mock_log):
"""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.
"""
decorator = rbac_rv.action(mock.sentinel.service, mock.sentinel.action)
mock_function = mock.Mock()
mock_function.side_effect = rbac_exceptions.RbacConflictingPolicies
wrapper = decorator(mock_function)
mock_policy.RbacPolicyParser.return_value.allowed.return_value = False
result = wrapper(self.mock_args)
self.assertIsNone(result)
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, 'rbac_policy_parser', autospec=True)
def test_rule_validation_rbac_conflicting_policies_negative(self,
mock_policy,
mock_log):
"""Test RbacConflictingPolicies error is thrown with permission fails.
Negative test case: if RbacConflictingPolicies is thrown and the user
is allowed to perform the action, then this is an expected failure.
"""
decorator = rbac_rv.action(mock.sentinel.service, mock.sentinel.action)
mock_function = mock.Mock()
mock_function.side_effect = rbac_exceptions.RbacConflictingPolicies
wrapper = decorator(mock_function)
mock_policy.RbacPolicyParser.return_value.allowed.return_value = True