kea: T7925: Improve error handling, validate IPv6 PD prefix length

4 files changed, 35 insertions, 6 deletions

diff --git a/python/vyos/kea.py b/python/vyos/kea.py
index 5ce6f47f1..f48cf293d 100644
--- a/python/vyos/kea.py
+++ b/python/vyos/kea.py
@@ -15,6 +15,7 @@
 
 import json
 import os
+import re
 import socket
 
 from datetime import datetime
@@ -25,7 +26,7 @@ from vyos.template import is_ipv6
 from vyos.template import netmask_from_cidr
 from vyos.utils.dict import dict_search_args
 from vyos.utils.file import file_permissions
-from vyos.utils.process import run
+from vyos.utils.process import run, rc_cmd
 
 kea4_options = {
     'name_server': 'domain-name-servers',
@@ -86,6 +87,14 @@ def _find_list_of_dict_index(lst, key='ip', value=''):
     idx = next((index for (index, d) in enumerate(lst) if d[key] == value), None)
     return idx
 
+def kea_test_config(process: str, config_path: str) -> tuple[bool, str]:
+    result, output = rc_cmd(f'{process} -t {config_path}')
+
+    if result == 0:
+        return (True, None)
+
+    find = re.search(r'Error encountered:\s([^\n$]+)', output)
+    return (False, find[1] if find else None)
 
 def kea_parse_options(config):
     options = []
diff --git a/smoketest/scripts/cli/test_service_dhcpv6-server.py b/smoketest/scripts/cli/test_service_dhcpv6-server.py
index 5e0dd7cc2..769c799b2 100755
--- a/smoketest/scripts/cli/test_service_dhcpv6-server.py
+++ b/smoketest/scripts/cli/test_service_dhcpv6-server.py
@@ -221,6 +221,7 @@ class TestServiceDHCPv6Server(VyOSUnitTestSHIM.TestCase):
         range_stop = inc_ip(subnet, 65535) # ::ffff
         delegate_start = '2001:db8:ee::'
         delegate_len = '64'
+        bad_prefix_len = '32'
         prefix_len = '56'
         exclude_len = '66'
 
@@ -229,11 +230,15 @@ 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 + ['prefix-delegation', 'prefix', delegate_start, 'delegated-length', delegate_len])
-        self.cli_set(pool + ['prefix-delegation', 'prefix', delegate_start, 'prefix-length', prefix_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])
 
         # commit changes
+        with self.assertRaises(ConfigSessionError):
+            self.cli_commit()
+
+        self.cli_set(pool + ['prefix-delegation', 'prefix', delegate_start, 'prefix-length', prefix_len])
         self.cli_commit()
 
         config = read_file(KEA6_CONF)
diff --git a/src/conf_mode/service_dhcp-server.py b/src/conf_mode/service_dhcp-server.py
index d92321ab8..e4730f8a8 100755
--- a/src/conf_mode/service_dhcp-server.py
+++ b/src/conf_mode/service_dhcp-server.py
@@ -25,6 +25,7 @@ from ipaddress import ip_network
 from netaddr import IPRange
 
 from vyos.config import Config
+from vyos.kea import kea_test_config
 from vyos.pki import wrap_certificate
 from vyos.pki import wrap_private_key
 from vyos.template import render
@@ -595,6 +596,10 @@ def apply(dhcp):
 
         return None
 
+    result, output = kea_test_config('kea-dhcp4', config_file)
+    if not result:
+        raise ConfigError(f'Unexpected error with Kea configuration:\n{output}')
+
     for service in services:
         action = 'restart'
 
diff --git a/src/conf_mode/service_dhcpv6-server.py b/src/conf_mode/service_dhcpv6-server.py
index b6a7a87c3..d7b2f2a6d 100755
--- a/src/conf_mode/service_dhcpv6-server.py
+++ b/src/conf_mode/service_dhcpv6-server.py
@@ -24,6 +24,7 @@ from ipaddress import ip_address
 from ipaddress import ip_network
 
 from vyos.config import Config
+from vyos.kea import kea_test_config
 from vyos.template import render
 from vyos.utils.process import call
 from vyos.utils.file import chmod_775
@@ -191,16 +192,21 @@ def verify(dhcpv6):
                     if 'prefix_length' not in prefix_config:
                         raise ConfigError('Length of delegated IPv6 prefix must be configured')
 
-                    if prefix_config['prefix_length'] > prefix_config['delegated_length']:
+                    prefix_len = prefix_config['prefix_length']
+                    prefix_obj = None
+
+                    if prefix_len > prefix_config['delegated_length']:
                         raise ConfigError('Length of delegated IPv6 prefix must be within parent prefix')
 
+                    try:
+                        prefix_obj = ip_network(f'{prefix}/{prefix_len}')
+                    except ValueError:
+                        raise ConfigError('Invalid prefix-length for delegated prefix')
+
                     if 'excluded_prefix' in prefix_config:
                         if 'excluded_prefix_length' not in prefix_config:
                             raise ConfigError('Length of excluded IPv6 prefix must be configured')
 
-                        prefix_len = prefix_config['prefix_length']
-                        prefix_obj = ip_network(f'{prefix}/{prefix_len}')
-
                         excluded_prefix = prefix_config['excluded_prefix']
                         excluded_len = prefix_config['excluded_prefix_length']
                         excluded_obj = ip_network(f'{excluded_prefix}/{excluded_len}')
@@ -301,6 +307,10 @@ def apply(dhcpv6):
             os.unlink(config_file)
         return None
 
+    result, output = kea_test_config('kea-dhcp6', config_file)
+    if not result:
+        raise ConfigError(f'Unexpected error with Kea configuration:\n{output}')
+
     call(f'systemctl restart {service_name}')
 
     return None