Merge pull request #3205 from vyos/mergify/bp/sagitta/pr-3202

ipsec: T5606: T5871: Use multi node for CA certificates (backport #3202)

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))