From f80d18a02c5e3a3292e2dfb56d410152a883b849 Mon Sep 17 00:00:00 2001 From: Slawek Kaplonski Date: Sat, 1 Oct 2022 22:06:11 +0200 Subject: [PATCH] Allow to pass EUI64 IP address as fixed ip for the port When port with IP address from the auto allocation subnet (SLAAC or dhcp_stateless) is created or updated and fixed IP address is specified, Neutron will fail as in case of such subnets IP address is assigned automatically based on the subnet's prefix and port's MAC address. But in case when given IP address is actually correct EUI64 generated IP address (the same as Neutron would generate for that port) there is no need to raise an exception and fail request. Additionally this patch fixes imports sections in the ipam_pluggable_backend module. Closes-bug: #1991398 Change-Id: Iaee5608c9581228a83f7ad75dbf2cc31dafaa9ea (cherry picked from commit d7b44f7218ff665045adb923b1aa92c35b371af9) --- neutron/db/ipam_pluggable_backend.py | 14 +++++---- .../unit/db/test_ipam_pluggable_backend.py | 30 +++++++++++++++++++ 2 files changed, 39 insertions(+), 5 deletions(-) diff --git a/neutron/db/ipam_pluggable_backend.py b/neutron/db/ipam_pluggable_backend.py index 5a08622c083..c084b2c3193 100644 --- a/neutron/db/ipam_pluggable_backend.py +++ b/neutron/db/ipam_pluggable_backend.py @@ -23,10 +23,10 @@ from neutron_lib import exceptions as n_exc from neutron_lib.objects import utils as obj_utils from neutron_lib.plugins import constants as plugin_consts from neutron_lib.plugins import directory - from oslo_db import exception as db_exc from oslo_log import log as logging from oslo_utils import excutils +from oslo_utils import netutils from neutron.common import ipv6_utils from neutron.db import ipam_backend_mixin @@ -272,6 +272,7 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin): p["network_id"], p['fixed_ips'], p['device_owner'], + p['mac_address'], subnets) else: ips = [] @@ -304,7 +305,7 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin): return ips def _test_fixed_ips_for_port(self, context, network_id, fixed_ips, - device_owner, subnets): + device_owner, mac_address, subnets): """Test fixed IPs for port. Check that configured subnets are valid prior to allocating any @@ -325,8 +326,11 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin): subnet['cidr'] != constants.PROVISIONAL_IPV6_PD_PREFIX): if (is_auto_addr_subnet and device_owner not in constants.ROUTER_INTERFACE_OWNERS): - raise ipam_exc.AllocationOnAutoAddressSubnet( - ip=fixed['ip_address'], subnet_id=subnet['id']) + eui64_ip = netutils.get_ipv6_addr_by_EUI64( + subnet['cidr'], mac_address) + if eui64_ip != netaddr.IPAddress(fixed['ip_address']): + raise ipam_exc.AllocationOnAutoAddressSubnet( + ip=fixed['ip_address'], subnet_id=subnet['id']) fixed_ip_list.append({'subnet_id': subnet['id'], 'ip_address': fixed['ip_address']}) else: @@ -383,7 +387,7 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin): # Check if the IP's to add are OK to_add = self._test_fixed_ips_for_port( context, port['network_id'], changes.add, - port['device_owner'], subnets) + port['device_owner'], port['mac_address'], subnets) if port['device_owner'] not in constants.ROUTER_INTERFACE_OWNERS: to_add += self._update_ips_for_pd_subnet( diff --git a/neutron/tests/unit/db/test_ipam_pluggable_backend.py b/neutron/tests/unit/db/test_ipam_pluggable_backend.py index c869bb1458d..f69e7a7d5c6 100644 --- a/neutron/tests/unit/db/test_ipam_pluggable_backend.py +++ b/neutron/tests/unit/db/test_ipam_pluggable_backend.py @@ -28,6 +28,7 @@ import webob.exc from neutron.db import ipam_backend_mixin from neutron.db import ipam_pluggable_backend +from neutron.ipam import exceptions as ipam_exc from neutron.ipam import requests as ipam_req from neutron.objects import network as network_obj from neutron.objects import ports as port_obj @@ -382,6 +383,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase): subnet['network_id'], fixed_ips, constants.DEVICE_OWNER_ROUTER_INTF, + "aa:bb:cc:dd:ee:ff", [subnet])) # Assert that ports created on prefix delegation subnets # will be returned without an ip address. This prevents router @@ -390,6 +392,31 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase): self.assertEqual(subnet['id'], filtered_ips[0]['subnet_id']) self.assertNotIn('ip_address', filtered_ips[0]) + def test_test_fixed_ips_for_port_allocation_on_auto_address_subnet(self): + context = mock.Mock() + pluggable_backend = ipam_pluggable_backend.IpamPluggableBackend() + with self.subnet(cidr="2001:db8::/64", + ip_version=constants.IP_VERSION_6, + ipv6_ra_mode=constants.IPV6_SLAAC, + ipv6_address_mode=constants.IPV6_SLAAC) as subnet: + subnet = subnet['subnet'] + bad_fixed_ip = [{'subnet_id': subnet['id'], + 'ip_address': '2001:db8::22'}] + eui64_fixed_ip = [{'subnet_id': subnet['id'], + 'ip_address': '2001:db8::a8bb:ccff:fedd:eeff'}] + self.assertRaises( + ipam_exc.AllocationOnAutoAddressSubnet, + pluggable_backend._test_fixed_ips_for_port, + context, subnet['network_id'], bad_fixed_ip, + "device_owner", "aa:bb:cc:dd:ee:ff", + [subnet]) + + filtered_ips = pluggable_backend._test_fixed_ips_for_port( + context, subnet['network_id'], eui64_fixed_ip, "device_owner", + "aa:bb:cc:dd:ee:ff", [subnet]) + self.assertEqual(1, len(filtered_ips)) + self.assertEqual(subnet['id'], filtered_ips[0]['subnet_id']) + @mock.patch('neutron.ipam.driver.Pool') def test_create_subnet_over_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) @@ -686,6 +713,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase): mocks['ipam']._update_ips_for_pd_subnet = mock.Mock(return_value=[]) port_dict = {'device_owner': uuidutils.generate_uuid(), + 'mac_address': 'aa:bb:cc:dd:ee:ff', 'network_id': uuidutils.generate_uuid()} mocks['ipam']._update_ips_for_port(context, port_dict, None, @@ -726,6 +754,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase): port_dict = { 'device_owner': constants.DEVICE_OWNER_DISTRIBUTED, 'device_id': 'ovnmeta-%s' % uuidutils.generate_uuid(), + 'mac_address': 'aa:bb:cc:dd:ee:ff', 'network_id': uuidutils.generate_uuid()} mocks['ipam']._update_ips_for_port(context, port_dict, None, @@ -749,6 +778,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase): 'subnet_id': uuidutils.generate_uuid()} port_dict = {'port': {'device_owner': uuidutils.generate_uuid(), 'network_id': network_id, + 'mac_address': 'aa:bb:cc:dd:ee:ff', 'fixed_ips': [ip_dict]}} subnets = [{'id': ip_dict['subnet_id'], 'network_id': network_id,