Merge "Allow to pass EUI64 IP address as fixed ip for the port"

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,6 +326,9 @@ 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):
+                    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'],
@@ -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(

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,