diff options
Diffstat (limited to 'python')
52 files changed, 1190 insertions, 1409 deletions
diff --git a/python/vyos/accel_ppp.py b/python/vyos/accel_ppp.py index 0af311e57..0b4f8a9fe 100644 --- a/python/vyos/accel_ppp.py +++ b/python/vyos/accel_ppp.py @@ -18,7 +18,7 @@ import sys import vyos.opmode -from vyos.util import rc_cmd +from vyos.utils.process import rc_cmd def get_server_statistics(accel_statistics, pattern, sep=':') -> dict: diff --git a/python/vyos/config.py b/python/vyos/config.py index c3bb68373..179f60c43 100644 --- a/python/vyos/config.py +++ b/python/vyos/config.py @@ -68,10 +68,16 @@ import json from copy import deepcopy import vyos.configtree -from vyos.xml_ref import multi_to_list, merge_defaults, relative_defaults +from vyos.xml_ref import multi_to_list, from_source +from vyos.xml_ref import merge_defaults, relative_defaults from vyos.utils.dict import get_sub_dict, mangle_dict_keys from vyos.configsource import ConfigSource, ConfigSourceSession +class ConfigDict(dict): + _from_defaults = {} + def from_defaults(self, path: list[str]): + return from_source(self._from_defaults, path) + class Config(object): """ The class of config access objects. @@ -239,9 +245,9 @@ class Config(object): """ lpath = self._make_path(path) root_dict = self.get_cached_root_dict(effective) - conf_dict = get_sub_dict(root_dict, lpath, get_first_key) + conf_dict = get_sub_dict(root_dict, lpath, get_first_key=get_first_key) - if key_mangling is None and no_multi_convert and not with_defaults: + if key_mangling is None and no_multi_convert and not (with_defaults or with_recursive_defaults): return deepcopy(conf_dict) rpath = lpath if get_first_key else lpath[:-1] @@ -250,6 +256,7 @@ class Config(object): conf_dict = multi_to_list(rpath, conf_dict) if with_defaults or with_recursive_defaults: + conf_dict = ConfigDict(conf_dict) conf_dict = merge_defaults(lpath, conf_dict, get_first_key=get_first_key, recursive=with_recursive_defaults) @@ -263,7 +270,17 @@ class Config(object): isinstance(key_mangling[1], str)): raise ValueError("key_mangling must be a tuple of two strings") - conf_dict = mangle_dict_keys(conf_dict, key_mangling[0], key_mangling[1], abs_path=rpath, no_tag_node_value_mangle=no_tag_node_value_mangle) + def mangle(obj): + return mangle_dict_keys(obj, key_mangling[0], key_mangling[1], + abs_path=rpath, + no_tag_node_value_mangle=no_tag_node_value_mangle) + + if isinstance(conf_dict, ConfigDict): + from_defaults = mangle(conf_dict._from_defaults) + conf_dict = mangle(conf_dict) + conf_dict._from_defaults = from_defaults + else: + conf_dict = mangle(conf_dict) return conf_dict diff --git a/python/vyos/config_mgmt.py b/python/vyos/config_mgmt.py index 26114149f..4ddabd6c2 100644 --- a/python/vyos/config_mgmt.py +++ b/python/vyos/config_mgmt.py @@ -18,16 +18,21 @@ import re import sys import gzip import logging + from typing import Optional, Tuple, Union from filecmp import cmp from datetime import datetime +from textwrap import dedent +from pathlib import Path from tabulate import tabulate from vyos.config import Config from vyos.configtree import ConfigTree, ConfigTreeError, show_diff from vyos.defaults import directories from vyos.version import get_full_version_data -from vyos.util import is_systemd_service_active, ask_yes_no, rc_cmd +from vyos.utils.io import ask_yes_no +from vyos.utils.process import is_systemd_service_active +from vyos.utils.process import rc_cmd SAVE_CONFIG = '/opt/vyatta/sbin/vyatta-save-config.pl' @@ -456,19 +461,18 @@ Proceed ?''' return ConfigTree(c) def _add_logrotate_conf(self): - conf = f"""{archive_config_file} {{ - su root vyattacfg - rotate {self.max_revisions} - start 0 - compress - copy -}}""" - mask = os.umask(0o133) - - with open(logrotate_conf, 'w') as f: - f.write(conf) - - os.umask(mask) + conf: str = dedent(f"""\ + {archive_config_file} {{ + su root vyattacfg + rotate {self.max_revisions} + start 0 + compress + copy + }} + """) + conf_file = Path(logrotate_conf) + conf_file.write_text(conf) + conf_file.chmod(0o644) def _archive_active_config(self) -> bool: mask = os.umask(0o113) diff --git a/python/vyos/configdep.py b/python/vyos/configdep.py index d4b2cc78f..7a8559839 100644 --- a/python/vyos/configdep.py +++ b/python/vyos/configdep.py @@ -18,7 +18,7 @@ import json import typing from inspect import stack -from vyos.util import load_as_module +from vyos.utils.system import load_as_module from vyos.defaults import directories from vyos.configsource import VyOSError from vyos import ConfigError diff --git a/python/vyos/configdict.py b/python/vyos/configdict.py index 1205342df..f642d38f2 100644 --- a/python/vyos/configdict.py +++ b/python/vyos/configdict.py @@ -19,9 +19,9 @@ A library for retrieving value dicts from VyOS configs in a declarative fashion. import os import json -from vyos.util import dict_search +from vyos.utils.dict import dict_search from vyos.xml import defaults -from vyos.util import cmd +from vyos.utils.process import cmd def retrieve_config(path_hash, base_path, config): """ @@ -590,7 +590,7 @@ def get_accel_dict(config, base, chap_secrets): Return a dictionary with the necessary interface config keys. """ - from vyos.util import get_half_cpus + from vyos.utils.system import get_half_cpus from vyos.template import is_ipv4 dict = config.get_config_dict(base, key_mangling=('-', '_'), diff --git a/python/vyos/configdiff.py b/python/vyos/configdiff.py index ac86af09c..0caa204c3 100644 --- a/python/vyos/configdiff.py +++ b/python/vyos/configdiff.py @@ -19,8 +19,9 @@ from vyos.config import Config from vyos.configtree import DiffTree from vyos.configdict import dict_merge from vyos.configdict import list_diff -from vyos.util import get_sub_dict, mangle_dict_keys -from vyos.util import dict_search_args +from vyos.utils.dict import get_sub_dict +from vyos.utils.dict import mangle_dict_keys +from vyos.utils.dict import dict_search_args from vyos.xml import defaults class ConfigDiffError(Exception): diff --git a/python/vyos/configquery.py b/python/vyos/configquery.py index 85fef8777..71ad5b4f0 100644 --- a/python/vyos/configquery.py +++ b/python/vyos/configquery.py @@ -1,4 +1,4 @@ -# Copyright 2021 VyOS maintainers and contributors <maintainers@vyos.io> +# Copyright 2021-2023 VyOS maintainers and contributors <maintainers@vyos.io> # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -19,9 +19,11 @@ settings from op mode, and execution of arbitrary op mode commands. ''' import os -from subprocess import STDOUT -from vyos.util import popen, boot_configuration_complete +from vyos.utils.process import STDOUT +from vyos.utils.process import popen + +from vyos.utils.boot import boot_configuration_complete from vyos.config import Config from vyos.configsource import ConfigSourceSession, ConfigSourceString from vyos.defaults import directories diff --git a/python/vyos/configsession.py b/python/vyos/configsession.py index decb82437..e8918d577 100644 --- a/python/vyos/configsession.py +++ b/python/vyos/configsession.py @@ -17,7 +17,7 @@ import re import sys import subprocess -from vyos.util import is_systemd_service_running +from vyos.utils.process import is_systemd_service_running from vyos.utils.dict import dict_to_paths CLI_SHELL_API = '/bin/cli-shell-api' diff --git a/python/vyos/configsource.py b/python/vyos/configsource.py index 510b5b65a..f582bdfab 100644 --- a/python/vyos/configsource.py +++ b/python/vyos/configsource.py @@ -1,5 +1,5 @@ -# Copyright 2020 VyOS maintainers and contributors <maintainers@vyos.io> +# Copyright 2020-2023 VyOS maintainers and contributors <maintainers@vyos.io> # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -19,7 +19,7 @@ import re import subprocess from vyos.configtree import ConfigTree -from vyos.util import boot_configuration_complete +from vyos.utils.boot import boot_configuration_complete class VyOSError(Exception): """ diff --git a/python/vyos/configverify.py b/python/vyos/configverify.py index 94dcdf4d9..5b94bd98b 100644 --- a/python/vyos/configverify.py +++ b/python/vyos/configverify.py @@ -22,8 +22,8 @@ # makes use of it! from vyos import ConfigError -from vyos.util import dict_search -from vyos.util import dict_search_recursive +from vyos.utils.dict import dict_search +from vyos.utils.dict import dict_search_recursive def verify_mtu(config): """ @@ -314,8 +314,6 @@ def verify_dhcpv6(config): recurring validation of DHCPv6 options which are mutually exclusive. """ if 'dhcpv6_options' in config: - from vyos.util import dict_search - if {'parameters_only', 'temporary'} <= set(config['dhcpv6_options']): raise ConfigError('DHCPv6 temporary and parameters-only options ' 'are mutually exclusive!') @@ -460,7 +458,7 @@ def verify_diffie_hellman_length(file, min_keysize): then or equal to min_keysize """ import os import re - from vyos.util import cmd + from vyos.utils.process import cmd try: keysize = str(min_keysize) diff --git a/python/vyos/dicts.py b/python/vyos/dicts.py deleted file mode 100644 index b12cda40f..000000000 --- a/python/vyos/dicts.py +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2019 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 vyos import ConfigError - - -class FixedDict(dict): - """ - FixedDict: A dictionnary not allowing new keys to be created after initialisation. - - >>> f = FixedDict(**{'count':1}) - >>> f['count'] = 2 - >>> f['king'] = 3 - File "...", line ..., in __setitem__ - raise ConfigError(f'Option "{k}" has no defined default') - """ - - def __init__(self, **options): - self._allowed = options.keys() - super().__init__(**options) - - def __setitem__(self, k, v): - """ - __setitem__ is a builtin which is called by python when setting dict values: - >>> d = dict() - >>> d['key'] = 'value' - >>> d - {'key': 'value'} - - is syntaxic sugar for - - >>> d = dict() - >>> d.__setitem__('key','value') - >>> d - {'key': 'value'} - """ - if k not in self._allowed: - raise ConfigError(f'Option "{k}" has no defined default') - super().__setitem__(k, v) diff --git a/python/vyos/ethtool.py b/python/vyos/ethtool.py index 9b7da89fa..ca3bcfc3d 100644 --- a/python/vyos/ethtool.py +++ b/python/vyos/ethtool.py @@ -16,7 +16,7 @@ import os import re -from vyos.util import popen +from vyos.utils.process import popen # These drivers do not support using ethtool to change the speed, duplex, or # flow control settings diff --git a/python/vyos/firewall.py b/python/vyos/firewall.py index 919032a41..2793b201c 100644 --- a/python/vyos/firewall.py +++ b/python/vyos/firewall.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2022 VyOS maintainers and contributors +# Copyright (C) 2021-2023 VyOS maintainers and contributors # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 or later as @@ -28,11 +28,11 @@ from time import strftime from vyos.remote import download from vyos.template import is_ipv4 from vyos.template import render -from vyos.util import call -from vyos.util import cmd -from vyos.util import dict_search_args -from vyos.util import dict_search_recursive -from vyos.util import run +from vyos.utils.dict import dict_search_args +from vyos.utils.dict import dict_search_recursive +from vyos.utils.process import call +from vyos.utils.process import cmd +from vyos.utils.process import run # Domain Resolver @@ -45,7 +45,7 @@ def fqdn_config_parse(firewall): rule = path[3] # rule id suffix = path[4][0] # source/destination (1 char) set_name = f'{fw_name}_{rule}_{suffix}' - + if path[0] == 'name': firewall['ip_fqdn'][set_name] = domain elif path[0] == 'ipv6_name': diff --git a/python/vyos/frr.py b/python/vyos/frr.py index a84f183ef..2e3c8a271 100644 --- a/python/vyos/frr.py +++ b/python/vyos/frr.py @@ -67,9 +67,12 @@ Apply the new configuration: import tempfile import re -from vyos import util -from vyos.util import chown -from vyos.util import cmd + +from vyos.utils.permission import chown +from vyos.utils.process import cmd +from vyos.utils.process import popen +from vyos.utils.process import STDOUT + import logging from logging.handlers import SysLogHandler import os @@ -144,7 +147,7 @@ def get_configuration(daemon=None, marked=False): if daemon: cmd += f' -d {daemon}' - output, code = util.popen(cmd, stderr=util.STDOUT) + output, code = popen(cmd, stderr=STDOUT) if code: raise OSError(code, output) @@ -166,7 +169,7 @@ def mark_configuration(config): config: The configuration string to mark/test return: The marked configuration from FRR """ - output, code = util.popen(f"{path_vtysh} -m -f -", stderr=util.STDOUT, input=config) + output, code = popen(f"{path_vtysh} -m -f -", stderr=STDOUT, input=config) if code == 2: raise ConfigurationNotValid(str(output)) @@ -206,7 +209,7 @@ def reload_configuration(config, daemon=None): cmd += f' {f.name}' LOG.debug(f'reload_configuration: Executing command against frr-reload: "{cmd}"') - output, code = util.popen(cmd, stderr=util.STDOUT) + output, code = popen(cmd, stderr=STDOUT) f.close() for i, e in enumerate(output.split('\n')): LOG.debug(f'frr-reload output: {i:3} {e}') @@ -235,7 +238,7 @@ def execute(command): cmd = f"{path_vtysh} -c '{command}'" - output, code = util.popen(cmd, stderr=util.STDOUT) + output, code = popen(cmd, stderr=STDOUT) if code: raise OSError(code, output) @@ -267,7 +270,7 @@ def configure(lines, daemon=False): for x in lines: cmd += f" -c '{x}'" - output, code = util.popen(cmd, stderr=util.STDOUT) + output, code = popen(cmd, stderr=STDOUT) if code == 1: raise ConfigurationNotValid(f'Configuration FRR failed: {repr(output)}') elif code: diff --git a/python/vyos/ifconfig/bond.py b/python/vyos/ifconfig/bond.py index 0edd17055..e88f860be 100644 --- a/python/vyos/ifconfig/bond.py +++ b/python/vyos/ifconfig/bond.py @@ -16,8 +16,8 @@ import os from vyos.ifconfig.interface import Interface -from vyos.util import cmd -from vyos.util import dict_search +from vyos.utils.process import cmd +from vyos.utils.dict import dict_search from vyos.validate import assert_list from vyos.validate import assert_positive diff --git a/python/vyos/ifconfig/bridge.py b/python/vyos/ifconfig/bridge.py index aa818bc5f..b103b49d8 100644 --- a/python/vyos/ifconfig/bridge.py +++ b/python/vyos/ifconfig/bridge.py @@ -19,8 +19,8 @@ import json from vyos.ifconfig.interface import Interface from vyos.validate import assert_boolean from vyos.validate import assert_positive -from vyos.util import cmd -from vyos.util import dict_search +from vyos.utils.process import cmd +from vyos.utils.dict import dict_search from vyos.configdict import get_vlan_ids from vyos.configdict import list_diff diff --git a/python/vyos/ifconfig/control.py b/python/vyos/ifconfig/control.py index 7a6b36e7c..c8366cb58 100644 --- a/python/vyos/ifconfig/control.py +++ b/python/vyos/ifconfig/control.py @@ -19,10 +19,10 @@ from inspect import signature from inspect import _empty from vyos.ifconfig.section import Section -from vyos.util import popen -from vyos.util import cmd -from vyos.util import read_file -from vyos.util import write_file +from vyos.utils.process import popen +from vyos.utils.process import cmd +from vyos.utils.file import read_file +from vyos.utils.file import write_file from vyos import debug class Control(Section): diff --git a/python/vyos/ifconfig/ethernet.py b/python/vyos/ifconfig/ethernet.py index 30bea3b86..4ff044c23 100644 --- a/python/vyos/ifconfig/ethernet.py +++ b/python/vyos/ifconfig/ethernet.py @@ -20,9 +20,9 @@ from glob import glob from vyos.base import Warning from vyos.ethtool import Ethtool from vyos.ifconfig.interface import Interface -from vyos.util import run -from vyos.util import dict_search -from vyos.util import read_file +from vyos.utils.dict import dict_search +from vyos.utils.file import read_file +from vyos.utils.process import run from vyos.validate import assert_list @Interface.register diff --git a/python/vyos/ifconfig/geneve.py b/python/vyos/ifconfig/geneve.py index 7a05e47a7..fbb261a35 100644 --- a/python/vyos/ifconfig/geneve.py +++ b/python/vyos/ifconfig/geneve.py @@ -14,7 +14,7 @@ # License along with this library. If not, see <http://www.gnu.org/licenses/>. from vyos.ifconfig import Interface -from vyos.util import dict_search +from vyos.utils.dict import dict_search @Interface.register class GeneveIf(Interface): diff --git a/python/vyos/ifconfig/interface.py b/python/vyos/ifconfig/interface.py index f6289a6e6..120f2131b 100644 --- a/python/vyos/ifconfig/interface.py +++ b/python/vyos/ifconfig/interface.py @@ -32,12 +32,12 @@ from vyos.configdict import list_diff from vyos.configdict import dict_merge from vyos.configdict import get_vlan_ids from vyos.template import render -from vyos.util import mac2eui64 -from vyos.util import dict_search -from vyos.util import read_file -from vyos.util import get_interface_config -from vyos.util import get_interface_namespace -from vyos.util import is_systemd_service_active +from vyos.utils.network import mac2eui64 +from vyos.utils.dict import dict_search +from vyos.utils.file import read_file +from vyos.utils.network import get_interface_config +from vyos.utils.network import get_interface_namespace +from vyos.utils.process import is_systemd_service_active from vyos.template import is_ipv4 from vyos.template import is_ipv6 from vyos.validate import is_intf_addr_assigned diff --git a/python/vyos/ifconfig/l2tpv3.py b/python/vyos/ifconfig/l2tpv3.py index fcd1fbf81..85a89ef8b 100644 --- a/python/vyos/ifconfig/l2tpv3.py +++ b/python/vyos/ifconfig/l2tpv3.py @@ -1,4 +1,4 @@ -# Copyright 2019-2021 VyOS maintainers and contributors <maintainers@vyos.io> +# Copyright 2019-2023 VyOS maintainers and contributors <maintainers@vyos.io> # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -15,7 +15,8 @@ from time import sleep from time import time -from vyos.util import run + +from vyos.utils.process import run from vyos.ifconfig.interface import Interface def wait_for_add_l2tpv3(timeout=10, sleep_interval=1, cmd=None): diff --git a/python/vyos/ifconfig/pppoe.py b/python/vyos/ifconfig/pppoe.py index 437fe0cae..fd4590beb 100644 --- a/python/vyos/ifconfig/pppoe.py +++ b/python/vyos/ifconfig/pppoe.py @@ -15,7 +15,7 @@ from vyos.ifconfig.interface import Interface from vyos.validate import assert_range -from vyos.util import get_interface_config +from vyos.utils.network import get_interface_config @Interface.register class PPPoEIf(Interface): diff --git a/python/vyos/ifconfig/tunnel.py b/python/vyos/ifconfig/tunnel.py index b7bf7d982..fb2f38e2b 100644 --- a/python/vyos/ifconfig/tunnel.py +++ b/python/vyos/ifconfig/tunnel.py @@ -17,7 +17,7 @@ # https://community.hetzner.com/tutorials/linux-setup-gre-tunnel from vyos.ifconfig.interface import Interface -from vyos.util import dict_search +from vyos.utils.dict import dict_search from vyos.validate import assert_list def enable_to_on(value): diff --git a/python/vyos/ifconfig/vrrp.py b/python/vyos/ifconfig/vrrp.py index 47aaadecd..fde903a53 100644 --- a/python/vyos/ifconfig/vrrp.py +++ b/python/vyos/ifconfig/vrrp.py @@ -1,4 +1,4 @@ -# Copyright 2019 VyOS maintainers and contributors <maintainers@vyos.io> +# Copyright 2019-2023 VyOS maintainers and contributors <maintainers@vyos.io> # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -21,8 +21,11 @@ from time import time from time import sleep from tabulate import tabulate -from vyos import util from vyos.configquery import ConfigTreeQuery +from vyos.utils.convert import seconds_to_human +from vyos.utils.file import read_file +from vyos.utils.file import wait_for_file_write_complete +from vyos.utils.process import process_running class VRRPError(Exception): pass @@ -84,21 +87,21 @@ class VRRP(object): def is_running(cls): if not os.path.exists(cls.location['pid']): return False - return util.process_running(cls.location['pid']) + return process_running(cls.location['pid']) @classmethod def collect(cls, what): fname = cls.location[what] try: # send signal to generate the configuration file - pid = util.read_file(cls.location['pid']) - util.wait_for_file_write_complete(fname, + pid = read_file(cls.location['pid']) + wait_for_file_write_complete(fname, pre_hook=(lambda: os.kill(int(pid), cls._signal[what])), timeout=30) - return util.read_file(fname) + return read_file(fname) except OSError: - # raised by vyos.util.read_file + # raised by vyos.utils.file.read_file raise VRRPNoData("VRRP data is not available (wait time exceeded)") except FileNotFoundError: raise VRRPNoData("VRRP data is not available (process not running or no active groups)") @@ -145,7 +148,7 @@ class VRRP(object): priority = data['effective_priority'] since = int(time() - float(data['last_transition'])) - last = util.seconds_to_human(since) + last = seconds_to_human(since) groups.append([name, intf, vrid, state, priority, last]) diff --git a/python/vyos/ifconfig/vti.py b/python/vyos/ifconfig/vti.py index dc99d365a..9ebbeb9ed 100644 --- a/python/vyos/ifconfig/vti.py +++ b/python/vyos/ifconfig/vti.py @@ -14,7 +14,7 @@ # License along with this library. If not, see <http://www.gnu.org/licenses/>. from vyos.ifconfig.interface import Interface -from vyos.util import dict_search +from vyos.utils.dict import dict_search @Interface.register class VTIIf(Interface): diff --git a/python/vyos/ifconfig/vxlan.py b/python/vyos/ifconfig/vxlan.py index 5baff10a9..6a9911588 100644 --- a/python/vyos/ifconfig/vxlan.py +++ b/python/vyos/ifconfig/vxlan.py @@ -15,7 +15,7 @@ from vyos import ConfigError from vyos.ifconfig import Interface -from vyos.util import dict_search +from vyos.utils.dict import dict_search @Interface.register class VXLANIf(Interface): diff --git a/python/vyos/initialsetup.py b/python/vyos/initialsetup.py index 574e7892d..3b280dc6b 100644 --- a/python/vyos/initialsetup.py +++ b/python/vyos/initialsetup.py @@ -1,7 +1,7 @@ # initialsetup -- functions for setting common values in config file, # for use in installation and first boot scripts # -# Copyright (C) 2018 VyOS maintainers and contributors +# Copyright (C) 2018-2023 VyOS maintainers and contributors # # This library is free software; you can redistribute it and/or modify it under the terms of # the GNU Lesser General Public License as published by the Free Software Foundation; @@ -12,10 +12,12 @@ # See the GNU Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public License along with this library; -# if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA +# if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA import vyos.configtree -import vyos.authutils + +from vyos.utils.auth import make_password_hash +from vyos.utils.auth import split_ssh_public_key def set_interface_address(config, intf, addr, intf_type="ethernet"): config.set(["interfaces", intf_type, intf, "address"], value=addr) @@ -35,8 +37,8 @@ def set_default_gateway(config, gateway): def set_user_password(config, user, password): # Make a password hash - hash = vyos.authutils.make_password_hash(password) - + hash = make_password_hash(password) + config.set(["system", "login", "user", user, "authentication", "encrypted-password"], value=hash) config.set(["system", "login", "user", user, "authentication", "plaintext-password"], value="") @@ -48,7 +50,7 @@ def set_user_level(config, user, level): config.set(["system", "login", "user", user, "level"], value=level) def set_user_ssh_key(config, user, key_string): - key = vyos.authutils.split_ssh_public_key(key_string, defaultname=user) + key = split_ssh_public_key(key_string, defaultname=user) config.set(["system", "login", "user", user, "authentication", "public-keys", key["name"], "key"], value=key["data"]) config.set(["system", "login", "user", user, "authentication", "public-keys", key["name"], "type"], value=key["type"]) diff --git a/python/vyos/migrator.py b/python/vyos/migrator.py index 87c74e1ea..872682bc0 100644 --- a/python/vyos/migrator.py +++ b/python/vyos/migrator.py @@ -20,7 +20,7 @@ import logging import vyos.defaults import vyos.component_version as component_version -from vyos.util import cmd +from vyos.utils.process import cmd log_file = os.path.join(vyos.defaults.directories['config'], 'vyos-migrate.log') diff --git a/python/vyos/nat.py b/python/vyos/nat.py index 53fd7fb33..603fedb9b 100644 --- a/python/vyos/nat.py +++ b/python/vyos/nat.py @@ -15,7 +15,7 @@ # along with this program. If not, see <http://www.gnu.org/licenses/>. from vyos.template import is_ip_network -from vyos.util import dict_search_args +from vyos.utils.dict import dict_search_args from vyos.template import bracketize_ipv6 @@ -54,28 +54,32 @@ def parse_nat_rule(rule_conf, rule_id, nat_type, ipv6=False): translation_str = 'return' log_suffix = '-EXCL' elif 'translation' in rule_conf: - translation_prefix = nat_type[:1] - translation_output = [f'{translation_prefix}nat'] addr = dict_search_args(rule_conf, 'translation', 'address') port = dict_search_args(rule_conf, 'translation', 'port') - - if addr and is_ip_network(addr): - if not ipv6: - map_addr = dict_search_args(rule_conf, nat_type, 'address') - translation_output.append(f'{ip_prefix} prefix to {ip_prefix} {translation_prefix}addr map {{ {map_addr} : {addr} }}') - ignore_type_addr = True - else: - translation_output.append(f'prefix to {addr}') - elif addr == 'masquerade': - if port: - addr = f'{addr} to ' - translation_output = [addr] - log_suffix = '-MASQ' + redirect_port = dict_search_args(rule_conf, 'translation', 'redirect', 'port') + if redirect_port: + translation_output = [f'redirect to {redirect_port}'] else: - translation_output.append('to') - if addr: - addr = bracketize_ipv6(addr) - translation_output.append(addr) + translation_prefix = nat_type[:1] + translation_output = [f'{translation_prefix}nat'] + + if addr and is_ip_network(addr): + if not ipv6: + map_addr = dict_search_args(rule_conf, nat_type, 'address') + translation_output.append(f'{ip_prefix} prefix to {ip_prefix} {translation_prefix}addr map {{ {map_addr} : {addr} }}') + ignore_type_addr = True + else: + translation_output.append(f'prefix to {addr}') + elif addr == 'masquerade': + if port: + addr = f'{addr} to ' + translation_output = [addr] + log_suffix = '-MASQ' + else: + translation_output.append('to') + if addr: + addr = bracketize_ipv6(addr) + translation_output.append(addr) options = [] addr_mapping = dict_search_args(rule_conf, 'translation', 'options', 'address_mapping') diff --git a/python/vyos/qos/base.py b/python/vyos/qos/base.py index 717e3c214..6c5a3d79c 100644 --- a/python/vyos/qos/base.py +++ b/python/vyos/qos/base.py @@ -16,9 +16,9 @@ import os from vyos.base import Warning -from vyos.util import cmd -from vyos.util import dict_search -from vyos.util import read_file +from vyos.utils.process import cmd +from vyos.utils.dict import dict_search +from vyos.utils.file import read_file from vyos.utils.network import get_protocol_by_name @@ -331,13 +331,15 @@ class QoSBase: # burst = cls_config['burst'] # filter_cmd += f' burst {burst}' + if 'default' in config: + default_cls_id = 1 + if 'class' in config: + class_id_max = self._get_class_max_id(config) + default_cls_id = int(class_id_max) +1 + self._build_base_qdisc(config['default'], default_cls_id) + if self.qostype == 'limiter': if 'default' in config: - if 'class' in config: - class_id_max = self._get_class_max_id(config) - default_cls_id = int(class_id_max) + 1 - self._build_base_qdisc(config['default'], default_cls_id) - filter_cmd = f'tc filter replace dev {self._interface} parent {self._parent:x}: ' filter_cmd += 'prio 255 protocol all basic' diff --git a/python/vyos/qos/priority.py b/python/vyos/qos/priority.py index 6d4a60a43..8182400f9 100644 --- a/python/vyos/qos/priority.py +++ b/python/vyos/qos/priority.py @@ -14,7 +14,7 @@ # License along with this library. If not, see <http://www.gnu.org/licenses/>. from vyos.qos.base import QoSBase -from vyos.util import dict_search +from vyos.utils.dict import dict_search class Priority(QoSBase): _parent = 1 diff --git a/python/vyos/remote.py b/python/vyos/remote.py index 66044fa52..cf731c881 100644 --- a/python/vyos/remote.py +++ b/python/vyos/remote.py @@ -25,22 +25,21 @@ import urllib.parse from ftplib import FTP from ftplib import FTP_TLS -from paramiko import SSHClient +from paramiko import SSHClient, SSHException from paramiko import MissingHostKeyPolicy from requests import Session from requests.adapters import HTTPAdapter from requests.packages.urllib3 import PoolManager -from vyos.util import ask_yes_no -from vyos.util import begin -from vyos.util import cmd -from vyos.util import make_incremental_progressbar -from vyos.util import make_progressbar -from vyos.util import print_error +from vyos.utils.io import ask_yes_no +from vyos.utils.io import make_incremental_progressbar +from vyos.utils.io import make_progressbar +from vyos.utils.io import print_error +from vyos.utils.misc import begin +from vyos.utils.process import cmd from vyos.version import get_version - CHUNK_SIZE = 8192 class InteractivePolicy(MissingHostKeyPolicy): @@ -51,7 +50,7 @@ class InteractivePolicy(MissingHostKeyPolicy): def missing_host_key(self, client, hostname, key): print_error(f"Host '{hostname}' not found in known hosts.") print_error('Fingerprint: ' + key.get_fingerprint().hex()) - if ask_yes_no('Do you wish to continue?'): + if sys.stdout.isatty() and ask_yes_no('Do you wish to continue?'): if client._host_keys_filename\ and ask_yes_no('Do you wish to permanently add this host/key pair to known hosts?'): client._host_keys.add(hostname, key.get_name(), key) @@ -97,7 +96,13 @@ def check_storage(path, size): class FtpC: - def __init__(self, url, progressbar=False, check_space=False, source_host='', source_port=0): + def __init__(self, + url, + progressbar=False, + check_space=False, + source_host='', + source_port=0, + timeout=10): self.secure = url.scheme == 'ftps' self.hostname = url.hostname self.path = url.path @@ -107,12 +112,15 @@ class FtpC: self.source = (source_host, source_port) self.progressbar = progressbar self.check_space = check_space + self.timeout = timeout def _establish(self): if self.secure: - return FTP_TLS(source_address=self.source, context=ssl.create_default_context()) + return FTP_TLS(source_address=self.source, + context=ssl.create_default_context(), + timeout=self.timeout) else: - return FTP(source_address=self.source) + return FTP(source_address=self.source, timeout=self.timeout) def download(self, location: str): # Open the file upfront before establishing connection. @@ -151,7 +159,13 @@ class FtpC: class SshC: known_hosts = os.path.expanduser('~/.ssh/known_hosts') - def __init__(self, url, progressbar=False, check_space=False, source_host='', source_port=0): + def __init__(self, + url, + progressbar=False, + check_space=False, + source_host='', + source_port=0, + timeout=10.0): self.hostname = url.hostname self.path = url.path self.username = url.username or os.getenv('REMOTE_USERNAME') @@ -160,6 +174,7 @@ class SshC: self.source = (source_host, source_port) self.progressbar = progressbar self.check_space = check_space + self.timeout = timeout def _establish(self): ssh = SSHClient() @@ -170,7 +185,7 @@ class SshC: ssh.set_missing_host_key_policy(InteractivePolicy()) # `socket.create_connection()` automatically picks a NIC and an IPv4/IPv6 address family # for us on dual-stack systems. - sock = socket.create_connection((self.hostname, self.port), socket.getdefaulttimeout(), self.source) + sock = socket.create_connection((self.hostname, self.port), self.timeout, self.source) ssh.connect(self.hostname, self.port, self.username, self.password, sock=sock) return ssh @@ -199,13 +214,20 @@ class SshC: class HttpC: - def __init__(self, url, progressbar=False, check_space=False, source_host='', source_port=0): + def __init__(self, + url, + progressbar=False, + check_space=False, + source_host='', + source_port=0, + timeout=10.0): self.urlstring = urllib.parse.urlunsplit(url) self.progressbar = progressbar self.check_space = check_space self.source_pair = (source_host, source_port) self.username = url.username or os.getenv('REMOTE_USERNAME') self.password = url.password or os.getenv('REMOTE_PASSWORD') + self.timeout = timeout def _establish(self): session = Session() @@ -221,8 +243,11 @@ class HttpC: # Not only would it potentially mess up with the progress bar but # `shutil.copyfileobj(request.raw, file)` does not handle automatic decoding. s.headers.update({'Accept-Encoding': 'identity'}) - with s.head(self.urlstring, allow_redirects=True) as r: + with s.head(self.urlstring, + allow_redirects=True, + timeout=self.timeout) as r: # Abort early if the destination is inaccessible. + print('pre-3') r.raise_for_status() # If the request got redirected, keep the last URL we ended up with. final_urlstring = r.url @@ -236,7 +261,8 @@ class HttpC: size = None if self.check_space: check_storage(location, size) - with s.get(final_urlstring, stream=True) as r, open(location, 'wb') as f: + with s.get(final_urlstring, stream=True, + timeout=self.timeout) as r, open(location, 'wb') as f: if self.progressbar and size: progress = make_incremental_progressbar(CHUNK_SIZE / size) next(progress) @@ -250,7 +276,10 @@ class HttpC: def upload(self, location: str): # Does not yet support progressbars. with self._establish() as s, open(location, 'rb') as f: - s.post(self.urlstring, data=f, allow_redirects=True) + s.post(self.urlstring, + data=f, + allow_redirects=True, + timeout=self.timeout) class TftpC: @@ -259,10 +288,16 @@ class TftpC: # 2. Since there's no concept authentication, we don't need to deal with keys/passwords. # 3. It would be a waste to import, audit and maintain a third-party library for TFTP. # 4. I'd rather not implement the entire protocol here, no matter how simple it is. - def __init__(self, url, progressbar=False, check_space=False, source_host=None, source_port=0): + def __init__(self, + url, + progressbar=False, + check_space=False, + source_host=None, + source_port=0, + timeout=10): source_option = f'--interface {source_host} --local-port {source_port}' if source_host else '' progress_flag = '--progress-bar' if progressbar else '-s' - self.command = f'curl {source_option} {progress_flag}' + self.command = f'curl {source_option} {progress_flag} --connect-timeout {timeout}' self.urlstring = urllib.parse.urlunsplit(url) def download(self, location: str): @@ -287,10 +322,16 @@ def urlc(urlstring, *args, **kwargs): raise ValueError(f'Unsupported URL scheme: "{url.scheme}"') def download(local_path, urlstring, *args, **kwargs): - urlc(urlstring, *args, **kwargs).download(local_path) + try: + urlc(urlstring, *args, **kwargs).download(local_path) + except Exception as err: + print_error(f'Unable to download "{urlstring}": {err}') def upload(local_path, urlstring, *args, **kwargs): - urlc(urlstring, *args, **kwargs).upload(local_path) + try: + urlc(urlstring, *args, **kwargs).upload(local_path) + except Exception as err: + print_error(f'Unable to upload "{urlstring}": {err}') def get_remote_config(urlstring, source_host='', source_port=0): """ diff --git a/python/vyos/template.py b/python/vyos/template.py index 254a15e3a..7d1c3970f 100644 --- a/python/vyos/template.py +++ b/python/vyos/template.py @@ -20,10 +20,10 @@ from jinja2 import Environment from jinja2 import FileSystemLoader from jinja2 import ChainableUndefined from vyos.defaults import directories -from vyos.util import chmod -from vyos.util import chown -from vyos.util import dict_search_args -from vyos.util import makedir +from vyos.utils.dict import dict_search_args +from vyos.utils.file import makedir +from vyos.utils.permission import chmod +from vyos.utils.permission import chown # Holds template filters registered via register_filter() _FILTERS = {} @@ -162,19 +162,19 @@ def force_to_list(value): @register_filter('seconds_to_human') def seconds_to_human(seconds, separator=""): """ Convert seconds to human-readable values like 1d6h15m23s """ - from vyos.util import seconds_to_human + from vyos.utils.convert import seconds_to_human return seconds_to_human(seconds, separator=separator) @register_filter('bytes_to_human') def bytes_to_human(bytes, initial_exponent=0, precision=2): """ Convert bytes to human-readable values like 1.44M """ - from vyos.util import bytes_to_human + from vyos.utils.convert import bytes_to_human return bytes_to_human(bytes, initial_exponent=initial_exponent, precision=precision) @register_filter('human_to_bytes') def human_to_bytes(value): """ Convert a data amount with a unit suffix to bytes, like 2K to 2048 """ - from vyos.util import human_to_bytes + from vyos.utils.convert import human_to_bytes return human_to_bytes(value) @register_filter('ip_from_cidr') @@ -424,7 +424,7 @@ def get_dhcp_router(interface): if not os.path.exists(lease_file): return None - from vyos.util import read_file + from vyos.utils.file import read_file for line in read_file(lease_file).splitlines(): if 'option routers' in line: (_, _, address) = line.split() diff --git a/python/vyos/util.py b/python/vyos/util.py deleted file mode 100644 index 33da5da40..000000000 --- a/python/vyos/util.py +++ /dev/null @@ -1,1175 +0,0 @@ -# Copyright 2020-2022 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 rc_cmd(command, flag='', shell=None, input=None, timeout=None, env=None, - stdout=PIPE, stderr=STDOUT, decode='utf-8'): - """ - A wrapper around popen, which returns the return code - of a command and stdout - - % rc_cmd('uname') - (0, 'Linux') - % rc_cmd('ip link show dev eth99') - (1, 'Device "eth99" does not exist.') - """ - out, code = popen( - command, flag, - stdout=stdout, stderr=stderr, - input=input, timeout=timeout, - env=env, shell=shell, - decode=decode, - ) - return code, out - - -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, append=False): - """ - 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' if not append else 'a') 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 and (len(match.groups()) == 2): - key = match.groups()[0].strip() - value = match.groups()[1].strip() - else: - raise ValueError(f"""Line "{l}" could not be parsed a colon-separated pair """, l) - 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, cmdline: str=None): - """ 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(['name', 'pid', 'cmdline']): - if cmdline: - if p.info['name'] == name and cmdline in p.info['cmdline']: - return p.info['pid'] - elif p.info['name'] == name: - return p.info['pid'] - return None - -def is_list_equal(first: list, second: list) -> bool: - """ Check if 2 lists are equal and list not empty """ - if len(first) != len(second) or len(first) == 0: - return False - return sorted(first) == sorted(second) - -def is_listen_port_bind_service(port: int, service: str) -> bool: - """Check if listen port bound to expected program name - :param port: Bind port - :param service: Program name - :return: bool - - Example: - % is_listen_port_bind_service(443, 'nginx') - True - % is_listen_port_bind_service(443, 'ocserv-main') - False - """ - from psutil import net_connections as connections - from psutil import Process as process - for connection in connections(): - addr = connection.laddr - pid = connection.pid - pid_name = process(pid).name() - pid_port = addr.port - if service == pid_name and port == pid_port: - return True - return False - -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, precision=2): - """ 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. - """ - - if bytes == 0: - return "0 B" - - 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:.{1}f} {2}".format(value, precision, suffix) - return size_string - -def human_to_bytes(value): - """ Converts a data amount with a unit suffix to bytes, like 2K to 2048 """ - - from re import match as re_match - - res = re_match(r'^\s*(\d+(?:\.\d+)?)\s*([a-zA-Z]+)\s*$', value) - - if not res: - raise ValueError(f"'{value}' is not a valid data amount") - else: - amount = float(res.group(1)) - unit = res.group(2).lower() - - if unit == 'b': - res = amount - elif (unit == 'k') or (unit == 'kb'): - res = amount * 1024 - elif (unit == 'm') or (unit == 'mb'): - res = amount * 1024**2 - elif (unit == 'g') or (unit == 'gb'): - res = amount * 1024**3 - elif (unit == 't') or (unit == 'tb'): - res = amount * 1024**4 - else: - raise ValueError(f"Unsupported data unit '{unit}'") - - # There cannot be fractional bytes, so we convert them to integer. - # However, truncating causes problems with conversion back to human unit, - # so we round instead -- that seems to work well enough. - return round(res) - -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, 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: - 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: - 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 convert_data(data): - """Convert multiple types of data to types usable in CLI - - Args: - data (str | bytes | list | OrderedDict): input data - - Returns: - str | list | dict: converted data - """ - from base64 import b64encode - from collections import OrderedDict - - if isinstance(data, str): - return data - if isinstance(data, bytes): - try: - return data.decode() - except UnicodeDecodeError: - return b64encode(data).decode() - if isinstance(data, list): - list_tmp = [] - for item in data: - list_tmp.append(convert_data(item)) - return list_tmp - if isinstance(data, OrderedDict): - dict_tmp = {} - for key, value in data.items(): - dict_tmp[key] = convert_data(value) - return dict_tmp - -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() - except Exception as e: - # errno.h: - #define EADDRINUSE 98 /* Address already in use */ - if e.errno == 98: - return False - - return True - -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 boot_configuration_success() -> bool: - from vyos.defaults import config_status - - try: - with open(config_status) as f: - res = f.read().strip() - except FileNotFoundError: - return False - - if int(res) == 0: - 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 - -def load_as_module(name: str, path: str): - import importlib.util - - spec = importlib.util.spec_from_file_location(name, path) - mod = importlib.util.module_from_spec(spec) - spec.loader.exec_module(mod) - return mod diff --git a/python/vyos/utils/__init__.py b/python/vyos/utils/__init__.py index 0d3998053..f2783113a 100644 --- a/python/vyos/utils/__init__.py +++ b/python/vyos/utils/__init__.py @@ -13,4 +13,17 @@ # 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/>. +from vyos.utils import auth +from vyos.utils import boot +from vyos.utils import commit +from vyos.utils import convert +from vyos.utils import dict +from vyos.utils import file +from vyos.utils import io +from vyos.utils import kernel +from vyos.utils import list +from vyos.utils import misc from vyos.utils import network +from vyos.utils import permission +from vyos.utils import process +from vyos.utils import system diff --git a/python/vyos/authutils.py b/python/vyos/utils/auth.py index 66b5f4a74..a59858d72 100644 --- a/python/vyos/authutils.py +++ b/python/vyos/utils/auth.py @@ -15,7 +15,7 @@ import re -from vyos.util import cmd +from vyos.utils.process import cmd def make_password_hash(password): diff --git a/python/vyos/utils/boot.py b/python/vyos/utils/boot.py new file mode 100644 index 000000000..3aecbec64 --- /dev/null +++ b/python/vyos/utils/boot.py @@ -0,0 +1,35 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +import os + +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 boot_configuration_success() -> bool: + from vyos.defaults import config_status + try: + with open(config_status) as f: + res = f.read().strip() + except FileNotFoundError: + return False + if int(res) == 0: + return True + return False diff --git a/python/vyos/utils/commit.py b/python/vyos/utils/commit.py new file mode 100644 index 000000000..105aed8c2 --- /dev/null +++ b/python/vyos/utils/commit.py @@ -0,0 +1,60 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +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) diff --git a/python/vyos/utils/convert.py b/python/vyos/utils/convert.py index 975c67e0a..ec2333ef0 100644 --- a/python/vyos/utils/convert.py +++ b/python/vyos/utils/convert.py @@ -143,3 +143,33 @@ def mac_to_eui64(mac, prefix=None): return str(net[euil]) except: # pylint: disable=bare-except return + +def convert_data(data): + """Convert multiple types of data to types usable in CLI + + Args: + data (str | bytes | list | OrderedDict): input data + + Returns: + str | list | dict: converted data + """ + from base64 import b64encode + from collections import OrderedDict + + if isinstance(data, str): + return data + if isinstance(data, bytes): + try: + return data.decode() + except UnicodeDecodeError: + return b64encode(data).decode() + if isinstance(data, list): + list_tmp = [] + for item in data: + list_tmp.append(convert_data(item)) + return list_tmp + if isinstance(data, OrderedDict): + dict_tmp = {} + for key, value in data.items(): + dict_tmp[key] = convert_data(value) + return dict_tmp diff --git a/python/vyos/utils/dict.py b/python/vyos/utils/dict.py index 28d32bb8d..9484eacdd 100644 --- a/python/vyos/utils/dict.py +++ b/python/vyos/utils/dict.py @@ -13,7 +13,6 @@ # 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/>. - 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. @@ -87,7 +86,7 @@ def mangle_dict_keys(data, regex, replacement, abs_path=None, no_tag_node_value_ if abs_path is None: abs_path = [] - new_dict = {} + new_dict = type(data)() for k in data.keys(): if no_tag_node_value_mangle and is_tag_value(abs_path + [k]): @@ -270,3 +269,39 @@ def check_mutually_exclusive_options(d, keys, required=False): if required and (len(present_keys) < 1): raise ValueError(f"At least one of the following options is required: {orig_keys}") + +class FixedDict(dict): + """ + FixedDict: A dictionnary not allowing new keys to be created after initialisation. + + >>> f = FixedDict(**{'count':1}) + >>> f['count'] = 2 + >>> f['king'] = 3 + File "...", line ..., in __setitem__ + raise ConfigError(f'Option "{k}" has no defined default') + """ + + from vyos import ConfigError + + def __init__(self, **options): + self._allowed = options.keys() + super().__init__(**options) + + def __setitem__(self, k, v): + """ + __setitem__ is a builtin which is called by python when setting dict values: + >>> d = dict() + >>> d['key'] = 'value' + >>> d + {'key': 'value'} + + is syntaxic sugar for + + >>> d = dict() + >>> d.__setitem__('key','value') + >>> d + {'key': 'value'} + """ + if k not in self._allowed: + raise ConfigError(f'Option "{k}" has no defined default') + super().__setitem__(k, v) diff --git a/python/vyos/utils/file.py b/python/vyos/utils/file.py index 2560a35be..667a2464b 100644 --- a/python/vyos/utils/file.py +++ b/python/vyos/utils/file.py @@ -14,7 +14,19 @@ # License along with this library. If not, see <http://www.gnu.org/licenses/>. import os +from vyos.utils.permission import chown +def makedir(path, user=None, group=None): + if os.path.exists(path): + return + os.makedirs(path, mode=0o755) + chown(path, user, group) + +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 read_file(fname, defaultonfailure=None): """ diff --git a/python/vyos/utils/kernel.py b/python/vyos/utils/kernel.py new file mode 100644 index 000000000..0eb113174 --- /dev/null +++ b/python/vyos/utils/kernel.py @@ -0,0 +1,27 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +import os + +def check_kmod(k_mod): + """ Common utility function to load required kernel modules on demand """ + from vyos import ConfigError + from vyos.utils.process import call + 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') diff --git a/python/vyos/utils/list.py b/python/vyos/utils/list.py new file mode 100644 index 000000000..63ef720ab --- /dev/null +++ b/python/vyos/utils/list.py @@ -0,0 +1,20 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +def is_list_equal(first: list, second: list) -> bool: + """ Check if 2 lists are equal and list not empty """ + if len(first) != len(second) or len(first) == 0: + return False + return sorted(first) == sorted(second) diff --git a/python/vyos/utils/misc.py b/python/vyos/utils/misc.py new file mode 100644 index 000000000..d82655914 --- /dev/null +++ b/python/vyos/utils/misc.py @@ -0,0 +1,66 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +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 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 + from vyos.utils.io import ask_yes_no + from vyos.utils.process import cmd + 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.') diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py index 72b7ca6da..3786caf26 100644 --- a/python/vyos/utils/network.py +++ b/python/vyos/utils/network.py @@ -15,7 +15,6 @@ import os - def get_protocol_by_name(protocol_name): """Get protocol number by protocol name @@ -28,3 +27,187 @@ def get_protocol_by_name(protocol_name): return protocol_number except socket.error: return protocol_name + +def interface_exists_in_netns(interface_name, netns): + from vyos.utils.process import rc_cmd + rc, out = rc_cmd(f'ip netns exec {netns} ip link show dev {interface_name}') + if rc == 0: + return True + return False + +def get_interface_vrf(interface): + """ Returns VRF of given interface """ + from vyos.utils.dict import dict_search + from vyos.utils.network import get_interface_config + tmp = get_interface_config(interface) + if dict_search('linkinfo.info_slave_kind', tmp) == 'vrf': + return tmp['master'] + return 'default' + +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 + from vyos.utils.process import cmd + tmp = loads(cmd(f'ip --detail --json link show dev {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 + from vyos.utils.process import cmd + tmp = loads(cmd(f'ip --detail --json addr show dev {interface}'))[0] + return tmp + +def get_interface_namespace(iface): + """ + Returns wich netns the interface belongs to + """ + from json import loads + from vyos.utils.process import cmd + # Check if netns exist + tmp = loads(cmd(f'ip --json netns ls')) + if len(tmp) == 0: + return None + + for ns in tmp: + netns = f'{ns["name"]}' + # Search interface in each netns + data = loads(cmd(f'ip netns exec {netns} ip --json link show')) + for tmp in data: + if iface == tmp["ifname"]: + return netns + + +def is_wwan_connected(interface): + """ Determine if a given WWAN interface, e.g. wwan0 is connected to the + carrier network or not """ + import json + from vyos.utils.process import cmd + + 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 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 + from vyos.utils.process import cmd + tmp = loads(cmd(f'bridge -j fdb show dev {interface}')) + return tmp + +def get_all_vrfs(): + """ Return a dictionary of all system wide known VRF instances """ + from json import loads + from vyos.utils.process import cmd + tmp = loads(cmd('ip --json 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 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 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() + except Exception as e: + # errno.h: + #define EADDRINUSE 98 /* Address already in use */ + if e.errno == 98: + return False + + return True + +def is_listen_port_bind_service(port: int, service: str) -> bool: + """Check if listen port bound to expected program name + :param port: Bind port + :param service: Program name + :return: bool + + Example: + % is_listen_port_bind_service(443, 'nginx') + True + % is_listen_port_bind_service(443, 'ocserv-main') + False + """ + from psutil import net_connections as connections + from psutil import Process as process + for connection in connections(): + addr = connection.laddr + pid = connection.pid + pid_name = process(pid).name() + pid_port = addr.port + if service == pid_name and port == pid_port: + return True + return False diff --git a/python/vyos/utils/permission.py b/python/vyos/utils/permission.py new file mode 100644 index 000000000..d938b494f --- /dev/null +++ b/python/vyos/utils/permission.py @@ -0,0 +1,78 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +import os + +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 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 get_cfg_group_id(): + from grp import getgrnam + from vyos.defaults import cfg_group + + group_data = getgrnam(cfg_group) + return group_data.gr_gid diff --git a/python/vyos/utils/process.py b/python/vyos/utils/process.py new file mode 100644 index 000000000..911547995 --- /dev/null +++ b/python/vyos/utils/process.py @@ -0,0 +1,232 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +import os + +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: + from sys import stderr + err_msg = f"returned (err):\n{str_err}" + # this message will also be send to syslog via airbag + debug.message(err_msg, flag, destination=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 rc_cmd(command, flag='', shell=None, input=None, timeout=None, env=None, + stdout=PIPE, stderr=STDOUT, decode='utf-8'): + """ + A wrapper around popen, which returns the return code + of a command and stdout + + % rc_cmd('uname') + (0, 'Linux') + % rc_cmd('ip link show dev eth99') + (1, 'Device "eth99" does not exist.') + """ + out, code = popen( + command, flag, + stdout=stdout, stderr=stderr, + input=input, timeout=timeout, + env=env, shell=shell, + decode=decode, + ) + return code, out + +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 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, cmdline: str=None): + """ 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(['name', 'pid', 'cmdline']): + if cmdline: + if p.info['name'] == name and cmdline in p.info['cmdline']: + return p.info['pid'] + elif p.info['name'] == name: + return p.info['pid'] + return None + +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')) diff --git a/python/vyos/utils/system.py b/python/vyos/utils/system.py new file mode 100644 index 000000000..5d41c0c05 --- /dev/null +++ b/python/vyos/utils/system.py @@ -0,0 +1,107 @@ +# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io> +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library. If not, see <http://www.gnu.org/licenses/>. + +import os +from subprocess import run + +def sysctl_read(name: str) -> str: + """Read and return current value of sysctl() option + + Args: + name (str): sysctl key name + + Returns: + str: sysctl key value + """ + tmp = run(['sysctl', '-nb', name], capture_output=True) + return tmp.stdout.decode() + +def sysctl_write(name: str, value: str | int) -> bool: + """Change value via sysctl() + + Args: + name (str): sysctl key name + value (str | int): sysctl key value + + Returns: + bool: True if changed, False otherwise + """ + # convert other types to string before comparison + if not isinstance(value, str): + value = str(value) + # do not change anything if a value is already configured + if sysctl_read(name) == value: + return True + # return False if sysctl call failed + if run(['sysctl', '-wq', f'{name}={value}']).returncode != 0: + return False + # compare old and new values + # sysctl may apply value, but its actual value will be + # different from requested + if sysctl_read(name) == value: + return True + # False in other cases + return False + +def sysctl_apply(sysctl_dict: dict[str, str], revert: bool = True) -> bool: + """Apply sysctl values. + + Args: + sysctl_dict (dict[str, str]): dictionary with sysctl keys with values + revert (bool, optional): Revert to original values if new were not + applied. Defaults to True. + + Returns: + bool: True if all params configured properly, False in other cases + """ + # get current values + sysctl_original: dict[str, str] = {} + for key_name in sysctl_dict.keys(): + sysctl_original[key_name] = sysctl_read(key_name) + # apply new values and revert in case one of them was not applied + for key_name, value in sysctl_dict.items(): + if not sysctl_write(key_name, value): + if revert: + sysctl_apply(sysctl_original, revert=False) + return False + # everything applied + return True + +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 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 load_as_module(name: str, path: str): + import importlib.util + + spec = importlib.util.spec_from_file_location(name, path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod diff --git a/python/vyos/validate.py b/python/vyos/validate.py index e5d8c6043..567f4c972 100644 --- a/python/vyos/validate.py +++ b/python/vyos/validate.py @@ -100,8 +100,9 @@ def is_intf_addr_assigned(intf, address) -> bool: def is_addr_assigned(ip_address, vrf=None) -> bool: """ Verify if the given IPv4/IPv6 address is assigned to any interface """ from netifaces import interfaces - from vyos.util import get_interface_config - from vyos.util import dict_search + from vyos.utils.network import get_interface_config + from vyos.utils.dict import dict_search + for interface in interfaces(): # Check if interface belongs to the requested VRF, if this is not the # case there is no need to proceed with this data set - continue loop @@ -218,7 +219,7 @@ def assert_mtu(mtu, ifname): assert_number(mtu) import json - from vyos.util import cmd + from vyos.utils.process import cmd out = cmd(f'ip -j -d link show dev {ifname}') # [{"ifindex":2,"ifname":"eth0","flags":["BROADCAST","MULTICAST","UP","LOWER_UP"],"mtu":1500,"qdisc":"pfifo_fast","operstate":"UP","linkmode":"DEFAULT","group":"default","txqlen":1000,"link_type":"ether","address":"08:00:27:d9:5b:04","broadcast":"ff:ff:ff:ff:ff:ff","promiscuity":0,"min_mtu":46,"max_mtu":16110,"inet6_addr_gen_mode":"none","num_tx_queues":1,"num_rx_queues":1,"gso_max_size":65536,"gso_max_segs":65535}] parsed = json.loads(out)[0] diff --git a/python/vyos/version.py b/python/vyos/version.py index fb706ad44..1c5651c83 100644 --- a/python/vyos/version.py +++ b/python/vyos/version.py @@ -1,4 +1,4 @@ -# Copyright 2017-2020 VyOS maintainers and contributors <maintainers@vyos.io> +# Copyright 2017-2023 VyOS maintainers and contributors <maintainers@vyos.io> # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -34,11 +34,11 @@ import json import requests import vyos.defaults -from vyos.util import read_file -from vyos.util import read_json -from vyos.util import popen -from vyos.util import run -from vyos.util import DEVNULL +from vyos.utils.file import read_file +from vyos.utils.file import read_json +from vyos.utils.process import popen +from vyos.utils.process import run +from vyos.utils.process import DEVNULL version_file = os.path.join(vyos.defaults.directories['data'], 'version.json') diff --git a/python/vyos/xml_ref/__init__.py b/python/vyos/xml_ref/__init__.py index 62d3680a1..ad2130dca 100644 --- a/python/vyos/xml_ref/__init__.py +++ b/python/vyos/xml_ref/__init__.py @@ -48,6 +48,9 @@ def is_leaf(path: list) -> bool: def cli_defined(path: list, node: str, non_local=False) -> bool: return load_reference().cli_defined(path, node, non_local=non_local) +def from_source(d: dict, path: list) -> bool: + return load_reference().from_source(d, path) + def component_version() -> dict: return load_reference().component_version() diff --git a/python/vyos/xml_ref/definition.py b/python/vyos/xml_ref/definition.py index 33a49ca69..d95d580e2 100644 --- a/python/vyos/xml_ref/definition.py +++ b/python/vyos/xml_ref/definition.py @@ -13,7 +13,12 @@ # 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/>. -from typing import Optional, Union, Any +from typing import Optional, Union, Any, TYPE_CHECKING + +# https://peps.python.org/pep-0484/#forward-references +# for type 'ConfigDict' +if TYPE_CHECKING: + from vyos.config import ConfigDict class Xml: def __init__(self): @@ -123,9 +128,6 @@ class Xml: return d def multi_to_list(self, rpath: list, conf: dict) -> dict: - if rpath and rpath[-1] in list(conf): - raise ValueError('rpath should be disjoint from conf keys') - res: Any = {} for k in list(conf): @@ -210,19 +212,42 @@ class Xml: return False return True + def _set_source_recursive(self, o: Union[dict, str, list], b: bool): + d = {} + if not isinstance(o, dict): + d = {'_source': b} + else: + for k, v in o.items(): + d[k] = self._set_source_recursive(v, b) + d |= {'_source': b} + return d + # use local copy of function in module configdict, to avoid circular # import + # + # extend dict_merge to keep track of keys only in source def _dict_merge(self, source, destination): from copy import deepcopy - tmp = deepcopy(destination) + dest = deepcopy(destination) + from_source = {} for key, value in source.items(): - if key not in tmp: - tmp[key] = value + if key not in dest: + dest[key] = value + from_source[key] = self._set_source_recursive(value, True) elif isinstance(source[key], dict): - tmp[key] = self._dict_merge(source[key], tmp[key]) + dest[key], f = self._dict_merge(source[key], dest[key]) + f |= {'_source': False} + from_source[key] = f + + return dest, from_source - return tmp + def from_source(self, d: dict, path: list) -> bool: + for key in path: + d = d[key] if key in d else {} + if not d or not isinstance(d, dict): + return False + return d.get('_source', False) def _relative_defaults(self, rpath: list, conf: dict, recursive=False) -> dict: res: dict = {} @@ -246,13 +271,14 @@ class Xml: if not conf: return self.get_defaults(path, get_first_key=get_first_key, recursive=recursive) - if path and path[-1] in list(conf): - conf = conf[path[-1]] - conf = {} if not isinstance(conf, dict) else conf - if not self._well_defined(path, conf): - print('path to config dict does not define full config paths') - return {} + # adjust for possible overlap: + if path and path[-1] in list(conf): + conf = conf[path[-1]] + conf = {} if not isinstance(conf, dict) else conf + if not self._well_defined(path, conf): + print('path to config dict does not define full config paths') + return {} res = self._relative_defaults(path, conf, recursive=recursive) @@ -264,13 +290,16 @@ class Xml: return res - def merge_defaults(self, path: list, conf: dict, get_first_key=False, - recursive=False) -> dict: + def merge_defaults(self, path: list, conf: Union[dict, 'ConfigDict'], + get_first_key=False, recursive=False) -> dict: """Return config dict with defaults non-destructively merged This merges non-recursive defaults relative to the config dict. """ d = self.relative_defaults(path, conf, get_first_key=get_first_key, recursive=recursive) - d = self._dict_merge(d, conf) + d, f = self._dict_merge(d, conf) + d = type(conf)(d) + if hasattr(d, '_from_defaults'): + setattr(d, '_from_defaults', f) return d |