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

8 files changed, 64 insertions, 58 deletions

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'