ipsec: T5606: T5871: Use multi node for CA certificates

This changes behaviour from fetching CA chain in PKI, to the user manually setting CA certificates.

Prevents unwanted parent CAs existing in PKI from being auto-included as may not be desired/intended.

3 files changed, 20 insertions, 30 deletions

diff --git a/interface-definitions/include/ipsec/authentication-x509.xml.i b/interface-definitions/include/ipsec/authentication-x509.xml.i
index db675c0bf..1d04c94ba 100644
--- a/interface-definitions/include/ipsec/authentication-x509.xml.i
+++ b/interface-definitions/include/ipsec/authentication-x509.xml.i
@@ -5,7 +5,7 @@
   </properties>
   <children>
     #include <include/pki/certificate-key.xml.i>
-    #include <include/pki/ca-certificate.xml.i>
+    #include <include/pki/ca-certificate-multi.xml.i>
   </children>
 </node>
 <!-- include end -->
diff --git a/smoketest/scripts/cli/test_vpn_ipsec.py b/smoketest/scripts/cli/test_vpn_ipsec.py
index 09e10a2c4..884394bac 100755
--- a/smoketest/scripts/cli/test_vpn_ipsec.py
+++ b/smoketest/scripts/cli/test_vpn_ipsec.py
@@ -413,6 +413,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase):
         self.cli_set(peer_base_path + ['authentication', 'local-id', peer_name])
         self.cli_set(peer_base_path + ['authentication', 'mode', 'x509'])
         self.cli_set(peer_base_path + ['authentication', 'remote-id', 'peer2'])
+        self.cli_set(peer_base_path + ['authentication', 'x509', 'ca-certificate', ca_name])
         self.cli_set(peer_base_path + ['authentication', 'x509', 'ca-certificate', int_ca_name])
         self.cli_set(peer_base_path + ['authentication', 'x509', 'certificate', peer_name])
         self.cli_set(peer_base_path + ['connection-type', 'initiate'])
@@ -465,8 +466,8 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase):
             self.assertIn(line, swanctl_conf)
 
         # Check Root CA, Intermediate CA and Peer cert/key pair is present
-        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{int_ca_name}_1.pem')))
-        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{int_ca_name}_2.pem')))
+        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}.pem')))
+        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{int_ca_name}.pem')))
         self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
 
         # There is only one VTI test so no need to delete this globally in tearDown()
@@ -666,7 +667,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase):
             self.assertIn(line, swanctl_conf)
 
         # Check Root CA, Intermediate CA and Peer cert/key pair is present
-        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}_1.pem')))
+        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}.pem')))
         self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
 
         self.tearDownPKI()
@@ -778,7 +779,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase):
             self.assertIn(line, swanctl_conf)
 
         # Check Root CA, Intermediate CA and Peer cert/key pair is present
-        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}_1.pem')))
+        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}.pem')))
         self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
 
         self.tearDownPKI()
@@ -893,7 +894,7 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase):
             self.assertIn(line, swanctl_conf)
 
         # Check Root CA, Intermediate CA and Peer cert/key pair is present
-        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}_1.pem')))
+        self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}.pem')))
         self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
 
         self.tearDownPKI()
diff --git a/src/conf_mode/vpn_ipsec.py b/src/conf_mode/vpn_ipsec.py
index 388f2a709..46f041cce 100755
--- a/src/conf_mode/vpn_ipsec.py
+++ b/src/conf_mode/vpn_ipsec.py
@@ -32,10 +32,7 @@ from vyos.configverify import verify_interface_exists
 from vyos.configverify import dynamic_interface_pattern
 from vyos.defaults import directories
 from vyos.ifconfig import Interface
-from vyos.pki import encode_certificate
 from vyos.pki import encode_public_key
-from vyos.pki import find_chain
-from vyos.pki import load_certificate
 from vyos.pki import load_private_key
 from vyos.pki import wrap_certificate
 from vyos.pki import wrap_crl
@@ -126,11 +123,11 @@ def verify_pki_x509(pki, x509_conf):
     if not pki or 'ca' not in pki or 'certificate' not in pki:
         raise ConfigError(f'PKI is not configured')
 
-    ca_cert_name = x509_conf['ca_certificate']
     cert_name = x509_conf['certificate']
 
-    if not dict_search_args(pki, 'ca', ca_cert_name, 'certificate'):
-        raise ConfigError(f'Missing CA certificate on specified PKI CA certificate "{ca_cert_name}"')
+    for ca_cert_name in x509_conf['ca_certificate']:
+        if not dict_search_args(pki, 'ca', ca_cert_name, 'certificate'):
+            raise ConfigError(f'Missing CA certificate on specified PKI CA certificate "{ca_cert_name}"')
 
     if not dict_search_args(pki, 'certificate', cert_name, 'certificate'):
         raise ConfigError(f'Missing certificate on specified PKI certificate "{cert_name}"')
@@ -443,32 +440,24 @@ def cleanup_pki_files():
                 os.unlink(file_path)
 
 def generate_pki_files_x509(pki, x509_conf):
-    ca_cert_name = x509_conf['ca_certificate']
-    ca_cert_data = dict_search_args(pki, 'ca', ca_cert_name, 'certificate')
-    ca_cert_crls = dict_search_args(pki, 'ca', ca_cert_name, 'crl') or []
-    ca_index = 1
-    crl_index = 1
+    for ca_cert_name in x509_conf['ca_certificate']:
+        ca_cert_data = dict_search_args(pki, 'ca', ca_cert_name, 'certificate')
+        ca_cert_crls = dict_search_args(pki, 'ca', ca_cert_name, 'crl') or []
+        crl_index = 1
 
-    ca_cert = load_certificate(ca_cert_data)
-    pki_ca_certs = [load_certificate(ca['certificate']) for ca in pki['ca'].values()]
+        with open(os.path.join(CA_PATH, f'{ca_cert_name}.pem'), 'w') as f:
+            f.write(wrap_certificate(ca_cert_data))
 
-    ca_cert_chain = find_chain(ca_cert, pki_ca_certs)
+        for crl in ca_cert_crls:
+            with open(os.path.join(CRL_PATH, f'{ca_cert_name}_{crl_index}.pem'), 'w') as f:
+                f.write(wrap_crl(crl))
+            crl_index += 1
 
     cert_name = x509_conf['certificate']
     cert_data = dict_search_args(pki, 'certificate', cert_name, 'certificate')
     key_data = dict_search_args(pki, 'certificate', cert_name, 'private', 'key')
     protected = 'passphrase' in x509_conf
 
-    for ca_cert_obj in ca_cert_chain:
-        with open(os.path.join(CA_PATH, f'{ca_cert_name}_{ca_index}.pem'), 'w') as f:
-            f.write(encode_certificate(ca_cert_obj))
-        ca_index += 1
-
-    for crl in ca_cert_crls:
-        with open(os.path.join(CRL_PATH, f'{ca_cert_name}_{crl_index}.pem'), 'w') as f:
-            f.write(wrap_crl(crl))
-        crl_index += 1
-
     with open(os.path.join(CERT_PATH, f'{cert_name}.pem'), 'w') as f:
         f.write(wrap_certificate(cert_data))