diff --git a/debian/control b/debian/control
index efc008af2..4f1207078 100644
--- a/debian/control
+++ b/debian/control
@@ -1,395 +1,397 @@
Source: vyos-1x
Section: contrib/net
Priority: extra
Maintainer: VyOS Package Maintainers <maintainers@vyos.net>
Build-Depends:
debhelper (>= 9),
dh-python,
fakeroot,
gcc,
iproute2,
libzmq3-dev,
python3 (>= 3.10),
# For QA
pylint,
# For generating command definitions
python3-lxml,
python3-xmltodict,
# For running tests
python3-coverage,
python3-hurry.filesize,
python3-netaddr,
python3-netifaces,
python3-nose,
python3-jinja2,
python3-paramiko,
python3-passlib,
python3-psutil,
python3-requests,
python3-setuptools,
python3-tabulate,
python3-zmq,
quilt,
whois
Standards-Version: 3.9.6
Package: vyos-1x
Architecture: amd64 arm64
Pre-Depends:
libpam-runtime [amd64],
libnss-tacplus [amd64],
libpam-tacplus [amd64],
libpam-radius-auth (= 1.5.0-cl3u7) [amd64],
libnss-mapuser (= 1.1.0-cl3u3) [amd64]
Depends:
## Fundamentals
${python3:Depends} (>= 3.10),
dialog,
libvyosconfig0,
libpam-cap,
bash-completion,
ipvsadm,
udev,
less,
at,
rsync,
vyatta-bash,
vyatta-biosdevname,
vyatta-cfg,
vyos-http-api-tools,
vyos-utils,
## End of Fundamentals
## Python libraries used in multiple modules and scripts
python3,
python3-cryptography,
python3-hurry.filesize,
python3-inotify,
python3-jinja2,
python3-jmespath,
python3-netaddr,
python3-netifaces,
python3-paramiko,
python3-passlib,
python3-pyroute2,
python3-psutil,
python3-pyhumps,
python3-pystache,
python3-pyudev,
python3-six,
python3-tabulate,
python3-voluptuous,
python3-xmltodict,
python3-zmq,
## End of Python libraries
## Basic System services and utilities
coreutils,
sudo,
systemd,
bsdmainutils,
openssl,
curl,
dbus,
file,
iproute2 (>= 6.0.0),
linux-cpupower,
# ipaddrcheck is widely used in IP value validators
ipaddrcheck,
ethtool (>= 6.10),
lm-sensors,
procps,
netplug,
sed,
ssl-cert,
tuned,
beep,
wide-dhcpv6-client,
# Generic colorizer
grc,
## End of System services and utilities
## For the installer
fdisk,
gdisk,
mdadm,
efibootmgr,
libefivar1,
dosfstools,
grub-efi-amd64-signed [amd64],
grub-efi-arm64-bin [arm64],
mokutil [amd64],
shim-signed [amd64],
sbsigntool [amd64],
# Image signature verification tool
minisign,
# Live filesystem tools
squashfs-tools,
fuse-overlayfs,
+# Tools for checking password strength
+ python3-cracklib,
## End installer
auditd,
iputils-arping,
iputils-ping,
isc-dhcp-client,
# For "vpn pptp", "vpn l2tp", "vpn sstp", "service ipoe-server"
accel-ppp,
# End "vpn pptp", "vpn l2tp", "vpn sstp", "service ipoe-server"
avahi-daemon,
conntrack,
conntrackd,
## Conf mode features
# For "interfaces wireless"
hostapd,
hsflowd,
iw,
wireless-regdb,
wpasupplicant (>= 0.6.7),
# End "interfaces wireless"
# For "interfaces wwan"
modemmanager,
usb-modeswitch,
libqmi-utils,
# End "interfaces wwan"
# For "interfaces openvpn"
openvpn,
openvpn-auth-ldap,
openvpn-auth-radius,
openvpn-otp,
openvpn-dco,
libpam-google-authenticator,
# End "interfaces openvpn"
# For "interfaces wireguard"
wireguard-tools,
qrencode,
# End "interfaces wireguard"
# For "interfaces pppoe"
pppoe,
# End "interfaces pppoe"
# For "interfaces sstpc"
sstp-client,
# End "interfaces sstpc"
# For "protocols *"
frr (>= 10.2),
frr-pythontools,
frr-rpki-rtrlib,
frr-snmp,
# End "protocols *"
# For "protocols igmp-proxy"
igmpproxy,
# End "protocols igmp-proxy"
# For "pki"
certbot,
# End "pki"
# For "service console-server"
conserver-client,
conserver-server,
console-data,
dropbear,
# End "service console-server"
# For "service aws glb"
aws-gwlbtun,
# For "service dns dynamic"
ddclient (>= 3.11.1),
# End "service dns dynamic"
# # For "service ids"
fastnetmon [amd64],
suricata,
suricata-update,
# End "service ids"
# # For "service ndp-proxy"
ndppd,
# End "service ndp-proxy"
# For "service router-advert"
radvd,
# End "service route-advert"
# For "load-balancing haproxy"
haproxy,
# End "load-balancing haproxy"
# For "service dhcp-relay"
isc-dhcp-relay,
# For "service dhcp-server"
kea,
# End "service dhcp-server"
# For "service lldp"
lldpd,
# End "service lldp"
# For "service https"
nginx-light,
# End "service https"
# For "service ssh"
openssh-server,
sshguard,
# End "service ssh"
# For "service salt-minion"
salt-minion,
# End "service salt-minion"
# For "service snmp"
snmp,
snmpd,
# End "service snmp"
# For "service webproxy"
squid,
squidclient,
squidguard,
# End "service webproxy"
# For "service monitoring prometheus node-exporter"
node-exporter,
# End "service monitoring prometheus node-exporter"
# For "service monitoring prometheus frr-exporter"
frr-exporter,
# End "service monitoring prometheus frr-exporter"
# For "service monitoring prometheus blackbox-exporter"
blackbox-exporter,
# End "service monitoring prometheus blackbox-exporter"
# For "service monitoring telegraf"
telegraf (>= 1.20),
# End "service monitoring telegraf"
# For "service monitoring zabbix-agent"
zabbix-agent2,
# End "service monitoring zabbix-agent"
# For "service tftp-server"
tftpd-hpa,
# End "service tftp-server"
# For "service dns forwarding"
pdns-recursor,
# End "service dns forwarding"
# For "service sla owamp"
owamp-client,
owamp-server,
# End "service sla owamp"
# For "service sla twamp"
twamp-client,
twamp-server,
# End "service sla twamp"
# For "service broadcast-relay"
udp-broadcast-relay,
# End "service broadcast-relay"
# For "high-availability vrrp"
keepalived (>=2.0.5),
# End "high-availability-vrrp"
# For "system console"
util-linux,
# End "system console"
# For "system task-scheduler"
cron,
# End "system task-scheduler"
# For "system lcd"
lcdproc,
lcdproc-extra-drivers,
# End "system lcd"
# For "system config-management commit-archive"
git,
# End "system config-management commit-archive"
# For firewall
libndp-tools,
libnetfilter-conntrack3,
libnfnetlink0,
nfct,
nftables (>= 0.9.3),
# For "vpn ipsec"
strongswan (>= 5.9),
strongswan-swanctl (>= 5.9),
charon-systemd,
libcharon-extra-plugins (>=5.9),
libcharon-extauth-plugins (>=5.9),
libstrongswan-extra-plugins (>=5.9),
libstrongswan-standard-plugins (>=5.9),
python3-vici (>= 5.7.2),
# End "vpn ipsec"
# For "nat64"
jool,
# End "nat64"
# For "system conntrack modules rtsp"
nat-rtsp,
# End "system conntrack modules rtsp"
# For "service ntp"
chrony,
# End "system ntp"
# For "vpn openconnect"
ocserv,
# End "vpn openconnect"
# For "system flow-accounting"
pmacct (>= 1.6.0),
# End "system flow-accounting"
# For "system syslog"
rsyslog,
# End "system syslog"
# For "system option keyboard-layout"
kbd,
# End "system option keyboard-layout"
# For "container"
podman (>=4.9.5),
netavark,
aardvark-dns,
# iptables is only used for containers now, not the the firewall CLI
iptables,
# End container
# For "vpp"
libvppinfra,
python3-vpp-api,
vpp,
vpp-dev,
vpp-plugin-core,
vpp-plugin-dpdk,
# End "vpp"
## End Configuration mode
## Operational mode
# Used for hypervisor model in "run show version"
hvinfo,
# For "run traceroute"
traceroute,
# For "run monitor traffic"
tcpdump,
# End "run monitor traffic"
# For "show hardware dmi"
dmidecode,
# For "run show hardware storage smart"
smartmontools,
# For "run show hardware scsi"
lsscsi,
# For "run show hardware pci"
pciutils,
# For "show hardware usb"
usbutils,
# For "run show hardware storage nvme"
nvme-cli,
# For "run monitor bandwidth-test"
iperf,
iperf3,
# End "run monitor bandwidth-test"
# For "run wake-on-lan"
etherwake,
# For "run force ipv6-nd"
ndisc6,
# For "run monitor bandwidth"
bmon,
# For "run format disk"
parted,
# End Operational mode
## TPM tools
cryptsetup,
tpm2-tools,
## End TPM tools
## Optional utilities
easy-rsa,
tcptraceroute,
mtr-tiny,
telnet,
stunnel4,
uidmap
## End optional utilities
Description: VyOS configuration scripts and data
VyOS configuration scripts, interface definitions, and everything
Package: vyos-1x-vmware
Architecture: amd64
Depends:
vyos-1x,
open-vm-tools
Description: VyOS configuration scripts and data for VMware
Adds configuration files required for VyOS running on VMware hosts.
Package: vyos-1x-smoketest
Architecture: all
Depends:
skopeo,
snmp,
vyos-1x
Description: VyOS build sanity checking toolkit
diff --git a/debian/vyos-1x.postinst b/debian/vyos-1x.postinst
index fde58651a..ba97f37f6 100644
--- a/debian/vyos-1x.postinst
+++ b/debian/vyos-1x.postinst
@@ -1,278 +1,290 @@
#!/bin/bash
# Turn off Debian default for %sudo
sed -i -e '/^%sudo/d' /etc/sudoers || true
# Add minion user for salt-minion
if ! grep -q '^minion' /etc/passwd; then
adduser --quiet --firstuid 100 --system --disabled-login --ingroup vyattacfg \
--gecos "salt minion user" --shell /bin/vbash minion
adduser --quiet minion frrvty
adduser --quiet minion sudo
adduser --quiet minion adm
adduser --quiet minion dip
adduser --quiet minion disk
adduser --quiet minion users
adduser --quiet minion frr
fi
# OpenVPN should get its own user
if ! grep -q '^openvpn' /etc/passwd; then
adduser --quiet --firstuid 100 --system --group --shell /usr/sbin/nologin openvpn
fi
# node_exporter should get its own user
if ! grep -q '^node_exporter' /etc/passwd; then
adduser --quiet --firstuid 100 --system --group --shell /bin/false node_exporter
fi
# We need to have a group for RADIUS service users to use it inside PAM rules
if ! grep -q '^radius' /etc/group; then
addgroup --firstgid 1000 --quiet radius
fi
# Remove TACACS user added by base package - we use our own UID range and group
# assignments - see below
if grep -q '^tacacs' /etc/passwd; then
if [ $(id -u tacacs0) -ge 1000 ]; then
level=0
vyos_group=vyattaop
while [ $level -lt 16 ]; do
userdel tacacs${level} || true
rm -rf /home/tacacs${level} || true
level=$(( level+1 ))
done 2>&1
fi
fi
# Remove TACACS+ PAM default profile
if [[ -e /usr/share/pam-configs/tacplus ]]; then
rm /usr/share/pam-configs/tacplus
fi
# Add TACACS system users required for TACACS based system authentication
if ! grep -q '^tacacs' /etc/passwd; then
# Add the tacacs group and all 16 possible tacacs privilege-level users to
# the password file, home directories, etc. The accounts are not enabled
# for local login, since they are only used to provide uid/gid/homedir for
# the mapped TACACS+ logins (and lookups against them). The tacacs15 user
# is also added to the sudo group, and vyattacfg group rather than vyattaop
# (used for tacacs0-14).
level=0
vyos_group=vyattaop
while [ $level -lt 16 ]; do
adduser --quiet --system --firstuid 900 --disabled-login --ingroup tacacs \
--no-create-home --gecos "TACACS+ mapped user at privilege level ${level}" \
--shell /bin/vbash tacacs${level}
adduser --quiet tacacs${level} frrvty
adduser --quiet tacacs${level} adm
adduser --quiet tacacs${level} dip
adduser --quiet tacacs${level} users
if [ $level -lt 15 ]; then
adduser --quiet tacacs${level} vyattaop
adduser --quiet tacacs${level} operator
else
adduser --quiet tacacs${level} vyattacfg
adduser --quiet tacacs${level} sudo
adduser --quiet tacacs${level} disk
adduser --quiet tacacs${level} frr
adduser --quiet tacacs${level} _kea
fi
level=$(( level+1 ))
done 2>&1 | grep -v "User tacacs${level} already exists"
fi
# Add RADIUS operator user for RADIUS authenticated users to map to
if ! grep -q '^radius_user' /etc/passwd; then
adduser --quiet --firstuid 1000 --disabled-login --ingroup radius \
--no-create-home --gecos "RADIUS mapped user at privilege level operator" \
--shell /sbin/radius_shell radius_user
adduser --quiet radius_user frrvty
adduser --quiet radius_user vyattaop
adduser --quiet radius_user operator
adduser --quiet radius_user adm
adduser --quiet radius_user dip
adduser --quiet radius_user users
fi
# Add RADIUS admin user for RADIUS authenticated users to map to
if ! grep -q '^radius_priv_user' /etc/passwd; then
adduser --quiet --firstuid 1000 --disabled-login --ingroup radius \
--no-create-home --gecos "RADIUS mapped user at privilege level admin" \
--shell /sbin/radius_shell radius_priv_user
adduser --quiet radius_priv_user frrvty
adduser --quiet radius_priv_user vyattacfg
adduser --quiet radius_priv_user sudo
adduser --quiet radius_priv_user adm
adduser --quiet radius_priv_user dip
adduser --quiet radius_priv_user disk
adduser --quiet radius_priv_user users
adduser --quiet radius_priv_user frr
adduser --quiet radius_priv_user _kea
fi
# add hostsd group for vyos-hostsd
if ! grep -q '^hostsd' /etc/group; then
addgroup --quiet --system hostsd
fi
# Add _kea user for kea-dhcp{4,6}-server to vyattacfg
# The user should exist via kea-common installed as transitive dependency
if grep -q '^_kea' /etc/passwd; then
adduser --quiet _kea vyattacfg
fi
# ensure the proxy user has a proper shell
chsh -s /bin/sh proxy
# Set file capabilities
setcap cap_net_admin=pe /sbin/ethtool
setcap cap_net_admin=pe /sbin/tc
setcap cap_net_admin=pe /bin/ip
setcap cap_net_admin=pe /sbin/xtables-legacy-multi
setcap cap_net_admin=pe /sbin/xtables-nft-multi
setcap cap_net_admin=pe /usr/sbin/conntrack
setcap cap_net_admin=pe /usr/sbin/arp
setcap cap_net_raw=pe /usr/bin/tcpdump
setcap cap_net_admin,cap_sys_admin=pe /sbin/sysctl
setcap cap_sys_module=pe /bin/kmod
setcap cap_sys_time=pe /bin/date
# create needed directories
mkdir -p /var/log/user
mkdir -p /var/core
mkdir -p /opt/vyatta/etc/config/auth
mkdir -p /opt/vyatta/etc/config/scripts
mkdir -p /opt/vyatta/etc/config/user-data
mkdir -p /opt/vyatta/etc/config/support
chown -R root:vyattacfg /opt/vyatta/etc/config
chmod -R 775 /opt/vyatta/etc/config
mkdir -p /opt/vyatta/etc/logrotate
mkdir -p /opt/vyatta/etc/netdevice.d
touch /etc/environment
if [ ! -f /etc/bash_completion ]; then
echo "source /etc/bash_completion.d/10vyatta-op" > /etc/bash_completion
echo "source /etc/bash_completion.d/20vyatta-cfg" >> /etc/bash_completion
fi
sed -i 's/^set /builtin set /' /etc/bash_completion
# Fix up PAM configuration for login so that invalid users are prompted
# for password
sed -i 's/requisite[ \t][ \t]*pam_securetty.so/required pam_securetty.so/' $rootfsdir/etc/pam.d/login
# Change default shell for new accounts
sed -i -e ':^DSHELL:s:/bin/bash:/bin/vbash:' /etc/adduser.conf
# Do not allow users to change full name field (controlled by vyos-1x)
sed -i -e 's/^CHFN_RESTRICT/#&/' /etc/login.defs
# Only allow root to use passwd command
if ! grep -q 'pam_succeed_if.so' /etc/pam.d/passwd ; then
sed -i -e '/^@include/i \
password requisite pam_succeed_if.so user = root
' /etc/pam.d/passwd
fi
# remove unnecessary ddclient script in /etc/ppp/ip-up.d/
# this logs unnecessary messages trying to start ddclient
rm -f /etc/ppp/ip-up.d/ddclient
# create /opt/vyatta/etc/config/scripts/vyos-preconfig-bootup.script
PRECONFIG_SCRIPT=/opt/vyatta/etc/config/scripts/vyos-preconfig-bootup.script
if [ ! -x $PRECONFIG_SCRIPT ]; then
mkdir -p $(dirname $PRECONFIG_SCRIPT)
touch $PRECONFIG_SCRIPT
chmod 755 $PRECONFIG_SCRIPT
cat <<EOF >>$PRECONFIG_SCRIPT
#!/bin/sh
# This script is executed at boot time before VyOS configuration is applied.
# Any modifications required to work around unfixed bugs or use
# services not available through the VyOS CLI system can be placed here.
EOF
fi
+# cracklib-runtime default database location
+CRACKLIB_DIR=/var/cache/cracklib
+CRACKLIB_DB=cracklib_dict
+
# create /opt/vyatta/etc/config/scripts/vyos-postconfig-bootup.script
POSTCONFIG_SCRIPT=/opt/vyatta/etc/config/scripts/vyos-postconfig-bootup.script
if [ ! -x $POSTCONFIG_SCRIPT ]; then
mkdir -p $(dirname $POSTCONFIG_SCRIPT)
touch $POSTCONFIG_SCRIPT
chmod 755 $POSTCONFIG_SCRIPT
cat <<EOF >>$POSTCONFIG_SCRIPT
#!/bin/sh
# This script is executed at boot time after VyOS configuration is fully applied.
# Any modifications required to work around unfixed bugs
# or use services not available through the VyOS CLI system can be placed here.
-
+#
+# T6353 - Just in case, check if cracklib was installed properly
+# If the database file is missing, re-install the runtime package
+#
+if [ ! -f "${CRACKLIB_DIR}/${CRACKLIB_DB}.pwd" ]; then
+ mkdir -p $CRACKLIB_DIR
+ /usr/sbin/create-cracklib-dict -o $CRACKLIB_DIR/$CRACKLIB_DB \
+ /usr/share/dict/cracklib-small
+fi
EOF
fi
# symlink destination is deleted during ISO assembly - this generates some noise
# when the system boots: systemd-sysv-generator[1881]: stat() failed on
# /etc/init.d/README, ignoring: No such file or directory. Thus we simply drop
# the file.
if [ -L /etc/init.d/README ]; then
rm -f /etc/init.d/README
fi
# Remove unwanted daemon files from /etc
# conntackd
# pmacct
# fastnetmon
# ntp
DELETE="/etc/logrotate.d/conntrackd.distrib /etc/init.d/conntrackd /etc/default/conntrackd
/etc/default/pmacctd /etc/pmacct
/etc/networks_list /etc/networks_whitelist /etc/fastnetmon.conf
/etc/ntp.conf /etc/default/ssh /etc/avahi/avahi-daemon.conf /etc/avahi/hosts
/etc/powerdns /etc/default/pdns-recursor
/etc/ppp/ip-up.d/0000usepeerdns /etc/ppp/ip-down.d/0000usepeerdns"
for tmp in $DELETE; do
if [ -e ${tmp} ]; then
rm -rf ${tmp}
fi
done
# Remove logrotate items controlled via CLI and VyOS defaults
sed -i '/^\/var\/log\/messages$/d' /etc/logrotate.d/rsyslog
sed -i '/^\/var\/log\/auth.log$/d' /etc/logrotate.d/rsyslog
# Fix FRR pam.d "vtysh_pam" vtysh_pam: Failed in account validation T5110
if test -f /etc/pam.d/frr; then
if grep -q 'pam_rootok.so' /etc/pam.d/frr; then
sed -i -re 's/rootok/permit/' /etc/pam.d/frr
fi
fi
# Enable Cloud-init pre-configuration service
systemctl enable vyos-config-cloud-init.service
# Enable Podman API
systemctl enable podman.service
# Generate API GraphQL schema
/usr/libexec/vyos/services/api/graphql/generate/generate_schema.py
# Update XML cache
python3 /usr/lib/python3/dist-packages/vyos/xml_ref/update_cache.py
# Generate hardlinks for systemd units for multi VRF support
# as softlinks will fail in systemd:
# symlink target name type "ssh.service" does not match source, rejecting.
if [ ! -f /lib/systemd/system/ssh@.service ]; then
ln /lib/systemd/system/ssh.service /lib/systemd/system/ssh@.service
fi
# T4287 - as we have a non-signed kernel use the upstream wireless reulatory database
update-alternatives --set regulatory.db /lib/firmware/regulatory.db-upstream
# Restart vyos-configd to apply changes in Python scripts/templates
if systemctl is-active --quiet vyos-configd; then
systemctl restart vyos-configd
fi
# Restart vyos-domain-resolver if running
if systemctl is-active --quiet vyos-domain-resolver; then
systemctl restart vyos-domain-resolver
fi
diff --git a/python/vyos/utils/auth.py b/python/vyos/utils/auth.py
index a0b3e1cae..a27d8a28a 100644
--- a/python/vyos/utils/auth.py
+++ b/python/vyos/utils/auth.py
@@ -1,51 +1,115 @@
# authutils -- miscelanneous functions for handling passwords and publis keys
#
# Copyright (C) 2023-2024 VyOS maintainers and contributors
#
# 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, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+import cracklib
+import math
import re
+import string
+from enum import StrEnum
+from decimal import Decimal
from vyos.utils.process import cmd
+
+DEFAULT_PASSWORD = 'vyos'
+LOW_ENTROPY_MSG = 'should be at least 8 characters long;'
+WEAK_PASSWORD_MSG= 'The password complexity is too low - @MSG@'
+
+
+class EPasswdStrength(StrEnum):
+ WEAK = 'Weak'
+ DECENT = 'Decent'
+ STRONG = 'Strong'
+
+
+def calculate_entropy(charset: str, passwd: str) -> float:
+ """
+ Calculate the entropy of a password based on the set of characters used
+ Uses E = log2(R**L) formula, where
+ - R is the range (length) of the character set
+ - L is the length of password
+ """
+ return math.log(math.pow(len(charset), len(passwd)), 2)
+
+def evaluate_strength(passwd: str) -> dict[str, str]:
+ """ Evaluates password strength and returns a check result dict """
+ charset = (cracklib.ASCII_UPPERCASE + cracklib.ASCII_LOWERCASE +
+ string.punctuation + string.digits)
+
+ result = {
+ 'strength': '',
+ 'error': '',
+ }
+
+ try:
+ cracklib.FascistCheck(passwd)
+ except ValueError as e:
+ # The password is vulnerable to dictionary attack no matter the entropy
+ if 'is' in str(e):
+ msg = str(e).replace('is', 'should not be')
+ else:
+ msg = f'should not be {e}'
+ result.update(strength=EPasswdStrength.WEAK)
+ result.update(error=WEAK_PASSWORD_MSG.replace('@MSG@', msg))
+ else:
+ # Now check the password's entropy
+ # Cast to Decimal for more precise rounding
+ entropy = Decimal.from_float(calculate_entropy(charset, passwd))
+
+ match round(entropy):
+ case e if e in range(0, 59):
+ result.update(strength=EPasswdStrength.WEAK)
+ result.update(
+ error=WEAK_PASSWORD_MSG.replace('@MSG@', LOW_ENTROPY_MSG)
+ )
+ case e if e in range(60, 119):
+ result.update(strength=EPasswdStrength.DECENT)
+ case e if e >= 120:
+ result.update(strength=EPasswdStrength.STRONG)
+
+ return result
+
def make_password_hash(password):
""" Makes a password hash for /etc/shadow using mkpasswd """
mkpassword = 'mkpasswd --method=sha-512 --stdin'
return cmd(mkpassword, input=password, timeout=5)
def split_ssh_public_key(key_string, defaultname=""):
""" Splits an SSH public key into its components """
key_string = key_string.strip()
parts = re.split(r'\s+', key_string)
if len(parts) == 3:
key_type, key_data, key_name = parts[0], parts[1], parts[2]
else:
key_type, key_data, key_name = parts[0], parts[1], defaultname
if key_type not in ['ssh-rsa', 'ssh-dss', 'ecdsa-sha2-nistp256', 'ecdsa-sha2-nistp384', 'ecdsa-sha2-nistp521', 'ssh-ed25519']:
raise ValueError("Bad key type \'{0}\', must be one of must be one of ssh-rsa, ssh-dss, ecdsa-sha2-nistp<256|384|521> or ssh-ed25519".format(key_type))
return({"type": key_type, "data": key_data, "name": key_name})
def get_current_user() -> str:
import os
current_user = 'nobody'
# During CLI "owner" script execution we use SUDO_USER
if 'SUDO_USER' in os.environ:
current_user = os.environ['SUDO_USER']
# During op-mode or config-mode interactive CLI we use USER
elif 'USER' in os.environ:
current_user = os.environ['USER']
return current_user
diff --git a/smoketest/scripts/cli/base_vyostest_shim.py b/smoketest/scripts/cli/base_vyostest_shim.py
index edf940efd..6da4ed9e6 100644
--- a/smoketest/scripts/cli/base_vyostest_shim.py
+++ b/smoketest/scripts/cli/base_vyostest_shim.py
@@ -1,219 +1,218 @@
# Copyright (C) 2021-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
import unittest
import paramiko
import pprint
from time import sleep
from typing import Type
from vyos.configsession import ConfigSession
from vyos.configsession import ConfigSessionError
from vyos import ConfigError
from vyos.defaults import commit_lock
from vyos.utils.process import cmd
from vyos.utils.process import run
save_config = '/tmp/vyos-smoketest-save'
# The commit process is not finished until all pending files from
# VYATTA_CHANGES_ONLY_DIR are copied to VYATTA_ACTIVE_CONFIGURATION_DIR. This
# is done inside libvyatta-cfg1 and the FUSE UnionFS part. On large non-
# interactive commits FUSE UnionFS might not replicate the real state in time,
# leading to errors when querying the working and effective configuration.
# TO BE DELETED AFTER SWITCH TO IN MEMORY CONFIG
CSTORE_GUARD_TIME = 4
# This class acts as shim between individual Smoketests developed for VyOS and
# the Python UnitTest framework. Before every test is loaded, we dump the current
# system configuration and reload it after the test - despite the test results.
#
# Using this approach we can not render a live system useless while running any
# kind of smoketest. In addition it adds debug capabilities like printing the
# command used to execute the test.
class VyOSUnitTestSHIM:
class TestCase(unittest.TestCase):
# if enabled in derived class, print out each and every set/del command
# on the CLI. This is usefull to grap all the commands required to
# trigger the certain failure condition.
# Use "self.debug = True" in derived classes setUp() method
debug = False
# Time to wait after a commit to ensure the CStore is up to date
# only required for testcases using FRR
_commit_guard_time = 0
@classmethod
def setUpClass(cls):
cls._session = ConfigSession(os.getpid())
cls._session.save_config(save_config)
if os.path.exists('/tmp/vyos.smoketest.debug'):
cls.debug = True
pass
@classmethod
def tearDownClass(cls):
# discard any pending changes which might caused a messed up config
cls._session.discard()
# ... and restore the initial state
cls._session.migrate_and_load_config(save_config)
try:
cls._session.commit()
except (ConfigError, ConfigSessionError):
cls._session.discard()
cls.fail(cls)
def cli_set(self, path, value=None):
if self.debug:
str = f'set {" ".join(path)} {value}' if value else f'set {" ".join(path)}'
print(str)
self._session.set(path, value)
def cli_delete(self, config):
if self.debug:
print('del ' + ' '.join(config))
self._session.delete(config)
def cli_discard(self):
if self.debug:
print('DISCARD')
self._session.discard()
def cli_commit(self):
if self.debug:
print('commit')
- self._session.commit()
# During a commit there is a process opening commit_lock, and run()
# returns 0
while run(f'sudo lsof -nP {commit_lock}') == 0:
sleep(0.250)
# Wait for CStore completion for fast non-interactive commits
sleep(self._commit_guard_time)
def op_mode(self, path : list) -> None:
"""
Execute OP-mode command and return stdout
"""
if self.debug:
print('commit')
path = ' '.join(path)
out = cmd(f'/opt/vyatta/bin/vyatta-op-cmd-wrapper {path}')
if self.debug:
print(f'\n\ncommand "{path}" returned:\n')
pprint.pprint(out)
return out
def getFRRconfig(self, string=None, end='$', endsection='^!',
substring=None, endsubsection=None, empty_retry=0):
"""
Retrieve current "running configuration" from FRR
string: search for a specific start string in the configuration
end: end of the section to search for (line ending)
endsection: end of the configuration
substring: search section under the result found by string
endsubsection: end of the subsection (usually something with "exit")
"""
command = f'vtysh -c "show run no-header"'
if string:
command += f' | sed -n "/^{string}{end}/,/{endsection}/p"'
if substring and endsubsection:
command += f' | sed -n "/^{substring}/,/{endsubsection}/p"'
out = cmd(command)
if self.debug:
print(f'\n\ncommand "{command}" returned:\n')
pprint.pprint(out)
if empty_retry > 0:
retry_count = 0
while not out and retry_count < empty_retry:
if self.debug and retry_count % 10 == 0:
print(f"Attempt {retry_count}: FRR config is still empty. Retrying...")
retry_count += 1
sleep(1)
out = cmd(command)
if not out:
print(f'FRR configuration still empty after {empty_retry} retires!')
return out
@staticmethod
def ssh_send_cmd(command, username, password, hostname='localhost'):
""" SSH command execution helper """
# Try to login via SSH
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname=hostname, username=username, password=password)
_, stdout, stderr = ssh_client.exec_command(command)
output = stdout.read().decode().strip()
error = stderr.read().decode().strip()
ssh_client.close()
return output, error
# Verify nftables output
def verify_nftables(self, nftables_search, table, inverse=False, args=''):
nftables_output = cmd(f'sudo nft {args} list table {table}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def verify_nftables_chain(self, nftables_search, table, chain, inverse=False, args=''):
nftables_output = cmd(f'sudo nft {args} list chain {table} {chain}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def verify_nftables_chain_exists(self, table, chain, inverse=False):
try:
cmd(f'sudo nft list chain {table} {chain}')
if inverse:
self.fail(f'Chain exists: {table} {chain}')
except OSError:
if not inverse:
self.fail(f'Chain does not exist: {table} {chain}')
# Verify ip rule output
def verify_rules(self, rules_search, inverse=False, addr_family='inet'):
rule_output = cmd(f'ip -family {addr_family} rule show')
for search in rules_search:
matched = False
for line in rule_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
# standard construction; typing suggestion: https://stackoverflow.com/a/70292317
def ignore_warning(warning: Type[Warning]):
import warnings
from functools import wraps
def inner(f):
@wraps(f)
def wrapped(*args, **kwargs):
with warnings.catch_warnings():
warnings.simplefilter("ignore", category=warning)
return f(*args, **kwargs)
return wrapped
return inner
diff --git a/smoketest/scripts/cli/test_system_login.py b/smoketest/scripts/cli/test_system_login.py
index d79f5521c..ed72f378e 100755
--- a/smoketest/scripts/cli/test_system_login.py
+++ b/smoketest/scripts/cli/test_system_login.py
@@ -1,538 +1,554 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-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
import re
import unittest
import jinja2
import secrets
import string
import paramiko
import shutil
from base_vyostest_shim import VyOSUnitTestSHIM
+from contextlib import redirect_stdout
from gzip import GzipFile
+from io import StringIO, TextIOWrapper
from subprocess import Popen
from subprocess import PIPE
from pwd import getpwall
from vyos.configsession import ConfigSessionError
from vyos.configquery import ConfigTreeQuery
from vyos.utils.auth import get_current_user
from vyos.utils.process import cmd
from vyos.utils.file import read_file
from vyos.utils.file import write_file
from vyos.template import inc_ip
from vyos.template import is_ipv6
from vyos.xml_ref import default_value
base_path = ['system', 'login']
users = ['vyos1', 'vyos-roxx123', 'VyOS-123_super.Nice']
+weak_passwd_user = ['test_user', 'passWord1']
ssh_test_command = '/opt/vyatta/bin/vyatta-op-cmd-wrapper show version'
ssh_pubkey = """
AAAAB3NzaC1yc2EAAAADAQABAAABgQD0NuhUOEtMIKnUVFIHoFatqX/c4mjerXyF
TlXYfVt6Ls2NZZsUSwHbnhK4BKDrPvVZMW/LycjQPzWW6TGtk6UbZP1WqdviQ9hP
jsEeKJSTKciMSvQpjBWyEQQPXSKYQC7ryQQilZDqnJgzqwzejKEe+nhhOdBvjuZc
uukxjT69E0UmWAwLxzvfiurwiQaC7tG+PwqvtfHOPL3i6yRO2C5ORpFarx8PeGDS
IfIXJCr3LoUbLHeuE7T2KaOKQcX0UsWJ4CoCapRLpTVYPDB32BYfgq7cW1Sal1re
EGH2PzuXBklinTBgCHA87lHjpwDIAqdmvMj7SXIW9LxazLtP+e37sexE7xEs0cpN
l68txdDbY2P2Kbz5mqGFfCvBYKv9V2clM5vyWNy/Xp5TsCis89nn83KJmgFS7sMx
pHJz8umqkxy3hfw0K7BRFtjWd63sbOP8Q/SDV7LPaIfIxenA9zv2rY7y+AIqTmSr
TTSb0X1zPGxPIRFy5GoGtO9Mm5h4OZk=
"""
tac_image = 'docker.io/lfkeitel/tacacs_plus:alpine'
tac_image_path = '/usr/share/vyos/tacplus-alpine.tar'
TAC_PLUS_TMPL_SRC = """
id = spawnd {
debug redirect = /dev/stdout
listen = { port = 49 }
spawn = {
instances min = 1
instances max = 10
}
background = no
}
id = tac_plus {
debug = ALL
log = stdout {
destination = /dev/stdout
}
authorization log group = yes
authentication log = stdout
authorization log = stdout
accounting log = stdout
host = smoketest {
address = {{ source_address }}/32
enable = clear enable
key = {{ tacacs_secret }}
}
group = admin {
default service = permit
enable = permit
service = shell {
default command = permit
default attribute = permit
set priv-lvl = 15
}
}
user = {{ username }} {
password = clear {{ password }}
member = admin
}
}
"""
radius_image = 'docker.io/dchidell/radius-web:latest'
radius_image_path = '/usr/share/vyos/radius-latest.tar'
RADIUS_CLIENTS_TMPL_SRC = """
client SMOKETEST {
secret = {{ radius_key }}
nastype = other
ipaddr = {{ source_address }}
}
"""
RADIUS_USERS_TMPL_SRC = """
# User configuration
{{ username }} Cleartext-Password := "{{ password }}"
Service-Type = NAS-Prompt-User,
Cisco-AVPair = "shell:priv-lvl=15"
"""
class TestSystemLogin(VyOSUnitTestSHIM.TestCase):
@classmethod
def setUpClass(cls):
super(TestSystemLogin, cls).setUpClass()
# ensure we can also run this test on a live system - so lets clean
# out the current configuration which will break this test
cls.cli_delete(cls, base_path + ['radius'])
cls.cli_delete(cls, base_path + ['tacacs'])
# Load images for smoketest provided in vyos-1x-smoketest
if not os.path.exists(tac_image_path):
cls.fail(cls, f'{tac_image} image not available')
cmd(f'sudo podman load -i {tac_image_path}')
if not os.path.exists(radius_image_path):
cls.fail(cls, f'{radius_image} image not available')
cmd(f'sudo podman load -i {radius_image_path}')
cls.ssh_test_command_result = cls.op_mode(cls, ['show', 'version'])
# Dynamically start SSH service if it's not running
config = ConfigTreeQuery()
cls.is_sshd_pre_test = config.exists(['service', 'sshd'])
if not cls.is_sshd_pre_test:
# Start SSH service
cls.cli_set(cls, ['service', 'ssh'])
@classmethod
def tearDownClass(cls):
# Stop SSH service - if it was not running before starting the test
if not cls.is_sshd_pre_test:
cls.cli_set(cls, ['service', 'ssh'])
cls.cli_commit(cls)
super(TestSystemLogin, cls).tearDownClass()
# Cleanup container images
cmd(f'sudo podman image rm -f {tac_image}')
cmd(f'sudo podman image rm -f {radius_image}')
def tearDown(self):
# Delete individual users from configuration
for user in users:
self.cli_delete(base_path + ['user', user])
self.cli_delete(base_path + ['radius'])
self.cli_delete(base_path + ['tacacs'])
self.cli_commit()
# After deletion, a user is not allowed to remain in /etc/passwd
usernames = [x[0] for x in getpwall()]
for user in users:
self.assertNotIn(user, usernames)
def test_add_linux_system_user(self):
# We are not allowed to re-use a username already taken by the Linux
# base system
system_user = 'backup'
self.cli_set(base_path + ['user', system_user, 'authentication', 'plaintext-password', system_user])
# check validate() - can not add username which exists on the Debian
# base system (UID < 1000)
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(base_path + ['user', system_user])
def test_system_login_user(self):
for user in users:
name = f'VyOS Roxx {user}'
+ passwd = f'{user}-pSWd-t3st'
home_dir = f'/tmp/smoketest/{user}'
- self.cli_set(base_path + ['user', user, 'authentication', 'plaintext-password', user])
+ self.cli_set(base_path + ['user', user, 'authentication', 'plaintext-password', passwd])
self.cli_set(base_path + ['user', user, 'full-name', name])
self.cli_set(base_path + ['user', user, 'home-directory', home_dir])
self.cli_commit()
for user in users:
+ passwd = f'{user}-pSWd-t3st'
tmp = ['su','-', user]
proc = Popen(tmp, stdin=PIPE, stdout=PIPE, stderr=PIPE)
- tmp = f'{user}\nuname -a'
+ tmp = f'{passwd}\nuname -a'
proc.stdin.write(tmp.encode())
proc.stdin.flush()
(stdout, stderr) = proc.communicate()
# stdout is something like this:
# b'Linux vyos 6.6.66-vyos 6.6.66-vyos #1 SMP Mon Dec 30 19:05:15 UTC 2024 x86_64 GNU/Linux\n'
self.assertTrue(len(stdout) > 40)
locked_user = users[0]
# disable the first user in list
self.cli_set(base_path + ['user', locked_user, 'disable'])
self.cli_commit()
# check if account is locked
tmp = cmd(f'sudo passwd -S {locked_user}')
self.assertIn(f'{locked_user} L ', tmp)
# unlock account
self.cli_delete(base_path + ['user', locked_user, 'disable'])
self.cli_commit()
# check if account is unlocked
tmp = cmd(f'sudo passwd -S {locked_user}')
self.assertIn(f'{locked_user} P ', tmp)
+ def test_system_login_weak_password_warning(self):
+ self.cli_set(base_path + [
+ 'user', weak_passwd_user[0], 'authentication',
+ 'plaintext-password', weak_passwd_user[1]
+ ])
+
+ out = self.cli_commit().strip()
+
+ self.assertIn('WARNING: The password complexity is too low', out)
+ self.cli_delete(base_path + ['user', weak_passwd_user[0]])
+
def test_system_login_otp(self):
otp_user = 'otp-test_user'
otp_password = 'SuperTestPassword'
otp_key = '76A3ZS6HFHBTOK2H4NDHTIVFPQ'
self.cli_set(base_path + ['user', otp_user, 'authentication', 'plaintext-password', otp_password])
self.cli_set(base_path + ['user', otp_user, 'authentication', 'otp', 'key', otp_key])
self.cli_commit()
# Check if OTP key was written properly
tmp = cmd(f'sudo head -1 /home/{otp_user}/.google_authenticator')
self.assertIn(otp_key, tmp)
self.cli_delete(base_path + ['user', otp_user])
def test_system_user_ssh_key(self):
ssh_user = 'ssh-test_user'
public_keys = 'vyos_test@domain-foo.com'
type = 'ssh-rsa'
self.cli_set(base_path + ['user', ssh_user, 'authentication', 'public-keys', public_keys, 'key', ssh_pubkey.replace('\n','')])
# check validate() - missing type for public-key
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(base_path + ['user', ssh_user, 'authentication', 'public-keys', public_keys, 'type', type])
self.cli_commit()
# Check that SSH key was written properly
tmp = cmd(f'sudo cat /home/{ssh_user}/.ssh/authorized_keys')
key = f'{type} ' + ssh_pubkey.replace('\n','')
self.assertIn(key, tmp)
self.cli_delete(base_path + ['user', ssh_user])
def test_radius_kernel_features(self):
# T2886: RADIUS requires some Kernel options to be present
kernel_config = GzipFile('/proc/config.gz').read().decode('UTF-8')
# T2886 - RADIUS authentication - check for statically compiled options
options = ['CONFIG_AUDIT', 'CONFIG_AUDITSYSCALL', 'CONFIG_AUDIT_ARCH']
for option in options:
self.assertIn(f'{option}=y', kernel_config)
def test_system_login_radius_ipv4(self):
radius_servers = ['100.64.0.4', '100.64.0.5']
radius_source = '100.64.0.1'
self._system_login_radius_test_helper(radius_servers, radius_source)
def test_system_login_radius_ipv6(self):
radius_servers = ['2001:db8::4', '2001:db8::5']
radius_source = '2001:db8::1'
self._system_login_radius_test_helper(radius_servers, radius_source)
def _system_login_radius_test_helper(self, radius_servers: list, radius_source: str):
# Verify generated RADIUS configuration files
radius_key = ''.join(secrets.choice(string.ascii_letters + string.digits) for i in range(10))
default_port = default_value(base_path + ['radius', 'server', radius_servers[0], 'port'])
default_timeout = default_value(base_path + ['radius', 'server', radius_servers[0], 'timeout'])
dummy_if = 'dum12760'
# Load container image for FreeRADIUS server
radius_config = '/tmp/smoketest-radius-server'
radius_container_path = ['container', 'name', 'radius-1']
# Generate random string with 10 digits
username = 'radius-admin'
password = ''.join(secrets.choice(string.ascii_letters + string.digits) for i in range(10))
radius_source_mask = '32'
if is_ipv6(radius_source):
radius_source_mask = '128'
radius_test_user = {
'username' : username,
'password' : password,
'radius_key' : radius_key,
'source_address' : f'{radius_source}/{radius_source_mask}'
}
tmpl = jinja2.Template(RADIUS_CLIENTS_TMPL_SRC)
write_file(f'{radius_config}/clients.cfg', tmpl.render(radius_test_user))
tmpl = jinja2.Template(RADIUS_USERS_TMPL_SRC)
write_file(f'{radius_config}/users', tmpl.render(radius_test_user))
# Start tac_plus container
self.cli_set(radius_container_path + ['allow-host-networks'])
self.cli_set(radius_container_path + ['image', radius_image])
self.cli_set(radius_container_path + ['volume', 'clients', 'destination', '/etc/raddb/clients.conf'])
self.cli_set(radius_container_path + ['volume', 'clients', 'mode', 'ro'])
self.cli_set(radius_container_path + ['volume', 'clients', 'source', f'{radius_config}/clients.cfg'])
self.cli_set(radius_container_path + ['volume', 'users', 'destination', '/etc/raddb/users'])
self.cli_set(radius_container_path + ['volume', 'users', 'mode', 'ro'])
self.cli_set(radius_container_path + ['volume', 'users', 'source', f'{radius_config}/users'])
# Start container
self.cli_commit()
# Deinfine RADIUS servers
for radius_server in radius_servers:
# Use this system as "remote" RADIUS server
dummy_address_mask = '32'
if is_ipv6(radius_server):
dummy_address_mask = '128'
self.cli_set(['interfaces', 'dummy', dummy_if, 'address', f'{radius_server}/{dummy_address_mask}'])
self.cli_set(base_path + ['radius', 'server', radius_server, 'key', radius_key])
# Define RADIUS traffic source address
self.cli_set(['interfaces', 'dummy', dummy_if, 'address', f'{radius_source}/{radius_source_mask}'])
self.cli_set(base_path + ['radius', 'source-address', radius_source])
self.cli_set(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)])
# check validate() - Only one IPv4 source-address supported
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)])
self.cli_commit()
# this file must be read with higher permissions
pam_radius_auth_conf = cmd('sudo cat /etc/pam_radius_auth.conf')
for radius_server in radius_servers:
if is_ipv6(radius_server):
# it is essential to escape the [] brackets when searching with a regex
radius_server = rf'\[{radius_server}\]'
tmp = re.findall(rf'\n?{radius_server}:{default_port}\s+{radius_key}\s+{default_timeout}\s+{radius_source}', pam_radius_auth_conf)
self.assertTrue(tmp)
# required, static options
self.assertIn('priv-lvl 15', pam_radius_auth_conf)
self.assertIn('mapped_priv_user radius_priv_user', pam_radius_auth_conf)
# PAM
pam_common_account = read_file('/etc/pam.d/common-account')
self.assertIn('pam_radius_auth.so', pam_common_account)
pam_common_auth = read_file('/etc/pam.d/common-auth')
self.assertIn('pam_radius_auth.so', pam_common_auth)
pam_common_session = read_file('/etc/pam.d/common-session')
self.assertIn('pam_radius_auth.so', pam_common_session)
pam_common_session_noninteractive = read_file('/etc/pam.d/common-session-noninteractive')
self.assertIn('pam_radius_auth.so', pam_common_session_noninteractive)
# NSS
nsswitch_conf = read_file('/etc/nsswitch.conf')
tmp = re.findall(r'passwd:\s+mapuid\s+files\s+mapname', nsswitch_conf)
self.assertTrue(tmp)
tmp = re.findall(r'group:\s+mapname\s+files', nsswitch_conf)
self.assertTrue(tmp)
# Login with proper credentials
out, err = self.ssh_send_cmd(ssh_test_command, username, password)
# verify login
self.assertFalse(err)
self.assertEqual(out, self.ssh_test_command_result)
# Login with invalid credentials
with self.assertRaises(paramiko.ssh_exception.AuthenticationException):
_, _ = self.ssh_send_cmd(ssh_test_command, username, f'{password}1')
# Remove RADIUS configuration
self.cli_delete(base_path + ['radius'])
# Remove RADIUS container
self.cli_delete(radius_container_path)
# Remove dummy interface
self.cli_delete(['interfaces', 'dummy', dummy_if])
self.cli_commit()
# Remove rendered tac_plus daemon configuration
shutil.rmtree(radius_config)
def test_system_login_max_login_session(self):
max_logins = '2'
timeout = '600'
self.cli_set(base_path + ['max-login-session', max_logins])
# 'max-login-session' must be only with 'timeout' option
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(base_path + ['timeout', timeout])
self.cli_commit()
security_limits = read_file('/etc/security/limits.d/10-vyos.conf')
self.assertIn(f'* - maxsyslogins {max_logins}', security_limits)
self.cli_delete(base_path + ['timeout'])
self.cli_delete(base_path + ['max-login-session'])
def test_system_login_tacacs(self):
tacacs_secret = ''.join(secrets.choice(string.ascii_letters + string.digits) for i in range(10))
tacacs_servers = ['100.64.0.11', '100.64.0.12']
source_address = '100.64.0.1'
dummy_if = 'dum12759'
# Load container image for lac_plus daemon
tac_plus_config = '/tmp/smoketest-tacacs-server'
tac_container_path = ['container', 'name', 'tacacs-1']
# Generate random string with 10 digits
username = 'tactest'
password = ''.join(secrets.choice(string.ascii_letters + string.digits) for i in range(10))
tac_test_user = {
'username' : username,
'password' : password,
'tacacs_secret' : tacacs_secret,
'source_address' : source_address,
}
tmpl = jinja2.Template(TAC_PLUS_TMPL_SRC)
write_file(f'{tac_plus_config}/tac_plus.cfg', tmpl.render(tac_test_user))
# Start tac_plus container
self.cli_set(tac_container_path + ['allow-host-networks'])
self.cli_set(tac_container_path + ['image', tac_image])
self.cli_set(tac_container_path + ['volume', 'config', 'destination', '/etc/tac_plus'])
self.cli_set(tac_container_path + ['volume', 'config', 'mode', 'ro'])
self.cli_set(tac_container_path + ['volume', 'config', 'source', tac_plus_config])
# Start container
self.cli_commit()
# Define TACACS traffic source address
self.cli_set(['interfaces', 'dummy', dummy_if, 'address', f'{source_address}/32'])
self.cli_set(base_path + ['tacacs', 'source-address', source_address])
# Define TACACS servers
for server in tacacs_servers:
# Use this system as "remote" TACACS server
self.cli_set(['interfaces', 'dummy', dummy_if, 'address', f'{server}/32'])
self.cli_set(base_path + ['tacacs', 'server', server, 'key', tacacs_secret])
self.cli_commit()
# NSS
nsswitch_conf = read_file('/etc/nsswitch.conf')
tmp = re.findall(r'passwd:\s+tacplus\s+files', nsswitch_conf)
self.assertTrue(tmp)
tmp = re.findall(r'group:\s+tacplus\s+files', nsswitch_conf)
self.assertTrue(tmp)
# PAM TACACS configuration
pam_tacacs_conf = read_file('/etc/tacplus_servers')
# NSS TACACS configuration
nss_tacacs_conf = read_file('/etc/tacplus_nss.conf')
# Users have individual home directories
self.assertIn('user_homedir=1', pam_tacacs_conf)
# specify services
self.assertIn('service=shell', pam_tacacs_conf)
self.assertIn('protocol=ssh', pam_tacacs_conf)
# Verify configured TACACS source address
self.assertIn(f'source_ip={source_address}', pam_tacacs_conf)
self.assertIn(f'source_ip={source_address}', nss_tacacs_conf)
# Verify configured TACACS servers
for server in tacacs_servers:
self.assertIn(f'secret={tacacs_secret}', pam_tacacs_conf)
self.assertIn(f'server={server}', pam_tacacs_conf)
self.assertIn(f'secret={tacacs_secret}', nss_tacacs_conf)
self.assertIn(f'server={server}', nss_tacacs_conf)
# Login with proper credentials
out, err = self.ssh_send_cmd(ssh_test_command, username, password)
# verify login
self.assertFalse(err)
self.assertEqual(out, self.ssh_test_command_result)
# Login with invalid credentials
with self.assertRaises(paramiko.ssh_exception.AuthenticationException):
_, _ = self.ssh_send_cmd(ssh_test_command, username, f'{password}1')
# Remove TACACS configuration
self.cli_delete(base_path + ['tacacs'])
# Remove tac_plus container
self.cli_delete(tac_container_path)
# Remove dummy interface
self.cli_delete(['interfaces', 'dummy', dummy_if])
self.cli_commit()
# Remove rendered tac_plus daemon configuration
shutil.rmtree(tac_plus_config)
def test_delete_current_user(self):
current_user = get_current_user()
# We are not allowed to delete the current user
self.cli_delete(base_path + ['user', current_user])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_discard()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/conf_mode/system_login.py b/src/conf_mode/system_login.py
index d3a969d9b..1e6061ecf 100755
--- a/src/conf_mode/system_login.py
+++ b/src/conf_mode/system_login.py
@@ -1,420 +1,439 @@
#!/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
+import warnings
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.configverify import verify_vrf
from vyos.template import render
from vyos.template import is_ipv4
-from vyos.utils.auth import get_current_user
+from vyos.utils.auth import (
+ DEFAULT_PASSWORD,
+ EPasswdStrength,
+ evaluate_strength,
+ 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
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
+
+ if plaintext_password is not None:
+ result = evaluate_strength(plaintext_password)
+ if result['strength'] == EPasswdStrength.WEAK:
+ 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')
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(f'pam-auth-update --enable mfa-google-authenticator')
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/op_mode/image_installer.py b/src/op_mode/image_installer.py
index 609b0b347..c6e9c7f6f 100755
--- a/src/op_mode/image_installer.py
+++ b/src/op_mode/image_installer.py
@@ -1,1092 +1,1111 @@
#!/usr/bin/env python3
#
# Copyright 2023-2025 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This file is part of VyOS.
#
# VyOS is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# VyOS 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
# VyOS. If not, see <https://www.gnu.org/licenses/>.
from argparse import ArgumentParser, Namespace
from pathlib import Path
from shutil import copy, chown, rmtree, copytree
from glob import glob
from sys import exit
from os import environ
from os import readlink
from os import getpid, getppid
from typing import Union
from urllib.parse import urlparse
from passlib.hosts import linux_context
from errno import ENOSPC
from psutil import disk_partitions
+from vyos.base import Warning
from vyos.configtree import ConfigTree
from vyos.remote import download
from vyos.system import disk, grub, image, compat, raid, SYSTEM_CFG_VER
from vyos.template import render
+from vyos.utils.auth import (
+ DEFAULT_PASSWORD,
+ EPasswdStrength,
+ evaluate_strength
+)
from vyos.utils.io import ask_input, ask_yes_no, select_entry
from vyos.utils.file import chmod_2775
from vyos.utils.process import cmd, run, rc_cmd
from vyos.version import get_version_data
# define text messages
MSG_ERR_NOT_LIVE: str = 'The system is already installed. Please use "add system image" instead.'
MSG_ERR_LIVE: str = 'The system is in live-boot mode. Please use "install image" instead.'
MSG_ERR_NO_DISK: str = 'No suitable disk was found. There must be at least one disk of 2GB or greater size.'
MSG_ERR_IMPROPER_IMAGE: str = 'Missing sha256sum.txt.\nEither this image is corrupted, or of era 1.2.x (md5sum) and would downgrade image tools;\ndisallowed in either case.'
MSG_ERR_INCOMPATIBLE_IMAGE: str = 'Image compatibility check failed, aborting installation.'
MSG_ERR_ARCHITECTURE_MISMATCH: str = 'The current architecture is "{0}", the new image is for "{1}". Upgrading to a different image architecture will break your system.'
MSG_ERR_FLAVOR_MISMATCH: str = 'The current image flavor is "{0}", the new image is "{1}". Upgrading to a non-matching flavor can have unpredictable consequences.'
MSG_ERR_MISSING_ARCHITECTURE: str = 'The new image version data does not specify architecture, cannot check compatibility (is it a legacy release image?)'
MSG_ERR_MISSING_FLAVOR: str = 'The new image version data does not specify flavor, cannot check compatibility (is it a legacy release image?)'
MSG_ERR_CORRUPT_CURRENT_IMAGE: str = 'Version data in the current image is malformed: missing flavor and/or architecture fields. Upgrade compatibility cannot be checked.'
MSG_INFO_INSTALL_WELCOME: str = 'Welcome to VyOS installation!\nThis command will install VyOS to your permanent storage.'
MSG_INFO_INSTALL_EXIT: str = 'Exiting from VyOS installation'
MSG_INFO_INSTALL_SUCCESS: str = 'The image installed successfully; please reboot now.'
MSG_INFO_INSTALL_DISKS_LIST: str = 'The following disks were found:'
MSG_INFO_INSTALL_DISK_SELECT: str = 'Which one should be used for installation?'
MSG_INFO_INSTALL_RAID_CONFIGURE: str = 'Would you like to configure RAID-1 mirroring?'
MSG_INFO_INSTALL_RAID_FOUND_DISKS: str = 'Would you like to configure RAID-1 mirroring on them?'
MSG_INFO_INSTALL_RAID_CHOOSE_DISKS: str = 'Would you like to choose two disks for RAID-1 mirroring?'
MSG_INFO_INSTALL_DISK_CONFIRM: str = 'Installation will delete all data on the drive. Continue?'
MSG_INFO_INSTALL_RAID_CONFIRM: str = 'Installation will delete all data on both drives. Continue?'
MSG_INFO_INSTALL_PARTITONING: str = 'Creating partition table...'
MSG_INPUT_CONFIG_FOUND: str = 'An active configuration was found. Would you like to copy it to the new image?'
MSG_INPUT_CONFIG_CHOICE: str = 'The following config files are available for boot:'
MSG_INPUT_CONFIG_CHOOSE: str = 'Which file would you like as boot config?'
MSG_INPUT_IMAGE_NAME: str = 'What would you like to name this image?'
MSG_INPUT_IMAGE_DEFAULT: str = 'Would you like to set the new image as the default one for boot?'
MSG_INPUT_PASSWORD: str = 'Please enter a password for the "vyos" user:'
MSG_INPUT_PASSWORD_CONFIRM: str = 'Please confirm password for the "vyos" user:'
MSG_INPUT_ROOT_SIZE_ALL: str = 'Would you like to use all the free space on the drive?'
MSG_INPUT_ROOT_SIZE_SET: str = 'Please specify the size (in GB) of the root partition (min is 1.5 GB)?'
MSG_INPUT_CONSOLE_TYPE: str = 'What console should be used by default? (K: KVM, S: Serial)?'
MSG_INPUT_COPY_DATA: str = 'Would you like to copy data to the new image?'
MSG_INPUT_CHOOSE_COPY_DATA: str = 'From which image would you like to save config information?'
MSG_INPUT_COPY_ENC_DATA: str = 'Would you like to copy the encrypted config to the new image?'
MSG_INPUT_CHOOSE_COPY_ENC_DATA: str = 'From which image would you like to copy the encrypted config?'
MSG_WARN_ISO_SIGN_INVALID: str = 'Signature is not valid. Do you want to continue with installation?'
MSG_WARN_ISO_SIGN_UNAVAL: str = 'Signature is not available. Do you want to continue with installation?'
MSG_WARN_ROOT_SIZE_TOOBIG: str = 'The size is too big. Try again.'
MSG_WARN_ROOT_SIZE_TOOSMALL: str = 'The size is too small. Try again'
MSG_WARN_IMAGE_NAME_WRONG: str = 'The suggested name is unsupported!\n'\
'It must be between 1 and 64 characters long and contains only the next characters: .+-_ a-z A-Z 0-9'
+
+MSG_WARN_CHANGE_PASSWORD: str = 'Default password used. Consider changing ' \
+ 'it on next login.'
MSG_WARN_PASSWORD_CONFIRM: str = 'The entered values did not match. Try again'
'Installing a different image flavor may cause functionality degradation or break your system.\n' \
'Do you want to continue with installation?'
CONST_MIN_DISK_SIZE: int = 2147483648 # 2 GB
CONST_MIN_ROOT_SIZE: int = 1610612736 # 1.5 GB
# a reserved space: 2MB for header, 1 MB for BIOS partition, 256 MB for EFI
CONST_RESERVED_SPACE: int = (2 + 1 + 256) * 1024**2
# define directories and paths
DIR_INSTALLATION: str = '/mnt/installation'
DIR_ROOTFS_SRC: str = f'{DIR_INSTALLATION}/root_src'
DIR_ROOTFS_DST: str = f'{DIR_INSTALLATION}/root_dst'
DIR_ISO_MOUNT: str = f'{DIR_INSTALLATION}/iso_src'
DIR_DST_ROOT: str = f'{DIR_INSTALLATION}/disk_dst'
DIR_KERNEL_SRC: str = '/boot/'
FILE_ROOTFS_SRC: str = '/usr/lib/live/mount/medium/live/filesystem.squashfs'
ISO_DOWNLOAD_PATH: str = ''
external_download_script = '/usr/libexec/vyos/simple-download.py'
external_latest_image_url_script = '/usr/libexec/vyos/latest-image-url.py'
# default boot variables
DEFAULT_BOOT_VARS: dict[str, str] = {
'timeout': '5',
'console_type': 'tty',
'console_num': '0',
'console_speed': '115200',
'bootmode': 'normal'
}
def bytes_to_gb(size: int) -> float:
"""Convert Bytes to GBytes, rounded to 1 decimal number
Args:
size (int): input size in bytes
Returns:
float: size in GB
"""
return round(size / 1024**3, 1)
def gb_to_bytes(size: float) -> int:
"""Convert GBytes to Bytes
Args:
size (float): input size in GBytes
Returns:
int: size in bytes
"""
return int(size * 1024**3)
def find_disks() -> dict[str, int]:
"""Find a target disk for installation
Returns:
dict[str, int]: a list of available disks by name and size
"""
# check for available disks
print('Probing disks')
disks_available: dict[str, int] = disk.disks_size()
for disk_name, disk_size in disks_available.copy().items():
if disk_size < CONST_MIN_DISK_SIZE:
del disks_available[disk_name]
if not disks_available:
print(MSG_ERR_NO_DISK)
exit(MSG_INFO_INSTALL_EXIT)
num_disks: int = len(disks_available)
print(f'{num_disks} disk(s) found')
return disks_available
def ask_root_size(available_space: int) -> int:
"""Define a size of root partition
Args:
available_space (int): available space in bytes for a root partition
Returns:
int: defined size
"""
if ask_yes_no(MSG_INPUT_ROOT_SIZE_ALL, default=True):
return available_space
while True:
root_size_gb: str = ask_input(MSG_INPUT_ROOT_SIZE_SET)
root_size_kbytes: int = (gb_to_bytes(float(root_size_gb))) // 1024
if root_size_kbytes > available_space:
print(MSG_WARN_ROOT_SIZE_TOOBIG)
continue
if root_size_kbytes < CONST_MIN_ROOT_SIZE / 1024:
print(MSG_WARN_ROOT_SIZE_TOOSMALL)
continue
return root_size_kbytes
def create_partitions(target_disk: str, target_size: int,
prompt: bool = True) -> None:
"""Create partitions on a target disk
Args:
target_disk (str): a target disk
target_size (int): size of disk in bytes
"""
# define target rootfs size in KB (smallest unit acceptable by sgdisk)
available_size: int = (target_size - CONST_RESERVED_SPACE) // 1024
if prompt:
rootfs_size: int = ask_root_size(available_size)
else:
rootfs_size: int = available_size
print(MSG_INFO_INSTALL_PARTITONING)
raid.clear()
disk.disk_cleanup(target_disk)
disk_details: disk.DiskDetails = disk.parttable_create(target_disk,
rootfs_size)
return disk_details
def search_format_selection(image: tuple[str, str]) -> str:
"""Format a string for selection of image
Args:
image (tuple[str, str]): a tuple of image name and drive
Returns:
str: formatted string
"""
return f'{image[0]} on {image[1]}'
def search_previous_installation(disks: list[str]) -> None:
"""Search disks for previous installation config and SSH keys
Args:
disks (list[str]): a list of available disks
"""
mnt_config = '/mnt/config'
mnt_encrypted_config = '/mnt/encrypted_config'
mnt_ssh = '/mnt/ssh'
mnt_tmp = '/mnt/tmp'
rmtree(Path(mnt_config), ignore_errors=True)
rmtree(Path(mnt_ssh), ignore_errors=True)
Path(mnt_tmp).mkdir(exist_ok=True)
Path(mnt_encrypted_config).unlink(missing_ok=True)
print('Searching for data from previous installations')
image_data = []
encrypted_configs = []
for disk_name in disks:
for partition in disk.partition_list(disk_name):
if disk.partition_mount(partition, mnt_tmp):
if Path(mnt_tmp + '/boot').exists():
for path in Path(mnt_tmp + '/boot').iterdir():
if path.joinpath('rw/config/.vyatta_config').exists():
image_data.append((path.name, partition))
if Path(mnt_tmp + '/luks').exists():
for path in Path(mnt_tmp + '/luks').iterdir():
encrypted_configs.append((path.name, partition))
disk.partition_umount(partition)
image_name = None
image_drive = None
encrypted = False
if len(image_data) > 0:
if len(image_data) == 1:
print('Found data from previous installation:')
print(f'\t{" on ".join(image_data[0])}')
if ask_yes_no(MSG_INPUT_COPY_DATA, default=True):
image_name, image_drive = image_data[0]
elif len(image_data) > 1:
print('Found data from previous installations')
if ask_yes_no(MSG_INPUT_COPY_DATA, default=True):
image_name, image_drive = select_entry(image_data,
'Available versions:',
MSG_INPUT_CHOOSE_COPY_DATA,
search_format_selection)
elif len(encrypted_configs) > 0:
if len(encrypted_configs) == 1:
print('Found encrypted config from previous installation:')
print(f'\t{" on ".join(encrypted_configs[0])}')
if ask_yes_no(MSG_INPUT_COPY_ENC_DATA, default=True):
image_name, image_drive = encrypted_configs[0]
encrypted = True
elif len(encrypted_configs) > 1:
print('Found encrypted configs from previous installations')
if ask_yes_no(MSG_INPUT_COPY_ENC_DATA, default=True):
image_name, image_drive = select_entry(encrypted_configs,
'Available versions:',
MSG_INPUT_CHOOSE_COPY_ENC_DATA,
search_format_selection)
encrypted = True
else:
print('No previous installation found')
return
if not image_name:
return
disk.partition_mount(image_drive, mnt_tmp)
if not encrypted:
copytree(f'{mnt_tmp}/boot/{image_name}/rw/config', mnt_config)
else:
copy(f'{mnt_tmp}/luks/{image_name}', mnt_encrypted_config)
Path(mnt_ssh).mkdir()
host_keys: list[str] = glob(f'{mnt_tmp}/boot/{image_name}/rw/etc/ssh/ssh_host*')
for host_key in host_keys:
copy(host_key, mnt_ssh)
disk.partition_umount(image_drive)
def copy_preserve_owner(src: str, dst: str, *, follow_symlinks=True):
if not Path(src).is_file():
return
if Path(dst).is_dir():
dst = Path(dst).joinpath(Path(src).name)
st = Path(src).stat()
copy(src, dst, follow_symlinks=follow_symlinks)
chown(dst, user=st.st_uid)
def copy_previous_installation_data(target_dir: str) -> None:
if Path('/mnt/config').exists():
copytree('/mnt/config', f'{target_dir}/opt/vyatta/etc/config',
dirs_exist_ok=True)
if Path('/mnt/ssh').exists():
copytree('/mnt/ssh', f'{target_dir}/etc/ssh',
dirs_exist_ok=True)
def copy_previous_encrypted_config(target_dir: str, image_name: str) -> None:
if Path('/mnt/encrypted_config').exists():
Path(target_dir).mkdir(exist_ok=True)
copy('/mnt/encrypted_config', Path(target_dir).joinpath(image_name))
def ask_single_disk(disks_available: dict[str, int]) -> str:
"""Ask user to select a disk for installation
Args:
disks_available (dict[str, int]): a list of available disks
"""
print(MSG_INFO_INSTALL_DISKS_LIST)
default_disk: str = list(disks_available)[0]
for disk_name, disk_size in disks_available.items():
disk_size_human: str = bytes_to_gb(disk_size)
print(f'Drive: {disk_name} ({disk_size_human} GB)')
disk_selected: str = ask_input(MSG_INFO_INSTALL_DISK_SELECT,
default=default_disk,
valid_responses=list(disks_available))
# create partitions
if not ask_yes_no(MSG_INFO_INSTALL_DISK_CONFIRM):
print(MSG_INFO_INSTALL_EXIT)
exit()
search_previous_installation(list(disks_available))
disk_details: disk.DiskDetails = create_partitions(disk_selected,
disks_available[disk_selected])
disk.filesystem_create(disk_details.partition['efi'], 'efi')
disk.filesystem_create(disk_details.partition['root'], 'ext4')
return disk_details
def check_raid_install(disks_available: dict[str, int]) -> Union[str, None]:
"""Ask user to select disks for RAID installation
Args:
disks_available (dict[str, int]): a list of available disks
"""
if len(disks_available) < 2:
return None
if not ask_yes_no(MSG_INFO_INSTALL_RAID_CONFIGURE, default=True):
return None
def format_selection(disk_name: str) -> str:
return f'{disk_name}\t({bytes_to_gb(disks_available[disk_name])} GB)'
disk0, disk1 = list(disks_available)[0], list(disks_available)[1]
disks_selected: dict[str, int] = { disk0: disks_available[disk0],
disk1: disks_available[disk1] }
target_size: int = min(disks_selected[disk0], disks_selected[disk1])
print(MSG_INFO_INSTALL_DISKS_LIST)
for disk_name, disk_size in disks_selected.items():
disk_size_human: str = bytes_to_gb(disk_size)
print(f'\t{disk_name} ({disk_size_human} GB)')
if not ask_yes_no(MSG_INFO_INSTALL_RAID_FOUND_DISKS, default=True):
if not ask_yes_no(MSG_INFO_INSTALL_RAID_CHOOSE_DISKS, default=True):
return None
else:
disks_selected = {}
disk0 = select_entry(list(disks_available), 'Disks available:',
'Select first disk:', format_selection)
disks_selected[disk0] = disks_available[disk0]
del disks_available[disk0]
disk1 = select_entry(list(disks_available), 'Remaining disks:',
'Select second disk:', format_selection)
disks_selected[disk1] = disks_available[disk1]
target_size: int = min(disks_selected[disk0],
disks_selected[disk1])
# create partitions
if not ask_yes_no(MSG_INFO_INSTALL_RAID_CONFIRM):
print(MSG_INFO_INSTALL_EXIT)
exit()
search_previous_installation(list(disks_available))
disks: list[disk.DiskDetails] = []
for disk_selected in list(disks_selected):
print(f'Creating partitions on {disk_selected}')
disk_details = create_partitions(disk_selected, target_size,
prompt=False)
disk.filesystem_create(disk_details.partition['efi'], 'efi')
disks.append(disk_details)
print('Creating RAID array')
members = [disk.partition['root'] for disk in disks]
raid_details: raid.RaidDetails = raid.raid_create(members)
# raid init stuff
print('Updating initramfs')
raid.update_initramfs()
# end init
print('Creating filesystem on RAID array')
disk.filesystem_create(raid_details.name, 'ext4')
return raid_details
def prepare_tmp_disr() -> None:
"""Create temporary directories for installation
"""
print('Creating temporary directories')
for dir in [DIR_ROOTFS_SRC, DIR_ROOTFS_DST, DIR_DST_ROOT]:
dirpath = Path(dir)
dirpath.mkdir(mode=0o755, parents=True)
def setup_grub(root_dir: str) -> None:
"""Install GRUB configurations
Args:
root_dir (str): a path to the root of target filesystem
"""
print('Installing GRUB configuration files')
grub_cfg_main = f'{root_dir}/{grub.GRUB_DIR_MAIN}/grub.cfg'
grub_cfg_vars = f'{root_dir}/{grub.CFG_VYOS_VARS}'
grub_cfg_modules = f'{root_dir}/{grub.CFG_VYOS_MODULES}'
grub_cfg_menu = f'{root_dir}/{grub.CFG_VYOS_MENU}'
grub_cfg_options = f'{root_dir}/{grub.CFG_VYOS_OPTIONS}'
# create new files
render(grub_cfg_main, grub.TMPL_GRUB_MAIN, {})
grub.common_write(root_dir)
grub.vars_write(grub_cfg_vars, DEFAULT_BOOT_VARS)
grub.modules_write(grub_cfg_modules, [])
grub.write_cfg_ver(1, root_dir)
render(grub_cfg_menu, grub.TMPL_GRUB_MENU, {})
render(grub_cfg_options, grub.TMPL_GRUB_OPTS, {})
def configure_authentication(config_file: str, password: str) -> None:
"""Write encrypted password to config file
Args:
config_file (str): path of target config file
password (str): plaintext password
N.B. this can not be deferred by simply setting the plaintext password
and relying on the config mode script to process at boot, as the config
will not automatically be saved in that case, thus leaving the
plaintext exposed
"""
encrypted_password = linux_context.hash(password)
with open(config_file) as f:
config_string = f.read()
config = ConfigTree(config_string)
config.set([
'system', 'login', 'user', 'vyos', 'authentication',
'encrypted-password'
],
value=encrypted_password,
replace=True)
config.set_tag(['system', 'login', 'user'])
with open(config_file, 'w') as f:
f.write(config.to_string())
def validate_signature(file_path: str, sign_type: str) -> None:
"""Validate a file by signature and delete a signature file
Args:
file_path (str): a path to file
sign_type (str): a signature type
"""
print('Validating signature')
signature_valid: bool = False
# validate with minisig
if sign_type == 'minisig':
pub_key_list = glob('/usr/share/vyos/keys/*.minisign.pub')
for pubkey in pub_key_list:
if run(f'minisign -V -q -p {pubkey} -m {file_path} -x {file_path}.minisig'
) == 0:
signature_valid = True
break
Path(f'{file_path}.minisig').unlink()
# validate with GPG
if sign_type == 'asc':
if run(f'gpg --verify ${file_path}.asc ${file_path}') == 0:
signature_valid = True
Path(f'{file_path}.asc').unlink()
# warn or pass
if not signature_valid:
if not ask_yes_no(MSG_WARN_ISO_SIGN_INVALID, default=False):
exit(MSG_INFO_INSTALL_EXIT)
else:
print('Signature is valid')
def download_file(local_file: str, remote_path: str, vrf: str,
username: str, password: str,
progressbar: bool = False, check_space: bool = False):
environ['REMOTE_USERNAME'] = username
environ['REMOTE_PASSWORD'] = password
if vrf is None:
download(local_file, remote_path, progressbar=progressbar,
check_space=check_space, raise_error=True)
else:
remote_auth = f'REMOTE_USERNAME={username} REMOTE_PASSWORD={password}'
vrf_cmd = f'ip vrf exec {vrf} {external_download_script} \
--local-file {local_file} --remote-path {remote_path}'
cmd(vrf_cmd, auth=remote_auth)
def image_fetch(image_path: str, vrf: str = None,
username: str = '', password: str = '',
no_prompt: bool = False) -> Path:
"""Fetch an ISO image
Args:
image_path (str): a path, remote or local
Returns:
Path: a path to a local file
"""
import os.path
from uuid import uuid4
global ISO_DOWNLOAD_PATH
# Latest version gets url from configured "system update-check url"
if image_path == 'latest':
command = external_latest_image_url_script
if vrf:
command = f'REMOTE_USERNAME={username} REMOTE_PASSWORD={password} \
ip vrf exec {vrf} ' + command
code, output = rc_cmd(command)
if code:
print(output)
exit(MSG_INFO_INSTALL_EXIT)
image_path = output if output else image_path
try:
# check a type of path
if urlparse(image_path).scheme:
# download an image
ISO_DOWNLOAD_PATH = os.path.join(os.path.expanduser("~"), '{0}.iso'.format(uuid4()))
download_file(ISO_DOWNLOAD_PATH, image_path, vrf,
username, password,
progressbar=True, check_space=True)
# download a signature
sign_file = (False, '')
for sign_type in ['minisig', 'asc']:
try:
download_file(f'{ISO_DOWNLOAD_PATH}.{sign_type}',
f'{image_path}.{sign_type}', vrf,
username, password)
sign_file = (True, sign_type)
break
except Exception:
print(f'{sign_type} signature is not available')
# validate a signature if it is available
if sign_file[0]:
validate_signature(ISO_DOWNLOAD_PATH, sign_file[1])
else:
if (not no_prompt and
not ask_yes_no(MSG_WARN_ISO_SIGN_UNAVAL, default=False)):
cleanup()
exit(MSG_INFO_INSTALL_EXIT)
return Path(ISO_DOWNLOAD_PATH)
else:
local_path: Path = Path(image_path)
if local_path.is_file():
return local_path
else:
raise FileNotFoundError
except Exception as e:
print(f'The image cannot be fetched from: {image_path} {e}')
exit(1)
def migrate_config() -> bool:
"""Check for active config and ask user for migration
Returns:
bool: user's decision
"""
active_config_path: Path = Path('/opt/vyatta/etc/config/config.boot')
if active_config_path.exists():
if ask_yes_no(MSG_INPUT_CONFIG_FOUND, default=True):
return True
return False
def copy_ssh_host_keys() -> bool:
"""Ask user to copy SSH host keys
Returns:
bool: user's decision
"""
if ask_yes_no('Would you like to copy SSH host keys?', default=True):
return True
return False
def console_hint() -> str:
pid = getppid() if 'SUDO_USER' in environ else getpid()
try:
path = readlink(f'/proc/{pid}/fd/1')
except OSError:
path = '/dev/tty'
name = Path(path).name
if name == 'ttyS0':
return 'S'
else:
return 'K'
def cleanup(mounts: list[str] = [], remove_items: list[str] = []) -> None:
"""Clean up after installation
Args:
mounts (list[str], optional): List of mounts to unmount.
Defaults to [].
remove_items (list[str], optional): List of files or directories
to remove. Defaults to [].
"""
print('Cleaning up')
# clean up installation directory by default
mounts_all = disk_partitions(all=True)
for mounted_device in mounts_all:
if mounted_device.mountpoint.startswith(DIR_INSTALLATION) and not (
mounted_device.device in mounts or
mounted_device.mountpoint in mounts):
mounts.append(mounted_device.mountpoint)
# add installation dir to cleanup list
if DIR_INSTALLATION not in remove_items:
remove_items.append(DIR_INSTALLATION)
# also delete an ISO file
if Path(ISO_DOWNLOAD_PATH).exists(
) and ISO_DOWNLOAD_PATH not in remove_items:
remove_items.append(ISO_DOWNLOAD_PATH)
if mounts:
print('Unmounting target filesystems')
for mountpoint in mounts:
disk.partition_umount(mountpoint)
for mountpoint in mounts:
disk.wait_for_umount(mountpoint)
if remove_items:
print('Removing temporary files')
for remove_item in remove_items:
if Path(remove_item).exists():
if Path(remove_item).is_file():
Path(remove_item).unlink()
if Path(remove_item).is_dir():
rmtree(remove_item, ignore_errors=True)
def cleanup_raid(details: raid.RaidDetails) -> None:
efiparts = []
for raid_disk in details.disks:
efiparts.append(raid_disk.partition['efi'])
cleanup([details.name, *efiparts],
['/mnt/installation'])
def is_raid_install(install_object: Union[disk.DiskDetails, raid.RaidDetails]) -> bool:
"""Check if installation target is a RAID array
Args:
install_object (Union[disk.DiskDetails, raid.RaidDetails]): a target disk
Returns:
bool: True if it is a RAID array
"""
if isinstance(install_object, raid.RaidDetails):
return True
return False
def validate_compatibility(iso_path: str, force: bool = False) -> None:
"""Check architecture and flavor compatibility with the running image
Args:
iso_path (str): a path to the mounted ISO image
"""
current_data = get_version_data()
current_flavor = current_data.get('flavor')
current_architecture = current_data.get('architecture') or cmd('dpkg --print-architecture')
new_data = get_version_data(f'{iso_path}/version.json')
new_flavor = new_data.get('flavor')
new_architecture = new_data.get('architecture')
if not current_flavor or not current_architecture:
# This may only happen if someone modified the version file.
# Unlikely but not impossible.
print(MSG_ERR_CORRUPT_CURRENT_IMAGE)
cleanup()
exit(MSG_INFO_INSTALL_EXIT)
success = True
if current_architecture != new_architecture:
success = False
if not new_architecture:
print(MSG_ERR_MISSING_ARCHITECTURE)
else:
print(MSG_ERR_ARCHITECTURE_MISMATCH.format(current_architecture, new_architecture))
if current_flavor != new_flavor:
if not force:
success = False
if not new_flavor:
print(MSG_ERR_MISSING_FLAVOR)
else:
print(MSG_ERR_FLAVOR_MISMATCH.format(current_flavor, new_flavor))
if not success:
print(MSG_ERR_INCOMPATIBLE_IMAGE)
cleanup()
exit(MSG_INFO_INSTALL_EXIT)
def install_image() -> None:
"""Install an image to a disk
"""
if not image.is_live_boot():
exit(MSG_ERR_NOT_LIVE)
print(MSG_INFO_INSTALL_WELCOME)
if not ask_yes_no('Would you like to continue?'):
print(MSG_INFO_INSTALL_EXIT)
exit()
# configure image name
running_image_name: str = image.get_running_image()
while True:
image_name: str = ask_input(MSG_INPUT_IMAGE_NAME,
running_image_name)
if image.validate_name(image_name):
break
print(MSG_WARN_IMAGE_NAME_WRONG)
# ask for password
while True:
user_password: str = ask_input(MSG_INPUT_PASSWORD, no_echo=True,
non_empty=True)
+
+ if user_password == DEFAULT_PASSWORD:
+ Warning(MSG_WARN_CHANGE_PASSWORD)
+ else:
+ result = evaluate_strength(user_password)
+ if result['strength'] == EPasswdStrength.WEAK:
+ Warning(result['error'])
+
confirm: str = ask_input(MSG_INPUT_PASSWORD_CONFIRM, no_echo=True,
non_empty=True)
+
if user_password == confirm:
break
+
print(MSG_WARN_PASSWORD_CONFIRM)
# ask for default console
console_type: str = ask_input(MSG_INPUT_CONSOLE_TYPE,
default=console_hint(),
valid_responses=['K', 'S'])
console_dict: dict[str, str] = {'K': 'tty', 'S': 'ttyS'}
config_boot_list = ['/opt/vyatta/etc/config/config.boot',
'/opt/vyatta/etc/config.boot.default']
default_config = config_boot_list[0]
disks: dict[str, int] = find_disks()
install_target: Union[disk.DiskDetails, raid.RaidDetails, None] = None
try:
install_target = check_raid_install(disks)
if install_target is None:
install_target = ask_single_disk(disks)
# if previous install was selected in search_previous_installation,
# directory /mnt/config was prepared for copy below; if not, prompt:
if not Path('/mnt/config').exists():
default_config: str = select_entry(config_boot_list,
MSG_INPUT_CONFIG_CHOICE,
MSG_INPUT_CONFIG_CHOOSE,
default_entry=1) # select_entry indexes from 1
# create directories for installation media
prepare_tmp_disr()
# mount target filesystem and create required dirs inside
print('Mounting new partitions')
if is_raid_install(install_target):
disk.partition_mount(install_target.name, DIR_DST_ROOT)
Path(f'{DIR_DST_ROOT}/boot/efi').mkdir(parents=True)
else:
disk.partition_mount(install_target.partition['root'], DIR_DST_ROOT)
Path(f'{DIR_DST_ROOT}/boot/efi').mkdir(parents=True)
disk.partition_mount(install_target.partition['efi'], f'{DIR_DST_ROOT}/boot/efi')
# a config dir. It is the deepest one, so the comand will
# create all the rest in a single step
print('Creating a configuration file')
target_config_dir: str = f'{DIR_DST_ROOT}/boot/{image_name}/rw/opt/vyatta/etc/config/'
Path(target_config_dir).mkdir(parents=True)
chown(target_config_dir, group='vyattacfg')
chmod_2775(target_config_dir)
# copy config
copy(default_config, f'{target_config_dir}/config.boot')
configure_authentication(f'{target_config_dir}/config.boot',
user_password)
Path(f'{target_config_dir}/.vyatta_config').touch()
# create a persistence.conf
Path(f'{DIR_DST_ROOT}/persistence.conf').write_text('/ union\n')
# copy system image and kernel files
print('Copying system image files')
for file in Path(DIR_KERNEL_SRC).iterdir():
if file.is_file():
copy(file, f'{DIR_DST_ROOT}/boot/{image_name}/')
copy(FILE_ROOTFS_SRC,
f'{DIR_DST_ROOT}/boot/{image_name}/{image_name}.squashfs')
# copy saved config data and SSH keys
# owner restored on copy of config data by chmod_2775, above
copy_previous_installation_data(f'{DIR_DST_ROOT}/boot/{image_name}/rw')
# copy saved encrypted config volume
copy_previous_encrypted_config(f'{DIR_DST_ROOT}/luks', image_name)
if is_raid_install(install_target):
write_dir: str = f'{DIR_DST_ROOT}/boot/{image_name}/rw'
raid.update_default(write_dir)
setup_grub(DIR_DST_ROOT)
# add information about version
grub.create_structure()
grub.version_add(image_name, DIR_DST_ROOT)
grub.set_default(image_name, DIR_DST_ROOT)
grub.set_console_type(console_dict[console_type], DIR_DST_ROOT)
if is_raid_install(install_target):
# add RAID specific modules
grub.modules_write(f'{DIR_DST_ROOT}/{grub.CFG_VYOS_MODULES}',
['part_msdos', 'part_gpt', 'diskfilter',
'ext2','mdraid1x'])
# install GRUB
if is_raid_install(install_target):
print('Installing GRUB to the drives')
l = install_target.disks
for disk_target in l:
disk.partition_mount(disk_target.partition['efi'], f'{DIR_DST_ROOT}/boot/efi')
grub.install(disk_target.name, f'{DIR_DST_ROOT}/boot/',
f'{DIR_DST_ROOT}/boot/efi',
id=f'VyOS (RAID disk {l.index(disk_target) + 1})')
disk.partition_umount(disk_target.partition['efi'])
else:
print('Installing GRUB to the drive')
grub.install(install_target.name, f'{DIR_DST_ROOT}/boot/',
f'{DIR_DST_ROOT}/boot/efi')
# sort inodes (to make GRUB read config files in alphabetical order)
grub.sort_inodes(f'{DIR_DST_ROOT}/{grub.GRUB_DIR_VYOS}')
grub.sort_inodes(f'{DIR_DST_ROOT}/{grub.GRUB_DIR_VYOS_VERS}')
# umount filesystems and remove temporary files
if is_raid_install(install_target):
cleanup([install_target.name],
['/mnt/installation'])
else:
cleanup([install_target.partition['efi'],
install_target.partition['root']],
['/mnt/installation'])
# we are done
print(MSG_INFO_INSTALL_SUCCESS)
exit()
except Exception as err:
print(f'Unable to install VyOS: {err}')
# unmount filesystems and clenup
try:
if install_target is not None:
if is_raid_install(install_target):
cleanup_raid(install_target)
else:
cleanup([install_target.partition['efi'],
install_target.partition['root']],
['/mnt/installation'])
except Exception as err:
print(f'Cleanup failed: {err}')
exit(1)
@compat.grub_cfg_update
def add_image(image_path: str, vrf: str = None, username: str = '',
password: str = '', no_prompt: bool = False, force: bool = False) -> None:
"""Add a new image
Args:
image_path (str): a path to an ISO image
"""
if image.is_live_boot():
exit(MSG_ERR_LIVE)
# fetch an image
iso_path: Path = image_fetch(image_path, vrf, username, password, no_prompt)
try:
# mount an ISO
Path(DIR_ISO_MOUNT).mkdir(mode=0o755, parents=True)
disk.partition_mount(iso_path, DIR_ISO_MOUNT, 'iso9660')
print('Validating image compatibility')
validate_compatibility(DIR_ISO_MOUNT, force=force)
# check sums
print('Validating image checksums')
if not Path(DIR_ISO_MOUNT).joinpath('sha256sum.txt').exists():
cleanup()
exit(MSG_ERR_IMPROPER_IMAGE)
if run(f'cd {DIR_ISO_MOUNT} && sha256sum --status -c sha256sum.txt'):
cleanup()
exit('Image checksum verification failed.')
# mount rootfs (to get a system version)
Path(DIR_ROOTFS_SRC).mkdir(mode=0o755, parents=True)
disk.partition_mount(f'{DIR_ISO_MOUNT}/live/filesystem.squashfs',
DIR_ROOTFS_SRC, 'squashfs')
cfg_ver: str = image.get_image_tools_version(DIR_ROOTFS_SRC)
version_name: str = image.get_image_version(DIR_ROOTFS_SRC)
disk.partition_umount(f'{DIR_ISO_MOUNT}/live/filesystem.squashfs')
if cfg_ver < SYSTEM_CFG_VER:
raise compat.DowngradingImageTools(
f'Adding image would downgrade image tools to v.{cfg_ver}; disallowed')
if not no_prompt:
while True:
image_name: str = ask_input(MSG_INPUT_IMAGE_NAME, version_name)
if image.validate_name(image_name):
break
print(MSG_WARN_IMAGE_NAME_WRONG)
set_as_default: bool = ask_yes_no(MSG_INPUT_IMAGE_DEFAULT, default=True)
else:
image_name: str = version_name
set_as_default: bool = True
# find target directory
root_dir: str = disk.find_persistence()
# a config dir. It is the deepest one, so the comand will
# create all the rest in a single step
target_config_dir: str = f'{root_dir}/boot/{image_name}/rw/opt/vyatta/etc/config/'
# copy config
if no_prompt or migrate_config():
print('Copying configuration directory')
# copytree preserves perms but not ownership:
Path(target_config_dir).mkdir(parents=True)
chown(target_config_dir, group='vyattacfg')
chmod_2775(target_config_dir)
copytree('/opt/vyatta/etc/config/', target_config_dir,
copy_function=copy_preserve_owner, dirs_exist_ok=True)
else:
Path(target_config_dir).mkdir(parents=True)
chown(target_config_dir, group='vyattacfg')
chmod_2775(target_config_dir)
Path(f'{target_config_dir}/.vyatta_config').touch()
target_ssh_dir: str = f'{root_dir}/boot/{image_name}/rw/etc/ssh/'
if no_prompt or copy_ssh_host_keys():
print('Copying SSH host keys')
Path(target_ssh_dir).mkdir(parents=True)
host_keys: list[str] = glob('/etc/ssh/ssh_host*')
for host_key in host_keys:
copy(host_key, target_ssh_dir)
# copy system image and kernel files
print('Copying system image files')
for file in Path(f'{DIR_ISO_MOUNT}/live').iterdir():
if file.is_file() and (file.match('initrd*') or
file.match('vmlinuz*')):
copy(file, f'{root_dir}/boot/{image_name}/')
copy(f'{DIR_ISO_MOUNT}/live/filesystem.squashfs',
f'{root_dir}/boot/{image_name}/{image_name}.squashfs')
# unmount an ISO and cleanup
cleanup([str(iso_path)])
# add information about version
grub.version_add(image_name, root_dir)
if set_as_default:
grub.set_default(image_name, root_dir)
except OSError as e:
# if no space error, remove image dir and cleanup
if e.errno == ENOSPC:
cleanup(mounts=[str(iso_path)],
remove_items=[f'{root_dir}/boot/{image_name}'])
else:
# unmount an ISO and cleanup
cleanup([str(iso_path)])
exit(f'Error: {e}')
except Exception as err:
# unmount an ISO and cleanup
cleanup([str(iso_path)])
exit(f'Error: {err}')
def parse_arguments() -> Namespace:
"""Parse arguments
Returns:
Namespace: a namespace with parsed arguments
"""
parser: ArgumentParser = ArgumentParser(
description='Install new system images')
parser.add_argument('--action',
choices=['install', 'add'],
required=True,
help='action to perform with an image')
parser.add_argument('--vrf',
help='vrf name for image download')
parser.add_argument('--no-prompt', action='store_true',
help='perform action non-interactively')
parser.add_argument('--username', default='',
help='username for image download')
parser.add_argument('--password', default='',
help='password for image download')
parser.add_argument('--image-path',
help='a path (HTTP or local file) to an image that needs to be installed'
)
parser.add_argument('--force', action='store_true',
help='Ignore flavor compatibility requirements.'
)
# parser.add_argument('--image_new_name', help='a new name for image')
args: Namespace = parser.parse_args()
# Validate arguments
if args.action == 'add' and not args.image_path:
exit('A path to image is required for add action')
return args
if __name__ == '__main__':
try:
args: Namespace = parse_arguments()
if args.action == 'install':
install_image()
if args.action == 'add':
add_image(args.image_path, args.vrf,
args.username, args.password,
args.no_prompt, args.force)
exit()
except KeyboardInterrupt:
print('Stopped by Ctrl+C')
cleanup()
exit()
except Exception as err:
exit(f'{err}')