fix directory data structure error and add validation for sid prefix length

Signed-off-by: BYGX-wcr <[email protected]>

Author: BYGX-wcr

src/sonic-bgpcfgd/bgpcfgd/managers_srv6.py

@@ -56,10 +56,16 @@ def sids_set_handler(self, key, data):
         locator_name = key.split("|")[0]
         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))
@@ -80,7 +86,7 @@ def sids_set_handler(self, key, data):
         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):
@@ -102,18 +108,18 @@ def sids_del_handler(self, key):
         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):

src/sonic-bgpcfgd/tests/test_srv6.py

@@ -87,7 +87,8 @@ def test_uN_add():
         '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::/48")
+    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()
@@ -103,7 +104,8 @@ 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::/64")
+    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()
@@ -123,7 +125,7 @@ def test_uN_del():
             '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::/48")
+    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()
@@ -149,8 +151,8 @@ def test_uDT46_del_vrf1():
             '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::/48")
-    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc1|fcbb:bbbb:1:f2::/64")
+    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()
@@ -160,4 +162,4 @@ def test_invalid_add():
         '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::/64")
+    assert not sid_mgr.directory.path_exist(sid_mgr.db_name, sid_mgr.table_name, "loc2|fcbb:bbbb:21:f1::\\64")