Merge "Fix creating policy rules from subattributes."

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',

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
 

neutron/tests/unit/test_policy.py

@@ -183,7 +183,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,
@@ -252,6 +258,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",
         }
@@ -470,6 +480,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"