Merge "NSX|V3: ensure external subnet does not conflicts with T0 address"

vmware_nsx/plugins/nsx_v3/plugin.py

@@ -1638,7 +1638,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
                 LOG.error("Unable to delete DHCP server mapping for "
                           "network %s", network_id)
 
-    def _validate_address_space(self, subnet):
+    def _validate_address_space(self, context, subnet):
         cidr = subnet.get('cidr')
         if (not validators.is_attr_set(cidr) or
             netaddr.IPNetwork(cidr).version != 4):
@@ -1649,6 +1649,28 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
             msg = _("Subnet overlaps with shared address space 100.64.0.0/10")
             LOG.error(msg)
             raise n_exc.InvalidInput(error_message=msg)
+        # Ensure that the NSX uplink does not lie on the same subnet as
+        # the external subnet
+        filters = {'id': [subnet['network_id']],
+                   'router:external': [True]}
+        nets = self.get_networks(context, filters=filters)
+        for net in nets:
+            tier0 = net.get(pnet.PHYSICAL_NETWORK)
+            if tier0:
+                ports = self.nsxlib.logical_router_port.get_by_router_id(tier0)
+                for port in ports:
+                    if (port.get('resource_type') ==
+                        'LogicalRouterUpLinkPort'):
+                        for subnet in port.get('subnets', []):
+                            for ip_address in subnet.get('ip_addresses'):
+                                if (netaddr.IPAddress(ip_address) in
+                                    netaddr.IPNetwork(cidr)):
+                                    msg = _("External subnet cannot "
+                                            "overlap with T0 router "
+                                            "address!")
+                                    LOG.error(msg)
+                                    raise n_exc.InvalidInput(
+                                            error_message=msg)
 
     def _create_bulk_with_callback(self, resource, context, request_items,
                                    post_create_func=None, rollback_func=None):
@@ -1746,7 +1768,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
             return self._create_bulk('subnet', context, subnets)
 
     def create_subnet(self, context, subnet):
-        self._validate_address_space(subnet['subnet'])
+        self._validate_address_space(context, subnet['subnet'])
         # TODO(berlin): public external subnet announcement
         if (cfg.CONF.nsx_v3.native_dhcp_metadata and
             subnet['subnet'].get('enable_dhcp', False)):

vmware_nsx/tests/unit/nsx_v3/test_plugin.py

@@ -616,6 +616,30 @@ class TestSubnetsV2(test_plugin.TestSubnetsV2, NsxV3PluginTestCaseMixin):
                               self.plugin.create_subnet,
                               context.get_admin_context(), data)
 
+    def _create_external_network(self):
+        data = {'network': {'name': 'net1',
+                            'router:external': 'True',
+                            'tenant_id': 'tenant_one',
+                            'provider:physical_network': 'stam'}}
+        network_req = self.new_create_request('networks', data)
+        network = self.deserialize(self.fmt,
+                                   network_req.get_response(self.api))
+        return network
+
+    def test_create_subnet_with_conflicting_t0_address(self):
+        network = self._create_external_network()
+        data = {'subnet': {'network_id': network['network']['id'],
+                           'cidr': '172.20.1.0/24'}}
+        ports = [{'subnets': [{'ip_addresses': [u'172.20.1.60'],
+                               'prefix_length': 24}],
+                  'resource_type': 'LogicalRouterUpLinkPort'}]
+        with mock.patch.object(self.plugin.nsxlib.logical_router_port,
+                               'get_by_router_id',
+                               return_value=ports):
+            self.assertRaises(n_exc.InvalidInput,
+                              self.plugin.create_subnet,
+                              context.get_admin_context(), data)
+
     def test_subnet_update_ipv4_and_ipv6_pd_v6stateless_subnets(self):
         self.skipTest('Multiple fixed ips on a port are not supported')