summaryrefslogtreecommitdiff
path: root/src/migration-scripts/interfaces/25-to-26
diff options
context:
space:
mode:
Diffstat (limited to 'src/migration-scripts/interfaces/25-to-26')
-rw-r--r--src/migration-scripts/interfaces/25-to-26329
1 files changed, 210 insertions, 119 deletions
diff --git a/src/migration-scripts/interfaces/25-to-26 b/src/migration-scripts/interfaces/25-to-26
index 7a4032d10..1f10e3dca 100644
--- a/src/migration-scripts/interfaces/25-to-26
+++ b/src/migration-scripts/interfaces/25-to-26
@@ -1,4 +1,4 @@
-# Copyright 2021-2024 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -29,6 +29,7 @@ from vyos.pki import encode_dh_parameters
from vyos.pki import encode_private_key
from vyos.pki import verify_crl
from vyos.utils.process import run
+from vyos.utils.file import read_file
def wrapped_pem_to_config_value(pem):
out = []
@@ -38,20 +39,28 @@ def wrapped_pem_to_config_value(pem):
out.append(line)
return "".join(out)
-def read_file_for_pki(config_auth_path):
- full_path = os.path.join(AUTH_DIR, config_auth_path)
- output = None
+def read_auth_file(config_auth_path):
+ full_path = os.path.normpath(os.path.join(AUTH_DIR, config_auth_path))
+
+ # If the file is not found under `/config/auth`, it may be because the `/config`
+ # partition has not been bind-mounted yet during early boot migration execution.
+ # Fall back to the equivalent path under `/opt/vyatta/etc/config/auth` which
+ # is accessible at all boot stages.
+ if not os.path.isfile(full_path) and full_path.startswith(f'{AUTH_DIR}/'):
+ full_path = AUTH_DIR_FALLBACK + full_path[len(AUTH_DIR): ]
if os.path.isfile(full_path):
if not os.access(full_path, os.R_OK):
- run(f'sudo chmod 644 {full_path}')
+ run(['sudo', 'chmod', '644', full_path])
+
+ return read_file(full_path)
- with open(full_path, 'r') as f:
- output = f.read()
+ return None
- return output
AUTH_DIR = '/config/auth'
+AUTH_DIR_FALLBACK = '/opt/vyatta/etc/config/auth'
+
pki_base = ['pki']
def migrate(config: ConfigTree) -> None:
@@ -69,13 +78,30 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['openvpn', 'shared-secret'])
key_file = config.return_value(base + [interface, 'shared-secret-key-file'])
- key = read_file_for_pki(key_file)
+ key = read_auth_file(key_file)
key_pki_name = f'{pki_name}_shared'
if key:
- config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'key'], value=wrapped_pem_to_config_value(key))
- config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'version'], value='1')
- config.set(base + [interface, 'shared-secret-key'], value=key_pki_name)
+ # Check if OpenVPN shared-secret already exists - no need to check node
+ # existence as it is always created above when entering this context
+ secret_exists = None
+ for secret_name in config.list_nodes(pki_base + ['openvpn', 'shared-secret']):
+ secret_path = pki_base + ['openvpn', 'shared-secret', secret_name, 'key']
+ if not config.exists(secret_path):
+ continue
+
+ secret = config.return_value(secret_path)
+ # Check for duplicate cert/key - and re-use if possible
+ if secret == wrapped_pem_to_config_value(key):
+ secret_exists = secret_name
+ break
+
+ if secret_exists:
+ config.set(base + [interface, 'shared-secret-key'], value=secret_exists)
+ else:
+ config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'key'], value=wrapped_pem_to_config_value(key))
+ config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'version'], value='1')
+ config.set(base + [interface, 'shared-secret-key'], value=key_pki_name)
else:
print(f'Failed to migrate shared-secret-key on openvpn interface {interface}')
@@ -90,13 +116,30 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['openvpn', 'shared-secret'])
key_file = config.return_value(base + [interface, 'tls', 'auth-file'])
- key = read_file_for_pki(key_file)
+ key = read_auth_file(key_file)
key_pki_name = f'{pki_name}_auth'
if key:
- config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'key'], value=wrapped_pem_to_config_value(key))
- config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'version'], value='1')
- config.set(base + [interface, 'tls', 'auth-key'], value=key_pki_name)
+ # Check if OpenVPN auth key already exists - no need to check node
+ # existence as it is always created above when entering this context
+ secret_exists = None
+ for secret_name in config.list_nodes(pki_base + ['openvpn', 'shared-secret']):
+ secret_path = pki_base + ['openvpn', 'shared-secret', secret_name, 'key']
+ if not config.exists(secret_path):
+ continue
+
+ secret = config.return_value(secret_path)
+ # Check for duplicate cert/key - and re-use if possible
+ if secret == wrapped_pem_to_config_value(key):
+ secret_exists = secret_name
+ break
+
+ if secret_exists:
+ config.set(base + [interface, 'tls', 'auth-key'], value=secret_exists)
+ else:
+ config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'key'], value=wrapped_pem_to_config_value(key))
+ config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'version'], value='1')
+ config.set(base + [interface, 'tls', 'auth-key'], value=key_pki_name)
else:
print(f'Failed to migrate auth-key on openvpn interface {interface}')
@@ -108,13 +151,30 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['openvpn', 'shared-secret'])
key_file = config.return_value(base + [interface, 'tls', 'crypt-file'])
- key = read_file_for_pki(key_file)
+ key = read_auth_file(key_file)
key_pki_name = f'{pki_name}_crypt'
if key:
- config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'key'], value=wrapped_pem_to_config_value(key))
- config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'version'], value='1')
- config.set(base + [interface, 'tls', 'crypt-key'], value=key_pki_name)
+ # Check if OpenVPN auth key already exists - no need to check node
+ # existence as it is always created above when entering this context
+ secret_exists = None
+ for secret_name in config.list_nodes(pki_base + ['openvpn', 'shared-secret']):
+ secret_path = pki_base + ['openvpn', 'shared-secret', secret_name, 'key']
+ if not config.exists(secret_path):
+ continue
+
+ secret = config.return_value(secret_path)
+ # Check for duplicate cert/key - and re-use if possible
+ if secret == wrapped_pem_to_config_value(key):
+ secret_exists = secret_name
+ break
+
+ if secret_exists:
+ config.set(base + [interface, 'tls', 'crypt-key'], value=secret_exists)
+ else:
+ config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'key'], value=wrapped_pem_to_config_value(key))
+ config.set(pki_base + ['openvpn', 'shared-secret', key_pki_name, 'version'], value='1')
+ config.set(base + [interface, 'tls', 'crypt-key'], value=key_pki_name)
else:
print(f'Failed to migrate crypt-key on openvpn interface {interface}')
@@ -128,28 +188,41 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['ca'])
cert_file = config.return_value(x509_base + ['ca-cert-file'])
- cert_path = os.path.join(AUTH_DIR, cert_file)
-
- if os.path.isfile(cert_path):
- if not os.access(cert_path, os.R_OK):
- run(f'sudo chmod 644 {cert_path}')
-
- with open(cert_path, 'r') as f:
- certs_str = f.read()
- certs_data = certs_str.split(CERT_BEGIN)
- index = 1
- for cert_data in certs_data[1:]:
- cert = load_certificate(CERT_BEGIN + cert_data, wrap_tags=False)
-
- if cert:
- ca_certs[f'{pki_name}_{index}'] = cert
- cert_pem = encode_certificate(cert)
+ certs_str = read_auth_file(cert_file)
+
+ if certs_str:
+ certs_data = certs_str.split(CERT_BEGIN)
+ index = 1
+ for cert_data in certs_data[1:]:
+ cert = load_certificate(CERT_BEGIN + cert_data, wrap_tags=False)
+
+ if cert:
+ ca_certs[f'{pki_name}_{index}'] = cert
+ cert_pem = encode_certificate(cert)
+
+ # Check if CA already exists - no need to check node existence as
+ # it is always created above when entering this context
+ ca_exists = None
+ for ca_name in config.list_nodes(pki_base + ['ca']):
+ ca_cert_path = pki_base + ['ca', ca_name, 'certificate']
+ if not config.exists(ca_cert_path):
+ continue
+
+ ca_base64 = config.return_value(ca_cert_path)
+ # Check for duplicate cert/key - and re-use if possible
+ if ca_base64 == wrapped_pem_to_config_value(cert_pem):
+ ca_exists = ca_name
+ break
+
+ if ca_exists:
+ config.set(x509_base + ['ca-certificate'], value=ca_exists, replace=False)
+ else:
config.set(pki_base + ['ca', f'{pki_name}_{index}', 'certificate'], value=wrapped_pem_to_config_value(cert_pem))
config.set(x509_base + ['ca-certificate'], value=f'{pki_name}_{index}', replace=False)
- else:
- print(f'Failed to migrate CA certificate on openvpn interface {interface}')
+ else:
+ print(f'Failed to migrate CA certificate on openvpn interface {interface}')
- index += 1
+ index += 1
else:
print(f'Failed to migrate CA certificate on openvpn interface {interface}')
@@ -161,26 +234,36 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['ca'])
crl_file = config.return_value(x509_base + ['crl-file'])
- crl_path = os.path.join(AUTH_DIR, crl_file)
- crl = None
- crl_ca_name = None
-
- if os.path.isfile(crl_path):
- if not os.access(crl_path, os.R_OK):
- run(f'sudo chmod 644 {crl_path}')
+ crl_data = read_auth_file(crl_file)
- with open(crl_path, 'r') as f:
- crl_data = f.read()
- crl = load_crl(crl_data, wrap_tags=False)
+ crl = load_crl(crl_data, wrap_tags=False) if crl_data else None
+ crl_ca_name = None
- for ca_name, ca_cert in ca_certs.items():
- if verify_crl(crl, ca_cert):
- crl_ca_name = ca_name
- break
+ if crl:
+ for ca_name, ca_cert in ca_certs.items():
+ if verify_crl(crl, ca_cert):
+ crl_ca_name = ca_name
+ break
- if crl and crl_ca_name:
+ if crl_ca_name:
crl_pem = encode_certificate(crl)
- config.set(pki_base + ['ca', crl_ca_name, 'crl'], value=wrapped_pem_to_config_value(crl_pem))
+
+ # Check if CRL already exists - no need to check node
+ # existence as it is always created above when entering this context
+ crl_exists = None
+ for ca_name in config.list_nodes(pki_base + ['ca']):
+ crl_path = pki_base + ['ca', ca_name, 'crl']
+ if not config.exists(crl_path):
+ continue
+
+ crl_base64 = config.return_value(crl_path)
+ # Check if CRL is a duplicate and we have already imported it
+ if crl_base64 == wrapped_pem_to_config_value(crl_pem):
+ crl_exists = ca_name
+ break
+
+ if not crl_exists:
+ config.set(pki_base + ['ca', crl_ca_name, 'crl'], value=wrapped_pem_to_config_value(crl_pem))
else:
print(f'Failed to migrate CRL on openvpn interface {interface}')
@@ -192,21 +275,31 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['certificate'])
cert_file = config.return_value(x509_base + ['cert-file'])
- cert_path = os.path.join(AUTH_DIR, cert_file)
- cert = None
-
- if os.path.isfile(cert_path):
- if not os.access(cert_path, os.R_OK):
- run(f'sudo chmod 644 {cert_path}')
+ cert_data = read_auth_file(cert_file)
- with open(cert_path, 'r') as f:
- cert_data = f.read()
- cert = load_certificate(cert_data, wrap_tags=False)
+ cert = load_certificate(cert_data, wrap_tags=False) if cert_data else None
if cert:
cert_pem = encode_certificate(cert)
- config.set(pki_base + ['certificate', pki_name, 'certificate'], value=wrapped_pem_to_config_value(cert_pem))
- config.set(x509_base + ['certificate'], value=pki_name)
+ # Check if certificate public key already exists - no need to check node
+ # existence as it is always created above when entering this context
+ cert_exists = None
+ for cert_name in config.list_nodes(pki_base + ['certificate']):
+ cert_path = pki_base + ['certificate', cert_name, 'certificate']
+ if not config.exists(cert_path):
+ continue
+
+ cert_base64 = config.return_value(cert_path)
+ # Check for duplicate cert/key - and re-use if possible
+ if cert_base64 == wrapped_pem_to_config_value(cert_pem):
+ cert_exists = cert_name
+ break
+
+ if cert_exists:
+ config.set(x509_base + ['certificate'], value=cert_exists)
+ else:
+ config.set(pki_base + ['certificate', pki_name, 'certificate'], value=wrapped_pem_to_config_value(cert_pem))
+ config.set(x509_base + ['certificate'], value=pki_name)
else:
print(f'Failed to migrate certificate on openvpn interface {interface}')
@@ -214,20 +307,28 @@ def migrate(config: ConfigTree) -> None:
if config.exists(x509_base + ['key-file']):
key_file = config.return_value(x509_base + ['key-file'])
- key_path = os.path.join(AUTH_DIR, key_file)
- key = None
-
- if os.path.isfile(key_path):
- if not os.access(key_path, os.R_OK):
- run(f'sudo chmod 644 {key_path}')
+ key_data = read_auth_file(key_file)
- with open(key_path, 'r') as f:
- key_data = f.read()
- key = load_private_key(key_data, passphrase=None, wrap_tags=False)
+ key = load_private_key(key_data, passphrase=None, wrap_tags=False) if key_data else None
if key:
key_pem = encode_private_key(key, passphrase=None)
- config.set(pki_base + ['certificate', pki_name, 'private', 'key'], value=wrapped_pem_to_config_value(key_pem))
+ # Check if certificate public key already exists - no need to check node
+ # existence as it is always created above when entering this context
+ key_exists = None
+ for key_name in config.list_nodes(pki_base + ['certificate']):
+ key_path = pki_base + ['certificate', key_name, 'private', 'key']
+ if not config.exists(key_path):
+ continue
+
+ key_base64 = config.return_value(key_path)
+ # Check for duplicate cert/key - and re-use if possible
+ if key_base64 == wrapped_pem_to_config_value(key_pem):
+ key_exists = key_name
+ break
+
+ if not key_exists:
+ config.set(pki_base + ['certificate', pki_name, 'private', 'key'], value=wrapped_pem_to_config_value(key_pem))
else:
print(f'Failed to migrate private key on openvpn interface {interface}')
@@ -239,21 +340,32 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['dh'])
dh_file = config.return_value(x509_base + ['dh-file'])
- dh_path = os.path.join(AUTH_DIR, dh_file)
- dh = None
-
- if os.path.isfile(dh_path):
- if not os.access(dh_path, os.R_OK):
- run(f'sudo chmod 644 {dh_path}')
+ dh_data = read_auth_file(dh_file)
- with open(dh_path, 'r') as f:
- dh_data = f.read()
- dh = load_dh_parameters(dh_data, wrap_tags=False)
+ dh = load_dh_parameters(dh_data, wrap_tags=False) if dh_data else None
if dh:
dh_pem = encode_dh_parameters(dh)
- config.set(pki_base + ['dh', pki_name, 'parameters'], value=wrapped_pem_to_config_value(dh_pem))
- config.set(x509_base + ['dh-params'], value=pki_name)
+
+ # Check if DH parameters already exists - no need to check node existence
+ # as it is always created above when entering this context
+ dh_exists = None
+ for dh_name in config.list_nodes(pki_base + ['dh']):
+ dh_param_path = pki_base + ['dh', dh_name, 'parameters']
+ if not config.exists(dh_param_path):
+ continue
+
+ dh_base64 = config.return_value(dh_param_path)
+ # Check for duplicate cert/key - and re-use if possible
+ if dh_base64 == wrapped_pem_to_config_value(dh_pem):
+ dh_exists = dh_name
+ break
+
+ if dh_exists:
+ config.set(x509_base + ['dh-params'], value=dh_exists)
+ else:
+ config.set(pki_base + ['dh', pki_name, 'parameters'], value=wrapped_pem_to_config_value(dh_pem))
+ config.set(x509_base + ['dh-params'], value=pki_name)
else:
print(f'Failed to migrate DH parameters on openvpn interface {interface}')
@@ -270,15 +382,15 @@ def migrate(config: ConfigTree) -> None:
if config.exists(private_key_path):
key_file = config.return_value(private_key_path)
- full_key_path = f'/config/auth/wireguard/{key_file}/private.key'
+ full_key_path = f'{AUTH_DIR}/wireguard/{key_file}/private.key'
+ key_data = read_auth_file(full_key_path)
- if not os.path.exists(full_key_path):
+ if not key_data:
print(f'Could not find wireguard private key for migration on interface "{interface}"')
continue
- with open(full_key_path, 'r') as f:
- key_data = f.read().strip()
- config.set(private_key_path, value=key_data)
+ key_data = key_data.strip()
+ config.set(private_key_path, value=key_data)
for peer in config.list_nodes(base + [interface, 'peer']):
config.rename(base + [interface, 'peer', peer, 'pubkey'], 'public-key')
@@ -300,16 +412,9 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['ca'])
cert_file = config.return_value(x509_base + ['ca-cert-file'])
- cert_path = os.path.join(AUTH_DIR, cert_file)
- cert = None
-
- if os.path.isfile(cert_path):
- if not os.access(cert_path, os.R_OK):
- run(f'sudo chmod 644 {cert_path}')
+ cert_data = read_auth_file(cert_file)
- with open(cert_path, 'r') as f:
- cert_data = f.read()
- cert = load_certificate(cert_data, wrap_tags=False)
+ cert = load_certificate(cert_data, wrap_tags=False) if cert_data else None
if cert:
cert_pem = encode_certificate(cert)
@@ -326,16 +431,9 @@ def migrate(config: ConfigTree) -> None:
config.set_tag(pki_base + ['certificate'])
cert_file = config.return_value(x509_base + ['cert-file'])
- cert_path = os.path.join(AUTH_DIR, cert_file)
- cert = None
+ cert_data = read_auth_file(cert_file)
- if os.path.isfile(cert_path):
- if not os.access(cert_path, os.R_OK):
- run(f'sudo chmod 644 {cert_path}')
-
- with open(cert_path, 'r') as f:
- cert_data = f.read()
- cert = load_certificate(cert_data, wrap_tags=False)
+ cert = load_certificate(cert_data, wrap_tags=False) if cert_data else None
if cert:
cert_pem = encode_certificate(cert)
@@ -348,16 +446,9 @@ def migrate(config: ConfigTree) -> None:
if config.exists(x509_base + ['key-file']):
key_file = config.return_value(x509_base + ['key-file'])
- key_path = os.path.join(AUTH_DIR, key_file)
- key = None
-
- if os.path.isfile(key_path):
- if not os.access(key_path, os.R_OK):
- run(f'sudo chmod 644 {key_path}')
+ key_data = read_auth_file(key_file)
- with open(key_path, 'r') as f:
- key_data = f.read()
- key = load_private_key(key_data, passphrase=None, wrap_tags=False)
+ key = load_private_key(key_data, passphrase=None, wrap_tags=False) if key_data else None
if key:
key_pem = encode_private_key(key, passphrase=None)
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-22 11:29:28 +0000