diff --git a/data/templates/https/nginx.default.j2 b/data/templates/https/nginx.default.j2
index 4619361e5..1dde66ebf 100644
--- a/data/templates/https/nginx.default.j2
+++ b/data/templates/https/nginx.default.j2
@@ -1,69 +1,69 @@
### Autogenerated by service_https.py ###
{% if enable_http_redirect is vyos_defined %}
server {
listen 80 default_server;
server_name {{ hostname }};
return 301 https://$host$request_uri;
}
{% endif %}
server {
{% if listen_address is vyos_defined %}
{% for address in listen_address %}
listen {{ address | bracketize_ipv6 }}:{{ port }} ssl;
{% endfor %}
{% else %}
listen {{ port }} ssl;
listen [::]:{{ port }} ssl;
{% endif %}
server_name {{ hostname }};
root /srv/localui;
{% if request_body_size_limit is vyos_defined %}
client_max_body_size {{ request_body_size_limit }}M;
{% endif %}
# SSL configuration
{% if certificates.cert_path is vyos_defined and certificates.key_path is vyos_defined %}
ssl_certificate {{ certificates.cert_path }};
ssl_certificate_key {{ certificates.key_path }};
{% if certificates.dh_file is vyos_defined %}
ssl_dhparam {{ certificates.dh_file }};
{% endif %}
{% else %}
# Self signed certs generated by the ssl-cert package
# Don't use them in a production server!
include snippets/snakeoil.conf;
{% endif %}
# Improve HTTPS performance with session resumption
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
ssl_protocols {{ 'TLSv' ~ ' TLSv'.join(tls_version) }};
# From LetsEncrypt
ssl_prefer_server_ciphers on;
ssl_ciphers 'ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384:DHE-RSA-AES128-GCM-SHA256:DHE-DSS-AES128-GCM-SHA256:kEDH+AESGCM:ECDHE-RSA-AES128-SHA256:ECDHE-ECDSA-AES128-SHA256:ECDHE-RSA-AES128-SHA:ECDHE-ECDSA-AES128-SHA:ECDHE-RSA-AES256-SHA384:ECDHE-ECDSA-AES256-SHA384:ECDHE-RSA-AES256-SHA:ECDHE-ECDSA-AES256-SHA:DHE-RSA-AES128-SHA256:DHE-RSA-AES128-SHA:DHE-DSS-AES128-SHA256:DHE-RSA-AES256-SHA256:DHE-DSS-AES256-SHA:DHE-RSA-AES256-SHA:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!3DES:!MD5:!PSK';
# proxy settings for HTTP API, if enabled; 503, if not
- location ~ ^/(retrieve|configure|config-file|image|container-image|generate|show|reboot|reset|poweroff|docs|openapi.json|redoc|graphql) {
+ location ~ ^/(retrieve|configure|config-file|image|import-pki|container-image|generate|show|reboot|reset|poweroff|docs|openapi.json|redoc|graphql) {
{% if api is vyos_defined %}
proxy_pass http://unix:/run/api.sock;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_read_timeout 600;
proxy_buffering off;
{% else %}
return 503;
{% endif %}
{% if allow_client.address is vyos_defined %}
{% for address in allow_client.address %}
allow {{ address }};
{% endfor %}
deny all;
{% endif %}
}
error_page 497 =301 https://$host:{{ port }}$request_uri;
}
diff --git a/python/vyos/configsession.py b/python/vyos/configsession.py
index ccf2ce8f2..7d51b94e4 100644
--- a/python/vyos/configsession.py
+++ b/python/vyos/configsession.py
@@ -1,276 +1,287 @@
# 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', '--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 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/pki.py b/python/vyos/pki.py
index 27fe793a8..5a0e2ddda 100644
--- a/python/vyos/pki.py
+++ b/python/vyos/pki.py
@@ -1,453 +1,453 @@
# Copyright (C) 2023-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 datetime
import ipaddress
from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.x509.extensions import ExtensionNotFound
from cryptography.x509.oid import NameOID
from cryptography.x509.oid import ExtendedKeyUsageOID
from cryptography.x509.oid import ExtensionOID
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import dh
from cryptography.hazmat.primitives.asymmetric import dsa
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric import rsa
CERT_BEGIN='-----BEGIN CERTIFICATE-----\n'
CERT_END='\n-----END CERTIFICATE-----'
KEY_BEGIN='-----BEGIN PRIVATE KEY-----\n'
KEY_END='\n-----END PRIVATE KEY-----'
KEY_ENC_BEGIN='-----BEGIN ENCRYPTED PRIVATE KEY-----\n'
KEY_ENC_END='\n-----END ENCRYPTED PRIVATE KEY-----'
KEY_PUB_BEGIN='-----BEGIN PUBLIC KEY-----\n'
KEY_PUB_END='\n-----END PUBLIC KEY-----'
CRL_BEGIN='-----BEGIN X509 CRL-----\n'
CRL_END='\n-----END X509 CRL-----'
CSR_BEGIN='-----BEGIN CERTIFICATE REQUEST-----\n'
CSR_END='\n-----END CERTIFICATE REQUEST-----'
DH_BEGIN='-----BEGIN DH PARAMETERS-----\n'
DH_END='\n-----END DH PARAMETERS-----'
OVPN_BEGIN = '-----BEGIN OpenVPN Static key V{0}-----\n'
OVPN_END = '\n-----END OpenVPN Static key V{0}-----'
OPENSSH_KEY_BEGIN='-----BEGIN OPENSSH PRIVATE KEY-----\n'
OPENSSH_KEY_END='\n-----END OPENSSH PRIVATE KEY-----'
# Print functions
encoding_map = {
'PEM': serialization.Encoding.PEM,
'OpenSSH': serialization.Encoding.OpenSSH
}
public_format_map = {
'SubjectPublicKeyInfo': serialization.PublicFormat.SubjectPublicKeyInfo,
'OpenSSH': serialization.PublicFormat.OpenSSH
}
private_format_map = {
'PKCS8': serialization.PrivateFormat.PKCS8,
'OpenSSH': serialization.PrivateFormat.OpenSSH
}
hash_map = {
'sha256': hashes.SHA256,
'sha384': hashes.SHA384,
'sha512': hashes.SHA512,
}
def get_certificate_fingerprint(cert, hash):
hash_algorithm = hash_map[hash]()
fp = cert.fingerprint(hash_algorithm)
return fp.hex(':').upper()
def encode_certificate(cert):
return cert.public_bytes(encoding=serialization.Encoding.PEM).decode('utf-8')
def encode_public_key(cert, encoding='PEM', key_format='SubjectPublicKeyInfo'):
if encoding not in encoding_map:
encoding = 'PEM'
if key_format not in public_format_map:
key_format = 'SubjectPublicKeyInfo'
return cert.public_bytes(
encoding=encoding_map[encoding],
format=public_format_map[key_format]).decode('utf-8')
def encode_private_key(private_key, encoding='PEM', key_format='PKCS8', passphrase=None):
if encoding not in encoding_map:
encoding = 'PEM'
if key_format not in private_format_map:
key_format = 'PKCS8'
encryption = serialization.NoEncryption() if not passphrase else serialization.BestAvailableEncryption(bytes(passphrase, 'utf-8'))
return private_key.private_bytes(
encoding=encoding_map[encoding],
format=private_format_map[key_format],
encryption_algorithm=encryption).decode('utf-8')
def encode_dh_parameters(dh_parameters):
return dh_parameters.parameter_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.ParameterFormat.PKCS3).decode('utf-8')
# EC Helper
def get_elliptic_curve(size):
curve_func = None
name = f'SECP{size}R1'
if hasattr(ec, name):
curve_func = getattr(ec, name)
else:
curve_func = ec.SECP256R1() # Default to SECP256R1
return curve_func()
# Creation functions
def create_private_key(key_type, key_size=None):
private_key = None
if key_type == 'rsa':
private_key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)
elif key_type == 'dsa':
private_key = dsa.generate_private_key(key_size=key_size)
elif key_type == 'ec':
curve = get_elliptic_curve(key_size)
private_key = ec.generate_private_key(curve)
return private_key
def create_certificate_request(subject, private_key, subject_alt_names=[]):
subject_obj = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, subject['country']),
x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, subject['state']),
x509.NameAttribute(NameOID.LOCALITY_NAME, subject['locality']),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, subject['organization']),
x509.NameAttribute(NameOID.COMMON_NAME, subject['common_name'])])
builder = x509.CertificateSigningRequestBuilder() \
.subject_name(subject_obj)
if subject_alt_names:
alt_names = []
for obj in subject_alt_names:
if isinstance(obj, ipaddress.IPv4Address) or isinstance(obj, ipaddress.IPv6Address):
alt_names.append(x509.IPAddress(obj))
elif isinstance(obj, str):
alt_names.append(x509.RFC822Name(obj) if '@' in obj else x509.DNSName(obj))
if alt_names:
builder = builder.add_extension(x509.SubjectAlternativeName(alt_names), critical=False)
return builder.sign(private_key, hashes.SHA256())
def add_key_identifier(ca_cert):
try:
ski_ext = ca_cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier)
return x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(ski_ext.value)
except:
return x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_cert.public_key())
def create_certificate(cert_req, ca_cert, ca_private_key, valid_days=365, cert_type='server', is_ca=False, is_sub_ca=False):
ext_key_usage = []
if is_ca:
ext_key_usage = [ExtendedKeyUsageOID.CLIENT_AUTH, ExtendedKeyUsageOID.SERVER_AUTH]
elif cert_type == 'client':
ext_key_usage = [ExtendedKeyUsageOID.CLIENT_AUTH]
elif cert_type == 'server':
ext_key_usage = [ExtendedKeyUsageOID.SERVER_AUTH]
builder = x509.CertificateBuilder() \
.subject_name(cert_req.subject) \
.issuer_name(ca_cert.subject) \
.public_key(cert_req.public_key()) \
.serial_number(x509.random_serial_number()) \
.not_valid_before(datetime.datetime.utcnow()) \
.not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=int(valid_days)))
builder = builder.add_extension(x509.BasicConstraints(ca=is_ca, path_length=0 if is_sub_ca else None), critical=True)
builder = builder.add_extension(x509.KeyUsage(
digital_signature=True,
content_commitment=False,
key_encipherment=False,
data_encipherment=False,
key_agreement=False,
key_cert_sign=is_ca,
crl_sign=is_ca,
encipher_only=False,
decipher_only=False), critical=True)
builder = builder.add_extension(x509.ExtendedKeyUsage(ext_key_usage), critical=False)
builder = builder.add_extension(x509.SubjectKeyIdentifier.from_public_key(cert_req.public_key()), critical=False)
if not is_ca or is_sub_ca:
builder = builder.add_extension(add_key_identifier(ca_cert), critical=False)
for ext in cert_req.extensions:
builder = builder.add_extension(ext.value, critical=False)
return builder.sign(ca_private_key, hashes.SHA256())
def create_certificate_revocation_list(ca_cert, ca_private_key, serial_numbers=[]):
if not serial_numbers:
return False
builder = x509.CertificateRevocationListBuilder() \
.issuer_name(ca_cert.subject) \
.last_update(datetime.datetime.today()) \
.next_update(datetime.datetime.today() + datetime.timedelta(1, 0, 0))
for serial_number in serial_numbers:
revoked_cert = x509.RevokedCertificateBuilder() \
.serial_number(serial_number) \
.revocation_date(datetime.datetime.today()) \
.build()
builder = builder.add_revoked_certificate(revoked_cert)
return builder.sign(private_key=ca_private_key, algorithm=hashes.SHA256())
def create_dh_parameters(bits=2048):
if not bits or bits < 512:
print("Invalid DH parameter key size")
return False
return dh.generate_parameters(generator=2, key_size=int(bits))
# Wrap functions
def wrap_public_key(raw_data):
return KEY_PUB_BEGIN + raw_data + KEY_PUB_END
def wrap_private_key(raw_data, passphrase=None):
return (KEY_ENC_BEGIN if passphrase else KEY_BEGIN) + raw_data + (KEY_ENC_END if passphrase else KEY_END)
def wrap_openssh_public_key(raw_data, type):
return f'{type} {raw_data}'
def wrap_openssh_private_key(raw_data):
return OPENSSH_KEY_BEGIN + raw_data + OPENSSH_KEY_END
def wrap_certificate_request(raw_data):
return CSR_BEGIN + raw_data + CSR_END
def wrap_certificate(raw_data):
return CERT_BEGIN + raw_data + CERT_END
def wrap_crl(raw_data):
return CRL_BEGIN + raw_data + CRL_END
def wrap_dh_parameters(raw_data):
return DH_BEGIN + raw_data + DH_END
def wrap_openvpn_key(raw_data, version='1'):
return OVPN_BEGIN.format(version) + raw_data + OVPN_END.format(version)
# Load functions
def load_public_key(raw_data, wrap_tags=True):
if wrap_tags:
raw_data = wrap_public_key(raw_data)
try:
return serialization.load_pem_public_key(bytes(raw_data, 'utf-8'))
except ValueError:
return False
def load_private_key(raw_data, passphrase=None, wrap_tags=True):
if wrap_tags:
raw_data = wrap_private_key(raw_data, passphrase)
if passphrase is not None:
passphrase = bytes(passphrase, 'utf-8')
try:
return serialization.load_pem_private_key(bytes(raw_data, 'utf-8'), password=passphrase)
- except ValueError:
+ except (ValueError, TypeError):
return False
def load_openssh_public_key(raw_data, type):
try:
return serialization.load_ssh_public_key(bytes(f'{type} {raw_data}', 'utf-8'))
except ValueError:
return False
def load_openssh_private_key(raw_data, passphrase=None, wrap_tags=True):
if wrap_tags:
raw_data = wrap_openssh_private_key(raw_data)
try:
return serialization.load_ssh_private_key(bytes(raw_data, 'utf-8'), password=passphrase)
except ValueError:
return False
def load_certificate_request(raw_data, wrap_tags=True):
if wrap_tags:
raw_data = wrap_certificate_request(raw_data)
try:
return x509.load_pem_x509_csr(bytes(raw_data, 'utf-8'))
except ValueError:
return False
def load_certificate(raw_data, wrap_tags=True):
if wrap_tags:
raw_data = wrap_certificate(raw_data)
try:
return x509.load_pem_x509_certificate(bytes(raw_data, 'utf-8'))
except ValueError:
return False
def load_crl(raw_data, wrap_tags=True):
if wrap_tags:
raw_data = wrap_crl(raw_data)
try:
return x509.load_pem_x509_crl(bytes(raw_data, 'utf-8'))
except ValueError:
return False
def load_dh_parameters(raw_data, wrap_tags=True):
if wrap_tags:
raw_data = wrap_dh_parameters(raw_data)
try:
return serialization.load_pem_parameters(bytes(raw_data, 'utf-8'))
except ValueError:
return False
# Verify
def is_ca_certificate(cert):
if not cert:
return False
try:
ext = cert.extensions.get_extension_for_oid(ExtensionOID.BASIC_CONSTRAINTS)
return ext.value.ca
except ExtensionNotFound:
return False
def verify_certificate(cert, ca_cert):
# Verify certificate was signed by specified CA
if ca_cert.subject != cert.issuer:
return False
ca_public_key = ca_cert.public_key()
try:
if isinstance(ca_public_key, rsa.RSAPublicKeyWithSerialization):
ca_public_key.verify(
cert.signature,
cert.tbs_certificate_bytes,
padding=padding.PKCS1v15(),
algorithm=cert.signature_hash_algorithm)
elif isinstance(ca_public_key, dsa.DSAPublicKeyWithSerialization):
ca_public_key.verify(
cert.signature,
cert.tbs_certificate_bytes,
algorithm=cert.signature_hash_algorithm)
elif isinstance(ca_public_key, ec.EllipticCurvePublicKeyWithSerialization):
ca_public_key.verify(
cert.signature,
cert.tbs_certificate_bytes,
signature_algorithm=ec.ECDSA(cert.signature_hash_algorithm))
else:
return False # We cannot verify it
return True
except InvalidSignature:
return False
def verify_crl(crl, ca_cert):
# Verify CRL was signed by specified CA
if ca_cert.subject != crl.issuer:
return False
ca_public_key = ca_cert.public_key()
try:
if isinstance(ca_public_key, rsa.RSAPublicKeyWithSerialization):
ca_public_key.verify(
crl.signature,
crl.tbs_certlist_bytes,
padding=padding.PKCS1v15(),
algorithm=crl.signature_hash_algorithm)
elif isinstance(ca_public_key, dsa.DSAPublicKeyWithSerialization):
ca_public_key.verify(
crl.signature,
crl.tbs_certlist_bytes,
algorithm=crl.signature_hash_algorithm)
elif isinstance(ca_public_key, ec.EllipticCurvePublicKeyWithSerialization):
ca_public_key.verify(
crl.signature,
crl.tbs_certlist_bytes,
signature_algorithm=ec.ECDSA(crl.signature_hash_algorithm))
else:
return False # We cannot verify it
return True
except InvalidSignature:
return False
def verify_ca_chain(sorted_names, pki_node):
if len(sorted_names) == 1: # Single cert, no chain
return True
for name in sorted_names:
cert = load_certificate(pki_node[name]['certificate'])
verified = False
for ca_name in sorted_names:
if name == ca_name:
continue
ca_cert = load_certificate(pki_node[ca_name]['certificate'])
if verify_certificate(cert, ca_cert):
verified = True
break
if not verified and name != sorted_names[-1]:
# Only permit top-most certificate to fail verify (e.g. signed by public CA not explicitly in chain)
return False
return True
# Certificate chain
def find_parent(cert, ca_certs):
for ca_cert in ca_certs:
if verify_certificate(cert, ca_cert):
return ca_cert
return None
def find_chain(cert, ca_certs):
remaining = ca_certs.copy()
chain = [cert]
while remaining:
parent = find_parent(chain[-1], remaining)
if parent is None:
# No parent in the list of remaining certificates or there's a circular dependency
break
elif parent == chain[-1]:
# Self-signed: must be root CA (end of chain)
break
else:
remaining.remove(parent)
chain.append(parent)
return chain
def sort_ca_chain(ca_names, pki_node):
def ca_cmp(ca_name1, ca_name2, pki_node):
cert1 = load_certificate(pki_node[ca_name1]['certificate'])
cert2 = load_certificate(pki_node[ca_name2]['certificate'])
if verify_certificate(cert1, cert2): # cert1 is child of cert2
return -1
return 1
from functools import cmp_to_key
return sorted(ca_names, key=cmp_to_key(lambda cert1, cert2: ca_cmp(cert1, cert2, pki_node)))
diff --git a/src/op_mode/pki.py b/src/op_mode/pki.py
index 84b080023..b1a42d6c3 100755
--- a/src/op_mode/pki.py
+++ b/src/op_mode/pki.py
@@ -1,1096 +1,1105 @@
#!/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
from cryptography import x509
from cryptography.x509.oid import ExtendedKeyUsageOID
from vyos.config import Config
from vyos.config import config_dict_mangle_acme
from vyos.pki import encode_certificate, encode_public_key, encode_private_key, encode_dh_parameters
from vyos.pki import get_certificate_fingerprint
from vyos.pki import create_certificate, create_certificate_request, create_certificate_revocation_list
from vyos.pki import create_private_key
from vyos.pki import create_dh_parameters
from vyos.pki import load_certificate, load_certificate_request, load_private_key
from vyos.pki import load_crl, load_dh_parameters, load_public_key
from vyos.pki import verify_certificate
from vyos.utils.io import ask_input
from vyos.utils.io import ask_yes_no
from vyos.utils.misc import install_into_config
from vyos.utils.process import cmd
CERT_REQ_END = '-----END CERTIFICATE REQUEST-----'
auth_dir = '/config/auth'
# Helper Functions
conf = Config()
def get_default_values():
# Fetch default x509 values
base = ['pki', 'x509', 'default']
x509_defaults = conf.get_config_dict(base, key_mangling=('-', '_'),
no_tag_node_value_mangle=True,
get_first_key=True,
with_recursive_defaults=True)
return x509_defaults
def get_config_ca_certificate(name=None):
# Fetch ca certificates from config
base = ['pki', 'ca']
if not conf.exists(base):
return False
if name:
base = base + [name]
if not conf.exists(base + ['private', 'key']) or not conf.exists(base + ['certificate']):
return False
return conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
def get_config_certificate(name=None):
# Get certificates from config
base = ['pki', 'certificate']
if not conf.exists(base):
return False
if name:
base = base + [name]
if not conf.exists(base + ['private', 'key']) or not conf.exists(base + ['certificate']):
return False
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 = {}
subject['country'] = ask_input('Enter country code:', default=default_values['country'])
subject['state'] = ask_input('Enter state:', default=default_values['state'])
subject['locality'] = ask_input('Enter locality:', default=default_values['locality'])
subject['organization'] = ask_input('Enter organization name:', default=default_values['organization'])
subject['common_name'] = ask_input('Enter common name:', default='vyos.io')
subject_alt_names = None
if ask_san and ask_yes_no('Do you want to configure Subject Alternative Names?'):
print("Enter alternative names in a comma separate list, example: ipv4:1.1.1.1,ipv6:fe80::1,dns:vyos.net,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):
+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
- passphrase = ask_input('Enter private key passphrase: ') or 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: {path}')
+ 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):
+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
- passphrase = ask_input('Enter private key passphrase: ') or 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: {path}')
+ 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):
+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
- passphrase = ask_input('Enter private key passphrase: ') or 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: {path}')
+ 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):
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):
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))
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):
headers = ['CA Name', 'Updated', 'Revokes']
data = []
certs = get_config_ca_certificate()
if certs:
for cert_name, cert_dict in certs.items():
if name and name != cert_name:
continue
if 'crl' not in cert_dict:
continue
crls = cert_dict['crl']
if isinstance(crls, str):
crls = [crls]
for crl_data in cert_dict['crl']:
crl = load_crl(crl_data)
if not crl:
continue
if name and pem:
print(encode_certificate(crl))
continue
certs = get_revoked_by_serial_numbers([revoked.serial_number for revoked in crl])
data.append([cert_name, crl.last_update, ", ".join(certs)])
if name and pem:
return
print("Certificate Revocation Lists:")
print(tabulate.tabulate(data, headers))
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--action', help='PKI action', required=True)
# X509
parser.add_argument('--ca', help='Certificate Authority', required=False)
parser.add_argument('--certificate', help='Certificate', required=False)
parser.add_argument('--crl', help='Certificate Revocation List', required=False)
parser.add_argument('--sign', help='Sign certificate with specified CA', required=False)
parser.add_argument('--self-sign', help='Self-sign the certificate', action='store_true')
parser.add_argument('--pem', help='Output using PEM encoding', action='store_true')
parser.add_argument('--fingerprint', help='Show fingerprint and exit', action='store')
# SSH
parser.add_argument('--ssh', help='SSH Key', required=False)
# DH
parser.add_argument('--dh', help='DH Parameters', required=False)
# Key pair
parser.add_argument('--keypair', help='Key pair', required=False)
# OpenVPN
parser.add_argument('--openvpn', help='OpenVPN TLS key', required=False)
# WireGuard
parser.add_argument('--wireguard', help='Wireguard', action='store_true')
group = parser.add_mutually_exclusive_group()
group.add_argument('--key', help='Wireguard key pair', action='store_true', required=False)
group.add_argument('--psk', help='Wireguard pre shared key', action='store_true', required=False)
parser.add_argument('--interface', help='Install generated keys into running-config for named interface', action='store')
parser.add_argument('--peer', help='Install generated keys into running-config for peer', action='store')
# Global
parser.add_argument('--file', help='Write generated keys into specified filename', action='store_true')
parser.add_argument('--install', help='Install generated keys into running-config', action='store_true')
parser.add_argument('--filename', help='Write certificate into specified filename', action='store')
parser.add_argument('--key-filename', help='Write key into specified filename', action='store')
+ 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()
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)
+ 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)
+ 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)
+ 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)
diff --git a/src/services/vyos-http-api-server b/src/services/vyos-http-api-server
index 7f5233c6b..97633577d 100755
--- a/src/services/vyos-http-api-server
+++ b/src/services/vyos-http-api-server
@@ -1,972 +1,1034 @@
#!/usr/share/vyos-http-api-tools/bin/python3
#
# Copyright (C) 2019-2024 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import sys
import grp
import copy
import json
import logging
import signal
import traceback
import threading
from enum import Enum
from time import sleep
from typing import List, Union, Callable, Dict, Self
from fastapi import FastAPI, Depends, Request, Response, HTTPException
from fastapi import BackgroundTasks
from fastapi.responses import HTMLResponse
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
from pydantic import BaseModel, StrictStr, validator, model_validator
from starlette.middleware.cors import CORSMiddleware
from starlette.datastructures import FormData
from starlette.formparsers import FormParser, MultiPartParser
from multipart.multipart import parse_options_header
from uvicorn import Config as UvicornConfig
from uvicorn import Server as UvicornServer
from ariadne.asgi import GraphQL
from vyos.config import Config
from vyos.configtree import ConfigTree
from vyos.configdiff import get_config_diff
from vyos.configsession import ConfigSession
from vyos.configsession import ConfigSessionError
from vyos.defaults import api_config_state
import api.graphql.state
CFG_GROUP = 'vyattacfg'
debug = True
logger = logging.getLogger(__name__)
logs_handler = logging.StreamHandler()
logger.addHandler(logs_handler)
if debug:
logger.setLevel(logging.DEBUG)
else:
logger.setLevel(logging.INFO)
# Giant lock!
lock = threading.Lock()
def load_server_config():
with open(api_config_state) as f:
config = json.load(f)
return config
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 error(code, msg):
resp = {"success": False, "error": msg, "data": None}
resp = json.dumps(resp)
return HTMLResponse(resp, status_code=code)
def success(data):
resp = {"success": True, "data": data, "error": None}
resp = json.dumps(resp)
return HTMLResponse(resp)
# Pydantic models for validation
# Pydantic will cast when possible, so use StrictStr
# validators added as needed for additional constraints
# schema_extra adds anotations to OpenAPI, to add examples
class ApiModel(BaseModel):
key: StrictStr
class BasePathModel(BaseModel):
op: StrictStr
path: List[StrictStr]
@validator("path")
def check_non_empty(cls, path):
if not len(path) > 0:
raise ValueError('path must be non-empty')
return path
class BaseConfigureModel(BasePathModel):
value: StrictStr = None
class ConfigureModel(ApiModel, BaseConfigureModel):
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "set | delete | comment",
"path": ['config', 'mode', 'path'],
}
}
class ConfigureListModel(ApiModel):
commands: List[BaseConfigureModel]
class Config:
schema_extra = {
"example": {
"key": "id_key",
"commands": "list of commands",
}
}
class BaseConfigSectionModel(BasePathModel):
section: Dict
class ConfigSectionModel(ApiModel, BaseConfigSectionModel):
pass
class ConfigSectionListModel(ApiModel):
commands: List[BaseConfigSectionModel]
class BaseConfigSectionTreeModel(BaseModel):
op: StrictStr
mask: Dict
config: Dict
class ConfigSectionTreeModel(ApiModel, BaseConfigSectionTreeModel):
pass
class RetrieveModel(ApiModel):
op: StrictStr
path: List[StrictStr]
configFormat: StrictStr = None
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "returnValue | returnValues | exists | showConfig",
"path": ['config', 'mode', 'path'],
"configFormat": "json (default) | json_ast | raw",
}
}
class ConfigFileModel(ApiModel):
op: StrictStr
file: StrictStr = None
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "save | load",
"file": "filename",
}
}
class ImageOp(str, Enum):
add = "add"
delete = "delete"
show = "show"
set_default = "set_default"
class ImageModel(ApiModel):
op: ImageOp
url: StrictStr = None
name: StrictStr = None
@model_validator(mode='after')
def check_data(self) -> Self:
if self.op == 'add':
if not self.url:
raise ValueError("Missing required field \"url\"")
elif self.op in ['delete', 'set_default']:
if not self.name:
raise ValueError("Missing required field \"name\"")
return self
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "add | delete | show | set_default",
"url": "imagelocation",
"name": "imagename",
}
}
+class ImportPkiModel(ApiModel):
+ op: StrictStr
+ path: List[StrictStr]
+ passphrase: StrictStr = None
+
+ class Config:
+ schema_extra = {
+ "example": {
+ "key": "id_key",
+ "op": "import_pki",
+ "path": ["op", "mode", "path"],
+ "passphrase": "passphrase",
+ }
+ }
+
+
class ContainerImageModel(ApiModel):
op: StrictStr
name: StrictStr = None
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "add | delete | show",
"name": "imagename",
}
}
class GenerateModel(ApiModel):
op: StrictStr
path: List[StrictStr]
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "generate",
"path": ["op", "mode", "path"],
}
}
class ShowModel(ApiModel):
op: StrictStr
path: List[StrictStr]
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "show",
"path": ["op", "mode", "path"],
}
}
class RebootModel(ApiModel):
op: StrictStr
path: List[StrictStr]
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "reboot",
"path": ["op", "mode", "path"],
}
}
class ResetModel(ApiModel):
op: StrictStr
path: List[StrictStr]
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "reset",
"path": ["op", "mode", "path"],
}
}
class PoweroffModel(ApiModel):
op: StrictStr
path: List[StrictStr]
class Config:
schema_extra = {
"example": {
"key": "id_key",
"op": "poweroff",
"path": ["op", "mode", "path"],
}
}
class Success(BaseModel):
success: bool
data: Union[str, bool, Dict]
error: str
class Error(BaseModel):
success: bool = False
data: Union[str, bool, Dict]
error: str
responses = {
200: {'model': Success},
400: {'model': Error},
422: {'model': Error, 'description': 'Validation Error'},
500: {'model': Error}
}
def auth_required(data: ApiModel):
key = data.key
api_keys = app.state.vyos_keys
key_id = check_auth(api_keys, key)
if not key_id:
raise HTTPException(status_code=401, detail="Valid API key is required")
app.state.vyos_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):
_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 form(self) -> 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, options = parse_options_header(content_type_header)
if content_type == b"multipart/form-data":
multipart_parser = MultiPartParser(self.orig_headers, self.stream())
self._form = await multipart_parser.parse()
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
logger.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
else:
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):
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
app = FastAPI(debug=True,
title="VyOS API",
version="0.1.0",
responses={**responses},
dependencies=[Depends(auth_required)])
app.router.route_class = MultipartRoute
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
return error(400, str(exc.errors()[0]))
self_ref_msg = "Requested HTTP API server configuration change; commit will be called in the background"
def call_commit(s: ConfigSession):
try:
s.commit()
except ConfigSessionError as e:
s.discard()
if app.state.vyos_debug:
logger.warning(f"ConfigSessionError:\n {traceback.format_exc()}")
else:
logger.warning(f"ConfigSessionError: {e}")
def _configure_op(data: Union[ConfigureModel, ConfigureListModel,
ConfigSectionModel, ConfigSectionListModel,
ConfigSectionTreeModel],
request: Request, background_tasks: BackgroundTasks):
session = app.state.vyos_session
env = session.get_session_env()
endpoint = request.url.path
# 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 app.state.vyos_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, session)
msg = self_ref_msg
else:
session.commit()
logger.info(f"Configuration modified via HTTP API using key '{app.state.vyos_id}'")
except ConfigSessionError as e:
session.discard()
status = 400
if app.state.vyos_debug:
logger.critical(f"ConfigSessionError:\n {traceback.format_exc()}")
error_msg = str(e)
except Exception as e:
session.discard()
logger.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:]
+
@app.post('/configure')
def configure_op(data: Union[ConfigureModel,
ConfigureListModel],
request: Request, background_tasks: BackgroundTasks):
return _configure_op(data, request, background_tasks)
@app.post('/configure-section')
def configure_section_op(data: Union[ConfigSectionModel,
ConfigSectionListModel,
ConfigSectionTreeModel],
request: Request, background_tasks: BackgroundTasks):
return _configure_op(data, request, background_tasks)
@app.post("/retrieve")
async def retrieve_op(data: RetrieveModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
@app.post('/config-file')
def config_file_op(data: ConfigFileModel, background_tasks: BackgroundTasks):
session = app.state.vyos_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, session)
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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(msg)
@app.post('/image')
def image_op(data: ImageModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
@app.post('/container-image')
def container_image_op(data: ContainerImageModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
@app.post('/generate')
def generate_op(data: GenerateModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
@app.post('/show')
def show_op(data: ShowModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
@app.post('/reboot')
def reboot_op(data: RebootModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
@app.post('/reset')
def reset_op(data: ResetModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
+@app.post('/import-pki')
+def import_pki(data: ImportPkiModel):
+ session = app.state.vyos_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 as e:
+ logger.critical(traceback.format_exc())
+ return error(500, "An internal error occured. Check the logs for details.")
+ finally:
+ lock.release()
+
+ return success(res)
+
@app.post('/poweroff')
def poweroff_op(data: PoweroffModel):
session = app.state.vyos_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 as e:
logger.critical(traceback.format_exc())
return error(500, "An internal error occured. Check the logs for details.")
return success(res)
###
# GraphQL integration
###
def graphql_init(app: FastAPI = app):
from api.graphql.libs.token_auth import get_user_context
api.graphql.state.init()
api.graphql.state.settings['app'] = app
# import after initializaion of state
from api.graphql.bindings import generate_schema
schema = generate_schema()
in_spec = app.state.vyos_introspection
if app.state.vyos_origins:
origins = app.state.vyos_origins
app.add_route('/graphql', CORSMiddleware(GraphQL(schema,
context_value=get_user_context,
debug=True,
introspection=in_spec),
allow_origins=origins,
allow_methods=("GET", "POST", "OPTIONS"),
allow_headers=("Authorization",)))
else:
app.add_route('/graphql', GraphQL(schema,
context_value=get_user_context,
debug=True,
introspection=in_spec))
###
# Modify uvicorn to allow reloading server within the configsession
###
server = None
shutdown = False
class ApiServerConfig(UvicornConfig):
pass
class ApiServer(UvicornServer):
def install_signal_handlers(self):
pass
def reload_handler(signum, frame):
global server
logger.debug('Reload signal received...')
if server is not None:
server.handle_exit(signum, frame)
server = None
logger.info('Server stopping for reload...')
else:
logger.warning('Reload called for non-running server...')
def shutdown_handler(signum, frame):
global shutdown
logger.debug('Shutdown signal received...')
server.handle_exit(signum, frame)
logger.info('Server shutdown...')
shutdown = True
def flatten_keys(d: dict) -> list[dict]:
keys_list = []
for el in list(d['keys'].get('id', {})):
key = d['keys']['id'][el].get('key', '')
if key:
keys_list.append({'id': el, 'key': key})
return keys_list
def initialization(session: ConfigSession, app: FastAPI = app):
global server
try:
server_config = load_server_config()
except Exception as e:
logger.critical(f'Failed to load the HTTP API server config: {e}')
sys.exit(1)
app.state.vyos_session = session
app.state.vyos_keys = []
if 'keys' in server_config:
app.state.vyos_keys = flatten_keys(server_config)
app.state.vyos_debug = bool('debug' in server_config)
app.state.vyos_strict = bool('strict' in server_config)
app.state.vyos_origins = server_config.get('cors', {}).get('allow_origin', [])
if 'graphql' in server_config:
app.state.vyos_graphql = True
if isinstance(server_config['graphql'], dict):
if 'introspection' in server_config['graphql']:
app.state.vyos_introspection = True
else:
app.state.vyos_introspection = False
# default values if not set explicitly
app.state.vyos_auth_type = server_config['graphql']['authentication']['type']
app.state.vyos_token_exp = server_config['graphql']['authentication']['expiration']
app.state.vyos_secret_len = server_config['graphql']['authentication']['secret_length']
else:
app.state.vyos_graphql = False
if app.state.vyos_graphql:
graphql_init(app)
config = ApiServerConfig(app, uds="/run/api.sock", proxy_headers=True)
server = ApiServer(config)
def run_server():
try:
server.run()
except OSError as e:
logger.critical(e)
sys.exit(1)
if __name__ == '__main__':
# systemd's user and group options don't work, do it by hand here,
# else no one else will be able to commit
cfg_group = grp.getgrnam(CFG_GROUP)
os.setgid(cfg_group.gr_gid)
# Need to set file permissions to 775 too so that every vyattacfg group member
# has write access to the running config
os.umask(0o002)
signal.signal(signal.SIGHUP, reload_handler)
signal.signal(signal.SIGTERM, shutdown_handler)
config_session = ConfigSession(os.getpid())
while True:
logger.debug('Enter main loop...')
if shutdown:
break
if server is None:
initialization(config_session)
server.run()
sleep(1)