openstack
/
neutron
Code Issues Proposed changes

Handle properly InvalidScope exceptions to not return error 500 

When new default policy rules and scope enforcement are enabled, Neutron
needs to handle properly not only PolicyNotAuthorized exception from
oslo_policy module but also InvalidScope exception.
This patch adds handling of that exception to the neutron policy
modules.

In the check() method from the neutron.policy module we are calling
ENFORCER.enforce() method with do_raise=False which means that
PolicyNotAuthorized isn't rasised. Unfortunately it seems that there is
bug in oslo.policy module and InvalidScope is raised even with
do_raise=False.
For now, lets workaround it in Neutron by properly handling InvalidScope
exception in the check() method.
This workaround can be cleaned when bug [1] will be fixed in
oslo.policy.

[1] https://bugs.launchpad.net/oslo.policy/+bug/1965315

Partial-Bug: #1959333
Change-Id: I973f8896248c8222031c53343bb53ce48254da74
This commit is contained in:
Slawek Kaplonski 2022-01-28 12:08:32 +01:00
parent 10250fa6d8
commit 51d1899bac
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:
# If a tenant is modifying it's own object, it's safe to
# return a 403. Otherwise, pretend that it doesn't exist
@ -187,7 +187,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'}