dhcpv6: T8849: Add time-zone support for Kea DHCPv6 (#5190)

* dhcpv6: T8849: Add time-zone support for Kea DHCPv6

Add DHCPv6 option support for time zone (RFC4833 options
41 and 42). This includes both the POSIX-style TZ string
(`new-posix-timezone`) and the IANA time zone name
(`new-tzdb-timezone`).

* dhcpv6: T8849: Refactor per code-review suggestion

* dhcpv6: T8849: Reformat for compliance

3 files changed, 182 insertions, 93 deletions

diff --git a/interface-definitions/include/dhcp/option-v6.xml.i b/interface-definitions/include/dhcp/option-v6.xml.i
index 202843ddf..c97d652b6 100644
--- a/interface-definitions/include/dhcp/option-v6.xml.i
+++ b/interface-definitions/include/dhcp/option-v6.xml.i
@@ -102,6 +102,17 @@
       </constraint>
       </properties>
     </leafNode>
+    <leafNode name="time-zone">
+      <properties>
+        <help>Time zone to send to clients. Uses RFC4833 options 41 and 42</help>
+        <completionHelp>
+          <script>timedatectl list-timezones</script>
+        </completionHelp>
+        <constraint>
+          <validator name="timezone" argument="--validate"/>
+        </constraint>
+      </properties>
+    </leafNode>
     <node name="vendor-option">
       <properties>
         <help>Vendor Specific Options</help>
diff --git a/python/vyos/kea.py b/python/vyos/kea.py
index 74ffa3fb1..068e2cf7a 100644
--- a/python/vyos/kea.py
+++ b/python/vyos/kea.py
@@ -89,6 +89,14 @@ def _find_list_of_dict_index(lst, key='ip', value=''):
     return idx
 
 
+def _read_posix_timezone(tz_name):
+    try:
+        with open(f'/usr/share/zoneinfo/{tz_name}', 'rb') as f:
+            return f.read().split(b'\n')[-2].decode('utf-8').replace(',', '\\,')
+    except (FileNotFoundError, IOError, IndexError) as e:
+        raise ConfigError(f'Failed to read timezone data for: {tz_name}') from e
+
+
 def kea_test_config(process: str, config_path: str) -> tuple[bool, str]:
     result, output = rc_cmd(f'{process} -t {config_path}')
 
@@ -143,9 +151,7 @@ def kea_parse_options(config):
         )
 
     if 'time_zone' in config:
-        with open('/usr/share/zoneinfo/' + config['time_zone'], 'rb') as f:
-            tz_string = f.read().split(b'\n')[-2].decode('utf-8').replace(',', '\\,')
-
+        tz_string = _read_posix_timezone(config['time_zone'])
         options.append({'name': 'pcode', 'data': tz_string})
         options.append({'name': 'tcode', 'data': config['time_zone']})
 
@@ -289,6 +295,11 @@ def kea6_parse_options(config):
         if hosts:
             options.append({'name': 'sip-server-dns', 'data': ', '.join(hosts)})
 
+    if 'time_zone' in config:
+        tz_string = _read_posix_timezone(config['time_zone'])
+        options.append({'name': 'new-posix-timezone', 'data': tz_string})
+        options.append({'name': 'new-tzdb-timezone', 'data': config['time_zone']})
+
     cisco_tftp = dict_search_args(config, 'vendor_option', 'cisco', 'tftp-server')
     if cisco_tftp:
         options.append(
diff --git a/smoketest/scripts/cli/test_service_dhcpv6-server.py b/smoketest/scripts/cli/test_service_dhcpv6-server.py
index 4e0a19444..cc0c99866 100755
--- a/smoketest/scripts/cli/test_service_dhcpv6-server.py
+++ b/smoketest/scripts/cli/test_service_dhcpv6-server.py
@@ -37,18 +37,22 @@ nis_servers = ['2001:db8:ffff::1', '2001:db8:ffff::2']
 interface = 'eth0'
 interface_addr = inc_ip(subnet, 1) + '/64'
 
+
 class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
     @classmethod
     def setUpClass(cls):
         super(TestServiceDHCPv6Server, cls).setUpClass()
         # Clear out current configuration to allow running this test on a live system
         cls.cli_delete(cls, base_path)
-
-        cls.cli_set(cls, ['interfaces', 'ethernet', interface, 'address', interface_addr])
+        cls.cli_set(
+            cls, ['interfaces', 'ethernet', interface, 'address', interface_addr]
+        )
 
     @classmethod
     def tearDownClass(cls):
-        cls.cli_delete(cls, ['interfaces', 'ethernet', interface, 'address', interface_addr])
+        cls.cli_delete(
+            cls, ['interfaces', 'ethernet', interface, 'address', interface_addr]
+        )
         cls.cli_commit(cls)
 
         super(TestServiceDHCPv6Server, cls).tearDownClass()
@@ -70,7 +74,7 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
                 self.assertTrue(isinstance(current, list), msg=f'Failed path: {path}')
                 self.assertTrue(0 <= key < len(current), msg=f'Failed path: {path}')
             else:
-                assert False, "Invalid type"
+                assert False, 'Invalid type'
 
             current = current[key]
 
@@ -91,7 +95,7 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
 
     def test_single_pool(self):
         shared_net_name = 'SMOKE-1'
-        search_domains  = ['foo.vyos.net', 'bar.vyos.net']
+        search_domains = ['foo.vyos.net', 'bar.vyos.net']
         lease_time = '1200'
         max_lease_time = '72000'
         min_lease_time = '600'
@@ -99,9 +103,11 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         sip_server = 'sip.vyos.net'
         sntp_server = inc_ip(subnet, 100)
         range_start = inc_ip(subnet, 256)  # ::100
-        range_stop = inc_ip(subnet, 65535) # ::ffff
+        range_stop = inc_ip(subnet, 65535)  # ::ffff
 
         pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
+        mapping = pool + ['static-mapping']
+        conf_path = ['Dhcp6', 'shared-networks']
 
         self.cli_set(base_path + ['preference', preference])
         self.cli_set(pool + ['interface', interface])
@@ -121,6 +127,8 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         self.cli_set(pool + ['range', '1', 'start', range_start])
         self.cli_set(pool + ['range', '1', 'stop', range_stop])
 
+        self.cli_set(pool + ['option', 'time-zone', 'Europe/London'])
+
         for server in nis_servers:
             self.cli_set(pool + ['option', 'nis-server', server])
             self.cli_set(pool + ['option', 'nisplus-server', server])
@@ -128,19 +136,20 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         for search in search_domains:
             self.cli_set(pool + ['option', 'domain-search', search])
 
-        client_base = 1
-        for client in ['client1', 'client2', 'client3']:
-            duid = f'00:01:00:01:12:34:56:78:aa:bb:cc:dd:ee:{client_base:02}'
-            self.cli_set(pool + ['static-mapping', client, 'duid', duid])
-            self.cli_set(pool + ['static-mapping', client, 'ipv6-address', inc_ip(subnet, client_base)])
-            self.cli_set(pool + ['static-mapping', client, 'ipv6-prefix', inc_ip(subnet, client_base << 64) + '/64'])
-            client_base += 1
+        for client_suffix in range(1, 4):
+            duid = f'00:01:00:01:12:34:56:78:aa:bb:cc:dd:ee:{client_suffix:02}'
+            ip = inc_ip(subnet, client_suffix)
+            prefix = inc_ip(subnet, client_suffix << 64) + '/64'
+
+            self.cli_set(mapping + [f'client{client_suffix}', 'duid', duid])
+            self.cli_set(mapping + [f'client{client_suffix}', 'ipv6-address', ip])
+            self.cli_set(mapping + [f'client{client_suffix}', 'ipv6-prefix', prefix])
 
         # cannot have both mac-address and duid set
         with self.assertRaises(ConfigSessionError):
-            self.cli_set(pool + ['static-mapping', 'client1', 'mac', '00:50:00:00:00:11'])
+            self.cli_set(mapping + ['client1', 'mac', '00:50:00:00:00:11'])
             self.cli_commit()
-        self.cli_delete(pool + ['static-mapping', 'client1', 'mac'])
+        self.cli_delete(mapping + ['client1', 'mac'])
 
         # commit changes
         self.cli_commit()
@@ -148,79 +157,120 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         config = read_file(KEA6_CONF)
         obj = loads(config)
 
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks'], 'name', shared_net_name)
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'subnet', subnet)
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'interface', interface)
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'id', 1)
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'valid-lifetime', int(lease_time))
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'min-valid-lifetime', int(min_lease_time))
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'max-valid-lifetime', int(max_lease_time))
+        self.verify_config_value(obj, conf_path, 'name', shared_net_name)
+        self.verify_config_value(obj, conf_path + [0, 'subnet6'], 'subnet', subnet)
+        self.verify_config_value(
+            obj, conf_path + [0, 'subnet6'], 'interface', interface
+        )
+        self.verify_config_value(obj, conf_path + [0, 'subnet6'], 'id', 1)
+        self.verify_config_value(
+            obj,
+            conf_path + [0, 'subnet6'],
+            'valid-lifetime',
+            int(lease_time),
+        )
+        self.verify_config_value(
+            obj,
+            conf_path + [0, 'subnet6'],
+            'min-valid-lifetime',
+            int(min_lease_time),
+        )
+        self.verify_config_value(
+            obj,
+            conf_path + [0, 'subnet6'],
+            'max-valid-lifetime',
+            int(max_lease_time),
+        )
 
         # Verify options
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'capwap-ac-v6', 'data': dns_1})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'capwap-ac-v6', 'data': dns_1},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'dns-servers', 'data': f'{dns_1}, {dns_2}'})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'dns-servers', 'data': f'{dns_1}, {dns_2}'},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'domain-search', 'data': ", ".join(search_domains)})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'domain-search', 'data': ', '.join(search_domains)},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'nis-domain-name', 'data': domain})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'nis-domain-name', 'data': domain},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'nis-servers', 'data': ", ".join(nis_servers)})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'nis-servers', 'data': ', '.join(nis_servers)},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'nisp-domain-name', 'data': domain})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'nisp-domain-name', 'data': domain},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'nisp-servers', 'data': ", ".join(nis_servers)})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'nisp-servers', 'data': ', '.join(nis_servers)},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'sntp-servers', 'data': sntp_server})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'sntp-servers', 'data': sntp_server},
+        )
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'option-data'],
-                {'name': 'sip-server-dns', 'data': sip_server})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'sip-server-dns', 'data': sip_server},
+        )
+
+        # Time zone
+        self.verify_config_object(
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'new-posix-timezone', 'data': 'GMT0BST\\,M3.5.0/1\\,M10.5.0'},
+        )
+        self.verify_config_object(
+            obj,
+            conf_path + [0, 'subnet6', 0, 'option-data'],
+            {'name': 'new-tzdb-timezone', 'data': 'Europe/London'},
+        )
 
         # Verify pools
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'pools'],
-                {'pool': f'{range_start} - {range_stop}'})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'pools'],
+            {'pool': f'{range_start} - {range_stop}'},
+        )
 
-        client_base = 1
-        for client in ['client1', 'client2', 'client3']:
-            duid = f'00:01:00:01:12:34:56:78:aa:bb:cc:dd:ee:{client_base:02}'
-            ip = inc_ip(subnet, client_base)
-            prefix = inc_ip(subnet, client_base << 64) + '/64'
+        for client_suffix in range(1, 4):
+            duid = f'00:01:00:01:12:34:56:78:aa:bb:cc:dd:ee:{client_suffix:02}'
+            ip = inc_ip(subnet, client_suffix)
+            prefix = inc_ip(subnet, client_suffix << 64) + '/64'
 
             self.verify_config_object(
-                    obj,
-                    ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'reservations'],
-                    {'hostname': client, 'duid': duid, 'ip-addresses': [ip], 'prefixes': [prefix]})
-
-            client_base += 1
+                obj,
+                conf_path + [0, 'subnet6', 0, 'reservations'],
+                {
+                    'hostname': f'client{client_suffix}',
+                    'duid': duid,
+                    'ip-addresses': [ip],
+                    'prefixes': [prefix],
+                },
+            )
 
         # Check for running process
         self.assertTrue(process_named_running(PROCESS_NAME))
 
-
     def test_prefix_delegation(self):
         shared_net_name = 'SMOKE-2'
         range_start = inc_ip(subnet, 256)  # ::100
-        range_stop = inc_ip(subnet, 65535) # ::ffff
+        range_stop = inc_ip(subnet, 65535)  # ::ffff
         delegate_start = '2001:db8:ee::'
         delegate_len = '64'
         bad_prefix_len = '32'
@@ -228,43 +278,49 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         exclude_len = '66'
 
         pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
+        pd_mapping = pool + ['prefix-delegation']
+        pd_mapping_prefix = pd_mapping + ['prefix', delegate_start]
+        conf_path = ['Dhcp6', 'shared-networks']
+
         self.cli_set(pool + ['subnet-id', '1'])
         self.cli_set(pool + ['range', '1', 'start', range_start])
         self.cli_set(pool + ['range', '1', 'stop', range_stop])
-        self.cli_set(pool + ['prefix-delegation', 'prefix', delegate_start, 'delegated-length', delegate_len])
-        self.cli_set(pool + ['prefix-delegation', 'prefix', delegate_start, 'prefix-length', bad_prefix_len])
-        self.cli_set(pool + ['prefix-delegation', 'prefix', delegate_start, 'excluded-prefix', delegate_start])
-        self.cli_set(pool + ['prefix-delegation', 'prefix', delegate_start, 'excluded-prefix-length', exclude_len])
+        self.cli_set(pd_mapping_prefix + ['delegated-length', delegate_len])
+        self.cli_set(pd_mapping_prefix + ['prefix-length', bad_prefix_len])
+        self.cli_set(pd_mapping_prefix + ['excluded-prefix', delegate_start])
+        self.cli_set(pd_mapping_prefix + ['excluded-prefix-length', exclude_len])
 
         # commit changes
         with self.assertRaises(ConfigSessionError):
             self.cli_commit()
 
-        self.cli_set(pool + ['prefix-delegation', 'prefix', delegate_start, 'prefix-length', prefix_len])
+        self.cli_set(pd_mapping_prefix + ['prefix-length', prefix_len])
         self.cli_commit()
 
         config = read_file(KEA6_CONF)
         obj = loads(config)
 
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks'], 'name', shared_net_name)
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'subnet', subnet)
+        self.verify_config_value(obj, conf_path, 'name', shared_net_name)
+        self.verify_config_value(obj, conf_path + [0, 'subnet6'], 'subnet', subnet)
 
         # Verify pools
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'pools'],
-                {'pool': f'{range_start} - {range_stop}'})
+            obj,
+            conf_path + [0, 'subnet6', 0, 'pools'],
+            {'pool': f'{range_start} - {range_stop}'},
+        )
 
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'shared-networks', 0, 'subnet6', 0, 'pd-pools'],
-                {
-                    'prefix': delegate_start,
-                    'prefix-len': int(prefix_len),
-                    'delegated-len': int(delegate_len),
-                    'excluded-prefix': delegate_start,
-                    'excluded-prefix-len': int(exclude_len)
-                })
+            obj,
+            conf_path + [0, 'subnet6', 0, 'pd-pools'],
+            {
+                'prefix': delegate_start,
+                'prefix-len': int(prefix_len),
+                'delegated-len': int(delegate_len),
+                'excluded-prefix': delegate_start,
+                'excluded-prefix-len': int(exclude_len),
+            },
+        )
 
         # Check for running process
         self.assertTrue(process_named_running(PROCESS_NAME))
@@ -274,9 +330,12 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         ns_global_1 = '2001:db8::1111'
         ns_global_2 = '2001:db8::2222'
 
+        pool = base_path + ['shared-network-name', shared_net_name, 'subnet', subnet]
+        conf_path = ['Dhcp6', 'shared-networks']
+
         self.cli_set(base_path + ['global-parameters', 'name-server', ns_global_1])
         self.cli_set(base_path + ['global-parameters', 'name-server', ns_global_2])
-        self.cli_set(base_path + ['shared-network-name', shared_net_name, 'subnet', subnet, 'subnet-id', '1'])
+        self.cli_set(pool + ['subnet-id', '1'])
 
         # commit changes
         self.cli_commit()
@@ -284,17 +343,25 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         config = read_file(KEA6_CONF)
         obj = loads(config)
 
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks'], 'name', shared_net_name)
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'subnet', subnet)
-        self.verify_config_value(obj, ['Dhcp6', 'shared-networks', 0, 'subnet6'], 'id', 1)
+        self.verify_config_value(obj, conf_path, 'name', shared_net_name)
+        self.verify_config_value(obj, conf_path + [0, 'subnet6'], 'subnet', subnet)
+        self.verify_config_value(obj, conf_path + [0, 'subnet6'], 'id', 1)
 
         self.verify_config_object(
-                obj,
-                ['Dhcp6', 'option-data'],
-                {'name': 'dns-servers', "code": 23, "space": "dhcp6", "csv-format": True, 'data': f'{ns_global_1}, {ns_global_2}'})
+            obj,
+            ['Dhcp6', 'option-data'],
+            {
+                'name': 'dns-servers',
+                'code': 23,
+                'space': 'dhcp6',
+                'csv-format': True,
+                'data': f'{ns_global_1}, {ns_global_2}',
+            },
+        )
 
         # Check for running process
         self.assertTrue(process_named_running(PROCESS_NAME))
 
+
 if __name__ == '__main__':
     unittest.main(verbosity=2, failfast=VyOSUnitTestSHIM.TestCase.debug_on())