Merge "NSX|P: Fix provider security groups"

vmware_nsx/plugins/nsx_p/plugin.py

@@ -728,8 +728,10 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
         return network_id
 
     def _build_port_tags(self, port_data):
-        sec_groups = port_data.get(ext_sg.SECURITYGROUPS, [])
-        sec_groups += port_data.get(provider_sg.PROVIDER_SECURITYGROUPS, [])
+        sec_groups = []
+        sec_groups.extend(port_data.get(ext_sg.SECURITYGROUPS, []))
+        sec_groups.extend(port_data.get(provider_sg.PROVIDER_SECURITYGROUPS,
+                                        []))
 
         tags = []
         for sg in sec_groups:
@@ -1042,7 +1044,6 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
 
             (port_security, has_ip) = self._determine_port_security_and_has_ip(
                 context, updated_port)
-            self._remove_provider_security_groups_from_list(updated_port)
             self._process_portbindings_create_and_update(
                 context, port_data, updated_port,
                 vif_type=self._vif_type_by_vnic_type(direct_vnic_type))
@@ -1057,6 +1058,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
                     raise n_exc.InvalidInput(error_message=msg)
                 self._update_mac_learning_state(context, port_id,
                                                 mac_learning_state)
+            self._remove_provider_security_groups_from_list(updated_port)
 
         # Update the QoS policy
         qos_policy_id = self._get_port_qos_policy_id(
@@ -1108,7 +1110,6 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
             'mac_address_updated': False,
             'original_port': original_port,
         }
-
         registry.notify(resources.PORT, events.AFTER_UPDATE, self, **kwargs)
         return updated_port
 
@@ -1986,7 +1987,8 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
         return '%s_local_group' % sg_rule['id']
 
     def _create_security_group_backend_rule(self, context, domain_id, map_id,
-                                            sg_rule, secgroup_logging):
+                                            sg_rule, secgroup_logging,
+                                            is_provider_sg=False):
         # The id of the map and group is the same as the security group id
         this_group_id = map_id
         # There is no rule name in neutron. Using ID instead
@@ -2037,12 +2039,14 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
         logging = (cfg.CONF.nsx_p.log_security_groups_allowed_traffic or
                    secgroup_logging)
         scope = [self.nsxpolicy.group.get_path(domain_id, this_group_id)]
+        action = (policy_constants.ACTION_DENY if is_provider_sg
+                  else policy_constants.ACTION_ALLOW)
         self.nsxpolicy.comm_map.create_entry(
             nsx_name, domain_id, map_id, entry_id=sg_rule['id'],
             description=sg_rule.get('description'),
             service_ids=[service] if service else None,
             ip_protocol=ip_protocol,
-            action=policy_constants.ACTION_ALLOW,
+            action=action,
             source_groups=[source] if source else None,
             dest_groups=[destination] if destination else None,
             scope=scope,
@@ -2205,11 +2209,13 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
                     context, rules_db[i], r['security_group_rule'])
 
         domain_id = sg['tenant_id']
+        is_provider_sg = sg.get(provider_sg.PROVIDER)
         secgroup_logging = self._is_security_group_logged(context, sg_id)
         for rule_data in rules_db:
             # create the NSX backend rule
             self._create_security_group_backend_rule(
-                context, domain_id, sg_id, rule_data, secgroup_logging)
+                context, domain_id, sg_id, rule_data, secgroup_logging,
+                is_provider_sg=is_provider_sg)
 
         return rules_db
 

vmware_nsx/tests/unit/extensions/test_provider_security_groups.py

@@ -28,6 +28,8 @@ from vmware_nsx.extensions import providersecuritygroup as provider_sg
 from vmware_nsx.tests.unit.nsx_p import test_plugin as test_nsxp_plugin
 from vmware_nsx.tests.unit.nsx_v import test_plugin as test_nsxv_plugin
 from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_nsxv3_plugin
+from vmware_nsxlib.v3 import nsx_constants
+from vmware_nsxlib.v3.policy import constants as policy_constants
 
 
 PLUGIN_NAME = ('vmware_nsx.tests.unit.extensions.'
@@ -397,33 +399,27 @@ class TestNSXvProviderSecurityGroup(test_nsxv_plugin.NsxVPluginV2TestCase,
 class TestNSXpProviderSecurityGrp(test_nsxp_plugin.NsxPPluginTestCaseMixin,
                                   ProviderSecurityGroupExtTestCase):
 
-    # Temporarily skip all port related tests until the plugin supports it
-    def test_update_port_security_groups(self):
-        self.skipTest('Temporarily not supported')
+    def test_create_provider_security_group_rule(self):
+        provider_secgroup = self._create_provider_security_group()
+        sg_id = provider_secgroup['security_group']['id']
 
-    def test_update_port_remove_provider_sg_with_empty_list(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_update_port_security_groups_only(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_create_port_with_no_provider_sg(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_create_port_gets_multi_provider_sg(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_cannot_update_port_with_provider_group_as_sec_group(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_update_port_remove_provider_sg_with_none(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_create_port_gets_provider_sg(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_cannot_update_port_with_different_tenant_provider_secgroup(self):
-        self.skipTest('Temporarily not supported')
-
-    def test_cannot_update_port_with_sec_group_as_provider(self):
-        self.skipTest('Temporarily not supported')
+        with mock.patch("vmware_nsxlib.v3.policy.core_resources."
+                        "NsxPolicyCommunicationMapApi.create_entry"
+                        ) as entry_create:
+            with self.security_group_rule(security_group_id=sg_id) as rule:
+                rule_data = rule['security_group_rule']
+                rule_id = rule_data['id']
+                project_id = rule_data['project_id']
+                scope = [self.plugin.nsxpolicy.group.get_path(
+                    project_id, sg_id)]
+                entry_create.assert_called_once_with(
+                    rule_id, project_id, sg_id, entry_id=rule_id,
+                    description='',
+                    direction=nsx_constants.IN,
+                    ip_protocol=nsx_constants.IPV4,
+                    action=policy_constants.ACTION_DENY,
+                    service_ids=mock.ANY,
+                    source_groups=mock.ANY,
+                    dest_groups=mock.ANY,
+                    scope=scope,
+                    logged=False)

vmware_nsx/tests/unit/nsx_p/test_plugin.py

@@ -52,9 +52,9 @@ from vmware_nsx.tests import unit as vmware
 from vmware_nsx.tests.unit.common_plugin import common_v3
 from vmware_nsxlib.v3 import exceptions as nsxlib_exc
 from vmware_nsxlib.v3 import nsx_constants
+from vmware_nsxlib.v3.policy import constants as policy_constants
 from vmware_nsxlib.v3 import utils as nsxlib_utils
 
-from vmware_nsxlib.v3.policy import constants as policy_constants
 
 PLUGIN_NAME = 'vmware_nsx.plugin.NsxPolicyPlugin'
 NSX_OVERLAY_TZ_NAME = 'OVERLAY_TZ'