pki: T3642: verify() that we can not delete certificates still referenced in CLI

1 files changed, 41 insertions, 5 deletions

diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
index ef1b57650..efa3578b4 100755
--- a/src/conf_mode/pki.py
+++ b/src/conf_mode/pki.py
@@ -16,8 +16,11 @@
 
 from sys import exit
 
+import jmespath
+
 from vyos.config import Config
 from vyos.configdict import dict_merge
+from vyos.configdict import node_changed
 from vyos.pki import is_ca_certificate
 from vyos.pki import load_certificate
 from vyos.pki import load_certificate_request
@@ -26,6 +29,7 @@ from vyos.pki import load_private_key
 from vyos.pki import load_crl
 from vyos.pki import load_dh_parameters
 from vyos.util import ask_input
+from vyos.util import dict_search_recursive
 from vyos.xml import defaults
 from vyos import ConfigError
 from vyos import airbag
@@ -37,14 +41,29 @@ def get_config(config=None):
     else:
         conf = Config()
     base = ['pki']
-    if not conf.exists(base):
-        return None
 
     pki = conf.get_config_dict(base, key_mangling=('-', '_'),
-                                     get_first_key=True, no_tag_node_value_mangle=True)
+                                     get_first_key=True,
+                                     no_tag_node_value_mangle=True)
+
+    pki['changed'] = {}
+    tmp = node_changed(conf, base + ['ca'], key_mangling=('-', '_'))
+    if tmp: pki['changed'].update({'ca' : tmp})
+
+    tmp = node_changed(conf, base + ['certificate'], key_mangling=('-', '_'))
+    if tmp: pki['changed'].update({'certificate' : tmp})
+
+    # We only merge on the defaults of there is a configuration at all
+    if conf.exists(base):
+        default_values = defaults(base)
+        pki = dict_merge(default_values, pki)
+
+    # We need to get the entire system configuration to verify that we are not
+    # deleting a certificate that is still referenced somewhere!
+    pki['system'] = conf.get_config_dict([], key_mangling=('-', '_'),
+                                         get_first_key=True,
+                                         no_tag_node_value_mangle=True)
 
-    default_values = defaults(base)
-    pki = dict_merge(default_values, pki)
     return pki
 
 def is_valid_certificate(raw_data):
@@ -142,6 +161,21 @@ def verify(pki):
                 if len(country) != 2 or not country.isalpha():
                     raise ConfigError(f'Invalid default country value. Value must be 2 alpha characters.')
 
+    if 'changed' in pki:
+        # if the list is getting longer, we can move to a dict() and also embed the
+        # search key as value from line 173 or 176
+        for cert_type in ['ca', 'certificate']:
+            if not cert_type in pki['changed']:
+                continue
+            for certificate in pki['changed'][cert_type]:
+                if cert_type not in pki or certificate not in pki['changed'][cert_type]:
+                    if cert_type == 'ca':
+                        if certificate in dict_search_recursive(pki['system'], 'ca_certificate'):
+                            raise ConfigError(f'CA certificate "{certificate}" is still in use!')
+                    elif cert_type == 'certificate':
+                        if certificate in dict_search_recursive(pki['system'], 'certificate'):
+                            raise ConfigError(f'Certificate "{certificate}" is still in use!')
+
     return None
 
 def generate(pki):
@@ -154,6 +188,8 @@ def apply(pki):
     if not pki:
         return None
 
+    # XXX: restart services if the content of a certificate changes
+
     return None
 
 if __name__ == '__main__':