No OneTemporary
Actions

Size

48 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/python/vyos/defaults.py b/python/vyos/defaults.py
	index 25d7fd50c..833ac1fbf 100644
	--- a/python/vyos/defaults.py
	+++ b/python/vyos/defaults.py
	@@ -1,100 +1,101 @@
	# 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
	# License as published by the Free Software Foundation; either
	# version 2.1 of the License, or (at your option) any later version.
	#
	# This library is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	# Lesser General Public License for more details.
	#
	# You should have received a copy of the GNU Lesser General Public
	# License along with this library. If not, see <http://www.gnu.org/licenses/>.

	import os

	base_dir = '/usr/libexec/vyos'

	directories = {
	'base' : f'{base_dir}',
	'data' : '/usr/share/vyos/',
	'conf_mode' : f'{base_dir}/conf_mode',
	'op_mode' : f'{base_dir}/op_mode',
	'services' : f'{base_dir}/services',
	'config' : '/opt/vyatta/etc/config',
	'migrate' : '/opt/vyatta/etc/config-migrate/migrate',
	'activate' : f'{base_dir}/activate',
	'log' : '/var/log/vyatta',
	'templates' : '/usr/share/vyos/templates/',
	'certbot' : '/config/auth/letsencrypt',
	'api_schema': f'{base_dir}/services/api/graphql/graphql/schema/',
	'api_client_op': f'{base_dir}/services/api/graphql/graphql/client_op/',
	'api_templates': f'{base_dir}/services/api/graphql/session/templates/',
	'vyos_udev_dir' : '/run/udev/vyos',
	'isc_dhclient_dir' : '/run/dhclient',
	'dhcp6_client_dir' : '/run/dhcp6c',
	'vyos_configdir' : '/opt/vyatta/config',
	'completion_dir' : f'{base_dir}/completion',
	'ca_certificates' : '/usr/local/share/ca-certificates/vyos',
	'podman_storage' : '/usr/lib/live/mount/persistence/container/storage',
	'ppp_nexthop_dir' : '/run/ppp_nexthop',
	'proto_path' : '/usr/share/vyos/vyconf',
	'vyconf_session_dir' : f'{base_dir}/vyconf/session'
	}

	systemd_services = {
	'haproxy' : 'haproxy.service',
	+ 'openconnect': 'ocserv.service',
	'syslog' : 'syslog.service',
	'snmpd' : 'snmpd.service',
	}

	internal_ports = {
	'certbot_haproxy' : 65080, # Certbot running behing haproxy
	}

	config_files = {
	'sshd_user_ca' : '/run/sshd/trusted_user_ca',
	}

	config_status = '/tmp/vyos-config-status'
	api_config_state = '/run/http-api-state'
	frr_debug_enable = '/tmp/vyos.frr.debug'
	static_route_dhcp_interfaces_path = '/tmp/static_dhcp_interfaces'
	vyos_configd_socket_path = 'ipc:///run/vyos-configd.sock'

	cfg_group = 'vyattacfg'

	cfg_vintage = 'vyos'

	commit_lock = os.path.join(directories['vyos_configdir'], '.lock')

	component_version_json = os.path.join(directories['data'], 'component-versions.json')

	config_default = os.path.join(directories['data'], 'config.boot.default')

	rt_symbolic_names = {
	# Standard routing tables for Linux & reserved IDs for VyOS
	'default': 253, # Confusingly, a final fallthru, not the default.
	'main': 254, # The actual global table used by iproute2 unless told otherwise.
	'local': 255, # Special kernel loopback table.
	}

	rt_global_vrf = rt_symbolic_names['main']
	rt_global_table = rt_symbolic_names['main']

	vyconfd_conf = '/etc/vyos/vyconfd.conf'

	DEFAULT_COMMIT_CONFIRM_MINUTES = 10

	commit_hooks = {'pre': '/etc/commit/pre-hooks.d',
	'post': '/etc/commit/post-hooks.d'
	}

	airbag_noteworthy_size = 20

	SSH_DSA_DEPRECATION_WARNING: str = \
	'Support for SSH-DSA keys is deprecated and will be removed in VyOS 1.6. ' \
	'Please update affected keys to a supported algorithm (e.g., RSA, ECDSA or ' \
	'ED25519) to avoid authentication failures after the upgrade.'
	diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
	index 98e5840a9..bf44dcfc7 100755
	--- a/src/conf_mode/pki.py
	+++ b/src/conf_mode/pki.py
	@@ -1,607 +1,619 @@
	#!/usr/bin/env python3
	#
	# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
	#
	# This program is free software; you can redistribute it and/or modify
	# it under the terms of the GNU General Public License version 2 or later as
	# published by the Free Software Foundation.
	#
	# This program is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	# GNU General Public License for more details.
	#
	# You should have received a copy of the GNU General Public License
	# along with this program. If not, see <http://www.gnu.org/licenses/>.

	import os

	from sys import argv
	from sys import exit

	from vyos.base import Message
	from vyos.config import Config
	from vyos.config import config_dict_merge
	from vyos.configdep import set_dependents
	from vyos.configdep import call_dependents
	from vyos.configdict import node_changed
	from vyos.configdiff import Diff
	from vyos.configdiff import get_config_diff
	from vyos.defaults import directories
	from vyos.defaults import internal_ports
	from vyos.defaults import systemd_services
	from vyos.pki import encode_certificate
	from vyos.pki import is_ca_certificate
	from vyos.pki import load_certificate
	from vyos.pki import load_public_key
	from vyos.pki import load_openssh_public_key
	from vyos.pki import load_openssh_private_key
	from vyos.pki import load_private_key
	from vyos.pki import load_crl
	from vyos.pki import load_dh_parameters
	from vyos.utils.boot import boot_configuration_complete
	from vyos.utils.configfs import add_cli_node
	from vyos.utils.dict import dict_search
	from vyos.utils.dict import dict_search_args
	from vyos.utils.dict import dict_search_recursive
	+from vyos.utils.dict import dict_set_nested
	from vyos.utils.file import read_file
	from vyos.utils.network import check_port_availability
	from vyos.utils.process import call
	from vyos.utils.process import cmd
	from vyos.utils.process import is_systemd_service_active
	from vyos.utils.process import is_systemd_service_running
	from vyos import ConfigError
	from vyos import airbag
	airbag.enable()

	vyos_certbot_dir = directories['certbot']
	vyos_ca_certificates_dir = directories['ca_certificates']

	# keys to recursively search for under specified path
	sync_search = [
	{
	'keys': ['certificate'],
	'path': ['service', 'https'],
	},
	{
	'keys': ['key'],
	'path': ['service', 'ssh'],
	},
	{
	'keys': ['certificate', 'ca_certificate'],
	'path': ['interfaces', 'ethernet'],
	},
	{
	'keys': ['certificate', 'ca_certificate', 'dh_params', 'shared_secret_key', 'auth_key', 'crypt_key'],
	'path': ['interfaces', 'openvpn'],
	},
	{
	'keys': ['ca_certificate'],
	'path': ['interfaces', 'sstpc'],
	},
	{
	'keys': ['certificate', 'ca_certificate'],
	'path': ['load_balancing', 'haproxy'],
	},
	{
	'keys': ['key'],
	'path': ['protocols', 'rpki', 'cache'],
	},
	{
	'keys': ['certificate', 'ca_certificate', 'local_key', 'remote_key'],
	'path': ['vpn', 'ipsec'],
	},
	{
	'keys': ['certificate', 'ca_certificate'],
	'path': ['vpn', 'openconnect'],
	},
	{
	'keys': ['certificate', 'ca_certificate'],
	'path': ['vpn', 'sstp'],
	},
	{
	'keys': ['certificate', 'ca_certificate'],
	'path': ['service', 'stunnel'],
	}
	]

	# key from other config nodes -> key in pki['changed'] and pki
	sync_translate = {
	'certificate': 'certificate',
	'ca_certificate': 'ca',
	'dh_params': 'dh',
	'local_key': 'key_pair',
	'remote_key': 'key_pair',
	'shared_secret_key': 'openvpn',
	'auth_key': 'openvpn',
	'crypt_key': 'openvpn',
	'key': 'openssh',
	}

	def certbot_delete(certificate):
	if not boot_configuration_complete():
	return
	if os.path.exists(f'{vyos_certbot_dir}/renewal/{certificate}.conf'):
	cmd(f'certbot delete --non-interactive --config-dir {vyos_certbot_dir} --cert-name {certificate}')

	-def certbot_request(name: str, config: dict, dry_run: bool=True):
	+def certbot_request(name: str, config: dict, dry_run: bool=True) -> None:
	# We do not call certbot when booting the system - there is no need to do so and
	# request new certificates during boot/image upgrade as the certbot configuration
	# is stored persistent under /config - thus we do not open the door to transient
	# errors
	if not boot_configuration_complete():
	- return
	+ return None

	domains = '--domains ' + ' --domains '.join(config['domain_name'])
	tmp = f'certbot certonly --non-interactive --config-dir {vyos_certbot_dir} --cert-name {name} '\
	f'--standalone --agree-tos --no-eff-email --expand --server {config["url"]} '\
	f'--email {config["email"]} --key-type rsa --rsa-key-size {config["rsa_key_size"]} '\
	f'{domains}'

	listen_address = None
	if 'listen_address' in config:
	listen_address = config['listen_address']

	# When ACME is used behind a reverse proxy, we always bind to localhost
	# whatever the CLI listen-address is configured for.
	if ('used_by' in config and 'haproxy' in config['used_by'] and
	is_systemd_service_running(systemd_services['haproxy']) and
	not check_port_availability(listen_address, 80)):
	tmp += f' --http-01-address 127.0.0.1 --http-01-port {internal_ports["certbot_haproxy"]}'
	elif listen_address:
	tmp += f' --http-01-address {listen_address}'

	# verify() does not need to actually request a cert but only test for plausability
	if dry_run:
	tmp += ' --dry-run'

	- cmd(tmp, raising=ConfigError, message=f'ACME certbot request failed for "{name}"!')
	+ cmd(tmp, raising=ConfigError, message=f'Certbot request failed for "{name}"!')
	+ return None

	def get_config(config=None):
	if config:
	conf = config
	else:
	conf = Config()
	base = ['pki']

	pki = conf.get_config_dict(base, key_mangling=('-', '_'),
	get_first_key=True,
	no_tag_node_value_mangle=True)

	if len(argv) > 1 and argv[1] == 'certbot_renew':
	pki['certbot_renew'] = {}


	# Walk through the list of sync_translate mapping and build a list
	# which is later used to check if the node was changed in the CLI config
	changed_keys = []
	for value in sync_translate.values():
	if value not in changed_keys:
	changed_keys.append(value)
	# Check for changes to said given keys in the CLI config
	for key in changed_keys:
	tmp = node_changed(conf, base + [key], recursive=True, expand_nodes=Diff.DELETE \| Diff.ADD)
	- if 'changed' not in pki:
	- pki.update({'changed':{}})
	- pki['changed'].update({key.replace('-', '_') : tmp})
	+ if tmp:
	+ dict_set_nested(f'changed.{key.replace("-", "_")}', tmp, pki)

	# We only merge on the defaults if there is a configuration at all
	if conf.exists(base):
	# We have gathered the dict representation of the CLI, but there are default
	# options which we need to update into the dictionary retrived.
	default_values = conf.get_config_defaults(**pki.kwargs, recursive=True)
	# remove ACME default configuration if unused by CLI
	if 'certificate' in pki:
	for name, cert_config in pki['certificate'].items():
	if 'acme' not in cert_config:
	# Remove ACME default values
	del default_values['certificate'][name]['acme']

	# merge CLI and default dictionary
	pki = config_dict_merge(default_values, pki)

	# Certbot triggered an external renew of the certificates.
	# Mark all ACME based certificates as "changed" to trigger
	# update of dependent services
	if 'certificate' in pki and 'certbot_renew' in pki:
	renew = []
	for name, cert_config in pki['certificate'].items():
	if 'acme' in cert_config:
	renew.append(name)
	- # If triggered externally by certbot, certificate key is not present in changed
	- if 'changed' not in pki: pki.update({'changed':{}})
	- pki['changed'].update({'certificate' : renew})
	+ if renew:
	+ # Get the current list of changed certificates
	+ tmp = pki.get('changed', {}).get('certificate', [])
	+ # and extend it with the list of ACME based certificates
	+ tmp += renew
	+ # remove any duplicates if necessary
	+ tmp = set(tmp)
	+ dict_set_nested('changed.certificate', tmp, 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)
	D = get_config_diff(conf)

	for search in sync_search:
	for key in search['keys']:
	changed_key = sync_translate[key]
	if 'changed' not in pki or changed_key not in pki['changed']:
	continue

	for item_name in pki['changed'][changed_key]:
	node_present = False
	if changed_key == 'openvpn':
	node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
	else:
	node_present = dict_search_args(pki, changed_key, item_name)

	if node_present:
	search_dict = dict_search_args(pki['system'], *search['path'])
	if not search_dict:
	continue
	for found_name, found_path in dict_search_recursive(search_dict, key):
	if isinstance(found_name, list) and item_name not in found_name:
	continue

	if isinstance(found_name, str) and found_name != item_name:
	continue

	path = search['path']
	- path_str = ' '.join(path + found_path).replace('_','-')
	- Message(f'Updating configuration: "{path_str} {item_name}"')
	+ # Only enable this for debug purposes - otherwise we will always
	+ # print this message for ACME certificates during renew tests -
	+ # even if they are not due for renew!
	+ # path_str = ' '.join(path + found_path).replace('_','-')
	+ # print(f'Updating configuration: "{path_str} {item_name}"')

	if path[0] == 'interfaces':
	ifname = found_path[0]
	if not D.node_changed_presence(path + [ifname]):
	set_dependents(path[1], conf, ifname)
	else:
	if not D.node_changed_presence(path):
	set_dependents(path[1], conf)

	# Check PKI certificates if they are auto-generated by ACME. If they are,
	# traverse the current configuration and determine the service where the
	# certificate is used by.
	# Required to check if we might need to run certbot behing a reverse proxy.
	if 'certificate' in pki:
	for name, cert_config in pki['certificate'].items():
	if 'acme' not in cert_config:
	continue
	if not dict_search('system.load_balancing.haproxy', pki):
	continue
	# Determine which service depends on ACME issued certificates
	+ # We only need to add services blocking the default certbot ports
	+ # 80 and 443. For instance there won't be a conflict with strongSwan
	+ # as it runs on different ports.
	used_by = []
	# We start with HAProxy
	for cert_list, _ in dict_search_recursive(
	pki['system']['load_balancing']['haproxy'], 'certificate'):
	if name in cert_list:
	used_by.append('haproxy')
	# Check if OpenConnect consumes an ACME certificate
	tmp = dict_search('system.vpn.openconnect.ssl.certificate', pki)
	if tmp and tmp in cert_list:
	used_by.append('openconnect')

	if used_by:
	pki['certificate'][name]['acme'].update({'used_by': used_by})

	return pki

	def is_valid_certificate(raw_data):
	# If it loads correctly we're good, or return False
	return load_certificate(raw_data, wrap_tags=True)

	def is_valid_ca_certificate(raw_data):
	# Check if this is a valid certificate with CA attributes
	cert = load_certificate(raw_data, wrap_tags=True)
	if not cert:
	return False
	return is_ca_certificate(cert)

	def is_valid_public_key(raw_data):
	# If it loads correctly we're good, or return False
	return load_public_key(raw_data, wrap_tags=True)

	def is_valid_private_key(raw_data, protected=False):
	# If it loads correctly we're good, or return False
	# With encrypted private keys, we always return true as we cannot ask for password to verify
	if protected:
	return True
	return load_private_key(raw_data, passphrase=None, wrap_tags=True)

	def is_valid_openssh_public_key(raw_data, type):
	# If it loads correctly we're good, or return False
	return load_openssh_public_key(raw_data, type)

	def is_valid_openssh_private_key(raw_data, protected=False):
	# If it loads correctly we're good, or return False
	# With encrypted private keys, we always return true as we cannot ask for password to verify
	if protected:
	return True
	return load_openssh_private_key(raw_data, passphrase=None, wrap_tags=True)

	def is_valid_crl(raw_data):
	# If it loads correctly we're good, or return False
	return load_crl(raw_data, wrap_tags=True)

	def is_valid_dh_parameters(raw_data):
	# If it loads correctly we're good, or return False
	return load_dh_parameters(raw_data, wrap_tags=True)

	def verify(pki):
	if not pki:
	return None

	if 'ca' in pki:
	for name, ca_conf in pki['ca'].items():
	if 'certificate' in ca_conf:
	if not is_valid_ca_certificate(ca_conf['certificate']):
	raise ConfigError(f'Invalid certificate on CA certificate "{name}"')

	if 'private' in ca_conf and 'key' in ca_conf['private']:
	private = ca_conf['private']
	protected = 'password_protected' in private

	if not is_valid_private_key(private['key'], protected):
	raise ConfigError(f'Invalid private key on CA certificate "{name}"')

	if 'crl' in ca_conf:
	ca_crls = ca_conf['crl']
	if isinstance(ca_crls, str):
	ca_crls = [ca_crls]

	for crl in ca_crls:
	if not is_valid_crl(crl):
	raise ConfigError(f'Invalid CRL on CA certificate "{name}"')

	if 'certificate' in pki:
	for name, cert_conf in pki['certificate'].items():
	if 'certificate' in cert_conf:
	if not is_valid_certificate(cert_conf['certificate']):
	raise ConfigError(f'Invalid certificate on certificate "{name}"')

	if 'private' in cert_conf and 'key' in cert_conf['private']:
	private = cert_conf['private']
	protected = 'password_protected' in private

	if not is_valid_private_key(private['key'], protected):
	raise ConfigError(f'Invalid private key on certificate "{name}"')

	if 'acme' in cert_conf:
	if 'domain_name' not in cert_conf['acme']:
	raise ConfigError(f'At least one domain-name is required to request '\
	f'certificate for "{name}" via ACME!')

	if 'email' not in cert_conf['acme']:
	raise ConfigError(f'An email address is required to request '\
	f'certificate for "{name}" via ACME!')

	listen_address = None
	if 'listen_address' in cert_conf['acme']:
	listen_address = cert_conf['acme']['listen_address']

	if 'used_by' not in cert_conf['acme']:
	# A call to check_port_availability() will always fail during system
	# boot when listen_address is set and the address is not yet assigned
	# to an interface. This happens b/c PKI subsystem is called prior
	# to any inteface - e.g. ethernet - and thus the OS will always
	# be unable to bind() a socket() to a non existing IP address.
	if boot_configuration_complete() and not check_port_availability(listen_address, 80):
	raise ConfigError('Port 80 is already in use and not available '\
	f'to provide ACME challenge for "{name}"!')

	+ # Only run the ACME command if something on this entity changed,
	+ # as this is time intensive
	if 'certbot_renew' not in pki:
	- # Only run the ACME command if something on this entity changed,
	- # as this is time intensive
	tmp = dict_search('changed.certificate', pki)
	if tmp != None and name in tmp:
	certbot_request(name, cert_conf['acme'])

	if 'dh' in pki:
	for name, dh_conf in pki['dh'].items():
	if 'parameters' in dh_conf:
	if not is_valid_dh_parameters(dh_conf['parameters']):
	raise ConfigError(f'Invalid DH parameters on "{name}"')

	if 'key_pair' in pki:
	for name, key_conf in pki['key_pair'].items():
	if 'public' in key_conf and 'key' in key_conf['public']:
	if not is_valid_public_key(key_conf['public']['key']):
	raise ConfigError(f'Invalid public key on key-pair "{name}"')

	if 'private' in key_conf and 'key' in key_conf['private']:
	private = key_conf['private']
	protected = 'password_protected' in private
	if not is_valid_private_key(private['key'], protected):
	raise ConfigError(f'Invalid private key on key-pair "{name}"')

	if 'openssh' in pki:
	for name, key_conf in pki['openssh'].items():
	if 'public' in key_conf and 'key' in key_conf['public']:
	if 'type' not in key_conf['public']:
	raise ConfigError(f'Must define OpenSSH public key type for "{name}"')
	if not is_valid_openssh_public_key(key_conf['public']['key'], key_conf['public']['type']):
	raise ConfigError(f'Invalid OpenSSH public key "{name}"')

	if 'private' in key_conf and 'key' in key_conf['private']:
	private = key_conf['private']
	protected = 'password_protected' in private
	if not is_valid_openssh_private_key(private['key'], protected):
	raise ConfigError(f'Invalid OpenSSH private key "{name}"')

	if 'x509' in pki:
	if 'default' in pki['x509']:
	default_values = pki['x509']['default']
	if 'country' in default_values:
	country = default_values['country']
	if len(country) != 2 or not country.isalpha():
	raise ConfigError('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 search in sync_search:
	for key in search['keys']:
	changed_key = sync_translate[key]
	if changed_key not in pki['changed']:
	continue
	for item_name in pki['changed'][changed_key]:
	node_present = False
	if changed_key == 'openvpn':
	node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
	else:
	node_present = dict_search_args(pki, changed_key, item_name)
	# If the node is still present, we can skip the check
	# as we are not deleting it
	if node_present:
	continue

	search_dict = dict_search_args(pki['system'], *search['path'])
	if not search_dict:
	continue

	for found_name, found_path in dict_search_recursive(search_dict, key):
	# Check if the name matches either by string compare, or beeing
	# part of a list
	if ((isinstance(found_name, str) and found_name == item_name) or
	(isinstance(found_name, list) and item_name in found_name)):
	# We do not support _ in CLI paths - this is only a convenience
	# as we mangle all - to _, now it's time to reverse this!
	path_str = ' '.join(search['path'] + found_path).replace('_','-')
	object = changed_key.replace('_','-')
	tmp = f'Embedded PKI {object} with name "{item_name}" is still '\
	f'in use by CLI path "{path_str}"'
	raise ConfigError(tmp)

	return None

	def cleanup_system_ca():
	if not os.path.exists(vyos_ca_certificates_dir):
	os.mkdir(vyos_ca_certificates_dir)
	else:
	for filename in os.listdir(vyos_ca_certificates_dir):
	full_path = os.path.join(vyos_ca_certificates_dir, filename)
	if os.path.isfile(full_path):
	os.unlink(full_path)

	def generate(pki):
	if not pki:
	cleanup_system_ca()
	return None

	# Create or cleanup CA install directory
	if 'changed' in pki and 'ca' in pki['changed']:
	cleanup_system_ca()

	if 'ca' in pki:
	for ca, ca_conf in pki['ca'].items():
	if 'system_install' in ca_conf:
	ca_obj = load_certificate(ca_conf['certificate'])
	ca_path = os.path.join(vyos_ca_certificates_dir, f'{ca}.crt')

	with open(ca_path, 'w') as f:
	f.write(encode_certificate(ca_obj))

	# Certbot renewal only needs to re-trigger the services to load up the
	# new PEM file
	if 'certbot_renew' in pki:
	return None

	certbot_list = []
	certbot_list_on_disk = []
	if os.path.exists(f'{vyos_certbot_dir}/live'):
	certbot_list_on_disk = [f.path.split('/')[-1] for f in os.scandir(f'{vyos_certbot_dir}/live') if f.is_dir()]

	if 'certificate' in pki:
	changed_certificates = dict_search('changed.certificate', pki)
	for name, cert_conf in pki['certificate'].items():
	if 'acme' in cert_conf:
	certbot_list.append(name)
	# There is no ACME/certbot managed certificate presend on the
	# system, generate it
	if name not in certbot_list_on_disk:
	certbot_request(name, cert_conf['acme'], dry_run=False)
	# Now that the certificate was properly generated we have
	# the PEM files on disk. We need to add the certificate to
	# certbot_list_on_disk to automatically import the CA chain
	certbot_list_on_disk.append(name)
	# We alredy had an ACME managed certificate on the system, but
	# something changed in the configuration
	elif changed_certificates != None and name in changed_certificates:
	# Delete old ACME certificate first
	if name in certbot_list_on_disk:
	certbot_delete(name)
	# Request new certificate via certbot
	certbot_request(name, cert_conf['acme'], dry_run=False)

	# Cleanup certbot configuration and certificates if no longer in use by CLI
	# Get foldernames under vyos_certbot_dir which each represent a certbot cert
	if os.path.exists(f'{vyos_certbot_dir}/live'):
	for cert in certbot_list_on_disk:
	# ACME certificate is no longer in use by CLI remove it
	if cert not in certbot_list:
	certbot_delete(cert)
	continue
	# ACME not enabled for individual certificate - bail out early
	if 'acme' not in pki['certificate'][cert]:
	continue

	# Read in ACME certificate chain information
	tmp = read_file(f'{vyos_certbot_dir}/live/{cert}/chain.pem')
	tmp = load_certificate(tmp, wrap_tags=False)
	cert_chain_base64 = "".join(encode_certificate(tmp).strip().split("\n")[1:-1])

	# Check if CA chain certificate is already present on CLI to avoid adding
	# a duplicate. This only checks for manual added CA certificates and not
	# auto added ones with the AUTOCHAIN_ prefix
	autochain_prefix = 'AUTOCHAIN_'
	ca_cert_present = False
	if 'ca' in pki:
	for ca_base64, cli_path in dict_search_recursive(pki['ca'], 'certificate'):
	# Ignore automatic added CA certificates
	if any(item.startswith(autochain_prefix) for item in cli_path):
	continue
	if cert_chain_base64 == ca_base64:
	ca_cert_present = True

	if not ca_cert_present:
	tmp = dict_search_args(pki, 'ca', f'{autochain_prefix}{cert}', 'certificate')
	if not bool(tmp) or tmp != cert_chain_base64:
	Message(f'Add/replace automatically imported CA certificate for "{cert}" ...')
	add_cli_node(['pki', 'ca', f'{autochain_prefix}{cert}', 'certificate'], value=cert_chain_base64)

	return None

	def apply(pki):
	systemd_certbot_name = 'certbot.timer'
	if not pki:
	call(f'systemctl stop {systemd_certbot_name}')
	call('update-ca-certificates')
	return None

	has_certbot = False
	if 'certificate' in pki:
	for name, cert_conf in pki['certificate'].items():
	if 'acme' in cert_conf:
	has_certbot = True
	break

	if not has_certbot:
	call(f'systemctl stop {systemd_certbot_name}')
	elif has_certbot and not is_systemd_service_active(systemd_certbot_name):
	call(f'systemctl restart {systemd_certbot_name}')

	if 'changed' in pki:
	call_dependents()

	# Rebuild ca-certificates bundle
	if 'ca' in pki['changed']:
	call('update-ca-certificates')

	return None

	if __name__ == '__main__':
	try:
	c = get_config()
	verify(c)
	generate(c)
	apply(c)
	except ConfigError as e:
	print(e)
	exit(1)
	diff --git a/src/conf_mode/vpn_openconnect.py b/src/conf_mode/vpn_openconnect.py
	index 78cdaa179..68cd363f0 100755
	--- a/src/conf_mode/vpn_openconnect.py
	+++ b/src/conf_mode/vpn_openconnect.py
	@@ -1,418 +1,420 @@
	#!/usr/bin/env python3
	#
	# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
	#
	# This program is free software; you can redistribute it and/or modify
	# it under the terms of the GNU General Public License version 2 or later as
	# published by the Free Software Foundation.
	#
	# This program is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	# GNU General Public License for more details.
	#
	# You should have received a copy of the GNU General Public License
	# along with this program. If not, see <http://www.gnu.org/licenses/>.

	import os
	from sys import exit

	from vyos.base import Warning
	from vyos.config import Config
	from vyos.configverify import verify_pki_certificate
	from vyos.configverify import verify_pki_ca_certificate
	+from vyos.defaults import systemd_services
	from vyos.pki import find_chain
	from vyos.pki import encode_certificate
	from vyos.pki import load_certificate
	from vyos.pki import wrap_private_key
	from vyos.template import render
	from vyos.utils.dict import dict_search
	from vyos.utils.file import write_file
	from vyos.utils.network import check_port_availability
	from vyos.utils.network import is_listen_port_bind_service
	from vyos.utils.process import call
	from vyos.utils.process import is_systemd_service_running
	from vyos import ConfigError
	from passlib.hash import sha512_crypt
	from time import sleep

	from vyos import airbag

	airbag.enable()

	cfg_dir = '/run/ocserv'
	ocserv_conf = cfg_dir + '/ocserv.conf'
	ocserv_passwd = cfg_dir + '/ocpasswd'
	ocserv_otp_usr = cfg_dir + '/users.oath'
	radius_cfg = cfg_dir + '/radiusclient.conf'
	radius_servers = cfg_dir + '/radius_servers'


	# Generate hash from user cleartext password
	def get_hash(password):
	return sha512_crypt.hash(password)


	def get_config(config=None):
	if config:
	conf = config
	else:
	conf = Config()
	base = ['vpn', 'openconnect']
	if not conf.exists(base):
	return None

	ocserv = conf.get_config_dict(
	base,
	key_mangling=('-', '_'),
	get_first_key=True,
	with_recursive_defaults=True,
	with_pki=True,
	)

	return ocserv


	def verify(ocserv):
	if ocserv is None:
	return None
	# Check if listen-ports not binded other services
	# It can be only listen by 'ocserv-main'
	for proto, port in ocserv.get('listen_ports').items():
	if check_port_availability(
	ocserv['listen_address'], int(port), proto
	) is not True and not is_listen_port_bind_service(int(port), 'ocserv-main'):
	raise ConfigError(f'"{proto}" port "{port}" is used by another service')

	# Check accounting
	if 'accounting' in ocserv:
	if 'mode' in ocserv['accounting'] and 'radius' in ocserv['accounting']['mode']:
	if not ocserv['accounting']['radius']['server']:
	raise ConfigError(
	'OpenConnect accounting mode radius requires at least one RADIUS server'
	)
	if 'authentication' not in ocserv or 'mode' not in ocserv['authentication']:
	raise ConfigError(
	'Accounting depends on OpenConnect authentication configuration'
	)
	elif 'radius' not in ocserv['authentication']['mode']:
	raise ConfigError(
	'RADIUS accounting must be used with RADIUS authentication'
	)

	# Check authentication
	if 'authentication' in ocserv:
	if 'mode' in ocserv['authentication']:
	if (
	'local' in ocserv['authentication']['mode']
	and 'radius' in ocserv['authentication']['mode']
	):
	raise ConfigError(
	'OpenConnect authentication modes are mutually-exclusive, remove either local or radius from your configuration'
	)
	if 'radius' in ocserv['authentication']['mode']:
	if 'server' not in ocserv['authentication']['radius']:
	raise ConfigError(
	'OpenConnect authentication mode radius requires at least one RADIUS server'
	)
	if 'local' in ocserv['authentication']['mode']:
	if not ocserv.get('authentication', {}).get('local_users'):
	raise ConfigError(
	'OpenConnect mode local required at least one user'
	)
	if not ocserv['authentication']['local_users']['username']:
	raise ConfigError(
	'OpenConnect mode local required at least one user'
	)
	else:
	# For OTP mode: verify that each local user has an OTP key
	if 'otp' in ocserv['authentication']['mode']['local']:
	users_wo_key = []
	for user, user_config in ocserv['authentication'][
	'local_users'
	]['username'].items():
	# User has no OTP key defined
	if dict_search('otp.key', user_config) is None:
	users_wo_key.append(user)
	if users_wo_key:
	raise ConfigError(
	f'OTP enabled, but no OTP key is configured for these users:\n{users_wo_key}'
	)
	# For password (and default) mode: verify that each local user has password
	if (
	'password' in ocserv['authentication']['mode']['local']
	or 'otp' not in ocserv['authentication']['mode']['local']
	):
	users_wo_pswd = []
	for user in ocserv['authentication']['local_users']['username']:
	if (
	'password'
	not in ocserv['authentication']['local_users'][
	'username'
	][user]
	):
	users_wo_pswd.append(user)
	if users_wo_pswd:
	raise ConfigError(
	f'password required for users:\n{users_wo_pswd}'
	)

	# Validate that if identity-based-config is configured all child config nodes are set
	if 'identity_based_config' in ocserv['authentication']:
	if 'disabled' not in ocserv['authentication']['identity_based_config']:
	Warning(
	'Identity based configuration files is a 3rd party addition. Use at your own risk, this might break the ocserv daemon!'
	)
	if 'mode' not in ocserv['authentication']['identity_based_config']:
	raise ConfigError(
	'OpenConnect radius identity-based-config enabled but mode not selected'
	)
	elif (
	'group'
	in ocserv['authentication']['identity_based_config']['mode']
	and 'radius' not in ocserv['authentication']['mode']
	):
	raise ConfigError(
	'OpenConnect config-per-group must be used with radius authentication'
	)
	if (
	'directory'
	not in ocserv['authentication']['identity_based_config']
	):
	raise ConfigError(
	'OpenConnect identity-based-config enabled but directory not set'
	)
	if (
	'default_config'
	not in ocserv['authentication']['identity_based_config']
	):
	raise ConfigError(
	'OpenConnect identity-based-config enabled but default-config not set'
	)
	else:
	raise ConfigError('OpenConnect authentication mode required')
	else:
	raise ConfigError('OpenConnect authentication credentials required')

	# Check ssl
	if 'ssl' not in ocserv:
	raise ConfigError('SSL missing on OpenConnect config!')

	if 'certificate' not in ocserv['ssl']:
	raise ConfigError('SSL certificate missing on OpenConnect config!')
	verify_pki_certificate(ocserv, ocserv['ssl']['certificate'])

	if 'ca_certificate' in ocserv['ssl']:
	for ca_cert in ocserv['ssl']['ca_certificate']:
	verify_pki_ca_certificate(ocserv, ca_cert)

	# Check network settings
	if 'network_settings' in ocserv:
	if 'push_route' in ocserv['network_settings']:
	# Replace default route
	if '0.0.0.0/0' in ocserv['network_settings']['push_route']:
	ocserv['network_settings']['push_route'].remove('0.0.0.0/0')
	ocserv['network_settings']['push_route'].append('default')
	else:
	ocserv['network_settings']['push_route'] = ['default']
	else:
	raise ConfigError('OpenConnect network settings required!')


	def generate(ocserv):
	if not ocserv:
	return None

	if 'radius' in ocserv['authentication']['mode']:
	if dict_search(ocserv, 'accounting.mode.radius'):
	# Render radius client configuration
	render(radius_cfg, 'ocserv/radius_conf.j2', ocserv)
	merged_servers = (
	ocserv['accounting']['radius']['server']
	\| ocserv['authentication']['radius']['server']
	)
	# Render radius servers
	# Merge the accounting and authentication servers into a single dictionary
	render(
	radius_servers, 'ocserv/radius_servers.j2', {'server': merged_servers}
	)
	else:
	# Render radius client configuration
	render(radius_cfg, 'ocserv/radius_conf.j2', ocserv)
	# Render radius servers
	render(
	radius_servers,
	'ocserv/radius_servers.j2',
	ocserv['authentication']['radius'],
	)
	elif 'local' in ocserv['authentication']['mode']:
	# if mode "OTP", generate OTP users file parameters
	if 'otp' in ocserv['authentication']['mode']['local']:
	if 'local_users' in ocserv['authentication']:
	for user in ocserv['authentication']['local_users']['username']:
	# OTP token type from CLI parameters:
	otp_interval = str(
	ocserv['authentication']['local_users']['username'][user][
	'otp'
	].get('interval')
	)
	token_type = ocserv['authentication']['local_users']['username'][
	user
	]['otp'].get('token_type')
	otp_length = str(
	ocserv['authentication']['local_users']['username'][user][
	'otp'
	].get('otp_length')
	)
	if token_type == 'hotp-time':
	otp_type = 'HOTP/T' + otp_interval
	elif token_type == 'hotp-event':
	otp_type = 'HOTP/E'
	else:
	otp_type = 'HOTP/T' + otp_interval
	ocserv['authentication']['local_users']['username'][user]['otp'][
	'token_tmpl'
	] = otp_type + '/' + otp_length
	# if there is a password, generate hash
	if (
	'password' in ocserv['authentication']['mode']['local']
	or 'otp' not in ocserv['authentication']['mode']['local']
	):
	if 'local_users' in ocserv['authentication']:
	for user in ocserv['authentication']['local_users']['username']:
	ocserv['authentication']['local_users']['username'][user][
	'hash'
	] = get_hash(
	ocserv['authentication']['local_users']['username'][user][
	'password'
	]
	)

	if 'password-otp' in ocserv['authentication']['mode']['local']:
	# Render local users ocpasswd
	render(
	ocserv_passwd,
	'ocserv/ocserv_passwd.j2',
	ocserv['authentication']['local_users'],
	)
	# Render local users OTP keys
	render(
	ocserv_otp_usr,
	'ocserv/ocserv_otp_usr.j2',
	ocserv['authentication']['local_users'],
	)
	elif 'password' in ocserv['authentication']['mode']['local']:
	# Render local users ocpasswd
	render(
	ocserv_passwd,
	'ocserv/ocserv_passwd.j2',
	ocserv['authentication']['local_users'],
	)
	elif 'otp' in ocserv['authentication']['mode']['local']:
	# Render local users OTP keys
	render(
	ocserv_otp_usr,
	'ocserv/ocserv_otp_usr.j2',
	ocserv['authentication']['local_users'],
	)
	else:
	# Render local users ocpasswd
	render(
	ocserv_passwd,
	'ocserv/ocserv_passwd.j2',
	ocserv['authentication']['local_users'],
	)
	else:
	if 'local_users' in ocserv['authentication']:
	for user in ocserv['authentication']['local_users']['username']:
	ocserv['authentication']['local_users']['username'][user]['hash'] = (
	get_hash(
	ocserv['authentication']['local_users']['username'][user][
	'password'
	]
	)
	)
	# Render local users
	render(
	ocserv_passwd,
	'ocserv/ocserv_passwd.j2',
	ocserv['authentication']['local_users'],
	)

	if 'ssl' in ocserv:
	cert_file_path = os.path.join(cfg_dir, 'cert.pem')
	cert_key_path = os.path.join(cfg_dir, 'cert.key')

	if 'certificate' in ocserv['ssl']:
	cert_name = ocserv['ssl']['certificate']
	pki_cert = ocserv['pki']['certificate'][cert_name]

	loaded_pki_cert = load_certificate(pki_cert['certificate'])
	loaded_ca_certs = (
	{
	load_certificate(c['certificate'])
	for c in ocserv['pki']['ca'].values()
	}
	if 'ca' in ocserv['pki']
	else {}
	)

	cert_full_chain = find_chain(loaded_pki_cert, loaded_ca_certs)

	write_file(
	cert_file_path,
	'\n'.join(encode_certificate(c) for c in cert_full_chain),
	)

	if 'private' in pki_cert and 'key' in pki_cert['private']:
	write_file(cert_key_path, wrap_private_key(pki_cert['private']['key']))

	if 'ca_certificate' in ocserv['ssl']:
	ca_cert_file_path = os.path.join(cfg_dir, 'ca.pem')
	ca_chains = []

	for ca_name in ocserv['ssl']['ca_certificate']:
	pki_ca_cert = ocserv['pki']['ca'][ca_name]
	loaded_ca_cert = load_certificate(pki_ca_cert['certificate'])
	ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs)
	ca_chains.append(
	'\n'.join(encode_certificate(c) for c in ca_full_chain)
	)

	write_file(ca_cert_file_path, '\n'.join(ca_chains))

	# Render config
	render(ocserv_conf, 'ocserv/ocserv_config.j2', ocserv)


	def apply(ocserv):
	+ service_name = systemd_services['openconnect']
	if not ocserv:
	- call('systemctl stop ocserv.service')
	+ call(f'systemctl stop {service_name}')
	for file in [ocserv_conf, ocserv_passwd, ocserv_otp_usr]:
	if os.path.exists(file):
	os.unlink(file)
	else:
	- call('systemctl reload-or-restart ocserv.service')
	+ call(f'systemctl reload-or-restart {service_name}')
	counter = 0
	while True:
	# exit early when service runs
	- if is_systemd_service_running('ocserv.service'):
	+ if is_systemd_service_running(service_name):
	break
	sleep(0.250)
	if counter > 5:
	raise ConfigError(
	'OpenConnect failed to start, check the logs for details'
	)
	break
	counter += 1


	if __name__ == '__main__':
	try:
	c = get_config()
	verify(c)
	generate(c)
	apply(c)
	except ConfigError as e:
	print(e)
	exit(1)

File Metadata

Mime Type: text/x-diff
Expires: Sat, Sep 26, 2:15 PM (1 d, 12 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4285371
Default Alt Text: (48 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions