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 d7b44f7)

Author: slawqo

neutron/db/ipam_pluggable_backend.py

@@ -23,10 +23,10 @@
 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 @@ def _allocate_ips_for_port(self, context, port):
                                                 p["network_id"],
                                                 p['fixed_ips'],
                                                 p['device_owner'],
+                                                p['mac_address'],
                                                 subnets)
         else:
             ips = []
@@ -304,7 +305,7 @@ def _get_auto_address_ips(self, v6_stateless_subnets, port,
         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 @@ def _test_fixed_ips_for_port(self, context, network_id, fixed_ips,
                     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 @@ def _update_ips_for_port(self, context, port, host,
         # 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(

neutron/tests/unit/db/test_ipam_pluggable_backend.py

@@ -28,6 +28,7 @@
 
 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 @@ def test_test_fixed_ips_for_port_pd_gateway(self):
                                 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 @@ def test_test_fixed_ips_for_port_pd_gateway(self):
             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 @@ def test_update_ips_for_port_passes_port_dict_to_factory(self, pool_mock):
         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 @@ def test_update_ips_for_port_ovn_distributed_svc(self, pool_mock):
         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 @@ def test_update_ips_for_port_passes_port_id_to_factory(self, pool_mock):
                    '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,