T8534: refactor sysctl_(read|write) to accept key parts

Interface names can contain dots (e.g. VLAN subinterfaces like eth0.10), which
conflicts with sysctl's dot-separated key syntax.

Change sysctl_read() and sysctl_write() to take key components as a list and
normalize each component by replacing . with / before invoking sysctl. This
fixes sysctl lookups/updates for VLAN subinterfaces.

Extend the SR-TE smoketest to cover a VLAN subinterface.

Previous error raising this issue:
  vyos-configd: sysctl: cannot stat /proc/sys/net/ipv6/conf/eth0/10/seg6_enabled: No such file or directory
  vyos-configd: sysctl: cannot stat /proc/sys/net/ipv6/conf/eth0/201/seg6_enabled: No such file or directory

17 files changed, 126 insertions, 106 deletions

diff --git a/python/vyos/utils/system.py b/python/vyos/utils/system.py
index 578537544..fd5f49645 100644
--- a/python/vyos/utils/system.py
+++ b/python/vyos/utils/system.py
@@ -16,28 +16,34 @@
 import os
 from subprocess import run
 
-def sysctl_read(name: str) -> str:
+def _sysctl_key(parts: list[str]) -> str:
+    normalized = (p.replace('.', '/') for p in parts)
+    return '.'.join(normalized)
+
+def sysctl_read(name: list[str]) -> str:
     """Read and return current value of sysctl() option
 
     Args:
-        name (str): sysctl key name
+        name (list[str]): sysctl key name components
 
     Returns:
         str: sysctl key value
     """
-    tmp = run(['sysctl', '-nb', name], capture_output=True)
-    return tmp.stdout.decode()
+    key = _sysctl_key(name)
+    tmp = run(['sysctl', '-nb', key], capture_output=True)
+    return tmp.stdout.decode().strip()
 
-def sysctl_write(name: str, value: str | int) -> bool:
+def sysctl_write(name: list[str], value: str | int) -> bool:
     """Change value via sysctl()
 
     Args:
-        name (str): sysctl key name
+        name (list[str]): sysctl key name components
         value (str | int): sysctl key value
 
     Returns:
         bool: True if changed, False otherwise
     """
+    key = _sysctl_key(name)
     # convert other types to string before comparison
     if not isinstance(value, str):
         value = str(value)
@@ -45,7 +51,7 @@ def sysctl_write(name: str, value: str | int) -> bool:
     if sysctl_read(name) == value:
         return True
     # return False if sysctl call failed
-    if run(['sysctl', '-wq', f'{name}={value}']).returncode != 0:
+    if run(['sysctl', '-wq', f'{key}={value}']).returncode != 0:
         return False
     # compare old and new values
     # sysctl may apply value, but its actual value will be
@@ -55,11 +61,11 @@ def sysctl_write(name: str, value: str | int) -> bool:
     # False in other cases
     return False
 
-def sysctl_apply(sysctl_dict: dict[str, str], revert: bool = True) -> bool:
+def sysctl_apply(sysctl_dict: dict[tuple[str, ...], str], revert: bool = True) -> bool:
     """Apply sysctl values.
 
     Args:
-        sysctl_dict (dict[str, str]): dictionary with sysctl keys with values
+        sysctl_dict (dict[tuple[str, ...], str]): dictionary with sysctl key components (tuple) with values
         revert (bool, optional): Revert to original values if new were not
         applied. Defaults to True.
 
@@ -67,12 +73,12 @@ def sysctl_apply(sysctl_dict: dict[str, str], revert: bool = True) -> bool:
         bool: True if all params configured properly, False in other cases
     """
     # get current values
-    sysctl_original: dict[str, str] = {}
-    for key_name in sysctl_dict.keys():
-        sysctl_original[key_name] = sysctl_read(key_name)
+    sysctl_original: dict[tuple[str, ...], str] = {}
+    for key_parts in sysctl_dict.keys():
+        sysctl_original[key_parts] = sysctl_read(list(key_parts))
     # apply new values and revert in case one of them was not applied
-    for key_name, value in sysctl_dict.items():
-        if not sysctl_write(key_name, value):
+    for key_parts, value in sysctl_dict.items():
+        if not sysctl_write(list(key_parts), value):
             if revert:
                 sysctl_apply(sysctl_original, revert=False)
             return False
diff --git a/smoketest/scripts/cli/test_nat64.py b/smoketest/scripts/cli/test_nat64.py
index c578da214..dc0eb0abc 100755
--- a/smoketest/scripts/cli/test_nat64.py
+++ b/smoketest/scripts/cli/test_nat64.py
@@ -65,7 +65,7 @@ class TestNAT64(VyOSUnitTestSHIM.TestCase):
         # https://nicmx.github.io/Jool/en/usr-flags-pool4.html#port-range
         # Jool is incapable of ensuring pool4 does not intersect with other defined
         # port ranges; this validation is the operator’s responsibility.
-        tmp = sysctl_read('net.ipv4.ip_local_port_range')
+        tmp = sysctl_read(['net', 'ipv4', 'ip_local_port_range'])
         _, ephemeral_port_max = map(int, tmp.split())
         pool_port = f'{ephemeral_port_max +1}-{ephemeral_port_max +1000}'
         self.cli_set(rule_path + ['translation', 'pool', translation_rule,
diff --git a/smoketest/scripts/cli/test_protocols_segment-routing.py b/smoketest/scripts/cli/test_protocols_segment-routing.py
index 85c25998c..7be76ee0c 100755
--- a/smoketest/scripts/cli/test_protocols_segment-routing.py
+++ b/smoketest/scripts/cli/test_protocols_segment-routing.py
@@ -26,7 +26,6 @@ from vyos.utils.system import sysctl_read
 
 base_path = ['protocols', 'segment-routing']
 
-
 class TestProtocolsSegmentRouting(VyOSUnitTestSHIM.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -37,6 +36,16 @@ class TestProtocolsSegmentRouting(VyOSUnitTestSHIM.TestCase):
         # ensure we can also run this test on a live system - so lets clean
         # out the current configuration :)
         cls.cli_delete(cls, base_path)
+        # create a VLAN interface for testing
+        cls.cli_set(cls, ['interfaces', 'ethernet', 'eth0', 'vif', '4000',
+                          'address', '192.168.40.1/24'])
+        cls.cli_commit(cls)
+        cls._interfaces = Section.interfaces('ethernet', vlan=True)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.cli_delete(cls, ['interfaces', 'ethernet', 'eth0', 'vif', '4000'])
+        super(TestProtocolsSegmentRouting, cls).tearDownClass()
 
     def tearDown(self):
         self.cli_delete(base_path)
@@ -48,7 +57,6 @@ class TestProtocolsSegmentRouting(VyOSUnitTestSHIM.TestCase):
         super().tearDown()
 
     def test_srv6(self):
-        interfaces = Section.interfaces('ethernet', vlan=False)
         locators = {
             'foo1': {'prefix': '2001:a::/64'},
             'foo2': {'prefix': '2001:b::/64', 'usid': {}},
@@ -112,17 +120,17 @@ class TestProtocolsSegmentRouting(VyOSUnitTestSHIM.TestCase):
         # verify() - SRv6 should be enabled on at least one interface!
         with self.assertRaises(ConfigSessionError):
             self.cli_commit()
-        for interface in interfaces:
+        for interface in self._interfaces:
             self.cli_set(base_path + ['interface', interface, 'srv6'])
 
         self.cli_commit()
 
-        for interface in interfaces:
+        for interface in self._interfaces:
             self.assertEqual(
-                sysctl_read(f'net.ipv6.conf.{interface}.seg6_enabled'), '1'
+                sysctl_read(['net', 'ipv6', 'conf', interface, 'seg6_enabled']), '1'
             )
             self.assertEqual(
-                sysctl_read(f'net.ipv6.conf.{interface}.seg6_require_hmac'), '0'
+                sysctl_read(['net', 'ipv6', 'conf', interface, 'seg6_require_hmac']), '0'
             )  # default
 
         frrconfig = self.getFRRconfig('segment-routing', stop_section='^exit')
@@ -163,7 +171,7 @@ class TestProtocolsSegmentRouting(VyOSUnitTestSHIM.TestCase):
         source6 = '2001:db8::1'
 
         # SRv6 must be enabled on at least one interface
-        for interface in Section.interfaces('ethernet', vlan=False):
+        for interface in self._interfaces:
             self.cli_set(base_path + ['interface', interface, 'srv6'])
 
         self.cli_set(base_path + ['srv6', 'encapsulation', 'source-address', source6])
@@ -176,43 +184,41 @@ class TestProtocolsSegmentRouting(VyOSUnitTestSHIM.TestCase):
         self.assertIn(f'   source-address {source6}', frrconfig)
 
     def test_srv6_sysctl(self):
-        interfaces = Section.interfaces('ethernet', vlan=False)
 
         # HMAC accept
-        for interface in interfaces:
+        for interface in self._interfaces:
             self.cli_set(base_path + ['interface', interface, 'srv6'])
             self.cli_set(base_path + ['interface', interface, 'srv6', 'hmac', 'ignore'])
         self.cli_commit()
 
-        for interface in interfaces:
+        for interface in self._interfaces:
             self.assertEqual(
-                sysctl_read(f'net.ipv6.conf.{interface}.seg6_enabled'), '1'
+                sysctl_read(['net', 'ipv6', 'conf', interface, 'seg6_enabled']), '1'
             )
             self.assertEqual(
-                sysctl_read(f'net.ipv6.conf.{interface}.seg6_require_hmac'), '-1'
+                sysctl_read(['net', 'ipv6', 'conf', interface, 'seg6_require_hmac']), '-1'
             )  # ignore
 
         # HMAC drop
-        for interface in interfaces:
+        for interface in self._interfaces:
             self.cli_set(base_path + ['interface', interface, 'srv6'])
             self.cli_set(base_path + ['interface', interface, 'srv6', 'hmac', 'drop'])
         self.cli_commit()
 
-        for interface in interfaces:
+        for interface in self._interfaces:
             self.assertEqual(
-                sysctl_read(f'net.ipv6.conf.{interface}.seg6_enabled'), '1'
+                sysctl_read(['net', 'ipv6', 'conf', interface, 'seg6_enabled']), '1'
             )
             self.assertEqual(
-                sysctl_read(f'net.ipv6.conf.{interface}.seg6_require_hmac'), '1'
+                sysctl_read(['net', 'ipv6', 'conf', interface, 'seg6_require_hmac']), '1'
             )  # drop
 
         # Disable SRv6 on first interface
-        first_if = interfaces[-1]
+        first_if = self._interfaces[-1]
         self.cli_delete(base_path + ['interface', first_if])
         self.cli_commit()
 
-        self.assertEqual(sysctl_read(f'net.ipv6.conf.{first_if}.seg6_enabled'), '0')
-
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', first_if, 'seg6_enabled']), '0')
 
 if __name__ == '__main__':
     unittest.main(verbosity=2, failfast=VyOSUnitTestSHIM.TestCase.debug_on())
diff --git a/smoketest/scripts/cli/test_system_conntrack.py b/smoketest/scripts/cli/test_system_conntrack.py
index b0b34b740..e84ad3644 100755
--- a/smoketest/scripts/cli/test_system_conntrack.py
+++ b/smoketest/scripts/cli/test_system_conntrack.py
@@ -208,7 +208,7 @@ class TestSystemConntrack(VyOSUnitTestSHIM.TestCase):
 
         # verify new configuration - only effective after reboot, but
         # a valid config file is sufficient
-        tmp = sysctl_read('net.netfilter.nf_conntrack_buckets')
+        tmp = sysctl_read(['net', 'netfilter', 'nf_conntrack_buckets'])
         self.assertIn(hash_size, tmp)
 
         # Test default value by deleting the configuration
@@ -219,7 +219,7 @@ class TestSystemConntrack(VyOSUnitTestSHIM.TestCase):
 
         # verify new configuration - only effective after reboot, but
         # a valid config file is sufficient
-        tmp = sysctl_read('net.netfilter.nf_conntrack_buckets')
+        tmp = sysctl_read(['net', 'netfilter', 'nf_conntrack_buckets'])
         self.assertIn(hash_size_default, tmp)
 
     def test_conntrack_ignore(self):
diff --git a/smoketest/scripts/cli/test_system_ip.py b/smoketest/scripts/cli/test_system_ip.py
index 25dbeef1d..43ae4e008 100755
--- a/smoketest/scripts/cli/test_system_ip.py
+++ b/smoketest/scripts/cli/test_system_ip.py
@@ -40,40 +40,40 @@ class TestSystemIP(VyOSUnitTestSHIM.TestCase):
     def test_system_ip_forwarding(self):
         # Test if IPv4 forwarding can be disabled globally, default is '1'
         # which means forwarding enabled
-        self.assertEqual(sysctl_read('net.ipv4.conf.all.forwarding'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'conf', 'all', 'forwarding']), '1')
 
         self.cli_set(base_path + ['disable-forwarding'])
         self.cli_commit()
 
-        self.assertEqual(sysctl_read('net.ipv4.conf.all.forwarding'), '0')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'conf', 'all', 'forwarding']), '0')
         frrconfig = self.getFRRconfig()
         self.assertIn('no ip forwarding', frrconfig)
 
         self.cli_delete(base_path + ['disable-forwarding'])
         self.cli_commit()
 
-        self.assertEqual(sysctl_read('net.ipv4.conf.all.forwarding'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'conf', 'all', 'forwarding']), '1')
         frrconfig = self.getFRRconfig()
         self.assertNotIn('no ip forwarding', frrconfig)
 
     def test_system_ip_multipath(self):
         # Test IPv4 multipathing options, options default to off -> '0'
-        self.assertEqual(sysctl_read('net.ipv4.fib_multipath_use_neigh'), '0')
-        self.assertEqual(sysctl_read('net.ipv4.fib_multipath_hash_policy'), '0')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'fib_multipath_use_neigh']), '0')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'fib_multipath_hash_policy']), '0')
 
         self.cli_set(base_path + ['multipath', 'ignore-unreachable-nexthops'])
         self.cli_set(base_path + ['multipath', 'layer4-hashing'])
         self.cli_commit()
 
-        self.assertEqual(sysctl_read('net.ipv4.fib_multipath_use_neigh'), '1')
-        self.assertEqual(sysctl_read('net.ipv4.fib_multipath_hash_policy'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'fib_multipath_use_neigh']), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'fib_multipath_hash_policy']), '1')
 
     def test_system_ip_arp_table_size(self):
         cli_default = int(default_value(base_path + ['arp', 'table-size']))
         def _verify_gc_thres(table_size):
-            self.assertEqual(sysctl_read('net.ipv4.neigh.default.gc_thresh3'), str(table_size))
-            self.assertEqual(sysctl_read('net.ipv4.neigh.default.gc_thresh2'), str(table_size // 2))
-            self.assertEqual(sysctl_read('net.ipv4.neigh.default.gc_thresh1'), str(table_size // 8))
+            self.assertEqual(sysctl_read(['net', 'ipv4', 'neigh', 'default', 'gc_thresh3']), str(table_size))
+            self.assertEqual(sysctl_read(['net', 'ipv4', 'neigh', 'default', 'gc_thresh2']), str(table_size // 2))
+            self.assertEqual(sysctl_read(['net', 'ipv4', 'neigh', 'default', 'gc_thresh1']), str(table_size // 8))
 
         _verify_gc_thres(cli_default)
 
diff --git a/smoketest/scripts/cli/test_system_ipv6.py b/smoketest/scripts/cli/test_system_ipv6.py
index f7d74eb58..eacf833ae 100755
--- a/smoketest/scripts/cli/test_system_ipv6.py
+++ b/smoketest/scripts/cli/test_system_ipv6.py
@@ -41,23 +41,23 @@ class TestSystemIPv6(VyOSUnitTestSHIM.TestCase):
     def test_system_ipv6_forwarding(self):
         # Test if IPv6 forwarding can be disabled globally, default is '1'
         # which means forwearding enabled
-        self.assertEqual(sysctl_read('net.ipv6.conf.all.forwarding'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', 'all', 'forwarding']), '1')
 
         self.cli_set(base_path + ['disable-forwarding'])
         self.cli_commit()
-        self.assertEqual(sysctl_read('net.ipv6.conf.all.forwarding'), '0')
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', 'all', 'forwarding']), '0')
         frrconfig = self.getFRRconfig()
         self.assertIn('no ipv6 forwarding', frrconfig)
 
         self.cli_delete(base_path + ['disable-forwarding'])
         self.cli_commit()
-        self.assertEqual(sysctl_read('net.ipv6.conf.all.forwarding'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', 'all', 'forwarding']), '1')
         frrconfig = self.getFRRconfig()
         self.assertNotIn('no ipv6 forwarding', frrconfig)
 
     def test_system_ipv6_strict_dad(self):
         # This defaults to 1
-        self.assertEqual(sysctl_read('net.ipv6.conf.all.accept_dad'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', 'all', 'accept_dad']), '1')
 
         # Do not assign any IPv6 address on interfaces, this requires a reboot
         # which can not be tested, but we can read the config file :)
@@ -65,11 +65,11 @@ class TestSystemIPv6(VyOSUnitTestSHIM.TestCase):
         self.cli_commit()
 
         # Verify configuration file
-        self.assertEqual(sysctl_read('net.ipv6.conf.all.accept_dad'), '2')
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', 'all', 'accept_dad']), '2')
 
     def test_system_ipv6_multipath(self):
         # This defaults to 0
-        self.assertEqual(sysctl_read('net.ipv6.fib_multipath_hash_policy'), '0')
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'fib_multipath_hash_policy']), '0')
 
         # Do not assign any IPv6 address on interfaces, this requires a reboot
         # which can not be tested, but we can read the config file :)
@@ -77,7 +77,7 @@ class TestSystemIPv6(VyOSUnitTestSHIM.TestCase):
         self.cli_commit()
 
         # Verify configuration file
-        self.assertEqual(sysctl_read('net.ipv6.fib_multipath_hash_policy'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv6', 'fib_multipath_hash_policy']), '1')
 
     def test_system_ipv6_neighbor_table_size(self):
         # Maximum number of entries to keep in the ARP cache, the
@@ -85,9 +85,9 @@ class TestSystemIPv6(VyOSUnitTestSHIM.TestCase):
         cli_default = int(default_value(base_path + ['neighbor', 'table-size']))
 
         def _verify_gc_thres(table_size):
-            self.assertEqual(sysctl_read('net.ipv6.neigh.default.gc_thresh3'), str(table_size))
-            self.assertEqual(sysctl_read('net.ipv6.neigh.default.gc_thresh2'), str(table_size // 2))
-            self.assertEqual(sysctl_read('net.ipv6.neigh.default.gc_thresh1'), str(table_size // 8))
+            self.assertEqual(sysctl_read(['net', 'ipv6', 'neigh', 'default', 'gc_thresh3']), str(table_size))
+            self.assertEqual(sysctl_read(['net', 'ipv6', 'neigh', 'default', 'gc_thresh2']), str(table_size // 2))
+            self.assertEqual(sysctl_read(['net', 'ipv6', 'neigh', 'default', 'gc_thresh1']), str(table_size // 8))
 
         _verify_gc_thres(cli_default)
 
diff --git a/smoketest/scripts/cli/test_system_option.py b/smoketest/scripts/cli/test_system_option.py
index ce53b6498..5a8e5b78c 100755
--- a/smoketest/scripts/cli/test_system_option.py
+++ b/smoketest/scripts/cli/test_system_option.py
@@ -84,9 +84,9 @@ class TestSystemOption(VyOSUnitTestSHIM.TestCase):
 
         self.assertTrue(is_systemd_service_active(tuned_service))
 
-        self.assertEqual(sysctl_read('net.ipv4.neigh.default.gc_thresh1'), gc_thresh1)
-        self.assertEqual(sysctl_read('net.ipv4.neigh.default.gc_thresh2'), gc_thresh2)
-        self.assertEqual(sysctl_read('net.ipv4.neigh.default.gc_thresh3'), gc_thresh3)
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'neigh', 'default', 'gc_thresh1']), gc_thresh1)
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'neigh', 'default', 'gc_thresh2']), gc_thresh2)
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'neigh', 'default', 'gc_thresh3']), gc_thresh3)
 
     def test_ssh_client_options(self):
         loopback = 'lo'
diff --git a/smoketest/scripts/cli/test_vpp.py b/smoketest/scripts/cli/test_vpp.py
index 8fcd541f0..e987da8df 100755
--- a/smoketest/scripts/cli/test_vpp.py
+++ b/smoketest/scripts/cli/test_vpp.py
@@ -1052,25 +1052,25 @@ class TestVPP(VyOSUnitTestSHIM.TestCase):
 
         # Check if max-map-count has default auto calculated value
         # but not less than '65530'
-        self.assertEqual(sysctl_read('vm.max_map_count'), '65530')
+        self.assertEqual(sysctl_read(['vm', 'max_map_count']), '65530')
         # The same is with: kernel.shmmax = '8589934592'
-        self.assertEqual(sysctl_read('kernel.shmmax'), '8589934592')
+        self.assertEqual(sysctl_read(['kernel', 'shmmax']), '8589934592')
 
         # Change max-map-count, shmmax and check
         self.cli_set(hr_path + ['max-map-count', max_map_count])
         self.cli_set(hr_path + ['shmmax', shmmax])
         self.cli_commit()
 
-        self.assertEqual(sysctl_read('vm.max_map_count'), max_map_count)
-        self.assertEqual(sysctl_read('kernel.shmmax'), shmmax)
+        self.assertEqual(sysctl_read(['vm', 'max_map_count']), max_map_count)
+        self.assertEqual(sysctl_read(['kernel', 'shmmax']), shmmax)
 
         # We expect max-map-count and shmmax will return auto calculated values
         self.cli_delete(hr_path + ['max-map-count'])
         self.cli_delete(hr_path + ['shmmax'])
         self.cli_commit()
 
-        self.assertEqual(sysctl_read('vm.max_map_count'), '65530')
-        self.assertEqual(sysctl_read('kernel.shmmax'), '8589934592')
+        self.assertEqual(sysctl_read(['vm', 'max_map_count']), '65530')
+        self.assertEqual(sysctl_read(['kernel', 'shmmax']), '8589934592')
 
     def test_17_1_vpp_pppoe_mapping(self):
         config_file = '/run/accel-pppd/pppoe.conf'
diff --git a/smoketest/scripts/cli/test_vrf.py b/smoketest/scripts/cli/test_vrf.py
index 63fe46d21..902308b9b 100755
--- a/smoketest/scripts/cli/test_vrf.py
+++ b/smoketest/scripts/cli/test_vrf.py
@@ -183,8 +183,8 @@ class VRFTest(VyOSUnitTestSHIM.TestCase):
             # Ensure VRF was created
             self.assertTrue(interface_exists(vrf))
             # Verify IP forwarding is 1 (enabled)
-            self.assertEqual(sysctl_read(f'net.ipv4.conf.{vrf}.forwarding'), '1')
-            self.assertEqual(sysctl_read(f'net.ipv6.conf.{vrf}.forwarding'), '1')
+            self.assertEqual(sysctl_read(['net', 'ipv4', 'conf', vrf, 'forwarding']), '1')
+            self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', vrf, 'forwarding']), '1')
 
             # Test for proper loopback IP assignment
             for addr in loopbacks:
@@ -203,11 +203,11 @@ class VRFTest(VyOSUnitTestSHIM.TestCase):
         self.cli_commit()
 
         # Verify VRF configuration
-        self.assertEqual(sysctl_read('net.ipv4.tcp_l3mdev_accept'), '1')
-        self.assertEqual(sysctl_read('net.ipv4.udp_l3mdev_accept'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'tcp_l3mdev_accept']), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'udp_l3mdev_accept']), '1')
 
         # If there is any VRF defined, strict_mode should be on
-        self.assertEqual(sysctl_read('net.vrf.strict_mode'), '1')
+        self.assertEqual(sysctl_read(['net', 'vrf', 'strict_mode']), '1')
 
     def test_vrf_table_id_is_unalterable(self):
         # Linux Kernel prohibits the change of a VRF table  on the fly.
@@ -386,8 +386,8 @@ class VRFTest(VyOSUnitTestSHIM.TestCase):
             # Ensure VRF was created
             self.assertTrue(interface_exists(vrf))
             # Verify IP forwarding is 0 (disabled)
-            self.assertEqual(sysctl_read(f'net.ipv4.conf.{vrf}.forwarding'), '0')
-            self.assertEqual(sysctl_read(f'net.ipv6.conf.{vrf}.forwarding'), '0')
+            self.assertEqual(sysctl_read(['net', 'ipv4', 'conf', vrf, 'forwarding']), '0')
+            self.assertEqual(sysctl_read(['net', 'ipv6', 'conf', vrf, 'forwarding']), '0')
 
     def test_vrf_ip_protocol_route_map(self):
         table = '6000'
diff --git a/src/conf_mode/nat64.py b/src/conf_mode/nat64.py
index 175e9eb1e..b7d0b586d 100755
--- a/src/conf_mode/nat64.py
+++ b/src/conf_mode/nat64.py
@@ -96,7 +96,7 @@ def verify(nat64) -> None:
     # https://nicmx.github.io/Jool/en/usr-flags-pool4.html#port-range
     # Jool is incapable of ensuring pool4 does not intersect with other defined
     # port ranges; this validation is the operator’s responsibility.
-    tmp = sysctl_read('net.ipv4.ip_local_port_range')
+    tmp = sysctl_read(['net', 'ipv4', 'ip_local_port_range'])
     ephemeral_port_min, ephemeral_port_max = map(int, tmp.split())
 
     if dict_search('source.rule', nat64):
diff --git a/src/conf_mode/protocols_mpls.py b/src/conf_mode/protocols_mpls.py
index d534ad2e3..841e5406f 100755
--- a/src/conf_mode/protocols_mpls.py
+++ b/src/conf_mode/protocols_mpls.py
@@ -85,21 +85,21 @@ def apply(config_dict):
     labels = '0'
     if 'interface' in mpls:
         labels = '1048575'
-    sysctl_write('net.mpls.platform_labels', labels)
+    sysctl_write(['net', 'mpls', 'platform_labels'], labels)
 
     # Check for changes in global MPLS options
     if 'parameters' in mpls:
             # Choose whether to copy IP TTL to MPLS header TTL
         if 'no_propagate_ttl' in mpls['parameters']:
-            sysctl_write('net.mpls.ip_ttl_propagate', 0)
+            sysctl_write(['net', 'mpls', 'ip_ttl_propagate'], 0)
             # Choose whether to limit maximum MPLS header TTL
         if 'maximum_ttl' in mpls['parameters']:
             ttl = mpls['parameters']['maximum_ttl']
-            sysctl_write('net.mpls.default_ttl', ttl)
+            sysctl_write(['net', 'mpls', 'default_ttl'], ttl)
     else:
         # Set default global MPLS options if not defined.
-        sysctl_write('net.mpls.ip_ttl_propagate', 1)
-        sysctl_write('net.mpls.default_ttl', 255)
+        sysctl_write(['net', 'mpls', 'ip_ttl_propagate'], 1)
+        sysctl_write(['net', 'mpls', 'default_ttl'], 255)
 
     # Enable and disable MPLS processing on interfaces per configuration
     if 'interface' in mpls:
@@ -112,20 +112,17 @@ def apply(config_dict):
             interface_state = read_file(f'/proc/sys/net/mpls/conf/{system_interface}/input')
             if '1' in interface_state:
                 if system_interface not in mpls['interface']:
-                    system_interface = system_interface.replace('.', '/')
-                    sysctl_write(f'net.mpls.conf.{system_interface}.input', 0)
+                    sysctl_write(['net', 'mpls', 'conf', system_interface, 'input'], 0)
             elif '0' in interface_state:
                 if system_interface in mpls['interface']:
-                    system_interface = system_interface.replace('.', '/')
-                    sysctl_write(f'net.mpls.conf.{system_interface}.input', 1)
+                    sysctl_write(['net', 'mpls', 'conf', system_interface, 'input'], 1)
     else:
         system_interfaces = []
         # If MPLS interfaces are not configured, set MPLS processing disabled
         for interface in glob('/proc/sys/net/mpls/conf/*'):
             system_interfaces.append(os.path.basename(interface))
         for system_interface in system_interfaces:
-            system_interface = system_interface.replace('.', '/')
-            sysctl_write(f'net.mpls.conf.{system_interface}.input', 0)
+            sysctl_write(['net', 'mpls', 'conf', system_interface, 'input'], 0)
 
     return None
 
diff --git a/src/conf_mode/protocols_segment-routing.py b/src/conf_mode/protocols_segment-routing.py
index 4020a0917..8a4d12bd8 100755
--- a/src/conf_mode/protocols_segment-routing.py
+++ b/src/conf_mode/protocols_segment-routing.py
@@ -70,24 +70,24 @@ def apply(config_dict):
 
     for interface in list_diff(current_interfaces, sr_interfaces):
         # Disable processing of IPv6-SR packets
-        sysctl_write(f'net.ipv6.conf.{interface}.seg6_enabled', '0')
+        sysctl_write(['net', 'ipv6', 'conf', interface, 'seg6_enabled'], '0')
 
     for interface, interface_config in sr.get('interface', {}).items():
         # Accept or drop SR-enabled IPv6 packets on this interface
         if 'srv6' in interface_config:
-            sysctl_write(f'net.ipv6.conf.{interface}.seg6_enabled', '1')
+            sysctl_write(['net', 'ipv6', 'conf', interface, 'seg6_enabled'], '1')
             # Define HMAC policy for ingress SR-enabled packets on this interface
             # It's a redundant check as HMAC has a default value - but better safe
             # then sorry
             tmp = dict_search('srv6.hmac', interface_config)
             if tmp == 'accept':
-                sysctl_write(f'net.ipv6.conf.{interface}.seg6_require_hmac', '0')
+                sysctl_write(['net', 'ipv6', 'conf', interface, 'seg6_require_hmac'], '0')
             elif tmp == 'drop':
-                sysctl_write(f'net.ipv6.conf.{interface}.seg6_require_hmac', '1')
+                sysctl_write(['net', 'ipv6', 'conf', interface, 'seg6_require_hmac'], '1')
             elif tmp == 'ignore':
-                sysctl_write(f'net.ipv6.conf.{interface}.seg6_require_hmac', '-1')
+                sysctl_write(['net', 'ipv6', 'conf', interface, 'seg6_require_hmac'], '-1')
         else:
-            sysctl_write(f'net.ipv6.conf.{interface}.seg6_enabled', '0')
+            sysctl_write(['net', 'ipv6', 'conf', interface, 'seg6_enabled'], '0')
 
     if config_dict and not is_systemd_service_running('vyos-configd.service'):
         FRRender().apply()
diff --git a/src/conf_mode/system_ip.py b/src/conf_mode/system_ip.py
index 13471eb3b..6aff982d7 100755
--- a/src/conf_mode/system_ip.py
+++ b/src/conf_mode/system_ip.py
@@ -75,20 +75,20 @@ def apply(config_dict):
     # table_size has a default value - thus the key always exists
     size = int(dict_search('arp.table_size', opt))
     # Amount upon reaching which the records begin to be cleared immediately
-    sysctl_write('net.ipv4.neigh.default.gc_thresh3', size)
+    sysctl_write(['net', 'ipv4', 'neigh', 'default', 'gc_thresh3'], size)
     # Amount after which the records begin to be cleaned after 5 seconds
-    sysctl_write('net.ipv4.neigh.default.gc_thresh2', size // 2)
+    sysctl_write(['net', 'ipv4', 'neigh', 'default', 'gc_thresh2'], size // 2)
     # Minimum number of stored records is indicated which is not cleared
-    sysctl_write('net.ipv4.neigh.default.gc_thresh1', size // 8)
+    sysctl_write(['net', 'ipv4', 'neigh', 'default', 'gc_thresh1'], size // 8)
 
     # configure multipath
     tmp = dict_search('multipath.ignore_unreachable_nexthops', opt)
     value = '1' if (tmp != None) else '0'
-    sysctl_write('net.ipv4.fib_multipath_use_neigh', value)
+    sysctl_write(['net', 'ipv4', 'fib_multipath_use_neigh'], value)
 
     tmp = dict_search('multipath.layer4_hashing', opt)
     value = '1' if (tmp != None) else '0'
-    sysctl_write('net.ipv4.fib_multipath_hash_policy', value)
+    sysctl_write(['net', 'ipv4', 'fib_multipath_hash_policy'], value)
 
     # configure TCP options (defaults as of Linux 6.4)
     tmp = dict_search('tcp.mss.probing', opt)
@@ -101,15 +101,15 @@ def apply(config_dict):
     else:
         # Shouldn't happen
         raise ValueError("TCP MSS probing is neither 'on-icmp-black-hole' nor 'force'!")
-    sysctl_write('net.ipv4.tcp_mtu_probing', value)
+    sysctl_write(['net', 'ipv4', 'tcp_mtu_probing'], value)
 
     tmp = dict_search('tcp.mss.base', opt)
     value = '1024' if (tmp is None) else tmp
-    sysctl_write('net.ipv4.tcp_base_mss', value)
+    sysctl_write(['net', 'ipv4', 'tcp_base_mss'], value)
 
     tmp = dict_search('tcp.mss.floor', opt)
     value = '48' if (tmp is None) else tmp
-    sysctl_write('net.ipv4.tcp_mtu_probe_floor', value)
+    sysctl_write(['net', 'ipv4', 'tcp_mtu_probe_floor'], value)
 
     # During startup of vyos-router that brings up FRR, the service is not yet
     # running when this script is called first. Skip this part and wait for initial
diff --git a/src/conf_mode/system_ipv6.py b/src/conf_mode/system_ipv6.py
index 35f343d82..80a7a386a 100755
--- a/src/conf_mode/system_ipv6.py
+++ b/src/conf_mode/system_ipv6.py
@@ -70,17 +70,17 @@ def apply(config_dict):
     # configure multipath
     tmp = dict_search('multipath.layer4_hashing', opt)
     value = '1' if (tmp != None) else '0'
-    sysctl_write('net.ipv6.fib_multipath_hash_policy', value)
+    sysctl_write(['net', 'ipv6', 'fib_multipath_hash_policy'], value)
 
     # Apply ND threshold values
     # table_size has a default value - thus the key always exists
     size = int(dict_search('neighbor.table_size', opt))
     # Amount upon reaching which the records begin to be cleared immediately
-    sysctl_write('net.ipv6.neigh.default.gc_thresh3', size)
+    sysctl_write(['net', 'ipv6', 'neigh', 'default', 'gc_thresh3'], size)
     # Amount after which the records begin to be cleaned after 5 seconds
-    sysctl_write('net.ipv6.neigh.default.gc_thresh2', size // 2)
+    sysctl_write(['net', 'ipv6', 'neigh', 'default', 'gc_thresh2'], size // 2)
     # Minimum number of stored records is indicated which is not cleared
-    sysctl_write('net.ipv6.neigh.default.gc_thresh1', size // 8)
+    sysctl_write(['net', 'ipv6', 'neigh', 'default', 'gc_thresh1'], size // 8)
 
     # configure IPv6 strict-dad
     tmp = dict_search('strict_dad', opt)
diff --git a/src/conf_mode/system_option.py b/src/conf_mode/system_option.py
index 41b9a23bb..ac6e7768d 100755
--- a/src/conf_mode/system_option.py
+++ b/src/conf_mode/system_option.py
@@ -571,7 +571,7 @@ def apply(options):
         }
 
         for parameter, value in parameters.items():
-            sysctl_write(parameter, value)
+            sysctl_write(parameter.split('.'), value)
 
 
 if __name__ == '__main__':
diff --git a/src/conf_mode/vrf.py b/src/conf_mode/vrf.py
index c9b34fe8e..92bb956ed 100755
--- a/src/conf_mode/vrf.py
+++ b/src/conf_mode/vrf.py
@@ -243,8 +243,8 @@ def apply(vrf):
     bind_all = '0'
     if 'bind_to_all' in vrf:
         bind_all = '1'
-    sysctl_write('net.ipv4.tcp_l3mdev_accept', bind_all)
-    sysctl_write('net.ipv4.udp_l3mdev_accept', bind_all)
+    sysctl_write(['net', 'ipv4', 'tcp_l3mdev_accept'], bind_all)
+    sysctl_write(['net', 'ipv4', 'udp_l3mdev_accept'], bind_all)
 
     for tmp in (dict_search('vrf_remove', vrf) or []):
         if interface_exists(tmp):
diff --git a/src/tests/test_utils.py b/src/tests/test_utils.py
index 9cf6f7a1e..bbf1cb523 100644
--- a/src/tests/test_utils.py
+++ b/src/tests/test_utils.py
@@ -13,6 +13,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from unittest import TestCase
+from unittest.mock import patch
 
 class TestVyOSUtils(TestCase):
     def test_key_mangling(self):
@@ -24,7 +25,17 @@ class TestVyOSUtils(TestCase):
 
     def test_sysctl_read(self):
         from vyos.utils.system import sysctl_read
-        self.assertEqual(sysctl_read('net.ipv4.conf.lo.forwarding'), '1')
+        self.assertEqual(sysctl_read(['net', 'ipv4', 'conf', 'lo', 'forwarding']), '1')
+
+    def test_sysctl_key_normalization(self):
+        from vyos.utils.system import sysctl_read
+        with patch('vyos.utils.system.run') as mock_run:
+            mock_run.return_value.stdout = b'1\n'
+            sysctl_read(['net', 'ipv4', 'conf', 'eth0.10', 'forwarding'])
+            mock_run.assert_called_with(
+                ['sysctl', '-nb', 'net.ipv4.conf.eth0/10.forwarding'],
+                capture_output=True,
+            )
 
     def test_list_strip(self):
         from vyos.utils.list import list_strip