No OneTemporary
Actions

Size

46 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/data/config-mode-dependencies/vyos-1x.json b/data/config-mode-dependencies/vyos-1x.json
	index 7506a0908..ccfc022f4 100644
	--- a/data/config-mode-dependencies/vyos-1x.json
	+++ b/data/config-mode-dependencies/vyos-1x.json
	@@ -1,80 +1,84 @@
	{
	"system_conntrack": {
	"conntrack_sync": ["service_conntrack-sync"],
	"vrf": ["vrf"]
	},
	"firewall": {
	"conntrack": ["system_conntrack"],
	"group_resync": ["system_conntrack", "nat", "policy_route"]
	},
	"interfaces_bonding": {
	"ethernet": ["interfaces_ethernet"]
	},
	"interfaces_bridge": {
	"vxlan": ["interfaces_vxlan"],
	"wlan": ["interfaces_wireless"]
	},
	"interfaces_wireguard": {
	"vxlan": ["interfaces_vxlan"]
	},
	"load_balancing_wan": {
	"conntrack": ["system_conntrack"]
	},
	"nat": {
	"conntrack": ["system_conntrack"]
	},
	"nat66": {
	"conntrack": ["system_conntrack"]
	},
	"pki": {
	"ethernet": ["interfaces_ethernet"],
	"openvpn": ["interfaces_openvpn"],
	"haproxy": ["load-balancing_haproxy"],
	"https": ["service_https"],
	"ipsec": ["vpn_ipsec"],
	"openconnect": ["vpn_openconnect"],
	"rpki": ["protocols_rpki"],
	+ "ssh": ["service_ssh"],
	"sstp": ["vpn_sstp"],
	"sstpc": ["interfaces_sstpc"],
	"stunnel": ["service_stunnel"]
	},
	"vpn_ipsec": {
	"nhrp": ["protocols_nhrp"]
	},
	"vpn_l2tp": {
	"ipsec": ["vpn_ipsec"]
	},
	"qos": {
	"bonding": ["interfaces_bonding"],
	"bridge": ["interfaces_bridge"],
	"dummy": ["interfaces_dummy"],
	"ethernet": ["interfaces_ethernet"],
	"geneve": ["interfaces_geneve"],
	"input": ["interfaces_input"],
	"l2tpv3": ["interfaces_l2tpv3"],
	"loopback": ["interfaces_loopback"],
	"macsec": ["interfaces_macsec"],
	"openvpn": ["interfaces_openvpn"],
	"pppoe": ["interfaces_pppoe"],
	"pseudo-ethernet": ["interfaces_pseudo-ethernet"],
	"tunnel": ["interfaces_tunnel"],
	"vti": ["interfaces_vti"],
	"vxlan": ["interfaces_vxlan"],
	"wireguard": ["interfaces_wireguard"],
	"wireless": ["interfaces_wireless"],
	"wwan": ["interfaces_wwan"]
	},
	"system_wireless": {
	"wireless": ["interfaces_wireless"]
	},
	"system_ip": {
	"sysctl": ["system_sysctl"]
	},
	"system_ipv6": {
	"sysctl": ["system_sysctl"]
	},
	+ "system_login": {
	+ "ssh": ["service_ssh"]
	+ },
	"system_option": {
	"ip_ipv6": ["system_ip", "system_ipv6"],
	"sysctl": ["system_sysctl"]
	}
	}
	diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
	index 869518dd9..14fe86d56 100755
	--- a/src/conf_mode/pki.py
	+++ b/src/conf_mode/pki.py
	@@ -1,590 +1,594 @@
	#!/usr/bin/env python3
	#
	# Copyright (C) 2021-2025 VyOS maintainers and contributors
	#
	# 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.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': ['ca_certificate'],
	+ '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):
	# 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

	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 ('haproxy' in dict_search('used_by', config) 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}"!')

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

	# We only merge on the defaults of 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})

	# 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}"')

	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
	used_by = []
	for cert_list, _ in dict_search_recursive(
	pki['system']['load_balancing']['haproxy'], 'certificate'):
	if name in cert_list:
	used_by.append('haproxy')
	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']:
	if 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}"!')

	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(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 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/system_login.py b/src/conf_mode/system_login.py
	index fa866c0ce..481fdd16e 100755
	--- a/src/conf_mode/system_login.py
	+++ b/src/conf_mode/system_login.py
	@@ -1,447 +1,451 @@
	#!/usr/bin/env python3
	#
	# Copyright (C) 2020-2024 VyOS maintainers and contributors
	#
	# 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 passlib.hosts import linux_context
	from psutil import users
	from pwd import getpwall
	from pwd import getpwnam
	from pwd import getpwuid
	from sys import exit
	from time import sleep

	from vyos.base import Warning
	from vyos.config import Config
	+from vyos.configdep import set_dependents
	+from vyos.configdep import call_dependents
	from vyos.configverify import verify_vrf
	from vyos.template import render
	from vyos.template import is_ipv4
	from vyos.utils.auth import EPasswdStrength
	from vyos.utils.auth import evaluate_strength
	from vyos.utils.auth import get_current_user
	from vyos.utils.configfs import delete_cli_node
	from vyos.utils.configfs import add_cli_node
	from vyos.utils.dict import dict_search
	from vyos.utils.file import chown
	from vyos.utils.process import cmd
	from vyos.utils.process import call
	from vyos.utils.process import run
	from vyos.utils.process import DEVNULL
	from vyos import ConfigError
	from vyos import airbag
	airbag.enable()

	autologout_file = "/etc/profile.d/autologout.sh"
	limits_file = "/etc/security/limits.d/10-vyos.conf"
	radius_config_file = "/etc/pam_radius_auth.conf"
	tacacs_pam_config_file = "/etc/tacplus_servers"
	tacacs_nss_config_file = "/etc/tacplus_nss.conf"
	nss_config_file = "/etc/nsswitch.conf"

	# Minimum UID used when adding system users
	MIN_USER_UID: int = 1000
	# Maximim UID used when adding system users
	MAX_USER_UID: int = 59999
	# LOGIN_TIMEOUT from /etc/loign.defs minus 10 sec
	MAX_RADIUS_TIMEOUT: int = 50
	# MAX_RADIUS_TIMEOUT divided by 2 sec (minimum recomended timeout)
	MAX_RADIUS_COUNT: int = 8
	# Maximum number of supported TACACS servers
	MAX_TACACS_COUNT: int = 8
	# Minimum USER id for TACACS users
	MIN_TACACS_UID = 900
	# List of local user accounts that must be preserved
	SYSTEM_USER_SKIP_LIST: list = ['radius_user', 'radius_priv_user', 'tacacs0', 'tacacs1',
	'tacacs2', 'tacacs3', 'tacacs4', 'tacacs5', 'tacacs6',
	'tacacs7', 'tacacs8', 'tacacs9', 'tacacs10',' tacacs11',
	'tacacs12', 'tacacs13', 'tacacs14', 'tacacs15']

	def get_local_users(min_uid=MIN_USER_UID, max_uid=MAX_USER_UID):
	"""Return list of dynamically allocated users (see Debian Policy Manual)"""
	local_users = []
	for s_user in getpwall():
	if getpwnam(s_user.pw_name).pw_uid < min_uid:
	continue
	if getpwnam(s_user.pw_name).pw_uid > max_uid:
	continue
	if s_user.pw_name in SYSTEM_USER_SKIP_LIST:
	continue
	local_users.append(s_user.pw_name)

	return local_users

	def get_shadow_password(username):
	with open('/etc/shadow') as f:
	for user in f.readlines():
	items = user.split(":")
	if username == items[0]:
	return items[1]
	return None

	def get_config(config=None):
	if config:
	conf = config
	else:
	conf = Config()
	base = ['system', 'login']
	login = conf.get_config_dict(base, key_mangling=('-', '_'),
	no_tag_node_value_mangle=True,
	get_first_key=True,
	with_recursive_defaults=True)

	# users no longer existing in the running configuration need to be deleted
	local_users = get_local_users()
	cli_users = []
	if 'user' in login:
	cli_users = list(login['user'])

	# prune TACACS global defaults if not set by user
	if login.from_defaults(['tacacs']):
	del login['tacacs']
	# same for RADIUS
	if login.from_defaults(['radius']):
	del login['radius']

	# create a list of all users, cli and users
	all_users = list(set(local_users + cli_users))
	# We will remove any normal users that dos not exist in the current
	# configuration. This can happen if user is added but configuration was not
	# saved and system is rebooted.
	rm_users = [tmp for tmp in all_users if tmp not in cli_users]
	if rm_users: login.update({'rm_users' : rm_users})

	# Build TACACS user mapping
	if 'tacacs' in login:
	login['exclude_users'] = get_local_users(min_uid=0,
	max_uid=MIN_TACACS_UID) + cli_users
	login['tacacs_min_uid'] = MIN_TACACS_UID

	+ set_dependents('ssh', conf)
	return login

	def verify(login):
	if 'rm_users' in login:
	# This check is required as the script is also executed from vyos-router
	# init script and there is no SUDO_USER environment variable available
	# during system boot.
	tmp = get_current_user()
	if tmp in login['rm_users']:
	raise ConfigError(f'Attempting to delete current user: {tmp}')

	if 'user' in login:
	system_users = getpwall()
	for user, user_config in login['user'].items():
	# Linux system users range up until UID 1000, we can not create a
	# VyOS CLI user which already exists as system user
	for s_user in system_users:
	if s_user.pw_name == user and s_user.pw_uid < MIN_USER_UID:
	raise ConfigError(f'User "{user}" can not be created, conflict with local system account!')

	# T6353: Check password for complexity using cracklib.
	# A user password should be sufficiently complex
	plaintext_password = dict_search(
	path='authentication.plaintext_password',
	dict_object=user_config
	) or None

	failed_check_status = [EPasswdStrength.WEAK, EPasswdStrength.ERROR]
	if plaintext_password is not None:
	result = evaluate_strength(plaintext_password)
	if result['strength'] in failed_check_status:
	Warning(result['error'])

	for pubkey, pubkey_options in (dict_search('authentication.public_keys', user_config) or {}).items():
	if 'type' not in pubkey_options:
	raise ConfigError(f'Missing type for public-key "{pubkey}"!')
	if 'key' not in pubkey_options:
	raise ConfigError(f'Missing key for public-key "{pubkey}"!')

	if {'radius', 'tacacs'} <= set(login):
	raise ConfigError('Using both RADIUS and TACACS at the same time is not supported!')

	# At lease one RADIUS server must not be disabled
	if 'radius' in login:
	if 'server' not in login['radius']:
	raise ConfigError('No RADIUS server defined!')
	sum_timeout: int = 0
	radius_servers_count: int = 0
	fail = True
	for server, server_config in dict_search('radius.server', login).items():
	if 'key' not in server_config:
	raise ConfigError(f'RADIUS server "{server}" requires key!')
	if 'disable' not in server_config:
	sum_timeout += int(server_config['timeout'])
	radius_servers_count += 1
	fail = False

	if fail:
	raise ConfigError('All RADIUS servers are disabled')

	if radius_servers_count > MAX_RADIUS_COUNT:
	raise ConfigError(f'Number of RADIUS servers exceeded maximum of {MAX_RADIUS_COUNT}!')

	if sum_timeout > MAX_RADIUS_TIMEOUT:
	raise ConfigError('Sum of RADIUS servers timeouts '
	'has to be less or eq 50 sec')

	verify_vrf(login['radius'])

	if 'source_address' in login['radius']:
	ipv4_count = 0
	ipv6_count = 0
	for address in login['radius']['source_address']:
	if is_ipv4(address): ipv4_count += 1
	else: ipv6_count += 1

	if ipv4_count > 1:
	raise ConfigError('Only one IPv4 source-address can be set!')
	if ipv6_count > 1:
	raise ConfigError('Only one IPv6 source-address can be set!')

	if 'tacacs' in login:
	tacacs_servers_count: int = 0
	fail = True
	for server, server_config in dict_search('tacacs.server', login).items():
	if 'key' not in server_config:
	raise ConfigError(f'TACACS server "{server}" requires key!')
	if 'disable' not in server_config:
	tacacs_servers_count += 1
	fail = False

	if fail:
	raise ConfigError('All RADIUS servers are disabled')

	if tacacs_servers_count > MAX_TACACS_COUNT:
	raise ConfigError(f'Number of TACACS servers exceeded maximum of {MAX_TACACS_COUNT}!')

	verify_vrf(login['tacacs'])

	if 'max_login_session' in login and 'timeout' not in login:
	raise ConfigError('"login timeout" must be configured!')

	return None


	def generate(login):
	# calculate users encrypted password
	if 'user' in login:
	for user, user_config in login['user'].items():
	tmp = dict_search('authentication.plaintext_password', user_config)
	if tmp:
	encrypted_password = linux_context.hash(tmp)
	login['user'][user]['authentication']['encrypted_password'] = encrypted_password
	del login['user'][user]['authentication']['plaintext_password']

	# Set default commands for re-adding user with encrypted password
	del_user_plain = ['system', 'login', 'user', user, 'authentication', 'plaintext-password']
	add_user_encrypt = ['system', 'login', 'user', user, 'authentication', 'encrypted-password']

	delete_cli_node(del_user_plain)
	add_cli_node(add_user_encrypt, value=encrypted_password)

	else:
	try:
	if get_shadow_password(user) == dict_search('authentication.encrypted_password', user_config):
	# If the current encrypted bassword matches the encrypted password
	# from the config - do not update it. This will remove the encrypted
	# value from the system logs.
	#
	# The encrypted password will be set only once during the first boot
	# after an image upgrade.
	del login['user'][user]['authentication']['encrypted_password']
	except:
	pass

	### RADIUS based user authentication
	if 'radius' in login:
	render(radius_config_file, 'login/pam_radius_auth.conf.j2', login,
	permission=0o600, user='root', group='root')
	else:
	if os.path.isfile(radius_config_file):
	os.unlink(radius_config_file)

	### TACACS+ based user authentication
	if 'tacacs' in login:
	render(tacacs_pam_config_file, 'login/tacplus_servers.j2', login,
	permission=0o644, user='root', group='root')
	render(tacacs_nss_config_file, 'login/tacplus_nss.conf.j2', login,
	permission=0o644, user='root', group='root')
	else:
	if os.path.isfile(tacacs_pam_config_file):
	os.unlink(tacacs_pam_config_file)
	if os.path.isfile(tacacs_nss_config_file):
	os.unlink(tacacs_nss_config_file)

	# NSS must always be present on the system
	render(nss_config_file, 'login/nsswitch.conf.j2', login,
	permission=0o644, user='root', group='root')

	# /etc/security/limits.d/10-vyos.conf
	if 'max_login_session' in login:
	render(limits_file, 'login/limits.j2', login,
	permission=0o644, user='root', group='root')
	else:
	if os.path.isfile(limits_file):
	os.unlink(limits_file)

	if 'timeout' in login:
	render(autologout_file, 'login/autologout.j2', login,
	permission=0o755, user='root', group='root')
	else:
	if os.path.isfile(autologout_file):
	os.unlink(autologout_file)

	return None


	def apply(login):
	enable_otp = False
	if 'user' in login:
	for user, user_config in login['user'].items():
	# make new user using vyatta shell and make home directory (-m),
	# default group of 100 (users)
	command = 'useradd --create-home --no-user-group '
	# check if user already exists:
	if user in get_local_users():
	# update existing account
	command = 'usermod'

	# all accounts use /bin/vbash
	command += ' --shell /bin/vbash'
	# we need to use '' quotes when passing formatted data to the shell
	# else it will not work as some data parts are lost in translation
	tmp = dict_search('authentication.encrypted_password', user_config)
	if tmp: command += f" --password '{tmp}'"

	tmp = dict_search('full_name', user_config)
	if tmp: command += f" --comment '{tmp}'"

	tmp = dict_search('home_directory', user_config)
	if tmp: command += f" --home '{tmp}'"
	else: command += f" --home '/home/{user}'"

	command += f' --groups frr,frrvty,vyattacfg,sudo,adm,dip,disk,_kea {user}'
	try:
	cmd(command)
	# we should not rely on the value stored in user_config['home_directory'], as a
	# crazy user will choose username root or any other system user which will fail.
	#
	# XXX: Should we deny using root at all?
	home_dir = getpwnam(user).pw_dir
	# always re-render SSH keys with appropriate permissions
	render(f'{home_dir}/.ssh/authorized_keys', 'login/authorized_keys.j2',
	user_config, permission=0o600,
	formater=lambda _: _.replace(""", '"'),
	user=user, group='users')

	principals_file = f'{home_dir}/.ssh/authorized_principals'
	if dict_search('authentication.principal', user_config):
	render(principals_file, 'login/authorized_principals.j2',
	user_config, permission=0o600,
	formater=lambda _: _.replace(""", '"'),
	user=user, group='users')
	else:
	if os.path.exists(principals_file):
	os.unlink(principals_file)

	except Exception as e:
	raise ConfigError(f'Adding user "{user}" raised exception: "{e}"')

	# T5875: ensure UID is properly set on home directory if user is re-added
	# the home directory will always exist, as it's created above by --create-home,
	# retrieve current owner of home directory and adjust on demand
	dir_owner = None
	try:
	dir_owner = getpwuid(os.stat(home_dir).st_uid).pw_name
	except:
	pass

	if dir_owner != user:
	chown(home_dir, user=user, recursive=True)

	# Generate 2FA/MFA One-Time-Pad configuration
	if dict_search('authentication.otp.key', user_config):
	enable_otp = True
	render(f'{home_dir}/.google_authenticator', 'login/pam_otp_ga.conf.j2',
	user_config, permission=0o400, user=user, group='users')
	else:
	# delete configuration as it's not enabled for the user
	if os.path.exists(f'{home_dir}/.google_authenticator'):
	os.remove(f'{home_dir}/.google_authenticator')

	# Lock/Unlock local user account
	lock_unlock = '--unlock'
	if 'disable' in user_config:
	lock_unlock = '--lock'
	cmd(f'usermod {lock_unlock} {user}')

	if 'rm_users' in login:
	for user in login['rm_users']:
	try:
	# Disable user to prevent re-login
	call(f'usermod -s /sbin/nologin {user}')

	# Logout user if he is still logged in
	if user in list(set([tmp[0] for tmp in users()])):
	print(f'{user} is logged in, forcing logout!')
	# re-run command until user is logged out
	while run(f'pkill -HUP -u {user}'):
	sleep(0.250)

	# Remove user account but leave home directory in place. Re-run
	# command until user is removed - userdel might return 8 as
	# SSH sessions are not all yet properly cleaned away, thus we
	# simply re-run the command until the account wen't away
	while run(f'userdel {user}', stderr=DEVNULL):
	sleep(0.250)

	except Exception as e:
	raise ConfigError(f'Deleting user "{user}" raised exception: {e}')

	# Enable/disable RADIUS in PAM configuration
	cmd('pam-auth-update --disable radius-mandatory radius-optional')
	if 'radius' in login:
	if login['radius'].get('security_mode', '') == 'mandatory':
	pam_profile = 'radius-mandatory'
	else:
	pam_profile = 'radius-optional'
	cmd(f'pam-auth-update --enable {pam_profile}')

	# Enable/disable TACACS+ in PAM configuration
	cmd('pam-auth-update --disable tacplus-mandatory tacplus-optional')
	if 'tacacs' in login:
	if login['tacacs'].get('security_mode', '') == 'mandatory':
	pam_profile = 'tacplus-mandatory'
	else:
	pam_profile = 'tacplus-optional'
	cmd(f'pam-auth-update --enable {pam_profile}')

	# Enable/disable Google authenticator
	cmd('pam-auth-update --disable mfa-google-authenticator')
	if enable_otp:
	cmd('pam-auth-update --enable mfa-google-authenticator')

	+ call_dependents()
	return None


	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: Fri, Nov 6, 10:04 AM (1 h, 42 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4654135
Default Alt Text: (46 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions