openstack
/
neutron
Code Issues Proposed changes

Subnet gateway should be a valid IP 

When a subnet is created and allocated, the gateway IP is created
based on the subnet CIDR. In case of IPv6 prefix delegation, this
CIDR is a temporary one. In this case the gateway IP cannot be
assigned yet and the value stored in the DB should be "None".

IpamBackendMixin._gateway_ip_str must read properly the IP version
stored in the "subnet" variable, under the key "ip_version"
instead of "version".

Closes-Bug: #1856726
Closes-Bug: #1856675

Change-Id: I9313c880cc458f08dc3a1b0ff13187b764ba7042
This commit is contained in:
Rodolfo Alonso Hernandez 2019-12-17 16:50:36 +00:00
parent 4b7c2dceb4
commit 0b3cff33c1
8 changed files with 128 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
neutron/db/ipam_backend_mixin.py
View File

@ -56,10 +56,14 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
@staticmethod
def _gateway_ip_str(subnet, cidr_net):
if subnet.get('gateway_ip') is const.ATTR_NOT_SPECIFIED:
if subnet.get('version') == const.IP_VERSION_6:
return str(netaddr.IPNetwork(cidr_net).network)
if subnet.get('ip_version') == const.IP_VERSION_6:
gateway_ip = netaddr.IPNetwork(cidr_net).network
pd_net = netaddr.IPNetwork(const.PROVISIONAL_IPV6_PD_PREFIX)
if gateway_ip == pd_net.network:
return
else:
return str(netaddr.IPNetwork(cidr_net).network + 1)
gateway_ip = netaddr.IPNetwork(cidr_net).network + 1
return str(gateway_ip)
return subnet.get('gateway_ip')
@staticmethod

16
neutron/ipam/utils.py
View File

@ -17,7 +17,7 @@ import netaddr
from neutron_lib import constants
def check_subnet_ip(cidr, ip_address, port_owner=None):
def check_subnet_ip(cidr, ip_address, port_owner=''):
"""Validate that the IP address is on the subnet."""
ip = netaddr.IPAddress(ip_address)
net = netaddr.IPNetwork(cidr)
@ -25,16 +25,12 @@ def check_subnet_ip(cidr, ip_address, port_owner=None):
# network or the broadcast address
if net.version == constants.IP_VERSION_6:
# NOTE(njohnston): In some cases the code cannot know the owner of the
# port. In these cases port_owner should be None, and we pass it
# through here.
return ((port_owner in constants.ROUTER_PORT_OWNERS or
port_owner is None or
ip != net.network) and
net.netmask & ip == net.network)
# port. In these cases port_owner should an empty string, and we pass
# it through here.
return (port_owner in (constants.ROUTER_PORT_OWNERS + ('', )) and
ip in net)
else:
return (ip != net.network and
ip != net[-1] and
net.netmask & ip == net.network)
return ip != net.network and ip != net.broadcast and ip in net
def check_gateway_invalid_in_subnet(cidr, gateway):

13
neutron/tests/fullstack/resources/client.py
View File

@ -107,20 +107,25 @@ class ClientFixture(fixtures.Fixture):
def create_subnet(self, tenant_id, network_id,
cidr, gateway_ip=None, name=None, enable_dhcp=True,
ipv6_address_mode='slaac', ipv6_ra_mode='slaac'):
ipv6_address_mode='slaac', ipv6_ra_mode='slaac',
subnetpool_id=None, ip_version=None):
resource_type = 'subnet'
name = name or utils.get_rand_name(prefix=resource_type)
ip_version = netaddr.IPNetwork(cidr).version
if cidr and not ip_version:
ip_version = netaddr.IPNetwork(cidr).version
spec = {'tenant_id': tenant_id, 'network_id': network_id, 'name': name,
'cidr': cidr, 'enable_dhcp': enable_dhcp,
'ip_version': ip_version}
'enable_dhcp': enable_dhcp, 'ip_version': ip_version}
if ip_version == constants.IP_VERSION_6:
spec['ipv6_address_mode'] = ipv6_address_mode
spec['ipv6_ra_mode'] = ipv6_ra_mode
if gateway_ip:
spec['gateway_ip'] = gateway_ip
if subnetpool_id:
spec['subnetpool_id'] = subnetpool_id
if cidr:
spec['cidr'] = cidr
return self._create_resource(resource_type, spec)

82
neutron/tests/fullstack/test_subnet.py Normal file
View File

@ -0,0 +1,82 @@
# Copyright 2019 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import netaddr
from neutron_lib import constants
from oslo_utils import uuidutils
from neutron.tests.common.exclusive_resources import ip_network
from neutron.tests.fullstack import base
from neutron.tests.fullstack.resources import environment
class TestSubnet(base.BaseFullStackTestCase):
def setUp(self):
host_descriptions = [
environment.HostDescription(l3_agent=True, dhcp_agent=True),
environment.HostDescription()]
env = environment.Environment(
environment.EnvironmentDescription(network_type='vlan',
l2_pop=False),
host_descriptions)
super(TestSubnet, self).setUp(env)
self._project_id = uuidutils.generate_uuid()
self._network = self._create_network(self._project_id)
def _create_network(self, project_id, name='test_network'):
return self.safe_client.create_network(project_id, name=name)
def _create_subnet(self, project_id, network_id, cidr,
ipv6_address_mode=None, ipv6_ra_mode=None,
subnetpool_id=None):
ip_version = None
if ipv6_address_mode or ipv6_ra_mode:
ip_version = constants.IP_VERSION_6
return self.safe_client.create_subnet(
project_id, network_id, cidr, enable_dhcp=True,
ipv6_address_mode=ipv6_address_mode, ipv6_ra_mode=ipv6_ra_mode,
subnetpool_id=subnetpool_id, ip_version=ip_version)
def _show_subnet(self, subnet_id):
return self.client.show_subnet(subnet_id)
def test_create_subnet_ipv4(self):
cidr = self.useFixture(
ip_network.ExclusiveIPNetwork(
'240.0.0.0', '240.255.255.255', '24')).network
subnet = self._create_subnet(self._project_id, self._network['id'],
cidr)
subnet = self._show_subnet(subnet['id'])
self.assertEqual(subnet['subnet']['gateway_ip'],
str(netaddr.IPNetwork(cidr).network + 1))
def test_create_subnet_ipv6_slaac(self):
cidr = self.useFixture(
ip_network.ExclusiveIPNetwork(
'2001:db8::', '2001:db8::ffff', '64')).network
subnet = self._create_subnet(self._project_id, self._network['id'],
cidr, ipv6_address_mode='slaac',
ipv6_ra_mode='slaac')
subnet = self._show_subnet(subnet['id'])
self.assertEqual(subnet['subnet']['gateway_ip'],
str(netaddr.IPNetwork(cidr).network))
def test_create_subnet_ipv6_prefix_delegation(self):
subnet = self._create_subnet(self._project_id, self._network['id'],
None, ipv6_address_mode='slaac',
ipv6_ra_mode='slaac',
subnetpool_id='prefix_delegation')
subnet = self._show_subnet(subnet['id'])
self.assertIsNone(subnet['subnet']['gateway_ip'])

7
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -3934,7 +3934,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
ipv6_ra_mode=constants.DHCPV6_STATEFUL,
ipv6_address_mode=constants.DHCPV6_STATEFUL)
# If gateway_ip is not specified, allocate first IP from the subnet
expected = {'gateway_ip': gateway,
expected = {'gateway_ip': str(netaddr.IPNetwork(cidr).network),
'cidr': cidr}
self._test_create_subnet(expected=expected,
cidr=cidr, ip_version=constants.IP_VERSION_6,
@ -3966,8 +3966,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
cidr=cidr, ip_version=constants.IP_VERSION_6,
ipv6_ra_mode=constants.DHCPV6_STATELESS,
ipv6_address_mode=constants.DHCPV6_STATELESS)
# If gateway_ip is not specified, allocate first IP from the subnet
expected = {'gateway_ip': gateway,
# If gateway_ip is not specified and the subnet is using prefix
# delegation, until the CIDR is assigned, this value should be "None"
expected = {'gateway_ip': None,
'cidr': cidr}
self._test_create_subnet(expected=expected,
cidr=cidr, ip_version=constants.IP_VERSION_6,

4
neutron/tests/unit/db/test_ipam_pluggable_backend.py
View File

@ -406,7 +406,9 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
mocks = self._prepare_mocks_with_pool_mock(pool_mock)
cfg.CONF.set_override('ipv6_pd_enabled', True)
cidr = constants.PROVISIONAL_IPV6_PD_PREFIX
allocation_pools = [netaddr.IPRange('::2', '::ffff:ffff:ffff:ffff')]
cidr_network = netaddr.IPNetwork(cidr)
allocation_pools = [netaddr.IPRange(cidr_network.ip + 1,
cidr_network.last)]
with self.subnet(cidr=None, ip_version=constants.IP_VERSION_6,
subnetpool_id=constants.IPV6_PD_POOL_ID,
ipv6_ra_mode=constants.IPV6_SLAAC,

25
neutron/tests/unit/extensions/test_network_ip_availability.py
View File

@ -13,6 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import netaddr
from neutron_lib import constants
import neutron.api.extensions as api_ext
@ -215,9 +216,11 @@ class TestNetworkIPAvailabilityAPI(
self.assertEqual(0, len(response[IP_AVAILS_KEY]))
def test_usages_query_ip_version_v6(self):
cidr_ipv6 = '2001:db8:1002:51::/64'
cidr_ipv6_net = netaddr.IPNetwork(cidr_ipv6)
with self.network() as net:
with self.subnet(
network=net, cidr='2607:f0d0:1002:51::/64',
network=net, cidr=cidr_ipv6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.DHCPV6_STATELESS):
# Get IPv6
@ -227,7 +230,7 @@ class TestNetworkIPAvailabilityAPI(
request.get_response(self.ext_api))
self.assertEqual(1, len(response[IP_AVAILS_KEY]))
self._validate_from_availabilities(
response[IP_AVAILS_KEY], net, 0, 18446744073709551614)
response[IP_AVAILS_KEY], net, 0, cidr_ipv6_net.size - 1)
# Get IPv4 should return empty array
params = 'ip_version=%s' % constants.IP_VERSION_4
@ -237,9 +240,11 @@ class TestNetworkIPAvailabilityAPI(
self.assertEqual(0, len(response[IP_AVAILS_KEY]))
def test_usages_ports_consumed_v6(self):
cidr_ipv6 = '2001:db8:1002:51::/64'
cidr_ipv6_net = netaddr.IPNetwork(cidr_ipv6)
with self.network() as net:
with self.subnet(
network=net, cidr='2607:f0d0:1002:51::/64',
network=net, cidr=cidr_ipv6,
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.DHCPV6_STATELESS) as subnet:
request = self.new_list_request(API_RESOURCE)
@ -252,7 +257,7 @@ class TestNetworkIPAvailabilityAPI(
self._validate_from_availabilities(response[IP_AVAILS_KEY],
net, 3,
18446744073709551614)
cidr_ipv6_net.size - 1)
def test_usages_query_network_id(self):
with self.network() as net:
@ -345,14 +350,16 @@ class TestNetworkIPAvailabilityAPI(
def test_usages_multi_net_multi_subnet_46(self):
# Setup mixed v4/v6 networks with IPs consumed on each
cidr_ipv6 = '2001:db8:1003:52::/64'
cidr_ipv6_net = netaddr.IPNetwork(cidr_ipv6)
with self.network(name='net-v6-1') as net_v6_1, \
self.network(name='net-v6-2') as net_v6_2, \
self.network(name='net-v4-1') as net_v4_1, \
self.network(name='net-v4-2') as net_v4_2:
with self.subnet(network=net_v6_1, cidr='2607:f0d0:1002:51::/64',
with self.subnet(network=net_v6_1, cidr='2001:db8:1002:51::/64',
ip_version=constants.IP_VERSION_6) as s61, \
self.subnet(network=net_v6_2,
cidr='2607:f0d0:1003:52::/64',
cidr=cidr_ipv6,
ip_version=constants.IP_VERSION_6) as s62, \
self.subnet(network=net_v4_1, cidr='10.0.0.0/24') as s41, \
self.subnet(network=net_v4_2, cidr='10.0.1.0/24') as s42:
@ -367,9 +374,9 @@ class TestNetworkIPAvailabilityAPI(
self.fmt, request.get_response(self.ext_api))
avails_list = response[IP_AVAILS_KEY]
self._validate_from_availabilities(
avails_list, net_v6_1, 1, 18446744073709551614)
avails_list, net_v6_1, 1, cidr_ipv6_net.size - 1)
self._validate_from_availabilities(
avails_list, net_v6_2, 2, 18446744073709551614)
avails_list, net_v6_2, 2, cidr_ipv6_net.size - 1)
self._validate_from_availabilities(
avails_list, net_v4_1, 1, 253)
self._validate_from_availabilities(
@ -397,6 +404,6 @@ class TestNetworkIPAvailabilityAPI(
self.fmt, request.get_response(self.ext_api))
avails_list = response[IP_AVAILS_KEY]
self._validate_from_availabilities(
avails_list, net_v6_2, 2, 18446744073709551614)
avails_list, net_v6_2, 2, cidr_ipv6_net.size - 1)
self._validate_from_availabilities(
avails_list, net_v4_2, 2, 253)

2
neutron/tests/unit/services/ovn_l3/test_plugin.py
View File

@ -1452,6 +1452,6 @@ class OVNL3ExtrarouteTests(test_l3_gw.ExtGwModeIntTestCase,
test_router_update_gateway_upon_subnet_create_max_ips_ipv6()
add_static_route_calls = [
mock.call(mock.ANY, ip_prefix='0.0.0.0/0', nexthop='10.0.0.1'),
mock.call(mock.ANY, ip_prefix='::/0', nexthop='2001:db8::1')]
mock.call(mock.ANY, ip_prefix='::/0', nexthop='2001:db8::')]
self.l3_inst._ovn.add_static_route.assert_has_calls(
add_static_route_calls, any_order=True)