openstack
/
neutron
Code Issues Proposed changes

Merge "Enable to apply policies to resources with special plural"

This commit is contained in:
Jenkins 2015-04-01 08:04:45 +00:00 committed by Gerrit Code Review
parent 8305c85898 88fb463b3c
commit b025e24998
3 changed files with 102 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -144,7 +144,8 @@ class Controller(object):
context,
'%s:%s' % (self._plugin_handlers[self.SHOW], attr_name),
data,
might_not_exist=True):
might_not_exist=True,
pluralized=self._collection):
# this attribute is visible, check next one
continue
# if the code reaches this point then either the policy check
@ -199,7 +200,10 @@ class Controller(object):
arg_list.append(body)
# It is ok to raise a 403 because accessibility to the
# object was checked earlier in this method
policy.enforce(request.context, name, resource)
policy.enforce(request.context,
name,
resource,
pluralized=self._collection)
return getattr(self._plugin, name)(*arg_list, **kwargs)
return _handle_action
else:
@ -254,7 +258,8 @@ class Controller(object):
if policy.check(request.context,
self._plugin_handlers[self.SHOW],
obj,
plugin=self._plugin)]
plugin=self._plugin,
pluralized=self._collection)]
# Use the first element in the list for discriminating which attributes
# should be filtered out because of authZ policies
# fields_to_add contains a list of attributes added for request policy
@ -287,7 +292,10 @@ class Controller(object):
# FIXME(salvatore-orlando): obj_getter might return references to
# other resources. Must check authZ on them too.
if do_authz:
policy.enforce(request.context, action, obj)
policy.enforce(request.context,
action,
obj,
pluralized=self._collection)
return obj
def _send_dhcp_notification(self, context, data, methodname):
@ -398,7 +406,8 @@ class Controller(object):
item[self._resource])
policy.enforce(request.context,
action,
item[self._resource])
item[self._resource],
pluralized=self._collection)
try:
tenant_id = item[self._resource]['tenant_id']
count = quota.QUOTAS.count(request.context, self._resource,
@ -469,7 +478,8 @@ class Controller(object):
try:
policy.enforce(request.context,
action,
obj)
obj,
pluralized=self._collection)
except common_policy.PolicyNotAuthorized:
# To avoid giving away information, pretend that it
# doesn't exist
@ -524,7 +534,8 @@ class Controller(object):
try:
policy.enforce(request.context,
action,
orig_obj)
orig_obj,
pluralized=self._collection)
except common_policy.PolicyNotAuthorized:
with excutils.save_and_reraise_exception() as ctxt:
# If a tenant is modifying it's own object, it's safe to return

37
neutron/policy.py
View File

@ -78,10 +78,11 @@ def refresh():
init()
def get_resource_and_action(action):
def get_resource_and_action(action, pluralized=None):
"""Extract resource and action (write, read) from api operation."""
data = action.split(':', 1)[0].split('_', 1)
return ("%ss" % data[-1], data[0] != 'get')
resource = pluralized or ("%ss" % data[-1])
return (resource, data[0] != 'get')
def set_rules(policies, overwrite=True):
@ -183,7 +184,7 @@ def _process_rules_list(rules, match_rule):
return rules
def _build_match_rule(action, target):
def _build_match_rule(action, target, pluralized):
"""Create the rule to match for a given action.
The policy rule to be matched is built in the following way:
@ -196,7 +197,7 @@ def _build_match_rule(action, target):
(e.g.: create_router:external_gateway_info:network_id)
"""
match_rule = policy.RuleCheck('rule', action)
resource, is_write = get_resource_and_action(action)
resource, is_write = get_resource_and_action(action, pluralized)
# Attribute-based checks shall not be enforced on GETs
if is_write:
# assigning to variable with short name for improving readability
@ -346,12 +347,12 @@ class FieldCheck(policy.Check):
return target_value == self.value
def _prepare_check(context, action, target):
def _prepare_check(context, action, target, pluralized):
"""Prepare rule, target, and credentials for the policy engine."""
# Compare with None to distinguish case in which target is {}
if target is None:
target = {}
match_rule = _build_match_rule(action, target)
match_rule = _build_match_rule(action, target, pluralized)
credentials = context.to_dict()
return match_rule, target, credentials
@ -362,7 +363,8 @@ def log_rule_list(match_rule):
LOG.debug("Enforcing rules: %s", rules)
def check(context, action, target, plugin=None, might_not_exist=False):
def check(context, action, target, plugin=None, might_not_exist=False,
pluralized=None):
"""Verifies that the action is valid on the target in this context.
:param context: neutron context
@ -376,20 +378,28 @@ def check(context, action, target, plugin=None, might_not_exist=False):
:param might_not_exist: If True the policy check is skipped (and the
function returns True) if the specified policy does not exist.
Defaults to false.
:param pluralized: pluralized case of resource
e.g. firewall_policy -> pluralized = "firewall_policies"
:return: Returns True if access is permitted else False.
"""
if might_not_exist and not (_ENFORCER.rules and action in _ENFORCER.rules):
return True
match_rule, target, credentials = _prepare_check(context, action, target)
result = _ENFORCER.enforce(match_rule, target, credentials)
match_rule, target, credentials = _prepare_check(context,
action,
target,
pluralized)
result = _ENFORCER.enforce(match_rule,
target,
credentials,
pluralized=pluralized)
# logging applied rules in case of failure
if not result:
log_rule_list(match_rule)
return result
def enforce(context, action, target, plugin=None):
def enforce(context, action, target, plugin=None, pluralized=None):
"""Verifies that the action is valid on the target in this context.
:param context: neutron context
@ -400,11 +410,16 @@ def enforce(context, action, target, plugin=None):
location of the object e.g. ``{'project_id': context.project_id}``
:param plugin: currently unused and deprecated.
Kept for backward compatibility.
:param pluralized: pluralized case of resource
e.g. firewall_policy -> pluralized = "firewall_policies"
:raises neutron.openstack.common.policy.PolicyNotAuthorized:
if verification fails.
"""
rule, target, credentials = _prepare_check(context, action, target)
rule, target, credentials = _prepare_check(context,
action,
target,
pluralized)
try:
result = _ENFORCER.enforce(rule, target, credentials, action=action,
do_raise=True)

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

@ -183,17 +183,29 @@ class DefaultPolicyTestCase(base.BaseTestCase):
policy.enforce(self.context, "example:noexist", {})
FAKE_RESOURCE_NAME = 'something'
FAKE_RESOURCE = {"%ss" % FAKE_RESOURCE_NAME:
{'attr': {'allow_post': True,
'allow_put': True,
'is_visible': True,
'default': None,
'enforce_policy': True,
'validate': {'type:dict':
{'sub_attr_1': {'type:string': None},
'sub_attr_2': {'type:string': None}}}
}}}
FAKE_RESOURCE_NAME = 'fake_resource'
FAKE_SPECIAL_RESOURCE_NAME = 'fake_policy'
FAKE_RESOURCES = {"%ss" % FAKE_RESOURCE_NAME:
{'attr': {'allow_post': True,
'allow_put': True,
'is_visible': True,
'default': None,
'enforce_policy': True,
'validate': {'type:dict':
{'sub_attr_1': {'type:string': None},
'sub_attr_2': {'type:string': None}}}
}},
# special plural name
"%s" % FAKE_SPECIAL_RESOURCE_NAME.replace('y', 'ies'):
{'attr': {'allow_post': True,
'allow_put': True,
'is_visible': True,
'default': None,
'enforce_policy': True,
'validate': {'type:dict':
{'sub_attr_1': {'type:string': None},
'sub_attr_2': {'type:string': None}}}
}}}
class NeutronPolicyTestCase(base.BaseTestCase):
@ -207,8 +219,8 @@ class NeutronPolicyTestCase(base.BaseTestCase):
policy.refresh()
self.admin_only_legacy = "role:admin"
self.admin_or_owner_legacy = "role:admin or tenant_id:%(tenant_id)s"
# Add a Fake 'something' resource to RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP.update(FAKE_RESOURCE)
# Add Fake resources to RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP.update(FAKE_RESOURCES)
self.rules = dict((k, common_policy.parse_rule(v)) for k, v in {
"context_is_admin": "role:admin",
"context_is_advsvc": "role:advsvc",
@ -234,11 +246,12 @@ class NeutronPolicyTestCase(base.BaseTestCase):
"update_port": "rule:admin_or_owner or rule:context_is_advsvc",
"get_port": "rule:admin_or_owner or rule:context_is_advsvc",
"delete_port": "rule:admin_or_owner or rule:context_is_advsvc",
"create_something": "rule:admin_or_owner",
"create_something:attr": "rule:admin_or_owner",
"create_something:attr:sub_attr_1": "rule:admin_or_owner",
"create_something:attr:sub_attr_2": "rule:admin_only",
"create_fake_resource": "rule:admin_or_owner",
"create_fake_resource:attr": "rule:admin_or_owner",
"create_fake_resource:attr:sub_attr_1": "rule:admin_or_owner",
"create_fake_resource:attr:sub_attr_2": "rule:admin_only",
"create_fake_policy:": "rule:admin_or_owner",
"get_firewall_policy": "rule:admin_or_owner or "
"rule:shared",
"get_firewall_rule": "rule:admin_or_owner or "
@ -374,17 +387,17 @@ class NeutronPolicyTestCase(base.BaseTestCase):
self.context, action, target)
def _test_build_subattribute_match_rule(self, validate_value):
bk = FAKE_RESOURCE['%ss' % FAKE_RESOURCE_NAME]['attr']['validate']
FAKE_RESOURCE['%ss' % FAKE_RESOURCE_NAME]['attr']['validate'] = (
bk = FAKE_RESOURCES['%ss' % FAKE_RESOURCE_NAME]['attr']['validate']
FAKE_RESOURCES['%ss' % FAKE_RESOURCE_NAME]['attr']['validate'] = (
validate_value)
action = "create_something"
action = "create_" + FAKE_RESOURCE_NAME
target = {'tenant_id': 'fake', 'attr': {'sub_attr_1': 'x'}}
self.assertFalse(policy._build_subattr_match_rule(
'attr',
FAKE_RESOURCE['%ss' % FAKE_RESOURCE_NAME]['attr'],
FAKE_RESOURCES['%ss' % FAKE_RESOURCE_NAME]['attr'],
action,
target))
FAKE_RESOURCE['%ss' % FAKE_RESOURCE_NAME]['attr']['validate'] = bk
FAKE_RESOURCES['%ss' % FAKE_RESOURCE_NAME]['attr']['validate'] = bk
def test_build_subattribute_match_rule_empty_dict_validator(self):
self._test_build_subattribute_match_rule({})
@ -393,14 +406,27 @@ class NeutronPolicyTestCase(base.BaseTestCase):
self._test_build_subattribute_match_rule(
{'type:dict': 'wrong_stuff'})
def test_build_match_rule_special_pluralized(self):
action = "create_" + FAKE_SPECIAL_RESOURCE_NAME
pluralized = "create_fake_policies"
target = {}
result = policy._build_match_rule(action, target, pluralized)
self.assertEqual("rule:" + action, str(result))
def test_build_match_rule_normal_pluralized_when_create(self):
action = "create_" + FAKE_RESOURCE_NAME
target = {}
result = policy._build_match_rule(action, target, None)
self.assertEqual("rule:" + action, str(result))
def test_enforce_subattribute(self):
action = "create_something"
action = "create_" + FAKE_RESOURCE_NAME
target = {'tenant_id': 'fake', 'attr': {'sub_attr_1': 'x'}}
result = policy.enforce(self.context, action, target, None)
self.assertEqual(result, True)
def test_enforce_admin_only_subattribute(self):
action = "create_something"
action = "create_" + FAKE_RESOURCE_NAME
target = {'tenant_id': 'fake', 'attr': {'sub_attr_1': 'x',
'sub_attr_2': 'y'}}
result = policy.enforce(context.get_admin_context(),
@ -408,7 +434,7 @@ class NeutronPolicyTestCase(base.BaseTestCase):
self.assertEqual(result, True)
def test_enforce_admin_only_subattribute_nonadminctx_returns_403(self):
action = "create_something"
action = "create_" + FAKE_RESOURCE_NAME
target = {'tenant_id': 'fake', 'attr': {'sub_attr_1': 'x',
'sub_attr_2': 'y'}}
self.assertRaises(common_policy.PolicyNotAuthorized, policy.enforce,
@ -576,11 +602,12 @@ class NeutronPolicyTestCase(base.BaseTestCase):
expected_policies))
def test_process_rules(self):
action = "create_something"
action = "create_" + FAKE_RESOURCE_NAME
# Construct RuleChecks for an action, attribute and subattribute
match_rule = common_policy.RuleCheck('rule', action)
attr_rule = common_policy.RuleCheck('rule', '%s:%s' %
(action, 'somethings'))
attr_rule = common_policy.RuleCheck('rule', '%s:%ss' %
(action,
FAKE_RESOURCE_NAME))
sub_attr_rules = [common_policy.RuleCheck('rule', '%s:%s:%s' %
(action, 'attr',
'sub_attr_1'))]
@ -593,8 +620,9 @@ class NeutronPolicyTestCase(base.BaseTestCase):
match_rule = common_policy.AndCheck([match_rule, attr_rule])
# Assert that the rules are correctly extracted from the match_rule
rules = policy._process_rules_list([], match_rule)
self.assertEqual(['create_something', 'create_something:somethings',
'create_something:attr:sub_attr_1'], rules)
self.assertEqual(['create_fake_resource',
'create_fake_resource:fake_resources',
'create_fake_resource:attr:sub_attr_1'], rules)
def test_log_rule_list(self):
with contextlib.nested(