Merge "Revert "DVS: Disable port security and security group""

vmware_nsx/plugins/dvs/plugin.py

@@ -444,21 +444,14 @@ class NsxDvsV2(addr_pair_db.AllowedAddressPairsMixin,
         # ATTR_NOT_SPECIFIED is for the case where a port is created on a
         # shared network that is not owned by the tenant.
         port_data = port['port']
-        network_type = self._dvs_get_network_type(context, port['port'][
-                                                  'network_id'])
         with db_api.CONTEXT_WRITER.using(context):
             # First we allocate port in neutron database
             neutron_db = super(NsxDvsV2, self).create_port(context, port)
             self._extension_manager.process_create_port(
                 context, port_data, neutron_db)
-            if network_type and network_type == 'vlan':
+            port_security = self._get_network_security_binding(
-                # Not allowed to enable port security on vlan DVS ports
+                context, neutron_db['network_id'])
-                port_data[psec.PORTSECURITY] = False
+            port_data[psec.PORTSECURITY] = port_security
-                port_security = False
-            else:
-                port_security = self._get_network_security_binding(
-                    context, neutron_db['network_id'])
-                port_data[psec.PORTSECURITY] = port_security
             self._process_port_port_security_create(
                 context, port_data, neutron_db)
             # Update fields obtained from neutron db (eg: MAC address)
@@ -466,17 +459,12 @@ class NsxDvsV2(addr_pair_db.AllowedAddressPairsMixin,
             has_ip = self._ip_on_port(neutron_db)
 
             # security group extension checks
-            if network_type and network_type != 'vlan':
+            if has_ip:
-                if has_ip:
+                self._ensure_default_security_group_on_port(context, port)
-                    self._ensure_default_security_group_on_port(context, port)
+            elif validators.is_attr_set(port_data.get(ext_sg.SECURITYGROUPS)):
-                elif validators.is_attr_set(port_data.get(
+                raise psec_exc.PortSecurityAndIPRequiredForSecurityGroups()
-                        ext_sg.SECURITYGROUPS)):
+            port_data[ext_sg.SECURITYGROUPS] = (
-                    raise psec_exc.PortSecurityAndIPRequiredForSecurityGroups()
+                self._get_security_groups_on_port(context, port))
-            if network_type and network_type == 'vlan':
-                port_data[ext_sg.SECURITYGROUPS] = []
-            else:
-                port_data[ext_sg.SECURITYGROUPS] = (
-                    self._get_security_groups_on_port(context, port))
             self._process_port_create_security_group(
                 context, port_data, port_data[ext_sg.SECURITYGROUPS])
             self._process_portbindings_create_and_update(context,
@@ -514,6 +502,7 @@ class NsxDvsV2(addr_pair_db.AllowedAddressPairsMixin,
         delete_addr_pairs = self._check_update_deletes_allowed_address_pairs(
             port)
         has_addr_pairs = self._check_update_has_allowed_address_pairs(port)
+
         with db_api.CONTEXT_WRITER.using(context):
             ret_port = super(NsxDvsV2, self).update_port(
                 context, id, port)
@@ -524,19 +513,10 @@ class NsxDvsV2(addr_pair_db.AllowedAddressPairsMixin,
             port['port'].pop('fixed_ips', None)
             ret_port.update(port['port'])
 
-            # populate port_security setting, ignoring vlan network ports.
+            # populate port_security setting
-            network_type = self._dvs_get_network_type(context,
+            if psec.PORTSECURITY not in port['port']:
-                                                      ret_port['network_id'])
-            if (psec.PORTSECURITY not in port['port'] and network_type !=
-                    'vlan'):
                 ret_port[psec.PORTSECURITY] = self._get_port_security_binding(
                     context, id)
-            elif (network_type == 'vlan' and
-                  psec.PORTSECURITY in port['port'] and
-                  port['port'][psec.PORTSECURITY]):
-                # Not allowed to enable port security on vlan DVS ports
-                err_msg = _("Cannot enable port security on port %s") % id
-                raise n_exc.InvalidInput(error_message=err_msg)
 
             # validate port security and allowed address pairs
             if not ret_port[psec.PORTSECURITY]:
@@ -557,11 +537,8 @@ class NsxDvsV2(addr_pair_db.AllowedAddressPairsMixin,
                     context, ret_port, ret_port[addr_apidef.ADDRESS_PAIRS])
 
             if psec.PORTSECURITY in port['port']:
-                if network_type != 'vlan':
+                self._process_port_port_security_update(
-                    self._process_port_port_security_update(
+                    context, port['port'], ret_port)
-                        context, port['port'], ret_port)
-                else:
-                    ret_port[psec.PORTSECURITY] = False
             self._process_vnic_type(context, port['port'], id)
             LOG.debug("Updating port: %s", port)
             self._extension_manager.process_update_port(

vmware_nsx/tests/unit/dvs/test_plugin.py

@@ -292,47 +292,6 @@ class NeutronSimpleDvsTest(NeutronSimpleDvsTestCase):
                 port_status = port['port']['status']
                 self.assertEqual(port_status, 'ACTIVE')
 
-    def test_create_dvs_port_vlan_no_port_security(self):
-        params = {'provider:network_type': 'vlan',
-                  'provider:physical_network': 'dvs',
-                  'provider:segmentation_id': 7}
-        params['arg_list'] = tuple(params.keys())
-        with mock.patch.object(self._plugin._dvs, 'add_port_group'),\
-                mock.patch.object(self._plugin._dvs, 'delete_port_group'),\
-                mock.patch.object(dvs.DvsManager, 'get_dvs_moref_by_name'),\
-                mock.patch.object(dvs.DvsManager, 'add_port_group'),\
-                mock.patch.object(dvs.DvsManager, 'delete_port_group'):
-            with self.network(**params) as network,\
-                    self.subnet(network) as subnet,\
-                    self.port(subnet) as port:
-                self.assertEqual('dvs',
-                                 port['port'][portbindings.VIF_TYPE])
-                port_security = port['port']['port_security_enabled']
-                security_groups = port['port']['security_groups']
-                self.assertEqual(port_security, False)
-                self.assertEqual(security_groups, [])
-
-    def test_update_dvs_port_vlan_no_port_security(self):
-        params = {'provider:network_type': 'vlan',
-                  'provider:physical_network': 'dvs',
-                  'provider:segmentation_id': 7}
-        params['arg_list'] = tuple(params.keys())
-        with mock.patch.object(self._plugin._dvs, 'add_port_group'),\
-                mock.patch.object(self._plugin._dvs, 'delete_port_group'),\
-                mock.patch.object(dvs.DvsManager, 'get_dvs_moref_by_name'),\
-                mock.patch.object(dvs.DvsManager, 'add_port_group'),\
-                mock.patch.object(dvs.DvsManager, 'delete_port_group'):
-            with self.network(**params) as network,\
-                    self.subnet(network) as subnet,\
-                    self.port(subnet) as port:
-                self.assertEqual('dvs',
-                                 port['port'][portbindings.VIF_TYPE])
-                data = {'port': {'port_security_enabled': True}}
-                req = self.new_update_request('ports',
-                                              data, port['port']['id'])
-                res = self.deserialize('json', req.get_response(self.api))
-                self.assertIn('NeutronError', res)
-
     def test_create_router_only_dvs_backend(self):
         data = {'router': {'tenant_id': 'whatever'}}
         data['router']['name'] = 'router1'