Use EUI64 for IPv6 SLAAC when subnet is specified

This commit uses EUI64 for SLAAC and stateless IPv6 address
when subnet id in fixed_ip is specified.

After this patch, all the ports created on a subnet which has
ipv6_address_mod=slaac or ipv6_address_mod=dhcpv6-stateless
will use EUI64 as the address.
This patch also checks if fixed IP address is specified
for a IPv6 subnet with address mode slaac or dhcpv6-stateless
during creating or updating a port. If yes, raise InvalidInput
error to stop the port creation or update.

Remove unit test test_generated_duplicate_ip_ipv6 because
fixed_ip should not be specified for a slaac subnet.

Change-Id: Ie481cfb2f4313baf44bf1a838ebda374a5c74c6a
Closes-Bug: 1330826

neutron/db/db_base_plugin_v2.py

@@ -449,7 +449,14 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
                     msg = _('IP address %s is not a valid IP for the defined '
                             'subnet') % fixed['ip_address']
                     raise n_exc.InvalidInput(error_message=msg)
-
+                if self._check_if_subnet_uses_eui64(subnet):
+                    msg = (_("IPv6 address %(address)s can not be directly "
+                            "assigned to a port on subnet %(id)s with "
+                            "%(mode)s address mode") %
+                           {'address': fixed['ip_address'],
+                            'id': subnet_id,
+                            'mode': subnet['ipv6_address_mode']})
+                    raise n_exc.InvalidInput(error_message=msg)
                 fixed_ip_set.append({'subnet_id': subnet_id,
                                      'ip_address': fixed['ip_address']})
             else:
@@ -459,7 +466,7 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
             raise n_exc.InvalidInput(error_message=msg)
         return fixed_ip_set
 
-    def _allocate_fixed_ips(self, context, fixed_ips):
+    def _allocate_fixed_ips(self, context, fixed_ips, mac_address):
         """Allocate IP addresses according to the configured fixed_ips."""
         ips = []
         for fixed in fixed_ips:
@@ -472,15 +479,24 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
             # Only subnet ID is specified => need to generate IP
             # from subnet
             else:
-                subnets = [self._get_subnet(context, fixed['subnet_id'])]
+                subnet = self._get_subnet(context, fixed['subnet_id'])
-                # IP address allocation
+                if (subnet['ip_version'] == 6 and
-                result = self._generate_ip(context, subnets)
+                        self._check_if_subnet_uses_eui64(subnet)):
-                ips.append({'ip_address': result['ip_address'],
+                    prefix = subnet['cidr']
-                            'subnet_id': result['subnet_id']})
+                    ip_address = ipv6_utils.get_ipv6_addr_by_EUI64(
+                        prefix, mac_address)
+                    ips.append({'ip_address': ip_address.format(),
+                                'subnet_id': subnet['id']})
+                else:
+                    subnets = [subnet]
+                    # IP address allocation
+                    result = self._generate_ip(context, subnets)
+                    ips.append({'ip_address': result['ip_address'],
+                                'subnet_id': result['subnet_id']})
         return ips
 
     def _update_ips_for_port(self, context, network_id, port_id, original_ips,
-                             new_ips):
+                             new_ips, mac_address):
         """Add or remove IPs from the port."""
         ips = []
         # These ips are still on the port and haven't been removed
@@ -511,7 +527,7 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
 
         if to_add:
             LOG.debug(_("Port update. Adding %s"), to_add)
-            ips = self._allocate_fixed_ips(context, to_add)
+            ips = self._allocate_fixed_ips(context, to_add, mac_address)
         return ips, prev_ips
 
     def _allocate_ips_for_port(self, context, port):
@@ -529,7 +545,9 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
             configured_ips = self._test_fixed_ips_for_port(context,
                                                            p["network_id"],
                                                            p['fixed_ips'])
-            ips = self._allocate_fixed_ips(context, configured_ips)
+            ips = self._allocate_fixed_ips(context,
+                                           configured_ips,
+                                           p['mac_address'])
         else:
             filter = {'network_id': [p['network_id']]}
             subnets = self.get_subnets(context, filters=filter)
@@ -548,10 +566,9 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
                     # subnet from the array of subnets that will be passed
                     # to the _generate_ip() function call, since we just
                     # generated an IP.
-                    mac = p['mac_address']
                     prefix = subnet['cidr']
                     ip_address = ipv6_utils.get_ipv6_addr_by_EUI64(
-                        prefix, mac)
+                        prefix, p['mac_address'])
                     if not self._check_unique_ip(
                         context, p['network_id'],
                         subnet['id'], ip_address.format()):
@@ -1377,8 +1394,9 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
                 changed_ips = True
                 original = self._make_port_dict(port, process_extensions=False)
                 added_ips, prev_ips = self._update_ips_for_port(
-                    context, port["network_id"], id, original["fixed_ips"],
+                    context, port["network_id"], id,
-                    p['fixed_ips'])
+                    original["fixed_ips"], p['fixed_ips'],
+                    original['mac_address'])
 
                 # Update ips if necessary
                 for ip in added_ips:

neutron/tests/unit/test_db_plugin.py

@@ -1233,6 +1233,34 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
                 self.assertEqual(ips[1]['ip_address'], '10.0.0.4')
                 self.assertEqual(ips[1]['subnet_id'], subnet['subnet']['id'])
 
+    def test_update_port_invalid_fixed_ip_address_v6_slaac(self):
+        with self.subnet(
+            cidr='2607:f0d0:1002:51::/64',
+            ip_version=6,
+            ipv6_address_mode=constants.IPV6_SLAAC,
+            gateway_ip=attributes.ATTR_NOT_SPECIFIED) as subnet:
+            with self.port(subnet=subnet) as port:
+                ips = port['port']['fixed_ips']
+                self.assertEqual(len(ips), 1)
+                port_mac = port['port']['mac_address']
+                subnet_cidr = subnet['subnet']['cidr']
+                eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr,
+                                                                 port_mac))
+                self.assertEqual(ips[0]['ip_address'], eui_addr)
+                self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id'])
+
+                data = {'port': {'fixed_ips': [{'subnet_id':
+                                                subnet['subnet']['id'],
+                                                'ip_address':
+                                                '2607:f0d0:1002:51::5'}]}}
+                req = self.new_update_request('ports', data,
+                                              port['port']['id'])
+                res = req.get_response(self.api)
+                err = self.deserialize(self.fmt, res)
+                self.assertEqual(res.status_int,
+                                 webob.exc.HTTPClientError.code)
+                self.assertEqual(err['NeutronError']['type'], 'InvalidInput')
+
     def test_requested_duplicate_mac(self):
         with self.port() as port:
             mac = port['port']['mac_address']
@@ -1295,20 +1323,6 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
                 res = self._create_port(self.fmt, net_id=net_id, **kwargs)
                 self.assertEqual(res.status_int, webob.exc.HTTPConflict.code)
 
-    def test_generated_duplicate_ip_ipv6(self):
-        with self.subnet(ip_version=6,
-                         cidr="2014::/64",
-                         ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
-            with self.port(subnet=subnet,
-                           fixed_ips=[{'subnet_id': subnet['subnet']['id'],
-                                       'ip_address':
-                                       "2014::1322:33ff:fe44:5566"}]) as port:
-                # Check configuring of duplicate IP
-                kwargs = {"mac_address": "11:22:33:44:55:66"}
-                net_id = port['port']['network_id']
-                res = self._create_port(self.fmt, net_id=net_id, **kwargs)
-                self.assertEqual(res.status_int, webob.exc.HTTPConflict.code)
-
     def test_requested_subnet_id(self):
         with self.subnet() as subnet:
             with self.port(subnet=subnet) as port:
@@ -1393,6 +1407,57 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
                 self._delete('ports', port3['port']['id'])
                 self._delete('ports', port4['port']['id'])
 
+    def test_requested_invalid_fixed_ip_address_v6_slaac(self):
+        with self.subnet(gateway_ip='fe80::1',
+                         cidr='2607:f0d0:1002:51::/64',
+                         ip_version=6,
+                         ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
+            kwargs = {"fixed_ips": [{'subnet_id': subnet['subnet']['id'],
+                                     'ip_address': '2607:f0d0:1002:51::5'}]}
+            net_id = subnet['subnet']['network_id']
+            res = self._create_port(self.fmt, net_id=net_id, **kwargs)
+            self.assertEqual(res.status_int,
+                             webob.exc.HTTPClientError.code)
+
+    def test_requested_subnet_id_v6_slaac(self):
+        with self.subnet(gateway_ip='fe80::1',
+                         cidr='2607:f0d0:1002:51::/64',
+                         ip_version=6,
+                         ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
+            with self.port(subnet,
+                           fixed_ips=[{'subnet_id':
+                                       subnet['subnet']['id']}]) as port:
+                port_mac = port['port']['mac_address']
+                subnet_cidr = subnet['subnet']['cidr']
+                eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr,
+                                                                 port_mac))
+                self.assertEqual(port['port']['fixed_ips'][0]['ip_address'],
+                                 eui_addr)
+
+    def test_requested_subnet_id_v4_and_v6_slaac(self):
+        with self.network() as network:
+            with contextlib.nested(
+                self.subnet(network),
+                self.subnet(network,
+                            cidr='2607:f0d0:1002:51::/64',
+                            ip_version=6,
+                            gateway_ip='fe80::1',
+                            ipv6_address_mode=constants.IPV6_SLAAC)
+            ) as (subnet, subnet2):
+                with self.port(
+                    subnet,
+                    fixed_ips=[{'subnet_id': subnet['subnet']['id']},
+                               {'subnet_id': subnet2['subnet']['id']}]
+                ) as port:
+                    ips = port['port']['fixed_ips']
+                    self.assertEqual(len(ips), 2)
+                    self.assertEqual(ips[0]['ip_address'], '10.0.0.2')
+                    port_mac = port['port']['mac_address']
+                    subnet_cidr = subnet2['subnet']['cidr']
+                    eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(
+                            subnet_cidr, port_mac))
+                    self.assertEqual(ips[1]['ip_address'], eui_addr)
+
     def test_ip_allocation_for_ipv6_subnet_slaac_address_mode(self):
         res = self._create_network(fmt=self.fmt, name='net',
                                    admin_state_up=True)