summaryrefslogtreecommitdiff
path: root/python
diff options
context:
space:
mode:
Diffstat (limited to 'python')
-rw-r--r--python/vyos/accel_ppp.py2
-rw-r--r--python/vyos/config.py25
-rw-r--r--python/vyos/config_mgmt.py32
-rw-r--r--python/vyos/configdep.py2
-rw-r--r--python/vyos/configdict.py6
-rw-r--r--python/vyos/configdiff.py5
-rw-r--r--python/vyos/configquery.py8
-rw-r--r--python/vyos/configsession.py2
-rw-r--r--python/vyos/configsource.py4
-rw-r--r--python/vyos/configverify.py8
-rw-r--r--python/vyos/dicts.py53
-rw-r--r--python/vyos/ethtool.py2
-rw-r--r--python/vyos/firewall.py14
-rw-r--r--python/vyos/frr.py19
-rw-r--r--python/vyos/ifconfig/bond.py4
-rw-r--r--python/vyos/ifconfig/bridge.py4
-rw-r--r--python/vyos/ifconfig/control.py8
-rw-r--r--python/vyos/ifconfig/ethernet.py6
-rw-r--r--python/vyos/ifconfig/geneve.py2
-rw-r--r--python/vyos/ifconfig/interface.py12
-rw-r--r--python/vyos/ifconfig/l2tpv3.py5
-rw-r--r--python/vyos/ifconfig/pppoe.py2
-rw-r--r--python/vyos/ifconfig/tunnel.py2
-rw-r--r--python/vyos/ifconfig/vrrp.py19
-rw-r--r--python/vyos/ifconfig/vti.py2
-rw-r--r--python/vyos/ifconfig/vxlan.py2
-rw-r--r--python/vyos/initialsetup.py14
-rw-r--r--python/vyos/migrator.py2
-rw-r--r--python/vyos/nat.py44
-rw-r--r--python/vyos/qos/base.py18
-rw-r--r--python/vyos/qos/priority.py2
-rw-r--r--python/vyos/remote.py85
-rw-r--r--python/vyos/template.py16
-rw-r--r--python/vyos/util.py1175
-rw-r--r--python/vyos/utils/__init__.py13
-rw-r--r--python/vyos/utils/auth.py (renamed from python/vyos/authutils.py)2
-rw-r--r--python/vyos/utils/boot.py35
-rw-r--r--python/vyos/utils/commit.py60
-rw-r--r--python/vyos/utils/convert.py30
-rw-r--r--python/vyos/utils/dict.py39
-rw-r--r--python/vyos/utils/file.py12
-rw-r--r--python/vyos/utils/kernel.py27
-rw-r--r--python/vyos/utils/list.py20
-rw-r--r--python/vyos/utils/misc.py66
-rw-r--r--python/vyos/utils/network.py185
-rw-r--r--python/vyos/utils/permission.py78
-rw-r--r--python/vyos/utils/process.py232
-rw-r--r--python/vyos/utils/system.py107
-rw-r--r--python/vyos/validate.py7
-rw-r--r--python/vyos/version.py12
-rw-r--r--python/vyos/xml_ref/__init__.py3
-rw-r--r--python/vyos/xml_ref/definition.py65
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