diff --git a/op-mode-definitions/pki.xml.in b/op-mode-definitions/pki.xml.in
index a1c55dcf4..346febec0 100644
--- a/op-mode-definitions/pki.xml.in
+++ b/op-mode-definitions/pki.xml.in
@@ -1,400 +1,535 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="generate">
<children>
<node name="pki">
<properties>
<help>Generate PKI certificates and keys</help>
</properties>
<children>
<node name="ca">
<properties>
<help>Generate CA certificate</help>
</properties>
<children>
<tagNode name="sign">
<properties>
<help>Sign generated CA certificate with another specified CA certificate</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated CA certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$7" --sign "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated CA certificate into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$7" --sign "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "noname" --sign "$5"</command>
</tagNode>
<tagNode name="file">
<properties>
<help>Write generated CA certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated CA certificate into running configuration</help>
<completionHelp>
<list><CA name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ca "noname"</command>
</node>
<node name="certificate">
<properties>
<help>Generate certificate request</help>
</properties>
<children>
<node name="self-signed">
<properties>
<help>Generate self-signed certificate</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated self-signed certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$6" --self-sign --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated self-signed certificate into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$6" --self-sign --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname" --self-sign</command>
</node>
<tagNode name="sign">
<properties>
<help>Sign generated certificate with specified CA certificate</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated signed certificate into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$7" --sign "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated signed certificate into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$7" --sign "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname" --sign "$5"</command>
</tagNode>
<tagNode name="file">
<properties>
<help>Write generated certificate request and key into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated certificate private key into running configuration</help>
<completionHelp>
<list><certificate name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --certificate "noname"</command>
</node>
<tagNode name="crl">
<properties>
<help>Generate CRL for specified CA certificate</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated CRL into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --crl "$4" --file</command>
</tagNode>
<leafNode name="install">
<properties>
<help>Commands for installing generated CRL into running configuration</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --crl "$4" --install</command>
</leafNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --crl "$4"</command>
</tagNode>
<node name="dh">
<properties>
<help>Generate DH parameters</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated DH parameters into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --dh "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated DH parameters into running configuration</help>
<completionHelp>
<list><DH name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --dh "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --dh "noname"</command>
</node>
<node name="key-pair">
<properties>
<help>Generate a key pair</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated key pair into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --keypair "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated key pair into running configuration</help>
<completionHelp>
<list><key name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --keypair "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --keypair "noname"</command>
</node>
<node name="openvpn">
<properties>
<help>Generate OpenVPN keys</help>
</properties>
<children>
<node name="shared-secret">
<properties>
<help>Generate OpenVPN shared secret key</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated OpenVPN shared secret key into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --openvpn "$6" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated OpenVPN shared secret key into running configuration</help>
<completionHelp>
<list><key name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --openvpn "$6" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --openvpn "noname"</command>
</node>
</children>
</node>
<node name="ssh-key">
<properties>
<help>Generate SSH key</help>
</properties>
<children>
<tagNode name="file">
<properties>
<help>Write generated SSH keys into the specified filename</help>
<completionHelp>
<list><filename></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ssh "$5" --file</command>
</tagNode>
<tagNode name="install">
<properties>
<help>Commands for installing generated SSH key into running configuration</help>
<completionHelp>
<list><key name></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ssh "$5" --install</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --ssh "noname"</command>
</node>
<node name="wireguard">
<properties>
<help>Generate WireGuard keys</help>
</properties>
<children>
<node name="key-pair">
<properties>
<help>Generate WireGuard public/private key-pair</help>
</properties>
<children>
<node name="install">
<properties>
<help>Generate CLI commands to install WireGuard key to configuration</help>
</properties>
<children>
<tagNode name="interface">
<properties>
<help>WireGuard Interface used in install command</help>
<completionHelp>
<path>interfaces wireguard</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --key --interface "$7" --install</command>
</tagNode>
</children>
</node>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --key</command>
</node>
<node name="preshared-key">
<properties>
<help>Generate WireGuard pre-shared key</help>
</properties>
<children>
<node name="install">
<properties>
<help>Generate CLI commands to install WireGuard key to configuration</help>
</properties>
<children>
<tagNode name="interface">
<properties>
<help>WireGuard Interface used in install command</help>
<completionHelp>
<path>interfaces wireguard</path>
</completionHelp>
</properties>
<children>
<tagNode name="peer">
<properties>
<help>Interface used for install command</help>
<completionHelp>
<path>interfaces wireguard ${COMP_WORDS[COMP_CWORD-2]} peer</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --psk --interface "$7" --peer "$9" --install</command>
</tagNode>
</children>
</tagNode>
</children>
</node>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action generate --wireguard --psk</command>
</node>
</children>
</node>
</children>
</node>
</children>
</node>
+ <node name="import">
+ <properties>
+ <help>Import an object</help>
+ </properties>
+ <children>
+ <node name="pki">
+ <properties>
+ <help>Import file into PKI configuration</help>
+ </properties>
+ <children>
+ <tagNode name="ca">
+ <properties>
+ <help>Import CA certificate into PKI</help>
+ <completionHelp>
+ <list><name></list>
+ </completionHelp>
+ </properties>
+ <children>
+ <tagNode name="file">
+ <properties>
+ <help>Path to CA certificate file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --ca "$4" --filename "$6"</command>
+ </tagNode>
+ <tagNode name="key-file">
+ <properties>
+ <help>Path to private key file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --ca "$4" --key-filename "$6"</command>
+ </tagNode>
+ </children>
+ </tagNode>
+ <tagNode name="certificate">
+ <properties>
+ <help>Import certificate into PKI</help>
+ <completionHelp>
+ <list><name></list>
+ </completionHelp>
+ </properties>
+ <children>
+ <tagNode name="file">
+ <properties>
+ <help>Path to certificate file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --certificate "$4" --filename "$6"</command>
+ </tagNode>
+ <tagNode name="key-file">
+ <properties>
+ <help>Path to private key file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --certificate "$4" --key-filename "$6"</command>
+ </tagNode>
+ </children>
+ </tagNode>
+ <tagNode name="crl">
+ <properties>
+ <help>Import certificate revocation list into PKI</help>
+ <completionHelp>
+ <list><CA name></list>
+ </completionHelp>
+ </properties>
+ <children>
+ <tagNode name="file">
+ <properties>
+ <help>Path to CRL file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --crl "$4" --filename "$6"</command>
+ </tagNode>
+ </children>
+ </tagNode>
+ <tagNode name="dh">
+ <properties>
+ <help>Import DH parameters into PKI</help>
+ <completionHelp>
+ <list><name></list>
+ </completionHelp>
+ </properties>
+ <children>
+ <tagNode name="file">
+ <properties>
+ <help>Path to DH parameters file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --dh "$4" --filename "$6"</command>
+ </tagNode>
+ </children>
+ </tagNode>
+ <tagNode name="key-pair">
+ <properties>
+ <help>Import key pair into PKI</help>
+ <completionHelp>
+ <list><name></list>
+ </completionHelp>
+ </properties>
+ <children>
+ <tagNode name="public-file">
+ <properties>
+ <help>Path to public key file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --keypair "$4" --filename "$6"</command>
+ </tagNode>
+ <tagNode name="private-file">
+ <properties>
+ <help>Path to private key file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --keypair "$4" --key-filename "$6"</command>
+ </tagNode>
+ </children>
+ </tagNode>
+ <node name="openvpn">
+ <properties>
+ <help>Import OpenVPN keys into PKI</help>
+ </properties>
+ <children>
+ <tagNode name="shared-secret">
+ <properties>
+ <help>Import OpenVPN shared secret key into PKI</help>
+ <completionHelp>
+ <list><name></list>
+ </completionHelp>
+ </properties>
+ <children>
+ <tagNode name="file">
+ <properties>
+ <help>Path to shared secret key file</help>
+ </properties>
+ <command>sudo ${vyos_op_scripts_dir}/pki.py --action import --openvpn "$5" --filename "$7"</command>
+ </tagNode>
+ </children>
+ </tagNode>
+ </children>
+ </node>
+ </children>
+ </node>
+ </children>
+ </node>
<node name="show">
<children>
<node name="pki">
<properties>
<help>Show PKI x509 certificates</help>
</properties>
<children>
<leafNode name="ca">
<properties>
<help>Show x509 CA certificates</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --ca "all"</command>
</leafNode>
<tagNode name="ca">
<properties>
<help>Show x509 CA certificate by name</help>
<completionHelp>
<path>pki ca</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --ca "$4"</command>
</tagNode>
<leafNode name="certificate">
<properties>
<help>Show x509 certificates</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --certificate "all"</command>
</leafNode>
<tagNode name="certificate">
<properties>
<help>Show x509 certificate by name</help>
<completionHelp>
<path>pki certificate</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --certificate "$4"</command>
</tagNode>
<leafNode name="crl">
<properties>
<help>Show x509 certificate revocation lists</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show --crl "all"</command>
</leafNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/pki.py --action show</command>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/python/vyos/pki.py b/python/vyos/pki.py
index 0b916eaae..fd91fc9bf 100644
--- a/python/vyos/pki.py
+++ b/python/vyos/pki.py
@@ -1,359 +1,359 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021 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, ExtendedKeyUsageOID, 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}-----'
# 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
}
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.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_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:
+ 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:
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
# 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
diff --git a/src/op_mode/pki.py b/src/op_mode/pki.py
index bc7813052..193d97744 100755
--- a/src/op_mode/pki.py
+++ b/src/op_mode/pki.py
@@ -1,867 +1,1051 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021 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.configquery import ConfigTreeQuery
from vyos.configdict import dict_merge
from vyos.pki import encode_certificate, encode_public_key, encode_private_key, encode_dh_parameters
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, load_crl
+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.xml import defaults
from vyos.util import ask_input, ask_yes_no
from vyos.util import cmd
from vyos.util import install_into_config
CERT_REQ_END = '-----END CERTIFICATE REQUEST-----'
auth_dir = '/config/auth'
# Helper Functions
conf = ConfigTreeQuery()
def get_default_values():
# Fetch default x509 values
base = ['pki', 'x509', 'default']
x509_defaults = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
default_values = defaults(base)
x509_defaults = dict_merge(default_values, x509_defaults)
return x509_defaults
def get_config_ca_certificate(name=None):
# Fetch ca certificates from config
base = ['pki', 'ca']
if not conf.exists(base):
return False
if name:
base = base + [name]
if not conf.exists(base + ['private', 'key']) or not conf.exists(base + ['certificate']):
return False
return conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
def get_config_certificate(name=None):
# Get certificates from config
base = ['pki', 'certificate']
if not conf.exists(base):
return False
if name:
base = base + [name]
if not conf.exists(base + ['private', 'key']) or not conf.exists(base + ['certificate']):
return False
return conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
def get_certificate_ca(cert, ca_certs):
# Find CA certificate for given certificate
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):
+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 = ask_yes_no('Do you want to install the public key?', default=True)
+ 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 = ask_yes_no('Do you want to install the private key?', default=True)
+ install_private_key = not prompt or ask_yes_no('Do you want to install the private key?', default=True)
private_key_pem = encode_private_key(private_key, passphrase=passphrase)
if install_private_key:
install_private_pem = "".join(private_key_pem.strip().split("\n")[1:-1])
config_paths.append(f"pki key-pair {name} private key '{install_private_pem}'")
if passphrase:
config_paths.append(f"pki key-pair {name} private password-protected")
else:
print("Private key:")
print(private_key_pem)
install_into_config(conf, config_paths)
+def install_openvpn_key(name, key_data, key_version='1'):
+ config_paths = [
+ f"pki openvpn shared-secret {name} key '{key_data}'",
+ f"pki openvpn shared-secret {name} version '{key_version}'"
+ ]
+ install_into_config(conf, config_paths)
+
def install_wireguard_key(interface, private_key, public_key):
# Show conf commands for installing wireguard key pairs
from vyos.ifconfig import Section
if Section.section(interface) != 'wireguard':
print(f'"{interface}" is not a WireGuard interface name!')
exit(1)
# Check if we are running in a config session - if yes, we can directly write to the CLI
install_into_config(conf, [f"interfaces wireguard {interface} private-key '{private_key}'"])
print(f"Corresponding public-key to use on peer system is: '{public_key}'")
def install_wireguard_psk(interface, peer, psk):
from vyos.ifconfig import Section
if Section.section(interface) != 'wireguard':
print(f'"{interface}" is not a WireGuard interface name!')
exit(1)
# Check if we are running in a config session - if yes, we can directly write to the CLI
install_into_config(conf, [f"interfaces wireguard {interface} peer {peer} preshared-key '{psk}'"])
def ask_passphrase():
passphrase = None
print("Note: If you plan to use the generated key on this router, do not encrypt the private key.")
if ask_yes_no('Do you want to encrypt the private key with a passphrase?'):
passphrase = ask_input('Enter passphrase:')
return passphrase
def write_file(filename, contents):
full_path = os.path.join(auth_dir, filename)
directory = os.path.dirname(full_path)
if not os.path.exists(directory):
print('Failed to write file: directory does not exist')
return False
if os.path.exists(full_path) and not ask_yes_no('Do you want to overwrite the existing file?'):
return False
with open(full_path, 'w') as f:
f.write(contents)
print(f'File written to {full_path}')
# Generation functions
def generate_private_key():
key_type = ask_input('Enter private key type: [rsa, dsa, ec]', default='rsa', valid_responses=['rsa', 'dsa', 'ec'])
size_valid = []
size_default = 0
if key_type in ['rsa', 'dsa']:
size_default = 2048
size_valid = [512, 1024, 2048, 4096]
elif key_type == 'ec':
size_default = 256
size_valid = [224, 256, 384, 521]
size = ask_input('Enter private key bits:', default=size_default, numeric_only=True, valid_responses=size_valid)
return create_private_key(key_type, size), key_type
def parse_san_string(san_string):
if not san_string:
return None
output = []
san_split = san_string.strip().split(",")
for pair_str in san_split:
tag, value = pair_str.strip().split(":", 1)
if tag == 'ipv4':
output.append(ipaddress.IPv4Address(value))
elif tag == 'ipv6':
output.append(ipaddress.IPv6Address(value))
elif tag == 'dns':
output.append(value)
return output
def generate_certificate_request(private_key=None, key_type=None, return_request=False, name=None, install=False, file=False, ask_san=True):
if not private_key:
private_key, key_type = generate_private_key()
default_values = get_default_values()
subject = {}
subject['country'] = ask_input('Enter country code:', default=default_values['country'])
subject['state'] = ask_input('Enter state:', default=default_values['state'])
subject['locality'] = ask_input('Enter locality:', default=default_values['locality'])
subject['organization'] = ask_input('Enter organization name:', default=default_values['organization'])
subject['common_name'] = ask_input('Enter common name:', default='vyos.io')
subject_alt_names = None
if ask_san and ask_yes_no('Do you want to configure Subject Alternative Names?'):
print("Enter alternative names in a comma separate list, example: ipv4:1.1.1.1,ipv6:fe80::1,dns:vyos.net")
san_string = ask_input('Enter Subject Alternative Names:')
subject_alt_names = parse_san_string(san_string)
cert_req = create_certificate_request(subject, private_key, subject_alt_names)
if return_request:
return cert_req
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert_req))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
print("Certificate request:")
print(encode_certificate(cert_req) + "\n")
install_certificate(name, private_key=private_key, key_type=key_type, key_passphrase=passphrase, is_ca=False)
if file:
write_file(f'{name}.csr', encode_certificate(cert_req))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate(cert_req, ca_cert, ca_private_key, is_ca=False, is_sub_ca=False):
valid_days = ask_input('Enter how many days certificate will be valid:', default='365' if not is_ca else '1825', numeric_only=True)
cert_type = None
if not is_ca:
cert_type = ask_input('Enter certificate type: (client, server)', default='server', valid_responses=['client', 'server'])
return create_certificate(cert_req, ca_cert, ca_private_key, valid_days, cert_type, is_ca, is_sub_ca)
def generate_ca_certificate(name, install=False, file=False):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True, ask_san=False)
cert = generate_certificate(cert_req, cert_req, private_key, is_ca=True)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key, key_type, key_passphrase=passphrase, is_ca=True)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_ca_certificate_sign(name, ca_name, install=False, file=False):
ca_dict = get_config_ca_certificate(ca_name)
if not ca_dict:
print(f"CA certificate or private key for '{ca_name}' not found")
return None
ca_cert = load_certificate(ca_dict['certificate'])
if not ca_cert:
print("Failed to load signing CA certificate, aborting")
return None
ca_private = ca_dict['private']
ca_private_passphrase = None
if 'password_protected' in ca_private:
ca_private_passphrase = ask_input('Enter signing CA private key passphrase:')
ca_private_key = load_private_key(ca_private['key'], passphrase=ca_private_passphrase)
if not ca_private_key:
print("Failed to load signing CA private key, aborting")
return None
private_key = None
key_type = None
cert_req = None
if not ask_yes_no('Do you already have a certificate request?'):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True, ask_san=False)
else:
print("Paste certificate request and press enter:")
lines = []
curr_line = ''
while True:
curr_line = input().strip()
if not curr_line or curr_line == CERT_REQ_END:
break
lines.append(curr_line)
if not lines:
print("Aborted")
return None
wrap = lines[0].find('-----') < 0 # Only base64 pasted, add the CSR tags for parsing
cert_req = load_certificate_request("\n".join(lines), wrap)
if not cert_req:
print("Invalid certificate request")
return None
cert = generate_certificate(cert_req, ca_cert, ca_private_key, is_ca=True, is_sub_ca=True)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key, key_type, key_passphrase=passphrase, is_ca=True)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate_sign(name, ca_name, install=False, file=False):
ca_dict = get_config_ca_certificate(ca_name)
if not ca_dict:
print(f"CA certificate or private key for '{ca_name}' not found")
return None
ca_cert = load_certificate(ca_dict['certificate'])
if not ca_cert:
print("Failed to load CA certificate, aborting")
return None
ca_private = ca_dict['private']
ca_private_passphrase = None
if 'password_protected' in ca_private:
ca_private_passphrase = ask_input('Enter CA private key passphrase:')
ca_private_key = load_private_key(ca_private['key'], passphrase=ca_private_passphrase)
if not ca_private_key:
print("Failed to load CA private key, aborting")
return None
private_key = None
key_type = None
cert_req = None
if not ask_yes_no('Do you already have a certificate request?'):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True)
else:
print("Paste certificate request and press enter:")
lines = []
curr_line = ''
while True:
curr_line = input().strip()
if not curr_line or curr_line == CERT_REQ_END:
break
lines.append(curr_line)
if not lines:
print("Aborted")
return None
wrap = lines[0].find('-----') < 0 # Only base64 pasted, add the CSR tags for parsing
cert_req = load_certificate_request("\n".join(lines), wrap)
if not cert_req:
print("Invalid certificate request")
return None
cert = generate_certificate(cert_req, ca_cert, ca_private_key, is_ca=False)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key, key_type, key_passphrase=passphrase, is_ca=False)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate_selfsign(name, install=False, file=False):
private_key, key_type = generate_private_key()
cert_req = generate_certificate_request(private_key, key_type, return_request=True)
cert = generate_certificate(cert_req, cert_req, private_key, is_ca=False)
passphrase = ask_passphrase()
if not install and not file:
print(encode_certificate(cert))
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_certificate(name, cert, private_key=private_key, key_type=key_type, key_passphrase=passphrase, is_ca=False)
if file:
write_file(f'{name}.pem', encode_certificate(cert))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_certificate_revocation_list(ca_name, install=False, file=False):
ca_dict = get_config_ca_certificate(ca_name)
if not ca_dict:
print(f"CA certificate or private key for '{ca_name}' not found")
return None
ca_cert = load_certificate(ca_dict['certificate'])
if not ca_cert:
print("Failed to load CA certificate, aborting")
return None
ca_private = ca_dict['private']
ca_private_passphrase = None
if 'password_protected' in ca_private:
ca_private_passphrase = ask_input('Enter CA private key passphrase:')
ca_private_key = load_private_key(ca_private['key'], passphrase=ca_private_passphrase)
if not ca_private_key:
print("Failed to load CA private key, aborting")
return None
revoked_certs = get_config_revoked_certificates()
to_revoke = []
for cert_dict in revoked_certs:
if 'certificate' not in cert_dict:
continue
cert_data = cert_dict['certificate']
try:
cert = load_certificate(cert_data)
if cert.issuer == ca_cert.subject:
to_revoke.append(cert.serial_number)
except ValueError:
continue
if not to_revoke:
print("No revoked certificates to add to the CRL")
return None
crl = create_certificate_revocation_list(ca_cert, ca_private_key, to_revoke)
if not crl:
print("Failed to create CRL")
return None
if not install and not file:
print(encode_certificate(crl))
return None
if install:
install_crl(ca_name, crl)
if file:
write_file(f'{name}.crl', encode_certificate(crl))
def generate_ssh_keypair(name, install=False, file=False):
private_key, key_type = generate_private_key()
public_key = private_key.public_key()
passphrase = ask_passphrase()
if not install and not file:
print(encode_public_key(public_key, encoding='OpenSSH', key_format='OpenSSH'))
print("")
print(encode_private_key(private_key, encoding='PEM', key_format='OpenSSH', passphrase=passphrase))
return None
if install:
install_ssh_key(name, public_key, private_key, passphrase)
if file:
write_file(f'{name}.pem', encode_public_key(public_key, encoding='OpenSSH', key_format='OpenSSH'))
write_file(f'{name}.key', encode_private_key(private_key, encoding='PEM', key_format='OpenSSH', passphrase=passphrase))
def generate_dh_parameters(name, install=False, file=False):
bits = ask_input('Enter DH parameters key size:', default=2048, numeric_only=True)
print("Generating parameters...")
dh_params = create_dh_parameters(bits)
if not dh_params:
print("Failed to create DH parameters")
return None
if not install and not file:
print("DH Parameters:")
print(encode_dh_parameters(dh_params))
if install:
install_dh_parameters(name, dh_params)
if file:
write_file(f'{name}.pem', encode_dh_parameters(dh_params))
def generate_keypair(name, install=False, file=False):
private_key, key_type = generate_private_key()
public_key = private_key.public_key()
passphrase = ask_passphrase()
if not install and not file:
print(encode_public_key(public_key))
print("")
print(encode_private_key(private_key, passphrase=passphrase))
return None
if install:
install_keypair(name, key_type, private_key, public_key, passphrase)
if file:
write_file(f'{name}.pem', encode_public_key(public_key))
write_file(f'{name}.key', encode_private_key(private_key, passphrase=passphrase))
def generate_openvpn_key(name, install=False, file=False):
result = cmd('openvpn --genkey secret /dev/stdout | grep -o "^[^#]*"')
if not result:
print("Failed to generate OpenVPN key")
return None
if not install and not file:
print(result)
return None
if install:
key_lines = result.split("\n")
key_data = "".join(key_lines[1:-1]) # Remove wrapper tags and line endings
key_version = '1'
- import re
version_search = re.search(r'BEGIN OpenVPN Static key V(\d+)', result) # Future-proofing (hopefully)
if version_search:
key_version = version_search[1]
base = f"set pki openvpn shared-secret {name}"
print("Configure mode commands to install OpenVPN key:")
print(f"{base} key '{key_data}'")
print(f"{base} version '{key_version}'")
if file:
write_file(f'{name}.key', result)
def generate_wireguard_key(interface=None, install=False):
private_key = cmd('wg genkey')
public_key = cmd('wg pubkey', input=private_key)
if interface and install:
install_wireguard_key(interface, private_key, public_key)
else:
print(f'Private key: {private_key}')
print(f'Public key: {public_key}', end='\n\n')
def generate_wireguard_psk(interface=None, peer=None, install=False):
psk = cmd('wg genpsk')
if interface and peer and install:
install_wireguard_psk(interface, peer, psk)
else:
print(f'Pre-shared key: {psk}')
+# Import functions
+def import_ca_certificate(name, path=None, key_path=None):
+ if path:
+ if not os.path.exists(path):
+ print(f'File not found: {path}')
+ return
+
+ cert = None
+
+ with open(path) as f:
+ cert_data = f.read()
+ cert = load_certificate(cert_data, wrap_tags=False)
+
+ if not cert:
+ print(f'Invalid certificate: {path}')
+ return
+
+ install_certificate(name, cert, is_ca=True)
+
+ if key_path:
+ if not os.path.exists(key_path):
+ print(f'File not found: {key_path}')
+ return
+
+ key = None
+ passphrase = ask_input('Enter private key passphrase: ') or None
+
+ with open(key_path) as f:
+ key_data = f.read()
+ key = load_private_key(key_data, passphrase=passphrase, wrap_tags=False)
+
+ if not key:
+ print(f'Invalid private key or passphrase: {path}')
+ return
+
+ install_certificate(name, private_key=key, is_ca=True)
+
+def import_certificate(name, path=None, key_path=None):
+ if path:
+ if not os.path.exists(path):
+ print(f'File not found: {path}')
+ return
+
+ cert = None
+
+ with open(path) as f:
+ cert_data = f.read()
+ cert = load_certificate(cert_data, wrap_tags=False)
+
+ if not cert:
+ print(f'Invalid certificate: {path}')
+ return
+
+ install_certificate(name, cert, is_ca=False)
+
+ if key_path:
+ if not os.path.exists(key_path):
+ print(f'File not found: {key_path}')
+ return
+
+ key = None
+ passphrase = ask_input('Enter private key passphrase: ') or None
+
+ with open(key_path) as f:
+ key_data = f.read()
+ key = load_private_key(key_data, passphrase=passphrase, wrap_tags=False)
+
+ if not key:
+ print(f'Invalid private key or passphrase: {path}')
+ return
+
+ install_certificate(name, private_key=key, is_ca=False)
+
+def import_crl(name, path):
+ if not os.path.exists(path):
+ print(f'File not found: {path}')
+ return
+
+ crl = None
+
+ with open(path) as f:
+ crl_data = f.read()
+ crl = load_crl(crl_data, wrap_tags=False)
+
+ if not crl:
+ print(f'Invalid certificate: {path}')
+ return
+
+ install_crl(name, crl)
+
+def import_dh_parameters(name, path):
+ if not os.path.exists(path):
+ print(f'File not found: {path}')
+ return
+
+ dh = None
+
+ with open(path) as f:
+ dh_data = f.read()
+ dh = load_dh_parameters(dh_data, wrap_tags=False)
+
+ if not dh:
+ print(f'Invalid DH parameters: {path}')
+ return
+
+ install_dh_parameters(name, dh)
+
+def import_keypair(name, path=None, key_path=None):
+ if path:
+ if not os.path.exists(path):
+ print(f'File not found: {path}')
+ return
+
+ key = None
+
+ with open(path) as f:
+ key_data = f.read()
+ key = load_public_key(key_data, wrap_tags=False)
+
+ if not key:
+ print(f'Invalid public key: {path}')
+ return
+
+ install_keypair(name, None, public_key=key, prompt=False)
+
+ if key_path:
+ if not os.path.exists(key_path):
+ print(f'File not found: {key_path}')
+ return
+
+ key = None
+ passphrase = ask_input('Enter private key passphrase: ') or None
+
+ with open(key_path) as f:
+ key_data = f.read()
+ key = load_private_key(key_data, passphrase=passphrase, wrap_tags=False)
+
+ if not key:
+ print(f'Invalid private key or passphrase: {path}')
+ return
+
+ install_keypair(name, None, private_key=key, prompt=False)
+
+def import_openvpn_secret(name, path):
+ if not os.path.exists(path):
+ print(f'File not found: {path}')
+ return
+
+ key_data = None
+ key_version = '1'
+
+ with open(path) as f:
+ key_lines = f.read().split("\n")
+ key_data = "".join(key_lines[1:-1]) # Remove wrapper tags and line endings
+
+ version_search = re.search(r'BEGIN OpenVPN Static key V(\d+)', key_lines[0]) # Future-proofing (hopefully)
+ if version_search:
+ key_version = version_search[1]
+
+ install_openvpn_key(name, key_data, key_version)
+
# Show functions
def show_certificate_authority(name=None):
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'])
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):
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
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'
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'
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):
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
certs = get_revoked_by_serial_numbers([revoked.serial_number for revoked in crl])
data.append([cert_name, crl.last_update, ", ".join(certs)])
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')
# SSH
parser.add_argument('--ssh', help='SSH Key', required=False)
# DH
parser.add_argument('--dh', help='DH Parameters', required=False)
# Key pair
parser.add_argument('--keypair', help='Key pair', required=False)
# OpenVPN
parser.add_argument('--openvpn', help='OpenVPN TLS key', required=False)
# WireGuard
parser.add_argument('--wireguard', help='Wireguard', action='store_true')
group = parser.add_mutually_exclusive_group()
group.add_argument('--key', help='Wireguard key pair', action='store_true', required=False)
group.add_argument('--psk', help='Wireguard pre shared key', action='store_true', required=False)
parser.add_argument('--interface', help='Install generated keys into running-config for named interface', action='store')
parser.add_argument('--peer', help='Install generated keys into running-config for peer', action='store')
# Global
parser.add_argument('--file', help='Write generated keys into specified filename', action='store_true')
parser.add_argument('--install', help='Install generated keys into running-config', action='store_true')
+ parser.add_argument('--filename', help='Write certificate into specified filename', action='store')
+ parser.add_argument('--key-filename', help='Write key into specified filename', action='store')
+
args = parser.parse_args()
try:
if args.action == 'generate':
if args.ca:
if args.sign:
generate_ca_certificate_sign(args.ca, args.sign, install=args.install, file=args.file)
else:
generate_ca_certificate(args.ca, install=args.install, file=args.file)
elif args.certificate:
if args.sign:
generate_certificate_sign(args.certificate, args.sign, install=args.install, file=args.file)
elif args.self_sign:
generate_certificate_selfsign(args.certificate, install=args.install, file=args.file)
else:
generate_certificate_request(name=args.certificate, install=args.install, file=args.file)
elif args.crl:
generate_certificate_revocation_list(args.crl, install=args.install, file=args.file)
elif args.ssh:
generate_ssh_keypair(args.ssh, install=args.install, file=args.file)
elif args.dh:
generate_dh_parameters(args.dh, install=args.install, file=args.file)
elif args.keypair:
generate_keypair(args.keypair, install=args.install, file=args.file)
elif args.openvpn:
generate_openvpn_key(args.openvpn, install=args.install, file=args.file)
elif args.wireguard:
# WireGuard supports writing key directly into the CLI, but this
# requires the vyos_libexec_dir environment variable to be set
os.environ["vyos_libexec_dir"] = "/usr/libexec/vyos"
if args.key:
generate_wireguard_key(args.interface, install=args.install)
if args.psk:
generate_wireguard_psk(args.interface, peer=args.peer, install=args.install)
-
+ elif args.action == 'import':
+ if args.ca:
+ import_ca_certificate(args.ca, path=args.filename, key_path=args.key_filename)
+ elif args.certificate:
+ import_certificate(args.certificate, path=args.filename, key_path=args.key_filename)
+ elif args.crl:
+ import_crl(args.crl, args.filename)
+ elif args.dh:
+ import_dh_parameters(args.dh, args.filename)
+ elif args.keypair:
+ import_keypair(args.keypair, path=args.filename, key_path=args.key_filename)
+ elif args.openvpn:
+ import_openvpn_secret(args.openvpn, args.filename)
elif args.action == 'show':
if args.ca:
ca_name = None if args.ca == 'all' else args.ca
if ca_name:
if not conf.exists(['pki', 'ca', ca_name]):
print(f'CA "{ca_name}" does not exist!')
exit(1)
show_certificate_authority(ca_name)
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)
show_certificate(None if args.certificate == 'all' else args.certificate)
elif args.crl:
show_crl(None if args.crl == 'all' else args.crl)
else:
show_certificate_authority()
show_certificate()
show_crl()
except KeyboardInterrupt:
print("Aborted")
sys.exit(0)