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, 42 insertions, 34 deletions

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