Remove backward compatibility for check_is_admin

This routine in policy.py used to have a backward compatibility
check to ensure proper behaviour even when the policy.json file
did not have a specific 'context_is_admin' policy.
However, this backward compatibility check does not work. It
appears indeed that it has been broken for several release cycles;
it is also possible that actually it never worked.
When the 'context_is_admin' policy is not in the policy.json file
the enforcer simply ends up evaluating whatever is the default
policy configured there.

Therefore this patch:
- Removes the backward compatibility check, since it does not work
- Fails, for safety, check_is_admin if 'context_is_admin' policy is
  not specified
- Fixeds check_is_advsvc in the same way (the backward compatibility
  check never made any sense for this function)
- Fixes unit tests adding appropriate tests for check_is_admin and
  check_is_advsvc

Change-Id: Ia47e5781d86a3f21b9d837c9ac70a62ac435d20b
Closes-Bug: #1445690

diff --git a/neutron/policy.py b/neutron/policy.py
index 9e586dd40ba8080fdfd49444f963f322172c18f2..ab25fb54e93e947e27a83159a90d134e99953985 100644 (file)
--- a/neutron/policy.py
+++ b/neutron/policy.py
@@ -435,12 +435,9 @@ def check_is_admin(context):
     init()
     # the target is user-self
     credentials = context.to_dict()
-    target = credentials
-    # Backward compatibility: if ADMIN_CTX_POLICY is not
-    # found, default to validating role:admin
-    admin_policy = (ADMIN_CTX_POLICY if ADMIN_CTX_POLICY in _ENFORCER.rules
-                    else 'role:admin')
-    return _ENFORCER.enforce(admin_policy, target, credentials)
+    if ADMIN_CTX_POLICY not in _ENFORCER.rules:
+        return False
+    return _ENFORCER.enforce(ADMIN_CTX_POLICY, credentials, credentials)
 
 
 def check_is_advsvc(context):
@@ -448,12 +445,9 @@ def check_is_advsvc(context):
     init()
     # the target is user-self
     credentials = context.to_dict()
-    target = credentials
-    # Backward compatibility: if ADVSVC_CTX_POLICY is not
-    # found, default to validating role:advsvc
-    advsvc_policy = (ADVSVC_CTX_POLICY in _ENFORCER.rules
-                    and ADVSVC_CTX_POLICY or 'role:advsvc')
-    return _ENFORCER.enforce(advsvc_policy, target, credentials)
+    if ADVSVC_CTX_POLICY not in _ENFORCER.rules:
+        return False
+    return _ENFORCER.enforce(ADVSVC_CTX_POLICY, credentials, credentials)
 
 
 def _extract_roles(rule, roles):

diff --git a/neutron/tests/unit/test_policy.py b/neutron/tests/unit/test_policy.py
index 3888ce3429306086192c6a574bc637d2222dab07..63fc16475d4a4d7b38ae6f9795b0afd4b33a3631 100644 (file)
--- a/neutron/tests/unit/test_policy.py
+++ b/neutron/tests/unit/test_policy.py
@@ -217,8 +217,6 @@ class NeutronPolicyTestCase(base.BaseTestCase):
     def setUp(self):
         super(NeutronPolicyTestCase, self).setUp()
         policy.refresh()
-        self.admin_only_legacy = "role:admin"
-        self.admin_or_owner_legacy = "role:admin or tenant_id:%(tenant_id)s"
         # 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 {
@@ -340,6 +338,34 @@ class NeutronPolicyTestCase(base.BaseTestCase):
     def test_nonadmin_read_on_shared_succeeds(self):
         self._test_nonadmin_action_on_attr('get', 'shared', True)
 
+    def test_check_is_admin_with_admin_context_succeeds(self):
+        admin_context = context.get_admin_context()
+        self.assertTrue(policy.check_is_admin(admin_context))
+
+    def test_check_is_admin_with_user_context_fails(self):
+        self.assertFalse(policy.check_is_admin(self.context))
+
+    def test_check_is_admin_with_no_admin_policy_fails(self):
+        del self.rules[policy.ADMIN_CTX_POLICY]
+        admin_context = context.get_admin_context()
+        self.assertFalse(policy.check_is_admin(admin_context))
+
+    def test_check_is_advsvc_with_admin_context_fails(self):
+        admin_context = context.get_admin_context()
+        self.assertFalse(policy.check_is_advsvc(admin_context))
+
+    def test_check_is_advsvc_with_svc_context_suceeds(self):
+        svc_context = context.Context('', 'svc', roles=['advsvc'])
+        self.assertTrue(policy.check_is_advsvc(svc_context))
+
+    def test_check_is_advsvc_with_no_advsvc_policy_fails(self):
+        del self.rules[policy.ADVSVC_CTX_POLICY]
+        svc_context = context.Context('', 'svc', roles=['advsvc'])
+        self.assertFalse(policy.check_is_advsvc(svc_context))
+
+    def test_check_is_advsvc_with_user_context_fails(self):
+        self.assertFalse(policy.check_is_advsvc(self.context))
+
     def _test_enforce_adminonly_attribute(self, action, **kwargs):
         admin_context = context.get_admin_context()
         target = {'shared': True}
@@ -361,31 +387,12 @@ class NeutronPolicyTestCase(base.BaseTestCase):
                                            common_policy.PolicyNotAuthorized,
                                            **kwargs)
 
-    def test_enforce_adminonly_attribute_no_context_is_admin_policy(self):
-        del self.rules[policy.ADMIN_CTX_POLICY]
-        self.rules['admin_only'] = common_policy.parse_rule(
-            self.admin_only_legacy)
-        self.rules['admin_or_owner'] = common_policy.parse_rule(
-            self.admin_or_owner_legacy)
-        self._test_enforce_adminonly_attribute('create_network')
-
     def test_enforce_adminonly_attribute_nonadminctx_returns_403(self):
         action = "create_network"
         target = {'shared': True, 'tenant_id': 'somebody_else'}
         self.assertRaises(common_policy.PolicyNotAuthorized, policy.enforce,
                           self.context, action, target)
 
-    def test_enforce_adminonly_nonadminctx_no_ctx_is_admin_policy_403(self):
-        del self.rules[policy.ADMIN_CTX_POLICY]
-        self.rules['admin_only'] = common_policy.parse_rule(
-            self.admin_only_legacy)
-        self.rules['admin_or_owner'] = common_policy.parse_rule(
-            self.admin_or_owner_legacy)
-        action = "create_network"
-        target = {'shared': True, 'tenant_id': 'somebody_else'}
-        self.assertRaises(common_policy.PolicyNotAuthorized, policy.enforce,
-                          self.context, action, target)
-
     def _test_build_subattribute_match_rule(self, validate_value):
         bk = FAKE_RESOURCES['%ss' % FAKE_RESOURCE_NAME]['attr']['validate']
         FAKE_RESOURCES['%ss' % FAKE_RESOURCE_NAME]['attr']['validate'] = (