Merge pull request sonic-net#545 from mssonicbld/cherry/msft-202412/21468

[action] [PR:21468] [Bgpcfgd] Update bgpcfgd to handle SRV6_MY_SIDS table's key as an ipv6 prefix

Author: r12f

src/sonic-bgpcfgd/bgpcfgd/managers_srv6.py

@@ -54,14 +54,18 @@ def locators_set_handler(self, key, data):
 
     def sids_set_handler(self, key, data):
         locator_name = key.split("|")[0]
-        ip_addr = key.split("|")[1].lower()
-        key = "{}|{}".format(locator_name, ip_addr)
+        ip_prefix = key.split("|")[1].lower()
+        key = "{}|{}".format(locator_name, ip_prefix)
+        prefix_len = int(ip_prefix.split("/")[1])
 
         if not self.directory.path_exist(self.db_name, "SRV6_MY_LOCATORS", locator_name):
             log_err("Found a SRv6 SID config entry with a locator that does not exist: {} | {}".format(key, data))
             return False
 
         locator = self.directory.get(self.db_name, "SRV6_MY_LOCATORS", locator_name)
+        if locator.block_len + locator.node_len > prefix_len:
+            log_err("Found a SRv6 SID config entry with an invalid prefix length {} | {}".format(key, data))
+            return False
 
         if 'action' not in data:
             log_err("Found a SRv6 SID config entry that does not specify action: {} | {}".format(key, data))
@@ -71,18 +75,18 @@ def sids_set_handler(self, key, data):
             log_err("Found a SRv6 SID config entry associated with unsupported action: {} | {}".format(key, data))
             return False
 
-        sid = SID(locator_name, ip_addr, data) # the information in data will be parsed into SID's attributes
+        sid = SID(locator_name, ip_prefix, data) # the information in data will be parsed into SID's attributes
 
         cmd_list = ['segment-routing', 'srv6', 'static-sids']
-        sid_cmd = 'sid {}/{} locator {} behavior {}'.format(ip_addr, locator.block_len + locator.node_len + locator.func_len, locator_name, sid.action)
+        sid_cmd = 'sid {} locator {} behavior {}'.format(ip_prefix, locator_name, sid.action)
         if sid.decap_vrf != DEFAULT_VRF:
             sid_cmd += ' vrf {}'.format(sid.decap_vrf)
         cmd_list.append(sid_cmd)
 
         self.cfg_mgr.push_list(cmd_list)
         log_debug("{} SRv6 static configuration {}|{} is scheduled for updates. {}".format(self.db_name, self.table_name, key, str(cmd_list)))
 
-        self.directory.put(self.db_name, self.table_name, key, (sid, sid_cmd))
+        self.directory.put(self.db_name, self.table_name, key.replace("/", "\\"), (sid, sid_cmd))
         return True
 
     def del_handler(self, key):
@@ -101,21 +105,21 @@ def locators_del_handler(self, key):
 
     def sids_del_handler(self, key):
         locator_name = key.split("|")[0]
-        ip_addr = key.split("|")[1].lower()
-        key = "{}|{}".format(locator_name, ip_addr)
+        ip_prefix = key.split("|")[1].lower()
+        key = "{}|{}".format(locator_name, ip_prefix)
 
-        if not self.directory.path_exist(self.db_name, self.table_name, key):
+        if not self.directory.path_exist(self.db_name, self.table_name, key.replace("/", "\\")):
             log_warn("Encountered a config deletion with a SRv6 SID that does not exist: {}".format(key))
             return
 
-        _, sid_cmd = self.directory.get(self.db_name, self.table_name, key)
+        _, sid_cmd = self.directory.get(self.db_name, self.table_name, key.replace("/", "\\"))
         cmd_list = ['segment-routing', 'srv6', "static-sids"]
         no_sid_cmd = 'no ' + sid_cmd
         cmd_list.append(no_sid_cmd)
 
         self.cfg_mgr.push_list(cmd_list)
         log_debug("{} SRv6 static configuration {}|{} is scheduled for updates. {}".format(self.db_name, self.table_name, key, str(cmd_list)))
-        self.directory.remove(self.db_name, self.table_name, key)
+        self.directory.remove(self.db_name, self.table_name, key.replace("/", "\\"))
 
 class Locator:
     def __init__(self, name, data):
@@ -127,9 +131,9 @@ def __init__(self, name, data):
         self.prefix = data['prefix'].lower() + "/{}".format(self.block_len + self.node_len)
 
 class SID:
-    def __init__(self, locator, ip_addr, data):
+    def __init__(self, locator, ip_prefix, data):
         self.locator_name = locator
-        self.ip_addr = ip_addr
+        self.ip_prefix = ip_prefix
 
         self.action = data['action']
         self.decap_vrf = data['decap_vrf'] if 'decap_vrf' in data else DEFAULT_VRF

src/sonic-bgpcfgd/tests/test_srv6.py

@@ -78,22 +78,23 @@ def test_uN_add():
     loc_mgr, sid_mgr = constructor()
     assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
 
-    op_test(sid_mgr, 'SET', ("loc1|FCBB:BBBB:1:F1::", {
+    op_test(sid_mgr, 'SET', ("loc1|FCBB:BBBB:1::/48", {
         'action': 'uN'
     }), expected_ret=True, expected_cmds=[
         'segment-routing',
         'srv6',
         'static-sids',
-        'sid fcbb:bbbb:1:f1::/64 locator loc1 behavior uN'
+        'sid fcbb:bbbb:1::/48 locator loc1 behavior uN'
     ])
 
-    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f1::")
+    print(loc_mgr.directory.data)
+    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1::\\48")
 
 def test_uDT46_add_vrf1():
     loc_mgr, sid_mgr = constructor()
     assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
 
-    op_test(sid_mgr, 'SET', ("loc1|FCBB:BBBB:1:F2::", {
+    op_test(sid_mgr, 'SET', ("loc1|FCBB:BBBB:1:F2::/64", {
         'action': 'uDT46',
         'decap_vrf': 'Vrf1'
     }), expected_ret=True, expected_cmds=[
@@ -103,61 +104,62 @@ def test_uDT46_add_vrf1():
         'sid fcbb:bbbb:1:f2::/64 locator loc1 behavior uDT46 vrf Vrf1'
     ])
 
-    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f2::")
+    print(loc_mgr.directory.data)
+    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f2::\\64")
 
 def test_uN_del():
     loc_mgr, sid_mgr = constructor()
     assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
 
     # add uN function first
-    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1:F1::", {
+    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1::/48", {
         'action': 'uN'
     })
 
     # test the deletion
-    op_test(sid_mgr, 'DEL', ("loc1|FCBB:BBBB:1:F1::",),
+    op_test(sid_mgr, 'DEL', ("loc1|FCBB:BBBB:1::/48",),
             expected_ret=True, expected_cmds=[
             'segment-routing',
             'srv6',
             'static-sids',
-            'no sid fcbb:bbbb:1:f1::/64 locator loc1 behavior uN'
+            'no sid fcbb:bbbb:1::/48 locator loc1 behavior uN'
     ])
 
-    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f1::")
+    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1::\\48")
 
 def test_uDT46_del_vrf1():
     loc_mgr, sid_mgr = constructor()
     assert loc_mgr.set_handler("loc1", {'prefix': 'fcbb:bbbb:1::'})
 
     # add a uN action first to make the uDT46 action not the last function
-    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1:F1::", {
+    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1::/48", {
         'action': 'uN'
     })
 
     # add the uDT46 action
-    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1:F2::", {
+    assert sid_mgr.set_handler("loc1|FCBB:BBBB:1:F2::/64", {
         'action': 'uDT46',
         "decap_vrf": "Vrf1"
     })
 
     # test the deletion of uDT46
-    op_test(sid_mgr, 'DEL', ("loc1|FCBB:BBBB:1:F2::",),
+    op_test(sid_mgr, 'DEL', ("loc1|FCBB:BBBB:1:F2::/64",),
             expected_ret=True, expected_cmds=[
             'segment-routing',
             'srv6',
             'static-sids',
             'no sid fcbb:bbbb:1:f2::/64 locator loc1 behavior uDT46 vrf Vrf1'
     ])
 
-    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f1::")
-    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f2::")
+    assert sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1::\\48")
+    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f2::\\64")
 
 def test_invalid_add():
     _, sid_mgr = constructor()
 
     # test the addition of a SID with a non-existent locator
-    op_test(sid_mgr, 'SET', ("loc2|FCBB:BBBB:21:F1::", {
+    op_test(sid_mgr, 'SET', ("loc2|FCBB:BBBB:21:F1::/64", {
         'action': 'uN'
     }), expected_ret=False, expected_cmds=[])
 
-    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc2|fcbb:bbbb:21:f1::")
+    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc2|fcbb:bbbb:21:f1::\\64")