diff --git a/interface-definitions/pki.xml.in b/interface-definitions/pki.xml.in
index 6d137c2ce..c4fde2c78 100644
--- a/interface-definitions/pki.xml.in
+++ b/interface-definitions/pki.xml.in
@@ -1,209 +1,210 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="pki" owner="${vyos_conf_scripts_dir}/pki.py">
<properties>
<help>VyOS PKI configuration</help>
+ <priority>300</priority>
</properties>
<children>
<tagNode name="ca">
<properties>
<help>Certificate Authority</help>
</properties>
<children>
<leafNode name="certificate">
<properties>
<help>CA certificate in PEM format</help>
</properties>
</leafNode>
<leafNode name="description">
<properties>
<help>Description</help>
</properties>
</leafNode>
<node name="private">
<properties>
<help>CA private key in PEM format</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>CA private key in PEM format</help>
</properties>
</leafNode>
<leafNode name="password-protected">
<properties>
<help>CA private key is password protected</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<leafNode name="crl">
<properties>
<help>Certificate revocation list in PEM format</help>
<multi/>
</properties>
</leafNode>
<leafNode name="revoke">
<properties>
<help>If parent CA is present, this CA certificate will be included in generated CRLs</help>
<valueless/>
</properties>
</leafNode>
</children>
</tagNode>
<tagNode name="certificate">
<properties>
<help>Certificate</help>
</properties>
<children>
<leafNode name="certificate">
<properties>
<help>Certificate in PEM format</help>
</properties>
</leafNode>
<leafNode name="description">
<properties>
<help>Description</help>
</properties>
</leafNode>
<node name="private">
<properties>
<help>Certificate private key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>Certificate private key in PEM format</help>
</properties>
</leafNode>
<leafNode name="password-protected">
<properties>
<help>Certificate private key is password protected</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<leafNode name="revoke">
<properties>
<help>If CA is present, this certificate will be included in generated CRLs</help>
<valueless/>
</properties>
</leafNode>
</children>
</tagNode>
<tagNode name="dh">
<properties>
<help>Diffie-Hellman parameters</help>
</properties>
<children>
<leafNode name="parameters">
<properties>
<help>DH parameters in PEM format</help>
</properties>
</leafNode>
</children>
</tagNode>
<tagNode name="key-pair">
<properties>
<help>Public and private keys</help>
</properties>
<children>
<node name="public">
<properties>
<help>Public key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>Public key in PEM format</help>
</properties>
</leafNode>
</children>
</node>
<node name="private">
<properties>
<help>Private key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>Private key in PEM format</help>
</properties>
</leafNode>
<leafNode name="password-protected">
<properties>
<help>Private key is password protected</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
</children>
</tagNode>
<node name="openvpn">
<properties>
<help>OpenVPN keys</help>
</properties>
<children>
<tagNode name="shared-secret">
<properties>
<help>OpenVPN shared secret key</help>
</properties>
<children>
<leafNode name="key">
<properties>
<help>OpenVPN shared secret key data</help>
</properties>
</leafNode>
<leafNode name="version">
<properties>
<help>OpenVPN shared secret key version</help>
</properties>
</leafNode>
</children>
</tagNode>
</children>
</node>
<node name="x509">
<properties>
<help>X509 Settings</help>
</properties>
<children>
<node name="default">
<properties>
<help>X509 Default Values</help>
</properties>
<children>
<leafNode name="country">
<properties>
<help>Default country</help>
</properties>
<defaultValue>GB</defaultValue>
</leafNode>
<leafNode name="state">
<properties>
<help>Default state</help>
</properties>
<defaultValue>Some-State</defaultValue>
</leafNode>
<leafNode name="locality">
<properties>
<help>Default locality</help>
</properties>
<defaultValue>Some-City</defaultValue>
</leafNode>
<leafNode name="organization">
<properties>
<help>Default organization</help>
</properties>
<defaultValue>VyOS</defaultValue>
</leafNode>
</children>
</node>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/python/vyos/util.py b/python/vyos/util.py
index de55e108b..0d62fbfe9 100644
--- a/python/vyos/util.py
+++ b/python/vyos/util.py
@@ -1,1026 +1,1031 @@
# Copyright 2020-2021 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
import os
import re
import sys
#
# NOTE: Do not import full classes here, move your import to the function
# where it is used so it is as local as possible to the execution
#
from subprocess import Popen
from subprocess import PIPE
from subprocess import STDOUT
from subprocess import DEVNULL
def popen(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=PIPE, stderr=PIPE, decode='utf-8'):
"""
popen is a wrapper helper aound subprocess.Popen
with it default setting it will return a tuple (out, err)
out: the output of the program run
err: the error code returned by the program
it can be affected by the following flags:
shell: do not try to auto-detect if a shell is required
for example if a pipe (|) or redirection (>, >>) is used
input: data to sent to the child process via STDIN
the data should be bytes but string will be converted
timeout: time after which the command will be considered to have failed
env: mapping that defines the environment variables for the new process
stdout: define how the output of the program should be handled
- PIPE (default), sends stdout to the output
- DEVNULL, discard the output
stderr: define how the output of the program should be handled
- None (default), send/merge the data to/with stderr
- PIPE, popen will append it to output
- STDOUT, send the data to be merged with stdout
- DEVNULL, discard the output
decode: specify the expected text encoding (utf-8, ascii, ...)
the default is explicitely utf-8 which is python's own default
usage:
get both stdout and stderr: popen('command', stdout=PIPE, stderr=STDOUT)
discard stdout and get stderr: popen('command', stdout=DEVNUL, stderr=PIPE)
"""
# airbag must be left as an import in the function as otherwise we have a
# a circual import dependency
from vyos import debug
from vyos import airbag
# log if the flag is set, otherwise log if command is set
if not debug.enabled(flag):
flag = 'command'
cmd_msg = f"cmd '{command}'"
debug.message(cmd_msg, flag)
use_shell = shell
stdin = None
if shell is None:
use_shell = False
if ' ' in command:
use_shell = True
if env:
use_shell = True
if input:
stdin = PIPE
input = input.encode() if type(input) is str else input
p = Popen(command, stdin=stdin, stdout=stdout, stderr=stderr,
env=env, shell=use_shell)
pipe = p.communicate(input, timeout)
pipe_out = b''
if stdout == PIPE:
pipe_out = pipe[0]
pipe_err = b''
if stderr == PIPE:
pipe_err = pipe[1]
str_out = pipe_out.decode(decode).replace('\r\n', '\n').strip()
str_err = pipe_err.decode(decode).replace('\r\n', '\n').strip()
out_msg = f"returned (out):\n{str_out}"
if str_out:
debug.message(out_msg, flag)
if str_err:
err_msg = f"returned (err):\n{str_err}"
# this message will also be send to syslog via airbag
debug.message(err_msg, flag, destination=sys.stderr)
# should something go wrong, report this too via airbag
airbag.noteworthy(cmd_msg)
airbag.noteworthy(out_msg)
airbag.noteworthy(err_msg)
return str_out, p.returncode
def run(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=DEVNULL, stderr=PIPE, decode='utf-8'):
"""
A wrapper around popen, which discard the stdout and
will return the error code of a command
"""
_, code = popen(
command, flag,
stdout=stdout, stderr=stderr,
input=input, timeout=timeout,
env=env, shell=shell,
decode=decode,
)
return code
def cmd(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=PIPE, stderr=PIPE, decode='utf-8', raising=None, message='',
expect=[0]):
"""
A wrapper around popen, which returns the stdout and
will raise the error code of a command
raising: specify which call should be used when raising
the class should only require a string as parameter
(default is OSError) with the error code
expect: a list of error codes to consider as normal
"""
decoded, code = popen(
command, flag,
stdout=stdout, stderr=stderr,
input=input, timeout=timeout,
env=env, shell=shell,
decode=decode,
)
if code not in expect:
feedback = message + '\n' if message else ''
feedback += f'failed to run command: {command}\n'
feedback += f'returned: {decoded}\n'
feedback += f'exit code: {code}'
if raising is None:
# error code can be recovered with .errno
raise OSError(code, feedback)
else:
raise raising(feedback)
return decoded
def call(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=PIPE, stderr=PIPE, decode='utf-8'):
"""
A wrapper around popen, which print the stdout and
will return the error code of a command
"""
out, code = popen(
command, flag,
stdout=stdout, stderr=stderr,
input=input, timeout=timeout,
env=env, shell=shell,
decode=decode,
)
if out:
print(out)
return code
def read_file(fname, defaultonfailure=None):
"""
read the content of a file, stripping any end characters (space, newlines)
should defaultonfailure be not None, it is returned on failure to read
"""
try:
""" Read a file to string """
with open(fname, 'r') as f:
data = f.read().strip()
return data
except Exception as e:
if defaultonfailure is not None:
return defaultonfailure
raise e
def write_file(fname, data, defaultonfailure=None, user=None, group=None, mode=None):
"""
Write content of data to given fname, should defaultonfailure be not None,
it is returned on failure to read.
If directory of file is not present, it is auto-created.
"""
dirname = os.path.dirname(fname)
if not os.path.isdir(dirname):
os.makedirs(dirname, mode=0o755, exist_ok=False)
chown(dirname, user, group)
try:
""" Write a file to string """
bytes = 0
with open(fname, 'w') as f:
bytes = f.write(data)
chown(fname, user, group)
chmod(fname, mode)
return bytes
except Exception as e:
if defaultonfailure is not None:
return defaultonfailure
raise e
def read_json(fname, defaultonfailure=None):
"""
read and json decode the content of a file
should defaultonfailure be not None, it is returned on failure to read
"""
import json
try:
with open(fname, 'r') as f:
data = json.load(f)
return data
except Exception as e:
if defaultonfailure is not None:
return defaultonfailure
raise e
def chown(path, user, group):
""" change file/directory owner """
from pwd import getpwnam
from grp import getgrnam
if user is None or group is None:
return False
# path may also be an open file descriptor
if not isinstance(path, int) and not os.path.exists(path):
return False
uid = getpwnam(user).pw_uid
gid = getgrnam(group).gr_gid
os.chown(path, uid, gid)
return True
def chmod(path, bitmask):
# path may also be an open file descriptor
if not isinstance(path, int) and not os.path.exists(path):
return
if bitmask is None:
return
os.chmod(path, bitmask)
def chmod_600(path):
""" make file only read/writable by owner """
from stat import S_IRUSR, S_IWUSR
bitmask = S_IRUSR | S_IWUSR
chmod(path, bitmask)
def chmod_750(path):
""" make file/directory only executable to user and group """
from stat import S_IRUSR, S_IWUSR, S_IXUSR, S_IRGRP, S_IXGRP
bitmask = S_IRUSR | S_IWUSR | S_IXUSR | S_IRGRP | S_IXGRP
chmod(path, bitmask)
def chmod_755(path):
""" make file executable by all """
from stat import S_IRUSR, S_IWUSR, S_IXUSR, S_IRGRP, S_IXGRP, S_IROTH, S_IXOTH
bitmask = S_IRUSR | S_IWUSR | S_IXUSR | S_IRGRP | S_IXGRP | \
S_IROTH | S_IXOTH
chmod(path, bitmask)
def makedir(path, user=None, group=None):
if os.path.exists(path):
return
os.makedirs(path, mode=0o755)
chown(path, user, group)
def colon_separated_to_dict(data_string, uniquekeys=False):
""" Converts a string containing newline-separated entries
of colon-separated key-value pairs into a dict.
Such files are common in Linux /proc filesystem
Args:
data_string (str): data string
uniquekeys (bool): whether to insist that keys are unique or not
Returns: dict
Raises:
ValueError: if uniquekeys=True and the data string has
duplicate keys.
Note:
If uniquekeys=True, then dict entries are always strings,
otherwise they are always lists of strings.
"""
import re
key_value_re = re.compile('([^:]+)\s*\:\s*(.*)')
data_raw = re.split('\n', data_string)
data = {}
for l in data_raw:
l = l.strip()
if l:
match = re.match(key_value_re, l)
if match:
key = match.groups()[0].strip()
value = match.groups()[1].strip()
if key in data.keys():
if uniquekeys:
raise ValueError("Data string has duplicate keys: {0}".format(key))
else:
data[key].append(value)
else:
if uniquekeys:
data[key] = value
else:
data[key] = [value]
else:
pass
return data
def _mangle_dict_keys(data, regex, replacement, abs_path=[], no_tag_node_value_mangle=False, mod=0):
""" Mangles dict keys according to a regex and replacement character.
Some libraries like Jinja2 do not like certain characters in dict keys.
This function can be used for replacing all offending characters
with something acceptable.
Args:
data (dict): Original dict to mangle
Returns: dict
"""
from vyos.xml import is_tag
new_dict = {}
for key in data.keys():
save_mod = mod
save_path = abs_path[:]
abs_path.append(key)
if not is_tag(abs_path):
new_key = re.sub(regex, replacement, key)
else:
if mod%2:
new_key = key
else:
new_key = re.sub(regex, replacement, key)
if no_tag_node_value_mangle:
mod += 1
value = data[key]
if isinstance(value, dict):
new_dict[new_key] = _mangle_dict_keys(value, regex, replacement, abs_path=abs_path, mod=mod, no_tag_node_value_mangle=no_tag_node_value_mangle)
else:
new_dict[new_key] = value
mod = save_mod
abs_path = save_path[:]
return new_dict
def mangle_dict_keys(data, regex, replacement, abs_path=[], no_tag_node_value_mangle=False):
return _mangle_dict_keys(data, regex, replacement, abs_path=abs_path, no_tag_node_value_mangle=no_tag_node_value_mangle, mod=0)
def _get_sub_dict(d, lpath):
k = lpath[0]
if k not in d.keys():
return {}
c = {k: d[k]}
lpath = lpath[1:]
if not lpath:
return c
elif not isinstance(c[k], dict):
return {}
return _get_sub_dict(c[k], lpath)
def get_sub_dict(source, lpath, get_first_key=False):
""" Returns the sub-dict of a nested dict, defined by path of keys.
Args:
source (dict): Source dict to extract from
lpath (list[str]): sequence of keys
Returns: source, if lpath is empty, else
{key : source[..]..[key]} for key the last element of lpath, if exists
{} otherwise
"""
if not isinstance(source, dict):
raise TypeError("source must be of type dict")
if not isinstance(lpath, list):
raise TypeError("path must be of type list")
if not lpath:
return source
ret = _get_sub_dict(source, lpath)
if get_first_key and lpath and ret:
tmp = next(iter(ret.values()))
if not isinstance(tmp, dict):
raise TypeError("Data under node is not of type dict")
ret = tmp
return ret
def process_running(pid_file):
""" Checks if a process with PID in pid_file is running """
from psutil import pid_exists
if not os.path.isfile(pid_file):
return False
with open(pid_file, 'r') as f:
pid = f.read().strip()
return pid_exists(int(pid))
def process_named_running(name):
""" Checks if process with given name is running and returns its PID.
If Process is not running, return None
"""
from psutil import process_iter
for p in process_iter():
if name in p.name():
return p.pid
return None
def seconds_to_human(s, separator=""):
""" Converts number of seconds passed to a human-readable
interval such as 1w4d18h35m59s
"""
s = int(s)
week = 60 * 60 * 24 * 7
day = 60 * 60 * 24
hour = 60 * 60
remainder = 0
result = ""
weeks = s // week
if weeks > 0:
result = "{0}w".format(weeks)
s = s % week
days = s // day
if days > 0:
result = "{0}{1}{2}d".format(result, separator, days)
s = s % day
hours = s // hour
if hours > 0:
result = "{0}{1}{2}h".format(result, separator, hours)
s = s % hour
minutes = s // 60
if minutes > 0:
result = "{0}{1}{2}m".format(result, separator, minutes)
s = s % 60
seconds = s
if seconds > 0:
result = "{0}{1}{2}s".format(result, separator, seconds)
return result
def bytes_to_human(bytes, initial_exponent=0):
""" Converts a value in bytes to a human-readable size string like 640 KB
The initial_exponent parameter is the exponent of 2,
e.g. 10 (1024) for kilobytes, 20 (1024 * 1024) for megabytes.
"""
from math import log2
bytes = bytes * (2**initial_exponent)
# log2 is a float, while range checking requires an int
exponent = int(log2(bytes))
if exponent < 10:
value = bytes
suffix = "B"
elif exponent in range(10, 20):
value = bytes / 1024
suffix = "KB"
elif exponent in range(20, 30):
value = bytes / 1024**2
suffix = "MB"
elif exponent in range(30, 40):
value = bytes / 1024**3
suffix = "GB"
else:
value = bytes / 1024**4
suffix = "TB"
# Add a new case when the first machine with petabyte RAM
# hits the market.
size_string = "{0:.2f} {1}".format(value, suffix)
return size_string
def get_cfg_group_id():
from grp import getgrnam
from vyos.defaults import cfg_group
group_data = getgrnam(cfg_group)
return group_data.gr_gid
def file_is_persistent(path):
import re
location = r'^(/config|/opt/vyatta/etc/config)'
absolute = os.path.abspath(os.path.dirname(path))
return re.match(location,absolute)
def wait_for_inotify(file_path, pre_hook=None, event_type=None, timeout=None, sleep_interval=0.1):
""" Waits for an inotify event to occur """
if not os.path.dirname(file_path):
raise ValueError(
"File path {} does not have a directory part (required for inotify watching)".format(file_path))
if not os.path.basename(file_path):
raise ValueError(
"File path {} does not have a file part, do not know what to watch for".format(file_path))
from inotify.adapters import Inotify
from time import time
from time import sleep
time_start = time()
i = Inotify()
i.add_watch(os.path.dirname(file_path))
if pre_hook:
pre_hook()
for event in i.event_gen(yield_nones=True):
if (timeout is not None) and ((time() - time_start) > timeout):
# If the function didn't return until this point,
# the file failed to have been written to and closed within the timeout
raise OSError("Waiting for file {} to be written has failed".format(file_path))
# Most such events don't take much time, so it's better to check right away
# and sleep later.
if event is not None:
(_, type_names, path, filename) = event
if filename == os.path.basename(file_path):
if event_type in type_names:
return
sleep(sleep_interval)
def wait_for_file_write_complete(file_path, pre_hook=None, timeout=None, sleep_interval=0.1):
""" Waits for a process to close a file after opening it in write mode. """
wait_for_inotify(file_path,
event_type='IN_CLOSE_WRITE', pre_hook=pre_hook, timeout=timeout, sleep_interval=sleep_interval)
def commit_in_progress():
""" Not to be used in normal op mode scripts! """
# The CStore backend locks the config by opening a file
# The file is not removed after commit, so just checking
# if it exists is insufficient, we need to know if it's open by anyone
# There are two ways to check if any other process keeps a file open.
# The first one is to try opening it and see if the OS objects.
# That's faster but prone to race conditions and can be intrusive.
# The other one is to actually check if any process keeps it open.
# It's non-intrusive but needs root permissions, else you can't check
# processes of other users.
#
# Since this will be used in scripts that modify the config outside of the CLI
# framework, those knowingly have root permissions.
# For everything else, we add a safeguard.
from psutil import process_iter
from psutil import NoSuchProcess
from getpass import getuser
from vyos.defaults import commit_lock
if getuser() != 'root':
raise OSError('This functions needs to be run as root to return correct results!')
for proc in process_iter():
try:
files = proc.open_files()
if files:
for f in files:
if f.path == commit_lock:
return True
except NoSuchProcess as err:
# Process died before we could examine it
pass
# Default case
return False
def wait_for_commit_lock():
""" Not to be used in normal op mode scripts! """
from time import sleep
# Very synchronous approach to multiprocessing
while commit_in_progress():
sleep(1)
def ask_input(question, default='', numeric_only=False, valid_responses=[]):
question_out = question
if default:
question_out += f' (Default: {default})'
response = ''
while True:
response = input(question_out + ' ').strip()
if not response and default:
return default
if numeric_only:
if not response.isnumeric():
print("Invalid value, try again.")
continue
response = int(response)
if valid_responses and response not in valid_responses:
print("Invalid value, try again.")
continue
break
return response
def ask_yes_no(question, default=False) -> bool:
"""Ask a yes/no question via input() and return their answer."""
from sys import stdout
default_msg = "[Y/n]" if default else "[y/N]"
while True:
try:
stdout.write("%s %s " % (question, default_msg))
c = input().lower()
if c == '':
return default
elif c in ("y", "ye", "yes"):
return True
elif c in ("n", "no"):
return False
else:
stdout.write("Please respond with yes/y or no/n\n")
except EOFError:
stdout.write("\nPlease respond with yes/y or no/n\n")
def is_admin() -> bool:
"""Look if current user is in sudo group"""
from getpass import getuser
from grp import getgrnam
current_user = getuser()
(_, _, _, admin_group_members) = getgrnam('sudo')
return current_user in admin_group_members
def mac2eui64(mac, prefix=None):
"""
Convert a MAC address to a EUI64 address or, with prefix provided, a full
IPv6 address.
Thankfully copied from https://gist.github.com/wido/f5e32576bb57b5cc6f934e177a37a0d3
"""
import re
from ipaddress import ip_network
# http://tools.ietf.org/html/rfc4291#section-2.5.1
eui64 = re.sub(r'[.:-]', '', mac).lower()
eui64 = eui64[0:6] + 'fffe' + eui64[6:]
eui64 = hex(int(eui64[0:2], 16) ^ 2)[2:].zfill(2) + eui64[2:]
if prefix is None:
return ':'.join(re.findall(r'.{4}', eui64))
else:
try:
net = ip_network(prefix, strict=False)
euil = int('0x{0}'.format(eui64), 16)
return str(net[euil])
except: # pylint: disable=bare-except
return
def get_half_cpus():
""" return 1/2 of the numbers of available CPUs """
cpu = os.cpu_count()
if cpu > 1:
cpu /= 2
return int(cpu)
def check_kmod(k_mod):
""" Common utility function to load required kernel modules on demand """
from vyos import ConfigError
if isinstance(k_mod, str):
k_mod = k_mod.split()
for module in k_mod:
if not os.path.exists(f'/sys/module/{module}'):
if call(f'modprobe {module}') != 0:
raise ConfigError(f'Loading Kernel module {module} failed')
def find_device_file(device):
""" Recurively search /dev for the given device file and return its full path.
If no device file was found 'None' is returned """
from fnmatch import fnmatch
for root, dirs, files in os.walk('/dev'):
for basename in files:
if fnmatch(basename, device):
return os.path.join(root, basename)
return None
def dict_search(path, dict_object):
""" Traverse Python dictionary (dict_object) delimited by dot (.).
Return value of key if found, None otherwise.
This is faster implementation then jmespath.search('foo.bar', dict_object)"""
if not isinstance(dict_object, dict) or not path:
return None
parts = path.split('.')
inside = parts[:-1]
if not inside:
if path not in dict_object:
return None
return dict_object[path]
c = dict_object
for p in parts[:-1]:
c = c.get(p, {})
return c.get(parts[-1], None)
def dict_search_args(dict_object, *path):
# Traverse dictionary using variable arguments
# Added due to above function not allowing for '.' in the key names
# Example: dict_search_args(some_dict, 'key', 'subkey', 'subsubkey', ...)
if not isinstance(dict_object, dict) or not path:
return None
for item in path:
if item not in dict_object:
return None
dict_object = dict_object[item]
return dict_object
-def dict_search_recursive(dict_object, key):
+def dict_search_recursive(dict_object, key, path=[]):
""" Traverse a dictionary recurisvely and return the value of the key
we are looking for.
Thankfully copied from https://stackoverflow.com/a/19871956
+
+ Modified to yield optional path to found keys
"""
if isinstance(dict_object, list):
for i in dict_object:
- for x in dict_search_recursive(i, key):
- yield x
+ new_path = path + [i]
+ for x in dict_search_recursive(i, key, new_path):
+ yield x
elif isinstance(dict_object, dict):
if key in dict_object:
- yield dict_object[key]
- for j in dict_object.values():
- for x in dict_search_recursive(j, key):
+ new_path = path + [key]
+ yield dict_object[key], new_path
+ for k, j in dict_object.items():
+ new_path = path + [k]
+ for x in dict_search_recursive(j, key, new_path):
yield x
def get_bridge_fdb(interface):
""" Returns the forwarding database entries for a given interface """
if not os.path.exists(f'/sys/class/net/{interface}'):
return None
from json import loads
tmp = loads(cmd(f'bridge -j fdb show dev {interface}'))
return tmp
def get_interface_config(interface):
""" Returns the used encapsulation protocol for given interface.
If interface does not exist, None is returned.
"""
if not os.path.exists(f'/sys/class/net/{interface}'):
return None
from json import loads
tmp = loads(cmd(f'ip -d -j link show {interface}'))[0]
return tmp
def get_interface_address(interface):
""" Returns the used encapsulation protocol for given interface.
If interface does not exist, None is returned.
"""
if not os.path.exists(f'/sys/class/net/{interface}'):
return None
from json import loads
tmp = loads(cmd(f'ip -d -j addr show {interface}'))[0]
return tmp
def get_interface_namespace(iface):
"""
Returns wich netns the interface belongs to
"""
from json import loads
# Check if netns exist
tmp = loads(cmd(f'ip --json netns ls'))
if len(tmp) == 0:
return None
for ns in tmp:
namespace = f'{ns["name"]}'
# Search interface in each netns
data = loads(cmd(f'ip netns exec {namespace} ip -j link show'))
for compare in data:
if iface == compare["ifname"]:
return namespace
def get_all_vrfs():
""" Return a dictionary of all system wide known VRF instances """
from json import loads
tmp = loads(cmd('ip -j vrf list'))
# Result is of type [{"name":"red","table":1000},{"name":"blue","table":2000}]
# so we will re-arrange it to a more nicer representation:
# {'red': {'table': 1000}, 'blue': {'table': 2000}}
data = {}
for entry in tmp:
name = entry.pop('name')
data[name] = entry
return data
def print_error(str='', end='\n'):
"""
Print `str` to stderr, terminated with `end`.
Used for warnings and out-of-band messages to avoid mangling precious
stdout output.
"""
sys.stderr.write(str)
sys.stderr.write(end)
sys.stderr.flush()
def make_progressbar():
"""
Make a procedure that takes two arguments `done` and `total` and prints a
progressbar based on the ratio thereof, whose length is determined by the
width of the terminal.
"""
import shutil, math
col, _ = shutil.get_terminal_size()
col = max(col - 15, 20)
def print_progressbar(done, total):
if done <= total:
increment = total / col
length = math.ceil(done / increment)
percentage = str(math.ceil(100 * done / total)).rjust(3)
print_error(f'[{length * "#"}{(col - length) * "_"}] {percentage}%', '\r')
# Print a newline so that the subsequent prints don't overwrite the full bar.
if done == total:
print_error()
return print_progressbar
def make_incremental_progressbar(increment: float):
"""
Make a generator that displays a progressbar that grows monotonically with
every iteration.
First call displays it at 0% and every subsequent iteration displays it
at `increment` increments where 0.0 < `increment` < 1.0.
Intended for FTP and HTTP transfers with stateless callbacks.
"""
print_progressbar = make_progressbar()
total = 0.0
while total < 1.0:
print_progressbar(total, 1.0)
yield
total += increment
print_progressbar(1, 1)
# Ignore further calls.
while True:
yield
def begin(*args):
"""
Evaluate arguments in order and return the result of the *last* argument.
For combining multiple expressions in one statement. Useful for lambdas.
"""
return args[-1]
def begin0(*args):
"""
Evaluate arguments in order and return the result of the *first* argument.
For combining multiple expressions in one statement. Useful for lambdas.
"""
return args[0]
def is_systemd_service_active(service):
""" Test is a specified systemd service is activated.
Returns True if service is active, false otherwise.
Copied from: https://unix.stackexchange.com/a/435317 """
tmp = cmd(f'systemctl show --value -p ActiveState {service}')
return bool((tmp == 'active'))
def is_systemd_service_running(service):
""" Test is a specified systemd service is actually running.
Returns True if service is running, false otherwise.
Copied from: https://unix.stackexchange.com/a/435317 """
tmp = cmd(f'systemctl show --value -p SubState {service}')
return bool((tmp == 'running'))
def check_port_availability(ipaddress, port, protocol):
"""
Check if port is available and not used by any service
Return False if a port is busy or IP address does not exists
Should be used carefully for services that can start listening
dynamically, because IP address may be dynamic too
"""
from socketserver import TCPServer, UDPServer
from ipaddress import ip_address
# verify arguments
try:
ipaddress = ip_address(ipaddress).compressed
except:
raise ValueError(f'The {ipaddress} is not a valid IPv4 or IPv6 address')
if port not in range(1, 65536):
raise ValueError(f'The port number {port} is not in the 1-65535 range')
if protocol not in ['tcp', 'udp']:
raise ValueError(
f'The protocol {protocol} is not supported. Only tcp and udp are allowed'
)
# check port availability
try:
if protocol == 'tcp':
server = TCPServer((ipaddress, port), None, bind_and_activate=True)
if protocol == 'udp':
server = UDPServer((ipaddress, port), None, bind_and_activate=True)
server.server_close()
return True
except:
return False
def install_into_config(conf, config_paths, override_prompt=True):
# Allows op-mode scripts to install values if called from an active config session
# config_paths: dict of config paths
# override_prompt: if True, user will be prompted before existing nodes are overwritten
if not config_paths:
return None
from vyos.config import Config
if not Config().in_session():
print('You are not in configure mode, commands to install manually from configure mode:')
for path in config_paths:
print(f'set {path}')
return None
count = 0
failed = []
for path in config_paths:
if override_prompt and conf.exists(path) and not conf.is_multi(path):
if not ask_yes_no(f'Config node "{node}" already exists. Do you want to overwrite it?'):
continue
try:
cmd(f'/opt/vyatta/sbin/my_set {path}')
count += 1
except:
failed.append(path)
if failed:
print(f'Failed to install {len(failed)} value(s). Commands to manually install:')
for path in failed:
print(f'set {path}')
if count > 0:
print(f'{count} value(s) installed. Use "compare" to see the pending changes, and "commit" to apply.')
def is_wwan_connected(interface):
""" Determine if a given WWAN interface, e.g. wwan0 is connected to the
carrier network or not """
import json
if not interface.startswith('wwan'):
raise ValueError(f'Specified interface "{interface}" is not a WWAN interface')
# ModemManager is required for connection(s) - if service is not running,
# there won't be any connection at all!
if not is_systemd_service_active('ModemManager.service'):
return False
modem = interface.lstrip('wwan')
tmp = cmd(f'mmcli --modem {modem} --output-json')
tmp = json.loads(tmp)
# return True/False if interface is in connected state
return dict_search('modem.generic.state', tmp) == 'connected'
def boot_configuration_complete() -> bool:
""" Check if the boot config loader has completed
"""
from vyos.defaults import config_status
if os.path.isfile(config_status):
return True
return False
def sysctl_read(name):
""" Read and return current value of sysctl() option """
tmp = cmd(f'sysctl {name}')
return tmp.split()[-1]
def sysctl_write(name, value):
""" Change value via sysctl() - return True if changed, False otherwise """
tmp = cmd(f'sysctl {name}')
# last list index contains the actual value - only write if value differs
if sysctl_read(name) != str(value):
call(f'sysctl -wq {name}={value}')
return True
return False
diff --git a/smoketest/scripts/cli/test_pki.py b/smoketest/scripts/cli/test_pki.py
index e92123dbc..cba5ffdde 100755
--- a/smoketest/scripts/cli/test_pki.py
+++ b/smoketest/scripts/cli/test_pki.py
@@ -1,193 +1,250 @@
#!/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 unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
base_path = ['pki']
valid_ca_cert = """
MIIDgTCCAmmgAwIBAgIUeM0mATGs+sKF7ViBM6DEf9fQ19swDQYJKoZIhvcNAQEL
BQAwVzELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcM
CVNvbWUtQ2l0eTENMAsGA1UECgwEVnlPUzEQMA4GA1UEAwwHVnlPUyBDQTAeFw0y
MTA2MjgxMzE2NDZaFw0yNjA2MjcxMzE2NDZaMFcxCzAJBgNVBAYTAkdCMRMwEQYD
VQQIDApTb21lLVN0YXRlMRIwEAYDVQQHDAlTb21lLUNpdHkxDTALBgNVBAoMBFZ5
T1MxEDAOBgNVBAMMB1Z5T1MgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
AoIBAQDK98WwZIqgC6teHPSsyKLLRtboy55aisJN0D3iHJ8WGKkDmIrdCR2LI4J5
C82ErfPOzl4Ck4vTmqh8wnuK/dhUxxzNdFJBMPHAe/E+UawYrubtJj5g8iHYowZJ
T5HQKnZbcqlPvl6EizA+etO48WGljKhpimj9/LVTp81+BtFNP4tJ/vOl+iqyJ0+P
xiqQNDJgAF18meQRKaT9CcXycsciG9snMlB1tdOR7KDbi8lJ86lOi5ukPJaiMgWE
u4UlyFVyHJ/68NvtwRhYerMoQquqDs21OXkOd8spZL6qEsxMeK8InedA7abPaxgx
ORpHguPQV4Ib5HBH9Chdb9zBMheZAgMBAAGjRTBDMA8GA1UdEwEB/wQFMAMBAf8w
IAYDVR0lAQH/BBYwFAYIKwYBBQUHAwIGCCsGAQUFBwMBMA4GA1UdDwEB/wQEAwIB
hjANBgkqhkiG9w0BAQsFAAOCAQEAbwJZifMEDbrKPQfGLp7ZA1muM728o4EYmmE7
9eWwH22wGMSZI7T2xr5zRlFLs+Jha917yQK4b5xBMjQRAJlHKjzNLJ+3XaGlnWja
TBJ2SC5YktrmXRAIS7PxTRk/r1bHs/D00+sEWewbFYr8Js4a1Cv4TksTNyjHx8pv
phA+KIx/4qdojTslz+oH/cakUz0M9fh2B2xsO4bab5vX+LGLCK7jjeAL4Zyjf1hD
yx+Ri79L5N8h4Q69fER4cIkW7KVKUOyjEg3N4ST56urdycmyq9bXFz5pRxuZLInA
6RRToJrL8i0aPLJ6SyMujfREfjqOxdW5vyNF5/RkY+5Nz8JMgQ==
"""
valid_ca_private_key = """
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDK98WwZIqgC6te
HPSsyKLLRtboy55aisJN0D3iHJ8WGKkDmIrdCR2LI4J5C82ErfPOzl4Ck4vTmqh8
wnuK/dhUxxzNdFJBMPHAe/E+UawYrubtJj5g8iHYowZJT5HQKnZbcqlPvl6EizA+
etO48WGljKhpimj9/LVTp81+BtFNP4tJ/vOl+iqyJ0+PxiqQNDJgAF18meQRKaT9
CcXycsciG9snMlB1tdOR7KDbi8lJ86lOi5ukPJaiMgWEu4UlyFVyHJ/68NvtwRhY
erMoQquqDs21OXkOd8spZL6qEsxMeK8InedA7abPaxgxORpHguPQV4Ib5HBH9Chd
b9zBMheZAgMBAAECggEAa/CK5L0DcAvkrd9OS9lDokFhJ1qqM1KZ9NHrJyW7gP/K
Wow0RUqEuKtAxuj8+jOcdn4PRuV6tiUIt5iiJQ/MjYF6ktTqrZq+5nPDnzXGBTZ2
vuXYxKvgThqczD4RuJfsa8O1wR/nmit/k6q0kCVmnakJI1+laHWNZRjXUs+DXcWb
rUN5D4/5kyjvFilH1c8arfrO2O4DcwfX1zNbxicgYrGmjE5m6WCZKWdcgpBcIQSh
ZfNATfXIEZ16WmDIFZnuOEUtFAzweR2ataLQNoyaTUeEe6g+ZDtUQIGKR/f0+Z4T
/JMJfPX/vRn0l3nRJWWC7Okpa2xb0hVdBmS/op+TNQKBgQDvNGAkS4uUx8xw724k
zCKQJRnzR80AQ6b2FoqRbAevWm+i0ntsCMyvCItAQS8Bw+9fgITvsmd9SdYPncMQ
Z1oQYPk5yso/SPUyuNPXtygDxUP1xS1yja5MObqyrq2O2EzcxiVxEHGlZMLTNxNA
1tE8nF4c0nQpV/EfLtkQFnnUSwKBgQDZOA2hiLaiDlPj03S4UXDu6aUD2o07782C
UKl6A331ZhH/8zGEiUvBKg8IG/2FyCHQDC0C6rbfoarAhrRGbDHKkDTKNmThTj+I
YBkLt/5OATvqkEw8eL0nB+PY5JKH04/jE0F/YM/StUsgxvMCVhtp0u/d2Hq4V9sk
xah6oFbtKwKBgGEvs3wroWtyffLIpMSYl9Ze7Js2aekYk4ZahDQvYzPwl3jc8b5k
GN1oqEMT+MhL1j7EFb7ZikiSLkGsBGvuwd3zuG6toNxzhQP1qkRzqvNVO5ZoZV2s
iMt5jQw6AlQON7RfYSj92F6tgKaWMuFeJibtFSO6se12SIY134U0zIzfAoGAQWF7
yNkrj4+cdICbKzdoNKEiyAwqYpYFV2oL+OvAJ/L3DAEZMHla0eNk7t3t6yyX8NUZ
Xz1imeFBUf25mVDLk9rf6NWCe8ZfnR6/qyVQaA47CJkyOSlmVa8sR4ZVDIkDUCfl
mP98zkE/QbhgQJ3GVo3lIPMdzQq0rVbJJU/Jmk0CgYEAtHRNaoKBsxKfb7N7ewla
MzwcULIORODjWM8MUXM+R50F/2uYMiTvpz6eIUVfXoFyQoioYI8kcDZ8NamiQIS7
uZsHfKpgMDJkV3kOoZQusoDhasGQ0SOnxbz/y0XmNUtAePipH0jPY1SYUvWbvm2y
a4aWVhBFly9hi2ZeHiVxVhk=
"""
valid_cert = """
MIIB9zCCAZygAwIBAgIUQ5G1nyASL/YsKGyLNGhRPPQyo4kwCgYIKoZIzj0EAwIw
XjELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv
bWUtQ2l0eTENMAsGA1UECgwEVnlPUzEXMBUGA1UEAwwOVnlPUyBUZXN0IENlcnQw
HhcNMjEwNjI4MTMyNjIyWhcNMjIwNjI4MTMyNjIyWjBeMQswCQYDVQQGEwJHQjET
MBEGA1UECAwKU29tZS1TdGF0ZTESMBAGA1UEBwwJU29tZS1DaXR5MQ0wCwYDVQQK
DARWeU9TMRcwFQYDVQQDDA5WeU9TIFRlc3QgQ2VydDBZMBMGByqGSM49AgEGCCqG
SM49AwEHA0IABBsebIt+8rr2UysTpL8NnYUtmt47e3sC3H9IO8iI/N4uFrmGVgTL
E2G+RDGzZgG/r7LviJSTuE9HX7wHLcIr0SmjODA2MAwGA1UdEwEB/wQCMAAwFgYD
VR0lAQH/BAwwCgYIKwYBBQUHAwEwDgYDVR0PAQH/BAQDAgeAMAoGCCqGSM49BAMC
A0kAMEYCIQD5xK5kdC3TJ7SZrBGvzIM7E7Cil/KZJUyQDR9eFNNZVQIhALg8DTfr
wAawf8L+Ncjn/l2gd5cB0nGij0D7uYnm3zf/
"""
valid_dh_params = """
MIIBCAKCAQEAnNldZCrJk5MxhFoUlvvaYmUO+TmtL0uL62H2RIHJ+O0R+8vzdGPh
6zDAzo46EJK735haUgu8+A1RTsXDOXcwBqDlVe0hYj9KaPHz1HpfNKntpoPCJAYJ
wiH8dd5zVMH+iBwEKlrfteV9vWHn0HUxgLJFSLp5o6y0qpKPREJu6k0XguGScrPa
Iw6RUwsoDy3unHfk+YeC0o040R18F75V1mXWTjQlEgM7ZO2JZkLGkhW30jB0vSHr
krFqOvtPUiyG7r3+j18IUYLTN0s+5FOCfCjvSVKibNlB1vUz5y/9Ve8roctpkRM/
5R5FA0mtbl7U/yMSX4FRIQ/A9BlHiu4bowIBAg==
"""
valid_public_ec_key = """
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEAoInInwjlu/3+wDqvRa/Eyg3EMvB
pPyq2v4jqEtEh2n4lOCi7ZgNjr+1sQSvrn8mccpALYl3/RKOougC5oQzCg==
"""
valid_private_rsa_key = """
MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDDoAVyJPpcLBFs
2NdS1qMOSj7mwKBKVZiBN3nqbLiOvEHbVe22UMNvUFU3sGs2Ta2zXwhPF3d6vPPs
GlYTkO3XAffMSNXhjCsvWHiIOR4JrWf598Bpt+txBsxsa12kM3/HM7RDf3zdN2gT
twzrcWzu+zOTXlqJ2OSq/BRRZO9IMbQLQ1/h42GJHEr4THnY4zDqUjmMmIuiBXn4
xoE4KFLH1+xPTVleeKvPPeJ1wsshoUjlXYOgcsrXasDUt5gtkkXsVQwR9Lvbh+Rc
BhT+tJmrX9Cwq4YAd3tLSNJARS9HanRZ8uV0RTyZsImdw1Fr5ySpG2oEp/Z5mbL6
QYqDmQ+DAgMBAAECggEAGu7qMQf0TEJo98J3CtmwQ2Rnep+ksfdM8uVvbJ4hXs1+
h7Mx8jr2XVoDEZLBgA17z8lSvIjvkz92mdgaZ8E5bbPAqSiSAeapf3A/0AmFIDH2
scyxehyvVrVn6blygAvzGLr+o5hm2ZIqSySVq8jHBbQiKrT/5CCvgvcH2Rj7dMXd
T5lL73tCRJZsgvFNlxyj4Omj9Lh7SjL+tIwEQaLFbvANXrZ/BPyw4OlK8daBNg9b
5GvJSDitAVMgDEEApGYu1iNwMM4UJSQAC27eJdr+qJO6DDqktWOyWcyXrxJ9mDVK
FNbb9QNQZDj7bFfm6rCuSdH9yYe3vly+SNJqtyCiwQKBgQDvemt/57KiwQffmoKR
65NAZsQvmA4PtELYOV8NPeYH1BZN/EPmCc74iELJdQPFDYy903aRJEPGt7jfqprd
PexLwt73P/XiUjPrsbqgJqfF/EMiczxAktyW3xBt2lIWU1MUUmO1ps+ZZEg8Ks4e
K/3+FWqbwZ8drDBUT9BthUA0oQKBgQDRHxU6bu938PGweFJcIG6U21nsYaWiwCiT
LXA5vWZ+UEqz81BUye6tIcCDgeku3HvC/0ycvrBM9F4AZCjnnEvrAJHKl6e4j+C4
IpghGQvRvQ9ihDs9JIHnaoUC1i8dE3ISbbp1r7CN+J/HnAC2OeECMJuffXdnkVWa
xRdxU+9towKBgCwFVeNyJO00DI126o+GPVA2U9Pn4JXUbgEvMqDNgw5nVx5Iw/Zy
USBwc85yexnq7rcqOv5dKzRJK2u6AbOvoVMf5DqRAFL1B2RJDGRKFscXIwQfKLE6
DeCR6oQ3AKXn9TqkFn4axsiMnZapy6/SKGNfbnRpOCWNNGkbLtYjC3VhAoGAN0kO
ZapaaM0sOEk3DOAOHBB5j4KpNYOztmU23Cz0YcR8W2KiBCh2jxLzQFEiAp+LoJu5
9156YX3hNB1GqySo9XHrGTJKxwJSmJucuHNUqphe7t6igqGaLkH89CkHv5oaeEDG
IMLX3FC0fSMDFSnsEJYlLl8PKDRF+2rLrcxQ6h0CgYAZllNu8a7tE6cM6QsCILQn
NjuLuZRX8/KYWRqBJxatwZXCcMe2jti1HKTVVVCyYffOFa1QcAjCPknAmAz80l3e
g6a75NnEXo0J6YLAOOxd8fD2/HidhbceCmTF+3msidIzCsBidBkgn6V5TXx2IyMS
xGsJxVHfSKeooUQn6q76sg==
"""
+valid_update_cert = """
+MIICJTCCAcugAwIBAgIUZJqjNmPfVQwePjNFBtB6WI31ThMwCgYIKoZIzj0EAwIw
+VzELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv
+bWUtQ2l0eTENMAsGA1UECgwEVnlPUzEQMA4GA1UEAwwHdnlvcy5pbzAeFw0yMjA1
+MzExNTE3NDlaFw0yMzA1MzExNTE3NDlaMFcxCzAJBgNVBAYTAkdCMRMwEQYDVQQI
+DApTb21lLVN0YXRlMRIwEAYDVQQHDAlTb21lLUNpdHkxDTALBgNVBAoMBFZ5T1Mx
+EDAOBgNVBAMMB3Z5b3MuaW8wWTATBgcqhkjOPQIBBggqhkjOPQMBBwNCAAQMe0h/
+3CdD8mEgy+klk55QfJ8R3ZycefxCn4abWjzTXz/TuCIxqb4wpRT8DZtIn4NRimFT
+mODYdEDOYxFtZm37o3UwczAMBgNVHRMBAf8EAjAAMA4GA1UdDwEB/wQEAwIHgDAT
+BgNVHSUEDDAKBggrBgEFBQcDAjAdBgNVHQ4EFgQUqH7KSZpzArpMFuxLXqI8e1QD
+fBkwHwYDVR0jBBgwFoAUqH7KSZpzArpMFuxLXqI8e1QDfBkwCgYIKoZIzj0EAwID
+SAAwRQIhAKofUgRtcUljmbubPF6sqHtn/3TRvuafl8VfPbk3s2bJAiBp3Q1AnU/O
+i7t5FGhCgnv5m8DW2F3LZPCJdW4ELQ3d9A==
+"""
+
+valid_update_private_key = """
+MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgvyODf22w/p7Zgfz9
+dyLIT09LqLOrUN6zbAecfukiiiyhRANCAAQMe0h/3CdD8mEgy+klk55QfJ8R3Zyc
+efxCn4abWjzTXz/TuCIxqb4wpRT8DZtIn4NRimFTmODYdEDOYxFtZm37
+"""
+
class TestPKI(VyOSUnitTestSHIM.TestCase):
@classmethod
def setUpClass(cls):
super(TestPKI, cls).setUpClass()
# ensure we can also run this test on a live system - so lets clean
# out the current configuration :)
cls.cli_delete(cls, base_path)
def tearDown(self):
self.cli_delete(base_path)
self.cli_commit()
def test_valid_pki(self):
# Valid CA
self.cli_set(base_path + ['ca', 'smoketest', 'certificate', valid_ca_cert.replace('\n','')])
self.cli_set(base_path + ['ca', 'smoketest', 'private', 'key', valid_ca_private_key.replace('\n','')])
# Valid cert
self.cli_set(base_path + ['certificate', 'smoketest', 'certificate', valid_cert.replace('\n','')])
# Valid DH
self.cli_set(base_path + ['dh', 'smoketest', 'parameters', valid_dh_params.replace('\n','')])
# Valid public key
self.cli_set(base_path + ['key-pair', 'smoketest', 'public', 'key', valid_public_ec_key.replace('\n','')])
# Valid private key
self.cli_set(base_path + ['key-pair', 'smoketest1', 'private', 'key', valid_private_rsa_key.replace('\n','')])
self.cli_commit()
def test_invalid_ca_valid_certificate(self):
self.cli_set(base_path + ['ca', 'smoketest', 'certificate', valid_cert.replace('\n','')])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
def test_invalid_certificate(self):
self.cli_set(base_path + ['certificate', 'smoketest', 'certificate', 'invalidcertdata'])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
def test_invalid_public_key(self):
self.cli_set(base_path + ['key-pair', 'smoketest', 'public', 'key', 'invalidkeydata'])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
def test_invalid_private_key(self):
self.cli_set(base_path + ['key-pair', 'smoketest', 'private', 'key', 'invalidkeydata'])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
def test_invalid_dh_parameters(self):
self.cli_set(base_path + ['dh', 'smoketest', 'parameters', 'thisisinvalid'])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
+ def test_certificate_in_use(self):
+ self.cli_set(base_path + ['certificate', 'smoketest', 'certificate', valid_ca_cert.replace('\n','')])
+ self.cli_set(base_path + ['certificate', 'smoketest', 'private', 'key', valid_ca_private_key.replace('\n','')])
+ self.cli_commit()
+
+ self.cli_set(['service', 'https', 'certificates', 'certificate', 'smoketest'])
+ self.cli_commit()
+
+ self.cli_delete(base_path + ['certificate', 'smoketest'])
+ with self.assertRaises(ConfigSessionError):
+ self.cli_commit()
+
+ self.cli_delete(['service', 'https', 'certificates', 'certificate'])
+
+ def test_certificate_https_update(self):
+ self.cli_set(base_path + ['certificate', 'smoketest', 'certificate', valid_ca_cert.replace('\n','')])
+ self.cli_set(base_path + ['certificate', 'smoketest', 'private', 'key', valid_ca_private_key.replace('\n','')])
+ self.cli_commit()
+
+ self.cli_set(['service', 'https', 'certificates', 'certificate', 'smoketest'])
+ self.cli_commit()
+
+ cert_data = None
+
+ with open('/etc/ssl/certs/smoketest.pem') as f:
+ cert_data = f.read()
+
+ self.cli_set(base_path + ['certificate', 'smoketest', 'certificate', valid_update_cert.replace('\n','')])
+ self.cli_set(base_path + ['certificate', 'smoketest', 'private', 'key', valid_update_private_key.replace('\n','')])
+ self.cli_commit()
+
+ with open('/etc/ssl/certs/smoketest.pem') as f:
+ self.assertNotEqual(cert_data, f.read())
+
+ self.cli_delete(['service', 'https', 'certificates', 'certificate'])
+
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/conf_mode/pki.py b/src/conf_mode/pki.py
index efa3578b4..29ed7b1b7 100755
--- a/src/conf_mode/pki.py
+++ b/src/conf_mode/pki.py
@@ -1,203 +1,304 @@
#!/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/>.
from sys import exit
import jmespath
from vyos.config import Config
from vyos.configdict import dict_merge
from vyos.configdict import node_changed
from vyos.pki import is_ca_certificate
from vyos.pki import load_certificate
from vyos.pki import load_certificate_request
from vyos.pki import load_public_key
from vyos.pki import load_private_key
from vyos.pki import load_crl
from vyos.pki import load_dh_parameters
from vyos.util import ask_input
+from vyos.util import call
+from vyos.util import dict_search_args
from vyos.util import dict_search_recursive
from vyos.xml import defaults
from vyos import ConfigError
from vyos import airbag
airbag.enable()
+# keys to recursively search for under specified path, script to call if update required
+sync_search = [
+ {
+ 'keys': ['certificate'],
+ 'path': ['service', 'https'],
+ 'script': '/usr/libexec/vyos/conf_mode/https.py'
+ },
+ {
+ 'keys': ['certificate', 'ca_certificate'],
+ 'path': ['interfaces', 'ethernet'],
+ 'script': '/usr/libexec/vyos/conf_mode/interfaces-ethernet.py'
+ },
+ {
+ 'keys': ['certificate', 'ca_certificate', 'dh_params', 'shared_secret_key', 'auth_key', 'crypt_key'],
+ 'path': ['interfaces', 'openvpn'],
+ 'script': '/usr/libexec/vyos/conf_mode/interfaces-openvpn.py'
+ },
+ {
+ 'keys': ['certificate', 'ca_certificate', 'local_key', 'remote_key'],
+ 'path': ['vpn', 'ipsec'],
+ 'script': '/usr/libexec/vyos/conf_mode/vpn_ipsec.py'
+ },
+ {
+ 'keys': ['certificate', 'ca_certificate'],
+ 'path': ['vpn', 'openconnect'],
+ 'script': '/usr/libexec/vyos/conf_mode/vpn_openconnect.py'
+ },
+ {
+ 'keys': ['certificate', 'ca_certificate'],
+ 'path': ['vpn', 'sstp'],
+ 'script': '/usr/libexec/vyos/conf_mode/vpn_sstp.py'
+ }
+]
+
+# key from other config nodes -> key in pki['changed'] and pki
+sync_translate = {
+ 'certificate': 'certificate',
+ 'ca_certificate': 'ca',
+ 'dh_params': 'dh',
+ 'local_key': 'key_pair',
+ 'remote_key': 'key_pair',
+ 'shared_secret_key': 'openvpn',
+ 'auth_key': 'openvpn',
+ 'crypt_key': 'openvpn'
+}
+
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['pki']
pki = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
pki['changed'] = {}
- tmp = node_changed(conf, base + ['ca'], key_mangling=('-', '_'))
+ tmp = node_changed(conf, base + ['ca'], key_mangling=('-', '_'), recursive=True)
if tmp: pki['changed'].update({'ca' : tmp})
- tmp = node_changed(conf, base + ['certificate'], key_mangling=('-', '_'))
+ tmp = node_changed(conf, base + ['certificate'], key_mangling=('-', '_'), recursive=True)
if tmp: pki['changed'].update({'certificate' : tmp})
+ tmp = node_changed(conf, base + ['dh'], key_mangling=('-', '_'), recursive=True)
+ if tmp: pki['changed'].update({'dh' : tmp})
+
+ tmp = node_changed(conf, base + ['key-pair'], key_mangling=('-', '_'), recursive=True)
+ if tmp: pki['changed'].update({'key_pair' : tmp})
+
+ tmp = node_changed(conf, base + ['openvpn', 'shared-secret'], key_mangling=('-', '_'), recursive=True)
+ if tmp: pki['changed'].update({'openvpn' : tmp})
+
# We only merge on the defaults of there is a configuration at all
if conf.exists(base):
default_values = defaults(base)
pki = dict_merge(default_values, pki)
# We need to get the entire system configuration to verify that we are not
# deleting a certificate that is still referenced somewhere!
pki['system'] = conf.get_config_dict([], key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
return pki
def is_valid_certificate(raw_data):
# If it loads correctly we're good, or return False
return load_certificate(raw_data, wrap_tags=True)
def is_valid_ca_certificate(raw_data):
# Check if this is a valid certificate with CA attributes
cert = load_certificate(raw_data, wrap_tags=True)
if not cert:
return False
return is_ca_certificate(cert)
def is_valid_public_key(raw_data):
# If it loads correctly we're good, or return False
return load_public_key(raw_data, wrap_tags=True)
def is_valid_private_key(raw_data, protected=False):
# If it loads correctly we're good, or return False
# With encrypted private keys, we always return true as we cannot ask for password to verify
if protected:
return True
return load_private_key(raw_data, passphrase=None, wrap_tags=True)
def is_valid_crl(raw_data):
# If it loads correctly we're good, or return False
return load_crl(raw_data, wrap_tags=True)
def is_valid_dh_parameters(raw_data):
# If it loads correctly we're good, or return False
return load_dh_parameters(raw_data, wrap_tags=True)
def verify(pki):
if not pki:
return None
if 'ca' in pki:
for name, ca_conf in pki['ca'].items():
if 'certificate' in ca_conf:
if not is_valid_ca_certificate(ca_conf['certificate']):
raise ConfigError(f'Invalid certificate on CA certificate "{name}"')
if 'private' in ca_conf and 'key' in ca_conf['private']:
private = ca_conf['private']
protected = 'password_protected' in private
if not is_valid_private_key(private['key'], protected):
raise ConfigError(f'Invalid private key on CA certificate "{name}"')
if 'crl' in ca_conf:
ca_crls = ca_conf['crl']
if isinstance(ca_crls, str):
ca_crls = [ca_crls]
for crl in ca_crls:
if not is_valid_crl(crl):
raise ConfigError(f'Invalid CRL on CA certificate "{name}"')
if 'certificate' in pki:
for name, cert_conf in pki['certificate'].items():
if 'certificate' in cert_conf:
if not is_valid_certificate(cert_conf['certificate']):
raise ConfigError(f'Invalid certificate on certificate "{name}"')
if 'private' in cert_conf and 'key' in cert_conf['private']:
private = cert_conf['private']
protected = 'password_protected' in private
if not is_valid_private_key(private['key'], protected):
raise ConfigError(f'Invalid private key on certificate "{name}"')
if 'dh' in pki:
for name, dh_conf in pki['dh'].items():
if 'parameters' in dh_conf:
if not is_valid_dh_parameters(dh_conf['parameters']):
raise ConfigError(f'Invalid DH parameters on "{name}"')
if 'key_pair' in pki:
for name, key_conf in pki['key_pair'].items():
if 'public' in key_conf and 'key' in key_conf['public']:
if not is_valid_public_key(key_conf['public']['key']):
raise ConfigError(f'Invalid public key on key-pair "{name}"')
if 'private' in key_conf and 'key' in key_conf['private']:
private = key_conf['private']
protected = 'password_protected' in private
if not is_valid_private_key(private['key'], protected):
raise ConfigError(f'Invalid private key on key-pair "{name}"')
if 'x509' in pki:
if 'default' in pki['x509']:
default_values = pki['x509']['default']
if 'country' in default_values:
country = default_values['country']
if len(country) != 2 or not country.isalpha():
raise ConfigError(f'Invalid default country value. Value must be 2 alpha characters.')
if 'changed' in pki:
# if the list is getting longer, we can move to a dict() and also embed the
# search key as value from line 173 or 176
- for cert_type in ['ca', 'certificate']:
- if not cert_type in pki['changed']:
- continue
- for certificate in pki['changed'][cert_type]:
- if cert_type not in pki or certificate not in pki['changed'][cert_type]:
- if cert_type == 'ca':
- if certificate in dict_search_recursive(pki['system'], 'ca_certificate'):
- raise ConfigError(f'CA certificate "{certificate}" is still in use!')
- elif cert_type == 'certificate':
- if certificate in dict_search_recursive(pki['system'], 'certificate'):
- raise ConfigError(f'Certificate "{certificate}" is still in use!')
+ for search in sync_search:
+ for key in search['keys']:
+ changed_key = sync_translate[key]
+
+ if changed_key not in pki['changed']:
+ continue
+
+ for item_name in pki['changed'][changed_key]:
+ node_present = False
+ if changed_key == 'openvpn':
+ node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
+ else:
+ node_present = dict_search_args(pki, changed_key, item_name)
+
+ if not node_present:
+ search_dict = dict_search_args(pki['system'], *search['path'])
+
+ if not search_dict:
+ continue
+
+ for found_name, found_path in dict_search_recursive(search_dict, key):
+ if found_name == item_name:
+ path_str = " ".join(search['path'] + found_path)
+ raise ConfigError(f'PKI object "{item_name}" still in use by "{path_str}"')
return None
def generate(pki):
if not pki:
return None
return None
def apply(pki):
if not pki:
return None
- # XXX: restart services if the content of a certificate changes
+ if 'changed' in pki:
+ for search in sync_search:
+ for key in search['keys']:
+ changed_key = sync_translate[key]
+
+ if changed_key not in pki['changed']:
+ continue
+
+ for item_name in pki['changed'][changed_key]:
+ node_present = False
+ if changed_key == 'openvpn':
+ node_present = dict_search_args(pki, 'openvpn', 'shared_secret', item_name)
+ else:
+ node_present = dict_search_args(pki, changed_key, item_name)
+
+ if node_present:
+ search_dict = dict_search_args(pki['system'], *search['path'])
+
+ if not search_dict:
+ continue
+
+ for found_name, found_path in dict_search_recursive(search_dict, key):
+ if found_name == item_name:
+ path_str = ' '.join(search['path'] + found_path)
+ print(f'pki: Updating config: {path_str} {found_name}')
+
+ script = search['script']
+ if found_path[0] == 'interfaces':
+ ifname = found_path[2]
+ call(f'VYOS_TAGNODE_VALUE={ifname} {script}')
+ else:
+ call(script)
return None
if __name__ == '__main__':
try:
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)