diff --git a/vmware_nsx/plugins/common_v3/plugin.py b/vmware_nsx/plugins/common_v3/plugin.py index 59ce040df8..cd320d82e7 100644 --- a/vmware_nsx/plugins/common_v3/plugin.py +++ b/vmware_nsx/plugins/common_v3/plugin.py @@ -249,10 +249,11 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin, self._validate_address_scope_for_router_interface( context.elevated(), router_db.id, gw_network_id, subnet['id']) - def _validate_ipv4_address_pairs(self, address_pairs): + def _validate_address_pairs(self, address_pairs): for pair in address_pairs: ip = pair.get('ip_address') - if not utils.is_ipv4_ip_address(ip): + if len(ip.split('/')) > 1: + LOG.error("cidr is not supported in allowed address pairs") raise nsx_exc.InvalidIPAddress(ip_address=ip) def _create_port_address_pairs(self, context, port_data): @@ -264,7 +265,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin, if not port_security: raise addr_exc.AddressPairAndPortSecurityRequired() else: - self._validate_ipv4_address_pairs(address_pairs) + self._validate_address_pairs(address_pairs) self._process_create_allowed_address_pairs(context, port_data, address_pairs) else: @@ -348,7 +349,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin, raise addr_exc.AddressPairAndPortSecurityRequired() if delete_addr_pairs or has_addr_pairs: - self._validate_ipv4_address_pairs( + self._validate_address_pairs( updated_port[addr_apidef.ADDRESS_PAIRS]) # delete address pairs and read them in self._delete_allowed_address_pairs(context, id) diff --git a/vmware_nsx/tests/unit/extensions/test_addresspairs.py b/vmware_nsx/tests/unit/extensions/test_addresspairs.py index 8b031ae597..dfe0a9950f 100644 --- a/vmware_nsx/tests/unit/extensions/test_addresspairs.py +++ b/vmware_nsx/tests/unit/extensions/test_addresspairs.py @@ -55,6 +55,18 @@ class TestAllowedAddressPairsNSXv3(test_v3_plugin.NsxV3PluginTestCaseMixin, 'ip_address': '10.0.0.1/24'}] self._create_port_with_address_pairs(address_pairs, 400) + def test_create_port_allowed_address_pairs_v6(self): + with self.network() as net: + address_pairs = [{'ip_address': '1001::12'}] + res = self._create_port(self.fmt, net['network']['id'], + arg_list=(addr_apidef.ADDRESS_PAIRS,), + allowed_address_pairs=address_pairs) + port = self.deserialize(self.fmt, res) + address_pairs[0]['mac_address'] = port['port']['mac_address'] + self.assertEqual(port['port'][addr_apidef.ADDRESS_PAIRS], + address_pairs) + self._delete('ports', port['port']['id']) + def test_update_add_bad_address_pairs_with_cidr(self): with self.network() as net: res = self._create_port(self.fmt, net['network']['id'])