diff --git a/neutron/conf/policies/port.py b/neutron/conf/policies/port.py index 80e3376f8b6..b13a449694c 100644 --- a/neutron/conf/policies/port.py +++ b/neutron/conf/policies/port.py @@ -68,7 +68,8 @@ rules = [ policy.DocumentedRuleDefault( 'create_port:fixed_ips', base.policy_or(base.RULE_ADVSVC, - base.RULE_ADMIN_OR_NET_OWNER), + base.RULE_ADMIN_OR_NET_OWNER, + 'rule:shared'), 'Specify ``fixed_ips`` information when creating a port', ACTION_POST ), @@ -118,6 +119,20 @@ rules = [ 'Specify ``allowed_address_pairs`` attribute when creating a port', ACTION_POST ), + policy.DocumentedRuleDefault( + 'create_port:allowed_address_pairs:mac_address', + base.RULE_ADMIN_OR_NET_OWNER, + ('Specify ``mac_address` of `allowed_address_pairs`` ' + 'attribute when creating a port'), + ACTION_POST + ), + policy.DocumentedRuleDefault( + 'create_port:allowed_address_pairs:ip_address', + base.RULE_ADMIN_OR_NET_OWNER, + ('Specify ``ip_address`` of ``allowed_address_pairs`` ' + 'attribute when creating a port'), + ACTION_POST + ), policy.DocumentedRuleDefault( 'get_port', @@ -235,6 +250,20 @@ rules = [ 'Update ``allowed_address_pairs`` attribute of a port', ACTION_PUT ), + policy.DocumentedRuleDefault( + 'update_port:allowed_address_pairs:mac_address', + base.RULE_ADMIN_OR_NET_OWNER, + ('Update ``mac_address`` of ``allowed_address_pairs`` ' + 'attribute of a port'), + ACTION_PUT + ), + policy.DocumentedRuleDefault( + 'update_port:allowed_address_pairs:ip_address', + base.RULE_ADMIN_OR_NET_OWNER, + ('Update ``ip_address`` of ``allowed_address_pairs`` ' + 'attribute of a port'), + ACTION_PUT + ), policy.DocumentedRuleDefault( 'update_port:data_plane_status', 'rule:admin_or_data_plane_int', diff --git a/neutron/policy.py b/neutron/policy.py index 6513734bc85..13a63a6f771 100644 --- a/neutron/policy.py +++ b/neutron/policy.py @@ -153,6 +153,17 @@ def _build_subattr_match_rule(attr_name, attr, action, target): return policy.AndCheck(sub_attr_rules) +def _build_list_of_subattrs_rule(attr_name, attribute_value, action): + rules = [] + for sub_attr in attribute_value: + if isinstance(sub_attr, dict): + for k in sub_attr: + rules.append(policy.RuleCheck( + 'rule', '%s:%s:%s' % (action, attr_name, k))) + if rules: + return policy.AndCheck(rules) + + def _process_rules_list(rules, match_rule): """Recursively walk a policy rule to extract a list of match entries.""" if isinstance(match_rule, policy.RuleCheck): @@ -188,8 +199,8 @@ def _build_match_rule(action, target, pluralized): target, action): attribute = res_map[resource][attribute_name] if 'enforce_policy' in attribute: - attr_rule = policy.RuleCheck('rule', '%s:%s' % - (action, attribute_name)) + attr_rule = policy.RuleCheck( + 'rule', '%s:%s' % (action, attribute_name)) # Build match entries for sub-attributes if _should_validate_sub_attributes( attribute, target[attribute_name]): @@ -197,6 +208,15 @@ def _build_match_rule(action, target, pluralized): [attr_rule, _build_subattr_match_rule( attribute_name, attribute, action, target)]) + + attribute_value = target[attribute_name] + if isinstance(attribute_value, list): + subattr_rule = _build_list_of_subattrs_rule( + attribute_name, attribute_value, action) + if subattr_rule: + attr_rule = policy.AndCheck( + [attr_rule, subattr_rule]) + match_rule = policy.AndCheck([match_rule, attr_rule]) return match_rule diff --git a/neutron/tests/unit/test_policy.py b/neutron/tests/unit/test_policy.py index 21513787e49..0b7092cd330 100644 --- a/neutron/tests/unit/test_policy.py +++ b/neutron/tests/unit/test_policy.py @@ -184,7 +184,13 @@ FAKE_RESOURCES = {"%ss" % FAKE_RESOURCE_NAME: 'validate': {'type:dict': {'sub_attr_1': {'type:string': None}, 'sub_attr_2': {'type:string': None}}} - }}, + }, + 'list_attr': {'allow_post': True, + 'allow_put': True, + 'is_visible': True, + 'default': None, + 'enforce_policy': True + }}, # special plural name "%s" % FAKE_SPECIAL_RESOURCE_NAME.replace('y', 'ies'): {'attr': {'allow_post': True, @@ -253,6 +259,10 @@ class NeutronPolicyTestCase(base.BaseTestCase): "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_resource:list_attr": "rule:admin_only_or_owner", + "create_fake_resource:list_attr:admin_element": "rule:admin_only", + "create_fake_resource:list_attr:user_element": ( + "rule:admin_or_owner"), "create_fake_policy:": "rule:admin_or_owner", } @@ -471,6 +481,23 @@ class NeutronPolicyTestCase(base.BaseTestCase): action, target) + def test_enforce_subattribute_as_list(self): + action = "create_" + FAKE_RESOURCE_NAME + target = { + 'tenant_id': 'fake', + 'list_attr': [{'user_element': 'x'}]} + result = policy.enforce(self.context, + action, target, None) + self.assertTrue(result) + + def test_enforce_subattribute_as_list_forbiden(self): + action = "create_" + FAKE_RESOURCE_NAME + target = { + 'tenant_id': 'fake', + 'list_attr': [{'admin_element': 'x'}]} + self.assertRaises(oslo_policy.PolicyNotAuthorized, policy.enforce, + self.context, action, target, None) + def test_retryrequest_on_notfound(self): failure = exceptions.NetworkNotFound(net_id='whatever') action = "create_port:mac"