Skip to content

Commit

Permalink
Make IPv6 and NAT pool work together
Browse files Browse the repository at this point in the history 
When providing multiple IP addresses on router create or router update,
we used to take all of them in consideration for NAT pooling. This
prevents user from updating their router with an IPv6 address (as "a
router with two addresses is not allowed for NAT pooling") and also will
fail when trying to create a NAT pool that contains IPv6 addresses. Now
we split all provided addresses in two lists, one per address family.
All IPv4 addresses go the regular NAT pooling / no NAT pooling path. For
IPv6 we allow zero or one address. Providing more than one IPv6 address
will show a nice error message to the user.
  • Loading branch information
@sebageek
sebageek committed Feb 13, 2025
1 parent 535f528 commit d202d0d
Show file tree
Hide file tree
Showing 3 changed files with 127 additions and 13 deletions.
8 changes: 8 additions & 0 deletions asr1k_neutron_l3/common/asr1k_exceptions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -148,3 +148,11 @@ class DynamicNatPoolGivenIPsDontBelongToNetwork(nexception.BadRequest):
class DynamicNatPoolExternalNetExhausted(nexception.BadRequest):
message = ("Could not find %(ip_count)s consecutive IP addresses in subnet %(subnet_id)s - "
"make sure there is enough undivided IP space")


class OnlyOneExternalIPv6AddressAllowed(nexception.BadRequest):
message = ("Only one external IPv6 address allowed per router")


class InvalidExternalGatewayIPDefinition(nexception.BadRequest):
message = ("Invalid external gateway ip definition found: %(ext_ip)s")
25 changes: 23 additions & 2 deletions asr1k_neutron_l3/plugins/l3/service_plugins/l3_extension_adapter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -405,8 +405,29 @@ def _update_router_gw_info(self, context, router_id, info,
ext_ips = info.get('external_fixed_ips', []) if info else []
orig_router_atts = self.db.get_router_att(context, router_id)
dynamic_nat_pool = None
if len(ext_ips) > 1:
dynamic_nat_pool = self._handle_dynamic_nat_pool_allocation(context, router_id, info, ext_ips)
ext_ips_v4 = []
ext_ips_v6 = []

for ext_ip in ext_ips:
ip_version = None
if 'subnet_id' in ext_ip:
# this will throw a "subnet not found" if the user specified an invalid subnet
ip_version = self.db.get_subnet(context, ext_ip['subnet_id'])['ip_version']
elif 'ip_address' in ext_ip:
ip_version = utils.get_ip_version(ext_ip['ip_address'])

if ip_version == 4:
ext_ips_v4.append(ext_ip)
elif ip_version == 6:
ext_ips_v6.append(ext_ip)
else:
# in theory we should always have an ip version, buf if not we have this safeguard here
raise asr1k_exc.InvalidExternalGatewayIPDefinition(ext_ip=ext_ip)

if len(ext_ips_v6) > 1:
raise asr1k_exc.OnlyOneExternalIPv6AddressAllowed()
if len(ext_ips_v4) > 1:
dynamic_nat_pool = self._handle_dynamic_nat_pool_allocation(context, router_id, info, ext_ips_v4)

result = super()._update_router_gw_info(context, router_id, info, request_body, router)
if dynamic_nat_pool is not None or orig_router_atts.dynamic_nat_pool:
Expand Down
107 changes: 96 additions & 11 deletions asr1k_neutron_l3/tests/unit/plugins/l3/service_plugins/test_l3_extension_adapter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -189,17 +189,19 @@ def test_router_create_with_extended_nat_pool_mixed_specified_pool(self, pc_mock
router["NeutronError"]["type"])

def test_router_create_with_extended_nat_pool_two_subnets_found(self, pc_mock):
with self.subnet(cidr="10.100.1.0/24") as s:
self._set_net_external(s['subnet']['network_id'])
with self.router(name="r1", admin_state_up=True, tenant_id=uuidutils.generate_uuid(),
external_gateway_info={'network_id': s['subnet']['network_id'],
'external_fixed_ips': [
{'subnet_id': uuidutils.generate_uuid()},
{'subnet_id': uuidutils.generate_uuid()},
{'subnet_id': s['subnet']['id']},
]}) as router:
self.assertEqual("DynamicNatPoolTwoSubnetsFound",
router["NeutronError"]["type"])
with self.network() as net:
self._set_net_external(net['network']['id'])
with self.subnet(cidr="10.100.1.0/24", network=net) as s1, \
self.subnet(cidr="10.100.4.0/24", network=net) as s2:
with self.router(name="r1", admin_state_up=True, tenant_id=uuidutils.generate_uuid(),
external_gateway_info={'network_id': net['network']['id'],
'external_fixed_ips': [
{'subnet_id': s1['subnet']['id']},
{'subnet_id': s2['subnet']['id']},
{'subnet_id': s2['subnet']['id']},
]}) as router:
self.assertEqual("DynamicNatPoolTwoSubnetsFound",
router["NeutronError"]["type"])

def test_router_create_with_extended_nat_pool_too_small(self, pc_mock):
with self.subnet(cidr="10.100.1.0/24") as s:
Expand Down Expand Up @@ -350,3 +352,86 @@ def test_router_create_with_extended_nat_pool_uses_smallest_segment(self, pc_moc
db = asr1k_db.get_db_plugin()
router_atts = db.get_router_att(ctx, router['router']['id'])
self.assertEqual("10.100.1.25-10.100.1.28/27", router_atts.dynamic_nat_pool)

def test_router_create_dualstack_without_nat_pool_given_subnet(self, pc_mock):
ctx = context.get_admin_context()

with self.network() as net:
self._set_net_external(net['network']['id'])
with self.subnet(cidr="10.100.1.0/24", network=net) as s1, \
self.subnet(cidr="2001:db8::/64", network=net, ip_version=6) as s2:
with self.router(name="r1", admin_state_up=True, tenant_id=uuidutils.generate_uuid(),
external_gateway_info={'network_id': net['network']['id'],
'external_fixed_ips': [
{'subnet_id': s1['subnet']['id']},
{'subnet_id': s2['subnet']['id']},
]}) as router:
db = asr1k_db.get_db_plugin()
router_atts = db.get_router_att(ctx, router['router']['id'])
self.assertIsNone(router_atts.dynamic_nat_pool)
self.assertEqual(2, len(router['router']['external_gateway_info']['external_fixed_ips']))

def test_router_create_dualstack_without_nat_pool_given_ipv6_addr(self, pc_mock):
ctx = context.get_admin_context()

with self.network() as net:
self._set_net_external(net['network']['id'])
with self.subnet(cidr="10.100.1.0/24", network=net) as s1, \
self.subnet(cidr="2001:db8::/64", network=net, ip_version=6):
with self.router(name="r1", admin_state_up=True, tenant_id=uuidutils.generate_uuid(),
external_gateway_info={'network_id': net['network']['id'],
'external_fixed_ips': [
{'subnet_id': s1['subnet']['id']},
{'ip_address': '2001:db8::23'},
]}) as router:
db = asr1k_db.get_db_plugin()
router_atts = db.get_router_att(ctx, router['router']['id'])
self.assertIsNone(router_atts.dynamic_nat_pool)
self.assertEqual(2, len(router['router']['external_gateway_info']['external_fixed_ips']))

def test_router_create_dualstack_with_nat_pool(self, pc_mock):
ctx = context.get_admin_context()

with self.network() as net:
self._set_net_external(net['network']['id'])
with self.subnet(cidr="10.100.1.0/24", network=net) as s1, \
self.subnet(cidr="2001:db8::/64", network=net, ip_version=6) as s2:
with self.router(name="r1", admin_state_up=True, tenant_id=uuidutils.generate_uuid(),
external_gateway_info={'network_id': net['network']['id'],
'external_fixed_ips': [
{'subnet_id': s1['subnet']['id']},
{'subnet_id': s1['subnet']['id']},
{'subnet_id': s2['subnet']['id']},
{'subnet_id': s1['subnet']['id']},
{'subnet_id': s1['subnet']['id']},
{'subnet_id': s1['subnet']['id']},
]}) as router:
db = asr1k_db.get_db_plugin()
router_atts = db.get_router_att(ctx, router['router']['id'])
self.assertEqual("10.100.1.2-10.100.1.5/24", router_atts.dynamic_nat_pool)
ext_ips = router['router']['external_gateway_info']['external_fixed_ips']
self.assertEqual(6, len(ext_ips))
self.assertEqual(1, len([ip for ip in ext_ips if ':' in ip['ip_address']]))

def test_router_create_two_v6_addresses_fail(self, pc_mock):
with self.subnet(cidr="2001:db8::/64", ip_version=6) as s:
self._set_net_external(s['subnet']['network_id'])
with self.router(name="r1", admin_state_up=True, tenant_id=uuidutils.generate_uuid(),
external_gateway_info={'network_id': s['subnet']['network_id'],
'external_fixed_ips': [
{'subnet_id': s['subnet']['id']},
{'subnet_id': s['subnet']['id']},
]}) as router:
self.assertEqual("OnlyOneExternalIPv6AddressAllowed",
router["NeutronError"]["type"])

def test_router_create_non_existant_subnet(self, pc_mock):
with self.subnet(cidr="10.100.1.0/24") as s:
self._set_net_external(s['subnet']['network_id'])
with self.router(name="r1", admin_state_up=True, tenant_id=uuidutils.generate_uuid(),
external_gateway_info={'network_id': s['subnet']['network_id'],
'external_fixed_ips': [
{'subnet_id': uuidutils.generate_uuid()},
]}) as router:
self.assertEqual("SubnetNotFound",
router["NeutronError"]["type"])

0 comments on commit d202d0d

Please sign in to comment.