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_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_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(exceptions.PolicyNotAuthorized, policy.enforce,
                       self.context, action, target)

 def test_get_roles_context_is_admin_rule_missing(self):
     rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         "some_other_rule": "role:admin",
     }.items())
     common_policy.set_rules(common_policy.Rules(rules))
     # 'admin' role is expected for bw compatibility
     self.assertEqual(['admin'], policy.get_admin_roles())

 def test_get_roles_with_rule_check(self):
     rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         policy.ADMIN_CTX_POLICY: "rule:some_other_rule",
         "some_other_rule": "role:admin",
     }.items())
     common_policy.set_rules(common_policy.Rules(rules))
     self.assertEqual(['admin'], policy.get_admin_roles())

 def _test_enforce_tenant_id_raises(self, bad_rule):
     self.rules["admin_or_owner"] = common_policy.parse_rule(bad_rule)
     # Trigger a policy with rule admin_or_owner
     action = "create_network"
     target = {"tenant_id": "fake"}
     policy.init()
     self.assertRaises(exceptions.PolicyCheckError, policy.enforce, self.context, action, target)

    def setUp(self):
        super(NeutronPolicyTestCase, self).setUp()
        policy.refresh()
        # 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",
            "admin_or_network_owner": "rule:context_is_admin or "
                                      "tenant_id:%(network:tenant_id)s",
            "admin_or_owner": ("rule:context_is_admin or "
                               "tenant_id:%(tenant_id)s"),
            "admin_only": "rule:context_is_admin",
            "regular_user": "role:user",
            "shared": "field:networks:shared=True",
            "external": "field:networks:router:external=True",
            "default": '@',

            "create_network": "rule:admin_or_owner",
            "create_network:shared": "rule:admin_only",
            "update_network": '@',
            "update_network:shared": "rule:admin_only",
            "get_network": "rule:admin_or_owner or rule:shared or "
                           "rule:external or rule:context_is_advsvc",
            "create_subnet": "rule:admin_or_network_owner",
            "create_port:mac": "rule:admin_or_network_owner or "
                               "rule:context_is_advsvc",
            "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_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 "
                            "rule:shared",

            "insert_rule": "rule:admin_or_owner",
            "remove_rule": "rule:admin_or_owner",
        }.items())

        def remove_fake_resource():
            del attributes.RESOURCE_ATTRIBUTE_MAP["%ss" % FAKE_RESOURCE_NAME]

        self.patcher = mock.patch.object(neutron.policy,
                                         'init',
                                         new=self.fakepolicyinit)
        self.patcher.start()
        self.addCleanup(remove_fake_resource)
        self.context = context.Context('fake', 'fake', roles=['user'])
        plugin_klass = importutils.import_class(
            "neutron.db.db_base_plugin_v2.NeutronDbPluginV2")
        self.manager_patcher = mock.patch('neutron.manager.NeutronManager')
        fake_manager = self.manager_patcher.start()
        fake_manager_instance = fake_manager.return_value
        fake_manager_instance.plugin = plugin_klass()

 def test_get_roles_with_or_check(self):
     self.rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         policy.ADMIN_CTX_POLICY: "rule:rule1 or rule:rule2",
         "rule1": "role:admin_1",
         "rule2": "role:admin_2"
     }.items())
     self.assertEqual(['admin_1', 'admin_2'],
                      policy.get_admin_roles())

    def test_enforce_tenant_id_check_parent_resource_bw_compatibility(self):
        def fakegetnetwork(*args, **kwargs):
            return {"tenant_id": "fake"}

        del self.rules["admin_or_network_owner"]
        self.rules["admin_or_network_owner"] = common_policy.parse_rule("role:admin or tenant_id:%(network_tenant_id)s")
        action = "create_port:mac"
        with mock.patch.object(manager.NeutronManager.get_instance().plugin, "get_network", new=fakegetnetwork):
            target = {"network_id": "whatever"}
            result = policy.enforce(self.context, action, target)
            self.assertTrue(result)

    def setUp(self):
        super(NeutronPolicyTestCase, self).setUp()
        policy.reset()
        policy.init()
        self.addCleanup(policy.reset)
        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)
        self.rules = dict(
            (k, common_policy.parse_rule(v))
            for k, v in {
                "context_is_admin": "role:admin",
                "admin_or_network_owner": "rule:context_is_admin or " "tenant_id:%(network:tenant_id)s",
                "admin_or_owner": ("rule:context_is_admin or " "tenant_id:%(tenant_id)s"),
                "admin_only": "rule:context_is_admin",
                "regular_user": "role:user",
                "shared": "field:networks:shared=True",
                "external": "field:networks:router:external=True",
                "default": "@",
                "create_network": "rule:admin_or_owner",
                "create_network:shared": "rule:admin_only",
                "update_network": "@",
                "update_network:shared": "rule:admin_only",
                "get_network": "rule:admin_or_owner or " "rule:shared or " "rule:external",
                "create_port:mac": "rule:admin_or_network_owner",
                "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",
                "get_firewall_policy": "rule:admin_or_owner or " "rule:shared",
                "get_firewall_rule": "rule:admin_or_owner or " "rule:shared",
            }.items()
        )

        def fakepolicyinit():
            common_policy.set_rules(common_policy.Rules(self.rules))

        def remove_fake_resource():
            del attributes.RESOURCE_ATTRIBUTE_MAP["%ss" % FAKE_RESOURCE_NAME]

        self.patcher = mock.patch.object(neutron.policy, "init", new=fakepolicyinit)
        self.patcher.start()
        self.addCleanup(self.patcher.stop)
        self.addCleanup(remove_fake_resource)
        self.context = context.Context("fake", "fake", roles=["user"])
        plugin_klass = importutils.import_class("neutron.db.db_base_plugin_v2.NeutronDbPluginV2")
        self.manager_patcher = mock.patch("neutron.manager.NeutronManager")
        fake_manager = self.manager_patcher.start()
        fake_manager_instance = fake_manager.return_value
        fake_manager_instance.plugin = plugin_klass()
        self.addCleanup(self.manager_patcher.stop)

 def setUp(self):
     super(UOSExtensionPolicyTestCase, self).setUp()
     policy.reset()
     policy.init()
     rules = {
         "associate_floatingip_router": "not role:project_observer",
         "get_router_details": "role:admin",
         "remove_router_portforwarding": "role:member"
     }
     common_policy.set_rules(common_policy.Rules(
         dict((k, common_policy.parse_rule(v))
              for k, v in rules.items())))
     self.context = context.Context('fake', 'fake', roles=['member'])
     self.request = FakeRequest(self.context)
     self.target = {}
     self.controller = uos.UosController()

 def test_policy_404(self):
     with self.subnet(cidr='12.0.0.0/24') as public_sub:
         self._set_net_external(public_sub['subnet']['network_id'])
         fip = self._make_floatingip(
             self.fmt,
             public_sub['subnet']['network_id'])
         policy.reset()
         policy.init()
         rules = {
             "delete_floatingip": "role:admin_only"
         }
         common_policy.set_rules(common_policy.Rules(
             dict((k, common_policy.parse_rule(v))
             for k, v in rules.items())))
         fip_id = fip['floatingip']['id']
         self.context = context.Context('fake', 'fake', roles=['member'])
         req = self.new_delete_request('floatingips', fip_id)
         req.environ['neutron.context'] = self.context
         res = req.get_response(self._api_for_resource('floatingips'))
         self.assertEqual(404, res.status_int)
         policy.reset()
         policy.init()
         self._delete('floatingips', fip_id)

 def test_get_roles_with_other_rules(self):
     self.rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         policy.ADMIN_CTX_POLICY: "role:xxx or other:value",
     }.items())
     self.assertEqual(['xxx'], policy.get_admin_roles())

 def test_get_roles_with_role_check(self):
     rules = dict((k, common_policy.parse_rule(v)) for k, v in {policy.ADMIN_CTX_POLICY: "role:admin"}.items())
     policy.set_rules(common_policy.Rules(rules))
     self.assertEqual(["admin"], policy.get_admin_roles())

 def _set_rules(self, default_rule):
     rules = common_policy.Rules(
         dict((k, common_policy.parse_rule(v))
              for k, v in self.rules.items()), default_rule)
     common_policy.set_rules(rules)