[Smartswitch] Update control plane ACL for smartswitch

scripts/caclmgrd

@@ -88,6 +88,7 @@ class ControlPlaneAclManager(logger.Logger):
     BFD_SESSION_TABLE = "BFD_SESSION_TABLE"
     VXLAN_TUNNEL_TABLE = "VXLAN_TUNNEL"
     DPU_TABLE = "DPU"
+    MID_PLANE_BRIDGE_TABLE = "MID_PLANE_BRIDGE"
 
     # 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"
@@ -328,6 +329,29 @@ class ControlPlaneAclManager(logger.Logger):
 
         return midplane_dev_name, midplane_ip
 
+    def get_midplane_bridge_ip_from_configdb(self, namespace=DEFAULT_NAMESPACE):
+        """
+        Method to get the IP prefix from ConfigDB using the path:
+        MID_PLANE_BRIDGE/GLOBAL/ip_prefix
+
+        Usage example:
+            caclmgr = ControlPlaneAclManager("test")
+            ip_prefix = caclmgr.get_midplane_bridge_ip_from_configdb()
+            if ip_prefix:
+                print("Midplane bridge IP prefix:", ip_prefix)
+        """
+        try:
+            config_db_connector = self.config_db_map[namespace]
+            midplane_bridge_table = config_db_connector.get_table(self.MID_PLANE_BRIDGE_TABLE)
+            if midplane_bridge_table and "GLOBAL" in midplane_bridge_table:
+                global_config = midplane_bridge_table["GLOBAL"]
+                if "ip_prefix" in global_config:
+                    self.log_info("Retrieved midplane bridge IP prefix from ConfigDB: {}".format(global_config["ip_prefix"]))
+                    return global_config["ip_prefix"].split("/")[0]
+        except Exception as e:
+            self.log_error("Failed to get midplane bridge IP from ConfigDB: {}".format(str(e)))
+        return "169.254.200.254"
+
     def generate_allow_internal_chasis_midplane_traffic(self, namespace):
         allow_internal_chassis_midplane_traffic = []
         if device_info.is_chassis() and not namespace:
@@ -337,6 +361,11 @@ class ControlPlaneAclManager(logger.Logger):
             allow_internal_chassis_midplane_traffic.append(['iptables', '-A', 'INPUT', '-s', chassis_midplane_ip, '-d', chassis_midplane_ip, '-j', 'ACCEPT'])
             allow_internal_chassis_midplane_traffic.append(['iptables', '-A', 'INPUT', '-i', chassis_midplane_dev_name, '-j', 'ACCEPT'])
 
+        if device_info.is_smartswitch():
+            # Allow traffic to the chassis midplane IP
+            midplane_bridge_ip = self.get_midplane_bridge_ip_from_configdb()
+            allow_internal_chassis_midplane_traffic.append(['iptables', '-A', 'INPUT', '-d', midplane_bridge_ip, '-j', 'ACCEPT'])
+
         return allow_internal_chassis_midplane_traffic
 
     def generate_allow_internal_docker_ip_traffic_commands(self, namespace):

tests/caclmgrd/caclmgrd_chassis_midplane_test.py

@@ -34,10 +34,62 @@ def test_caclmgrd_chassis_midplane(self, test_name, test_data, fs):
         if not os.path.exists(DBCONFIG_PATH):
             fs.create_file(DBCONFIG_PATH) # fake database_config.json
 
-        with mock.patch("sonic_py_common.device_info.is_chassis", return_value=True):
-            with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe", side_effect=["eth1-midplane", "1.0.0.33", "eth1-midplane", "1.0.0.33"]):
-                    caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
-                    ret = caclmgrd_daemon.generate_allow_internal_chasis_midplane_traffic('')
-                    self.assertListEqual(test_data["return"], ret)
-                    ret = caclmgrd_daemon.generate_allow_internal_chasis_midplane_traffic('asic0')
-                    self.assertListEqual([], ret)
+        mock_is_chassis = mock.MagicMock(return_value=True)
+        mock_is_smartswitch = mock.MagicMock(return_value=False)
+
+        with mock.patch("sonic_py_common.device_info.is_chassis", mock_is_chassis):
+            with mock.patch("sonic_py_common.device_info.is_smartswitch", mock_is_smartswitch):
+                with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe", side_effect=["eth1-midplane", "1.0.0.33", "eth1-midplane", "1.0.0.33"]):
+                        caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+                        ret = caclmgrd_daemon.generate_allow_internal_chasis_midplane_traffic('')
+                        self.assertListEqual(test_data["return"], ret)
+                        ret = caclmgrd_daemon.generate_allow_internal_chasis_midplane_traffic('asic0')
+                        self.assertListEqual([], ret)
+                        mock_is_chassis.return_value = False
+                        mock_is_smartswitch.return_value = True
+                        # Mock the get_midplane_bridge_ip_from_configdb method for smartswitch test
+                        with mock.patch.object(caclmgrd_daemon, 'get_midplane_bridge_ip_from_configdb', return_value="169.254.200.254"):
+                            ret = caclmgrd_daemon.generate_allow_internal_chasis_midplane_traffic('')
+                            self.assertListEqual(test_data["return_smartswitch"], ret)
+
+    def test_get_midplane_bridge_ip_from_configdb(self):
+        """Test the get_midplane_bridge_ip_from_configdb method"""
+        # Mock ConfigDB data with required tables
+        mock_config_db = {
+            "DEVICE_METADATA": {
+                "localhost": {
+                    "subtype": "ToR"
+                }
+            },
+            "FEATURE": {
+                "feature1": {
+                    "state": "enabled"
+                }
+            },
+            "MID_PLANE_BRIDGE": {
+                "GLOBAL": {
+                    "ip_prefix": "169.254.200.254/24"
+                }
+            }
+        }
+
+        # Set up the mock ConfigDB
+        MockConfigDb.set_config_db(mock_config_db)
+
+        caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+
+        # Test with valid ConfigDB data
+        ip_address = caclmgrd_daemon.get_midplane_bridge_ip_from_configdb()
+        self.assertEqual(ip_address, "169.254.200.254")
+
+        # Test with different IP prefix format
+        mock_config_db["MID_PLANE_BRIDGE"]["GLOBAL"]["ip_prefix"] = "10.0.0.1/16"
+        MockConfigDb.mod_config_db(mock_config_db)
+        ip_address = caclmgrd_daemon.get_midplane_bridge_ip_from_configdb()
+        self.assertEqual(ip_address, "10.0.0.1")
+
+        # Test with missing ConfigDB data (should return default)
+        mock_config_db["MID_PLANE_BRIDGE"]["GLOBAL"] = {}
+        MockConfigDb.mod_config_db(mock_config_db)
+        ip_address = caclmgrd_daemon.get_midplane_bridge_ip_from_configdb()
+        self.assertEqual(ip_address, "169.254.200.254")

tests/caclmgrd/test_chassis_midplane_vectors.py

@@ -10,6 +10,9 @@
             "return": [
                 ['iptables', '-A', 'INPUT', '-s', '1.0.0.33', '-d', '1.0.0.33', '-j', 'ACCEPT'],
                 ['iptables', '-A', 'INPUT', '-i', 'eth1-midplane', '-j', 'ACCEPT']
+            ],
+            "return_smartswitch": [
+                ['iptables', '-A', 'INPUT', '-d', '169.254.200.254','-j', 'ACCEPT']
             ]
         }
     ]