diff --git a/op-mode-definitions/pki.xml.in b/op-mode-definitions/pki.xml.in
index 254ef08cc..866f482bf 100644
--- a/op-mode-definitions/pki.xml.in
+++ b/op-mode-definitions/pki.xml.in
@@ -1,587 +1,587 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="generate">
<children>
<node name="pki">
<properties>
<help>Generate public key infrastructure (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 -E ${vyos_op_scripts_dir}/pki.py --action generate --ca "$7" --sign "$5" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ca --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --ca "$7" --sign "$5" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ca --name "$7" --sign "$5" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --ca "noname" --sign "$5"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ca --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 -E ${vyos_op_scripts_dir}/pki.py --action generate --ca "$5" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ca --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --ca "$5" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ca --name "$5" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --ca "noname"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ca</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 -E ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$6" --self-sign --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --certificate "$6" --self-sign --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --name "$6" --self-sign --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname" --self-sign</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --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 -E ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$7" --sign "$5" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --certificate "$7" --sign "$5" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --name "$7" --sign "$5" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname" --sign "$5"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --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 -E ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$5" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --certificate "$5" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate --name "$5" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type certificate</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 -E ${vyos_op_scripts_dir}/pki.py --action generate --crl "$4" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type crl --name "$4" --file</command>
</tagNode>
<leafNode name="install">
<properties>
<help>Commands for installing generated CRL into running configuration</help>
</properties>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --crl "$4" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type crl --name "$4" --install</command>
</leafNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --crl "$4"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type crl --name "$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 -E ${vyos_op_scripts_dir}/pki.py --action generate --dh "$5" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type dh --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --dh "$5" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type dh --name "$5" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --dh "noname"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type dh</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 -E ${vyos_op_scripts_dir}/pki.py --action generate --keypair "$5" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type key-pair --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --keypair "$5" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type key-pair --name "$5" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --keypair "noname"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type key-pair</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 -E ${vyos_op_scripts_dir}/pki.py --action generate --openvpn "$6" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type openvpn --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --openvpn "$6" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type openvpn --name "$6" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --openvpn "noname"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type openvpn</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 -E ${vyos_op_scripts_dir}/pki.py --action generate --ssh "$5" --file</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ssh --name "$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>${vyos_op_scripts_dir}/pki.py --action generate --ssh "$5" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ssh --name "$5" --install</command>
</tagNode>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --ssh "noname"</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type ssh</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>${vyos_op_scripts_dir}/pki.py --action generate --wireguard --key --interface "$7" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type wireguard --key --interface "$7" --install</command>
</tagNode>
</children>
</node>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --wireguard --key</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type 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>${vyos_op_scripts_dir}/pki.py --action generate --wireguard --psk --interface "$7" --peer "$9" --install</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type wireguard --psk --interface "$7" --peer "$9" --install</command>
</tagNode>
</children>
</tagNode>
</children>
</node>
</children>
- <command>${vyos_op_scripts_dir}/pki.py --action generate --wireguard --psk</command>
+ <command>${vyos_op_scripts_dir}/pki.py generate_pki --pki-type 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 -E ${vyos_op_scripts_dir}/pki.py --action import --ca "$4" --filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type ca --name "$4" --filename "$6"</command>
</tagNode>
<tagNode name="key-file">
<properties>
<help>Path to private key file</help>
</properties>
- <command>sudo -E ${vyos_op_scripts_dir}/pki.py --action import --ca "$4" --key-filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type ca --name "$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 -E ${vyos_op_scripts_dir}/pki.py --action import --certificate "$4" --filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type certificate --name "$4" --filename "$6"</command>
</tagNode>
<tagNode name="key-file">
<properties>
<help>Path to private key file</help>
</properties>
- <command>sudo -E ${vyos_op_scripts_dir}/pki.py --action import --certificate "$4" --key-filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type certificate --name "$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 -E ${vyos_op_scripts_dir}/pki.py --action import --crl "$4" --filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type crl --name "$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 -E ${vyos_op_scripts_dir}/pki.py --action import --dh "$4" --filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type dh --name "$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 -E ${vyos_op_scripts_dir}/pki.py --action import --keypair "$4" --filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type key-pair --name "$4" --filename "$6"</command>
</tagNode>
<tagNode name="private-file">
<properties>
<help>Path to private key file</help>
</properties>
- <command>sudo -E ${vyos_op_scripts_dir}/pki.py --action import --keypair "$4" --key-filename "$6"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type key-pair --name "$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 -E ${vyos_op_scripts_dir}/pki.py --action import --openvpn "$5" --filename "$7"</command>
+ <command>sudo -E ${vyos_op_scripts_dir}/pki.py import_pki --pki-type openvpn --name "$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>
- <command>sudo ${vyos_op_scripts_dir}/pki.py --action show</command>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_all</command>
<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>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_certificate_authority</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>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_certificate_authority --name "$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>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_certificate_authority --name "$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>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_certificate</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>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_certificate --name "$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>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_certificate --name "$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>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py show_certificate --name "$4" --fingerprint "$6"</command>
</tagNode>
</children>
</tagNode>
<leafNode name="crl">
<properties>
<help>Show x509 certificate revocation lists</help>
</properties>
- <command>${vyos_op_scripts_dir}/pki.py --action show --crl "all"</command>
+ <command>${vyos_op_scripts_dir}/pki.py show_crl</command>
</leafNode>
<tagNode name="crl">
<properties>
<help>Show x509 certificate revocation lists by CA name</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
- <command>${vyos_op_scripts_dir}/pki.py --action show --crl "$4"</command>
+ <command>${vyos_op_scripts_dir}/pki.py show_crl --name "$4"</command>
<children>
<leafNode name="pem">
<properties>
<help>Show x509 certificate revocation lists by CA name in PEM format</help>
</properties>
- <command>${vyos_op_scripts_dir}/pki.py --action show --crl "$4" --pem</command>
+ <command>${vyos_op_scripts_dir}/pki.py show_crl --name "$4" --pem</command>
</leafNode>
</children>
</tagNode>
</children>
</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/python/vyos/configsession.py b/python/vyos/configsession.py
index 9c56d246a..5876dc5b0 100644
--- a/python/vyos/configsession.py
+++ b/python/vyos/configsession.py
@@ -1,338 +1,337 @@
# Copyright (C) 2019-2024 VyOS maintainers and contributors
#
# This library is free software; you can redistribute it and/or modify it under the terms of
# the GNU Lesser General Public License as published by the Free Software Foundation;
# either version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
# without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
# See the GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License along with this library;
# if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
# configsession -- the write API for the VyOS running config
import os
import re
import sys
import subprocess
from vyos.defaults import directories
from vyos.utils.process import is_systemd_service_running
from vyos.utils.dict import dict_to_paths
CLI_SHELL_API = '/bin/cli-shell-api'
SET = '/opt/vyatta/sbin/my_set'
DELETE = '/opt/vyatta/sbin/my_delete'
COMMENT = '/opt/vyatta/sbin/my_comment'
COMMIT = '/opt/vyatta/sbin/my_commit'
DISCARD = '/opt/vyatta/sbin/my_discard'
SHOW_CONFIG = ['/bin/cli-shell-api', 'showConfig']
LOAD_CONFIG = ['/bin/cli-shell-api', 'loadFile']
MIGRATE_LOAD_CONFIG = ['/usr/libexec/vyos/vyos-load-config.py']
SAVE_CONFIG = ['/usr/libexec/vyos/vyos-save-config.py']
INSTALL_IMAGE = [
'/usr/libexec/vyos/op_mode/image_installer.py',
'--action',
'add',
'--no-prompt',
'--image-path',
]
IMPORT_PKI = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'import']
IMPORT_PKI_NO_PROMPT = [
'/usr/libexec/vyos/op_mode/pki.py',
- '--action',
- 'import',
+ 'import_pki',
'--no-prompt',
]
REMOVE_IMAGE = [
'/usr/libexec/vyos/op_mode/image_manager.py',
'--action',
'delete',
'--no-prompt',
'--image-name',
]
SET_DEFAULT_IMAGE = [
'/usr/libexec/vyos/op_mode/image_manager.py',
'--action',
'set',
'--no-prompt',
'--image-name',
]
GENERATE = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'generate']
SHOW = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'show']
RESET = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'reset']
REBOOT = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'reboot']
POWEROFF = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'poweroff']
OP_CMD_ADD = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'add']
OP_CMD_DELETE = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'delete']
# Default "commit via" string
APP = 'vyos-http-api'
# When started as a service rather than from a user shell,
# the process lacks the VyOS-specific environment that comes
# from bash configs, so we have to inject it
# XXX: maybe it's better to do via a systemd environment file
def inject_vyos_env(env):
env['VYATTA_CFG_GROUP_NAME'] = 'vyattacfg'
env['VYATTA_USER_LEVEL_DIR'] = '/opt/vyatta/etc/shell/level/admin'
env['VYATTA_PROCESS_CLIENT'] = 'gui2_rest'
env['VYOS_HEADLESS_CLIENT'] = 'vyos_http_api'
env['vyatta_bindir'] = '/opt/vyatta/bin'
env['vyatta_cfg_templates'] = '/opt/vyatta/share/vyatta-cfg/templates'
env['vyatta_configdir'] = directories['vyos_configdir']
env['vyatta_datadir'] = '/opt/vyatta/share'
env['vyatta_datarootdir'] = '/opt/vyatta/share'
env['vyatta_libdir'] = '/opt/vyatta/lib'
env['vyatta_libexecdir'] = '/opt/vyatta/libexec'
env['vyatta_op_templates'] = '/opt/vyatta/share/vyatta-op/templates'
env['vyatta_prefix'] = '/opt/vyatta'
env['vyatta_sbindir'] = '/opt/vyatta/sbin'
env['vyatta_sysconfdir'] = '/opt/vyatta/etc'
env['vyos_bin_dir'] = '/usr/bin'
env['vyos_cfg_templates'] = '/opt/vyatta/share/vyatta-cfg/templates'
env['vyos_completion_dir'] = '/usr/libexec/vyos/completion'
env['vyos_configdir'] = directories['vyos_configdir']
env['vyos_conf_scripts_dir'] = '/usr/libexec/vyos/conf_mode'
env['vyos_datadir'] = '/opt/vyatta/share'
env['vyos_datarootdir'] = '/opt/vyatta/share'
env['vyos_libdir'] = '/opt/vyatta/lib'
env['vyos_libexec_dir'] = '/usr/libexec/vyos'
env['vyos_op_scripts_dir'] = '/usr/libexec/vyos/op_mode'
env['vyos_op_templates'] = '/opt/vyatta/share/vyatta-op/templates'
env['vyos_prefix'] = '/opt/vyatta'
env['vyos_sbin_dir'] = '/usr/sbin'
env['vyos_validators_dir'] = '/usr/libexec/vyos/validators'
# if running the vyos-configd daemon, inject the vyshim env var
if is_systemd_service_running('vyos-configd.service'):
env['vyshim'] = '/usr/sbin/vyshim'
return env
class ConfigSessionError(Exception):
pass
class ConfigSession(object):
"""
The write API of VyOS.
"""
def __init__(self, session_id, app=APP):
"""
Creates a new config session.
Args:
session_id (str): Session identifier
app (str): Application name, purely informational
Note:
The session identifier MUST be globally unique within the system.
The best practice is to only have one ConfigSession object per process
and used the PID for the session identifier.
"""
env_str = subprocess.check_output(
[CLI_SHELL_API, 'getSessionEnv', str(session_id)]
)
self.__session_id = session_id
# Extract actual variables from the chunk of shell it outputs
# XXX: it's better to extend cli-shell-api to provide easily readable output
env_list = re.findall(r'([A-Z_]+)=([^;\s]+)', env_str.decode())
session_env = os.environ
session_env = inject_vyos_env(session_env)
for k, v in env_list:
session_env[k] = v
self.__session_env = session_env
self.__session_env['COMMIT_VIA'] = app
self.__run_command([CLI_SHELL_API, 'setupSession'])
def __del__(self):
try:
output = (
subprocess.check_output(
[CLI_SHELL_API, 'teardownSession'], env=self.__session_env
)
.decode()
.strip()
)
if output:
print(
'cli-shell-api teardownSession output for sesion {0}: {1}'.format(
self.__session_id, output
),
file=sys.stderr,
)
except Exception as e:
print(
'Could not tear down session {0}: {1}'.format(self.__session_id, e),
file=sys.stderr,
)
def __run_command(self, cmd_list):
p = subprocess.Popen(
cmd_list,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
env=self.__session_env,
)
(stdout_data, stderr_data) = p.communicate()
output = stdout_data.decode()
result = p.wait()
if result != 0:
raise ConfigSessionError(output)
return output
def get_session_env(self):
return self.__session_env
def set(self, path, value=None):
if not value:
value = []
else:
value = [value]
self.__run_command([SET] + path + value)
def set_section(self, path: list, d: dict):
try:
for p in dict_to_paths(d):
self.set(path + p)
except (ValueError, ConfigSessionError) as e:
raise ConfigSessionError(e)
def delete(self, path, value=None):
if not value:
value = []
else:
value = [value]
self.__run_command([DELETE] + path + value)
def load_section(self, path: list, d: dict):
try:
self.delete(path)
if d:
for p in dict_to_paths(d):
self.set(path + p)
except (ValueError, ConfigSessionError) as e:
raise ConfigSessionError(e)
def set_section_tree(self, d: dict):
try:
if d:
for p in dict_to_paths(d):
self.set(p)
except (ValueError, ConfigSessionError) as e:
raise ConfigSessionError(e)
def load_section_tree(self, mask: dict, d: dict):
try:
if mask:
for p in dict_to_paths(mask):
self.delete(p)
if d:
for p in dict_to_paths(d):
self.set(p)
except (ValueError, ConfigSessionError) as e:
raise ConfigSessionError(e)
def comment(self, path, value=None):
if not value:
value = ['']
else:
value = [value]
self.__run_command([COMMENT] + path + value)
def commit(self):
out = self.__run_command([COMMIT])
return out
def discard(self):
self.__run_command([DISCARD])
def show_config(self, path, format='raw'):
config_data = self.__run_command(SHOW_CONFIG + path)
if format == 'raw':
return config_data
def load_config(self, file_path):
out = self.__run_command(LOAD_CONFIG + [file_path])
return out
def load_explicit(self, file_path):
from vyos.load_config import load
from vyos.load_config import LoadConfigError
try:
load(file_path, switch='explicit')
except LoadConfigError as e:
raise ConfigSessionError(e) from e
def migrate_and_load_config(self, file_path):
out = self.__run_command(MIGRATE_LOAD_CONFIG + [file_path])
return out
def save_config(self, file_path):
out = self.__run_command(SAVE_CONFIG + [file_path])
return out
def install_image(self, url):
out = self.__run_command(INSTALL_IMAGE + [url])
return out
def remove_image(self, name):
out = self.__run_command(REMOVE_IMAGE + [name])
return out
def import_pki(self, path):
out = self.__run_command(IMPORT_PKI + path)
return out
def import_pki_no_prompt(self, path):
out = self.__run_command(IMPORT_PKI_NO_PROMPT + path)
return out
def set_default_image(self, name):
out = self.__run_command(SET_DEFAULT_IMAGE + [name])
return out
def generate(self, path):
out = self.__run_command(GENERATE + path)
return out
def show(self, path):
out = self.__run_command(SHOW + path)
return out
def reboot(self, path):
out = self.__run_command(REBOOT + path)
return out
def reset(self, path):
out = self.__run_command(RESET + path)
return out
def poweroff(self, path):
out = self.__run_command(POWEROFF + path)
return out
def add_container_image(self, name):
out = self.__run_command(OP_CMD_ADD + ['container', 'image'] + [name])
return out
def delete_container_image(self, name):
out = self.__run_command(OP_CMD_DELETE + ['container', 'image'] + [name])
return out
def show_container_image(self):
out = self.__run_command(SHOW + ['container', 'image'])
return out
diff --git a/python/vyos/opmode.py b/python/vyos/opmode.py
index 066c8058f..136ac35e2 100644
--- a/python/vyos/opmode.py
+++ b/python/vyos/opmode.py
@@ -1,285 +1,288 @@
# Copyright 2022-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
import re
import sys
import typing
from humps import decamelize
class Error(Exception):
""" Any error that makes requested operation impossible to complete
for reasons unrelated to the user input or script logic.
This is the base class, scripts should not use it directly
and should raise more specific errors instead,
whenever possible.
"""
pass
class UnconfiguredSubsystem(Error):
""" Requested operation is valid, but cannot be completed
because corresponding subsystem is not configured
and thus is not running.
"""
pass
class UnconfiguredObject(UnconfiguredSubsystem):
""" Requested operation is valid but cannot be completed
because its parameter refers to an object that does not exist
in the system configuration.
"""
pass
class DataUnavailable(Error):
""" Requested operation is valid, but cannot be completed
because data for it is not available.
This error MAY be treated as temporary because such issues
are often caused by transient events such as service restarts.
"""
pass
class PermissionDenied(Error):
""" Requested operation is valid, but the caller has no permission
to perform it.
"""
pass
class InsufficientResources(Error):
""" Requested operation and its arguments are valid but the system
does not have enough resources (such as drive space or memory)
to complete it.
"""
pass
class UnsupportedOperation(Error):
""" Requested operation is technically valid but is not implemented yet. """
pass
class IncorrectValue(Error):
""" Requested operation is valid, but an argument provided has an
incorrect value, preventing successful completion.
"""
pass
class CommitInProgress(Error):
""" Requested operation is valid, but not possible at the time due
to a commit being in progress.
"""
pass
class InternalError(Error):
""" Any situation when VyOS detects that it could not perform
an operation correctly due to logic errors in its own code
or errors in underlying software.
"""
pass
def _is_op_mode_function_name(name):
- if re.match(r"^(show|clear|reset|restart|add|update|delete|generate|set|renew|release|execute)", name):
+ if re.match(
+ r'^(show|clear|reset|restart|add|update|delete|generate|set|renew|release|execute|import)',
+ name,
+ ):
return True
else:
return False
def _capture_output(name):
if re.match(r"^(show|generate)", name):
return True
else:
return False
def _get_op_mode_functions(module):
from inspect import getmembers, isfunction
# Get all functions in that module
funcs = getmembers(module, isfunction)
# getmembers returns (name, func) tuples
funcs = list(filter(lambda ft: _is_op_mode_function_name(ft[0]), funcs))
funcs_dict = {}
for (name, thunk) in funcs:
funcs_dict[name] = thunk
return funcs_dict
def _is_optional_type(t):
# Optional[t] is internally an alias for Union[t, NoneType]
# and there's no easy way to get union members it seems
if (type(t) == typing._UnionGenericAlias):
if (len(t.__args__) == 2):
if t.__args__[1] == type(None):
return True
return False
def _get_arg_type(t):
""" Returns the type itself if it's a primitive type,
or the "real" type of typing.Optional
Doesn't work with anything else at the moment!
"""
if _is_optional_type(t):
return t.__args__[0]
else:
return t
def _is_literal_type(t):
if _is_optional_type(t):
t = _get_arg_type(t)
if typing.get_origin(t) == typing.Literal:
return True
return False
def _get_literal_values(t):
""" Returns the tuple of allowed values for a Literal type
"""
if not _is_literal_type(t):
return tuple()
if _is_optional_type(t):
t = _get_arg_type(t)
return typing.get_args(t)
def _normalize_field_name(name):
# Convert the name to string if it is not
# (in some cases they may be numbers)
name = str(name)
# Replace all separators with underscores
name = re.sub(r'(\s|[\(\)\[\]\{\}\-\.\,:\"\'\`])+', '_', name)
# Replace specific characters with textual descriptions
name = re.sub(r'@', '_at_', name)
name = re.sub(r'%', '_percentage_', name)
name = re.sub(r'~', '_tilde_', name)
# Force all letters to lowercase
name = name.lower()
# Remove leading and trailing underscores, if any
name = re.sub(r'(^(_+)(?=[^_])|_+$)', '', name)
# Ensure there are only single underscores
name = re.sub(r'_+', '_', name)
return name
def _normalize_dict_field_names(old_dict):
new_dict = {}
for key in old_dict:
new_key = _normalize_field_name(key)
new_dict[new_key] = _normalize_field_names(old_dict[key])
# Sanity check
if len(old_dict) != len(new_dict):
raise InternalError("Dictionary fields do not allow unique normalization")
else:
return new_dict
def _normalize_field_names(value):
if isinstance(value, dict):
return _normalize_dict_field_names(value)
elif isinstance(value, list):
return list(map(lambda v: _normalize_field_names(v), value))
else:
return value
def run(module):
from argparse import ArgumentParser
functions = _get_op_mode_functions(module)
parser = ArgumentParser()
subparsers = parser.add_subparsers(dest="subcommand")
for function_name in functions:
subparser = subparsers.add_parser(function_name, help=functions[function_name].__doc__)
type_hints = typing.get_type_hints(functions[function_name])
if 'return' in type_hints:
del type_hints['return']
for opt in type_hints:
th = type_hints[opt]
# Function argument names use underscores as separators
# but command-line options should use hyphens
# Without this, we'd get options like "--foo_bar"
opt = re.sub(r'_', '-', opt)
if _get_arg_type(th) == bool:
subparser.add_argument(f"--{opt}", action='store_true')
else:
if _is_optional_type(th):
if _is_literal_type(th):
subparser.add_argument(f"--{opt}",
choices=list(_get_literal_values(th)),
default=None)
else:
subparser.add_argument(f"--{opt}",
type=_get_arg_type(th), default=None)
else:
if _is_literal_type(th):
subparser.add_argument(f"--{opt}",
choices=list(_get_literal_values(th)),
required=True)
else:
subparser.add_argument(f"--{opt}",
type=_get_arg_type(th), required=True)
# Get options as a dict rather than a namespace,
# so that we can modify it and pack for passing to functions
args = vars(parser.parse_args())
if not args["subcommand"]:
print("Subcommand required!")
parser.print_usage()
sys.exit(1)
function_name = args["subcommand"]
func = functions[function_name]
# Remove the subcommand from the arguments,
# it would cause an extra argument error when we pass the dict to a function
del args["subcommand"]
# Show and generate commands must always get the "raw" argument,
# but other commands (clear/reset/restart/add/delete) should not,
# because they produce no output and it makes no sense for them.
if ("raw" not in args) and _capture_output(function_name):
args["raw"] = False
if _capture_output(function_name):
# Show and generate commands are slightly special:
# they may return human-formatted output
# or a raw dict that we need to serialize in JSON for printing
res = func(**args)
if not args["raw"]:
return res
else:
if not isinstance(res, dict) and not isinstance(res, list):
raise InternalError(f"Bare literal is not an acceptable raw output, must be a list or an object.\
The output was:{res}")
res = decamelize(res)
res = _normalize_field_names(res)
from json import dumps
return dumps(res, indent=4)
else:
# Other functions should not return anything,
# although they may print their own warnings or status messages
func(**args)
diff --git a/src/op_mode/pki.py b/src/op_mode/pki.py
index 5652a5d74..56b873bb1 100755
--- a/src/op_mode/pki.py
+++ b/src/op_mode/pki.py
@@ -1,1120 +1,1151 @@
#!/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 argparse
import ipaddress
import os
import re
import sys
import tabulate
+import typing
from cryptography import x509
from cryptography.x509.oid import ExtendedKeyUsageOID
+import vyos.opmode
+
from vyos.config import Config
from vyos.config import config_dict_mangle_acme
from vyos.pki import encode_certificate
from vyos.pki import encode_public_key
from vyos.pki import encode_private_key
from vyos.pki import encode_dh_parameters
from vyos.pki import get_certificate_fingerprint
from vyos.pki import create_certificate
from vyos.pki import create_certificate_request
from vyos.pki import create_certificate_revocation_list
from vyos.pki import create_private_key
from vyos.pki import create_dh_parameters
from vyos.pki import load_certificate
from vyos.pki import load_certificate_request
from vyos.pki import load_private_key
from vyos.pki import load_crl
from vyos.pki import load_dh_parameters
from vyos.pki import 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'
+ArgsPkiType = typing.Literal['ca', 'certificate', 'dh', 'key-pair', 'openvpn', 'crl']
+ArgsPkiTypeGen = typing.Literal[ArgsPkiType, typing.Literal['ssh', 'wireguard']]
+ArgsFingerprint = typing.Literal['sha256', 'sha384', 'sha512']
+
# Helper Functions
conf = Config()
+
+
+def _verify(target):
+ """Decorator checks if config for PKI exists"""
+ from functools import wraps
+
+ if target not in ['ca', 'certificate']:
+ raise ValueError('Invalid PKI')
+
+ def _verify_target(func):
+ @wraps(func)
+ def _wrapper(*args, **kwargs):
+ name = kwargs.get('name')
+ unconf_message = f'PKI {target} "{name}" does not exist!'
+ if name:
+ if not conf.exists(['pki', target, name]):
+ raise vyos.opmode.UnconfiguredSubsystem(unconf_message)
+ return func(*args, **kwargs)
+
+ return _wrapper
+
+ return _verify_target
+
+
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
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' or tag == 'rfc822':
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 = {}
while True:
country = ask_input('Enter country code:', default=default_values['country'])
if len(country) != 2:
print("Country name must be a 2 character country code")
continue
subject['country'] = country
break
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,rfc822:user@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 = None
if private_key is not None:
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
if private_key is not None:
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))
if private_key is not None:
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 = None
if private_key is not None:
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
if private_key is not None:
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))
if private_key is not None:
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, no_prompt=False, passphrase=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
if not no_prompt:
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: {key_path}')
return
install_certificate(name, private_key=key, is_ca=True)
def import_certificate(name, path=None, key_path=None, no_prompt=False, passphrase=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
if not no_prompt:
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: {key_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, no_prompt=False, passphrase=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
if not no_prompt:
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: {key_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().strip().split("\n")
key_lines = list(filter(lambda line: not line.strip().startswith('#'), key_lines)) # Remove commented lines
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):
+
+def generate_pki(
+ raw: bool,
+ pki_type: ArgsPkiTypeGen,
+ name: typing.Optional[str],
+ file: typing.Optional[bool],
+ install: typing.Optional[bool],
+ sign: typing.Optional[str],
+ self_sign: typing.Optional[bool],
+ key: typing.Optional[bool],
+ psk: typing.Optional[bool],
+ interface: typing.Optional[str],
+ peer: typing.Optional[str],
+):
+ try:
+ if pki_type == 'ca':
+ if sign:
+ generate_ca_certificate_sign(name, sign, install=install, file=file)
+ else:
+ generate_ca_certificate(name, install=install, file=file)
+ elif pki_type == 'certificate':
+ if sign:
+ generate_certificate_sign(name, sign, install=install, file=file)
+ elif self_sign:
+ generate_certificate_selfsign(name, install=install, file=file)
+ else:
+ generate_certificate_request(name=name, install=install, file=file)
+
+ elif pki_type == 'crl':
+ generate_certificate_revocation_list(name, install=install, file=file)
+
+ elif pki_type == 'ssh':
+ generate_ssh_keypair(name, install=install, file=file)
+
+ elif pki_type == 'dh':
+ generate_dh_parameters(name, install=install, file=file)
+
+ elif pki_type == 'key-pair':
+ generate_keypair(name, install=install, file=file)
+
+ elif pki_type == 'openvpn':
+ generate_openvpn_key(name, install=install, file=file)
+
+ elif pki_type == '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 key:
+ generate_wireguard_key(interface, install=install)
+ if psk:
+ generate_wireguard_psk(interface, peer=peer, install=install)
+ except KeyboardInterrupt:
+ print('Aborted')
+ sys.exit(0)
+
+
+def import_pki(
+ name: str,
+ pki_type: ArgsPkiType,
+ filename: typing.Optional[str],
+ key_filename: typing.Optional[str],
+ no_prompt: typing.Optional[bool],
+ passphrase: typing.Optional[str],
+):
+ try:
+ if pki_type == 'ca':
+ import_ca_certificate(
+ name,
+ path=filename,
+ key_path=key_filename,
+ no_prompt=no_prompt,
+ passphrase=passphrase,
+ )
+ elif pki_type == 'certificate':
+ import_certificate(
+ name,
+ path=filename,
+ key_path=key_filename,
+ no_prompt=no_prompt,
+ passphrase=passphrase,
+ )
+ elif pki_type == 'crl':
+ import_crl(name, filename)
+ elif pki_type == 'dh':
+ import_dh_parameters(name, filename)
+ elif pki_type == 'key-pair':
+ import_keypair(
+ name,
+ path=filename,
+ key_path=key_filename,
+ no_prompt=no_prompt,
+ passphrase=passphrase,
+ )
+ elif pki_type == 'openvpn':
+ import_openvpn_secret(name, filename)
+ except KeyboardInterrupt:
+ print('Aborted')
+ sys.exit(0)
+
+
+@_verify('ca')
+def show_certificate_authority(
+ raw: bool, name: typing.Optional[str] = None, pem: typing.Optional[bool] = 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, fingerprint_hash=None):
+
+@_verify('certificate')
+def show_certificate(
+ raw: bool,
+ name: typing.Optional[str] = None,
+ pem: typing.Optional[bool] = False,
+ fingerprint: typing.Optional[ArgsFingerprint] = None,
+):
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
- elif name and fingerprint_hash:
- print(get_certificate_fingerprint(cert, fingerprint_hash))
+ elif name and fingerprint:
+ print(get_certificate_fingerprint(cert, fingerprint))
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_crl(name=None, pem=False):
+
+def show_crl(
+ raw: bool, name: typing.Optional[str] = None, pem: typing.Optional[bool] = 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)
+def show_all(raw: bool):
+ show_certificate_authority(raw)
+ print('\n')
+ show_certificate(raw)
+ print('\n')
+ show_crl(raw)
- # 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')
-
- parser.add_argument('--no-prompt', action='store_true', help='Perform action non-interactively')
- parser.add_argument('--passphrase', help='A passphrase to decrypt the private key')
-
- args = parser.parse_args()
+if __name__ == '__main__':
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,
- no_prompt=args.no_prompt, passphrase=args.passphrase)
- elif args.certificate:
- import_certificate(args.certificate, path=args.filename, key_path=args.key_filename,
- no_prompt=args.no_prompt, passphrase=args.passphrase)
- 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,
- no_prompt=args.no_prompt, passphrase=args.passphrase)
- 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(args.certificate, fingerprint_hash=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)
+ res = vyos.opmode.run(sys.modules[__name__])
+ if res:
+ print(res)
+ except (ValueError, vyos.opmode.Error) as e:
+ print(e)
+ sys.exit(1)
diff --git a/src/services/api/rest/routers.py b/src/services/api/rest/routers.py
index da981d5bf..5612e947c 100644
--- a/src/services/api/rest/routers.py
+++ b/src/services/api/rest/routers.py
@@ -1,754 +1,754 @@
# Copyright 2024 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/>.
# pylint: disable=line-too-long,raise-missing-from,invalid-name
# pylint: disable=wildcard-import,unused-wildcard-import
# pylint: disable=broad-exception-caught
import json
import copy
import logging
import traceback
from threading import Lock
from typing import Union
from typing import Callable
from typing import TYPE_CHECKING
from fastapi import Depends
from fastapi import Request
from fastapi import Response
from fastapi import HTTPException
from fastapi import APIRouter
from fastapi import BackgroundTasks
from fastapi.routing import APIRoute
from starlette.datastructures import FormData
from starlette.formparsers import FormParser
from starlette.formparsers import MultiPartParser
from starlette.formparsers import MultiPartException
from multipart.multipart import parse_options_header
from vyos.config import Config
from vyos.configtree import ConfigTree
from vyos.configdiff import get_config_diff
from vyos.configsession import ConfigSessionError
from ..session import SessionState
from .models import success
from .models import error
from .models import responses
from .models import ApiModel
from .models import ConfigureModel
from .models import ConfigureListModel
from .models import ConfigSectionModel
from .models import ConfigSectionListModel
from .models import ConfigSectionTreeModel
from .models import BaseConfigSectionTreeModel
from .models import BaseConfigureModel
from .models import BaseConfigSectionModel
from .models import RetrieveModel
from .models import ConfigFileModel
from .models import ImageModel
from .models import ContainerImageModel
from .models import GenerateModel
from .models import ShowModel
from .models import RebootModel
from .models import ResetModel
from .models import ImportPkiModel
from .models import PoweroffModel
if TYPE_CHECKING:
from fastapi import FastAPI
LOG = logging.getLogger('http_api.routers')
lock = Lock()
def check_auth(key_list, key):
key_id = None
for k in key_list:
if k['key'] == key:
key_id = k['id']
return key_id
def auth_required(data: ApiModel):
session = SessionState()
key = data.key
api_keys = session.keys
key_id = check_auth(api_keys, key)
if not key_id:
raise HTTPException(status_code=401, detail='Valid API key is required')
session.id = key_id
# override Request and APIRoute classes in order to convert form request to json;
# do all explicit validation here, for backwards compatability of error messages;
# the explicit validation may be dropped, if desired, in favor of native
# validation by FastAPI/Pydantic, as is used for application/json requests
class MultipartRequest(Request):
"""Override Request class to convert form request to json"""
# pylint: disable=attribute-defined-outside-init
# pylint: disable=too-many-branches,too-many-statements
_form_err = ()
@property
def form_err(self):
return self._form_err
@form_err.setter
def form_err(self, val):
if not self._form_err:
self._form_err = val
@property
def orig_headers(self):
self._orig_headers = super().headers
return self._orig_headers
@property
def headers(self):
self._headers = super().headers.mutablecopy()
self._headers['content-type'] = 'application/json'
return self._headers
async def _get_form(
self, *, max_files: int | float = 1000, max_fields: int | float = 1000
) -> FormData:
if self._form is None:
assert (
parse_options_header is not None
), 'The `python-multipart` library must be installed to use form parsing.'
content_type_header = self.orig_headers.get('Content-Type')
content_type: bytes
content_type, _ = parse_options_header(content_type_header)
if content_type == b'multipart/form-data':
try:
multipart_parser = MultiPartParser(
self.orig_headers,
self.stream(),
max_files=max_files,
max_fields=max_fields,
)
self._form = await multipart_parser.parse()
except MultiPartException as exc:
if 'app' in self.scope:
raise HTTPException(status_code=400, detail=exc.message)
raise exc
elif content_type == b'application/x-www-form-urlencoded':
form_parser = FormParser(self.orig_headers, self.stream())
self._form = await form_parser.parse()
else:
self._form = FormData()
return self._form
async def body(self) -> bytes:
if not hasattr(self, '_body'):
forms = {}
merge = {}
body = await super().body()
self._body = body
form_data = await self.form()
if form_data:
endpoint = self.url.path
LOG.debug('processing form data')
for k, v in form_data.multi_items():
forms[k] = v
if 'data' not in forms:
self.form_err = (422, 'Non-empty data field is required')
return self._body
try:
tmp = json.loads(forms['data'])
except json.JSONDecodeError as e:
self.form_err = (400, f'Failed to parse JSON: {e}')
return self._body
if isinstance(tmp, list):
merge['commands'] = tmp
else:
merge = tmp
if 'commands' in merge:
cmds = merge['commands']
else:
cmds = copy.deepcopy(merge)
cmds = [cmds]
for c in cmds:
if not isinstance(c, dict):
self.form_err = (
400,
f"Malformed command '{c}': any command must be JSON of dict",
)
return self._body
if 'op' not in c:
self.form_err = (
400,
f"Malformed command '{c}': missing 'op' field",
)
if endpoint not in (
'/config-file',
'/container-image',
'/image',
'/configure-section',
):
if 'path' not in c:
self.form_err = (
400,
f"Malformed command '{c}': missing 'path' field",
)
elif not isinstance(c['path'], list):
self.form_err = (
400,
f"Malformed command '{c}': 'path' field must be a list",
)
elif not all(isinstance(el, str) for el in c['path']):
self.form_err = (
400,
f"Malformed command '{0}': 'path' field must be a list of strings",
)
if endpoint in ('/configure'):
if not c['path']:
self.form_err = (
400,
f"Malformed command '{c}': 'path' list must be non-empty",
)
if 'value' in c and not isinstance(c['value'], str):
self.form_err = (
400,
f"Malformed command '{c}': 'value' field must be a string",
)
if endpoint in ('/configure-section'):
if 'section' not in c and 'config' not in c:
self.form_err = (
400,
f"Malformed command '{c}': missing 'section' or 'config' field",
)
if 'key' not in forms and 'key' not in merge:
self.form_err = (401, 'Valid API key is required')
if 'key' in forms and 'key' not in merge:
merge['key'] = forms['key']
new_body = json.dumps(merge)
new_body = new_body.encode()
self._body = new_body
return self._body
class MultipartRoute(APIRoute):
"""Override APIRoute class to convert form request to json"""
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
request = MultipartRequest(request.scope, request.receive)
try:
response: Response = await original_route_handler(request)
except HTTPException as e:
return error(e.status_code, e.detail)
except Exception as e:
form_err = request.form_err
if form_err:
return error(*form_err)
raise e
return response
return custom_route_handler
router = APIRouter(
route_class=MultipartRoute,
responses={**responses},
dependencies=[Depends(auth_required)],
)
self_ref_msg = 'Requested HTTP API server configuration change; commit will be called in the background'
def call_commit(s: SessionState):
try:
s.session.commit()
except ConfigSessionError as e:
s.session.discard()
if s.debug:
LOG.warning(f'ConfigSessionError:\n {traceback.format_exc()}')
else:
LOG.warning(f'ConfigSessionError: {e}')
def _configure_op(
data: Union[
ConfigureModel,
ConfigureListModel,
ConfigSectionModel,
ConfigSectionListModel,
ConfigSectionTreeModel,
],
_request: Request,
background_tasks: BackgroundTasks,
):
# pylint: disable=too-many-branches,too-many-locals,too-many-nested-blocks,too-many-statements
# pylint: disable=consider-using-with
state = SessionState()
session = state.session
env = session.get_session_env()
# Allow users to pass just one command
if not isinstance(data, (ConfigureListModel, ConfigSectionListModel)):
data = [data]
else:
data = data.commands
# We don't want multiple people/apps to be able to commit at once,
# or modify the shared session while someone else is doing the same,
# so the lock is really global
lock.acquire()
config = Config(session_env=env)
status = 200
msg = None
error_msg = None
try:
for c in data:
op = c.op
if not isinstance(c, BaseConfigSectionTreeModel):
path = c.path
if isinstance(c, BaseConfigureModel):
if c.value:
value = c.value
else:
value = ''
# For vyos.configsession calls that have no separate value arguments,
# and for type checking too
cfg_path = ' '.join(path + [value]).strip()
elif isinstance(c, BaseConfigSectionModel):
section = c.section
elif isinstance(c, BaseConfigSectionTreeModel):
mask = c.mask
config = c.config
if isinstance(c, BaseConfigureModel):
if op == 'set':
session.set(path, value=value)
elif op == 'delete':
if state.strict and not config.exists(cfg_path):
raise ConfigSessionError(
f'Cannot delete [{cfg_path}]: path/value does not exist'
)
session.delete(path, value=value)
elif op == 'comment':
session.comment(path, value=value)
else:
raise ConfigSessionError(f"'{op}' is not a valid operation")
elif isinstance(c, BaseConfigSectionModel):
if op == 'set':
session.set_section(path, section)
elif op == 'load':
session.load_section(path, section)
else:
raise ConfigSessionError(f"'{op}' is not a valid operation")
elif isinstance(c, BaseConfigSectionTreeModel):
if op == 'set':
session.set_section_tree(config)
elif op == 'load':
session.load_section_tree(mask, config)
else:
raise ConfigSessionError(f"'{op}' is not a valid operation")
# end for
config = Config(session_env=env)
d = get_config_diff(config)
if d.is_node_changed(['service', 'https']):
background_tasks.add_task(call_commit, state)
msg = self_ref_msg
else:
# capture non-fatal warnings
out = session.commit()
msg = out if out else msg
LOG.info(f"Configuration modified via HTTP API using key '{state.id}'")
except ConfigSessionError as e:
session.discard()
status = 400
if state.debug:
LOG.critical(f'ConfigSessionError:\n {traceback.format_exc()}')
error_msg = str(e)
except Exception:
session.discard()
LOG.critical(traceback.format_exc())
status = 500
# Don't give the details away to the outer world
error_msg = 'An internal error occured. Check the logs for details.'
finally:
lock.release()
if status != 200:
return error(status, error_msg)
return success(msg)
def create_path_import_pki_no_prompt(path):
correct_paths = ['ca', 'certificate', 'key-pair']
if path[1] not in correct_paths:
return False
- path[1] = '--' + path[1].replace('-', '')
path[3] = '--key-filename'
- return path[1:]
+ path.insert(2, '--name')
+ return ['--pki-type'] + path[1:]
@router.post('/configure')
def configure_op(
data: Union[ConfigureModel, ConfigureListModel],
request: Request,
background_tasks: BackgroundTasks,
):
return _configure_op(data, request, background_tasks)
@router.post('/configure-section')
def configure_section_op(
data: Union[ConfigSectionModel, ConfigSectionListModel, ConfigSectionTreeModel],
request: Request,
background_tasks: BackgroundTasks,
):
return _configure_op(data, request, background_tasks)
@router.post('/retrieve')
async def retrieve_op(data: RetrieveModel):
state = SessionState()
session = state.session
env = session.get_session_env()
config = Config(session_env=env)
op = data.op
path = ' '.join(data.path)
try:
if op == 'returnValue':
res = config.return_value(path)
elif op == 'returnValues':
res = config.return_values(path)
elif op == 'exists':
res = config.exists(path)
elif op == 'showConfig':
config_format = 'json'
if data.configFormat:
config_format = data.configFormat
res = session.show_config(path=data.path)
if config_format == 'json':
config_tree = ConfigTree(res)
res = json.loads(config_tree.to_json())
elif config_format == 'json_ast':
config_tree = ConfigTree(res)
res = json.loads(config_tree.to_json_ast())
elif config_format == 'raw':
pass
else:
return error(400, f"'{config_format}' is not a valid config format")
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
@router.post('/config-file')
def config_file_op(data: ConfigFileModel, background_tasks: BackgroundTasks):
state = SessionState()
session = state.session
env = session.get_session_env()
op = data.op
msg = None
try:
if op == 'save':
if data.file:
path = data.file
else:
path = '/config/config.boot'
msg = session.save_config(path)
elif op == 'load':
if data.file:
path = data.file
else:
return error(400, 'Missing required field "file"')
session.migrate_and_load_config(path)
config = Config(session_env=env)
d = get_config_diff(config)
if d.is_node_changed(['service', 'https']):
background_tasks.add_task(call_commit, state)
msg = self_ref_msg
else:
session.commit()
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(msg)
@router.post('/image')
def image_op(data: ImageModel):
state = SessionState()
session = state.session
op = data.op
try:
if op == 'add':
res = session.install_image(data.url)
elif op == 'delete':
res = session.remove_image(data.name)
elif op == 'show':
res = session.show(['system', 'image'])
elif op == 'set_default':
res = session.set_default_image(data.name)
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
@router.post('/container-image')
def container_image_op(data: ContainerImageModel):
state = SessionState()
session = state.session
op = data.op
try:
if op == 'add':
if data.name:
name = data.name
else:
return error(400, 'Missing required field "name"')
res = session.add_container_image(name)
elif op == 'delete':
if data.name:
name = data.name
else:
return error(400, 'Missing required field "name"')
res = session.delete_container_image(name)
elif op == 'show':
res = session.show_container_image()
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
@router.post('/generate')
def generate_op(data: GenerateModel):
state = SessionState()
session = state.session
op = data.op
path = data.path
try:
if op == 'generate':
res = session.generate(path)
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
@router.post('/show')
def show_op(data: ShowModel):
state = SessionState()
session = state.session
op = data.op
path = data.path
try:
if op == 'show':
res = session.show(path)
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
@router.post('/reboot')
def reboot_op(data: RebootModel):
state = SessionState()
session = state.session
op = data.op
path = data.path
try:
if op == 'reboot':
res = session.reboot(path)
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
@router.post('/reset')
def reset_op(data: ResetModel):
state = SessionState()
session = state.session
op = data.op
path = data.path
try:
if op == 'reset':
res = session.reset(path)
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
@router.post('/import-pki')
def import_pki(data: ImportPkiModel):
# pylint: disable=consider-using-with
state = SessionState()
session = state.session
op = data.op
path = data.path
lock.acquire()
try:
if op == 'import-pki':
# need to get rid or interactive mode for private key
if len(path) == 5 and path[3] in ['key-file', 'private-key']:
path_no_prompt = create_path_import_pki_no_prompt(path)
if not path_no_prompt:
return error(400, f"Invalid command: {' '.join(path)}")
if data.passphrase:
path_no_prompt += ['--passphrase', data.passphrase]
res = session.import_pki_no_prompt(path_no_prompt)
else:
res = session.import_pki(path)
if not res[0].isdigit():
return error(400, res)
# commit changes
session.commit()
res = res.split('. ')[0]
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
finally:
lock.release()
return success(res)
@router.post('/poweroff')
def poweroff_op(data: PoweroffModel):
state = SessionState()
session = state.session
op = data.op
path = data.path
try:
if op == 'poweroff':
res = session.poweroff(path)
else:
return error(400, f"'{op}' is not a valid operation")
except ConfigSessionError as e:
return error(400, str(e))
except Exception:
LOG.critical(traceback.format_exc())
return error(500, 'An internal error occured. Check the logs for details.')
return success(res)
def rest_init(app: 'FastAPI'):
if all(r in app.routes for r in router.routes):
return
app.include_router(router)
def rest_clear(app: 'FastAPI'):
for r in router.routes:
if r in app.routes:
app.routes.remove(r)