Modify cacl rules to allow incoming packets to dash-ha (#256)

why
swbusd in dash-ha runs grpc client and server for connections within the switch and outside the switch. We need to config CACL iptables rules if dash-ha feature is enabled.

what this PR does
if dash-ha feature is enabled, subscribe to DPU table in config_db. Configure iptables based on the swbus_port in each DPU entries (up to 8). If swbus_port is changed, remove old iptables rules and add new rules. If DPU entry is deleted, remove corresponding iptables rules.

how to verify
Add dash-ha to feature table.
"dash-ha": {
"auto_restart": "disabled",
"delayed": "False",
"has_global_scope": "False",
"has_per_asic_scope": "False",
"has_per_dpu_scope": "True",
"high_mem_alert": "disabled",
"state": "enabled",
"support_syslog_rate_limit": "true"
}

Add DPU table in config_db
"DPU": {
"dpu0": {
"dpu_id": "0",
"gnmi_port": "50051",
"local_port": "8080",
"orchagent_zmq_port": "5555",
"pa_ipv4": "18.0.202.1",
"state": "up",
"swbus_port": "23606",
"vdpu_id": "vdpu0",
"vip_ipv4": "3.2.1.0"
},
...
}

verify iptables rule is created for both ipv4 and ipv6 to tcp port 23606.

change swbus_port and verify old rules are removed and new rules are created.

Author: yue-fred-gao

scripts/caclmgrd

@@ -85,6 +85,7 @@ class ControlPlaneAclManager(logger.Logger):
 
     BFD_SESSION_TABLE = "BFD_SESSION_TABLE"
     VXLAN_TUNNEL_TABLE = "VXLAN_TUNNEL"
+    DPU_TABLE = "DPU"
 
     # To specify a port range instead of a single port, use iptables format:
     # separate start and end ports with a colon, e.g., "1000:2000"
@@ -121,7 +122,9 @@ class ControlPlaneAclManager(logger.Logger):
     bfdAllowed = False
     VxlanAllowed = False
     VxlanSrcIP = ""
-
+    # a map from dpu name to port
+    dashHaPortMap = {}
+    
     def __init__(self, log_identifier):
         super(ControlPlaneAclManager, self).__init__(log_identifier)
 
@@ -601,6 +604,10 @@ class ControlPlaneAclManager(logger.Logger):
             fvs = swsscommon.FieldValuePairs([("src_ip", self.VxlanSrcIP)])
             iptables_cmds += self.get_vxlan_port_iptable_commands(namespace, fvs)
 
+        for dpu, port in self.dashHaPortMap.items():
+            iptables_cmds += self.make_dash_ha_rules(namespace, port)
+            self.log_info(f"Enabled dash-ha port {port} for {dpu}")
+            
         # Add iptables commands to allow internal docker traffic
         iptables_cmds += self.generate_allow_internal_docker_ip_traffic_commands(namespace)
 
@@ -972,6 +979,58 @@ class ControlPlaneAclManager(logger.Logger):
         self.VxlanSrcIP = ""
         return True
 
+    def remove_dash_ha_rules(self, namespace, port):
+        iptables_cmds = []
+        # Remove iptables/ip6tables commands that allow DASH-HA packets
+        iptables_cmds.append(self.iptables_cmd_ns_prefix[namespace] + 
+                ['iptables', '-D', 'INPUT', '-p', 'tcp', '--dport', str(port), '-j', 'ACCEPT'])
+        iptables_cmds.append(self.iptables_cmd_ns_prefix[namespace] + 
+                ['ip6tables', '-D', 'INPUT', '-p', 'tcp', '--dport', str(port), '-j', 'ACCEPT'])
+        self.run_commands(iptables_cmds)
+    
+    def add_dash_ha_rules(self, namespace, port):
+        iptables_cmds = self.make_dash_ha_rules(namespace, port)
+        self.run_commands(iptables_cmds)
+
+    def make_dash_ha_rules(self, namespace, port):
+        iptables_cmds = []
+        # make iptables/ip6tables commands that allow DASH-HA packets
+        iptables_cmds.append(self.iptables_cmd_ns_prefix[namespace] + 
+                ['iptables', '-I', 'INPUT', '2', '-p', 'tcp', '--dport', str(port), '-j', 'ACCEPT'])
+        iptables_cmds.append(self.iptables_cmd_ns_prefix[namespace] + 
+                ['ip6tables', '-I', 'INPUT', '2', '-p', 'tcp', '--dport', str(port), '-j', 'ACCEPT'])
+        return iptables_cmds
+    
+    def update_dash_ha_rules(self, namespace, key, op, data):
+        port = self.dashHaPortMap.get(key)
+        if op == "DEL" and not port:
+            return
+                
+        if op == "DEL" and port:
+            # Remove iptables/ip6tables commands that allow DASH-HA packets
+            self.remove_dash_ha_rules(namespace, port)
+            self.dashHaPortMap.pop(key)
+            return
+        
+        if op == "SET":
+            new_port = None
+            for fv in data:
+                if (fv[0] == "swbus_port"):
+                    new_port = fv[1]
+                    break
+            if not new_port:
+                self.log_info("Received DPU configuration without swbus_port. Ignore it.")
+                return
+            if new_port == port:
+                return
+            if port:
+                # Remove iptables/ip6tables commands that allow DASH-HA packets
+                self.remove_dash_ha_rules(namespace, port)
+            # Add iptables/ip6tables commands to allow DASH-HA packets
+            self.add_dash_ha_rules(namespace, new_port)
+            self.dashHaPortMap[key] = new_port
+            return
+
     def run(self):
         # Set select timeout to 1 second
         SELECT_TIMEOUT_MS = 1000
@@ -1022,6 +1081,9 @@ class ControlPlaneAclManager(logger.Logger):
 
         subscribe_vxlan_table = swsscommon.SubscriberStateTable(config_db_connector, self.VXLAN_TUNNEL_TABLE)
         sel.addSelectable(subscribe_vxlan_table)
+
+        subscribe_dpu_table = swsscommon.SubscriberStateTable(config_db_connector, self.DPU_TABLE)
+        sel.addSelectable(subscribe_dpu_table)
         # Map of Namespace <--> susbcriber table's object
         config_db_subscriber_table_map = {}
 
@@ -1106,6 +1168,13 @@ class ControlPlaneAclManager(logger.Logger):
                     elif op == 'DEL' and self.VxlanAllowed:
                         self.block_vxlan_port(namespace)
 
+                while True:
+                    key, op, fvs = subscribe_dpu_table.pop()
+                    if not key:
+                        break
+                    if "dash-ha" in self.feature_present:
+                        self.update_dash_ha_rules(namespace, key, op, fvs)
+                
             ctrl_plane_acl_notification = set()
 
             # Pop data of both Subscriber Table object of namespace that got config db acl table event

tests/caclmgrd/caclmgrd_dash_ha_test.py

@@ -0,0 +1,67 @@
+import os
+import sys
+from swsscommon import swsscommon
+
+from parameterized import parameterized
+from sonic_py_common.general import load_module_from_source
+from unittest import TestCase, mock
+from pyfakefs.fake_filesystem_unittest import patchfs
+
+from .test_dash_ha_vectors import CACLMGRD_DASH_HA_TEST_VECTOR
+from tests.common.mock_configdb import MockConfigDb
+from unittest.mock import MagicMock, patch
+
+DBCONFIG_PATH = '/var/run/redis/sonic-db/database_config.json'
+
+class TestCaclmgrdDashHa(TestCase):
+    """
+        Test caclmgrd bfd
+    """
+    def setUp(self):
+        swsscommon.ConfigDBConnector = MockConfigDb
+        test_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        modules_path = os.path.dirname(test_path)
+        scripts_path = os.path.join(modules_path, "scripts")
+        sys.path.insert(0, modules_path)
+        caclmgrd_path = os.path.join(scripts_path, 'caclmgrd')
+        self.caclmgrd = load_module_from_source('caclmgrd', caclmgrd_path)
+
+    @parameterized.expand(CACLMGRD_DASH_HA_TEST_VECTOR)
+    @patchfs
+    def test_caclmgrd_dash_ha(self, test_name, test_data, fs):
+        if not os.path.exists(DBCONFIG_PATH):
+            fs.create_file(DBCONFIG_PATH) # fake database_config.json
+
+        MockConfigDb.set_config_db(test_data["config_db"])
+
+        with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe", return_value='sonic'):
+            with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
+                popen_mock = mock.Mock()
+                popen_attrs = test_data["popen_attributes"]
+                popen_mock.configure_mock(**popen_attrs)
+                mocked_subprocess.Popen.return_value = popen_mock
+                mocked_subprocess.PIPE = -1
+
+                call_rc = test_data["call_rc"]
+                mocked_subprocess.call.return_value = call_rc
+
+                caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+                assert(caclmgrd_daemon.feature_present["dash-ha"] == True)
+                # a set operation without swbus_port. It is no-op.
+                caclmgrd_daemon.update_dash_ha_rules('', "dpu0", "SET", ["somekey", "somevalue"])
+                
+                caclmgrd_daemon.update_dash_ha_rules('', "dpu0", "SET", test_data["input_add"])
+                mocked_subprocess.Popen.assert_has_calls(test_data["expected_add_subprocess_calls"], any_order=True)
+                
+                caclmgrd_daemon.update_dash_ha_rules('', "dpu0", "SET", test_data["input_upd"])
+                mocked_subprocess.Popen.assert_has_calls(test_data["expected_upd_subprocess_calls"], any_order=True)
+                
+                caclmgrd_daemon.update_dash_ha_rules('', "dpu0", "DEL", {})
+                mocked_subprocess.Popen.assert_has_calls(test_data["expected_del_subprocess_calls"], any_order=True)
+                
+                caclmgrd_daemon.update_dash_ha_rules('', "dpu0", "SET", test_data["input_add"])
+                mocked_subprocess.Popen.reset_mock()
+                caclmgrd_daemon.num_changes[''] = 1
+                caclmgrd_daemon.check_and_update_control_plane_acls('', 1)
+                mocked_subprocess.Popen.assert_has_calls(test_data["expected_add_subprocess_calls"], any_order=True)
+

tests/caclmgrd/test_dash_ha_vectors.py

@@ -0,0 +1,49 @@
+from unittest.mock import call
+import subprocess
+
+"""
+    caclmgrd dash-ha test vector
+"""
+CACLMGRD_DASH_HA_TEST_VECTOR = [
+    [
+        "DASH_HA_TEST",
+        {
+            "config_db": {
+                "DEVICE_METADATA": {
+                    "localhost": {
+                        "subtype": "SmartSwitch",
+                        "type": "LeafRouter",
+                    }
+                },                
+                "FEATURE": {
+                    "dash-ha": {
+                        "auto_restart": "enabled",
+                        "has_per_dpu_scope": "True",
+                        "state": "enabled",
+                    }
+                }            
+            },
+            "input_add" : [("swbus_port", "23606")],
+            "input_upd" : [("swbus_port", "23607")],
+            "expected_add_subprocess_calls": [
+                call(['iptables', '-I', 'INPUT', '2', '-p', 'tcp', '--dport', '23606', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+                call(['ip6tables', '-I', 'INPUT', '2', '-p', 'tcp', '--dport', '23606', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+            ],
+            "expected_upd_subprocess_calls": [
+                call(['iptables', '-D', 'INPUT', '-p', 'tcp', '--dport', '23606', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+                call(['ip6tables', '-D', 'INPUT', '-p', 'tcp', '--dport', '23606', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+                call(['iptables', '-I', 'INPUT', '2', '-p', 'tcp', '--dport', '23607', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+                call(['ip6tables', '-I', 'INPUT', '2', '-p', 'tcp', '--dport', '23607', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+            ],
+            "expected_del_subprocess_calls": [
+                call(['iptables', '-D', 'INPUT', '-p', 'tcp', '--dport', '23607', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+                call(['ip6tables', '-D', 'INPUT', '-p', 'tcp', '--dport', '23607', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+            ],
+            
+            "popen_attributes": {
+                'communicate.return_value': ('output', 'error'),
+            },
+            "call_rc": 0,
+        }
+    ]
+]