ovn: update ACL actions on stateful field change

When SG 'stateful' field changes its value, all 'allow-related' /
'allow-stateless' ACLs that belong to the group should flip their
action value in OVN nbdb accordingly.

Closes-Bug: #1952653
Reported-At: https://bugzilla.redhat.com/show_bug.cgi?id=2017048
Change-Id: Ic4621dfaedb83287845e3336ba846b054230f087

neutron/common/ovn/acl.py

@@ -261,12 +261,15 @@ def _acl_columns_name_severity_supported(nb_idl):
     return ('name' in columns) and ('severity' in columns)
 
 
+def is_sg_stateful(sg, stateless_supported):
+    if stateless_supported:
+        return sg.get("stateful", True)
+    return True
+
+
 def add_acls_for_sg_port_group(ovn, security_group, txn,
                                stateless_supported=True):
-    if stateless_supported:
-        stateful = security_group.get("stateful", True)
-    else:
-        stateful = True
+    stateful = is_sg_stateful(security_group, stateless_supported)
     for r in security_group['security_group_rules']:
         acl = _add_sg_rule_acl_for_port_group(
             utils.ovn_port_group_name(security_group['id']), stateful, r)
@@ -288,11 +291,8 @@ def update_acls_for_security_group(plugin,
     # Check if ACL log name and severity supported or not
     keep_name_severity = _acl_columns_name_severity_supported(ovn)
 
-    if stateless_supported:
-        sg = plugin.get_security_group(admin_context, security_group_id)
-        stateful = sg.get("stateful", True)
-    else:
-        stateful = True
+    sg = plugin.get_security_group(admin_context, security_group_id)
+    stateful = is_sg_stateful(sg, stateless_supported)
 
     acl = _add_sg_rule_acl_for_port_group(
         utils.ovn_port_group_name(security_group_id),

neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py

@@ -418,11 +418,23 @@ class OVNMechanismDriver(api.MechanismDriver):
                                                security_group_id)
 
     def _update_security_group(self, resource, event, trigger, payload):
-        # OVN doesn't care about updates to security groups, only if they
-        # exist or not. We are bumping the revision number here so it
-        # doesn't show as inconsistent to the maintenance periodic task
         context = payload.context
         security_group = payload.latest_state
+
+        old_state, new_state = payload.states
+        is_allow_stateless_supported = (
+            self._ovn_client.is_allow_stateless_supported()
+        )
+        old_stateful = ovn_acl.is_sg_stateful(
+            old_state, is_allow_stateless_supported)
+        new_stateful = ovn_acl.is_sg_stateful(
+            new_state, is_allow_stateless_supported)
+        if old_stateful != new_stateful:
+            for rule in self._plugin.get_security_group_rules(
+                    context, {'security_group_id': [security_group['id']]}):
+                self._ovn_client.delete_security_group_rule(context, rule)
+                self._ovn_client.create_security_group_rule(context, rule)
+
         ovn_revision_numbers_db.bump_revision(
             context, security_group, ovn_const.TYPE_SECURITY_GROUPS)
 

neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py

@@ -673,6 +673,61 @@ class TestCreateDefaultDropPortGroup(base.BaseLoggingTestCase,
         self._test_pg_with_ports(expected_ports=[])
 
 
+class TestSecurityGroup(base.TestOVNFunctionalBase):
+
+    def setUp(self):
+        super(TestSecurityGroup, self).setUp()
+        self._ovn_client = self.mech_driver._ovn_client
+        self.plugin = self.mech_driver._plugin
+
+    def _find_acls_for_sg(self, sg_id):
+        rows = self.nb_api.db_find_rows('ACL').execute(check_error=True)
+        if rows:
+            rule_ids = {
+                r['id'] for r in self.plugin.get_security_group_rules(
+                    self.context, {'security_group_id': [sg_id]})
+            }
+
+            def get_api_id(r):
+                return r.external_ids.get(
+                    ovn_const.OVN_SG_RULE_EXT_ID_KEY, '')
+
+            return [r for r in rows if get_api_id(r) in rule_ids]
+        return []
+
+    def test_sg_stateful_toggle_updates_ovn_acls(self):
+        def check_acl_actions(sg_id, expected):
+            self.assertEqual(
+                {expected},
+                set(a.action for a in self._find_acls_for_sg(sg_id))
+            )
+
+        sg_data = {
+            'name': 'testsg',
+            'description': 'Test Security Group',
+            'tenant_id': self._tenant_id,
+            'is_default': True,
+        }
+        sg = self.plugin.create_security_group(
+            self.context, security_group={'security_group': sg_data})
+        check_acl_actions(sg['id'], 'allow-related')
+
+        def update_sg(stateful):
+            sg_data['stateful'] = stateful
+            self.plugin.update_security_group(
+                self.context, sg['id'],
+                security_group={'security_group': sg_data})
+
+        update_sg(False)
+        check_acl_actions(sg['id'], 'allow-stateless')
+
+        update_sg(True)
+        check_acl_actions(sg['id'], 'allow-related')
+
+        update_sg(False)
+        check_acl_actions(sg['id'], 'allow-stateless')
+
+
 class TestProvnetPorts(base.TestOVNFunctionalBase):
 
     def setUp(self):