diff --git a/neutron/db/ipam_pluggable_backend.py b/neutron/db/ipam_pluggable_backend.py index f31b66ab52f..fc22527d5c0 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 748cb68a4e0..69fc9fc5c60 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,