Add iptables rules to drop all bgp packets destined for loopback1 IP addresses in dual ToR

scripts/caclmgrd

@@ -361,6 +361,28 @@ class ControlPlaneAclManager(logger.Logger):
 
         return allow_internal_docker_ip_cmds
 
+    def generate_block_bgp_loopback1(self, namespace, config_db_connector):
+
+        drop_dulator_bgp_loopback1_cmds = []
+        if self.DualToR:
+            loopback_table = config_db_connector.get_table(self.LOOPBACK_TABLE)
+            loopback_name1 = 'Loopback1'
+            loopback_networks1 = get_ipv4_networks_from_interface_table(loopback_table, loopback_name1)
+            if len(loopback_networks1) == 0:
+                self.log_warning("Loopback 1 IP not available from DualToR config")
+                return drop_dulator_bgp_loopback1_cmds
+
+            loopback_address_vals1 = list(loopback_networks1.values())
+            if not isinstance(loopback_address_vals1[0], ipaddress.IPv4Address):
+                self.log_warning("Loopback 1 IP Network not available from DualToR config")
+                return drop_dulator_bgp_loopback1_cmds
+
+            # Add iptables rules to drop all packets destined for loopback1 IP addresses
+            loopback_address1 = loopback_address_vals1[0]
+            drop_dulator_bgp_loopback1_cmds.append(self.iptables_cmd_ns_prefix[namespace] +
+                    ['iptables', '-A', 'INPUT', '-d', str(loopback_address1), '-p', 'tcp', '--dport', '179', '-j', 'DROP'])
+
+        return drop_dulator_bgp_loopback1_cmds
 
     def generate_fwd_traffic_from_host_to_soc(self, namespace, config_db_connector):
 
@@ -861,6 +883,7 @@ class ControlPlaneAclManager(logger.Logger):
 
         if self.DualToR:
             dualtor_iptables_cmds = self.generate_fwd_traffic_from_host_to_soc(namespace, config_db_connector)
+            dualtor_iptables_cmds += self.generate_block_bgp_loopback1(namespace, config_db_connector)
             for cmd in dualtor_iptables_cmds:
                 self.log_info("  " + ' '.join(cmd))
             self.run_commands(dualtor_iptables_cmds)

tests/caclmgrd/caclmgrd_bgp_loopback1_test.py

@@ -0,0 +1,50 @@
+import os
+import sys
+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_bgp_loopback1_vectors import BGP_LOOPBACK1_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 TestCaclmgrdSoc(TestCase):
+    """
+        Test caclmgrd soc
+    """
+    def setUp(self):
+        swsscommon.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(BGP_LOOPBACK1_TEST_VECTOR)
+    @patchfs
+    def test_caclmgrd_soc(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")
+                caclmgrd_daemon.update_control_plane_nat_acls('', {}, MockConfigDb())
+                mocked_subprocess.Popen.assert_has_calls(test_data["expected_subprocess_calls"], any_order=True)

tests/caclmgrd/test_bgp_loopback1_vectors.py

@@ -0,0 +1,46 @@
+from unittest.mock import call
+import subprocess
+
+"""
+    caclmgrd soc test vector
+"""
+BGP_LOOPBACK1_TEST_VECTOR = [
+    [
+        "BGP_LOOPBACK1_SESSION_TEST",
+        {
+            "config_db": {
+                "DEVICE_METADATA": {
+                    "localhost": {
+                        "subtype": "DualToR",
+                        "type": "ToRRouter",
+                    }
+                },
+                "MUX_CABLE": {
+                    "Ethernet4": {
+                        "cable_type": "active-active",
+                        "soc_ipv4": "10.10.10.7/32",
+                    }
+                },
+                "VLAN_INTERFACE": {
+                    "Vlan1000|10.10.10.3/24": {
+                        "NULL": "NULL",
+                    }
+                },
+                "LOOPBACK_INTERFACE": {
+                    "Loopback1|10.10.10.34/32": {
+                        "NULL": "NULL",
+                    }
+                },
+                "FEATURE": {
+                },
+            },
+            "expected_subprocess_calls": [
+                call(['iptables', '-A', 'INPUT', '-d', "10.10.10.34", '-p', 'tcp', '--dport', '179', '-j', 'DROP'], universal_newlines=True, stdout=-1)
+            ],
+            "popen_attributes": {
+                'communicate.return_value': ('output', 'error'),
+            },
+            "call_rc": 0,
+        }
+    ]
+]