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

scripts/caclmgrd

@@ -361,6 +361,33 @@ 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)
+            loopback1_name = 'Loopback1'
+
+            if loopback_table:
+                for key, _ in loopback_table.items():
+                    if not _ip_prefix_in_key(key):
+                        continue
+
+                    iface_name, iface_cidr = key
+                    if iface_name.startswith(loopback1_name):
+                        loopback1_intf = ipaddress.ip_interface(iface_cidr)
+                        loopback1_addr = loopback1_intf.ip
+                        # Add iptables rules to drop all packets destined for loopback1 IP addresses
+                        if isinstance(loopback1_addr, ipaddress.IPv4Address):
+                            drop_dulator_bgp_loopback1_cmds.append(self.iptables_cmd_ns_prefix[namespace] +
+                                                                   ['iptables', '-I', 'INPUT', '1', '-d', str(loopback1_addr), '-p', 'tcp', '--dport', '179', '-j', 'DROP'])
+                        elif isinstance(loopback1_addr, ipaddress.IPv6Address):
+                            drop_dulator_bgp_loopback1_cmds.append(self.iptables_cmd_ns_prefix[namespace] +
+                                                                   ['ip6tables', '-I', 'INPUT', '1', '-d', str(loopback1_addr), '-p', 'tcp', '--dport', '179', '-j', 'DROP'])
+                        else:
+                            self.log_warning("Unrecognized Loopback 1 IP {}".format(loopback1_addr))
+
+        return drop_dulator_bgp_loopback1_cmds
 
     def generate_fwd_traffic_from_host_to_soc(self, namespace, config_db_connector):
 
@@ -861,6 +888,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 TestCaclmgrdLoopback1Drop(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,50 @@
+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.1.0.10/32": {},
+                    "Loopback1|10.1.0.12/32": {},
+                    "Loopback1|FC00:1:0:10::/128": {},
+                    "Loopback1|FC00:1:0:34::/128": {}
+                },
+                "FEATURE": {
+                },
+            },
+            "expected_subprocess_calls": [
+                call(['iptables', '-I', 'INPUT', '1', '-d', "10.1.0.10", '-p', 'tcp', '--dport', '179', '-j', 'DROP'], universal_newlines=True, stdout=-1),
+                call(['iptables', '-I', 'INPUT', '1', '-d', "10.1.0.12", '-p', 'tcp', '--dport', '179', '-j', 'DROP'], universal_newlines=True, stdout=-1),
+                call(['ip6tables', '-I', 'INPUT', '1', '-d', "fc00:1:0:10::", '-p', 'tcp', '--dport', '179', '-j', 'DROP'], universal_newlines=True, stdout=-1),
+                call(['ip6tables', '-I', 'INPUT', '1', '-d', "fc00:1:0:34::", '-p', 'tcp', '--dport', '179', '-j', 'DROP'], universal_newlines=True, stdout=-1)
+            ],
+            "popen_attributes": {
+                'communicate.return_value': ('output', 'error'),
+            },
+            "call_rc": 0,
+        }
+    ]
+]