import netaddr
+from neutron.common import constants
from neutron.openstack.common.gettextutils import _LI
from neutron.openstack.common import log
if not _IS_IPV6_ENABLED:
LOG.info(_LI("IPv6 is not enabled on this system."))
return _IS_IPV6_ENABLED
+
+
+def is_slaac_subnet(subnet):
+ """Check if subnet uses SLAAC addressing."""
+ return (subnet['ipv6_address_mode'] == constants.IPV6_SLAAC
+ or subnet['ipv6_address_mode'] == constants.DHCPV6_STATELESS)
ip_address=ip_address,
subnet_id=subnet_id).delete()
- @staticmethod
- def _check_if_subnet_uses_eui64(subnet):
- """Check if ipv6 address will be calculated via EUI64."""
- return (subnet['ipv6_address_mode'] == constants.IPV6_SLAAC
- or subnet['ipv6_address_mode'] == constants.DHCPV6_STATELESS)
-
@staticmethod
def _store_ip_allocation(context, ip_address, network_id, subnet_id,
port_id):
return True
return False
- def _test_fixed_ips_for_port(self, context, network_id, fixed_ips):
+ def _test_fixed_ips_for_port(self, context, network_id, fixed_ips,
+ device_owner):
"""Test fixed IPs for port.
Check that configured subnets are valid prior to allocating any
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):
+ if (ipv6_utils.is_slaac_subnet(subnet) and device_owner not in
+ (constants.DEVICE_OWNER_ROUTER_INTF,
+ constants.DEVICE_OWNER_DVR_INTERFACE)):
msg = (_("IPv6 address %(address)s can not be directly "
"assigned to a port on subnet %(id)s with "
"%(mode)s address mode") %
else:
subnet = self._get_subnet(context, fixed['subnet_id'])
if (subnet['ip_version'] == 6 and
- self._check_if_subnet_uses_eui64(subnet)):
+ ipv6_utils.is_slaac_subnet(subnet)):
prefix = subnet['cidr']
ip_address = ipv6_utils.get_ipv6_addr_by_EUI64(
prefix, mac_address)
return ips
def _update_ips_for_port(self, context, network_id, port_id, original_ips,
- new_ips, mac_address):
+ new_ips, mac_address, device_owner):
"""Add or remove IPs from the port."""
ips = []
# These ips are still on the port and haven't been removed
prev_ips.append(original_ip)
# Check if the IP's to add are OK
- to_add = self._test_fixed_ips_for_port(context, network_id, new_ips)
+ to_add = self._test_fixed_ips_for_port(context, network_id, new_ips,
+ device_owner)
for ip in original_ips:
LOG.debug(_("Port update. Hold %s"), ip)
NeutronDbPluginV2._delete_ip_allocation(context,
if fixed_configured:
configured_ips = self._test_fixed_ips_for_port(context,
p["network_id"],
- p['fixed_ips'])
+ p['fixed_ips'],
+ p['device_owner'])
ips = self._allocate_fixed_ips(context,
configured_ips,
p['mac_address'])
else:
v6.append(subnet)
for subnet in v6:
- if self._check_if_subnet_uses_eui64(subnet):
+ if ipv6_utils.is_slaac_subnet(subnet):
#(dzyu) If true, calculate an IPv6 address
# by mac address and prefix, then remove this
# subnet from the array of subnets that will be passed
# id together should be equal to 128. Currently neutron supports
# EUI64 interface id only, thus limiting the prefix
# length to be 64 only.
- if self._check_if_subnet_uses_eui64(subnet):
+ if ipv6_utils.is_slaac_subnet(subnet):
if netaddr.IPNetwork(subnet['cidr']).prefixlen != 64:
msg = _('Invalid CIDR %s for IPv6 address mode. '
'OpenStack uses the EUI-64 address format, '
s = subnet['subnet']
if s['gateway_ip'] is attributes.ATTR_NOT_SPECIFIED:
- s['gateway_ip'] = str(netaddr.IPAddress(net.first + 1))
+ if s['ip_version'] == 6 and ipv6_utils.is_slaac_subnet(s):
+ s['gateway_ip'] = None
+ else:
+ s['gateway_ip'] = str(netaddr.IPAddress(net.first + 1))
if s['allocation_pools'] == attributes.ATTR_NOT_SPECIFIED:
s['allocation_pools'] = self._allocate_pools_for_subnet(context, s)
added_ips, prev_ips = self._update_ips_for_port(
context, port["network_id"], id,
original["fixed_ips"], p['fixed_ips'],
- original['mac_address'])
+ original['mac_address'], port['device_owner'])
# Update ips if necessary
for ip in added_ips:
from neutron.api.v2 import attributes
from neutron.common import constants as l3_constants
from neutron.common import exceptions as n_exc
+from neutron.common import ipv6_utils
from neutron.common import rpc as n_rpc
from neutron.common import utils
from neutron.db import model_base
def _add_interface_by_subnet(self, context, router, subnet_id, owner):
subnet = self._core_plugin._get_subnet(context, subnet_id)
- if not subnet['gateway_ip']:
+ if (not subnet['gateway_ip']
+ and not ipv6_utils.is_slaac_subnet(subnet)):
msg = _('Subnet for router interface must have a gateway IP')
raise n_exc.BadRequest(resource='router', msg=msg)
self._check_for_dup_router_subnet(context, router,
subnet['network_id'],
subnet_id,
subnet['cidr'])
- fixed_ip = {'ip_address': subnet['gateway_ip'],
- 'subnet_id': subnet['id']}
+ if subnet['gateway_ip']:
+ fixed_ip = {'ip_address': subnet['gateway_ip'],
+ 'subnet_id': subnet['id']}
+ else:
+ fixed_ip = {'subnet_id': subnet['id']}
+
return self._core_plugin.create_port(context, {
'port':
{'tenant_id': subnet['tenant_id'],
_unsupported = (
'test_create_subnet_default_gw_conflict_allocation_pool_returns_409',
'test_create_subnet_defaults', 'test_create_subnet_gw_values',
+ 'test_create_subnet_ipv6_gw_values',
'test_update_subnet_gateway_in_allocation_pool_returns_409',
'test_update_subnet_allocation_pools',
'test_update_subnet_allocation_pools_invalid_pool_for_cidr')
self.assertEqual(res.status_int,
webob.exc.HTTPClientError.code)
+ def test_requested_fixed_ip_address_v6_slaac_router_iface(self):
+ with self.subnet(gateway_ip='fe80::1',
+ cidr='fe80::/64',
+ ip_version=6,
+ ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
+ kwargs = {"fixed_ips": [{'subnet_id': subnet['subnet']['id'],
+ 'ip_address': 'fe80::1'}]}
+ net_id = subnet['subnet']['network_id']
+ device_owner = constants.DEVICE_OWNER_ROUTER_INTF
+ res = self._create_port(self.fmt, net_id=net_id,
+ device_owner=device_owner, **kwargs)
+ port = self.deserialize(self.fmt, res)
+ self.assertEqual(len(port['port']['fixed_ips']), 1)
+ self.assertEqual(port['port']['fixed_ips'][0]['ip_address'],
+ 'fe80::1')
+
def test_requested_subnet_id_v6_slaac(self):
with self.subnet(gateway_ip='fe80::1',
cidr='2607:f0d0:1002:51::/64',
self._test_create_subnet(expected=expected,
gateway_ip=gateway)
+ def test_create_subnet_ipv6_gw_values(self):
+ cidr = '2001::/64'
+ # Gateway is last IP in IPv6 DHCPv6 stateful subnet
+ gateway = '2001::ffff:ffff:ffff:fffe'
+ allocation_pools = [{'start': '2001::1',
+ 'end': '2001::ffff:ffff:ffff:fffd'}]
+ expected = {'gateway_ip': gateway,
+ 'cidr': cidr,
+ 'allocation_pools': allocation_pools}
+ self._test_create_subnet(expected=expected, gateway_ip=gateway,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode=constants.DHCPV6_STATEFUL,
+ ipv6_address_mode=constants.DHCPV6_STATEFUL)
+ # Gateway is first IP in IPv6 DHCPv6 stateful subnet
+ gateway = '2001::1'
+ allocation_pools = [{'start': '2001::2',
+ 'end': '2001::ffff:ffff:ffff:fffe'}]
+ expected = {'gateway_ip': gateway,
+ 'cidr': cidr,
+ 'allocation_pools': allocation_pools}
+ self._test_create_subnet(expected=expected, gateway_ip=gateway,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode=constants.DHCPV6_STATEFUL,
+ ipv6_address_mode=constants.DHCPV6_STATEFUL)
+ # Gateway not specified for IPv6 SLAAC subnet
+ expected = {'gateway_ip': None,
+ 'cidr': cidr}
+ self._test_create_subnet(expected=expected,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode=constants.IPV6_SLAAC,
+ ipv6_address_mode=constants.IPV6_SLAAC)
+
def test_create_subnet_gw_outside_cidr_returns_400(self):
with self.network() as network:
self._create_subnet(self.fmt,
Code Review / openstack-build / neutron-build.git / commitdiff