diff --git a/data/templates/https/nginx.default.j2 b/data/templates/https/nginx.default.j2
index 80239ea56..a530c14ba 100644
--- a/data/templates/https/nginx.default.j2
+++ b/data/templates/https/nginx.default.j2
@@ -1,60 +1,60 @@
### Autogenerated by service_https.py ###
# Default server configuration
{% for server in server_block_list %}
server {
# SSL configuration
#
{% if server.address == '*' %}
listen {{ server.port }} ssl;
listen [::]:{{ server.port }} ssl;
{% else %}
listen {{ server.address | bracketize_ipv6 }}:{{ server.port }} ssl;
{% endif %}
{% for name in server.name %}
server_name {{ name }};
{% endfor %}
root /srv/localui;
-{% if server.certbot %}
- ssl_certificate {{ server.certbot_dir }}/live/{{ server.certbot_domain_dir }}/fullchain.pem;
- ssl_certificate_key {{ server.certbot_dir }}/live/{{ server.certbot_domain_dir }}/privkey.pem;
- include {{ server.certbot_dir }}/options-ssl-nginx.conf;
- ssl_dhparam {{ server.certbot_dir }}/ssl-dhparams.pem;
-{% elif server.vyos_cert %}
+{% if server.vyos_cert %}
ssl_certificate {{ server.vyos_cert.crt }};
ssl_certificate_key {{ server.vyos_cert.key }};
{% else %}
#
# Self signed certs generated by the ssl-cert package
# Don't use them in a production server!
#
include snippets/snakeoil.conf;
{% endif %}
+ ssl_session_cache shared:le_nginx_SSL:10m;
+ ssl_session_timeout 1440m;
+ ssl_session_tickets off;
+
ssl_protocols TLSv1.2 TLSv1.3;
+ ssl_ciphers 'ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384:DHE-RSA-AES128-GCM-SHA256:DHE-DSS-AES128-GCM-SHA256:kEDH+AESGCM:ECDHE-RSA-AES128-SHA256:ECDHE-ECDSA-AES128-SHA256:ECDHE-RSA-AES128-SHA:ECDHE-ECDSA-AES128-SHA:ECDHE-RSA-AES256-SHA384:ECDHE-ECDSA-AES256-SHA384:ECDHE-RSA-AES256-SHA:ECDHE-ECDSA-AES256-SHA:DHE-RSA-AES128-SHA256:DHE-RSA-AES128-SHA:DHE-DSS-AES128-SHA256:DHE-RSA-AES256-SHA256:DHE-DSS-AES256-SHA:DHE-RSA-AES256-SHA:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!3DES:!MD5:!PSK';
# proxy settings for HTTP API, if enabled; 503, if not
location ~ ^/(retrieve|configure|config-file|image|container-image|generate|show|reboot|reset|poweroff|docs|openapi.json|redoc|graphql) {
{% if server.api %}
proxy_pass http://unix:/run/api.sock;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 600;
proxy_buffering off;
{% else %}
return 503;
{% endif %}
{% if server.allow_client %}
{% for client in server.allow_client %}
allow {{ client }};
{% endfor %}
deny all;
{% endif %}
}
error_page 497 =301 https://$host:{{ server.port }}$request_uri;
}
{% endfor %}
diff --git a/debian/control b/debian/control
index 3f1ba1a63..871a1f0f7 100644
--- a/debian/control
+++ b/debian/control
@@ -1,212 +1,212 @@
Source: vyos-1x
Section: contrib/net
Priority: extra
Maintainer: VyOS Package Maintainers <maintainers@vyos.net>
Build-Depends:
debhelper (>= 9),
dh-python,
fakeroot,
gcc,
iproute2,
libvyosconfig0 (>= 0.0.7),
libzmq3-dev,
python3 (>= 3.10),
python3-coverage,
python3-lxml,
python3-netifaces,
python3-nose,
python3-jinja2,
python3-psutil,
python3-setuptools,
python3-sphinx,
python3-xmltodict,
quilt,
whois
Standards-Version: 3.9.6
Package: vyos-1x
Architecture: amd64 arm64
Pre-Depends:
libnss-tacplus [amd64],
libpam-tacplus [amd64],
libpam-radius-auth [amd64]
Depends:
${python3:Depends} (>= 3.10),
aardvark-dns,
accel-ppp,
auditd,
avahi-daemon,
beep,
bmon,
bsdmainutils,
+ certbot,
charon-systemd,
conntrack,
conntrackd,
conserver-client,
conserver-server,
console-data,
cron,
curl,
dbus,
ddclient (>= 3.11.1),
dmidecode,
dropbear,
easy-rsa,
etherwake,
ethtool,
fdisk,
fastnetmon [amd64],
file,
frr (>= 7.5),
frr-pythontools,
frr-rpki-rtrlib,
frr-snmp,
fuse-overlayfs,
libpam-google-authenticator,
git,
grc,
haproxy,
hostapd,
hsflowd,
hvinfo,
igmpproxy,
ipaddrcheck,
iperf,
iperf3,
iproute2 (>= 6.0.0),
iptables,
iputils-arping,
isc-dhcp-client,
isc-dhcp-relay,
isc-dhcp-server,
iw,
jool,
keepalived (>=2.0.5),
lcdproc,
lcdproc-extra-drivers,
libatomic1,
libauparse0,
libcharon-extra-plugins (>=5.9),
libcharon-extauth-plugins (>=5.9),
libndp-tools,
libnetfilter-conntrack3,
libnfnetlink0,
libqmi-utils,
libstrongswan-extra-plugins (>=5.9),
libstrongswan-standard-plugins (>=5.9),
libvyosconfig0,
linux-cpupower,
lldpd,
lm-sensors,
lsscsi,
minisign,
modemmanager,
mtr-tiny,
ndisc6,
ndppd,
netavark,
netplug,
nfct,
nftables (>= 0.9.3),
nginx-light,
chrony,
nvme-cli,
ocserv,
opennhrp,
openssh-server,
openssl,
openvpn,
openvpn-auth-ldap,
openvpn-auth-radius,
openvpn-otp,
owamp-client,
owamp-server,
pciutils,
pdns-recursor,
pmacct (>= 1.6.0),
podman,
pppoe,
procps,
python3,
- python3-certbot-nginx,
python3-cryptography,
python3-hurry.filesize,
python3-inotify,
python3-isc-dhcp-leases,
python3-jinja2,
python3-jmespath,
python3-netaddr,
python3-netifaces,
python3-paramiko,
python3-passlib,
python3-psutil,
python3-pyhumps,
python3-pystache,
python3-pyudev,
python3-six,
python3-tabulate,
python3-vici (>= 5.7.2),
python3-voluptuous,
python3-xmltodict,
python3-zmq,
qrencode,
radvd,
salt-minion,
sed,
smartmontools,
snmp,
snmpd,
squashfs-tools,
squid,
squidclient,
squidguard,
sshguard,
ssl-cert,
sstp-client,
strongswan (>= 5.9),
strongswan-swanctl (>= 5.9),
stunnel4,
sudo,
systemd,
telegraf (>= 1.20),
tcpdump,
tcptraceroute,
telnet,
tftpd-hpa,
traceroute,
tuned,
twamp-client,
twamp-server,
udp-broadcast-relay,
uidmap,
usb-modeswitch,
usbutils,
vyatta-bash,
vyatta-cfg,
vyos-http-api-tools,
vyos-utils,
wide-dhcpv6-client,
wireguard-tools,
wireless-regdb,
wpasupplicant (>= 0.6.7),
zabbix-agent2,
ndppd,
miniupnpd-nftables
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/interface-definitions/include/constraint/email.xml.i b/interface-definitions/include/constraint/email.xml.i
new file mode 100644
index 000000000..b19a88d64
--- /dev/null
+++ b/interface-definitions/include/constraint/email.xml.i
@@ -0,0 +1,3 @@
+<!-- include start from constraint/email.xml.i -->
+<regex>[^\s@]+@([^\s@.,]+\.)+[^\s@.,]{2,}</regex>
+<!-- include end -->
diff --git a/interface-definitions/include/version/https-version.xml.i b/interface-definitions/include/version/https-version.xml.i
index fa18278f3..525314dbd 100644
--- a/interface-definitions/include/version/https-version.xml.i
+++ b/interface-definitions/include/version/https-version.xml.i
@@ -1,3 +1,3 @@
<!-- include start from include/version/https-version.xml.i -->
-<syntaxVersion component='https' version='5'></syntaxVersion>
+<syntaxVersion component='https' version='6'></syntaxVersion>
<!-- include end -->
diff --git a/interface-definitions/pki.xml.in b/interface-definitions/pki.xml.in
index 097c541ac..0ed199539 100644
--- a/interface-definitions/pki.xml.in
+++ b/interface-definitions/pki.xml.in
@@ -1,243 +1,297 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="pki" owner="${vyos_conf_scripts_dir}/pki.py">
<properties>
<help>VyOS PKI configuration</help>
<priority>300</priority>
</properties>
<children>
<tagNode name="ca">
<properties>
<help>Certificate Authority</help>
<constraint>
#include <include/constraint/alpha-numeric-hyphen-underscore-dot.xml.i>
</constraint>
</properties>
<children>
<leafNode name="certificate">
<properties>
<help>CA certificate in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>CA certificate is not base64-encoded</constraintErrorMessage>
</properties>
</leafNode>
#include <include/generic-description.xml.i>
<node name="private">
<properties>
<help>CA private key in PEM format</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>CA private key in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>CA private key is not base64-encoded</constraintErrorMessage>
</properties>
</leafNode>
<leafNode name="password-protected">
<properties>
<help>CA private key is password protected</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<leafNode name="crl">
<properties>
<help>Certificate revocation list in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>CRL is not base64-encoded</constraintErrorMessage>
<multi/>
</properties>
</leafNode>
<leafNode name="revoke">
<properties>
<help>If parent CA is present, this CA certificate will be included in generated CRLs</help>
<valueless/>
</properties>
</leafNode>
</children>
</tagNode>
<tagNode name="certificate">
<properties>
<help>Certificate</help>
<constraint>
#include <include/constraint/alpha-numeric-hyphen-underscore-dot.xml.i>
</constraint>
</properties>
<children>
<leafNode name="certificate">
<properties>
<help>Certificate in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>Certificate is not base64-encoded</constraintErrorMessage>
</properties>
</leafNode>
+ <node name="acme">
+ <properties>
+ <help>Automatic Certificate Management Environment (ACME) request</help>
+ </properties>
+ <children>
+ #include <include/url-http-https.xml.i>
+ <leafNode name="url">
+ <defaultValue>https://acme-v02.api.letsencrypt.org/directory</defaultValue>
+ </leafNode>
+ <leafNode name="domain-name">
+ <properties>
+ <help>Domain Name</help>
+ <constraint>
+ <validator name="fqdn"/>
+ </constraint>
+ <constraintErrorMessage>Invalid domain name (RFC 1123 section 2).\nMay only contain letters, numbers and .-_</constraintErrorMessage>
+ <multi/>
+ </properties>
+ </leafNode>
+ <leafNode name="email">
+ <properties>
+ <help>Email address to associate with certificate</help>
+ <constraint>
+ #include <include/constraint/email.xml.i>
+ </constraint>
+ </properties>
+ </leafNode>
+ #include <include/listen-address-ipv4-single.xml.i>
+ <leafNode name="rsa-key-size">
+ <properties>
+ <help>Size of the RSA key</help>
+ <completionHelp>
+ <list>2048 3072 4096</list>
+ </completionHelp>
+ <valueHelp>
+ <format>2048</format>
+ <description>RSA key length 2048 bit</description>
+ </valueHelp>
+ <valueHelp>
+ <format>3072</format>
+ <description>RSA key length 3072 bit</description>
+ </valueHelp>
+ <valueHelp>
+ <format>4096</format>
+ <description>RSA key length 4096 bit</description>
+ </valueHelp>
+ <constraint>
+ <regex>(2048|3072|4096)</regex>
+ </constraint>
+ </properties>
+ <defaultValue>2048</defaultValue>
+ </leafNode>
+ </children>
+ </node>
#include <include/generic-description.xml.i>
<node name="private">
<properties>
<help>Certificate private key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>Certificate private key in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>Certificate private key is not base64-encoded</constraintErrorMessage>
</properties>
</leafNode>
<leafNode name="password-protected">
<properties>
<help>Certificate private key is password protected</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<leafNode name="revoke">
<properties>
<help>If CA is present, this certificate will be included in generated CRLs</help>
<valueless/>
</properties>
</leafNode>
</children>
</tagNode>
<tagNode name="dh">
<properties>
<help>Diffie-Hellman parameters</help>
<constraint>
#include <include/constraint/alpha-numeric-hyphen-underscore-dot.xml.i>
</constraint>
</properties>
<children>
<leafNode name="parameters">
<properties>
<help>DH parameters in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>DH parameters are not base64-encoded</constraintErrorMessage>
</properties>
</leafNode>
</children>
</tagNode>
<tagNode name="key-pair">
<properties>
<help>Public and private keys</help>
</properties>
<children>
<node name="public">
<properties>
<help>Public key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>Public key in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>Public key is not base64-encoded</constraintErrorMessage>
</properties>
</leafNode>
</children>
</node>
<node name="private">
<properties>
<help>Private key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>Private key in PEM format</help>
<constraint>
<validator name="base64"/>
</constraint>
<constraintErrorMessage>Private key is not base64-encoded</constraintErrorMessage>
</properties>
</leafNode>
<leafNode name="password-protected">
<properties>
<help>Private key is password protected</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
</children>
</tagNode>
<node name="openvpn">
<properties>
<help>OpenVPN keys</help>
</properties>
<children>
<tagNode name="shared-secret">
<properties>
<help>OpenVPN shared secret key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>OpenVPN shared secret key data</help>
</properties>
</leafNode>
<leafNode name="version">
<properties>
<help>OpenVPN shared secret key version</help>
</properties>
</leafNode>
</children>
</tagNode>
</children>
</node>
<node name="x509">
<properties>
<help>X509 Settings</help>
</properties>
<children>
<node name="default">
<properties>
<help>X509 Default Values</help>
</properties>
<children>
<leafNode name="country">
<properties>
<help>Default country</help>
</properties>
<defaultValue>GB</defaultValue>
</leafNode>
<leafNode name="state">
<properties>
<help>Default state</help>
</properties>
<defaultValue>Some-State</defaultValue>
</leafNode>
<leafNode name="locality">
<properties>
<help>Default locality</help>
</properties>
<defaultValue>Some-City</defaultValue>
</leafNode>
<leafNode name="organization">
<properties>
<help>Default organization</help>
</properties>
<defaultValue>VyOS</defaultValue>
</leafNode>
</children>
</node>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/interface-definitions/service_https.xml.in b/interface-definitions/service_https.xml.in
index 223f10962..57f36a982 100644
--- a/interface-definitions/service_https.xml.in
+++ b/interface-definitions/service_https.xml.in
@@ -1,220 +1,202 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="service">
<children>
<node name="https" owner="${vyos_conf_scripts_dir}/service_https.py">
<properties>
<help>HTTPS configuration</help>
<priority>1001</priority>
</properties>
<children>
<tagNode name="virtual-host">
<properties>
<help>Identifier for virtual host</help>
<constraint>
<regex>[a-zA-Z0-9-_.:]{1,255}</regex>
</constraint>
<constraintErrorMessage>illegal characters in identifier or identifier longer than 255 characters</constraintErrorMessage>
</properties>
<children>
<leafNode name="listen-address">
<properties>
<help>Address to listen for HTTPS requests</help>
<completionHelp>
<script>${vyos_completion_dir}/list_local_ips.sh --both</script>
</completionHelp>
<valueHelp>
<format>ipv4</format>
<description>HTTPS IPv4 address</description>
</valueHelp>
<valueHelp>
<format>ipv6</format>
<description>HTTPS IPv6 address</description>
</valueHelp>
<valueHelp>
<format>'*'</format>
<description>any</description>
</valueHelp>
<constraint>
<validator name="ip-address"/>
<regex>\*</regex>
</constraint>
</properties>
</leafNode>
#include <include/port-number.xml.i>
<leafNode name='port'>
<defaultValue>443</defaultValue>
</leafNode>
<leafNode name="server-name">
<properties>
<help>Server names: exact, wildcard, or regex</help>
<multi/>
</properties>
</leafNode>
#include <include/allow-client.xml.i>
</children>
</tagNode>
<node name="api">
<properties>
<help>VyOS HTTP API configuration</help>
</properties>
<children>
<node name="keys">
<properties>
<help>HTTP API keys</help>
</properties>
<children>
<tagNode name="id">
<properties>
<help>HTTP API id</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>HTTP API plaintext key</help>
</properties>
</leafNode>
</children>
</tagNode>
</children>
</node>
<leafNode name="strict">
<properties>
<help>Enforce strict path checking</help>
<valueless/>
</properties>
</leafNode>
<leafNode name="debug">
<properties>
<help>Debug</help>
<valueless/>
<hidden/>
</properties>
</leafNode>
<node name="graphql">
<properties>
<help>GraphQL support</help>
</properties>
<children>
<leafNode name="introspection">
<properties>
<help>Schema introspection</help>
<valueless/>
</properties>
</leafNode>
<node name="authentication">
<properties>
<help>GraphQL authentication</help>
</properties>
<children>
<leafNode name="type">
<properties>
<help>Authentication type</help>
<completionHelp>
<list>key token</list>
</completionHelp>
<valueHelp>
<format>key</format>
<description>Use API keys</description>
</valueHelp>
<valueHelp>
<format>token</format>
<description>Use JWT token</description>
</valueHelp>
<constraint>
<regex>(key|token)</regex>
</constraint>
</properties>
<defaultValue>key</defaultValue>
</leafNode>
<leafNode name="expiration">
<properties>
<help>Token time to expire in seconds</help>
<valueHelp>
<format>u32:60-31536000</format>
<description>Token lifetime in seconds</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 60-31536000"/>
</constraint>
</properties>
<defaultValue>3600</defaultValue>
</leafNode>
<leafNode name="secret-length">
<properties>
<help>Length of shared secret in bytes</help>
<valueHelp>
<format>u32:16-65535</format>
<description>Byte length of generated shared secret</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 16-65535"/>
</constraint>
</properties>
<defaultValue>32</defaultValue>
</leafNode>
</children>
</node>
</children>
</node>
<node name="cors">
<properties>
<help>Set CORS options</help>
</properties>
<children>
<leafNode name="allow-origin">
<properties>
<help>Allow resource request from origin</help>
<multi/>
</properties>
</leafNode>
</children>
</node>
</children>
</node>
<node name="api-restrict">
<properties>
<help>Restrict api proxy to subset of virtual hosts</help>
</properties>
<children>
<leafNode name="virtual-host">
<properties>
<help>Restrict proxy to virtual host(s)</help>
<multi/>
</properties>
</leafNode>
</children>
</node>
<node name="certificates">
<properties>
<help>TLS certificates</help>
</properties>
<children>
#include <include/pki/ca-certificate.xml.i>
#include <include/pki/certificate.xml.i>
- <node name="certbot" owner="${vyos_conf_scripts_dir}/service_https_certificates_certbot.py">
- <properties>
- <help>Request or apply a letsencrypt certificate for domain-name</help>
- </properties>
- <children>
- <leafNode name="domain-name">
- <properties>
- <help>Domain name(s) for which to obtain certificate</help>
- <multi/>
- </properties>
- </leafNode>
- <leafNode name="email">
- <properties>
- <help>Email address to associate with certificate</help>
- </properties>
- </leafNode>
- </children>
- </node>
</children>
</node>
#include <include/interface/vrf.xml.i>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/op-mode-definitions/monitor-log.xml.in b/op-mode-definitions/monitor-log.xml.in
index df17371cc..f01c715cb 100644
--- a/op-mode-definitions/monitor-log.xml.in
+++ b/op-mode-definitions/monitor-log.xml.in
@@ -1,360 +1,366 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="monitor">
<properties>
<help>Monitor system information</help>
</properties>
<children>
<node name="log">
<properties>
<help>Monitor last lines of messages file</help>
</properties>
<command>SYSTEMD_LOG_COLOR=false journalctl --no-hostname --follow --boot</command>
<children>
<node name="color">
<properties>
<help>Output log in a colored fashion</help>
</properties>
<command>SYSTEMD_LOG_COLOR=false grc journalctl --no-hostname --follow --boot</command>
</node>
<node name="ids">
<properties>
<help>Monitor Intrusion Detection System log</help>
</properties>
<children>
<leafNode name="ddos-protection">
<properties>
<help>Monitor last lines of DDOS protection</help>
</properties>
<command>journalctl --no-hostname --follow --boot --unit fastnetmon.service</command>
</leafNode>
</children>
</node>
+ <leafNode name="certbot">
+ <properties>
+ <help>Monitor last lines of certbot log</help>
+ </properties>
+ <command>if sudo test -f /var/log/letsencrypt/letsencrypt.log; then sudo tail --follow=name /var/log/letsencrypt/letsencrypt.log; else echo "Cerbot log does not exist"; fi</command>
+ </leafNode>
<leafNode name="conntrack-sync">
<properties>
<help>Monitor last lines of conntrack-sync log</help>
</properties>
<command>journalctl --no-hostname --follow --boot --unit conntrackd.service</command>
</leafNode>
<leafNode name="console-server">
<properties>
<help>Monitor last lines of console server log</help>
</properties>
<command>journalctl --no-hostname --follow --boot --unit conserver-server.service</command>
</leafNode>
<node name="dhcp">
<properties>
<help>Monitor last lines of Dynamic Host Control Protocol log</help>
</properties>
<children>
<node name="server">
<properties>
<help>Monitor last lines of DHCP server log</help>
</properties>
<command>journalctl --no-hostname --follow --boot --unit isc-dhcp-server.service</command>
</node>
<node name="client">
<properties>
<help>Monitor last lines of DHCP client log</help>
</properties>
<command>journalctl --no-hostname --follow --boot --unit "dhclient@*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Show DHCP client log on specific interface</help>
<completionHelp>
<script>${vyos_completion_dir}/list_interfaces --broadcast</script>
</completionHelp>
</properties>
<command>journalctl --no-hostname --follow --boot --unit "dhclient@$6.service"</command>
</tagNode>
</children>
</node>
</children>
</node>
<node name="dhcpv6">
<properties>
<help>Monitor last lines of Dynamic Host Control Protocol IPv6 log</help>
</properties>
<children>
<node name="server">
<properties>
<help>Monitor last lines of DHCPv6 server log</help>
</properties>
<command>journalctl --no-hostname --follow --boot --unit isc-dhcp-server6.service</command>
</node>
<node name="client">
<properties>
<help>Monitor last lines of DHCPv6 client log</help>
</properties>
<command>journalctl --no-hostname --follow --boot --unit "dhcp6c@*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Show DHCPv6 client log on specific interface</help>
<completionHelp>
<script>${vyos_completion_dir}/list_interfaces</script>
</completionHelp>
</properties>
<command>journalctl --no-hostname --follow --boot --unit "dhcp6c@$6.service"</command>
</tagNode>
</children>
</node>
</children>
</node>
<leafNode name="flow-accounting">
<properties>
<help>Monitor last lines of flow-accounting log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit uacctd.service</command>
</leafNode>
<leafNode name="ipoe-server">
<properties>
<help>Monitor last lines of IP over Ethernet server log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit accel-ppp@ipoe.service</command>
</leafNode>
<leafNode name="kernel">
<properties>
<help>Monitor last lines of Linux Kernel log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --dmesg</command>
</leafNode>
<leafNode name="ndp-proxy">
<properties>
<help>Monitor last lines of Neighbor Discovery Protocol (NDP) Proxy</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit ndppd.service</command>
</leafNode>
<leafNode name="nhrp">
<properties>
<help>Monitor last lines of Next Hop Resolution Protocol log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit opennhrp.service</command>
</leafNode>
<leafNode name="ntp">
<properties>
<help>Monitor last lines of Network Time Protocol log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit chrony.service</command>
</leafNode>
<node name="openvpn">
<properties>
<help>Monitor last lines of OpenVPN log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit openvpn@*.service</command>
<children>
<tagNode name="interface">
<properties>
<help>Monitor last lines of specific OpenVPN interface log</help>
<completionHelp>
<path>interfaces openvpn</path>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --unit openvpn@$5.service</command>
</tagNode>
</children>
</node>
<node name="pppoe">
<properties>
<help>Monitor last lines of PPPoE interface log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit "ppp@pppoe*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Monitor last lines of PPPoE log for specific interface</help>
<completionHelp>
<path>interfaces pppoe</path>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --follow --unit "ppp@$5.service"</command>
</tagNode>
</children>
</node>
<leafNode name="pppoe-server">
<properties>
<help>Monitor last lines of PPPoE server log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit accel-ppp@pppoe.service</command>
</leafNode>
<node name="protocol">
<properties>
<help>Monitor routing protocol logs</help>
</properties>
<children>
<leafNode name="ospf">
<properties>
<help>Monitor log for OSPF</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/ospfd</command>
</leafNode>
<leafNode name="ospfv3">
<properties>
<help>Monitor log for OSPF for IPv6</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/ospf6d</command>
</leafNode>
<leafNode name="bgp">
<properties>
<help>Monitor log for BGP</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/bgpd</command>
</leafNode>
<leafNode name="rip">
<properties>
<help>Monitor log for RIP</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/ripd</command>
</leafNode>
<leafNode name="ripng">
<properties>
<help>Monitor log for RIPng</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/ripngd</command>
</leafNode>
<leafNode name="static">
<properties>
<help>Monitor log for static route</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/staticd</command>
</leafNode>
<leafNode name="multicast">
<properties>
<help>Monitor log for Multicast protocol</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/pimd</command>
</leafNode>
<leafNode name="isis">
<properties>
<help>Monitor log for ISIS</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/isisd</command>
</leafNode>
<leafNode name="nhrp">
<properties>
<help>Monitor log for NHRP</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/nhrpd</command>
</leafNode>
<leafNode name="bfd">
<properties>
<help>Monitor log for BFD</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/bfdd</command>
</leafNode>
<leafNode name="mpls">
<properties>
<help>Monitor log for MPLS</help>
</properties>
<command>journalctl --follow --no-hostname --boot /usr/lib/frr/ldpd</command>
</leafNode>
</children>
</node>
<node name="macsec">
<properties>
<help>Monitor last lines of MACsec</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit "wpa_supplicant-macsec@*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Monitor last lines of specific MACsec interface</help>
<completionHelp>
<path>interfaces macsec</path>
</completionHelp>
</properties>
<command>SRC=$(cli-shell-api returnValue interfaces macsec "$5" source-interface); journalctl --no-hostname --boot --follow --unit "wpa_supplicant-macsec@$SRC.service"</command>
</tagNode>
</children>
</node>
<leafNode name="router-advert">
<properties>
<help>Monitor last lines of Router Advertisement Daemon log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit radvd.service</command>
</leafNode>
<leafNode name="snmp">
<properties>
<help>Monitor last lines of Simple Network Monitoring Protocol log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit snmpd.service</command>
</leafNode>
<node name="ssh">
<properties>
<help>Monitor last lines of Secure Shell log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit ssh.service</command>
<children>
<node name="dynamic-protection">
<properties>
<help>Monitor last lines of SSH guard log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit sshguard.service</command>
</node>
</children>
</node>
<leafNode name="vpn">
<properties>
<help>Monitor last lines of ALL Virtual Private Network services</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit strongswan.service --unit accel-ppp@*.service --unit ocserv.service</command>
</leafNode>
<leafNode name="ipsec">
<properties>
<help>Monitor last lines of IPsec log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit strongswan.service</command>
</leafNode>
<leafNode name="l2tp">
<properties>
<help>Monitor last lines of L2TP log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit accel-ppp@l2tp.service</command>
</leafNode>
<leafNode name="openconnect">
<properties>
<help>Monitor last lines of OpenConnect log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit ocserv.service</command>
</leafNode>
<leafNode name="pptp">
<properties>
<help>Monitor last lines of PPTP log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit accel-ppp@pptp.service</command>
</leafNode>
<leafNode name="sstp">
<properties>
<help>Monitor last lines of Secure Socket Tunneling Protocol server</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit accel-ppp@sstp.service</command>
</leafNode>
<node name="sstpc">
<properties>
<help>Monitor last lines of Secure Socket Tunneling Protocol client</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit "ppp@sstpc*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Monitor last lines of SSTP client log for specific interface</help>
<completionHelp>
<path>interfaces sstpc</path>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --follow --unit "ppp@$5.service"</command>
</tagNode>
</children>
</node>
<leafNode name="vrrp">
<properties>
<help>Monitor last lines of Virtual Router Redundancy Protocol log</help>
</properties>
<command>journalctl --no-hostname --boot --follow --unit keepalived.service</command>
</leafNode>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/op-mode-definitions/pki.xml.in b/op-mode-definitions/pki.xml.in
index ca0eb3687..4b8d9c47a 100644
--- a/op-mode-definitions/pki.xml.in
+++ b/op-mode-definitions/pki.xml.in
@@ -1,577 +1,587 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="generate">
<children>
<node name="pki">
<properties>
<help>Generate PKI certificates and keys</help>
</properties>
<children>
<node name="ca">
<properties>
<help>Generate CA certificate</help>
</properties>
<children>
<tagNode name="sign">
<properties>
<help>Sign generated CA certificate with another specified CA certificate</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated CA certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$7" --sign "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated CA certificate into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$7" --sign "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "noname" --sign "$5"</command>
</tagNode>
<tagNode name="file">
<properties>
<help>Write generated CA certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated CA certificate into running configuration</help>
<completionHelp>
<list><CA name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "noname"</command>
</node>
<node name="certificate">
<properties>
<help>Generate certificate request</help>
</properties>
<children>
<node name="self-signed">
<properties>
<help>Generate self-signed certificate</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated self-signed certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$6" --self-sign --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated self-signed certificate into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$6" --self-sign --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname" --self-sign</command>
</node>
<tagNode name="sign">
<properties>
<help>Sign generated certificate with specified CA certificate</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated signed certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$7" --sign "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated signed certificate into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$7" --sign "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname" --sign "$5"</command>
</tagNode>
<tagNode name="file">
<properties>
<help>Write generated certificate request and key into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated certificate private key into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname"</command>
</node>
<tagNode name="crl">
<properties>
<help>Generate CRL for specified CA certificate</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated CRL into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --crl "$4" --file</command>
</tagNode>
<leafNode name="install">
<properties>
<help>Commands for installing generated CRL into running configuration</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --crl "$4" --install</command>
</leafNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --crl "$4"</command>
</tagNode>
<node name="dh">
<properties>
<help>Generate DH parameters</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated DH parameters into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --dh "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated DH parameters into running configuration</help>
<completionHelp>
<list><DH name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --dh "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --dh "noname"</command>
</node>
<node name="key-pair">
<properties>
<help>Generate a key pair</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated key pair into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --keypair "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated key pair into running configuration</help>
<completionHelp>
<list><key name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --keypair "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --keypair "noname"</command>
</node>
<node name="openvpn">
<properties>
<help>Generate OpenVPN keys</help>
</properties>
<children>
<node name="shared-secret">
<properties>
<help>Generate OpenVPN shared secret key</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated OpenVPN shared secret key into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --openvpn "$6" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated OpenVPN shared secret key into running configuration</help>
<completionHelp>
<list><key name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --openvpn "$6" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --openvpn "noname"</command>
</node>
</children>
</node>
<node name="ssh-key">
<properties>
<help>Generate SSH key</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated SSH keys into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ssh "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated SSH key into running configuration</help>
<completionHelp>
<list><key name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ssh "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ssh "noname"</command>
</node>
<node name="wireguard">
<properties>
<help>Generate WireGuard keys</help>
</properties>
<children>
<node name="key-pair">
<properties>
<help>Generate WireGuard public/private key-pair</help>
</properties>
<children>
<node name="install">
<properties>
<help>Generate CLI commands to install WireGuard key to configuration</help>
</properties>
<children>
<tagNode name="interface">
<properties>
<help>WireGuard Interface used in install command</help>
<completionHelp>
<path>interfaces wireguard</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --key --interface "$7" --install</command>
</tagNode>
</children>
</node>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --key</command>
</node>
<node name="preshared-key">
<properties>
<help>Generate WireGuard pre-shared key</help>
</properties>
<children>
<node name="install">
<properties>
<help>Generate CLI commands to install WireGuard key to configuration</help>
</properties>
<children>
<tagNode name="interface">
<properties>
<help>WireGuard Interface used in install command</help>
<completionHelp>
<path>interfaces wireguard</path>
</completionHelp>
</properties>
<children>
<tagNode name="peer">
<properties>
<help>Interface used for install command</help>
<completionHelp>
<path>interfaces wireguard ${COMP_WORDS[COMP_CWORD-2]} peer</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --psk --interface "$7" --peer "$9" --install</command>
</tagNode>
</children>
</tagNode>
</children>
</node>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --psk</command>
</node>
</children>
</node>
</children>
</node>
</children>
</node>
<node name="import">
<properties>
<help>Import an object</help>
</properties>
<children>
<node name="pki">
<properties>
<help>Import file into PKI configuration</help>
</properties>
<children>
<tagNode name="ca">
<properties>
<help>Import CA certificate into PKI</help>
<completionHelp>
<list><name></list>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Path to CA certificate file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --ca "$4" --filename "$6"</command>
</tagNode>
<tagNode name="key-file">
<properties>
<help>Path to private key file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --ca "$4" --key-filename "$6"</command>
</tagNode>
</children>
</tagNode>
<tagNode name="certificate">
<properties>
<help>Import certificate into PKI</help>
<completionHelp>
<list><name></list>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Path to certificate file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --certificate "$4" --filename "$6"</command>
</tagNode>
<tagNode name="key-file">
<properties>
<help>Path to private key file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --certificate "$4" --key-filename "$6"</command>
</tagNode>
</children>
</tagNode>
<tagNode name="crl">
<properties>
<help>Import certificate revocation list into PKI</help>
<completionHelp>
<list><CA name></list>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Path to CRL file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --crl "$4" --filename "$6"</command>
</tagNode>
</children>
</tagNode>
<tagNode name="dh">
<properties>
<help>Import DH parameters into PKI</help>
<completionHelp>
<list><name></list>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Path to DH parameters file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --dh "$4" --filename "$6"</command>
</tagNode>
</children>
</tagNode>
<tagNode name="key-pair">
<properties>
<help>Import key pair into PKI</help>
<completionHelp>
<list><name></list>
</completionHelp>
</properties>
<children>
<tagNode name="public-file">
<properties>
<help>Path to public key file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --keypair "$4" --filename "$6"</command>
</tagNode>
<tagNode name="private-file">
<properties>
<help>Path to private key file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --keypair "$4" --key-filename "$6"</command>
</tagNode>
</children>
</tagNode>
<node name="openvpn">
<properties>
<help>Import OpenVPN keys into PKI</help>
</properties>
<children>
<tagNode name="shared-secret">
<properties>
<help>Import OpenVPN shared secret key into PKI</help>
<completionHelp>
<list><name></list>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Path to shared secret key file</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action import --openvpn "$5" --filename "$7"</command>
</tagNode>
</children>
</tagNode>
</children>
</node>
</children>
</node>
</children>
</node>
<node name="show">
<children>
<node name="pki">
<properties>
<help>Show PKI x509 certificates</help>
</properties>
<children>
<leafNode name="ca">
<properties>
<help>Show x509 CA certificates</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --ca "all"</command>
</leafNode>
<tagNode name="ca">
<properties>
<help>Show x509 CA certificate by name</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --ca "$4"</command>
<children>
<leafNode name="pem">
<properties>
<help>Show x509 CA certificate in PEM format</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --ca "$4" --pem</command>
</leafNode>
</children>
</tagNode>
<leafNode name="certificate">
<properties>
<help>Show x509 certificates</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --certificate "all"</command>
</leafNode>
<tagNode name="certificate">
<properties>
<help>Show x509 certificate by name</help>
<completionHelp>
<path>pki certificate</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --certificate "$4"</command>
<children>
<leafNode name="pem">
<properties>
<help>Show x509 certificate in PEM format</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --certificate "$4" --pem</command>
</leafNode>
<tagNode name="fingerprint">
<properties>
<help>Show x509 certificate fingerprint</help>
<completionHelp>
<list>sha256 sha384 sha512</list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --certificate "$4" --fingerprint "$6"</command>
</tagNode>
</children>
</tagNode>
<leafNode name="crl">
<properties>
<help>Show x509 certificate revocation lists</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --crl "all"</command>
</leafNode>
<tagNode name="crl">
<properties>
<help>Show x509 certificate revocation lists by CA name</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --crl "$4"</command>
<children>
<leafNode name="pem">
<properties>
<help>Show x509 certificate revocation lists by CA name in PEM format</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --crl "$4" --pem</command>
</leafNode>
</children>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show</command>
</node>
</children>
</node>
+ <node name="renew">
+ <children>
+ <leafNode name="certbot">
+ <properties>
+ <help>Start manual certbot renewal</help>
+ </properties>
+ <command>sudo systemctl start certbot.service</command>
+ </leafNode>
+ </children>
+ </node>
</interfaceDefinition>
diff --git a/op-mode-definitions/show-log.xml.in b/op-mode-definitions/show-log.xml.in
index 6cd53882d..432a21b59 100644
--- a/op-mode-definitions/show-log.xml.in
+++ b/op-mode-definitions/show-log.xml.in
@@ -1,505 +1,511 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="show">
<properties>
<help>Show system information</help>
</properties>
<children>
<tagNode name="log">
<properties>
<help>Show last number of messages in master logging buffer</help>
<completionHelp>
<list><1-9999></list>
</completionHelp>
</properties>
<command>if ${vyos_validators_dir}/numeric --range 1-9999 "$3"; then journalctl --no-hostname --boot --lines "$3"; fi</command>
</tagNode>
<node name="log">
<properties>
<help>Show contents of current master logging buffer</help>
</properties>
<command>journalctl --no-hostname --boot</command>
<children>
<leafNode name="audit">
<properties>
<help>Show audit logs</help>
</properties>
<command>cat /var/log/audit/audit.log</command>
</leafNode>
<leafNode name="all">
<properties>
<help>Show contents of all master log files</help>
</properties>
<command>sudo bash -c 'eval $(lesspipe); less $_vyatta_less_options --prompt=".logm, file %i of %m., page %dt of %D" -- `printf "%s\n" /var/log/messages* | sort -nr`'</command>
</leafNode>
<leafNode name="authorization">
<properties>
<help>Show listing of authorization attempts</help>
</properties>
<command>journalctl --no-hostname --boot --quiet SYSLOG_FACILITY=10 SYSLOG_FACILITY=4</command>
</leafNode>
+ <leafNode name="certbot">
+ <properties>
+ <help>Show log for certbot</help>
+ </properties>
+ <command>if sudo test -f /var/log/letsencrypt/letsencrypt.log; then sudo cat /var/log/letsencrypt/letsencrypt.log; else echo "Cerbot log does not exist"; fi</command>
+ </leafNode>
<leafNode name="cluster">
<properties>
<help>Show log for Cluster</help>
</properties>
<command>cat $(printf "%s\n" /var/log/messages* | sort -nr) | grep -e heartbeat -e cl_status -e mach_down -e ha_log</command>
</leafNode>
<leafNode name="conntrack-sync">
<properties>
<help>Show log for Conntrack-sync</help>
</properties>
<command>journalctl --no-hostname --boot --unit conntrackd.service</command>
</leafNode>
<leafNode name="console-server">
<properties>
<help>Show log for console server</help>
</properties>
<command>journalctl --no-hostname --boot --unit conserver-server.service</command>
</leafNode>
<node name="ids">
<properties>
<help>Show log for for Intrusion Detection System</help>
</properties>
<children>
<leafNode name="ddos-protection">
<properties>
<help>Show log for DDOS protection</help>
</properties>
<command>journalctl --no-hostname --boot --unit fastnetmon.service</command>
</leafNode>
</children>
</node>
<node name="dhcp">
<properties>
<help>Show log for Dynamic Host Control Protocol (DHCP)</help>
</properties>
<children>
<node name="server">
<properties>
<help>Show log for DHCP server</help>
</properties>
<command>journalctl --no-hostname --boot --unit isc-dhcp-server.service</command>
</node>
<node name="client">
<properties>
<help>Show DHCP client logs</help>
</properties>
<command>journalctl --no-hostname --boot --unit "dhclient@*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Show DHCP client log on specific interface</help>
<completionHelp>
<script>${vyos_completion_dir}/list_interfaces --broadcast</script>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --unit "dhclient@$6.service"</command>
</tagNode>
</children>
</node>
</children>
</node>
<node name="dhcpv6">
<properties>
<help>Show log for Dynamic Host Control Protocol IPv6 (DHCPv6)</help>
</properties>
<children>
<node name="server">
<properties>
<help>Show log for DHCPv6 server</help>
</properties>
<command>journalctl --no-hostname --boot --unit isc-dhcp-server6.service</command>
</node>
<node name="client">
<properties>
<help>Show DHCPv6 client logs</help>
</properties>
<command>journalctl --no-hostname --boot --unit "dhcp6c@*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Show DHCPv6 client log on specific interface</help>
<completionHelp>
<script>${vyos_completion_dir}/list_interfaces</script>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --unit "dhcp6c@$6.service"</command>
</tagNode>
</children>
</node>
</children>
</node>
<node name="firewall">
<properties>
<help>Show log for Firewall</help>
</properties>
<children>
<tagNode name="ipv6-name">
<properties>
<help>Show log for a specified firewall (IPv6)</help>
<completionHelp>
<path>firewall ipv6-name</path>
</completionHelp>
</properties>
<command>cat $(printf "%s\n" /var/log/messages* | sort -nr ) | egrep "\[$5-([0-9]+|default)-[ADR]\]"</command>
<children>
<tagNode name="rule">
<properties>
<help>Show log for a rule in the specified firewall</help>
<completionHelp>
<path>firewall ipv6-name ${COMP_WORDS[4]} rule</path>
</completionHelp>
</properties>
<command>cat $(printf "%s\n" /var/log/messages* | sort -nr) | grep -e "\[$5-$7-[ADR]\]"</command>
</tagNode>
</children>
</tagNode>
<tagNode name="name">
<properties>
<help>Show log for a specified firewall (IPv4)</help>
<completionHelp>
<path>firewall name</path>
</completionHelp>
</properties>
<command>cat $(printf "%s\n" /var/log/messages* | sort -nr ) | egrep "\[$5-([0-9]+|default)-[ADR]\]"</command>
<children>
<tagNode name="rule">
<properties>
<help>Show log for a rule in the specified firewall</help>
<completionHelp>
<path>firewall name ${COMP_WORDS[4]} rule</path>
</completionHelp>
</properties>
<command>cat $(printf "%s\n" /var/log/messages* | sort -nr) | egrep "\[$5-$7-[ADR]\]"</command>
</tagNode>
</children>
</tagNode>
</children>
</node>
<leafNode name="flow-accounting">
<properties>
<help>Show log for flow-accounting</help>
</properties>
<command>journalctl --no-hostname --boot --unit uacctd.service</command>
</leafNode>
<leafNode name="https">
<properties>
<help>Show log for HTTPs</help>
</properties>
<command>journalctl --no-hostname --boot --unit nginx.service</command>
</leafNode>
<tagNode name="image">
<properties>
<help>Show contents of master log file for image</help>
<completionHelp>
<script>compgen -f /lib/live/mount/persistence/boot/ | grep -v grub | sed -e s@/lib/live/mount/persistence/boot/@@</script>
</completionHelp>
</properties>
<command>less $_vyatta_less_options --prompt=".log, page %dt of %D" -- /lib/live/mount/persistence/boot/$4/rw/var/log/messages</command>
<children>
<leafNode name="all">
<properties>
<help>Show contents of all master log files for image</help>
</properties>
<command>eval $(lesspipe); less $_vyatta_less_options --prompt=".log?m, file %i of %m., page %dt of %D" -- `printf "%s\n" /lib/live/mount/persistence/boot/$4/rw/var/log/messages* | sort -nr`</command>
</leafNode>
<leafNode name="authorization">
<properties>
<help>Show listing of authorization attempts for image</help>
</properties>
<command>less $_vyatta_less_options --prompt=".log, page %dt of %D" -- /lib/live/mount/persistence/boot/$4/rw/var/log/auth.log</command>
</leafNode>
<tagNode name="tail">
<properties>
<help>Show last changes to messages</help>
<completionHelp>
<list><NUMBER></list>
</completionHelp>
</properties>
<command>tail -n "$6" /lib/live/mount/persistence/boot/$4/rw/var/log/messages | ${VYATTA_PAGER:-cat}</command>
</tagNode>
</children>
</tagNode>
<leafNode name="ipoe-server">
<properties>
<help>Show log for IPoE server</help>
</properties>
<command>journalctl --no-hostname --boot --unit accel-ppp@ipoe.service</command>
</leafNode>
<leafNode name="kernel">
<properties>
<help>Show log for Linux Kernel</help>
</properties>
<command>journalctl --no-hostname --boot --dmesg</command>
</leafNode>
<leafNode name="lldp">
<properties>
<help>Show log for Link Layer Discovery Protocol (LLDP)</help>
</properties>
<command>journalctl --no-hostname --boot --unit lldpd.service</command>
</leafNode>
<leafNode name="nat">
<properties>
<help>Show log for Network Address Translation (NAT)</help>
</properties>
<command>egrep -i "kernel:.*\[NAT-[A-Z]{3,}-[0-9]+(-MASQ)?\]" $(find /var/log -maxdepth 1 -type f -name messages\* | sort -t. -k2nr)</command>
</leafNode>
<leafNode name="ndp-proxy">
<properties>
<help>Show log for Neighbor Discovery Protocol (NDP) Proxy</help>
</properties>
<command>journalctl --no-hostname --boot --unit ndppd.service</command>
</leafNode>
<leafNode name="nhrp">
<properties>
<help>Show log for Next Hop Resolution Protocol (NHRP)</help>
</properties>
<command>journalctl --no-hostname --boot --unit opennhrp.service</command>
</leafNode>
<leafNode name="ntp">
<properties>
<help>Show log for Network Time Protocol (NTP)</help>
</properties>
<command>journalctl --no-hostname --boot --unit chrony.service</command>
</leafNode>
<node name="macsec">
<properties>
<help>Show log for MACsec</help>
</properties>
<command>journalctl --no-hostname --boot --unit "wpa_supplicant-macsec@*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Show MACsec log on specific interface</help>
<completionHelp>
<path>interfaces macsec</path>
</completionHelp>
</properties>
<command>SRC=$(cli-shell-api returnValue interfaces macsec "$5" source-interface); journalctl --no-hostname --boot --unit "wpa_supplicant-macsec@$SRC.service"</command>
</tagNode>
</children>
</node>
<node name="openvpn">
<properties>
<help>Show log for OpenVPN</help>
</properties>
<command>journalctl --no-hostname --boot --unit openvpn@*.service</command>
<children>
<tagNode name="interface">
<properties>
<help>Show OpenVPN log on specific interface</help>
<completionHelp>
<path>interfaces openvpn</path>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --unit openvpn@$5.service</command>
</tagNode>
</children>
</node>
<node name="pppoe">
<properties>
<help>Show log for PPPoE interface</help>
</properties>
<command>journalctl --no-hostname --boot --unit "ppp@pppoe*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Show PPPoE log on specific interface</help>
<completionHelp>
<path>interfaces pppoe</path>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --unit "ppp@$5.service"</command>
</tagNode>
</children>
</node>
<leafNode name="pppoe-server">
<properties>
<help>Show log for PPPoE server</help>
</properties>
<command>journalctl --no-hostname --boot --unit accel-ppp@pppoe.service</command>
</leafNode>
<node name="protocol">
<properties>
<help>Show log for Routing Protocol</help>
</properties>
<children>
<leafNode name="ospf">
<properties>
<help>Show log for OSPF</help>
</properties>
<command>journalctl --boot /usr/lib/frr/ospfd</command>
</leafNode>
<leafNode name="ospfv3">
<properties>
<help>Show log for OSPF for IPv6</help>
</properties>
<command>journalctl --boot /usr/lib/frr/ospf6d</command>
</leafNode>
<leafNode name="bgp">
<properties>
<help>Show log for BGP</help>
</properties>
<command>journalctl --boot /usr/lib/frr/bgpd</command>
</leafNode>
<leafNode name="rip">
<properties>
<help>Show log for RIP</help>
</properties>
<command>journalctl --boot /usr/lib/frr/ripd</command>
</leafNode>
<leafNode name="ripng">
<properties>
<help>Show log for RIPng</help>
</properties>
<command>journalctl --boot /usr/lib/frr/ripngd</command>
</leafNode>
<leafNode name="static">
<properties>
<help>Show log for static route</help>
</properties>
<command>journalctl --boot /usr/lib/frr/staticd</command>
</leafNode>
<leafNode name="multicast">
<properties>
<help>Show log for Multicast protocol</help>
</properties>
<command>journalctl --boot /usr/lib/frr/pimd</command>
</leafNode>
<leafNode name="isis">
<properties>
<help>Show log for ISIS</help>
</properties>
<command>journalctl --boot /usr/lib/frr/isisd</command>
</leafNode>
<leafNode name="nhrp">
<properties>
<help>Show log for NHRP</help>
</properties>
<command>journalctl --boot /usr/lib/frr/nhrpd</command>
</leafNode>
<leafNode name="bfd">
<properties>
<help>Show log for BFD</help>
</properties>
<command>journalctl --boot /usr/lib/frr/bfdd</command>
</leafNode>
<leafNode name="mpls">
<properties>
<help>Show log for MPLS</help>
</properties>
<command>journalctl --boot /usr/lib/frr/ldpd</command>
</leafNode>
</children>
</node>
<leafNode name="router-advert">
<properties>
<help>Show log for Router Advertisement Daemon (radvd)</help>
</properties>
<command>journalctl --no-hostname --boot --unit radvd.service</command>
</leafNode>
<leafNode name="snmp">
<properties>
<help>Show log for Simple Network Monitoring Protocol (SNMP)</help>
</properties>
<command>journalctl --no-hostname --boot --unit snmpd.service</command>
</leafNode>
<node name="ssh">
<properties>
<help>Show log for Secure Shell (SSH)</help>
</properties>
<command>journalctl --no-hostname --boot --unit ssh.service</command>
<children>
<node name="dynamic-protection">
<properties>
<help>Show SSH guard log</help>
</properties>
<command>journalctl --no-hostname --boot --unit sshguard.service</command>
</node>
</children>
</node>
<tagNode name="tail">
<properties>
<help>Show last n changes to messages</help>
<completionHelp>
<list><NUMBER></list>
</completionHelp>
</properties>
<command>tail -n "$4" /var/log/messages | ${VYATTA_PAGER:-cat}</command>
</tagNode>
<node name="tail">
<properties>
<help>Show last 10 lines of /var/log/messages file</help>
</properties>
<command>tail -n 10 /var/log/messages</command>
</node>
<leafNode name="vpn">
<properties>
<help>Monitor last lines of ALL Virtual Private Network services</help>
</properties>
<command>journalctl --no-hostname --boot --unit strongswan.service --unit accel-ppp@*.service --unit ocserv.service</command>
</leafNode>
<leafNode name="ipsec">
<properties>
<help>Show log for IPsec</help>
</properties>
<command>journalctl --no-hostname --boot --unit strongswan.service</command>
</leafNode>
<leafNode name="l2tp">
<properties>
<help>Show log for L2TP</help>
</properties>
<command>journalctl --no-hostname --boot --unit accel-ppp@l2tp.service</command>
</leafNode>
<leafNode name="openconnect">
<properties>
<help>Show log for OpenConnect</help>
</properties>
<command>journalctl --no-hostname --boot --unit ocserv.service</command>
</leafNode>
<leafNode name="pptp">
<properties>
<help>Show log for PPTP</help>
</properties>
<command>journalctl --no-hostname --boot --unit accel-ppp@pptp.service</command>
</leafNode>
<leafNode name="sstp">
<properties>
<help>Show log for Secure Socket Tunneling Protocol (SSTP) server</help>
</properties>
<command>journalctl --no-hostname --boot --unit accel-ppp@sstp.service</command>
</leafNode>
<node name="sstpc">
<properties>
<help>Show log for Secure Socket Tunneling Protocol (SSTP) client</help>
</properties>
<command>journalctl --no-hostname --boot --unit "ppp@sstpc*.service"</command>
<children>
<tagNode name="interface">
<properties>
<help>Show SSTP client log on specific interface</help>
<completionHelp>
<path>interfaces sstpc</path>
</completionHelp>
</properties>
<command>journalctl --no-hostname --boot --unit "ppp@$5.service"</command>
</tagNode>
</children>
</node>
<leafNode name="vrrp">
<properties>
<help>Show log for Virtual Router Redundancy Protocol (VRRP)</help>
</properties>
<command>journalctl --no-hostname --boot --unit keepalived.service</command>
</leafNode>
<leafNode name="webproxy">
<properties>
<help>Show log for Webproxy</help>
</properties>
<command>journalctl --no-hostname --boot --unit squid.service</command>
</leafNode>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/python/vyos/config.py b/python/vyos/config.py
index ca7b035e5..bee85315d 100644
--- a/python/vyos/config.py
+++ b/python/vyos/config.py
@@ -1,583 +1,620 @@
# Copyright 2017, 2019-2023 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
"""
A library for reading VyOS running config data.
This library is used internally by all config scripts of VyOS,
but its API should be considered stable and safe to use
in user scripts.
Note that this module will not work outside VyOS.
Node taxonomy
#############
There are multiple types of config tree nodes in VyOS, each requires
its own set of operations.
*Leaf nodes* (such as "address" in interfaces) can have values, but cannot
have children.
Leaf nodes can have one value, multiple values, or no values at all.
For example, "system host-name" is a single-value leaf node,
"system name-server" is a multi-value leaf node (commonly abbreviated "multi node"),
and "system ip disable-forwarding" is a valueless leaf node.
Non-leaf nodes cannot have values, but they can have child nodes. They are divided into
two classes depending on whether the names of their children are fixed or not.
For example, under "system", the names of all valid child nodes are predefined
("login", "name-server" etc.).
To the contrary, children of the "system task-scheduler task" node can have arbitrary names.
Such nodes are called *tag nodes*. This terminology is confusing but we keep using it for lack
of a better word. No one remembers if the "tag" in "task Foo" is "task" or "Foo",
but the distinction is irrelevant in practice.
Configuration modes
###################
VyOS has two distinct modes: operational mode and configuration mode. When a user logins,
the CLI is in the operational mode. In this mode, only the running (effective) config is accessible for reading.
When a user enters the "configure" command, a configuration session is setup. Every config session
has its *proposed* (or *session*) config built on top of the current running config. When changes are commited, if commit succeeds,
the proposed config is merged into the running config.
In configuration mode, "base" functions like `exists`, `return_value` return values from the session config,
while functions prefixed "effective" return values from the running config.
In operational mode, all functions return values from the running config.
"""
import re
import json
from copy import deepcopy
from typing import Union
import vyos.configtree
from vyos.xml_ref import multi_to_list
from vyos.xml_ref import from_source
from vyos.xml_ref import ext_dict_merge
from vyos.xml_ref import relative_defaults
from vyos.utils.dict import get_sub_dict
from vyos.utils.dict import mangle_dict_keys
from vyos.configsource import ConfigSource
from vyos.configsource import ConfigSourceSession
class ConfigDict(dict):
_from_defaults = {}
_dict_kwargs = {}
def from_defaults(self, path: list[str]) -> bool:
return from_source(self._from_defaults, path)
@property
def kwargs(self) -> dict:
return self._dict_kwargs
def config_dict_merge(src: dict, dest: Union[dict, ConfigDict]) -> ConfigDict:
if not isinstance(dest, ConfigDict):
dest = ConfigDict(dest)
return ext_dict_merge(src, dest)
+def config_dict_mangle_acme(name, cli_dict):
+ """
+ Load CLI PKI dictionary and if an ACME certificate is used, load it's content
+ and place it into the CLI dictionary as it would be a "regular" CLI PKI based
+ certificate with private key
+ """
+ from vyos.base import ConfigError
+ from vyos.defaults import directories
+ from vyos.utils.file import read_file
+ from vyos.pki import encode_certificate
+ from vyos.pki import encode_private_key
+ from vyos.pki import load_certificate
+ from vyos.pki import load_private_key
+
+ try:
+ vyos_certbot_dir = directories['certbot']
+
+ if 'acme' in cli_dict:
+ tmp = read_file(f'{vyos_certbot_dir}/live/{name}/cert.pem')
+ tmp = load_certificate(tmp, wrap_tags=False)
+ cert_base64 = "".join(encode_certificate(tmp).strip().split("\n")[1:-1])
+
+ tmp = read_file(f'{vyos_certbot_dir}/live/{name}/privkey.pem')
+ tmp = load_private_key(tmp, wrap_tags=False)
+ key_base64 = "".join(encode_private_key(tmp).strip().split("\n")[1:-1])
+ # install ACME based PEM keys into "regular" CLI config keys
+ cli_dict.update({'certificate' : cert_base64, 'private' : {'key' : key_base64}})
+ except:
+ raise ConfigError(f'Unable to load ACME certificates for "{name}"!')
+
+ return cli_dict
+
class Config(object):
"""
The class of config access objects.
Internally, in the current implementation, this object is *almost* stateless,
the only state it keeps is relative *config path* for convenient access to config
subtrees.
"""
def __init__(self, session_env=None, config_source=None):
if config_source is None:
self._config_source = ConfigSourceSession(session_env)
else:
if not isinstance(config_source, ConfigSource):
raise TypeError("config_source not of type ConfigSource")
self._config_source = config_source
self._level = []
self._dict_cache = {}
(self._running_config,
self._session_config) = self._config_source.get_configtree_tuple()
def get_config_tree(self, effective=False):
if effective:
return self._running_config
return self._session_config
def _make_path(self, path):
# Backwards-compatibility stuff: original implementation used string paths
# libvyosconfig paths are lists, but since node names cannot contain whitespace,
# splitting at whitespace is reasonably safe.
# It may cause problems with exists() when it's used for checking values,
# since values may contain whitespace.
if isinstance(path, str):
path = re.split(r'\s+', path)
elif isinstance(path, list):
pass
else:
raise TypeError("Path must be a whitespace-separated string or a list")
return (self._level + path)
def set_level(self, path):
"""
Set the *edit level*, that is, a relative config tree path.
Once set, all operations will be relative to this path,
for example, after ``set_level("system")``, calling
``exists("name-server")`` is equivalent to calling
``exists("system name-server"`` without ``set_level``.
Args:
path (str|list): relative config path
"""
# Make sure there's always a space between default path (level)
# and path supplied as method argument
# XXX: for small strings in-place concatenation is not a problem
if isinstance(path, str):
if path:
self._level = re.split(r'\s+', path)
else:
self._level = []
elif isinstance(path, list):
self._level = path.copy()
else:
raise TypeError("Level path must be either a whitespace-separated string or a list")
def get_level(self):
"""
Gets the current edit level.
Returns:
str: current edit level
"""
return(self._level.copy())
def exists(self, path):
"""
Checks if a node or value with given path exists in the proposed config.
Args:
path (str): Configuration tree path
Returns:
True if node or value exists in the proposed config, False otherwise
Note:
This function should not be used outside of configuration sessions.
In operational mode scripts, use ``exists_effective``.
"""
if self._session_config is None:
return False
# Assume the path is a node path first
if self._session_config.exists(self._make_path(path)):
return True
else:
# If that check fails, it may mean the path has a value at the end.
# libvyosconfig exists() works only for _nodes_, not _values_
# libvyattacfg also worked for values, so we emulate that case here
if isinstance(path, str):
path = re.split(r'\s+', path)
path_without_value = path[:-1]
try:
# return_values() is safe to use with single-value nodes,
# it simply returns a single-item list in that case.
values = self._session_config.return_values(self._make_path(path_without_value))
# If we got this far, the node does exist and has values,
# so we need to check if it has the value in question among its values.
return (path[-1] in values)
except vyos.configtree.ConfigTreeError:
# Even the parent node doesn't exist at all
return False
def session_changed(self):
"""
Returns:
True if the config session has uncommited changes, False otherwise.
"""
return self._config_source.session_changed()
def in_session(self):
"""
Returns:
True if called from a configuration session, False otherwise.
"""
return self._config_source.in_session()
def show_config(self, path=[], default=None, effective=False):
"""
Args:
path (str list): Configuration tree path, or empty
default (str): Default value to return
Returns:
str: working configuration
"""
return self._config_source.show_config(path, default, effective)
def get_cached_root_dict(self, effective=False):
cached = self._dict_cache.get(effective, {})
if cached:
return cached
if effective:
config = self._running_config
else:
config = self._session_config
if config:
config_dict = json.loads(config.to_json())
else:
config_dict = {}
self._dict_cache[effective] = config_dict
return config_dict
def verify_mangling(self, key_mangling):
if not (isinstance(key_mangling, tuple) and \
(len(key_mangling) == 2) and \
isinstance(key_mangling[0], str) and \
isinstance(key_mangling[1], str)):
raise ValueError("key_mangling must be a tuple of two strings")
def get_config_dict(self, path=[], effective=False, key_mangling=None,
get_first_key=False, no_multi_convert=False,
no_tag_node_value_mangle=False,
with_defaults=False,
with_recursive_defaults=False,
with_pki=False):
"""
Args:
path (str list): Configuration tree path, can be empty
effective=False: effective or session config
key_mangling=None: mangle dict keys according to regex and replacement
get_first_key=False: if k = path[:-1], return sub-dict d[k] instead of {k: d[k]}
no_multi_convert=False: if convert, return single value of multi node as list
Returns: a dict representation of the config under path
"""
kwargs = locals().copy()
del kwargs['self']
del kwargs['no_multi_convert']
del kwargs['with_defaults']
del kwargs['with_recursive_defaults']
del kwargs['with_pki']
lpath = self._make_path(path)
root_dict = self.get_cached_root_dict(effective)
conf_dict = get_sub_dict(root_dict, lpath, get_first_key=get_first_key)
rpath = lpath if get_first_key else lpath[:-1]
if not no_multi_convert:
conf_dict = multi_to_list(rpath, conf_dict)
if key_mangling is not None:
self.verify_mangling(key_mangling)
conf_dict = mangle_dict_keys(conf_dict,
key_mangling[0], key_mangling[1],
abs_path=rpath,
no_tag_node_value_mangle=no_tag_node_value_mangle)
if with_defaults or with_recursive_defaults:
defaults = self.get_config_defaults(**kwargs,
recursive=with_recursive_defaults)
conf_dict = config_dict_merge(defaults, conf_dict)
else:
conf_dict = ConfigDict(conf_dict)
if with_pki and conf_dict:
pki_dict = self.get_config_dict(['pki'], key_mangling=('-', '_'),
no_tag_node_value_mangle=True,
get_first_key=True)
if pki_dict:
+ if 'certificate' in pki_dict:
+ for certificate in pki_dict['certificate']:
+ pki_dict['certificate'][certificate] = config_dict_mangle_acme(
+ certificate, pki_dict['certificate'][certificate])
+
conf_dict['pki'] = pki_dict
# save optional args for a call to get_config_defaults
setattr(conf_dict, '_dict_kwargs', kwargs)
return conf_dict
def get_config_defaults(self, path=[], effective=False, key_mangling=None,
no_tag_node_value_mangle=False, get_first_key=False,
recursive=False) -> dict:
lpath = self._make_path(path)
root_dict = self.get_cached_root_dict(effective)
conf_dict = get_sub_dict(root_dict, lpath, get_first_key)
defaults = relative_defaults(lpath, conf_dict,
get_first_key=get_first_key,
recursive=recursive)
rpath = lpath if get_first_key else lpath[:-1]
if key_mangling is not None:
self.verify_mangling(key_mangling)
defaults = mangle_dict_keys(defaults,
key_mangling[0], key_mangling[1],
abs_path=rpath,
no_tag_node_value_mangle=no_tag_node_value_mangle)
return defaults
def merge_defaults(self, config_dict: ConfigDict, recursive=False):
if not isinstance(config_dict, ConfigDict):
raise TypeError('argument is not of type ConfigDict')
if not config_dict.kwargs:
raise ValueError('argument missing metadata')
args = config_dict.kwargs
d = self.get_config_defaults(**args, recursive=recursive)
config_dict = config_dict_merge(d, config_dict)
return config_dict
def is_multi(self, path):
"""
Args:
path (str): Configuration tree path
Returns:
True if a node can have multiple values, False otherwise.
Note:
It also returns False if node doesn't exist.
"""
self._config_source.set_level(self.get_level)
return self._config_source.is_multi(path)
def is_tag(self, path):
"""
Args:
path (str): Configuration tree path
Returns:
True if a node is a tag node, False otherwise.
Note:
It also returns False if node doesn't exist.
"""
self._config_source.set_level(self.get_level)
return self._config_source.is_tag(path)
def is_leaf(self, path):
"""
Args:
path (str): Configuration tree path
Returns:
True if a node is a leaf node, False otherwise.
Note:
It also returns False if node doesn't exist.
"""
self._config_source.set_level(self.get_level)
return self._config_source.is_leaf(path)
def return_value(self, path, default=None):
"""
Retrieve a value of single-value leaf node in the running or proposed config
Args:
path (str): Configuration tree path
default (str): Default value to return if node does not exist
Returns:
str: Node value, if it has any
None: if node is valueless *or* if it doesn't exist
Note:
Due to the issue with treatment of valueless nodes by this function,
valueless nodes should be checked with ``exists`` instead.
This function cannot be used outside a configuration session.
In operational mode scripts, use ``return_effective_value``.
"""
if self._session_config:
try:
value = self._session_config.return_value(self._make_path(path))
except vyos.configtree.ConfigTreeError:
value = None
else:
value = None
if not value:
return(default)
else:
return(value)
def return_values(self, path, default=[]):
"""
Retrieve all values of a multi-value leaf node in the running or proposed config
Args:
path (str): Configuration tree path
Returns:
str list: Node values, if it has any
[]: if node does not exist
Note:
This function cannot be used outside a configuration session.
In operational mode scripts, use ``return_effective_values``.
"""
if self._session_config:
try:
values = self._session_config.return_values(self._make_path(path))
except vyos.configtree.ConfigTreeError:
values = []
else:
values = []
if not values:
return(default.copy())
else:
return(values)
def list_nodes(self, path, default=[]):
"""
Retrieve names of all children of a tag node in the running or proposed config
Args:
path (str): Configuration tree path
Returns:
string list: child node names
"""
if self._session_config:
try:
nodes = self._session_config.list_nodes(self._make_path(path))
except vyos.configtree.ConfigTreeError:
nodes = []
else:
nodes = []
if not nodes:
return(default.copy())
else:
return(nodes)
def exists_effective(self, path):
"""
Checks if a node or value exists in the running (effective) config.
Args:
path (str): Configuration tree path
Returns:
True if node exists in the running config, False otherwise
Note:
This function is safe to use in operational mode. In configuration mode,
it ignores uncommited changes.
"""
if self._running_config is None:
return False
# Assume the path is a node path first
if self._running_config.exists(self._make_path(path)):
return True
else:
# If that check fails, it may mean the path has a value at the end.
# libvyosconfig exists() works only for _nodes_, not _values_
# libvyattacfg also worked for values, so we emulate that case here
if isinstance(path, str):
path = re.split(r'\s+', path)
path_without_value = path[:-1]
try:
# return_values() is safe to use with single-value nodes,
# it simply returns a single-item list in that case.
values = self._running_config.return_values(self._make_path(path_without_value))
# If we got this far, the node does exist and has values,
# so we need to check if it has the value in question among its values.
return (path[-1] in values)
except vyos.configtree.ConfigTreeError:
# Even the parent node doesn't exist at all
return False
def return_effective_value(self, path, default=None):
"""
Retrieve a values of a single-value leaf node in a running (effective) config
Args:
path (str): Configuration tree path
default (str): Default value to return if node does not exist
Returns:
str: Node value
"""
if self._running_config:
try:
value = self._running_config.return_value(self._make_path(path))
except vyos.configtree.ConfigTreeError:
value = None
else:
value = None
if not value:
return(default)
else:
return(value)
def return_effective_values(self, path, default=[]):
"""
Retrieve all values of a multi-value node in a running (effective) config
Args:
path (str): Configuration tree path
Returns:
str list: A list of values
"""
if self._running_config:
try:
values = self._running_config.return_values(self._make_path(path))
except vyos.configtree.ConfigTreeError:
values = []
else:
values = []
if not values:
return(default.copy())
else:
return(values)
def list_effective_nodes(self, path, default=[]):
"""
Retrieve names of all children of a tag node in the running config
Args:
path (str): Configuration tree path
Returns:
str list: child node names
"""
if self._running_config:
try:
nodes = self._running_config.list_nodes(self._make_path(path))
except vyos.configtree.ConfigTreeError:
nodes = []
else:
nodes = []
if not nodes:
return(default.copy())
else:
return(nodes)
diff --git a/smoketest/scripts/cli/test_vpn_ipsec.py b/smoketest/scripts/cli/test_vpn_ipsec.py
index 17e12bcaf..f5369ee7a 100755
--- a/smoketest/scripts/cli/test_vpn_ipsec.py
+++ b/smoketest/scripts/cli/test_vpn_ipsec.py
@@ -1,898 +1,899 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021-2023 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
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.utils.process import call
from vyos.utils.process import process_named_running
from vyos.utils.file import read_file
ethernet_path = ['interfaces', 'ethernet']
tunnel_path = ['interfaces', 'tunnel']
vti_path = ['interfaces', 'vti']
nhrp_path = ['protocols', 'nhrp']
base_path = ['vpn', 'ipsec']
charon_file = '/etc/strongswan.d/charon.conf'
dhcp_waiting_file = '/tmp/ipsec_dhcp_waiting'
swanctl_file = '/etc/swanctl/swanctl.conf'
peer_ip = '203.0.113.45'
connection_name = 'main-branch'
local_id = 'left'
remote_id = 'right'
interface = 'eth1'
vif = '100'
esp_group = 'MyESPGroup'
ike_group = 'MyIKEGroup'
secret = 'MYSECRETKEY'
PROCESS_NAME = 'charon-systemd'
regex_uuid4 = '[0-9a-fA-F]{8}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{12}'
ca_name = 'MyVyOS-CA'
ca_pem = """
MIICMDCCAdegAwIBAgIUBCzIjYvD7SPbx5oU18IYg7NVxQ0wCgYIKoZIzj0EAwIw
ZzELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv
bWUtQ2l0eTENMAsGA1UECgwEVnlPUzEgMB4GA1UEAwwXSVBTZWMgU21va2V0ZXN0
IFJvb3QgQ0EwHhcNMjMwOTI0MTIwMzQxWhcNMzMwOTIxMTIwMzQxWjBnMQswCQYD
VQQGEwJHQjETMBEGA1UECAwKU29tZS1TdGF0ZTESMBAGA1UEBwwJU29tZS1DaXR5
MQ0wCwYDVQQKDARWeU9TMSAwHgYDVQQDDBdJUFNlYyBTbW9rZXRlc3QgUm9vdCBD
QTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABEh8/yU572B3zmFxrGgHk+H7grYt
EHUJodY3gXNWMHz0gySrbGhsGtECDfP/G+T4Suk7cuVzB1wnLocSafD8TcqjYTBf
MA8GA1UdEwEB/wQFMAMBAf8wDgYDVR0PAQH/BAQDAgGGMB0GA1UdJQQWMBQGCCsG
AQUFBwMCBggrBgEFBQcDATAdBgNVHQ4EFgQUTYoQJNlk7X87/gRegHnCnPef39Aw
CgYIKoZIzj0EAwIDRwAwRAIgX1spXjrUc10r3g/Zm4O31LU5O08J2vVqFo94zHE5
0VgCIG4JK9Zg5O/yn4mYksZux7efiHRUzL2y2TXQ9IqrqM8W
"""
int_ca_name = 'MyVyOS-IntCA'
int_ca_pem = """
MIICYDCCAgWgAwIBAgIUcFx2BVYErHI+SneyPYHijxXt1cgwCgYIKoZIzj0EAwIw
ZzELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv
bWUtQ2l0eTENMAsGA1UECgwEVnlPUzEgMB4GA1UEAwwXSVBTZWMgU21va2V0ZXN0
IFJvb3QgQ0EwHhcNMjMwOTI0MTIwNTE5WhcNMzMwOTIwMTIwNTE5WjBvMQswCQYD
VQQGEwJHQjETMBEGA1UECAwKU29tZS1TdGF0ZTESMBAGA1UEBwwJU29tZS1DaXR5
MQ0wCwYDVQQKDARWeU9TMSgwJgYDVQQDDB9JUFNlYyBTbW9rZXRlc3QgSW50ZXJt
ZWRpYXRlIENBMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEIHw2G5dq3c715AcA
tzR++dYu1fLRFmHzRGTZOT7hLrh2Fg4hnKFPLOeUA5Qi50xCvjJ9JnonTyy2RfRH
axYizKOBhjCBgzASBgNVHRMBAf8ECDAGAQH/AgEAMA4GA1UdDwEB/wQEAwIBhjAd
BgNVHSUEFjAUBggrBgEFBQcDAgYIKwYBBQUHAwEwHQYDVR0OBBYEFC9KrFYtA+hO
l7vdMbWxTMAyLB7BMB8GA1UdIwQYMBaAFE2KECTZZO1/O/4EXoB5wpz3n9/QMAoG
CCqGSM49BAMCA0kAMEYCIQCnqWbElgOL9dGO3iLxasFNq/hM7vM/DzaiHi4BowxW
0gIhAMohefNj+QgLfPhvyODHIPE9LMyfp7lJEaCC2K8PCSFD
"""
peer_name = 'peer1'
peer_cert = """
MIICSTCCAfCgAwIBAgIUPxYleUgCo/glVVePze3QmAFgi6MwCgYIKoZIzj0EAwIw
bzELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv
bWUtQ2l0eTENMAsGA1UECgwEVnlPUzEoMCYGA1UEAwwfSVBTZWMgU21va2V0ZXN0
IEludGVybWVkaWF0ZSBDQTAeFw0yMzA5MjQxMjA2NDJaFw0yODA5MjIxMjA2NDJa
MGQxCzAJBgNVBAYTAkdCMRMwEQYDVQQIDApTb21lLVN0YXRlMRIwEAYDVQQHDAlT
b21lLUNpdHkxDTALBgNVBAoMBFZ5T1MxHTAbBgNVBAMMFElQU2VjIFNtb2tldGVz
dCBQZWVyMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEZJtuTDu84uy++GMwRNLl
10JAXZxXQSDl+CdTWwjbQZURcdY+ia7BoaoYX/0VKPel3Se64rIUQQLQoY/9MJb9
UKN1MHMwDAYDVR0TAQH/BAIwADAOBgNVHQ8BAf8EBAMCB4AwEwYDVR0lBAwwCgYI
KwYBBQUHAwEwHQYDVR0OBBYEFNJCdnkm3cAmf04UwOKL7IqMJ6OXMB8GA1UdIwQY
MBaAFC9KrFYtA+hOl7vdMbWxTMAyLB7BMAoGCCqGSM49BAMCA0cAMEQCIGVnDRUy
UJ0U/deDvrBo1+AakZndkNAMN/XNo5a5GzhEAiBCY7E/3b0BIO8FiIbVB3iDcaxg
g7ET2RgWxvhEoN3ZRw==
"""
peer_key = """
MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgVDEZDK7q/T+tiJUV
WLKS3ZYDfZ4lZv0C1gJpYq0gWP2hRANCAARkm25MO7zi7L74YzBE0uXXQkBdnFdB
IOX4J1NbCNtBlRFx1j6JrsGhqhhf/RUo96XdJ7rishRBAtChj/0wlv1Q
"""
swanctl_dir = '/etc/swanctl'
CERT_PATH = f'{swanctl_dir}/x509/'
CA_PATH = f'{swanctl_dir}/x509ca/'
class TestVPNIPsec(VyOSUnitTestSHIM.TestCase):
skip_process_check = False
@classmethod
def setUpClass(cls):
super(TestVPNIPsec, cls).setUpClass()
# ensure we can also run this test on a live system - so lets clean
# out the current configuration :)
cls.cli_delete(cls, base_path)
+ cls.cli_delete(cls, ['pki'])
cls.cli_set(cls, base_path + ['interface', f'{interface}.{vif}'])
@classmethod
def tearDownClass(cls):
super(TestVPNIPsec, cls).tearDownClass()
cls.cli_delete(cls, base_path + ['interface', f'{interface}.{vif}'])
def setUp(self):
# Set IKE/ESP Groups
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'encryption', 'aes128'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'hash', 'sha1'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'dh-group', '2'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'encryption', 'aes128'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'hash', 'sha1'])
def tearDown(self):
# Check for running process
if not self.skip_process_check:
self.assertTrue(process_named_running(PROCESS_NAME))
else:
self.skip_process_check = False # Reset
self.cli_delete(base_path)
self.cli_delete(tunnel_path)
self.cli_commit()
# Check for no longer running process
self.assertFalse(process_named_running(PROCESS_NAME))
def setupPKI(self):
self.cli_set(['pki', 'ca', ca_name, 'certificate', ca_pem.replace('\n','')])
self.cli_set(['pki', 'ca', int_ca_name, 'certificate', int_ca_pem.replace('\n','')])
self.cli_set(['pki', 'certificate', peer_name, 'certificate', peer_cert.replace('\n','')])
self.cli_set(['pki', 'certificate', peer_name, 'private', 'key', peer_key.replace('\n','')])
def tearDownPKI(self):
self.cli_delete(['pki'])
def test_01_dhcp_fail_handling(self):
# Skip process check - connection is not created for this test
self.skip_process_check = True
# Interface for dhcp-interface
self.cli_set(ethernet_path + [interface, 'vif', vif, 'address', 'dhcp']) # Use VLAN to avoid getting IP from qemu dhcp server
# vpn ipsec auth psk <tag> id <x.x.x.x>
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', local_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', remote_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', peer_ip])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'secret', secret])
# Site to site
peer_base_path = base_path + ['site-to-site', 'peer', connection_name]
self.cli_set(peer_base_path + ['authentication', 'mode', 'pre-shared-secret'])
self.cli_set(peer_base_path + ['ike-group', ike_group])
self.cli_set(peer_base_path + ['default-esp-group', esp_group])
self.cli_set(peer_base_path + ['dhcp-interface', f'{interface}.{vif}'])
self.cli_set(peer_base_path + ['tunnel', '1', 'protocol', 'gre'])
self.cli_commit()
self.assertTrue(os.path.exists(dhcp_waiting_file))
dhcp_waiting = read_file(dhcp_waiting_file)
self.assertIn(f'{interface}.{vif}', dhcp_waiting) # Ensure dhcp-failed interface was added for dhclient hook
self.cli_delete(ethernet_path + [interface, 'vif', vif, 'address'])
def test_02_site_to_site(self):
self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2'])
local_address = '192.0.2.10'
priority = '20'
life_bytes = '100000'
life_packets = '2000000'
# vpn ipsec auth psk <tag> id <x.x.x.x>
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', local_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', remote_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', local_address])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', peer_ip])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'secret', secret])
# Site to site
peer_base_path = base_path + ['site-to-site', 'peer', connection_name]
self.cli_set(base_path + ['esp-group', esp_group, 'life-bytes', life_bytes])
self.cli_set(base_path + ['esp-group', esp_group, 'life-packets', life_packets])
self.cli_set(peer_base_path + ['authentication', 'mode', 'pre-shared-secret'])
self.cli_set(peer_base_path + ['ike-group', ike_group])
self.cli_set(peer_base_path + ['default-esp-group', esp_group])
self.cli_set(peer_base_path + ['local-address', local_address])
self.cli_set(peer_base_path + ['remote-address', peer_ip])
self.cli_set(peer_base_path + ['tunnel', '1', 'protocol', 'tcp'])
self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.10.0/24'])
self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.11.0/24'])
self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'port', '443'])
self.cli_set(peer_base_path + ['tunnel', '1', 'remote', 'prefix', '172.17.10.0/24'])
self.cli_set(peer_base_path + ['tunnel', '1', 'remote', 'prefix', '172.17.11.0/24'])
self.cli_set(peer_base_path + ['tunnel', '1', 'remote', 'port', '443'])
self.cli_set(peer_base_path + ['tunnel', '2', 'local', 'prefix', '10.1.0.0/16'])
self.cli_set(peer_base_path + ['tunnel', '2', 'remote', 'prefix', '10.2.0.0/16'])
self.cli_set(peer_base_path + ['tunnel', '2', 'priority', priority])
self.cli_commit()
# Verify strongSwan configuration
swanctl_conf = read_file(swanctl_file)
swanctl_conf_lines = [
f'version = 2',
f'auth = psk',
f'life_bytes = {life_bytes}',
f'life_packets = {life_packets}',
f'rekey_time = 28800s', # default value
f'proposals = aes128-sha1-modp1024',
f'esp_proposals = aes128-sha1-modp1024',
f'life_time = 3600s', # default value
f'local_addrs = {local_address} # dhcp:no',
f'remote_addrs = {peer_ip}',
f'mode = tunnel',
f'{connection_name}-tunnel-1',
f'local_ts = 172.16.10.0/24[tcp/443],172.16.11.0/24[tcp/443]',
f'remote_ts = 172.17.10.0/24[tcp/443],172.17.11.0/24[tcp/443]',
f'mode = tunnel',
f'{connection_name}-tunnel-2',
f'local_ts = 10.1.0.0/16',
f'remote_ts = 10.2.0.0/16',
f'priority = {priority}',
f'mode = tunnel',
]
for line in swanctl_conf_lines:
self.assertIn(line, swanctl_conf)
swanctl_secrets_lines = [
f'id-{regex_uuid4} = "{local_id}"',
f'id-{regex_uuid4} = "{remote_id}"',
f'id-{regex_uuid4} = "{local_address}"',
f'id-{regex_uuid4} = "{peer_ip}"',
f'secret = "{secret}"'
]
for line in swanctl_secrets_lines:
self.assertRegex(swanctl_conf, fr'{line}')
def test_03_site_to_site_vti(self):
local_address = '192.0.2.10'
vti = 'vti10'
# IKE
self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2'])
self.cli_set(base_path + ['ike-group', ike_group, 'disable-mobike'])
# ESP
self.cli_set(base_path + ['esp-group', esp_group, 'compression'])
# VTI interface
self.cli_set(vti_path + [vti, 'address', '10.1.1.1/24'])
# vpn ipsec auth psk <tag> id <x.x.x.x>
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', local_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', remote_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', peer_ip])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'secret', secret])
# Site to site
peer_base_path = base_path + ['site-to-site', 'peer', connection_name]
self.cli_set(peer_base_path + ['authentication', 'mode', 'pre-shared-secret'])
self.cli_set(peer_base_path + ['connection-type', 'none'])
self.cli_set(peer_base_path + ['force-udp-encapsulation'])
self.cli_set(peer_base_path + ['ike-group', ike_group])
self.cli_set(peer_base_path + ['default-esp-group', esp_group])
self.cli_set(peer_base_path + ['local-address', local_address])
self.cli_set(peer_base_path + ['remote-address', peer_ip])
self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.10.0/24'])
self.cli_set(peer_base_path + ['tunnel', '1', 'local', 'prefix', '172.16.11.0/24'])
self.cli_set(peer_base_path + ['tunnel', '1', 'remote', 'prefix', '172.17.10.0/24'])
self.cli_set(peer_base_path + ['tunnel', '1', 'remote', 'prefix', '172.17.11.0/24'])
self.cli_set(peer_base_path + ['vti', 'bind', vti])
self.cli_set(peer_base_path + ['vti', 'esp-group', esp_group])
self.cli_commit()
swanctl_conf = read_file(swanctl_file)
if_id = vti.lstrip('vti')
# The key defaults to 0 and will match any policies which similarly do
# not have a lookup key configuration - thus we shift the key by one
# to also support a vti0 interface
if_id = str(int(if_id) +1)
swanctl_conf_lines = [
f'version = 2',
f'auth = psk',
f'proposals = aes128-sha1-modp1024',
f'esp_proposals = aes128-sha1-modp1024',
f'local_addrs = {local_address} # dhcp:no',
f'mobike = no',
f'remote_addrs = {peer_ip}',
f'mode = tunnel',
f'local_ts = 172.16.10.0/24,172.16.11.0/24',
f'remote_ts = 172.17.10.0/24,172.17.11.0/24',
f'ipcomp = yes',
f'start_action = none',
f'if_id_in = {if_id}', # will be 11 for vti10 - shifted by one
f'if_id_out = {if_id}',
f'updown = "/etc/ipsec.d/vti-up-down {vti}"'
]
for line in swanctl_conf_lines:
self.assertIn(line, swanctl_conf)
swanctl_secrets_lines = [
f'id-{regex_uuid4} = "{local_id}"',
f'id-{regex_uuid4} = "{remote_id}"',
f'secret = "{secret}"'
]
for line in swanctl_secrets_lines:
self.assertRegex(swanctl_conf, fr'{line}')
def test_04_dmvpn(self):
tunnel_if = 'tun100'
nhrp_secret = 'secret'
ike_lifetime = '3600'
esp_lifetime = '1800'
# Tunnel
self.cli_set(tunnel_path + [tunnel_if, 'address', '172.16.253.134/29'])
self.cli_set(tunnel_path + [tunnel_if, 'encapsulation', 'gre'])
self.cli_set(tunnel_path + [tunnel_if, 'source-address', '192.0.2.1'])
self.cli_set(tunnel_path + [tunnel_if, 'enable-multicast'])
self.cli_set(tunnel_path + [tunnel_if, 'parameters', 'ip', 'key', '1'])
# NHRP
self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'cisco-authentication', nhrp_secret])
self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'holding-time', '300'])
self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'multicast', 'dynamic'])
self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'redirect'])
self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'shortcut'])
# IKE/ESP Groups
self.cli_set(base_path + ['esp-group', esp_group, 'lifetime', esp_lifetime])
self.cli_set(base_path + ['esp-group', esp_group, 'mode', 'transport'])
self.cli_set(base_path + ['esp-group', esp_group, 'pfs', 'dh-group2'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'hash', 'sha1'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'encryption', '3des'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'hash', 'md5'])
self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev1'])
self.cli_set(base_path + ['ike-group', ike_group, 'lifetime', ike_lifetime])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'dh-group', '2'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'hash', 'sha1'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'prf', 'prfsha1'])
# Profile
self.cli_set(base_path + ['profile', 'NHRPVPN', 'authentication', 'mode', 'pre-shared-secret'])
self.cli_set(base_path + ['profile', 'NHRPVPN', 'authentication', 'pre-shared-secret', nhrp_secret])
self.cli_set(base_path + ['profile', 'NHRPVPN', 'bind', 'tunnel', tunnel_if])
self.cli_set(base_path + ['profile', 'NHRPVPN', 'esp-group', esp_group])
self.cli_set(base_path + ['profile', 'NHRPVPN', 'ike-group', ike_group])
self.cli_commit()
swanctl_conf = read_file(swanctl_file)
swanctl_lines = [
f'proposals = aes128-sha1-modp1024,aes256-sha1-prfsha1-modp1024',
f'version = 1',
f'rekey_time = {ike_lifetime}s',
f'rekey_time = {esp_lifetime}s',
f'esp_proposals = aes128-sha1-modp1024,aes256-sha1-modp1024,3des-md5-modp1024',
f'local_ts = dynamic[gre]',
f'remote_ts = dynamic[gre]',
f'mode = transport',
f'secret = {nhrp_secret}'
]
for line in swanctl_lines:
self.assertIn(line, swanctl_conf)
# There is only one NHRP test so no need to delete this globally in tearDown()
self.cli_delete(nhrp_path)
def test_05_x509_site2site(self):
# Enable PKI
self.setupPKI()
vti = 'vti20'
self.cli_set(vti_path + [vti, 'address', '192.168.0.1/31'])
peer_ip = '172.18.254.202'
connection_name = 'office'
local_address = '172.18.254.201'
peer_base_path = base_path + ['site-to-site', 'peer', connection_name]
self.cli_set(peer_base_path + ['authentication', 'local-id', peer_name])
self.cli_set(peer_base_path + ['authentication', 'mode', 'x509'])
self.cli_set(peer_base_path + ['authentication', 'remote-id', 'peer2'])
self.cli_set(peer_base_path + ['authentication', 'x509', 'ca-certificate', int_ca_name])
self.cli_set(peer_base_path + ['authentication', 'x509', 'certificate', peer_name])
self.cli_set(peer_base_path + ['connection-type', 'initiate'])
self.cli_set(peer_base_path + ['ike-group', ike_group])
self.cli_set(peer_base_path + ['ikev2-reauth', 'inherit'])
self.cli_set(peer_base_path + ['local-address', local_address])
self.cli_set(peer_base_path + ['remote-address', peer_ip])
self.cli_set(peer_base_path + ['vti', 'bind', vti])
self.cli_set(peer_base_path + ['vti', 'esp-group', esp_group])
self.cli_commit()
swanctl_conf = read_file(swanctl_file)
tmp = peer_ip.replace('.', '-')
if_id = vti.lstrip('vti')
# The key defaults to 0 and will match any policies which similarly do
# not have a lookup key configuration - thus we shift the key by one
# to also support a vti0 interface
if_id = str(int(if_id) +1)
swanctl_lines = [
f'{connection_name}',
f'version = 0', # key-exchange not set - defaulting to 0 for ikev1 and ikev2
f'send_cert = always',
f'mobike = yes',
f'keyingtries = 0',
f'id = "{peer_name}"',
f'auth = pubkey',
f'certs = {peer_name}.pem',
f'proposals = aes128-sha1-modp1024',
f'esp_proposals = aes128-sha1-modp1024',
f'local_addrs = {local_address} # dhcp:no',
f'remote_addrs = {peer_ip}',
f'local_ts = 0.0.0.0/0,::/0',
f'remote_ts = 0.0.0.0/0,::/0',
f'updown = "/etc/ipsec.d/vti-up-down {vti}"',
f'if_id_in = {if_id}', # will be 11 for vti10
f'if_id_out = {if_id}',
f'ipcomp = no',
f'mode = tunnel',
f'start_action = start',
]
for line in swanctl_lines:
self.assertIn(line, swanctl_conf)
swanctl_secrets_lines = [
f'{connection_name}',
f'file = {peer_name}.pem',
]
for line in swanctl_secrets_lines:
self.assertIn(line, swanctl_conf)
# Check Root CA, Intermediate CA and Peer cert/key pair is present
self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{int_ca_name}_1.pem')))
self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{int_ca_name}_2.pem')))
self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
# There is only one VTI test so no need to delete this globally in tearDown()
self.cli_delete(vti_path)
# Disable PKI
self.tearDownPKI()
def test_06_flex_vpn_vips(self):
local_address = '192.0.2.5'
local_id = 'vyos-r1'
remote_id = 'vyos-r2'
peer_base_path = base_path + ['site-to-site', 'peer', connection_name]
self.cli_set(tunnel_path + ['tun1', 'encapsulation', 'gre'])
self.cli_set(tunnel_path + ['tun1', 'source-address', local_address])
self.cli_set(base_path + ['interface', interface])
self.cli_set(base_path + ['options', 'flexvpn'])
self.cli_set(base_path + ['options', 'interface', 'tun1'])
self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2'])
# vpn ipsec auth psk <tag> id <x.x.x.x>
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', local_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', remote_id])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', local_address])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'id', peer_ip])
self.cli_set(base_path + ['authentication', 'psk', connection_name, 'secret', secret])
self.cli_set(peer_base_path + ['authentication', 'local-id', local_id])
self.cli_set(peer_base_path + ['authentication', 'mode', 'pre-shared-secret'])
self.cli_set(peer_base_path + ['authentication', 'remote-id', remote_id])
self.cli_set(peer_base_path + ['connection-type', 'initiate'])
self.cli_set(peer_base_path + ['ike-group', ike_group])
self.cli_set(peer_base_path + ['default-esp-group', esp_group])
self.cli_set(peer_base_path + ['local-address', local_address])
self.cli_set(peer_base_path + ['remote-address', peer_ip])
self.cli_set(peer_base_path + ['tunnel', '1', 'protocol', 'gre'])
self.cli_set(peer_base_path + ['virtual-address', '203.0.113.55'])
self.cli_set(peer_base_path + ['virtual-address', '203.0.113.56'])
self.cli_commit()
# Verify strongSwan configuration
swanctl_conf = read_file(swanctl_file)
swanctl_conf_lines = [
f'version = 2',
f'vips = 203.0.113.55, 203.0.113.56',
f'life_time = 3600s', # default value
f'local_addrs = {local_address} # dhcp:no',
f'remote_addrs = {peer_ip}',
f'{connection_name}-tunnel-1',
f'mode = tunnel',
]
for line in swanctl_conf_lines:
self.assertIn(line, swanctl_conf)
swanctl_secrets_lines = [
f'id-{regex_uuid4} = "{local_id}"',
f'id-{regex_uuid4} = "{remote_id}"',
f'id-{regex_uuid4} = "{peer_ip}"',
f'id-{regex_uuid4} = "{local_address}"',
f'secret = "{secret}"',
]
for line in swanctl_secrets_lines:
self.assertRegex(swanctl_conf, fr'{line}')
# Verify charon configuration
charon_conf = read_file(charon_file)
charon_conf_lines = [
f'# Cisco FlexVPN',
f'cisco_flexvpn = yes',
f'install_virtual_ip = yes',
f'install_virtual_ip_on = tun1',
]
for line in charon_conf_lines:
self.assertIn(line, charon_conf)
def test_07_ikev2_road_warrior(self):
# This is a known to be good configuration for Microsoft Windows 10 and Apple iOS 17
self.setupPKI()
ike_group = 'IKE-RW'
esp_group = 'ESP-RW'
conn_name = 'vyos-rw'
local_address = '192.0.2.1'
ip_pool_name = 'ra-rw-ipv4'
username = 'vyos'
password = 'secret'
ike_lifetime = '7200'
eap_lifetime = '3600'
local_id = 'ipsec.vyos.net'
name_servers = ['172.16.254.100', '172.16.254.101']
prefix = '172.16.250.0/28'
# IKE
self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2'])
self.cli_set(base_path + ['ike-group', ike_group, 'lifetime', ike_lifetime])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'hash', 'sha512'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'hash', 'sha256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'dh-group', '2'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'hash', 'sha256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'encryption', 'aes128gcm128'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'hash', 'sha256'])
# ESP
self.cli_set(base_path + ['esp-group', esp_group, 'lifetime', eap_lifetime])
self.cli_set(base_path + ['esp-group', esp_group, 'pfs', 'disable'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'hash', 'sha512'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'hash', 'sha384'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'hash', 'sha256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '4', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '4', 'hash', 'sha1'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '10', 'encryption', 'aes128gcm128'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '10', 'hash', 'sha256'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'local-id', local_id])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'local-users', 'username', username, 'password', password])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'server-mode', 'x509'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'x509', 'certificate', peer_name])
# verify() - CA cert required for x509 auth
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'x509', 'ca-certificate', ca_name])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'esp-group', esp_group])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'ike-group', ike_group])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'local-address', local_address])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'pool', ip_pool_name])
for ns in name_servers:
self.cli_set(base_path + ['remote-access', 'pool', ip_pool_name, 'name-server', ns])
self.cli_set(base_path + ['remote-access', 'pool', ip_pool_name, 'prefix', prefix])
self.cli_commit()
# verify applied configuration
swanctl_conf = read_file(swanctl_file)
swanctl_lines = [
f'{conn_name}',
f'remote_addrs = %any',
f'local_addrs = {local_address}',
f'proposals = aes256-sha512-modp2048,aes256-sha256-modp2048,aes256-sha256-modp1024,aes128gcm128-sha256-modp2048',
f'version = 2',
f'send_certreq = no',
f'rekey_time = {ike_lifetime}s',
f'keyingtries = 0',
f'pools = {ip_pool_name}',
f'id = "{local_id}"',
f'auth = pubkey',
f'certs = peer1.pem',
f'auth = eap-mschapv2',
f'eap_id = %any',
f'esp_proposals = aes256-sha512,aes256-sha384,aes256-sha256,aes256-sha1,aes128gcm128-sha256',
f'rekey_time = {eap_lifetime}s',
f'rand_time = 540s',
f'dpd_action = clear',
f'inactivity = 28800',
f'local_ts = 0.0.0.0/0,::/0',
]
for line in swanctl_lines:
self.assertIn(line, swanctl_conf)
swanctl_secrets_lines = [
f'eap-{conn_name}-{username}',
f'secret = "{password}"',
f'id-{conn_name}-{username} = "{username}"',
]
for line in swanctl_secrets_lines:
self.assertIn(line, swanctl_conf)
swanctl_pool_lines = [
f'{ip_pool_name}',
f'addrs = {prefix}',
f'dns = {",".join(name_servers)}',
]
for line in swanctl_pool_lines:
self.assertIn(line, swanctl_conf)
# Check Root CA, Intermediate CA and Peer cert/key pair is present
self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}_1.pem')))
self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
self.tearDownPKI()
def test_08_ikev2_road_warrior_client_auth_eap_tls(self):
# This is a known to be good configuration for Microsoft Windows 10 and Apple iOS 17
self.setupPKI()
ike_group = 'IKE-RW'
esp_group = 'ESP-RW'
conn_name = 'vyos-rw'
local_address = '192.0.2.1'
ip_pool_name = 'ra-rw-ipv4'
username = 'vyos'
password = 'secret'
ike_lifetime = '7200'
eap_lifetime = '3600'
local_id = 'ipsec.vyos.net'
name_servers = ['172.16.254.100', '172.16.254.101']
prefix = '172.16.250.0/28'
# IKE
self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2'])
self.cli_set(base_path + ['ike-group', ike_group, 'lifetime', ike_lifetime])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'hash', 'sha512'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'hash', 'sha256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'dh-group', '2'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'hash', 'sha256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'encryption', 'aes128gcm128'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'hash', 'sha256'])
# ESP
self.cli_set(base_path + ['esp-group', esp_group, 'lifetime', eap_lifetime])
self.cli_set(base_path + ['esp-group', esp_group, 'pfs', 'disable'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'hash', 'sha512'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'hash', 'sha384'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'hash', 'sha256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '4', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '4', 'hash', 'sha1'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '10', 'encryption', 'aes128gcm128'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '10', 'hash', 'sha256'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'local-id', local_id])
# Use EAP-TLS auth instead of default EAP-MSCHAPv2
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'client-mode', 'eap-tls'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'server-mode', 'x509'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'x509', 'certificate', peer_name])
# verify() - CA cert required for x509 auth
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'x509', 'ca-certificate', ca_name])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'esp-group', esp_group])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'ike-group', ike_group])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'local-address', local_address])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'pool', ip_pool_name])
for ns in name_servers:
self.cli_set(base_path + ['remote-access', 'pool', ip_pool_name, 'name-server', ns])
self.cli_set(base_path + ['remote-access', 'pool', ip_pool_name, 'prefix', prefix])
self.cli_commit()
# verify applied configuration
swanctl_conf = read_file(swanctl_file)
swanctl_lines = [
f'{conn_name}',
f'remote_addrs = %any',
f'local_addrs = {local_address}',
f'proposals = aes256-sha512-modp2048,aes256-sha256-modp2048,aes256-sha256-modp1024,aes128gcm128-sha256-modp2048',
f'version = 2',
f'send_certreq = no',
f'rekey_time = {ike_lifetime}s',
f'keyingtries = 0',
f'pools = {ip_pool_name}',
f'id = "{local_id}"',
f'auth = pubkey',
f'certs = peer1.pem',
f'auth = eap-tls',
f'eap_id = %any',
f'esp_proposals = aes256-sha512,aes256-sha384,aes256-sha256,aes256-sha1,aes128gcm128-sha256',
f'rekey_time = {eap_lifetime}s',
f'rand_time = 540s',
f'dpd_action = clear',
f'inactivity = 28800',
f'local_ts = 0.0.0.0/0,::/0',
]
for line in swanctl_lines:
self.assertIn(line, swanctl_conf)
swanctl_pool_lines = [
f'{ip_pool_name}',
f'addrs = {prefix}',
f'dns = {",".join(name_servers)}',
]
for line in swanctl_pool_lines:
self.assertIn(line, swanctl_conf)
# Check Root CA, Intermediate CA and Peer cert/key pair is present
self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}_1.pem')))
self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
self.tearDownPKI()
def test_09_ikev2_road_warrior_client_auth_x509(self):
# This is a known to be good configuration for Microsoft Windows 10 and Apple iOS 17
self.setupPKI()
ike_group = 'IKE-RW'
esp_group = 'ESP-RW'
conn_name = 'vyos-rw'
local_address = '192.0.2.1'
ip_pool_name = 'ra-rw-ipv4'
ike_lifetime = '7200'
eap_lifetime = '3600'
local_id = 'ipsec.vyos.net'
name_servers = ['172.16.254.100', '172.16.254.101']
prefix = '172.16.250.0/28'
# IKE
self.cli_set(base_path + ['ike-group', ike_group, 'key-exchange', 'ikev2'])
self.cli_set(base_path + ['ike-group', ike_group, 'lifetime', ike_lifetime])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '1', 'hash', 'sha512'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '2', 'hash', 'sha256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'dh-group', '2'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'encryption', 'aes256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '3', 'hash', 'sha256'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'dh-group', '14'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'encryption', 'aes128gcm128'])
self.cli_set(base_path + ['ike-group', ike_group, 'proposal', '10', 'hash', 'sha256'])
# ESP
self.cli_set(base_path + ['esp-group', esp_group, 'lifetime', eap_lifetime])
self.cli_set(base_path + ['esp-group', esp_group, 'pfs', 'disable'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '1', 'hash', 'sha512'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '2', 'hash', 'sha384'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '3', 'hash', 'sha256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '4', 'encryption', 'aes256'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '4', 'hash', 'sha1'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '10', 'encryption', 'aes128gcm128'])
self.cli_set(base_path + ['esp-group', esp_group, 'proposal', '10', 'hash', 'sha256'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'local-id', local_id])
# Use client-mode x509 instead of default EAP-MSCHAPv2
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'client-mode', 'x509'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'server-mode', 'x509'])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'x509', 'certificate', peer_name])
# verify() - CA cert required for x509 auth
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'authentication', 'x509', 'ca-certificate', ca_name])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'esp-group', esp_group])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'ike-group', ike_group])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'local-address', local_address])
self.cli_set(base_path + ['remote-access', 'connection', conn_name, 'pool', ip_pool_name])
for ns in name_servers:
self.cli_set(base_path + ['remote-access', 'pool', ip_pool_name, 'name-server', ns])
self.cli_set(base_path + ['remote-access', 'pool', ip_pool_name, 'prefix', prefix])
self.cli_commit()
# verify applied configuration
swanctl_conf = read_file(swanctl_file)
swanctl_lines = [
f'{conn_name}',
f'remote_addrs = %any',
f'local_addrs = {local_address}',
f'proposals = aes256-sha512-modp2048,aes256-sha256-modp2048,aes256-sha256-modp1024,aes128gcm128-sha256-modp2048',
f'version = 2',
f'send_certreq = no',
f'rekey_time = {ike_lifetime}s',
f'keyingtries = 0',
f'pools = {ip_pool_name}',
f'id = "{local_id}"',
f'auth = pubkey',
f'certs = peer1.pem',
f'esp_proposals = aes256-sha512,aes256-sha384,aes256-sha256,aes256-sha1,aes128gcm128-sha256',
f'rekey_time = {eap_lifetime}s',
f'rand_time = 540s',
f'dpd_action = clear',
f'inactivity = 28800',
f'local_ts = 0.0.0.0/0,::/0',
]
for line in swanctl_lines:
self.assertIn(line, swanctl_conf)
swanctl_unexpected_lines = [
f'auth = eap-',
f'eap_id'
]
for unexpected_line in swanctl_unexpected_lines:
self.assertNotIn(unexpected_line, swanctl_conf)
swanctl_pool_lines = [
f'{ip_pool_name}',
f'addrs = {prefix}',
f'dns = {",".join(name_servers)}',
]
for line in swanctl_pool_lines:
self.assertIn(line, swanctl_conf)
# Check Root CA, Intermediate CA and Peer cert/key pair is present
self.assertTrue(os.path.exists(os.path.join(CA_PATH, f'{ca_name}_1.pem')))
self.assertTrue(os.path.exists(os.path.join(CERT_PATH, f'{peer_name}.pem')))
self.tearDownPKI()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
index f7e14aa16..239e44c3b 100755
--- a/src/conf_mode/pki.py
+++ b/src/conf_mode/pki.py
@@ -1,305 +1,433 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2021 VyOS maintainers and contributors
+# 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
+
+from sys import argv
from sys import exit
from vyos.config import Config
-from vyos.configdep import set_dependents, call_dependents
+from vyos.config import config_dict_merge
+from vyos.configdep import set_dependents
+from vyos.configdep import call_dependents
from vyos.configdict import node_changed
+from vyos.configdiff import Diff
+from vyos.defaults import directories
from vyos.pki import is_ca_certificate
from vyos.pki import load_certificate
from vyos.pki import load_public_key
from vyos.pki import load_private_key
from vyos.pki import load_crl
from vyos.pki import load_dh_parameters
+from vyos.utils.boot import boot_configuration_complete
+from vyos.utils.dict import dict_search
from vyos.utils.dict import dict_search_args
from vyos.utils.dict import dict_search_recursive
+from vyos.utils.process import call
+from vyos.utils.process import cmd
+from vyos.utils.process import is_systemd_service_active
from vyos import ConfigError
from vyos import airbag
airbag.enable()
-# keys to recursively search for under specified path, script to call if update required
+vyos_certbot_dir = directories['certbot']
+
+# keys to recursively search for under specified path
sync_search = [
{
'keys': ['certificate'],
'path': ['service', 'https'],
- 'script': '/usr/libexec/vyos/conf_mode/service_https.py'
},
{
'keys': ['certificate', 'ca_certificate'],
'path': ['interfaces', 'ethernet'],
- 'script': '/usr/libexec/vyos/conf_mode/interfaces_ethernet.py'
},
{
'keys': ['certificate', 'ca_certificate', 'dh_params', 'shared_secret_key', 'auth_key', 'crypt_key'],
'path': ['interfaces', 'openvpn'],
- 'script': '/usr/libexec/vyos/conf_mode/interfaces_openvpn.py'
},
{
'keys': ['ca_certificate'],
'path': ['interfaces', 'sstpc'],
- 'script': '/usr/libexec/vyos/conf_mode/interfaces_sstpc.py'
},
{
'keys': ['certificate', 'ca_certificate', 'local_key', 'remote_key'],
'path': ['vpn', 'ipsec'],
- 'script': '/usr/libexec/vyos/conf_mode/vpn_ipsec.py'
},
{
'keys': ['certificate', 'ca_certificate'],
'path': ['vpn', 'openconnect'],
- 'script': '/usr/libexec/vyos/conf_mode/vpn_openconnect.py'
},
{
'keys': ['certificate', 'ca_certificate'],
'path': ['vpn', 'sstp'],
- 'script': '/usr/libexec/vyos/conf_mode/vpn_sstp.py'
}
]
# key from other config nodes -> key in pki['changed'] and pki
sync_translate = {
'certificate': 'certificate',
'ca_certificate': 'ca',
'dh_params': 'dh',
'local_key': 'key_pair',
'remote_key': 'key_pair',
'shared_secret_key': 'openvpn',
'auth_key': 'openvpn',
'crypt_key': 'openvpn'
}
+def certbot_delete(certificate):
+ if not boot_configuration_complete():
+ return
+ if os.path.exists(f'{vyos_certbot_dir}/renewal/{certificate}.conf'):
+ cmd(f'certbot delete --non-interactive --config-dir {vyos_certbot_dir} --cert-name {certificate}')
+
+def certbot_request(name: str, config: dict, dry_run: bool=True):
+ # We do not call certbot when booting the system - there is no need to do so and
+ # request new certificates during boot/image upgrade as the certbot configuration
+ # is stored persistent under /config - thus we do not open the door to transient
+ # errors
+ if not boot_configuration_complete():
+ return
+
+ domains = '--domains ' + ' --domains '.join(config['domain_name'])
+ tmp = f'certbot certonly --non-interactive --config-dir {vyos_certbot_dir} --cert-name {name} '\
+ f'--standalone --agree-tos --no-eff-email --expand --server {config["url"]} '\
+ f'--email {config["email"]} --key-type rsa --rsa-key-size {config["rsa_key_size"]} '\
+ f'{domains}'
+ if 'listen_address' in config:
+ tmp += f' --http-01-address {config["listen_address"]}'
+ # verify() does not need to actually request a cert but only test for plausability
+ if dry_run:
+ tmp += ' --dry-run'
+
+ cmd(tmp, raising=ConfigError, message=f'ACME certbot request failed for "{name}"!')
+
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['pki']
pki = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
- pki['changed'] = {}
+ if len(argv) > 1 and argv[1] == 'certbot_renew':
+ pki['certbot_renew'] = {}
+
tmp = node_changed(conf, base + ['ca'], key_mangling=('-', '_'), recursive=True)
- if tmp: pki['changed'].update({'ca' : tmp})
+ if tmp:
+ if 'changed' not in pki: pki.update({'changed':{}})
+ pki['changed'].update({'ca' : tmp})
tmp = node_changed(conf, base + ['certificate'], key_mangling=('-', '_'), recursive=True)
- if tmp: pki['changed'].update({'certificate' : tmp})
+ if tmp:
+ if 'changed' not in pki: pki.update({'changed':{}})
+ pki['changed'].update({'certificate' : tmp})
tmp = node_changed(conf, base + ['dh'], key_mangling=('-', '_'), recursive=True)
- if tmp: pki['changed'].update({'dh' : tmp})
+ if tmp:
+ if 'changed' not in pki: pki.update({'changed':{}})
+ pki['changed'].update({'dh' : tmp})
tmp = node_changed(conf, base + ['key-pair'], key_mangling=('-', '_'), recursive=True)
- if tmp: pki['changed'].update({'key_pair' : tmp})
+ if tmp:
+ if 'changed' not in pki: pki.update({'changed':{}})
+ pki['changed'].update({'key_pair' : tmp})
- tmp = node_changed(conf, base + ['openvpn', 'shared-secret'], key_mangling=('-', '_'), recursive=True)
- if tmp: pki['changed'].update({'openvpn' : tmp})
+ tmp = node_changed(conf, base + ['openvpn', 'shared-secret'], key_mangling=('-', '_'),
+ recursive=True)
+ if tmp:
+ if 'changed' not in pki: pki.update({'changed':{}})
+ pki['changed'].update({'openvpn' : tmp})
# We only merge on the defaults of there is a configuration at all
if conf.exists(base):
- pki = conf.merge_defaults(pki, recursive=True)
+ # We have gathered the dict representation of the CLI, but there are default
+ # options which we need to update into the dictionary retrived.
+ default_values = conf.get_config_defaults(**pki.kwargs, recursive=True)
+ # remove ACME default configuration if unused by CLI
+ if 'certificate' in pki:
+ for name, cert_config in pki['certificate'].items():
+ if 'acme' not in cert_config:
+ # Remove ACME default values
+ del default_values['certificate'][name]['acme']
+
+ # merge CLI and default dictionary
+ pki = config_dict_merge(default_values, pki)
+
+ # Certbot triggered an external renew of the certificates.
+ # Mark all ACME based certificates as "changed" to trigger
+ # update of dependent services
+ if 'certificate' in pki and 'certbot_renew' in pki:
+ renew = []
+ for name, cert_config in pki['certificate'].items():
+ if 'acme' in cert_config:
+ renew.append(name)
+ # If triggered externally by certbot, certificate key is not present in changed
+ if 'changed' not in pki: pki.update({'changed':{}})
+ pki['changed'].update({'certificate' : renew})
# We need to get the entire system configuration to verify that we are not
# deleting a certificate that is still referenced somewhere!
pki['system'] = conf.get_config_dict([], key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
- if 'changed' in pki:
- for search in sync_search:
- for key in search['keys']:
- changed_key = sync_translate[key]
-
- if changed_key not in pki['changed']:
- continue
-
- for item_name in pki['changed'][changed_key]:
- node_present = False
- if changed_key == 'openvpn':
- node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
- else:
- node_present = dict_search_args(pki, changed_key, item_name)
-
- if node_present:
- search_dict = dict_search_args(pki['system'], *search['path'])
-
- if not search_dict:
- continue
-
- for found_name, found_path in dict_search_recursive(search_dict, key):
- if found_name == item_name:
- path = search['path']
- path_str = ' '.join(path + found_path)
- print(f'pki: Updating config: {path_str} {found_name}')
-
- if path[0] == 'interfaces':
- ifname = found_path[0]
- set_dependents(path[1], conf, ifname)
- else:
- set_dependents(path[1], conf)
+ for search in sync_search:
+ for key in search['keys']:
+ changed_key = sync_translate[key]
+ if 'changed' not in pki or changed_key not in pki['changed']:
+ continue
+
+ for item_name in pki['changed'][changed_key]:
+ node_present = False
+ if changed_key == 'openvpn':
+ node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
+ else:
+ node_present = dict_search_args(pki, changed_key, item_name)
+
+ if node_present:
+ search_dict = dict_search_args(pki['system'], *search['path'])
+ if not search_dict:
+ continue
+ for found_name, found_path in dict_search_recursive(search_dict, key):
+ if found_name == item_name:
+ path = search['path']
+ path_str = ' '.join(path + found_path)
+ print(f'PKI: Updating config: {path_str} {found_name}')
+
+ if path[0] == 'interfaces':
+ ifname = found_path[0]
+ set_dependents(path[1], conf, ifname)
+ else:
+ set_dependents(path[1], conf)
return pki
def is_valid_certificate(raw_data):
# If it loads correctly we're good, or return False
return load_certificate(raw_data, wrap_tags=True)
def is_valid_ca_certificate(raw_data):
# Check if this is a valid certificate with CA attributes
cert = load_certificate(raw_data, wrap_tags=True)
if not cert:
return False
return is_ca_certificate(cert)
def is_valid_public_key(raw_data):
# If it loads correctly we're good, or return False
return load_public_key(raw_data, wrap_tags=True)
def is_valid_private_key(raw_data, protected=False):
# If it loads correctly we're good, or return False
# With encrypted private keys, we always return true as we cannot ask for password to verify
if protected:
return True
return load_private_key(raw_data, passphrase=None, wrap_tags=True)
def is_valid_crl(raw_data):
# If it loads correctly we're good, or return False
return load_crl(raw_data, wrap_tags=True)
def is_valid_dh_parameters(raw_data):
# If it loads correctly we're good, or return False
return load_dh_parameters(raw_data, wrap_tags=True)
def verify(pki):
if not pki:
return None
if 'ca' in pki:
for name, ca_conf in pki['ca'].items():
if 'certificate' in ca_conf:
if not is_valid_ca_certificate(ca_conf['certificate']):
raise ConfigError(f'Invalid certificate on CA certificate "{name}"')
if 'private' in ca_conf and 'key' in ca_conf['private']:
private = ca_conf['private']
protected = 'password_protected' in private
if not is_valid_private_key(private['key'], protected):
raise ConfigError(f'Invalid private key on CA certificate "{name}"')
if 'crl' in ca_conf:
ca_crls = ca_conf['crl']
if isinstance(ca_crls, str):
ca_crls = [ca_crls]
for crl in ca_crls:
if not is_valid_crl(crl):
raise ConfigError(f'Invalid CRL on CA certificate "{name}"')
if 'certificate' in pki:
for name, cert_conf in pki['certificate'].items():
if 'certificate' in cert_conf:
if not is_valid_certificate(cert_conf['certificate']):
raise ConfigError(f'Invalid certificate on certificate "{name}"')
if 'private' in cert_conf and 'key' in cert_conf['private']:
private = cert_conf['private']
protected = 'password_protected' in private
if not is_valid_private_key(private['key'], protected):
raise ConfigError(f'Invalid private key on certificate "{name}"')
+ if 'acme' in cert_conf:
+ if 'domain_name' not in cert_conf['acme']:
+ raise ConfigError(f'At least one domain-name is required to request '\
+ f'certificate for "{name}" via ACME!')
+
+ if 'email' not in cert_conf['acme']:
+ raise ConfigError(f'An email address is required to request '\
+ f'certificate for "{name}" via ACME!')
+
+ if 'certbot_renew' not in pki:
+ # Only run the ACME command if something on this entity changed,
+ # as this is time intensive
+ tmp = dict_search('changed.certificate', pki)
+ if tmp != None and name in tmp:
+ certbot_request(name, cert_conf['acme'])
+
if 'dh' in pki:
for name, dh_conf in pki['dh'].items():
if 'parameters' in dh_conf:
if not is_valid_dh_parameters(dh_conf['parameters']):
raise ConfigError(f'Invalid DH parameters on "{name}"')
if 'key_pair' in pki:
for name, key_conf in pki['key_pair'].items():
if 'public' in key_conf and 'key' in key_conf['public']:
if not is_valid_public_key(key_conf['public']['key']):
raise ConfigError(f'Invalid public key on key-pair "{name}"')
if 'private' in key_conf and 'key' in key_conf['private']:
private = key_conf['private']
protected = 'password_protected' in private
if not is_valid_private_key(private['key'], protected):
raise ConfigError(f'Invalid private key on key-pair "{name}"')
if 'x509' in pki:
if 'default' in pki['x509']:
default_values = pki['x509']['default']
if 'country' in default_values:
country = default_values['country']
if len(country) != 2 or not country.isalpha():
raise ConfigError(f'Invalid default country value. Value must be 2 alpha characters.')
if 'changed' in pki:
# if the list is getting longer, we can move to a dict() and also embed the
# search key as value from line 173 or 176
for search in sync_search:
for key in search['keys']:
changed_key = sync_translate[key]
if changed_key not in pki['changed']:
continue
for item_name in pki['changed'][changed_key]:
node_present = False
if changed_key == 'openvpn':
node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
else:
node_present = dict_search_args(pki, changed_key, item_name)
if not node_present:
search_dict = dict_search_args(pki['system'], *search['path'])
if not search_dict:
continue
for found_name, found_path in dict_search_recursive(search_dict, key):
if found_name == item_name:
path_str = " ".join(search['path'] + found_path)
raise ConfigError(f'PKI object "{item_name}" still in use by "{path_str}"')
return None
def generate(pki):
if not pki:
return None
+ # Certbot renewal only needs to re-trigger the services to load up the
+ # new PEM file
+ if 'certbot_renew' in pki:
+ return None
+
+ certbot_list = []
+ certbot_list_on_disk = []
+ if os.path.exists(f'{vyos_certbot_dir}/live'):
+ certbot_list_on_disk = [f.path.split('/')[-1] for f in os.scandir(f'{vyos_certbot_dir}/live') if f.is_dir()]
+
+ if 'certificate' in pki:
+ changed_certificates = dict_search('changed.certificate', pki)
+ for name, cert_conf in pki['certificate'].items():
+ if 'acme' in cert_conf:
+ certbot_list.append(name)
+ # generate certificate if not found on disk
+ if name not in certbot_list_on_disk:
+ certbot_request(name, cert_conf['acme'], dry_run=False)
+ elif changed_certificates != None and name in changed_certificates:
+ # when something for the certificate changed, we should delete it
+ if name in certbot_list_on_disk:
+ certbot_delete(name)
+ certbot_request(name, cert_conf['acme'], dry_run=False)
+
+ # Cleanup certbot configuration and certificates if no longer in use by CLI
+ # Get foldernames under vyos_certbot_dir which each represent a certbot cert
+ if os.path.exists(f'{vyos_certbot_dir}/live'):
+ for cert in certbot_list_on_disk:
+ if cert not in certbot_list:
+ # certificate is no longer active on the CLI - remove it
+ certbot_delete(cert)
+
return None
def apply(pki):
+ systemd_certbot_name = 'certbot.timer'
if not pki:
+ call(f'systemctl stop {systemd_certbot_name}')
return None
+ has_certbot = False
+ if 'certificate' in pki:
+ for name, cert_conf in pki['certificate'].items():
+ if 'acme' in cert_conf:
+ has_certbot = True
+ break
+
+ if not has_certbot:
+ call(f'systemctl stop {systemd_certbot_name}')
+ elif has_certbot and not is_systemd_service_active(systemd_certbot_name):
+ call(f'systemctl restart {systemd_certbot_name}')
+
if 'changed' in pki:
call_dependents()
return None
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)
diff --git a/src/conf_mode/service_https.py b/src/conf_mode/service_https.py
index cb40acc9f..2e7ebda5a 100755
--- a/src/conf_mode/service_https.py
+++ b/src/conf_mode/service_https.py
@@ -1,330 +1,289 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-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 sys
import json
from copy import deepcopy
from time import sleep
import vyos.defaults
-import vyos.certbot_util
from vyos.base import Warning
from vyos.config import Config
from vyos.configdiff import get_config_diff
from vyos.configverify import verify_vrf
from vyos import ConfigError
from vyos.pki import wrap_certificate
from vyos.pki import wrap_private_key
from vyos.template import render
from vyos.utils.process import call
-from vyos.utils.process import is_systemd_service_running
-from vyos.utils.process import is_systemd_service_active
from vyos.utils.network import check_port_availability
from vyos.utils.network import is_listen_port_bind_service
from vyos.utils.file import write_file
from vyos import airbag
airbag.enable()
config_file = '/etc/nginx/sites-available/default'
systemd_override = r'/run/systemd/system/nginx.service.d/override.conf'
cert_dir = '/etc/ssl/certs'
key_dir = '/etc/ssl/private'
-certbot_dir = vyos.defaults.directories['certbot']
api_config_state = '/run/http-api-state'
systemd_service = '/run/systemd/system/vyos-http-api.service'
-# https config needs to coordinate several subsystems: api, certbot,
+# https config needs to coordinate several subsystems: api,
# self-signed certificate, as well as the virtual hosts defined within the
# https config definition itself. Consequently, one needs a general dict,
# encompassing the https and other configs, and a list of such virtual hosts
# (server blocks in nginx terminology) to pass to the jinja2 template.
default_server_block = {
'id' : '',
'address' : '*',
'port' : '443',
'name' : ['_'],
'api' : False,
'vyos_cert' : {},
- 'certbot' : False
}
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['service', 'https']
if not conf.exists(base):
return None
diff = get_config_diff(conf)
https = conf.get_config_dict(base, get_first_key=True, with_pki=True)
- https['children_changed'] = diff.node_changed_children(base)
https['api_add_or_delete'] = diff.node_changed_presence(base + ['api'])
if 'api' not in https:
return https
http_api = conf.get_config_dict(base + ['api'], key_mangling=('-', '_'),
no_tag_node_value_mangle=True,
get_first_key=True,
with_recursive_defaults=True)
if http_api.from_defaults(['graphql']):
del http_api['graphql']
# Do we run inside a VRF context?
vrf_path = ['service', 'https', 'vrf']
if conf.exists(vrf_path):
http_api['vrf'] = conf.return_value(vrf_path)
https['api'] = http_api
-
return https
def verify(https):
from vyos.utils.dict import dict_search
if https is None:
return None
if 'certificates' in https:
certificates = https['certificates']
if 'certificate' in certificates:
if not https['pki']:
raise ConfigError('PKI is not configured')
cert_name = certificates['certificate']
if cert_name not in https['pki']['certificate']:
- raise ConfigError("Invalid certificate on https configuration")
+ raise ConfigError('Invalid certificate on https configuration')
pki_cert = https['pki']['certificate'][cert_name]
if 'certificate' not in pki_cert:
- raise ConfigError("Missing certificate on https configuration")
+ raise ConfigError('Missing certificate on https configuration')
if 'private' not in pki_cert or 'key' not in pki_cert['private']:
raise ConfigError("Missing certificate private key on https configuration")
-
- if 'certbot' in https['certificates']:
- vhost_names = []
- for _, vh_conf in https.get('virtual-host', {}).items():
- vhost_names += vh_conf.get('server-name', [])
- domains = https['certificates']['certbot'].get('domain-name', [])
- domains_found = [domain for domain in domains if domain in vhost_names]
- if not domains_found:
- raise ConfigError("At least one 'virtual-host <id> server-name' "
- "matching the 'certbot domain-name' is required.")
+ else:
+ Warning('No certificate specified, using buildin self-signed certificates!')
server_block_list = []
# organize by vhosts
vhost_dict = https.get('virtual-host', {})
if not vhost_dict:
# no specified virtual hosts (server blocks); use default
server_block_list.append(default_server_block)
else:
for vhost in list(vhost_dict):
server_block = deepcopy(default_server_block)
data = vhost_dict.get(vhost, {})
server_block['address'] = data.get('listen-address', '*')
server_block['port'] = data.get('port', '443')
server_block_list.append(server_block)
for entry in server_block_list:
_address = entry.get('address')
_address = '0.0.0.0' if _address == '*' else _address
_port = entry.get('port')
proto = 'tcp'
if check_port_availability(_address, int(_port), proto) is not True and \
not is_listen_port_bind_service(int(_port), 'nginx'):
raise ConfigError(f'"{proto}" port "{_port}" is used by another service')
verify_vrf(https)
# Verify API server settings, if present
if 'api' in https:
keys = dict_search('api.keys.id', https)
gql_auth_type = dict_search('api.graphql.authentication.type', https)
# If "api graphql" is not defined and `gql_auth_type` is None,
# there's certainly no JWT auth option, and keys are required
jwt_auth = (gql_auth_type == "token")
# Check for incomplete key configurations in every case
valid_keys_exist = False
if keys:
for k in keys:
if 'key' not in keys[k]:
raise ConfigError(f'Missing HTTPS API key string for key id "{k}"')
else:
valid_keys_exist = True
# If only key-based methods are enabled,
# fail the commit if no valid key configurations are found
if (not valid_keys_exist) and (not jwt_auth):
raise ConfigError('At least one HTTPS API key is required unless GraphQL token authentication is enabled')
if (not valid_keys_exist) and jwt_auth:
Warning(f'API keys are not configured: the classic (non-GraphQL) API will be unavailable.')
return None
def generate(https):
if https is None:
return None
if 'api' not in https:
if os.path.exists(systemd_service):
os.unlink(systemd_service)
else:
render(systemd_service, 'https/vyos-http-api.service.j2', https['api'])
with open(api_config_state, 'w') as f:
json.dump(https['api'], f, indent=2)
server_block_list = []
# organize by vhosts
vhost_dict = https.get('virtual-host', {})
if not vhost_dict:
# no specified virtual hosts (server blocks); use default
server_block_list.append(default_server_block)
else:
for vhost in list(vhost_dict):
server_block = deepcopy(default_server_block)
server_block['id'] = vhost
data = vhost_dict.get(vhost, {})
server_block['address'] = data.get('listen-address', '*')
server_block['port'] = data.get('port', '443')
name = data.get('server-name', ['_'])
server_block['name'] = name
allow_client = data.get('allow-client', {})
server_block['allow_client'] = allow_client.get('address', [])
server_block_list.append(server_block)
# get certificate data
cert_dict = https.get('certificates', {})
if 'certificate' in cert_dict:
cert_name = cert_dict['certificate']
pki_cert = https['pki']['certificate'][cert_name]
cert_path = os.path.join(cert_dir, f'{cert_name}.pem')
key_path = os.path.join(key_dir, f'{cert_name}.pem')
server_cert = str(wrap_certificate(pki_cert['certificate']))
if 'ca-certificate' in cert_dict:
ca_cert = cert_dict['ca-certificate']
server_cert += '\n' + str(wrap_certificate(https['pki']['ca'][ca_cert]['certificate']))
write_file(cert_path, server_cert)
write_file(key_path, wrap_private_key(pki_cert['private']['key']))
vyos_cert_data = {
'crt': cert_path,
'key': key_path
}
for block in server_block_list:
block['vyos_cert'] = vyos_cert_data
- # letsencrypt certificate using certbot
-
- certbot = False
- cert_domains = cert_dict.get('certbot', {}).get('domain-name', [])
- if cert_domains:
- certbot = True
- for domain in cert_domains:
- sub_list = vyos.certbot_util.choose_server_block(server_block_list,
- domain)
- if sub_list:
- for sb in sub_list:
- sb['certbot'] = True
- sb['certbot_dir'] = certbot_dir
- # certbot organizes certificates by first domain
- sb['certbot_domain_dir'] = cert_domains[0]
-
if 'api' in list(https):
vhost_list = https.get('api-restrict', {}).get('virtual-host', [])
if not vhost_list:
for block in server_block_list:
block['api'] = True
else:
for block in server_block_list:
if block['id'] in vhost_list:
block['api'] = True
data = {
'server_block_list': server_block_list,
- 'certbot': certbot
}
render(config_file, 'https/nginx.default.j2', data)
render(systemd_override, 'https/override.conf.j2', https)
return None
def apply(https):
# Reload systemd manager configuration
call('systemctl daemon-reload')
http_api_service_name = 'vyos-http-api.service'
https_service_name = 'nginx.service'
if https is None:
- if is_systemd_service_active(f'{http_api_service_name}'):
- call(f'systemctl stop {http_api_service_name}')
+ call(f'systemctl stop {http_api_service_name}')
call(f'systemctl stop {https_service_name}')
return
- if 'api' in https['children_changed']:
- if 'api' in https:
- if is_systemd_service_running(f'{http_api_service_name}'):
- call(f'systemctl reload {http_api_service_name}')
- else:
- call(f'systemctl restart {http_api_service_name}')
- # Let uvicorn settle before (possibly) restarting nginx
- sleep(1)
- else:
- if is_systemd_service_active(f'{http_api_service_name}'):
- call(f'systemctl stop {http_api_service_name}')
+ if 'api' in https:
+ call(f'systemctl reload-or-restart {http_api_service_name}')
+ # Let uvicorn settle before (possibly) restarting nginx
+ sleep(1)
+ else:
+ call(f'systemctl stop {http_api_service_name}')
- if (not is_systemd_service_running(f'{https_service_name}') or
- https['api_add_or_delete'] or
- set(https['children_changed']) - set(['api'])):
- call(f'systemctl restart {https_service_name}')
+ call(f'systemctl reload-or-restart {https_service_name}')
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
sys.exit(1)
diff --git a/src/conf_mode/service_https_certificates_certbot.py b/src/conf_mode/service_https_certificates_certbot.py
deleted file mode 100755
index 1a6a498de..000000000
--- a/src/conf_mode/service_https_certificates_certbot.py
+++ /dev/null
@@ -1,114 +0,0 @@
-#!/usr/bin/env python3
-#
-# Copyright (C) 2019-2020 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 sys
-import os
-
-import vyos.defaults
-from vyos.config import Config
-from vyos import ConfigError
-from vyos.utils.process import cmd
-from vyos.utils.process import call
-from vyos.utils.process import is_systemd_service_running
-
-from vyos import airbag
-airbag.enable()
-
-vyos_conf_scripts_dir = vyos.defaults.directories['conf_mode']
-vyos_certbot_dir = vyos.defaults.directories['certbot']
-
-dependencies = [
- 'service_https.py',
-]
-
-def request_certbot(cert):
- email = cert.get('email')
- if email is not None:
- email_flag = '-m {0}'.format(email)
- else:
- email_flag = ''
-
- domains = cert.get('domains')
- if domains is not None:
- domain_flag = '-d ' + ' -d '.join(domains)
- else:
- domain_flag = ''
-
- certbot_cmd = f'certbot certonly --config-dir {vyos_certbot_dir} -n --nginx --agree-tos --no-eff-email --expand {email_flag} {domain_flag}'
-
- cmd(certbot_cmd,
- raising=ConfigError,
- message="The certbot request failed for the specified domains.")
-
-def get_config():
- conf = Config()
- if not conf.exists('service https certificates certbot'):
- return None
- else:
- conf.set_level('service https certificates certbot')
-
- cert = {}
-
- if conf.exists('domain-name'):
- cert['domains'] = conf.return_values('domain-name')
-
- if conf.exists('email'):
- cert['email'] = conf.return_value('email')
-
- return cert
-
-def verify(cert):
- if cert is None:
- return None
-
- if 'domains' not in cert:
- raise ConfigError("At least one domain name is required to"
- " request a letsencrypt certificate.")
-
- if 'email' not in cert:
- raise ConfigError("An email address is required to request"
- " a letsencrypt certificate.")
-
-def generate(cert):
- if cert is None:
- return None
-
- # certbot will attempt to reload nginx, even with 'certonly';
- # start nginx if not active
- if not is_systemd_service_running('nginx.service'):
- call('systemctl start nginx.service')
-
- request_certbot(cert)
-
-def apply(cert):
- if cert is not None:
- call('systemctl restart certbot.timer')
- else:
- call('systemctl stop certbot.timer')
- return None
-
- for dep in dependencies:
- cmd(f'{vyos_conf_scripts_dir}/{dep}', raising=ConfigError)
-
-if __name__ == '__main__':
- try:
- c = get_config()
- verify(c)
- generate(c)
- apply(c)
- except ConfigError as e:
- print(e)
- sys.exit(1)
diff --git a/src/conf_mode/vpn_ipsec.py b/src/conf_mode/vpn_ipsec.py
index 7fd32c230..5bdcf2fa1 100755
--- a/src/conf_mode/vpn_ipsec.py
+++ b/src/conf_mode/vpn_ipsec.py
@@ -1,594 +1,595 @@
#!/usr/bin/env python3
#
# 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 ipaddress
import os
import re
import jmespath
from sys import exit
from time import sleep
from time import time
from vyos.base import Warning
from vyos.config import Config
from vyos.configdict import leaf_node_changed
from vyos.configverify import verify_interface_exists
from vyos.defaults import directories
from vyos.ifconfig import Interface
from vyos.pki import encode_certificate
from vyos.pki import encode_public_key
from vyos.pki import find_chain
from vyos.pki import load_certificate
from vyos.pki import load_private_key
from vyos.pki import wrap_certificate
from vyos.pki import wrap_crl
from vyos.pki import wrap_public_key
from vyos.pki import wrap_private_key
from vyos.template import ip_from_cidr
from vyos.template import is_ipv4
from vyos.template import is_ipv6
from vyos.template import render
from vyos.utils.network import is_ipv6_link_local
+from vyos.utils.network import interface_exists
from vyos.utils.dict import dict_search
from vyos.utils.dict import dict_search_args
from vyos.utils.process import call
from vyos.utils.process import run
from vyos import ConfigError
from vyos import airbag
airbag.enable()
dhcp_wait_attempts = 2
dhcp_wait_sleep = 1
swanctl_dir = '/etc/swanctl'
charon_conf = '/etc/strongswan.d/charon.conf'
charon_dhcp_conf = '/etc/strongswan.d/charon/dhcp.conf'
charon_radius_conf = '/etc/strongswan.d/charon/eap-radius.conf'
interface_conf = '/etc/strongswan.d/interfaces_use.conf'
swanctl_conf = f'{swanctl_dir}/swanctl.conf'
default_install_routes = 'yes'
vici_socket = '/var/run/charon.vici'
-CERT_PATH = f'{swanctl_dir}/x509/'
+CERT_PATH = f'{swanctl_dir}/x509/'
PUBKEY_PATH = f'{swanctl_dir}/pubkey/'
-KEY_PATH = f'{swanctl_dir}/private/'
-CA_PATH = f'{swanctl_dir}/x509ca/'
-CRL_PATH = f'{swanctl_dir}/x509crl/'
+KEY_PATH = f'{swanctl_dir}/private/'
+CA_PATH = f'{swanctl_dir}/x509ca/'
+CRL_PATH = f'{swanctl_dir}/x509crl/'
DHCP_HOOK_IFLIST = '/tmp/ipsec_dhcp_waiting'
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['vpn', 'ipsec']
l2tp_base = ['vpn', 'l2tp', 'remote-access', 'ipsec-settings']
if not conf.exists(base):
return None
# retrieve common dictionary keys
ipsec = conf.get_config_dict(base, key_mangling=('-', '_'),
no_tag_node_value_mangle=True,
get_first_key=True,
with_recursive_defaults=True,
with_pki=True)
ipsec['dhcp_no_address'] = {}
ipsec['install_routes'] = 'no' if conf.exists(base + ["options", "disable-route-autoinstall"]) else default_install_routes
ipsec['interface_change'] = leaf_node_changed(conf, base + ['interface'])
ipsec['nhrp_exists'] = conf.exists(['protocols', 'nhrp', 'tunnel'])
tmp = conf.get_config_dict(l2tp_base, key_mangling=('-', '_'),
no_tag_node_value_mangle=True,
get_first_key=True)
if tmp:
ipsec['l2tp'] = conf.merge_defaults(tmp, recursive=True)
ipsec['l2tp_outside_address'] = conf.return_value(['vpn', 'l2tp', 'remote-access', 'outside-address'])
ipsec['l2tp_ike_default'] = 'aes256-sha1-modp1024,3des-sha1-modp1024'
ipsec['l2tp_esp_default'] = 'aes256-sha1,3des-sha1'
return ipsec
def get_dhcp_address(iface):
addresses = Interface(iface).get_addr()
if not addresses:
return None
for address in addresses:
if not is_ipv6_link_local(address):
return ip_from_cidr(address)
return None
def verify_pki_x509(pki, x509_conf):
if not pki or 'ca' not in pki or 'certificate' not in pki:
raise ConfigError(f'PKI is not configured')
ca_cert_name = x509_conf['ca_certificate']
cert_name = x509_conf['certificate']
if not dict_search_args(pki, 'ca', ca_cert_name, 'certificate'):
raise ConfigError(f'Missing CA certificate on specified PKI CA certificate "{ca_cert_name}"')
if not dict_search_args(pki, 'certificate', cert_name, 'certificate'):
raise ConfigError(f'Missing certificate on specified PKI certificate "{cert_name}"')
if not dict_search_args(pki, 'certificate', cert_name, 'private', 'key'):
raise ConfigError(f'Missing private key on specified PKI certificate "{cert_name}"')
return True
def verify_pki_rsa(pki, rsa_conf):
if not pki or 'key_pair' not in pki:
raise ConfigError(f'PKI is not configured')
local_key = rsa_conf['local_key']
remote_key = rsa_conf['remote_key']
if not dict_search_args(pki, 'key_pair', local_key, 'private', 'key'):
raise ConfigError(f'Missing private key on specified local-key "{local_key}"')
if not dict_search_args(pki, 'key_pair', remote_key, 'public', 'key'):
raise ConfigError(f'Missing public key on specified remote-key "{remote_key}"')
return True
def verify(ipsec):
if not ipsec:
return None
if 'authentication' in ipsec:
if 'psk' in ipsec['authentication']:
for psk, psk_config in ipsec['authentication']['psk'].items():
if 'id' not in psk_config or 'secret' not in psk_config:
raise ConfigError(f'Authentication psk "{psk}" missing "id" or "secret"')
if 'interfaces' in ipsec :
for ifname in ipsec['interface']:
verify_interface_exists(ifname)
if 'l2tp' in ipsec:
if 'esp_group' in ipsec['l2tp']:
if 'esp_group' not in ipsec or ipsec['l2tp']['esp_group'] not in ipsec['esp_group']:
raise ConfigError(f"Invalid esp-group on L2TP remote-access config")
if 'ike_group' in ipsec['l2tp']:
if 'ike_group' not in ipsec or ipsec['l2tp']['ike_group'] not in ipsec['ike_group']:
raise ConfigError(f"Invalid ike-group on L2TP remote-access config")
if 'authentication' not in ipsec['l2tp']:
raise ConfigError(f'Missing authentication settings on L2TP remote-access config')
if 'mode' not in ipsec['l2tp']['authentication']:
raise ConfigError(f'Missing authentication mode on L2TP remote-access config')
if not ipsec['l2tp_outside_address']:
raise ConfigError(f'Missing outside-address on L2TP remote-access config')
if ipsec['l2tp']['authentication']['mode'] == 'pre-shared-secret':
if 'pre_shared_secret' not in ipsec['l2tp']['authentication']:
raise ConfigError(f'Missing pre shared secret on L2TP remote-access config')
if ipsec['l2tp']['authentication']['mode'] == 'x509':
if 'x509' not in ipsec['l2tp']['authentication']:
raise ConfigError(f'Missing x509 settings on L2TP remote-access config')
x509 = ipsec['l2tp']['authentication']['x509']
if 'ca_certificate' not in x509 or 'certificate' not in x509:
raise ConfigError(f'Missing x509 certificates on L2TP remote-access config')
verify_pki_x509(ipsec['pki'], x509)
if 'profile' in ipsec:
for profile, profile_conf in ipsec['profile'].items():
if 'esp_group' in profile_conf:
if 'esp_group' not in ipsec or profile_conf['esp_group'] not in ipsec['esp_group']:
raise ConfigError(f"Invalid esp-group on {profile} profile")
else:
raise ConfigError(f"Missing esp-group on {profile} profile")
if 'ike_group' in profile_conf:
if 'ike_group' not in ipsec or profile_conf['ike_group'] not in ipsec['ike_group']:
raise ConfigError(f"Invalid ike-group on {profile} profile")
else:
raise ConfigError(f"Missing ike-group on {profile} profile")
if 'authentication' not in profile_conf:
raise ConfigError(f"Missing authentication on {profile} profile")
if 'remote_access' in ipsec:
if 'connection' in ipsec['remote_access']:
for name, ra_conf in ipsec['remote_access']['connection'].items():
if 'esp_group' in ra_conf:
if 'esp_group' not in ipsec or ra_conf['esp_group'] not in ipsec['esp_group']:
raise ConfigError(f"Invalid esp-group on {name} remote-access config")
else:
raise ConfigError(f"Missing esp-group on {name} remote-access config")
if 'ike_group' in ra_conf:
if 'ike_group' not in ipsec or ra_conf['ike_group'] not in ipsec['ike_group']:
raise ConfigError(f"Invalid ike-group on {name} remote-access config")
ike = ra_conf['ike_group']
if dict_search(f'ike_group.{ike}.key_exchange', ipsec) != 'ikev2':
raise ConfigError('IPsec remote-access connections requires IKEv2!')
else:
raise ConfigError(f"Missing ike-group on {name} remote-access config")
if 'authentication' not in ra_conf:
raise ConfigError(f"Missing authentication on {name} remote-access config")
if ra_conf['authentication']['server_mode'] == 'x509':
if 'x509' not in ra_conf['authentication']:
raise ConfigError(f"Missing x509 settings on {name} remote-access config")
x509 = ra_conf['authentication']['x509']
if 'ca_certificate' not in x509 or 'certificate' not in x509:
raise ConfigError(f"Missing x509 certificates on {name} remote-access config")
verify_pki_x509(ipsec['pki'], x509)
elif ra_conf['authentication']['server_mode'] == 'pre-shared-secret':
if 'pre_shared_secret' not in ra_conf['authentication']:
raise ConfigError(f"Missing pre-shared-key on {name} remote-access config")
if 'client_mode' not in ra_conf['authentication']:
raise ConfigError('Client authentication method is required!')
if dict_search('authentication.client_mode', ra_conf) == 'eap-radius':
if dict_search('remote_access.radius.server', ipsec) == None:
raise ConfigError('RADIUS authentication requires at least one server')
if 'pool' in ra_conf:
if {'dhcp', 'radius'} <= set(ra_conf['pool']):
raise ConfigError(f'Can not use both DHCP and RADIUS for address allocation '\
f'at the same time for "{name}"!')
if 'dhcp' in ra_conf['pool'] and len(ra_conf['pool']) > 1:
raise ConfigError(f'Can not use DHCP and a predefined address pool for "{name}"!')
if 'radius' in ra_conf['pool'] and len(ra_conf['pool']) > 1:
raise ConfigError(f'Can not use RADIUS and a predefined address pool for "{name}"!')
for pool in ra_conf['pool']:
if pool == 'dhcp':
if dict_search('remote_access.dhcp.server', ipsec) == None:
raise ConfigError('IPsec DHCP server is not configured!')
elif pool == 'radius':
if dict_search('remote_access.radius.server', ipsec) == None:
raise ConfigError('IPsec RADIUS server is not configured!')
if dict_search('authentication.client_mode', ra_conf) != 'eap-radius':
raise ConfigError('RADIUS IP pool requires eap-radius client authentication!')
elif 'pool' not in ipsec['remote_access'] or pool not in ipsec['remote_access']['pool']:
raise ConfigError(f'Requested pool "{pool}" does not exist!')
if 'pool' in ipsec['remote_access']:
for pool, pool_config in ipsec['remote_access']['pool'].items():
if 'prefix' not in pool_config:
raise ConfigError(f'Missing madatory prefix option for pool "{pool}"!')
if 'name_server' in pool_config:
if len(pool_config['name_server']) > 2:
raise ConfigError(f'Only two name-servers are supported for remote-access pool "{pool}"!')
for ns in pool_config['name_server']:
v4_addr_and_ns = is_ipv4(ns) and not is_ipv4(pool_config['prefix'])
v6_addr_and_ns = is_ipv6(ns) and not is_ipv6(pool_config['prefix'])
if v4_addr_and_ns or v6_addr_and_ns:
raise ConfigError('Must use both IPv4 or IPv6 addresses for pool prefix and name-server adresses!')
if 'exclude' in pool_config:
for exclude in pool_config['exclude']:
v4_addr_and_exclude = is_ipv4(exclude) and not is_ipv4(pool_config['prefix'])
v6_addr_and_exclude = is_ipv6(exclude) and not is_ipv6(pool_config['prefix'])
if v4_addr_and_exclude or v6_addr_and_exclude:
raise ConfigError('Must use both IPv4 or IPv6 addresses for pool prefix and exclude prefixes!')
if 'radius' in ipsec['remote_access'] and 'server' in ipsec['remote_access']['radius']:
for server, server_config in ipsec['remote_access']['radius']['server'].items():
if 'key' not in server_config:
raise ConfigError(f'Missing RADIUS secret key for server "{server}"')
if 'site_to_site' in ipsec and 'peer' in ipsec['site_to_site']:
for peer, peer_conf in ipsec['site_to_site']['peer'].items():
has_default_esp = False
# Peer name it is swanctl connection name and shouldn't contain dots or colons, T4118
if bool(re.search(':|\.', peer)):
raise ConfigError(f'Incorrect peer name "{peer}" '
f'Peer name can contain alpha-numeric letters, hyphen and underscore')
if 'remote_address' not in peer_conf:
print(f'You should set correct remote-address "peer {peer} remote-address x.x.x.x"\n')
if 'default_esp_group' in peer_conf:
has_default_esp = True
if 'esp_group' not in ipsec or peer_conf['default_esp_group'] not in ipsec['esp_group']:
raise ConfigError(f"Invalid esp-group on site-to-site peer {peer}")
if 'ike_group' in peer_conf:
if 'ike_group' not in ipsec or peer_conf['ike_group'] not in ipsec['ike_group']:
raise ConfigError(f"Invalid ike-group on site-to-site peer {peer}")
else:
raise ConfigError(f"Missing ike-group on site-to-site peer {peer}")
if 'authentication' not in peer_conf or 'mode' not in peer_conf['authentication']:
raise ConfigError(f"Missing authentication on site-to-site peer {peer}")
if {'id', 'use_x509_id'} <= set(peer_conf['authentication']):
raise ConfigError(f"Manually set peer id and use-x509-id are mutually exclusive!")
if peer_conf['authentication']['mode'] == 'x509':
if 'x509' not in peer_conf['authentication']:
raise ConfigError(f"Missing x509 settings on site-to-site peer {peer}")
x509 = peer_conf['authentication']['x509']
if 'ca_certificate' not in x509 or 'certificate' not in x509:
raise ConfigError(f"Missing x509 certificates on site-to-site peer {peer}")
verify_pki_x509(ipsec['pki'], x509)
elif peer_conf['authentication']['mode'] == 'rsa':
if 'rsa' not in peer_conf['authentication']:
raise ConfigError(f"Missing RSA settings on site-to-site peer {peer}")
rsa = peer_conf['authentication']['rsa']
if 'local_key' not in rsa:
raise ConfigError(f"Missing RSA local-key on site-to-site peer {peer}")
if 'remote_key' not in rsa:
raise ConfigError(f"Missing RSA remote-key on site-to-site peer {peer}")
verify_pki_rsa(ipsec['pki'], rsa)
if 'local_address' not in peer_conf and 'dhcp_interface' not in peer_conf:
raise ConfigError(f"Missing local-address or dhcp-interface on site-to-site peer {peer}")
if 'dhcp_interface' in peer_conf:
dhcp_interface = peer_conf['dhcp_interface']
verify_interface_exists(dhcp_interface)
dhcp_base = directories['isc_dhclient_dir']
if not os.path.exists(f'{dhcp_base}/dhclient_{dhcp_interface}.conf'):
raise ConfigError(f"Invalid dhcp-interface on site-to-site peer {peer}")
address = get_dhcp_address(dhcp_interface)
count = 0
while not address and count < dhcp_wait_attempts:
address = get_dhcp_address(dhcp_interface)
count += 1
sleep(dhcp_wait_sleep)
if not address:
ipsec['dhcp_no_address'][peer] = dhcp_interface
print(f"Failed to get address from dhcp-interface on site-to-site peer {peer} -- skipped")
continue
if 'vti' in peer_conf:
if 'local_address' in peer_conf and 'dhcp_interface' in peer_conf:
raise ConfigError(f"A single local-address or dhcp-interface is required when using VTI on site-to-site peer {peer}")
if dict_search('options.disable_route_autoinstall',
ipsec) == None:
Warning('It\'s recommended to use ipsec vti with the next command\n[set vpn ipsec option disable-route-autoinstall]')
if 'bind' in peer_conf['vti']:
vti_interface = peer_conf['vti']['bind']
- if not os.path.exists(f'/sys/class/net/{vti_interface}'):
+ if not interface_exists(vti_interface):
raise ConfigError(f'VTI interface {vti_interface} for site-to-site peer {peer} does not exist!')
if 'vti' not in peer_conf and 'tunnel' not in peer_conf:
raise ConfigError(f"No VTI or tunnel specified on site-to-site peer {peer}")
if 'tunnel' in peer_conf:
for tunnel, tunnel_conf in peer_conf['tunnel'].items():
if 'esp_group' not in tunnel_conf and not has_default_esp:
raise ConfigError(f"Missing esp-group on tunnel {tunnel} for site-to-site peer {peer}")
esp_group_name = tunnel_conf['esp_group'] if 'esp_group' in tunnel_conf else peer_conf['default_esp_group']
if esp_group_name not in ipsec['esp_group']:
raise ConfigError(f"Invalid esp-group on tunnel {tunnel} for site-to-site peer {peer}")
esp_group = ipsec['esp_group'][esp_group_name]
if 'mode' in esp_group and esp_group['mode'] == 'transport':
if 'protocol' in tunnel_conf and ((peer in ['any', '0.0.0.0']) or ('local_address' not in peer_conf or peer_conf['local_address'] in ['any', '0.0.0.0'])):
raise ConfigError(f"Fixed local-address or peer required when a protocol is defined with ESP transport mode on tunnel {tunnel} for site-to-site peer {peer}")
if ('local' in tunnel_conf and 'prefix' in tunnel_conf['local']) or ('remote' in tunnel_conf and 'prefix' in tunnel_conf['remote']):
raise ConfigError(f"Local/remote prefix cannot be used with ESP transport mode on tunnel {tunnel} for site-to-site peer {peer}")
def cleanup_pki_files():
for path in [CERT_PATH, CA_PATH, CRL_PATH, KEY_PATH, PUBKEY_PATH]:
if not os.path.exists(path):
continue
for file in os.listdir(path):
file_path = os.path.join(path, file)
if os.path.isfile(file_path):
os.unlink(file_path)
def generate_pki_files_x509(pki, x509_conf):
ca_cert_name = x509_conf['ca_certificate']
ca_cert_data = dict_search_args(pki, 'ca', ca_cert_name, 'certificate')
ca_cert_crls = dict_search_args(pki, 'ca', ca_cert_name, 'crl') or []
ca_index = 1
crl_index = 1
ca_cert = load_certificate(ca_cert_data)
pki_ca_certs = [load_certificate(ca['certificate']) for ca in pki['ca'].values()]
ca_cert_chain = find_chain(ca_cert, pki_ca_certs)
cert_name = x509_conf['certificate']
cert_data = dict_search_args(pki, 'certificate', cert_name, 'certificate')
key_data = dict_search_args(pki, 'certificate', cert_name, 'private', 'key')
protected = 'passphrase' in x509_conf
for ca_cert_obj in ca_cert_chain:
with open(os.path.join(CA_PATH, f'{ca_cert_name}_{ca_index}.pem'), 'w') as f:
f.write(encode_certificate(ca_cert_obj))
ca_index += 1
for crl in ca_cert_crls:
with open(os.path.join(CRL_PATH, f'{ca_cert_name}_{crl_index}.pem'), 'w') as f:
f.write(wrap_crl(crl))
crl_index += 1
with open(os.path.join(CERT_PATH, f'{cert_name}.pem'), 'w') as f:
f.write(wrap_certificate(cert_data))
with open(os.path.join(KEY_PATH, f'x509_{cert_name}.pem'), 'w') as f:
f.write(wrap_private_key(key_data, protected))
def generate_pki_files_rsa(pki, rsa_conf):
local_key_name = rsa_conf['local_key']
local_key_data = dict_search_args(pki, 'key_pair', local_key_name, 'private', 'key')
protected = 'passphrase' in rsa_conf
remote_key_name = rsa_conf['remote_key']
remote_key_data = dict_search_args(pki, 'key_pair', remote_key_name, 'public', 'key')
local_key = load_private_key(local_key_data, rsa_conf['passphrase'] if protected else None)
with open(os.path.join(KEY_PATH, f'rsa_{local_key_name}.pem'), 'w') as f:
f.write(wrap_private_key(local_key_data, protected))
with open(os.path.join(PUBKEY_PATH, f'{local_key_name}.pem'), 'w') as f:
f.write(encode_public_key(local_key.public_key()))
with open(os.path.join(PUBKEY_PATH, f'{remote_key_name}.pem'), 'w') as f:
f.write(wrap_public_key(remote_key_data))
def generate(ipsec):
cleanup_pki_files()
if not ipsec:
for config_file in [charon_dhcp_conf, charon_radius_conf, interface_conf, swanctl_conf]:
if os.path.isfile(config_file):
os.unlink(config_file)
render(charon_conf, 'ipsec/charon.j2', {'install_routes': default_install_routes})
return
if ipsec['dhcp_no_address']:
with open(DHCP_HOOK_IFLIST, 'w') as f:
f.write(" ".join(ipsec['dhcp_no_address'].values()))
elif os.path.exists(DHCP_HOOK_IFLIST):
os.unlink(DHCP_HOOK_IFLIST)
for path in [swanctl_dir, CERT_PATH, CA_PATH, CRL_PATH, PUBKEY_PATH]:
if not os.path.exists(path):
os.mkdir(path, mode=0o755)
if not os.path.exists(KEY_PATH):
os.mkdir(KEY_PATH, mode=0o700)
if 'l2tp' in ipsec:
if 'authentication' in ipsec['l2tp'] and 'x509' in ipsec['l2tp']['authentication']:
generate_pki_files_x509(ipsec['pki'], ipsec['l2tp']['authentication']['x509'])
if 'remote_access' in ipsec and 'connection' in ipsec['remote_access']:
for rw, rw_conf in ipsec['remote_access']['connection'].items():
if 'authentication' in rw_conf and 'x509' in rw_conf['authentication']:
generate_pki_files_x509(ipsec['pki'], rw_conf['authentication']['x509'])
if 'site_to_site' in ipsec and 'peer' in ipsec['site_to_site']:
for peer, peer_conf in ipsec['site_to_site']['peer'].items():
if peer in ipsec['dhcp_no_address']:
continue
if peer_conf['authentication']['mode'] == 'x509':
generate_pki_files_x509(ipsec['pki'], peer_conf['authentication']['x509'])
elif peer_conf['authentication']['mode'] == 'rsa':
generate_pki_files_rsa(ipsec['pki'], peer_conf['authentication']['rsa'])
local_ip = ''
if 'local_address' in peer_conf:
local_ip = peer_conf['local_address']
elif 'dhcp_interface' in peer_conf:
local_ip = get_dhcp_address(peer_conf['dhcp_interface'])
ipsec['site_to_site']['peer'][peer]['local_address'] = local_ip
if 'tunnel' in peer_conf:
for tunnel, tunnel_conf in peer_conf['tunnel'].items():
local_prefixes = dict_search_args(tunnel_conf, 'local', 'prefix')
remote_prefixes = dict_search_args(tunnel_conf, 'remote', 'prefix')
if not local_prefixes or not remote_prefixes:
continue
passthrough = None
for local_prefix in local_prefixes:
for remote_prefix in remote_prefixes:
local_net = ipaddress.ip_network(local_prefix)
remote_net = ipaddress.ip_network(remote_prefix)
if local_net.overlaps(remote_net):
if passthrough is None:
passthrough = []
passthrough.append(local_prefix)
ipsec['site_to_site']['peer'][peer]['tunnel'][tunnel]['passthrough'] = passthrough
# auth psk <tag> dhcp-interface <xxx>
if jmespath.search('authentication.psk.*.dhcp_interface', ipsec):
for psk, psk_config in ipsec['authentication']['psk'].items():
if 'dhcp_interface' in psk_config:
for iface in psk_config['dhcp_interface']:
id = get_dhcp_address(iface)
if id:
ipsec['authentication']['psk'][psk]['id'].append(id)
render(charon_conf, 'ipsec/charon.j2', ipsec)
render(charon_dhcp_conf, 'ipsec/charon/dhcp.conf.j2', ipsec)
render(charon_radius_conf, 'ipsec/charon/eap-radius.conf.j2', ipsec)
render(interface_conf, 'ipsec/interfaces_use.conf.j2', ipsec)
render(swanctl_conf, 'ipsec/swanctl.conf.j2', ipsec)
def resync_nhrp(ipsec):
if ipsec and not ipsec['nhrp_exists']:
return
tmp = run('/usr/libexec/vyos/conf_mode/protocols_nhrp.py')
if tmp > 0:
print('ERROR: failed to reapply NHRP settings!')
def apply(ipsec):
systemd_service = 'strongswan.service'
if not ipsec:
call(f'systemctl stop {systemd_service}')
else:
call(f'systemctl reload-or-restart {systemd_service}')
resync_nhrp(ipsec)
if __name__ == '__main__':
try:
ipsec = get_config()
verify(ipsec)
generate(ipsec)
apply(ipsec)
except ConfigError as e:
print(e)
exit(1)
diff --git a/src/etc/systemd/system/certbot.service.d/10-override.conf b/src/etc/systemd/system/certbot.service.d/10-override.conf
new file mode 100644
index 000000000..542f77eb2
--- /dev/null
+++ b/src/etc/systemd/system/certbot.service.d/10-override.conf
@@ -0,0 +1,7 @@
+[Unit]
+After=
+After=vyos-router.service
+
+[Service]
+ExecStart=
+ExecStart=/usr/bin/certbot renew --config-dir /config/auth/letsencrypt --no-random-sleep-on-renew --post-hook "/usr/libexec/vyos/vyos-certbot-renew-pki.sh"
diff --git a/src/helpers/vyos-certbot-renew-pki.sh b/src/helpers/vyos-certbot-renew-pki.sh
new file mode 100755
index 000000000..d0b663f7b
--- /dev/null
+++ b/src/helpers/vyos-certbot-renew-pki.sh
@@ -0,0 +1,3 @@
+#!/bin/sh
+source /opt/vyatta/etc/functions/script-template
+/usr/libexec/vyos/conf_mode/pki.py certbot_renew
diff --git a/src/migration-scripts/https/5-to-6 b/src/migration-scripts/https/5-to-6
new file mode 100755
index 000000000..b4159f02f
--- /dev/null
+++ b/src/migration-scripts/https/5-to-6
@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 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/>.
+
+# T5886: Add support for ACME protocol (LetsEncrypt), migrate https certbot
+# to new "pki certificate" CLI tree
+
+import os
+import sys
+
+from vyos.configtree import ConfigTree
+from vyos.defaults import directories
+
+vyos_certbot_dir = directories['certbot']
+
+if len(sys.argv) < 2:
+ print("Must specify file name!")
+ sys.exit(1)
+
+file_name = sys.argv[1]
+
+with open(file_name, 'r') as f:
+ config_file = f.read()
+
+config = ConfigTree(config_file)
+
+base = ['service', 'https', 'certificates']
+if not config.exists(base):
+ # Nothing to do
+ sys.exit(0)
+
+# both domain-name and email must be set on CLI - ensured by previous verify()
+domain_names = config.return_values(base + ['certbot', 'domain-name'])
+email = config.return_value(base + ['certbot', 'email'])
+config.delete(base)
+
+# Set default certname based on domain-name
+cert_name = 'https-' + domain_names[0].split('.')[0]
+# Overwrite certname from previous certbot calls if available
+if os.path.exists(f'{vyos_certbot_dir}/live'):
+ for cert in [f.path.split('/')[-1] for f in os.scandir(f'{vyos_certbot_dir}/live') if f.is_dir()]:
+ cert_name = cert
+ break
+
+for domain in domain_names:
+ config.set(['pki', 'certificate', cert_name, 'acme', 'domain-name'], value=domain, replace=False)
+ config.set(['pki', 'certificate', cert_name, 'acme', 'email'], value=email)
+
+# Update Webserver certificate
+config.set(base + ['certificate'], value=cert_name)
+
+try:
+ with open(file_name, 'w') as f:
+ f.write(config.to_string())
+except OSError as e:
+ print("Failed to save the modified config: {}".format(e))
+ sys.exit(1)
diff --git a/src/op_mode/pki.py b/src/op_mode/pki.py
index 6c854afb5..ad2c1ada0 100755
--- a/src/op_mode/pki.py
+++ b/src/op_mode/pki.py
@@ -1,1080 +1,1088 @@
#!/usr/bin/env python3
#
-# Copyright (C) 2021-2023 VyOS maintainers and contributors
+# 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 argparse
import ipaddress
import os
import re
import sys
import tabulate
from cryptography import x509
from cryptography.x509.oid import ExtendedKeyUsageOID
from vyos.config import Config
+from vyos.config import config_dict_mangle_acme
from vyos.pki import encode_certificate, encode_public_key, encode_private_key, encode_dh_parameters
from vyos.pki import get_certificate_fingerprint
from vyos.pki import create_certificate, create_certificate_request, create_certificate_revocation_list
from vyos.pki import create_private_key
from vyos.pki import create_dh_parameters
from vyos.pki import load_certificate, load_certificate_request, load_private_key
from vyos.pki import load_crl, load_dh_parameters, load_public_key
from vyos.pki import verify_certificate
from vyos.utils.io import ask_input
from vyos.utils.io import ask_yes_no
from vyos.utils.misc import install_into_config
from vyos.utils.process import cmd
CERT_REQ_END = '-----END CERTIFICATE REQUEST-----'
auth_dir = '/config/auth'
# Helper Functions
conf = Config()
def get_default_values():
# Fetch default x509 values
base = ['pki', 'x509', 'default']
x509_defaults = conf.get_config_dict(base, key_mangling=('-', '_'),
no_tag_node_value_mangle=True,
get_first_key=True,
with_recursive_defaults=True)
return x509_defaults
def get_config_ca_certificate(name=None):
# Fetch ca certificates from config
base = ['pki', 'ca']
if not conf.exists(base):
return False
if name:
base = base + [name]
if not conf.exists(base + ['private', 'key']) or not conf.exists(base + ['certificate']):
return False
return conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
def get_config_certificate(name=None):
# Get certificates from config
base = ['pki', 'certificate']
if not conf.exists(base):
return False
if name:
base = base + [name]
if not conf.exists(base + ['private', 'key']) or not conf.exists(base + ['certificate']):
return False
- return conf.get_config_dict(base, key_mangling=('-', '_'),
+ pki = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
+ if pki:
+ for certificate in pki:
+ pki[certificate] = config_dict_mangle_acme(certificate, pki[certificate])
+
+ return pki
def get_certificate_ca(cert, ca_certs):
# Find CA certificate for given certificate
if not ca_certs:
return None
for ca_name, ca_dict in ca_certs.items():
if 'certificate' not in ca_dict:
continue
ca_cert = load_certificate(ca_dict['certificate'])
if not ca_cert:
continue
if verify_certificate(cert, ca_cert):
return ca_name
return None
def get_config_revoked_certificates():
# Fetch revoked certificates from config
ca_base = ['pki', 'ca']
cert_base = ['pki', 'certificate']
certs = []
if conf.exists(ca_base):
ca_certificates = conf.get_config_dict(ca_base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
certs.extend(ca_certificates.values())
if conf.exists(cert_base):
certificates = conf.get_config_dict(cert_base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
certs.extend(certificates.values())
return [cert_dict for cert_dict in certs if 'revoke' in cert_dict]
def get_revoked_by_serial_numbers(serial_numbers=[]):
# Return serial numbers of revoked certificates
certs_out = []
certs = get_config_certificate()
ca_certs = get_config_ca_certificate()
if certs:
for cert_name, cert_dict in certs.items():
if 'certificate' not in cert_dict:
continue
cert = load_certificate(cert_dict['certificate'])
if cert.serial_number in serial_numbers:
certs_out.append(cert_name)
if ca_certs:
for cert_name, cert_dict in ca_certs.items():
if 'certificate' not in cert_dict:
continue
cert = load_certificate(cert_dict['certificate'])
if cert.serial_number in serial_numbers:
certs_out.append(cert_name)
return certs_out
def install_certificate(name, cert='', private_key=None, key_type=None, key_passphrase=None, is_ca=False):
# Show/install conf commands for certificate
prefix = 'ca' if is_ca else 'certificate'
base = f"pki {prefix} {name}"
config_paths = []
if cert:
cert_pem = "".join(encode_certificate(cert).strip().split("\n")[1:-1])
config_paths.append(f"{base} certificate '{cert_pem}'")
if private_key:
key_pem = "".join(encode_private_key(private_key, passphrase=key_passphrase).strip().split("\n")[1:-1])
config_paths.append(f"{base} private key '{key_pem}'")
if key_passphrase:
config_paths.append(f"{base} private password-protected")
install_into_config(conf, config_paths)
def install_crl(ca_name, crl):
# Show/install conf commands for crl
crl_pem = "".join(encode_certificate(crl).strip().split("\n")[1:-1])
install_into_config(conf, [f"pki ca {ca_name} crl '{crl_pem}'"])
def install_dh_parameters(name, params):
# Show/install conf commands for dh params
dh_pem = "".join(encode_dh_parameters(params).strip().split("\n")[1:-1])
install_into_config(conf, [f"pki dh {name} parameters '{dh_pem}'"])
def install_ssh_key(name, public_key, private_key, passphrase=None):
# Show/install conf commands for ssh key
key_openssh = encode_public_key(public_key, encoding='OpenSSH', key_format='OpenSSH')
username = os.getlogin()
type_key_split = key_openssh.split(" ")
base = f"system login user {username} authentication public-keys {name}"
install_into_config(conf, [
f"{base} key '{type_key_split[1]}'",
f"{base} type '{type_key_split[0]}'"
])
print(encode_private_key(private_key, encoding='PEM', key_format='OpenSSH', passphrase=passphrase))
def install_keypair(name, key_type, private_key=None, public_key=None, passphrase=None, prompt=True):
# Show/install conf commands for key-pair
config_paths = []
if public_key:
install_public_key = not prompt or ask_yes_no('Do you want to install the public key?', default=True)
public_key_pem = encode_public_key(public_key)
if install_public_key:
install_public_pem = "".join(public_key_pem.strip().split("\n")[1:-1])
config_paths.append(f"pki key-pair {name} public key '{install_public_pem}'")
else:
print("Public key:")
print(public_key_pem)
if private_key:
install_private_key = not prompt or ask_yes_no('Do you want to install the private key?', default=True)
private_key_pem = encode_private_key(private_key, passphrase=passphrase)
if install_private_key:
install_private_pem = "".join(private_key_pem.strip().split("\n")[1:-1])
config_paths.append(f"pki key-pair {name} private key '{install_private_pem}'")
if passphrase:
config_paths.append(f"pki key-pair {name} private password-protected")
else:
print("Private key:")
print(private_key_pem)
install_into_config(conf, config_paths)
def install_openvpn_key(name, key_data, key_version='1'):
config_paths = [
f"pki openvpn shared-secret {name} key '{key_data}'",
f"pki openvpn shared-secret {name} version '{key_version}'"
]
install_into_config(conf, config_paths)
def install_wireguard_key(interface, private_key, public_key):
# Show conf commands for installing wireguard key pairs
from vyos.ifconfig import Section
if Section.section(interface) != 'wireguard':
print(f'"{interface}" is not a WireGuard interface name!')
exit(1)
# Check if we are running in a config session - if yes, we can directly write to the CLI
install_into_config(conf, [f"interfaces wireguard {interface} private-key '{private_key}'"])
print(f"Corresponding public-key to use on peer system is: '{public_key}'")
def install_wireguard_psk(interface, peer, psk):
from vyos.ifconfig import Section
if Section.section(interface) != 'wireguard':
print(f'"{interface}" is not a WireGuard interface name!')
exit(1)
# Check if we are running in a config session - if yes, we can directly write to the CLI
install_into_config(conf, [f"interfaces wireguard {interface} peer {peer} preshared-key '{psk}'"])
def ask_passphrase():
passphrase = None
print("Note: If you plan to use the generated key on this router, do not encrypt the private key.")
if ask_yes_no('Do you want to encrypt the private key with a passphrase?'):
passphrase = ask_input('Enter passphrase:')
return passphrase
def write_file(filename, contents):
full_path = os.path.join(auth_dir, filename)
directory = os.path.dirname(full_path)
if not os.path.exists(directory):
print('Failed to write file: directory does not exist')
return False
if os.path.exists(full_path) and not ask_yes_no('Do you want to overwrite the existing file?'):
return False
with open(full_path, 'w') as f:
f.write(contents)
print(f'File written to {full_path}')
# Generation functions
def generate_private_key():
key_type = ask_input('Enter private key type: [rsa, dsa, ec]', default='rsa', valid_responses=['rsa', 'dsa', 'ec'])
size_valid = []
size_default = 0
if key_type in ['rsa', 'dsa']:
size_default = 2048
size_valid = [512, 1024, 2048, 4096]
elif key_type == 'ec':
size_default = 256
size_valid = [224, 256, 384, 521]
size = ask_input('Enter private key bits:', default=size_default, numeric_only=True, valid_responses=size_valid)
return create_private_key(key_type, size), key_type
def parse_san_string(san_string):
if not san_string:
return None
output = []
san_split = san_string.strip().split(",")
for pair_str in san_split:
tag, value = pair_str.strip().split(":", 1)
if tag == 'ipv4':
output.append(ipaddress.IPv4Address(value))
elif tag == 'ipv6':
output.append(ipaddress.IPv6Address(value))
elif tag == 'dns':
output.append(value)
return output
def generate_certificate_request(private_key=None, key_type=None, return_request=False, name=None, install=False, file=False, ask_san=True):
if not private_key:
private_key, key_type = generate_private_key()
default_values = get_default_values()
subject = {}
subject['country'] = ask_input('Enter country code:', default=default_values['country'])
subject['state'] = ask_input('Enter state:', default=default_values['state'])
subject['locality'] = ask_input('Enter locality:', default=default_values['locality'])
subject['organization'] = ask_input('Enter organization name:', default=default_values['organization'])
subject['common_name'] = ask_input('Enter common name:', default='vyos.io')
subject_alt_names = None
if ask_san and ask_yes_no('Do you want to configure Subject Alternative Names?'):
print("Enter alternative names in a comma separate list, example: ipv4:1.1.1.1,ipv6:fe80::1,dns:vyos.net")
san_string = ask_input('Enter Subject Alternative Names:')
subject_alt_names = parse_san_string(san_string)
cert_req = create_certificate_request(subject, private_key, subject_alt_names)
if return_request:
return cert_req
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert_req))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
print("Certificate request:")
print(encode_certificate(cert_req) + "\n")
install_certificate(name, private_key=private_key, key_type=key_type, key_passphrase=passphrase, is_ca=False)
if file:
write_file(f'{name}.csr', encode_certificate(cert_req))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate(cert_req, ca_cert, ca_private_key, is_ca=False, is_sub_ca=False):
valid_days = ask_input('Enter how many days certificate will be valid:', default='365' if not is_ca else '1825', numeric_only=True)
cert_type = None
if not is_ca:
cert_type = ask_input('Enter certificate type: (client, server)', default='server', valid_responses=['client', 'server'])
return create_certificate(cert_req, ca_cert, ca_private_key, valid_days, cert_type, is_ca, is_sub_ca)
def generate_ca_certificate(name, install=False, file=False):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True, ask_san=False)
cert = generate_certificate(cert_req, cert_req, private_key, is_ca=True)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key, key_type, key_passphrase=passphrase, is_ca=True)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_ca_certificate_sign(name, ca_name, install=False, file=False):
ca_dict = get_config_ca_certificate(ca_name)
if not ca_dict:
print(f"CA certificate or private key for '{ca_name}' not found")
return None
ca_cert = load_certificate(ca_dict['certificate'])
if not ca_cert:
print("Failed to load signing CA certificate, aborting")
return None
ca_private = ca_dict['private']
ca_private_passphrase = None
if 'password_protected' in ca_private:
ca_private_passphrase = ask_input('Enter signing CA private key passphrase:')
ca_private_key = load_private_key(ca_private['key'], passphrase=ca_private_passphrase)
if not ca_private_key:
print("Failed to load signing CA private key, aborting")
return None
private_key = None
key_type = None
cert_req = None
if not ask_yes_no('Do you already have a certificate request?'):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True, ask_san=False)
else:
print("Paste certificate request and press enter:")
lines = []
curr_line = ''
while True:
curr_line = input().strip()
if not curr_line or curr_line == CERT_REQ_END:
break
lines.append(curr_line)
if not lines:
print("Aborted")
return None
wrap = lines[0].find('-----') < 0 # Only base64 pasted, add the CSR tags for parsing
cert_req = load_certificate_request("\n".join(lines), wrap)
if not cert_req:
print("Invalid certificate request")
return None
cert = generate_certificate(cert_req, ca_cert, ca_private_key, is_ca=True, is_sub_ca=True)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key, key_type, key_passphrase=passphrase, is_ca=True)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate_sign(name, ca_name, install=False, file=False):
ca_dict = get_config_ca_certificate(ca_name)
if not ca_dict:
print(f"CA certificate or private key for '{ca_name}' not found")
return None
ca_cert = load_certificate(ca_dict['certificate'])
if not ca_cert:
print("Failed to load CA certificate, aborting")
return None
ca_private = ca_dict['private']
ca_private_passphrase = None
if 'password_protected' in ca_private:
ca_private_passphrase = ask_input('Enter CA private key passphrase:')
ca_private_key = load_private_key(ca_private['key'], passphrase=ca_private_passphrase)
if not ca_private_key:
print("Failed to load CA private key, aborting")
return None
private_key = None
key_type = None
cert_req = None
if not ask_yes_no('Do you already have a certificate request?'):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True)
else:
print("Paste certificate request and press enter:")
lines = []
curr_line = ''
while True:
curr_line = input().strip()
if not curr_line or curr_line == CERT_REQ_END:
break
lines.append(curr_line)
if not lines:
print("Aborted")
return None
wrap = lines[0].find('-----') < 0 # Only base64 pasted, add the CSR tags for parsing
cert_req = load_certificate_request("\n".join(lines), wrap)
if not cert_req:
print("Invalid certificate request")
return None
cert = generate_certificate(cert_req, ca_cert, ca_private_key, is_ca=False)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key, key_type, key_passphrase=passphrase, is_ca=False)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate_selfsign(name, install=False, file=False):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True)
cert = generate_certificate(cert_req, cert_req, private_key, is_ca=False)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key=private_key, key_type=key_type, key_passphrase=passphrase, is_ca=False)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate_revocation_list(ca_name, install=False, file=False):
ca_dict = get_config_ca_certificate(ca_name)
if not ca_dict:
print(f"CA certificate or private key for '{ca_name}' not found")
return None
ca_cert = load_certificate(ca_dict['certificate'])
if not ca_cert:
print("Failed to load CA certificate, aborting")
return None
ca_private = ca_dict['private']
ca_private_passphrase = None
if 'password_protected' in ca_private:
ca_private_passphrase = ask_input('Enter CA private key passphrase:')
ca_private_key = load_private_key(ca_private['key'], passphrase=ca_private_passphrase)
if not ca_private_key:
print("Failed to load CA private key, aborting")
return None
revoked_certs = get_config_revoked_certificates()
to_revoke = []
for cert_dict in revoked_certs:
if 'certificate' not in cert_dict:
continue
cert_data = cert_dict['certificate']
try:
cert = load_certificate(cert_data)
if cert.issuer == ca_cert.subject:
to_revoke.append(cert.serial_number)
except ValueError:
continue
if not to_revoke:
print("No revoked certificates to add to the CRL")
return None
crl = create_certificate_revocation_list(ca_cert, ca_private_key, to_revoke)
if not crl:
print("Failed to create CRL")
return None
if not install and not file:
print(encode_certificate(crl))
return None
if install:
install_crl(ca_name, crl)
if file:
write_file(f'{name}.crl', encode_certificate(crl))
def generate_ssh_keypair(name, install=False, file=False):
private_key, key_type = generate_private_key()
public_key = private_key.public_key()
passphrase = ask_passphrase()
if not install and not file:
print(encode_public_key(public_key, encoding='OpenSSH', key_format='OpenSSH'))
print("")
print(encode_private_key(private_key, encoding='PEM', key_format='OpenSSH', passphrase=passphrase))
return None
if install:
install_ssh_key(name, public_key, private_key, passphrase)
if file:
write_file(f'{name}.pem', encode_public_key(public_key, encoding='OpenSSH', key_format='OpenSSH'))
write_file(f'{name}.key', encode_private_key(private_key, encoding='PEM', key_format='OpenSSH', passphrase=passphrase))
def generate_dh_parameters(name, install=False, file=False):
bits = ask_input('Enter DH parameters key size:', default=2048, numeric_only=True)
print("Generating parameters...")
dh_params = create_dh_parameters(bits)
if not dh_params:
print("Failed to create DH parameters")
return None
if not install and not file:
print("DH Parameters:")
print(encode_dh_parameters(dh_params))
if install:
install_dh_parameters(name, dh_params)
if file:
write_file(f'{name}.pem', encode_dh_parameters(dh_params))
def generate_keypair(name, install=False, file=False):
private_key, key_type = generate_private_key()
public_key = private_key.public_key()
passphrase = ask_passphrase()
if not install and not file:
print(encode_public_key(public_key))
print("")
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_keypair(name, key_type, private_key, public_key, passphrase)
if file:
write_file(f'{name}.pem', encode_public_key(public_key))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_openvpn_key(name, install=False, file=False):
result = cmd('openvpn --genkey secret /dev/stdout | grep -o "^[^#]*"')
if not result:
print("Failed to generate OpenVPN key")
return None
if not install and not file:
print(result)
return None
if install:
key_lines = result.split("\n")
key_data = "".join(key_lines[1:-1]) # Remove wrapper tags and line endings
key_version = '1'
version_search = re.search(r'BEGIN OpenVPN Static key V(\d+)', result) # Future-proofing (hopefully)
if version_search:
key_version = version_search[1]
install_openvpn_key(name, key_data, key_version)
if file:
write_file(f'{name}.key', result)
def generate_wireguard_key(interface=None, install=False):
private_key = cmd('wg genkey')
public_key = cmd('wg pubkey', input=private_key)
if interface and install:
install_wireguard_key(interface, private_key, public_key)
else:
print(f'Private key: {private_key}')
print(f'Public key: {public_key}', end='\n\n')
def generate_wireguard_psk(interface=None, peer=None, install=False):
psk = cmd('wg genpsk')
if interface and peer and install:
install_wireguard_psk(interface, peer, psk)
else:
print(f'Pre-shared key: {psk}')
# Import functions
def import_ca_certificate(name, path=None, key_path=None):
if path:
if not os.path.exists(path):
print(f'File not found: {path}')
return
cert = None
with open(path) as f:
cert_data = f.read()
cert = load_certificate(cert_data, wrap_tags=False)
if not cert:
print(f'Invalid certificate: {path}')
return
install_certificate(name, cert, is_ca=True)
if key_path:
if not os.path.exists(key_path):
print(f'File not found: {key_path}')
return
key = None
passphrase = ask_input('Enter private key passphrase: ') or None
with open(key_path) as f:
key_data = f.read()
key = load_private_key(key_data, passphrase=passphrase, wrap_tags=False)
if not key:
print(f'Invalid private key or passphrase: {path}')
return
install_certificate(name, private_key=key, is_ca=True)
def import_certificate(name, path=None, key_path=None):
if path:
if not os.path.exists(path):
print(f'File not found: {path}')
return
cert = None
with open(path) as f:
cert_data = f.read()
cert = load_certificate(cert_data, wrap_tags=False)
if not cert:
print(f'Invalid certificate: {path}')
return
install_certificate(name, cert, is_ca=False)
if key_path:
if not os.path.exists(key_path):
print(f'File not found: {key_path}')
return
key = None
passphrase = ask_input('Enter private key passphrase: ') or None
with open(key_path) as f:
key_data = f.read()
key = load_private_key(key_data, passphrase=passphrase, wrap_tags=False)
if not key:
print(f'Invalid private key or passphrase: {path}')
return
install_certificate(name, private_key=key, is_ca=False)
def import_crl(name, path):
if not os.path.exists(path):
print(f'File not found: {path}')
return
crl = None
with open(path) as f:
crl_data = f.read()
crl = load_crl(crl_data, wrap_tags=False)
if not crl:
print(f'Invalid certificate: {path}')
return
install_crl(name, crl)
def import_dh_parameters(name, path):
if not os.path.exists(path):
print(f'File not found: {path}')
return
dh = None
with open(path) as f:
dh_data = f.read()
dh = load_dh_parameters(dh_data, wrap_tags=False)
if not dh:
print(f'Invalid DH parameters: {path}')
return
install_dh_parameters(name, dh)
def import_keypair(name, path=None, key_path=None):
if path:
if not os.path.exists(path):
print(f'File not found: {path}')
return
key = None
with open(path) as f:
key_data = f.read()
key = load_public_key(key_data, wrap_tags=False)
if not key:
print(f'Invalid public key: {path}')
return
install_keypair(name, None, public_key=key, prompt=False)
if key_path:
if not os.path.exists(key_path):
print(f'File not found: {key_path}')
return
key = None
passphrase = ask_input('Enter private key passphrase: ') or None
with open(key_path) as f:
key_data = f.read()
key = load_private_key(key_data, passphrase=passphrase, wrap_tags=False)
if not key:
print(f'Invalid private key or passphrase: {path}')
return
install_keypair(name, None, private_key=key, prompt=False)
def import_openvpn_secret(name, path):
if not os.path.exists(path):
print(f'File not found: {path}')
return
key_data = None
key_version = '1'
with open(path) as f:
key_lines = f.read().split("\n")
key_data = "".join(key_lines[1:-1]) # Remove wrapper tags and line endings
version_search = re.search(r'BEGIN OpenVPN Static key V(\d+)', key_lines[0]) # Future-proofing (hopefully)
if version_search:
key_version = version_search[1]
install_openvpn_key(name, key_data, key_version)
# Show functions
def show_certificate_authority(name=None, pem=False):
headers = ['Name', 'Subject', 'Issuer CN', 'Issued', 'Expiry', 'Private Key', 'Parent']
data = []
certs = get_config_ca_certificate()
if certs:
for cert_name, cert_dict in certs.items():
if name and name != cert_name:
continue
if 'certificate' not in cert_dict:
continue
cert = load_certificate(cert_dict['certificate'])
if name and pem:
print(encode_certificate(cert))
return
parent_ca_name = get_certificate_ca(cert, certs)
cert_issuer_cn = cert.issuer.rfc4514_string().split(",")[0]
if not parent_ca_name or parent_ca_name == cert_name:
parent_ca_name = 'N/A'
if not cert:
continue
have_private = 'Yes' if 'private' in cert_dict and 'key' in cert_dict['private'] else 'No'
data.append([cert_name, cert.subject.rfc4514_string(), cert_issuer_cn, cert.not_valid_before, cert.not_valid_after, have_private, parent_ca_name])
print("Certificate Authorities:")
print(tabulate.tabulate(data, headers))
def show_certificate(name=None, pem=False):
headers = ['Name', 'Type', 'Subject CN', 'Issuer CN', 'Issued', 'Expiry', 'Revoked', 'Private Key', 'CA Present']
data = []
certs = get_config_certificate()
if certs:
ca_certs = get_config_ca_certificate()
for cert_name, cert_dict in certs.items():
if name and name != cert_name:
continue
if 'certificate' not in cert_dict:
continue
cert = load_certificate(cert_dict['certificate'])
if not cert:
continue
if name and pem:
print(encode_certificate(cert))
return
ca_name = get_certificate_ca(cert, ca_certs)
cert_subject_cn = cert.subject.rfc4514_string().split(",")[0]
cert_issuer_cn = cert.issuer.rfc4514_string().split(",")[0]
cert_type = 'Unknown'
try:
ext = cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage)
if ext and ExtendedKeyUsageOID.SERVER_AUTH in ext.value:
cert_type = 'Server'
elif ext and ExtendedKeyUsageOID.CLIENT_AUTH in ext.value:
cert_type = 'Client'
except:
pass
revoked = 'Yes' if 'revoke' in cert_dict else 'No'
have_private = 'Yes' if 'private' in cert_dict and 'key' in cert_dict['private'] else 'No'
have_ca = f'Yes ({ca_name})' if ca_name else 'No'
data.append([
cert_name, cert_type, cert_subject_cn, cert_issuer_cn,
cert.not_valid_before, cert.not_valid_after,
revoked, have_private, have_ca])
print("Certificates:")
print(tabulate.tabulate(data, headers))
def show_certificate_fingerprint(name, hash):
cert = get_config_certificate(name=name)
cert = load_certificate(cert['certificate'])
print(get_certificate_fingerprint(cert, hash))
def show_crl(name=None, pem=False):
headers = ['CA Name', 'Updated', 'Revokes']
data = []
certs = get_config_ca_certificate()
if certs:
for cert_name, cert_dict in certs.items():
if name and name != cert_name:
continue
if 'crl' not in cert_dict:
continue
crls = cert_dict['crl']
if isinstance(crls, str):
crls = [crls]
for crl_data in cert_dict['crl']:
crl = load_crl(crl_data)
if not crl:
continue
if name and pem:
print(encode_certificate(crl))
continue
certs = get_revoked_by_serial_numbers([revoked.serial_number for revoked in crl])
data.append([cert_name, crl.last_update, ", ".join(certs)])
if name and pem:
return
print("Certificate Revocation Lists:")
print(tabulate.tabulate(data, headers))
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--action', help='PKI action', required=True)
# X509
parser.add_argument('--ca', help='Certificate Authority', required=False)
parser.add_argument('--certificate', help='Certificate', required=False)
parser.add_argument('--crl', help='Certificate Revocation List', required=False)
parser.add_argument('--sign', help='Sign certificate with specified CA', required=False)
parser.add_argument('--self-sign', help='Self-sign the certificate', action='store_true')
parser.add_argument('--pem', help='Output using PEM encoding', action='store_true')
parser.add_argument('--fingerprint', help='Show fingerprint and exit', action='store')
# SSH
parser.add_argument('--ssh', help='SSH Key', required=False)
# DH
parser.add_argument('--dh', help='DH Parameters', required=False)
# Key pair
parser.add_argument('--keypair', help='Key pair', required=False)
# OpenVPN
parser.add_argument('--openvpn', help='OpenVPN TLS key', required=False)
# WireGuard
parser.add_argument('--wireguard', help='Wireguard', action='store_true')
group = parser.add_mutually_exclusive_group()
group.add_argument('--key', help='Wireguard key pair', action='store_true', required=False)
group.add_argument('--psk', help='Wireguard pre shared key', action='store_true', required=False)
parser.add_argument('--interface', help='Install generated keys into running-config for named interface', action='store')
parser.add_argument('--peer', help='Install generated keys into running-config for peer', action='store')
# Global
parser.add_argument('--file', help='Write generated keys into specified filename', action='store_true')
parser.add_argument('--install', help='Install generated keys into running-config', action='store_true')
parser.add_argument('--filename', help='Write certificate into specified filename', action='store')
parser.add_argument('--key-filename', help='Write key into specified filename', action='store')
args = parser.parse_args()
try:
if args.action == 'generate':
if args.ca:
if args.sign:
generate_ca_certificate_sign(args.ca, args.sign, install=args.install, file=args.file)
else:
generate_ca_certificate(args.ca, install=args.install, file=args.file)
elif args.certificate:
if args.sign:
generate_certificate_sign(args.certificate, args.sign, install=args.install, file=args.file)
elif args.self_sign:
generate_certificate_selfsign(args.certificate, install=args.install, file=args.file)
else:
generate_certificate_request(name=args.certificate, install=args.install, file=args.file)
elif args.crl:
generate_certificate_revocation_list(args.crl, install=args.install, file=args.file)
elif args.ssh:
generate_ssh_keypair(args.ssh, install=args.install, file=args.file)
elif args.dh:
generate_dh_parameters(args.dh, install=args.install, file=args.file)
elif args.keypair:
generate_keypair(args.keypair, install=args.install, file=args.file)
elif args.openvpn:
generate_openvpn_key(args.openvpn, install=args.install, file=args.file)
elif args.wireguard:
# WireGuard supports writing key directly into the CLI, but this
# requires the vyos_libexec_dir environment variable to be set
os.environ["vyos_libexec_dir"] = "/usr/libexec/vyos"
if args.key:
generate_wireguard_key(args.interface, install=args.install)
if args.psk:
generate_wireguard_psk(args.interface, peer=args.peer, install=args.install)
elif args.action == 'import':
if args.ca:
import_ca_certificate(args.ca, path=args.filename, key_path=args.key_filename)
elif args.certificate:
import_certificate(args.certificate, path=args.filename, key_path=args.key_filename)
elif args.crl:
import_crl(args.crl, args.filename)
elif args.dh:
import_dh_parameters(args.dh, args.filename)
elif args.keypair:
import_keypair(args.keypair, path=args.filename, key_path=args.key_filename)
elif args.openvpn:
import_openvpn_secret(args.openvpn, args.filename)
elif args.action == 'show':
if args.ca:
ca_name = None if args.ca == 'all' else args.ca
if ca_name:
if not conf.exists(['pki', 'ca', ca_name]):
print(f'CA "{ca_name}" does not exist!')
exit(1)
show_certificate_authority(ca_name, args.pem)
elif args.certificate:
cert_name = None if args.certificate == 'all' else args.certificate
if cert_name:
if not conf.exists(['pki', 'certificate', cert_name]):
print(f'Certificate "{cert_name}" does not exist!')
exit(1)
if args.fingerprint is None:
show_certificate(None if args.certificate == 'all' else args.certificate, args.pem)
else:
show_certificate_fingerprint(args.certificate, args.fingerprint)
elif args.crl:
show_crl(None if args.crl == 'all' else args.crl, args.pem)
else:
show_certificate_authority()
+ print('\n')
show_certificate()
+ print('\n')
show_crl()
except KeyboardInterrupt:
print("Aborted")
sys.exit(0)