summaryrefslogtreecommitdiff
path: root/src/conf_mode
diff options
context:
space:
mode:
Diffstat (limited to 'src/conf_mode')
-rwxr-xr-xsrc/conf_mode/container.py2
-rwxr-xr-xsrc/conf_mode/interfaces_bridge.py8
-rwxr-xr-xsrc/conf_mode/interfaces_wwan.py5
-rwxr-xr-xsrc/conf_mode/nat.py7
-rwxr-xr-xsrc/conf_mode/pki.py7
-rwxr-xr-xsrc/conf_mode/protocols_ospf.py13
-rwxr-xr-xsrc/conf_mode/service_ssh.py76
-rwxr-xr-xsrc/conf_mode/system_conntrack.py2
-rwxr-xr-xsrc/conf_mode/system_ip.py5
-rwxr-xr-xsrc/conf_mode/system_login.py40
-rwxr-xr-xsrc/conf_mode/vpn_openconnect.py325
11 files changed, 335 insertions, 155 deletions
diff --git a/src/conf_mode/container.py b/src/conf_mode/container.py
index 94882fc14..83e6dee11 100755
--- a/src/conf_mode/container.py
+++ b/src/conf_mode/container.py
@@ -315,7 +315,7 @@ def generate_run_arguments(name, container_config):
sysctl_opt = ''
if 'sysctl' in container_config and 'parameter' in container_config['sysctl']:
for k, v in container_config['sysctl']['parameter'].items():
- sysctl_opt += f" --sysctl {k}={v['value']}"
+ sysctl_opt += f" --sysctl \"{k}={v['value']}\""
# Add capability options. Should be in uppercase
capabilities = ''
diff --git a/src/conf_mode/interfaces_bridge.py b/src/conf_mode/interfaces_bridge.py
index c14e6a599..fce07ae0a 100755
--- a/src/conf_mode/interfaces_bridge.py
+++ b/src/conf_mode/interfaces_bridge.py
@@ -111,6 +111,11 @@ def get_config(config=None):
elif interface.startswith('wlan') and interface_exists(interface):
set_dependents('wlan', conf, interface)
+ if interface.startswith('vtun'):
+ _, tmp_config = get_interface_dict(conf, ['interfaces', 'openvpn'], interface)
+ tmp = tmp_config.get('device_type') == 'tap'
+ bridge['member']['interface'][interface].update({'valid_ovpn' : tmp})
+
# delete empty dictionary keys - no need to run code paths if nothing is there to do
if 'member' in bridge:
if 'interface' in bridge['member'] and len(bridge['member']['interface']) == 0:
@@ -178,6 +183,9 @@ def verify(bridge):
if option in interface_config:
raise ConfigError('Can not use VLAN options on non VLAN aware bridge')
+ if interface.startswith('vtun') and not interface_config['valid_ovpn']:
+ raise ConfigError(error_msg + 'OpenVPN device-type must be set to "tap"')
+
if 'enable_vlan' in bridge:
if dict_search('vif.1', bridge):
raise ConfigError(f'VLAN 1 sub interface cannot be set for VLAN aware bridge {ifname}, and VLAN 1 is always the parent interface')
diff --git a/src/conf_mode/interfaces_wwan.py b/src/conf_mode/interfaces_wwan.py
index ddbebfb4a..fb71731d8 100755
--- a/src/conf_mode/interfaces_wwan.py
+++ b/src/conf_mode/interfaces_wwan.py
@@ -29,6 +29,7 @@ from vyos.configverify import verify_vrf
from vyos.configverify import verify_mtu_ipv6
from vyos.ifconfig import WWANIf
from vyos.utils.dict import dict_search
+from vyos.utils.network import is_wwan_connected
from vyos.utils.process import cmd
from vyos.utils.process import call
from vyos.utils.process import DEVNULL
@@ -137,7 +138,7 @@ def apply(wwan):
break
sleep(0.250)
- if 'shutdown_required' in wwan:
+ if 'shutdown_required' in wwan or (not is_wwan_connected(wwan['ifname'])):
# we only need the modem number. wwan0 -> 0, wwan1 -> 1
modem = wwan['ifname'].lstrip('wwan')
base_cmd = f'mmcli --modem {modem}'
@@ -159,7 +160,7 @@ def apply(wwan):
return None
- if 'shutdown_required' in wwan:
+ if 'shutdown_required' in wwan or (not is_wwan_connected(wwan['ifname'])):
ip_type = 'ipv4'
slaac = dict_search('ipv6.address.autoconf', wwan) != None
if 'address' in wwan:
diff --git a/src/conf_mode/nat.py b/src/conf_mode/nat.py
index 6c88e5cfd..a938021ba 100755
--- a/src/conf_mode/nat.py
+++ b/src/conf_mode/nat.py
@@ -31,7 +31,6 @@ from vyos.utils.file import write_file
from vyos.utils.process import cmd
from vyos.utils.process import run
from vyos.utils.process import call
-from vyos.utils.network import is_addr_assigned
from vyos.utils.network import interface_exists
from vyos.firewall import fqdn_config_parse
from vyos import ConfigError
@@ -176,12 +175,6 @@ def verify(nat):
if 'exclude' not in config and 'backend' not in config['load_balance']:
raise ConfigError(f'{err_msg} translation requires address and/or port')
- addr = dict_search('translation.address', config)
- if addr != None and addr != 'masquerade' and not is_ip_network(addr):
- for ip in addr.split('-'):
- if not is_addr_assigned(ip):
- Warning(f'IP address {ip} does not exist on the system!')
-
# common rule verification
verify_rule(config, err_msg, nat['firewall_group'])
diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
index 869518dd9..7d01b6642 100755
--- a/src/conf_mode/pki.py
+++ b/src/conf_mode/pki.py
@@ -64,6 +64,10 @@ sync_search = [
'path': ['service', 'https'],
},
{
+ 'keys': ['key'],
+ 'path': ['service', 'ssh'],
+ },
+ {
'keys': ['certificate', 'ca_certificate'],
'path': ['interfaces', 'ethernet'],
},
@@ -414,7 +418,8 @@ def verify(pki):
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.')
+ 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
diff --git a/src/conf_mode/protocols_ospf.py b/src/conf_mode/protocols_ospf.py
index c06c0aafc..467c9611b 100755
--- a/src/conf_mode/protocols_ospf.py
+++ b/src/conf_mode/protocols_ospf.py
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2021-2024 VyOS maintainers and contributors
+# 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
@@ -17,6 +17,7 @@
from sys import exit
from sys import argv
+from vyos.base import Warning
from vyos.config import Config
from vyos.configverify import verify_common_route_maps
from vyos.configverify import verify_route_map
@@ -62,6 +63,16 @@ def verify(config_dict):
if 'area' in ospf:
networks = []
for area, area_config in ospf['area'].items():
+ # Implemented as warning to not break existing configurations
+ if area == '0' and dict_search('area_type.nssa', area_config) != None:
+ Warning('You cannot configure NSSA to backbone!')
+ # Implemented as warning to not break existing configurations
+ if area == '0' and dict_search('area_type.stub', area_config) != None:
+ Warning('You cannot configure STUB to backbone!')
+ # Implemented as warning to not break existing configurations
+ if len(area_config['area_type']) > 1:
+ Warning(f'Only one area-type is supported for area "{area}"!')
+
if 'import_list' in area_config:
acl_import = area_config['import_list']
if acl_import: verify_access_list(acl_import, ospf)
diff --git a/src/conf_mode/service_ssh.py b/src/conf_mode/service_ssh.py
index 759f87bb2..3d38d940a 100755
--- a/src/conf_mode/service_ssh.py
+++ b/src/conf_mode/service_ssh.py
@@ -23,14 +23,15 @@ from syslog import LOG_INFO
from vyos.config import Config
from vyos.configdict import is_node_changed
from vyos.configverify import verify_vrf
-from vyos.configverify import verify_pki_ca_certificate
+from vyos.configverify import verify_pki_openssh_key
+from vyos.defaults import config_files
from vyos.utils.process import call
from vyos.template import render
from vyos import ConfigError
from vyos import airbag
-from vyos.pki import find_chain
-from vyos.pki import encode_certificate
-from vyos.pki import load_certificate
+from vyos.pki import encode_public_key
+from vyos.pki import load_openssh_public_key
+from vyos.utils.dict import dict_search_recursive
from vyos.utils.file import write_file
airbag.enable()
@@ -44,8 +45,7 @@ key_rsa = '/etc/ssh/ssh_host_rsa_key'
key_dsa = '/etc/ssh/ssh_host_dsa_key'
key_ed25519 = '/etc/ssh/ssh_host_ed25519_key'
-trusted_user_ca_key = '/etc/ssh/trusted_user_ca_key'
-
+trusted_user_ca = config_files['sshd_user_ca']
def get_config(config=None):
if config:
@@ -55,10 +55,8 @@ def get_config(config=None):
base = ['service', 'ssh']
if not conf.exists(base):
return None
-
- ssh = conf.get_config_dict(
- base, key_mangling=('-', '_'), get_first_key=True, with_pki=True
- )
+ ssh = conf.get_config_dict(base, key_mangling=('-', '_'),
+ get_first_key=True, with_pki=True)
tmp = is_node_changed(conf, base + ['vrf'])
if tmp:
@@ -68,14 +66,27 @@ def get_config(config=None):
# options which we need to update into the dictionary retrived.
ssh = conf.merge_defaults(ssh, recursive=True)
- # pass config file path - used in override template
- ssh['config_file'] = config_file
-
# Ignore default XML values if config doesn't exists
# Delete key from dict
if not conf.exists(base + ['dynamic-protection']):
del ssh['dynamic_protection']
+ # See if any user has specified a list of principal names that are accepted
+ # for certificate authentication.
+ tmp = conf.get_config_dict(['system', 'login', 'user'],
+ key_mangling=('-', '_'),
+ no_tag_node_value_mangle=True,
+ get_first_key=True)
+
+ for value, _ in dict_search_recursive(tmp, 'principal'):
+ # Only enable principal handling if SSH trusted-user-ca is set
+ if 'trusted_user_ca' in ssh:
+ ssh['has_principals'] = {}
+ # We do only need to execute this code path once as we need to know
+ # if any one of the local users has a principal set or not - this
+ # accounts for the entire system.
+ break
+
return ssh
@@ -86,15 +97,8 @@ def verify(ssh):
if 'rekey' in ssh and 'data' not in ssh['rekey']:
raise ConfigError('Rekey data is required!')
- if 'trusted_user_ca_key' in ssh:
- if 'ca_certificate' not in ssh['trusted_user_ca_key']:
- raise ConfigError('CA certificate is required for TrustedUserCAKey')
-
- ca_key_name = ssh['trusted_user_ca_key']['ca_certificate']
- verify_pki_ca_certificate(ssh, ca_key_name)
- pki_ca_cert = ssh['pki']['ca'][ca_key_name]
- if 'certificate' not in pki_ca_cert or not pki_ca_cert['certificate']:
- raise ConfigError(f"CA certificate '{ca_key_name}' is not valid or missing")
+ if 'trusted_user_ca' in ssh:
+ verify_pki_openssh_key(ssh, ssh['trusted_user_ca'])
verify_vrf(ssh)
return None
@@ -119,23 +123,17 @@ def generate(ssh):
syslog(LOG_INFO, 'SSH ed25519 host key not found, generating new key!')
call(f'ssh-keygen -q -N "" -t ed25519 -f {key_ed25519}')
- if 'trusted_user_ca_key' in ssh:
- ca_key_name = ssh['trusted_user_ca_key']['ca_certificate']
- pki_ca_cert = ssh['pki']['ca'][ca_key_name]
-
- loaded_ca_cert = load_certificate(pki_ca_cert['certificate'])
- loaded_ca_certs = {
- load_certificate(c['certificate'])
- for c in ssh['pki']['ca'].values()
- if 'certificate' in c
- }
-
- ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs)
- write_file(
- trusted_user_ca_key, '\n'.join(encode_certificate(c) for c in ca_full_chain)
- )
- elif os.path.exists(trusted_user_ca_key):
- os.unlink(trusted_user_ca_key)
+ if 'trusted_user_ca' in ssh:
+ key_name = ssh['trusted_user_ca']
+ openssh_cert = ssh['pki']['openssh'][key_name]
+ loaded_ca_cert = load_openssh_public_key(openssh_cert['public']['key'],
+ openssh_cert['public']['type'])
+ tmp = encode_public_key(loaded_ca_cert, encoding='OpenSSH',
+ key_format='OpenSSH')
+ write_file(trusted_user_ca, tmp, trailing_newline=True)
+ else:
+ if os.path.exists(trusted_user_ca):
+ os.unlink(trusted_user_ca)
render(config_file, 'ssh/sshd_config.j2', ssh)
diff --git a/src/conf_mode/system_conntrack.py b/src/conf_mode/system_conntrack.py
index f25ed8d10..8909d9cba 100755
--- a/src/conf_mode/system_conntrack.py
+++ b/src/conf_mode/system_conntrack.py
@@ -32,7 +32,6 @@ from vyos import ConfigError
from vyos import airbag
airbag.enable()
-conntrack_config = r'/etc/modprobe.d/vyatta_nf_conntrack.conf'
sysctl_file = r'/run/sysctl/10-vyos-conntrack.conf'
nftables_ct_file = r'/run/nftables-ct.conf'
vyos_conntrack_logger_config = r'/run/vyos-conntrack-logger.conf'
@@ -204,7 +203,6 @@ def generate(conntrack):
elif path[0] == 'ipv6':
conntrack['ipv6_firewall_action'] = 'accept'
- render(conntrack_config, 'conntrack/vyos_nf_conntrack.conf.j2', conntrack)
render(sysctl_file, 'conntrack/sysctl.conf.j2', conntrack)
render(nftables_ct_file, 'conntrack/nftables-ct.j2', conntrack)
diff --git a/src/conf_mode/system_ip.py b/src/conf_mode/system_ip.py
index 7f3796168..7f8b00ceb 100755
--- a/src/conf_mode/system_ip.py
+++ b/src/conf_mode/system_ip.py
@@ -53,6 +53,11 @@ def verify(config_dict):
for protocol, protocol_options in opt['protocol'].items():
if 'route_map' in protocol_options:
verify_route_map(protocol_options['route_map'], opt)
+
+ if dict_search('import_table', opt):
+ for table_num, import_config in opt['import_table'].items():
+ if dict_search('route_map', import_config):
+ verify_route_map(import_config['route_map'], opt)
return
def generate(config_dict):
diff --git a/src/conf_mode/system_login.py b/src/conf_mode/system_login.py
index 4febb6494..22b6fcc98 100755
--- a/src/conf_mode/system_login.py
+++ b/src/conf_mode/system_login.py
@@ -26,6 +26,8 @@ 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
@@ -129,6 +131,7 @@ def get_config(config=None):
max_uid=MIN_TACACS_UID) + cli_users
login['tacacs_min_uid'] = MIN_TACACS_UID
+ set_dependents('ssh', conf)
return login
def verify(login):
@@ -345,6 +348,17 @@ def apply(login):
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}"')
@@ -361,14 +375,15 @@ def apply(login):
chown(home_dir, user=user, recursive=True)
# Generate 2FA/MFA One-Time-Pad configuration
+ google_auth_file = f'{home_dir}/.google_authenticator'
if dict_search('authentication.otp.key', user_config):
enable_otp = True
- render(f'{home_dir}/.google_authenticator', 'login/pam_otp_ga.conf.j2',
+ render(google_auth_file, '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')
+ if os.path.exists(google_auth_file):
+ os.unlink(google_auth_file)
# Lock/Unlock local user account
lock_unlock = '--unlock'
@@ -382,6 +397,22 @@ def apply(login):
# Disable user to prevent re-login
call(f'usermod -s /sbin/nologin {user}')
+ home_dir = getpwnam(user).pw_dir
+ # Remove SSH authorized keys file
+ authorized_keys_file = f'{home_dir}/.ssh/authorized_keys'
+ if os.path.exists(authorized_keys_file):
+ os.unlink(authorized_keys_file)
+
+ # Remove SSH authorized principals file
+ principals_file = f'{home_dir}/.ssh/authorized_principals'
+ if os.path.exists(principals_file):
+ os.unlink(principals_file)
+
+ # Remove Google Authenticator file
+ google_auth_file = f'{home_dir}/.google_authenticator'
+ if os.path.exists(google_auth_file):
+ os.unlink(google_auth_file)
+
# 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!')
@@ -420,8 +451,9 @@ def apply(login):
# Enable/disable Google authenticator
cmd('pam-auth-update --disable mfa-google-authenticator')
if enable_otp:
- cmd(f'pam-auth-update --enable mfa-google-authenticator')
+ cmd('pam-auth-update --enable mfa-google-authenticator')
+ call_dependents()
return None
diff --git a/src/conf_mode/vpn_openconnect.py b/src/conf_mode/vpn_openconnect.py
index 0346c7819..1708b9d26 100755
--- a/src/conf_mode/vpn_openconnect.py
+++ b/src/conf_mode/vpn_openconnect.py
@@ -37,19 +37,22 @@ 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'
+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_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
@@ -59,78 +62,133 @@ def get_config(config=None):
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)
+ 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'):
+ 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 origin["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')
+ 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 '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')
+ 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"]:
+ if 'otp' in ocserv['authentication']['mode']['local']:
users_wo_key = []
- for user, user_config in ocserv["authentication"]["local_users"]["username"].items():
+ for user, user_config in ocserv['authentication'][
+ 'local_users'
+ ]['username'].items():
# User has no OTP key defined
- if dict_search('otp.key', user_config) == None:
+ 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}')
+ 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"]:
+ 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 not "password" in ocserv["authentication"]["local_users"]["username"][user]:
+ 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}')
+ 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')
+ 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:
@@ -149,94 +207,162 @@ def verify(ocserv):
verify_pki_ca_certificate(ocserv, ca_cert)
# Check network settings
- if "network_settings" in ocserv:
- if "push_route" in ocserv["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")
+ 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"]
+ 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 '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"]
+ 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})
+ 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"]:
+ 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"]:
+ 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"
+ 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
+ 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 not "otp" 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"]:
+ 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(
+ 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(
+ 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(
+ 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"])
+ 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"])
+ 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"])
+ 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"])
+ render(
+ ocserv_passwd,
+ 'ocserv/ocserv_passwd.j2',
+ ocserv['authentication']['local_users'],
+ )
- if "ssl" in ocserv:
+ 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 {}
+ 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))
+ 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']))
@@ -250,7 +376,8 @@ def generate(ocserv):
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))
+ '\n'.join(encode_certificate(c) for c in ca_full_chain)
+ )
write_file(ca_cert_file_path, '\n'.join(ca_chains))
@@ -269,11 +396,13 @@ def apply(ocserv):
counter = 0
while True:
# exit early when service runs
- if is_systemd_service_running("ocserv.service"):
+ if is_systemd_service_running('ocserv.service'):
break
sleep(0.250)
if counter > 5:
- raise ConfigError('OpenConnect failed to start, check the logs for details')
+ raise ConfigError(
+ 'OpenConnect failed to start, check the logs for details'
+ )
break
counter += 1
generated by cgit v1.2.3 (git 2.39.1) at 2025-08-07 05:37:34 +0000