diff --git a/neutron/conf/policies/__init__.py b/neutron/conf/policies/__init__.py index dbb7aad0e2d..8557f01bf85 100644 --- a/neutron/conf/policies/__init__.py +++ b/neutron/conf/policies/__init__.py @@ -34,6 +34,7 @@ from neutron.conf.policies import network from neutron.conf.policies import network_ip_availability from neutron.conf.policies import network_segment_range from neutron.conf.policies import port +from neutron.conf.policies import port_bindings from neutron.conf.policies import qos from neutron.conf.policies import quotas from neutron.conf.policies import rbac @@ -67,6 +68,7 @@ def list_rules(): network.list_rules(), network_ip_availability.list_rules(), network_segment_range.list_rules(), + port_bindings.list_rules(), port.list_rules(), qos.list_rules(), quotas.list_rules(), diff --git a/neutron/conf/policies/port_bindings.py b/neutron/conf/policies/port_bindings.py new file mode 100644 index 00000000000..7e8b136a065 --- /dev/null +++ b/neutron/conf/policies/port_bindings.py @@ -0,0 +1,51 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from oslo_policy import policy + +from neutron.conf.policies import base + + +GET_BINDING_PATH = '/ports/{port_id}/bindings/' +ACTIVATE_BINDING_PATH = '/ports/{port_id}/bindings/{host}' + + +rules = [ + policy.DocumentedRuleDefault( + name='get_port_binding', + check_str=base.ADMIN, + scope_types=['project'], + description='Get port binding information', + operations=[ + { + 'method': 'GET', + 'path': GET_BINDING_PATH, + }, + ], + ), + policy.DocumentedRuleDefault( + name='activate', + check_str=base.ADMIN, + scope_types=['project'], + description='Activate port binding on the host', + operations=[ + { + 'method': 'GET', + 'path': ACTIVATE_BINDING_PATH, + }, + ], + ), +] + + +def list_rules(): + return rules diff --git a/neutron/tests/unit/conf/policies/test_port_bindings.py b/neutron/tests/unit/conf/policies/test_port_bindings.py new file mode 100644 index 00000000000..a267f8010e9 --- /dev/null +++ b/neutron/tests/unit/conf/policies/test_port_bindings.py @@ -0,0 +1,100 @@ +# Copyright (c) 2021 Red Hat Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +# implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from oslo_policy import policy as base_policy + +from neutron import policy +from neutron.tests.unit.conf.policies import test_base as base + + +class PortBindingsAPITestCase(base.PolicyBaseTestCase): + + def setUp(self): + super(PortBindingsAPITestCase, self).setUp() + self.target = {} + + +class SystemAdminTests(PortBindingsAPITestCase): + + def setUp(self): + super(SystemAdminTests, self).setUp() + self.context = self.system_admin_ctx + + def test_get_port_binding(self): + self.assertRaises( + base_policy.InvalidScope, + policy.enforce, + self.context, "get_port_binding", self.target) + + def test_activate_port_binding(self): + self.assertRaises( + base_policy.InvalidScope, + policy.enforce, + self.context, "activate", self.target) + + +class SystemMemberTests(SystemAdminTests): + + def setUp(self): + super(SystemMemberTests, self).setUp() + self.context = self.system_member_ctx + + +class SystemReaderTests(SystemMemberTests): + + def setUp(self): + super(SystemReaderTests, self).setUp() + self.context = self.system_reader_ctx + + +class AdminTests(PortBindingsAPITestCase): + + def setUp(self): + super(AdminTests, self).setUp() + self.context = self.project_admin_ctx + + def test_get_port_binding(self): + self.assertTrue( + policy.enforce(self.context, "get_port_binding", self.target)) + + def test_activate_port_binding(self): + self.assertTrue( + policy.enforce(self.context, "activate", self.target)) + + +class ProjectMemberTests(AdminTests): + + def setUp(self): + super(ProjectMemberTests, self).setUp() + self.context = self.project_member_ctx + + def test_get_port_binding(self): + self.assertRaises( + base_policy.PolicyNotAuthorized, + policy.enforce, + self.context, "get_port_binding", self.target) + + def test_activate_port_binding(self): + self.assertRaises( + base_policy.PolicyNotAuthorized, + policy.enforce, + self.context, "activate", self.target) + + +class ProjectReaderTests(ProjectMemberTests): + + def setUp(self): + super(ProjectReaderTests, self).setUp() + self.context = self.project_reader_ctx