packages/openstack-neutron
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
openstack-neutron/CVE-2024-53916-Fix-the-tagging-policy-engine.patch

452 lines
20 KiB
Diff
Raw Permalink Blame History

diff --git a/neutron/extensions/tagging.py b/neutron/extensions/tagging.py
index b65978e..8ffd8a6 100644
--- a/neutron/extensions/tagging.py
+++ b/neutron/extensions/tagging.py
@@ -12,6 +12,9 @@
# under the License.
import abc
+import collections
+import functools
+import itertools
from neutron_lib.api import extensions as api_extensions
from neutron_lib.api import faults
@@ -28,6 +31,16 @@ from neutron.api import extensions
from neutron.api.v2 import resource as api_resource
from neutron.db import standard_attr
+from neutron.objects import network as network_obj
+from neutron.objects import network_segment_range as network_segment_range_obj
+from neutron.objects import ports as ports_obj
+from neutron.objects.qos import policy as policy_obj
+from neutron.objects import router as router_obj
+from neutron.objects import securitygroup as securitygroup_obj
+from neutron.objects import subnet as subnet_obj
+from neutron.objects import subnetpool as subnetpool_obj
+from neutron.objects import trunk as trunk_obj
+from neutron import policy
TAG = 'tag'
TAGS = TAG + 's'
@@ -49,6 +62,33 @@ TAG_ATTRIBUTE_MAP = {
'is_visible': False, 'is_filter': True},
}
+PARENTS = {
+ 'floatingips': router_obj.FloatingIP,
+ 'network_segment_ranges': network_segment_range_obj.NetworkSegmentRange,
+ 'networks': network_obj.Network,
+ 'policies': policy_obj.QosPolicy,
+ 'ports': ports_obj.Port,
+ 'routers': router_obj.Router,
+ 'security_groups': securitygroup_obj.SecurityGroup,
+ 'subnets': ('networks', subnet_obj.Subnet),
+ 'subnetpools': subnetpool_obj.SubnetPool,
+ 'trunks': trunk_obj.Trunk,
+}
+ResourceInfo = collections.namedtuple(
+ 'ResourceInfo', ['project_id',
+ 'parent_type',
+ 'parent_id',
+ 'upper_parent_type',
+ 'upper_parent_id',
+ ])
+EMPTY_RESOURCE_INFO = ResourceInfo(None, None, None, None, None)
+
+def _policy_init(f):
+ @functools.wraps(f)
+ def func(self, *args, **kwargs):
+ policy.init()
+ return f(self, *args, **kwargs)
+ return func
class TagResourceNotFound(exceptions.NotFound):
message = _("Resource %(resource)s %(resource_id)s could not be found.")
@@ -88,75 +128,161 @@ class TaggingController(object):
self.plugin = directory.get_plugin(TAG_PLUGIN_TYPE)
self.supported_resources = TAG_SUPPORTED_RESOURCES
+ def _get_target(self, res_info):
+ target = {'id': res_info.parent_id,
+ 'tenant_id': res_info.project_id,
+ 'project_id': res_info.project_id}
+ if res_info.upper_parent_type:
+ res_id = (self.supported_resources[res_info.upper_parent_type] +
+ '_id')
+ target[res_id] = res_info.upper_parent_id
+ return target
+
+ def _get_resource_info(self, context, kwargs):
+ """Return the tag parent resource information
+
+ Some parent resources, like the subnets, depend on other upper parent
+ resources (networks). In that case, it is needed to provide the upper
+ parent resource information.
+
+ :param kwargs: dictionary with the parent resource ID, along with other
+ information not needed. It is formated as
+ {"resource_id": "id", ...}
+ :return: ``ResourceInfo`` named tuple with the parent and upper parent
+ information and the project ID (of the parent or upper
+ parent).
+ """
+ for key, parent_type in itertools.product(
+ kwargs.keys(), self.supported_resources.keys()):
+ if key != self.supported_resources[parent_type] + '_id':
+ continue
+
+ parent_id = kwargs[key]
+ parent_obj = PARENTS[parent_type]
+ if isinstance(parent_obj, tuple):
+ upper_parent_type = parent_obj[0]
+ parent_obj = parent_obj[1]
+ res_id = (self.supported_resources[upper_parent_type] +
+ '_id')
+ upper_parent_id = parent_obj.get_values(
+ context.elevated(), res_id, id=parent_id)[0]
+ else:
+ upper_parent_type = upper_parent_id = None
+
+ try:
+ project_id = parent_obj.get_values(
+ context.elevated(), 'project_id', id=parent_id)[0]
+ except IndexError:
+ return EMPTY_RESOURCE_INFO
+
+ return ResourceInfo(project_id, parent_type, parent_id,
+ upper_parent_type, upper_parent_id)
+
+ # This should never be returned.
+ return EMPTY_RESOURCE_INFO
+
def _get_parent_resource_and_id(self, kwargs):
for key in kwargs:
for resource in self.supported_resources:
if key == self.supported_resources[resource] + '_id':
return resource, kwargs[key]
return None, None
-
+ @_policy_init
def index(self, request, **kwargs):
- # GET /v2.0/networks/{network_id}/tags
- parent, parent_id = self._get_parent_resource_and_id(kwargs)
- return self.plugin.get_tags(request.context, parent, parent_id)
-
+ # GET /v2.0/{parent_resource}/{parent_resource_id}/tags
+ ctx = request.context
+ rinfo = self._get_resource_info(ctx, kwargs)
+ target = self._get_target(rinfo)
+ policy.enforce(ctx, 'get_{}_{}'.format(rinfo.parent_type, TAGS),
+ target)
+ return self.plugin.get_tags(ctx, rinfo.parent_type, rinfo.parent_id)
+
+ @_policy_init
def show(self, request, id, **kwargs):
- # GET /v2.0/networks/{network_id}/tags/{tag}
+ # GET /v2.0/{parent_resource}/{parent_resource_id}/tags/{tag}
# id == tag
validate_tag(id)
- parent, parent_id = self._get_parent_resource_and_id(kwargs)
- return self.plugin.get_tag(request.context, parent, parent_id, id)
-
+ ctx = request.context
+ rinfo = self._get_resource_info(ctx, kwargs)
+ target = self._get_target(rinfo)
+ policy.enforce(ctx, 'get_{}_{}'.format(rinfo.parent_type, TAGS),
+ target)
+ return self.plugin.get_tag(ctx, rinfo.parent_type, rinfo.parent_id, id)
+
+ @_policy_init
def create(self, request, **kwargs):
# not supported
# POST /v2.0/networks/{network_id}/tags
raise webob.exc.HTTPNotFound("not supported")
+ @_policy_init
def update(self, request, id, **kwargs):
- # PUT /v2.0/networks/{network_id}/tags/{tag}
+ # PUT /v2.0/{parent_resource}/{parent_resource_id}/tags/{tag}
# id == tag
validate_tag(id)
- parent, parent_id = self._get_parent_resource_and_id(kwargs)
- notify_tag_action(request.context, 'create.start',
- parent, parent_id, [id])
- result = self.plugin.update_tag(request.context, parent, parent_id, id)
- notify_tag_action(request.context, 'create.end',
- parent, parent_id, [id])
+ ctx = request.context
+ rinfo = self._get_resource_info(ctx, kwargs)
+ target = self._get_target(rinfo)
+ policy.enforce(ctx, 'update_{}_{}'.format(rinfo.parent_type, TAGS),
+ target)
+ notify_tag_action(ctx, 'create.start', rinfo.parent_type,
+ rinfo.parent_id, [id])
+ result = self.plugin.update_tag(ctx, rinfo.parent_type,
+ rinfo.parent_id, id)
+ notify_tag_action(ctx, 'create.end', rinfo.parent_type,
+ rinfo.parent_id, [id])
return result
+ @_policy_init
def update_all(self, request, body, **kwargs):
- # PUT /v2.0/networks/{network_id}/tags
+ # PUT /v2.0/{parent_resource}/{parent_resource_id}/tags
# body: {"tags": ["aaa", "bbb"]}
validate_tags(body)
- parent, parent_id = self._get_parent_resource_and_id(kwargs)
- notify_tag_action(request.context, 'update.start',
- parent, parent_id, body['tags'])
- result = self.plugin.update_tags(request.context, parent,
- parent_id, body)
- notify_tag_action(request.context, 'update.end',
- parent, parent_id, body['tags'])
+ ctx = request.context
+ rinfo = self._get_resource_info(ctx, kwargs)
+ target = self._get_target(rinfo)
+ policy.enforce(ctx, 'update_{}_{}'.format(rinfo.parent_type, TAGS),
+ target)
+ notify_tag_action(ctx, 'update.start', rinfo.parent_type,
+ rinfo.parent_id, body['tags'])
+ result = self.plugin.update_tags(ctx, rinfo.parent_type,
+ rinfo.parent_id, body)
+ notify_tag_action(ctx, 'update.end', rinfo.parent_type,
+ rinfo.parent_id, body['tags'])
return result
+ @_policy_init
def delete(self, request, id, **kwargs):
- # DELETE /v2.0/networks/{network_id}/tags/{tag}
+ # DELETE /v2.0/{parent_resource}/{parent_resource_id}/tags/{tag}
# id == tag
validate_tag(id)
- parent, parent_id = self._get_parent_resource_and_id(kwargs)
- notify_tag_action(request.context, 'delete.start',
- parent, parent_id, [id])
- result = self.plugin.delete_tag(request.context, parent, parent_id, id)
- notify_tag_action(request.context, 'delete.end',
- parent, parent_id, [id])
+ ctx = request.context
+ rinfo = self._get_resource_info(ctx, kwargs)
+ target = self._get_target(rinfo)
+ policy.enforce(ctx, 'delete_{}_{}'.format(rinfo.parent_type, TAGS),
+ target)
+ notify_tag_action(ctx, 'delete.start', rinfo.parent_type,
+ rinfo.parent_id, [id])
+ result = self.plugin.delete_tag(ctx, rinfo.parent_type,
+ rinfo.parent_id, id)
+ notify_tag_action(ctx, 'delete.end', rinfo.parent_type,
+ rinfo.parent_id, [id])
return result
+ @_policy_init
def delete_all(self, request, **kwargs):
- # DELETE /v2.0/networks/{network_id}/tags
- parent, parent_id = self._get_parent_resource_and_id(kwargs)
- notify_tag_action(request.context, 'delete_all.start',
- parent, parent_id)
- result = self.plugin.delete_tags(request.context, parent, parent_id)
- notify_tag_action(request.context, 'delete_all.end',
- parent, parent_id)
+ # DELETE /v2.0/{parent_resource}/{parent_resource_id}/tags
+ ctx = request.context
+ rinfo = self._get_resource_info(ctx, kwargs)
+ target = self._get_target(rinfo)
+ policy.enforce(ctx, 'delete_{}_{}'.format(rinfo.parent_type, TAGS),
+ target)
+ notify_tag_action(ctx, 'delete_all.start', rinfo.parent_type,
+ rinfo.parent_id)
+ result = self.plugin.delete_tags(ctx, rinfo.parent_type,
+ rinfo.parent_id)
+ notify_tag_action(ctx, 'delete_all.end', rinfo.parent_type,
+ rinfo.parent_id)
return result
diff --git a/neutron/tests/unit/extensions/test_tagging.py b/neutron/tests/unit/extensions/test_tagging.py
new file mode 100644
index 0000000..d9eea32
--- /dev/null
+++ b/neutron/tests/unit/extensions/test_tagging.py
@@ -0,0 +1,179 @@
+# Copyright 2024 Red Hat, Inc.
+# All rights reserved.
+#
+# 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.
+#
+
+import netaddr
+from neutron_lib import constants as n_const
+from neutron_lib import context
+from neutron_lib.utils import net as net_utils
+from oslo_utils import uuidutils
+
+from neutron.extensions import tagging
+from neutron.objects import network as network_obj
+from neutron.objects import network_segment_range as network_segment_range_obj
+from neutron.objects import ports as ports_obj
+from neutron.objects.qos import policy as policy_obj
+from neutron.objects import router as router_obj
+from neutron.objects import securitygroup as securitygroup_obj
+from neutron.objects import subnet as subnet_obj
+from neutron.objects import subnetpool as subnetpool_obj
+from neutron.objects import trunk as trunk_obj
+from neutron.tests.unit import testlib_api
+
+
+class TaggingControllerDbTestCase(testlib_api.WebTestCase):
+ def setUp(self):
+ super().setUp()
+ self.user_id = uuidutils.generate_uuid()
+ self.project_id = uuidutils.generate_uuid()
+ self.ctx = context.Context(user_id=self.user_id,
+ tenant_id=self.project_id,
+ is_admin=False)
+ self.tc = tagging.TaggingController()
+
+ def test_all_parents_have_a_reference(self):
+ tc_supported_resources = set(self.tc.supported_resources.keys())
+ parent_resources = set(tagging.PARENTS.keys())
+ self.assertEqual(tc_supported_resources, parent_resources)
+
+ def _check_resource_info(self, parent_id, parent_type,
+ upper_parent_id=None, upper_parent_type=None):
+ p_id = self.tc.supported_resources[parent_type] + '_id'
+ res = self.tc._get_resource_info(self.ctx, {p_id: parent_id})
+ reference = tagging.ResourceInfo(
+ self.project_id, parent_type, parent_id,
+ upper_parent_type, upper_parent_id)
+ self.assertEqual(reference, res)
+
+ def test__get_resource_info_floatingips(self):
+ ext_net_id = uuidutils.generate_uuid()
+ fip_port_id = uuidutils.generate_uuid()
+ fip_id = uuidutils.generate_uuid()
+ network_obj.Network(
+ self.ctx, id=ext_net_id, project_id=self.project_id).create()
+ network_obj.ExternalNetwork(
+ self.ctx, project_id=self.project_id,
+ network_id=ext_net_id).create()
+ mac_str = next(net_utils.random_mac_generator(
+ ['ca', 'fe', 'ca', 'fe']))
+ mac = netaddr.EUI(mac_str)
+ ports_obj.Port(
+ self.ctx, id=fip_port_id, project_id=self.project_id,
+ mac_address=mac, network_id=ext_net_id, admin_state_up=True,
+ status='UP', device_id='', device_owner='').create()
+ ip_address = netaddr.IPAddress('1.2.3.4')
+ router_obj.FloatingIP(
+ self.ctx, id=fip_id, project_id=self.project_id,
+ floating_network_id=ext_net_id, floating_port_id=fip_port_id,
+ floating_ip_address=ip_address).create()
+ self._check_resource_info(fip_id, 'floatingips')
+
+ def test__get_resource_info_network_segment_ranges(self):
+ srange_id = uuidutils.generate_uuid()
+ network_segment_range_obj.NetworkSegmentRange(
+ self.ctx, id=srange_id, project_id=self.project_id,
+ shared=False, network_type=n_const.TYPE_GENEVE).create()
+ self._check_resource_info(srange_id, 'network_segment_ranges')
+
+ def test__get_resource_info_networks(self):
+ net_id = uuidutils.generate_uuid()
+ network_obj.Network(
+ self.ctx, id=net_id, project_id=self.project_id).create()
+ self._check_resource_info(net_id, 'networks')
+
+ def test__get_resource_info_policies(self):
+ qos_id = uuidutils.generate_uuid()
+ policy_obj.QosPolicy(
+ self.ctx, id=qos_id, project_id=self.project_id).create()
+ self._check_resource_info(qos_id, 'policies')
+
+ def test__get_resource_info_ports(self):
+ net_id = uuidutils.generate_uuid()
+ port_id = uuidutils.generate_uuid()
+ network_obj.Network(
+ self.ctx, id=net_id, project_id=self.project_id).create()
+ mac_str = next(net_utils.random_mac_generator(
+ ['ca', 'fe', 'ca', 'fe']))
+ mac = netaddr.EUI(mac_str)
+ ports_obj.Port(
+ self.ctx, id=port_id, project_id=self.project_id,
+ mac_address=mac, network_id=net_id, admin_state_up=True,
+ status='UP', device_id='', device_owner='').create()
+ self._check_resource_info(port_id, 'ports')
+
+ def test__get_resource_info_routers(self):
+ router_id = uuidutils.generate_uuid()
+ router_obj.Router(
+ self.ctx, id=router_id, project_id=self.project_id).create()
+ self._check_resource_info(router_id, 'routers')
+
+ def test__get_resource_info_security_groups(self):
+ sg_id = uuidutils.generate_uuid()
+ securitygroup_obj.SecurityGroup(
+ self.ctx, id=sg_id, project_id=self.project_id,
+ is_default=True).create()
+ self._check_resource_info(sg_id, 'security_groups')
+
+ def test__get_resource_info_subnets(self):
+ net_id = uuidutils.generate_uuid()
+ subnet_id = uuidutils.generate_uuid()
+ network_obj.Network(
+ self.ctx, id=net_id, project_id=self.project_id).create()
+ cidr = netaddr.IPNetwork('1.2.3.0/24')
+ subnet_obj.Subnet(
+ self.ctx, id=subnet_id, project_id=self.project_id,
+ ip_version=n_const.IP_VERSION_4, cidr=cidr,
+ network_id=net_id).create()
+ self._check_resource_info(subnet_id, 'subnets',
+ upper_parent_id=net_id,
+ upper_parent_type='networks')
+
+ def test__get_resource_info_subnetpools(self):
+ sp_id = uuidutils.generate_uuid()
+ subnetpool_obj.SubnetPool(
+ self.ctx, id=sp_id, project_id=self.project_id,
+ ip_version=n_const.IP_VERSION_4, default_prefixlen=26,
+ min_prefixlen=28, max_prefixlen=26).create()
+ self._check_resource_info(sp_id, 'subnetpools')
+
+ def test__get_resource_info_trunks(self):
+ trunk_id = uuidutils.generate_uuid()
+ net_id = uuidutils.generate_uuid()
+ port_id = uuidutils.generate_uuid()
+ network_obj.Network(
+ self.ctx, id=net_id, project_id=self.project_id).create()
+ mac_str = next(net_utils.random_mac_generator(
+ ['ca', 'fe', 'ca', 'fe']))
+ mac = netaddr.EUI(mac_str)
+ ports_obj.Port(
+ self.ctx, id=port_id, project_id=self.project_id,
+ mac_address=mac, network_id=net_id, admin_state_up=True,
+ status='UP', device_id='', device_owner='').create()
+ trunk_obj.Trunk(
+ self.ctx, id=trunk_id, project_id=self.project_id,
+ port_id=port_id).create()
+ self._check_resource_info(trunk_id, 'trunks')
+
+ def test__get_resource_info_parent_not_present(self):
+ missing_id = uuidutils.generate_uuid()
+ p_id = self.tc.supported_resources['trunks'] + '_id'
+ res = self.tc._get_resource_info(self.ctx, {p_id: missing_id})
+ self.assertEqual(tagging.EMPTY_RESOURCE_INFO, res)
+
+ def test__get_resource_info_wrong_resource(self):
+ missing_id = uuidutils.generate_uuid()
+ res = self.tc._get_resource_info(self.ctx,
+ {'wrong_resource_id': missing_id})
+ self.assertEqual(tagging.EMPTY_RESOURCE_INFO, res)
\ No newline at end of file
Reference in New Issue View Git Blame Copy Permalink