openstack/neutron
Code Issues Proposed changes

Merge "Handle properly InvalidScope exceptions to not return error 500"

Browse Source
This commit is contained in:
Zuul 2022-04-19 06:42:29 +00:00 committed by Gerrit Code Review
commit 26877af56b
4 changed files with 73 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
neutron/api/v2/base.py
View File

@ -236,7 +236,8 @@ class Controller(object):
do_authz=True,
field_list=None,
parent_id=parent_id)
except oslo_policy.PolicyNotAuthorized:
except (oslo_policy.PolicyNotAuthorized,
oslo_policy.InvalidScope):
msg = _('The resource could not be found.')
raise webob.exc.HTTPNotFound(msg)
body = kwargs.pop('body', None)
@ -388,7 +389,7 @@ class Controller(object):
field_list=field_list,
parent_id=parent_id),
fields_to_strip=added_fields)}
except oslo_policy.PolicyNotAuthorized:
except (oslo_policy.PolicyNotAuthorized, oslo_policy.InvalidScope):
# To avoid giving away information, pretend that it
# doesn't exist
msg = _('The resource could not be found.')
@ -585,7 +586,7 @@ class Controller(object):
action,
obj,
pluralized=self._collection)
except oslo_policy.PolicyNotAuthorized:
except (oslo_policy.PolicyNotAuthorized, oslo_policy.InvalidScope):
# To avoid giving away information, pretend that it
# doesn't exist if policy does not authorize SHOW
with excutils.save_and_reraise_exception() as ctxt:
@ -672,7 +673,7 @@ class Controller(object):
action,
orig_obj,
pluralized=self._collection)
except oslo_policy.PolicyNotAuthorized:
except (oslo_policy.PolicyNotAuthorized, oslo_policy.InvalidScope):
# To avoid giving away information, pretend that it
# doesn't exist if policy does not authorize SHOW
with excutils.save_and_reraise_exception() as ctxt:

4
neutron/pecan_wsgi/hooks/policy_enforcement.py
View File

@ -134,7 +134,7 @@ class PolicyHook(hooks.PecanHook):
policy.enforce(
neutron_context, action, item,
pluralized=collection)
except oslo_policy.PolicyNotAuthorized:
except (oslo_policy.PolicyNotAuthorized, oslo_policy.InvalidScope):
with excutils.save_and_reraise_exception() as ctxt:
controller = utils.get_controller(state)
# If a tenant is modifying it's own object, it's safe to
@ -191,7 +191,7 @@ class PolicyHook(hooks.PecanHook):
policy_method(neutron_context, action, item,
plugin=plugin,
pluralized=collection))]
except oslo_policy.PolicyNotAuthorized:
except (oslo_policy.PolicyNotAuthorized, oslo_policy.InvalidScope):
# This exception must be explicitly caught as the exception
# translation hook won't be called if an error occurs in the
# 'after' handler. Instead of raising an HTTPNotFound exception,

20
neutron/policy.py
View File

@ -483,10 +483,18 @@ def check(context, action, target, plugin=None, might_not_exist=False,
action,
target,
pluralized)
result = _ENFORCER.enforce(match_rule,
target,
credentials,
pluralized=pluralized)
# TODO(slaweq): this try..except.. block can be removed when bug
# https://bugs.launchpad.net/oslo.policy/+bug/1965315 will be fixed in
# oslo.policy
try:
result = _ENFORCER.enforce(match_rule,
target,
credentials,
pluralized=pluralized)
except policy.InvalidScope:
log_rule_list(match_rule)
LOG.debug("Failed policy check for '%s'", action)
result = False
return result
@ -518,10 +526,10 @@ def enforce(context, action, target, plugin=None, pluralized=None):
try:
result = _ENFORCER.enforce(rule, target, context, action=action,
do_raise=True)
except policy.PolicyNotAuthorized:
except (policy.PolicyNotAuthorized, policy.InvalidScope):
with excutils.save_and_reraise_exception():
log_rule_list(rule)
LOG.debug("Failed policy check for '%s'", action)
LOG.debug("Failed policy enforce for '%s'", action)
return result

54
neutron/tests/unit/test_policy.py
View File

@ -80,13 +80,31 @@ class PolicyTestCase(base.BaseTestCase):
"example:uppercase_admin": "role:ADMIN or role:sysadmin",
"example:only_system_admin_allowed": (
"role:admin and system_scope:all"),
"example:only_project_user_allowed": (
"role:reader and tenant_id:%(tenant_id)s")
}
policy.refresh()
self._register_default_rules()
# NOTE(vish): then overload underlying rules
policy.set_rules(oslo_policy.Rules.from_dict(rules))
self.context = context.Context('fake', 'fake', roles=['member'])
self.target = {}
def _register_default_rules(self):
rules_with_scope = [
oslo_policy.DocumentedRuleDefault(
name='get_example:only_project_user_allowed',
description="Test rule only",
check_str='role:reader and project_id:%(project_id)s',
operations=[
{
'method': 'GET',
'path': '/example',
},
],
scope_types=['project'])]
policy._ENFORCER.register_defaults(rules_with_scope)
def _test_check_system_admin_allowed_action(self, enforce_new_defaults):
action = "example:only_system_admin_allowed"
cfg.CONF.set_override(
@ -95,9 +113,11 @@ class PolicyTestCase(base.BaseTestCase):
user="fake", project_id="fake",
roles=['admin', 'member', 'reader'])
system_admin_ctx = context.Context(
user="fake", project_id="fake",
user="fake",
roles=['admin', 'member', 'reader'],
system_scope='all')
if not enforce_new_defaults:
system_admin_ctx.project_id = 'fake'
self.assertTrue(policy.check(system_admin_ctx, action, self.target))
if enforce_new_defaults:
self.assertFalse(
@ -121,9 +141,11 @@ class PolicyTestCase(base.BaseTestCase):
user="fake", project_id="fake",
roles=['admin', 'member', 'reader'])
system_admin_ctx = context.Context(
user="fake", project_id="fake",
user="fake",
roles=['admin', 'member', 'reader'],
system_scope='all')
if not enforce_new_defaults:
system_admin_ctx.project_id = 'fake'
self.assertTrue(policy.enforce(system_admin_ctx, action, self.target))
if enforce_new_defaults:
self.assertRaises(
@ -164,6 +186,19 @@ class PolicyTestCase(base.BaseTestCase):
might_not_exist=True)
self.assertTrue(result_2)
def test_check_invalid_scope(self):
cfg.CONF.set_override(
'enforce_new_defaults', True, group='oslo_policy')
cfg.CONF.set_override(
'enforce_scope', True, group='oslo_policy')
action = "get_example:only_project_user_allowed"
target = {'project_id': 'some-project'}
system_admin_ctx = context.Context(
user="fake",
roles=['admin', 'member', 'reader'],
system_scope='all')
self.assertFalse(policy.check(system_admin_ctx, action, target))
def test_enforce_good_action(self):
action = "example:allowed"
result = policy.enforce(self.context, action, self.target)
@ -184,6 +219,21 @@ class PolicyTestCase(base.BaseTestCase):
policy.enforce, self.context,
action, target)
def test_enforce_invalid_scope(self):
cfg.CONF.set_override(
'enforce_new_defaults', True, group='oslo_policy')
cfg.CONF.set_override(
'enforce_scope', True, group='oslo_policy')
action = "get_example:only_project_user_allowed"
target = {'project_id': 'some-project'}
system_admin_ctx = context.Context(
user="fake",
roles=['admin', 'member', 'reader'],
system_scope='all')
self.assertRaises(
oslo_policy.InvalidScope,
policy.enforce, system_admin_ctx, action, target)
def test_templatized_enforcement(self):
target_mine = {'tenant_id': 'fake'}
target_not_mine = {'tenant_id': 'another'}