summaryrefslogtreecommitdiff
path: root/python
diff options
context:
space:
mode:
Diffstat (limited to 'python')
-rw-r--r--python/vyos/base.py9
-rw-r--r--python/vyos/configdict.py17
-rw-r--r--python/vyos/configdiff.py30
-rw-r--r--python/vyos/configquery.py46
-rw-r--r--python/vyos/configsource.py5
-rw-r--r--python/vyos/defaults.py7
-rw-r--r--python/vyos/firewall.py235
-rw-r--r--python/vyos/frr.py66
-rw-r--r--python/vyos/hostsd_client.py12
-rw-r--r--python/vyos/ifconfig/__init__.py1
-rw-r--r--python/vyos/ifconfig/ethernet.py17
-rwxr-xr-xpython/vyos/ifconfig/interface.py92
-rw-r--r--python/vyos/ifconfig/section.py10
-rw-r--r--python/vyos/ifconfig/vxlan.py11
-rw-r--r--python/vyos/ifconfig/wwan.py17
-rw-r--r--python/vyos/range_regex.py142
-rw-r--r--python/vyos/remote.py563
-rw-r--r--python/vyos/template.py78
-rw-r--r--python/vyos/util.py70
19 files changed, 1023 insertions, 405 deletions
diff --git a/python/vyos/base.py b/python/vyos/base.py
index 4e23714e5..c78045548 100644
--- a/python/vyos/base.py
+++ b/python/vyos/base.py
@@ -1,4 +1,4 @@
-# Copyright 2018 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2018-2021 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
@@ -13,6 +13,11 @@
# 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 textwrap import fill
class ConfigError(Exception):
- pass
+ def __init__(self, message):
+ # Reformat the message and trim it to 72 characters in length
+ message = fill(message, width=72)
+ # Call the base class constructor with the parameters it needs
+ super().__init__(message)
diff --git a/python/vyos/configdict.py b/python/vyos/configdict.py
index 425a2e416..d974a7565 100644
--- a/python/vyos/configdict.py
+++ b/python/vyos/configdict.py
@@ -459,7 +459,10 @@ def get_interface_dict(config, base, ifname=''):
# Only add defaults if interface is not about to be deleted - this is
# to keep a cleaner config dict.
if 'deleted' not in dict:
- dict['vif'][vif] = dict_merge(default_vif_values, vif_config)
+ address = leaf_node_changed(config, ['vif', vif, 'address'])
+ if address: dict['vif'][vif].update({'address_old' : address})
+
+ dict['vif'][vif] = dict_merge(default_vif_values, dict['vif'][vif])
# XXX: T2665: blend in proper DHCPv6-PD default values
dict['vif'][vif] = T2665_set_dhcpv6pd_defaults(dict['vif'][vif])
@@ -480,7 +483,11 @@ def get_interface_dict(config, base, ifname=''):
# Only add defaults if interface is not about to be deleted - this is
# to keep a cleaner config dict.
if 'deleted' not in dict:
- dict['vif_s'][vif_s] = dict_merge(default_vif_s_values, vif_s_config)
+ address = leaf_node_changed(config, ['vif-s', vif_s, 'address'])
+ if address: dict['vif_s'][vif_s].update({'address_old' : address})
+
+ dict['vif_s'][vif_s] = dict_merge(default_vif_s_values,
+ dict['vif_s'][vif_s])
# XXX: T2665: blend in proper DHCPv6-PD default values
dict['vif_s'][vif_s] = T2665_set_dhcpv6pd_defaults(dict['vif_s'][vif_s])
@@ -499,8 +506,12 @@ def get_interface_dict(config, base, ifname=''):
# Only add defaults if interface is not about to be deleted - this is
# to keep a cleaner config dict.
if 'deleted' not in dict:
+ address = leaf_node_changed(config, ['vif-s', vif_s, 'vif-c', vif_c, 'address'])
+ if address: dict['vif_s'][vif_s]['vif_c'][vif_c].update(
+ {'address_old' : address})
+
dict['vif_s'][vif_s]['vif_c'][vif_c] = dict_merge(
- default_vif_c_values, vif_c_config)
+ default_vif_c_values, dict['vif_s'][vif_s]['vif_c'][vif_c])
# XXX: T2665: blend in proper DHCPv6-PD default values
dict['vif_s'][vif_s]['vif_c'][vif_c] = T2665_set_dhcpv6pd_defaults(
dict['vif_s'][vif_s]['vif_c'][vif_c])
diff --git a/python/vyos/configdiff.py b/python/vyos/configdiff.py
index 0e41fbe27..4ad7443d7 100644
--- a/python/vyos/configdiff.py
+++ b/python/vyos/configdiff.py
@@ -17,7 +17,9 @@ from enum import IntFlag, auto
from vyos.config import Config
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.xml import defaults
class ConfigDiffError(Exception):
@@ -134,6 +136,34 @@ class ConfigDiff(object):
self._key_mangling[1])
return config_dict
+ def get_child_nodes_diff_str(self, path=[]):
+ ret = {'add': {}, 'change': {}, 'delete': {}}
+
+ diff = self.get_child_nodes_diff(path,
+ expand_nodes=Diff.ADD | Diff.DELETE | Diff.MERGE | Diff.STABLE,
+ no_defaults=True)
+
+ def parse_dict(diff_dict, diff_type, prefix=[]):
+ for k, v in diff_dict.items():
+ if isinstance(v, dict):
+ parse_dict(v, diff_type, prefix + [k])
+ else:
+ path_str = ' '.join(prefix + [k])
+ if diff_type == 'add' or diff_type == 'delete':
+ if isinstance(v, list):
+ v = ', '.join(v)
+ ret[diff_type][path_str] = v
+ elif diff_type == 'merge':
+ old_value = dict_search_args(diff['stable'], *prefix, k)
+ if old_value and old_value != v:
+ ret['change'][path_str] = [old_value, v]
+
+ parse_dict(diff['merge'], 'merge')
+ parse_dict(diff['add'], 'add')
+ parse_dict(diff['delete'], 'delete')
+
+ return ret
+
def get_child_nodes_diff(self, path=[], expand_nodes=Diff(0), no_defaults=False):
"""
Args:
diff --git a/python/vyos/configquery.py b/python/vyos/configquery.py
index 1cdcbcf39..5b097b312 100644
--- a/python/vyos/configquery.py
+++ b/python/vyos/configquery.py
@@ -18,16 +18,15 @@ A small library that allows querying existence or value(s) of config
settings from op mode, and execution of arbitrary op mode commands.
'''
-import re
-import json
-from copy import deepcopy
+import os
from subprocess import STDOUT
-import vyos.util
-import vyos.xml
+from vyos.util import popen, boot_configuration_complete
from vyos.config import Config
-from vyos.configtree import ConfigTree
-from vyos.configsource import ConfigSourceSession
+from vyos.configsource import ConfigSourceSession, ConfigSourceString
+from vyos.defaults import directories
+
+config_file = os.path.join(directories['config'], 'config.boot')
class ConfigQueryError(Exception):
pass
@@ -58,21 +57,21 @@ class CliShellApiConfigQuery(GenericConfigQuery):
def exists(self, path: list):
cmd = ' '.join(path)
- (_, err) = vyos.util.popen(f'cli-shell-api existsActive {cmd}')
+ (_, err) = popen(f'cli-shell-api existsActive {cmd}')
if err:
return False
return True
def value(self, path: list):
cmd = ' '.join(path)
- (out, err) = vyos.util.popen(f'cli-shell-api returnActiveValue {cmd}')
+ (out, err) = popen(f'cli-shell-api returnActiveValue {cmd}')
if err:
raise ConfigQueryError('No value for given path')
return out
def values(self, path: list):
cmd = ' '.join(path)
- (out, err) = vyos.util.popen(f'cli-shell-api returnActiveValues {cmd}')
+ (out, err) = popen(f'cli-shell-api returnActiveValues {cmd}')
if err:
raise ConfigQueryError('No values for given path')
return out
@@ -81,25 +80,36 @@ class ConfigTreeQuery(GenericConfigQuery):
def __init__(self):
super().__init__()
- config_source = ConfigSourceSession()
- self.configtree = Config(config_source=config_source)
+ if boot_configuration_complete():
+ config_source = ConfigSourceSession()
+ self.config = Config(config_source=config_source)
+ else:
+ try:
+ with open(config_file) as f:
+ config_string = f.read()
+ except OSError as err:
+ raise ConfigQueryError('No config file available') from err
+
+ config_source = ConfigSourceString(running_config_text=config_string,
+ session_config_text=config_string)
+ self.config = Config(config_source=config_source)
def exists(self, path: list):
- return self.configtree.exists(path)
+ return self.config.exists(path)
def value(self, path: list):
- return self.configtree.return_value(path)
+ return self.config.return_value(path)
def values(self, path: list):
- return self.configtree.return_values(path)
+ return self.config.return_values(path)
def list_nodes(self, path: list):
- return self.configtree.list_nodes(path)
+ return self.config.list_nodes(path)
def get_config_dict(self, path=[], effective=False, key_mangling=None,
get_first_key=False, no_multi_convert=False,
no_tag_node_value_mangle=False):
- return self.configtree.get_config_dict(path, effective=effective,
+ return self.config.get_config_dict(path, effective=effective,
key_mangling=key_mangling, get_first_key=get_first_key,
no_multi_convert=no_multi_convert,
no_tag_node_value_mangle=no_tag_node_value_mangle)
@@ -110,7 +120,7 @@ class VbashOpRun(GenericOpRun):
def run(self, path: list, **kwargs):
cmd = ' '.join(path)
- (out, err) = vyos.util.popen(f'. /opt/vyatta/share/vyatta-op/functions/interpreter/vyatta-op-run; _vyatta_op_run {cmd}', stderr=STDOUT, **kwargs)
+ (out, err) = popen(f'/opt/vyatta/bin/vyatta-op-cmd-wrapper {cmd}', stderr=STDOUT, **kwargs)
if err:
raise ConfigQueryError(out)
return out
diff --git a/python/vyos/configsource.py b/python/vyos/configsource.py
index b0981d25e..510b5b65a 100644
--- a/python/vyos/configsource.py
+++ b/python/vyos/configsource.py
@@ -19,6 +19,7 @@ import re
import subprocess
from vyos.configtree import ConfigTree
+from vyos.util import boot_configuration_complete
class VyOSError(Exception):
"""
@@ -117,7 +118,7 @@ class ConfigSourceSession(ConfigSource):
# Running config can be obtained either from op or conf mode, it always succeeds
# once the config system is initialized during boot;
# before initialization, set to empty string
- if os.path.isfile('/tmp/vyos-config-status'):
+ if boot_configuration_complete():
try:
running_config_text = self._run([self._cli_shell_api, '--show-active-only', '--show-show-defaults', '--show-ignore-edit', 'showConfig'])
except VyOSError:
@@ -161,7 +162,7 @@ class ConfigSourceSession(ConfigSource):
if p.returncode != 0:
raise VyOSError()
else:
- return out.decode('ascii', 'ignore')
+ return out.decode()
def set_level(self, path):
"""
diff --git a/python/vyos/defaults.py b/python/vyos/defaults.py
index 00b14a985..c77b695bd 100644
--- a/python/vyos/defaults.py
+++ b/python/vyos/defaults.py
@@ -29,6 +29,8 @@ directories = {
"vyos_udev_dir": "/run/udev/vyos"
}
+config_status = '/tmp/vyos-config-status'
+
cfg_group = 'vyattacfg'
cfg_vintage = 'vyos'
@@ -44,8 +46,9 @@ https_data = {
api_data = {
'listen_address' : '127.0.0.1',
'port' : '8080',
- 'strict' : 'false',
- 'debug' : 'false',
+ 'socket' : False,
+ 'strict' : False,
+ 'debug' : False,
'api_keys' : [ {"id": "testapp", "key": "qwerty"} ]
}
diff --git a/python/vyos/firewall.py b/python/vyos/firewall.py
new file mode 100644
index 000000000..4993d855e
--- /dev/null
+++ b/python/vyos/firewall.py
@@ -0,0 +1,235 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2021 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import re
+
+from vyos.util import cmd
+from vyos.util import dict_search_args
+
+def find_nftables_rule(table, chain, rule_matches=[]):
+ # Find rule in table/chain that matches all criteria and return the handle
+ results = cmd(f'sudo nft -a list chain {table} {chain}').split("\n")
+ for line in results:
+ if all(rule_match in line for rule_match in rule_matches):
+ handle_search = re.search('handle (\d+)', line)
+ if handle_search:
+ return handle_search[1]
+ return None
+
+def remove_nftables_rule(table, chain, handle):
+ cmd(f'sudo nft delete rule {table} {chain} handle {handle}')
+
+# Functions below used by template generation
+
+def nft_action(vyos_action):
+ if vyos_action == 'accept':
+ return 'return'
+ return vyos_action
+
+def parse_rule(rule_conf, fw_name, rule_id, ip_name):
+ output = []
+ def_suffix = '6' if ip_name == 'ip6' else ''
+
+ if 'state' in rule_conf and rule_conf['state']:
+ states = ",".join([s for s, v in rule_conf['state'].items() if v == 'enable'])
+
+ if states:
+ output.append(f'ct state {{{states}}}')
+
+ if 'protocol' in rule_conf and rule_conf['protocol'] != 'all':
+ proto = rule_conf['protocol']
+ operator = ''
+ if proto[0] == '!':
+ operator = '!='
+ proto = proto[1:]
+ if proto == 'tcp_udp':
+ proto = '{tcp, udp}'
+ output.append(f'meta l4proto {operator} {proto}')
+
+ for side in ['destination', 'source']:
+ if side in rule_conf:
+ prefix = side[0]
+ side_conf = rule_conf[side]
+
+ if 'address' in side_conf:
+ suffix = side_conf['address']
+ if suffix[0] == '!':
+ suffix = f'!= {suffix[1:]}'
+ output.append(f'{ip_name} {prefix}addr {suffix}')
+
+ if 'mac_address' in side_conf:
+ suffix = side_conf["mac_address"]
+ if suffix[0] == '!':
+ suffix = f'!= {suffix[1:]}'
+ output.append(f'ether {prefix}addr {suffix}')
+
+ if 'port' in side_conf:
+ proto = rule_conf['protocol']
+ port = side_conf['port'].split(',')
+
+ ports = []
+ negated_ports = []
+
+ for p in port:
+ if p[0] == '!':
+ negated_ports.append(p[1:])
+ else:
+ ports.append(p)
+
+ if proto == 'tcp_udp':
+ proto = 'th'
+
+ if ports:
+ ports_str = ','.join(ports)
+ output.append(f'{proto} {prefix}port {{{ports_str}}}')
+
+ if negated_ports:
+ negated_ports_str = ','.join(negated_ports)
+ output.append(f'{proto} {prefix}port != {{{negated_ports_str}}}')
+
+ if 'group' in side_conf:
+ group = side_conf['group']
+ if 'address_group' in group:
+ group_name = group['address_group']
+ output.append(f'{ip_name} {prefix}addr $A{def_suffix}_{group_name}')
+ elif 'network_group' in group:
+ group_name = group['network_group']
+ output.append(f'{ip_name} {prefix}addr $N{def_suffix}_{group_name}')
+ if 'mac_group' in group:
+ group_name = group['mac_group']
+ output.append(f'ether {prefix}addr $M_{group_name}')
+ if 'port_group' in group:
+ proto = rule_conf['protocol']
+ group_name = group['port_group']
+
+ if proto == 'tcp_udp':
+ proto = 'th'
+
+ output.append(f'{proto} {prefix}port $P_{group_name}')
+
+ if 'log' in rule_conf and rule_conf['log'] == 'enable':
+ action = rule_conf['action'] if 'action' in rule_conf else 'accept'
+ output.append(f'log prefix "[{fw_name[:19]}-{rule_id}-{action[:1].upper()}] "')
+
+ if 'hop_limit' in rule_conf:
+ operators = {'eq': '==', 'gt': '>', 'lt': '<'}
+ for op, operator in operators.items():
+ if op in rule_conf['hop_limit']:
+ value = rule_conf['hop_limit'][op]
+ output.append(f'ip6 hoplimit {operator} {value}')
+
+ for icmp in ['icmp', 'icmpv6']:
+ if icmp in rule_conf:
+ if 'type_name' in rule_conf[icmp]:
+ output.append(icmp + ' type ' + rule_conf[icmp]['type_name'])
+ else:
+ if 'code' in rule_conf[icmp]:
+ output.append(icmp + ' code ' + rule_conf[icmp]['code'])
+ if 'type' in rule_conf[icmp]:
+ output.append(icmp + ' type ' + rule_conf[icmp]['type'])
+
+ if 'ipsec' in rule_conf:
+ if 'match_ipsec' in rule_conf['ipsec']:
+ output.append('meta ipsec == 1')
+ if 'match_non_ipsec' in rule_conf['ipsec']:
+ output.append('meta ipsec == 0')
+
+ if 'fragment' in rule_conf:
+ # Checking for fragmentation after priority -400 is not possible,
+ # so we use a priority -450 hook to set a mark
+ if 'match_frag' in rule_conf['fragment']:
+ output.append('meta mark 0xffff1')
+ if 'match_non_frag' in rule_conf['fragment']:
+ output.append('meta mark != 0xffff1')
+
+ if 'limit' in rule_conf:
+ if 'rate' in rule_conf['limit']:
+ output.append(f'limit rate {rule_conf["limit"]["rate"]}/second')
+ if 'burst' in rule_conf['limit']:
+ output.append(f'burst {rule_conf["limit"]["burst"]} packets')
+
+ if 'recent' in rule_conf:
+ count = rule_conf['recent']['count']
+ time = rule_conf['recent']['time']
+ # output.append(f'meter {fw_name}_{rule_id} {{ ip saddr and 255.255.255.255 limit rate over {count}/{time} burst {count} packets }}')
+ # Waiting on input from nftables developers due to
+ # bug with above line and atomic chain flushing.
+
+ if 'time' in rule_conf:
+ output.append(parse_time(rule_conf['time']))
+
+ tcp_flags = dict_search_args(rule_conf, 'tcp', 'flags')
+ if tcp_flags:
+ output.append(parse_tcp_flags(tcp_flags))
+
+ output.append('counter')
+
+ if 'set' in rule_conf:
+ output.append(parse_policy_set(rule_conf['set'], def_suffix))
+
+ if 'action' in rule_conf:
+ output.append(nft_action(rule_conf['action']))
+ else:
+ output.append('return')
+
+ output.append(f'comment "{fw_name}-{rule_id}"')
+ return " ".join(output)
+
+def parse_tcp_flags(flags):
+ include = [flag for flag in flags if flag != 'not']
+ exclude = flags['not'].keys() if 'not' in flags else []
+ return f'tcp flags & ({"|".join(include + exclude)}) == {"|".join(include)}'
+
+def parse_time(time):
+ out = []
+ if 'startdate' in time:
+ start = time['startdate']
+ if 'T' not in start and 'starttime' in time:
+ start += f' {time["starttime"]}'
+ out.append(f'time >= "{start}"')
+ if 'starttime' in time and 'startdate' not in time:
+ out.append(f'hour >= "{time["starttime"]}"')
+ if 'stopdate' in time:
+ stop = time['stopdate']
+ if 'T' not in stop and 'stoptime' in time:
+ stop += f' {time["stoptime"]}'
+ out.append(f'time < "{stop}"')
+ if 'stoptime' in time and 'stopdate' not in time:
+ out.append(f'hour < "{time["stoptime"]}"')
+ if 'weekdays' in time:
+ days = time['weekdays'].split(",")
+ out_days = [f'"{day}"' for day in days if day[0] != '!']
+ out.append(f'day {{{",".join(out_days)}}}')
+ return " ".join(out)
+
+def parse_policy_set(set_conf, def_suffix):
+ out = []
+ if 'dscp' in set_conf:
+ dscp = set_conf['dscp']
+ out.append(f'ip{def_suffix} dscp set {dscp}')
+ if 'mark' in set_conf:
+ mark = set_conf['mark']
+ out.append(f'meta mark set {mark}')
+ if 'table' in set_conf:
+ table = set_conf['table']
+ if table == 'main':
+ table = '254'
+ mark = 0x7FFFFFFF - int(table)
+ out.append(f'meta mark set {mark}')
+ if 'tcp_mss' in set_conf:
+ mss = set_conf['tcp_mss']
+ out.append(f'tcp option maxseg size set {mss}')
+ return " ".join(out)
diff --git a/python/vyos/frr.py b/python/vyos/frr.py
index df6849472..cbba19ab7 100644
--- a/python/vyos/frr.py
+++ b/python/vyos/frr.py
@@ -73,23 +73,25 @@ from vyos.util import cmd
import logging
from logging.handlers import SysLogHandler
import os
+import sys
+
LOG = logging.getLogger(__name__)
+DEBUG = False
-DEBUG = os.path.exists('/tmp/vyos.frr.debug')
-if DEBUG:
- LOG.setLevel(logging.DEBUG)
- ch = SysLogHandler(address='/dev/log')
- ch2 = logging.StreamHandler()
- LOG.addHandler(ch)
- LOG.addHandler(ch2)
+ch = SysLogHandler(address='/dev/log')
+ch2 = logging.StreamHandler(stream=sys.stdout)
+LOG.addHandler(ch)
+LOG.addHandler(ch2)
_frr_daemons = ['zebra', 'bgpd', 'fabricd', 'isisd', 'ospf6d', 'ospfd', 'pbrd',
- 'pimd', 'ripd', 'ripngd', 'sharpd', 'staticd', 'vrrpd', 'ldpd']
+ 'pimd', 'ripd', 'ripngd', 'sharpd', 'staticd', 'vrrpd', 'ldpd',
+ 'bfdd']
path_vtysh = '/usr/bin/vtysh'
path_frr_reload = '/usr/lib/frr/frr-reload.py'
path_config = '/run/frr'
+default_add_before = r'(ip prefix-list .*|route-map .*|line vty|end)'
class FrrError(Exception):
pass
@@ -119,6 +121,12 @@ class ConfigSectionNotFound(FrrError):
"""
pass
+def init_debugging():
+ global DEBUG
+
+ DEBUG = os.path.exists('/tmp/vyos.frr.debug')
+ if DEBUG:
+ LOG.setLevel(logging.DEBUG)
def get_configuration(daemon=None, marked=False):
""" Get current running FRR configuration
@@ -214,13 +222,8 @@ def reload_configuration(config, daemon=None):
def save_configuration():
- """Save FRR configuration to /run/frr/config/frr.conf
- It save configuration on each commit. T3217
- """
-
- cmd(f'{path_vtysh} -n -w')
-
- return
+ """ T3217: Save FRR configuration to /run/frr/config/frr.conf """
+ return cmd(f'{path_vtysh} -n -w')
def execute(command):
@@ -427,6 +430,8 @@ class FRRConfig:
Using this overwrites the current loaded config objects and replaces the original loaded config
'''
+ init_debugging()
+
self.imported_config = get_configuration(daemon=daemon)
if daemon:
LOG.debug(f'load_configuration: Configuration loaded from FRR daemon {daemon}')
@@ -448,16 +453,37 @@ class FRRConfig:
mark_configuration('\n'.join(self.config))
def commit_configuration(self, daemon=None):
- '''Commit the current configuration to FRR
- daemon: str with name of the FRR daemon to commit to or
- None to use the consolidated config
+ '''
+ Commit the current configuration to FRR daemon: str with name of the
+ FRR daemon to commit to or None to use the consolidated config.
+
+ Configuration is automatically saved after apply
'''
LOG.debug('commit_configuration: Commiting configuration')
for i, e in enumerate(self.config):
LOG.debug(f'commit_configuration: new_config {i:3} {e}')
- reload_configuration('\n'.join(self.config), daemon=daemon)
- def modify_section(self, start_pattern, replacement=[], stop_pattern=r'\S+', remove_stop_mark=False, count=0):
+ # https://github.com/FRRouting/frr/issues/10132
+ # https://github.com/FRRouting/frr/issues/10133
+ count = 0
+ count_max = 5
+ while count < count_max:
+ count += 1
+ try:
+ reload_configuration('\n'.join(self.config), daemon=daemon)
+ break
+ except:
+ # we just need to re-try the commit of the configuration
+ # for the listed FRR issues above
+ pass
+ if count >= count_max:
+ raise ConfigurationNotValid(f'Config commit retry counter ({count_max}) exceeded')
+
+ # Save configuration to /run/frr/config/frr.conf
+ save_configuration()
+
+
+ def modify_section(self, start_pattern, replacement='!', stop_pattern=r'\S+', remove_stop_mark=False, count=0):
if isinstance(replacement, str):
replacement = replacement.split('\n')
elif not isinstance(replacement, list):
diff --git a/python/vyos/hostsd_client.py b/python/vyos/hostsd_client.py
index 303b6ea47..f31ef51cf 100644
--- a/python/vyos/hostsd_client.py
+++ b/python/vyos/hostsd_client.py
@@ -79,6 +79,18 @@ class Client(object):
msg = {'type': 'forward_zones', 'op': 'get'}
return self._communicate(msg)
+ def add_authoritative_zones(self, data):
+ msg = {'type': 'authoritative_zones', 'op': 'add', 'data': data}
+ self._communicate(msg)
+
+ def delete_authoritative_zones(self, data):
+ msg = {'type': 'authoritative_zones', 'op': 'delete', 'data': data}
+ self._communicate(msg)
+
+ def get_authoritative_zones(self):
+ msg = {'type': 'authoritative_zones', 'op': 'get'}
+ return self._communicate(msg)
+
def add_search_domains(self, data):
msg = {'type': 'search_domains', 'op': 'add', 'data': data}
self._communicate(msg)
diff --git a/python/vyos/ifconfig/__init__.py b/python/vyos/ifconfig/__init__.py
index 2d3e406ac..a37615c8f 100644
--- a/python/vyos/ifconfig/__init__.py
+++ b/python/vyos/ifconfig/__init__.py
@@ -26,6 +26,7 @@ from vyos.ifconfig.ethernet import EthernetIf
from vyos.ifconfig.geneve import GeneveIf
from vyos.ifconfig.loopback import LoopbackIf
from vyos.ifconfig.macvlan import MACVLANIf
+from vyos.ifconfig.input import InputIf
from vyos.ifconfig.vxlan import VXLANIf
from vyos.ifconfig.wireguard import WireGuardIf
from vyos.ifconfig.vtun import VTunIf
diff --git a/python/vyos/ifconfig/ethernet.py b/python/vyos/ifconfig/ethernet.py
index 2e59a7afc..9d54dc78e 100644
--- a/python/vyos/ifconfig/ethernet.py
+++ b/python/vyos/ifconfig/ethernet.py
@@ -80,6 +80,23 @@ class EthernetIf(Interface):
super().__init__(ifname, **kargs)
self.ethtool = Ethtool(ifname)
+ def remove(self):
+ """
+ Remove interface from config. Removing the interface deconfigures all
+ assigned IP addresses.
+ Example:
+ >>> from vyos.ifconfig import WWANIf
+ >>> i = EthernetIf('eth0')
+ >>> i.remove()
+ """
+
+ if self.exists(self.ifname):
+ # interface is placed in A/D state when removed from config! It
+ # will remain visible for the operating system.
+ self.set_admin_state('down')
+
+ super().remove()
+
def set_flow_control(self, enable):
"""
Changes the pause parameters of the specified Ethernet device.
diff --git a/python/vyos/ifconfig/interface.py b/python/vyos/ifconfig/interface.py
index 58d130ef6..91c7f0c33 100755
--- a/python/vyos/ifconfig/interface.py
+++ b/python/vyos/ifconfig/interface.py
@@ -37,6 +37,7 @@ 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.template import is_ipv4
from vyos.template import is_ipv6
@@ -135,6 +136,9 @@ class Interface(Control):
'validate': assert_mtu,
'shellcmd': 'ip link set dev {ifname} mtu {value}',
},
+ 'netns': {
+ 'shellcmd': 'ip link set dev {ifname} netns {value}',
+ },
'vrf': {
'convert': lambda v: f'master {v}' if v else 'nomaster',
'shellcmd': 'ip link set dev {ifname} {value}',
@@ -512,6 +516,35 @@ class Interface(Control):
if prev_state == 'up':
self.set_admin_state('up')
+ def del_netns(self, netns):
+ """
+ Remove interface from given NETNS.
+ """
+
+ # If NETNS does not exist then there is nothing to delete
+ if not os.path.exists(f'/run/netns/{netns}'):
+ return None
+
+ # As a PoC we only allow 'dummy' interfaces
+ if 'dum' not in self.ifname:
+ return None
+
+ # Check if interface realy exists in namespace
+ if get_interface_namespace(self.ifname) != None:
+ self._cmd(f'ip netns exec {get_interface_namespace(self.ifname)} ip link del dev {self.ifname}')
+ return
+
+ def set_netns(self, netns):
+ """
+ Add interface from given NETNS.
+
+ Example:
+ >>> from vyos.ifconfig import Interface
+ >>> Interface('dum0').set_netns('foo')
+ """
+
+ self.set_interface('netns', netns)
+
def set_vrf(self, vrf):
"""
Add/Remove interface from given VRF instance.
@@ -544,6 +577,15 @@ class Interface(Control):
return None
return self.set_interface('arp_cache_tmo', tmo)
+ def _cleanup_mss_rules(self, table, ifname):
+ commands = []
+ results = self._cmd(f'nft -a list chain {table} VYOS_TCP_MSS').split("\n")
+ for line in results:
+ if f'oifname "{ifname}"' in line:
+ handle_search = re.search('handle (\d+)', line)
+ if handle_search:
+ self._cmd(f'nft delete rule {table} VYOS_TCP_MSS handle {handle_search[1]}')
+
def set_tcp_ipv4_mss(self, mss):
"""
Set IPv4 TCP MSS value advertised when TCP SYN packets leave this
@@ -555,22 +597,14 @@ class Interface(Control):
>>> from vyos.ifconfig import Interface
>>> Interface('eth0').set_tcp_ipv4_mss(1340)
"""
- iptables_bin = 'iptables'
- base_options = f'-A FORWARD -o {self.ifname} -p tcp -m tcp --tcp-flags SYN,RST SYN'
- out = self._cmd(f'{iptables_bin}-save -t mangle')
- for line in out.splitlines():
- if line.startswith(base_options):
- # remove OLD MSS mangling configuration
- line = line.replace('-A FORWARD', '-D FORWARD')
- self._cmd(f'{iptables_bin} -t mangle {line}')
-
- cmd_mss = f'{iptables_bin} -t mangle {base_options} --jump TCPMSS'
+ self._cleanup_mss_rules('raw', self.ifname)
+ nft_prefix = 'nft add rule raw VYOS_TCP_MSS'
+ base_cmd = f'oifname "{self.ifname}" tcp flags & (syn|rst) == syn'
if mss == 'clamp-mss-to-pmtu':
- self._cmd(f'{cmd_mss} --clamp-mss-to-pmtu')
+ self._cmd(f"{nft_prefix} '{base_cmd} tcp option maxseg size set rt mtu'")
elif int(mss) > 0:
- # probably add option to clamp only if bigger:
low_mss = str(int(mss) + 1)
- self._cmd(f'{cmd_mss} -m tcpmss --mss {low_mss}:65535 --set-mss {mss}')
+ self._cmd(f"{nft_prefix} '{base_cmd} tcp option maxseg size {low_mss}-65535 tcp option maxseg size set {mss}'")
def set_tcp_ipv6_mss(self, mss):
"""
@@ -583,22 +617,14 @@ class Interface(Control):
>>> from vyos.ifconfig import Interface
>>> Interface('eth0').set_tcp_mss(1320)
"""
- iptables_bin = 'ip6tables'
- base_options = f'-A FORWARD -o {self.ifname} -p tcp -m tcp --tcp-flags SYN,RST SYN'
- out = self._cmd(f'{iptables_bin}-save -t mangle')
- for line in out.splitlines():
- if line.startswith(base_options):
- # remove OLD MSS mangling configuration
- line = line.replace('-A FORWARD', '-D FORWARD')
- self._cmd(f'{iptables_bin} -t mangle {line}')
-
- cmd_mss = f'{iptables_bin} -t mangle {base_options} --jump TCPMSS'
+ self._cleanup_mss_rules('ip6 raw', self.ifname)
+ nft_prefix = 'nft add rule ip6 raw VYOS_TCP_MSS'
+ base_cmd = f'oifname "{self.ifname}" tcp flags & (syn|rst) == syn'
if mss == 'clamp-mss-to-pmtu':
- self._cmd(f'{cmd_mss} --clamp-mss-to-pmtu')
+ self._cmd(f"{nft_prefix} '{base_cmd} tcp option maxseg size set rt mtu'")
elif int(mss) > 0:
- # probably add option to clamp only if bigger:
low_mss = str(int(mss) + 1)
- self._cmd(f'{cmd_mss} -m tcpmss --mss {low_mss}:65535 --set-mss {mss}')
+ self._cmd(f"{nft_prefix} '{base_cmd} tcp option maxseg size {low_mss}-65535 tcp option maxseg size set {mss}'")
def set_arp_filter(self, arp_filter):
"""
@@ -1271,8 +1297,8 @@ class Interface(Control):
source_if = next(iter(self._config['is_mirror_intf']))
config = self._config['is_mirror_intf'][source_if].get('mirror', None)
- # Check configuration stored by old perl code before delete T3782
- if not 'redirect' in self._config:
+ # Check configuration stored by old perl code before delete T3782/T4056
+ if not 'redirect' in self._config and not 'traffic_policy' in self._config:
# Please do not clear the 'set $? = 0 '. It's meant to force a return of 0
# Remove existing mirroring rules
delete_tc_cmd = f'tc qdisc del dev {source_if} handle ffff: ingress 2> /dev/null;'
@@ -1353,6 +1379,16 @@ class Interface(Control):
if mac:
self.set_mac(mac)
+ # If interface is connected to NETNS we don't have to check all other
+ # settings like MTU/IPv6/sysctl values, etc.
+ # Since the interface is pushed onto a separate logical stack
+ # Configure NETNS
+ if dict_search('netns', config) != None:
+ self.set_netns(config.get('netns', ''))
+ return
+ else:
+ self.del_netns(config.get('netns', ''))
+
# Update interface description
self.set_alias(config.get('description', ''))
diff --git a/python/vyos/ifconfig/section.py b/python/vyos/ifconfig/section.py
index 0e4447b9e..91f667b65 100644
--- a/python/vyos/ifconfig/section.py
+++ b/python/vyos/ifconfig/section.py
@@ -52,12 +52,12 @@ class Section:
name: name of the interface
vlan: if vlan is True, do not stop at the vlan number
"""
- name = name.rstrip('0123456789')
- name = name.rstrip('.')
- if vlan:
- name = name.rstrip('0123456789.')
if vrrp:
- name = name.rstrip('0123456789v')
+ name = re.sub(r'\d(\d|v|\.)*$', '', name)
+ elif vlan:
+ name = re.sub(r'\d(\d|\.)*$', '', name)
+ else:
+ name = re.sub(r'\d+$', '', name)
return name
@classmethod
diff --git a/python/vyos/ifconfig/vxlan.py b/python/vyos/ifconfig/vxlan.py
index d73fb47b8..0c5282db4 100644
--- a/python/vyos/ifconfig/vxlan.py
+++ b/python/vyos/ifconfig/vxlan.py
@@ -54,18 +54,21 @@ class VXLANIf(Interface):
# arguments used by iproute2. For more information please refer to:
# - https://man7.org/linux/man-pages/man8/ip-link.8.html
mapping = {
- 'source_address' : 'local',
- 'source_interface' : 'dev',
- 'remote' : 'remote',
'group' : 'group',
+ 'external' : 'external',
+ 'gpe' : 'gpe',
'parameters.ip.dont_fragment': 'df set',
'parameters.ip.tos' : 'tos',
'parameters.ip.ttl' : 'ttl',
'parameters.ipv6.flowlabel' : 'flowlabel',
'parameters.nolearning' : 'nolearning',
+ 'remote' : 'remote',
+ 'source_address' : 'local',
+ 'source_interface' : 'dev',
+ 'vni' : 'id',
}
- cmd = 'ip link add {ifname} type {type} id {vni} dstport {port}'
+ cmd = 'ip link add {ifname} type {type} dstport {port}'
for vyos_key, iproute2_key in mapping.items():
# dict_search will return an empty dict "{}" for valueless nodes like
# "parameters.nolearning" - thus we need to test the nodes existence
diff --git a/python/vyos/ifconfig/wwan.py b/python/vyos/ifconfig/wwan.py
index f18959a60..845c9bef9 100644
--- a/python/vyos/ifconfig/wwan.py
+++ b/python/vyos/ifconfig/wwan.py
@@ -26,3 +26,20 @@ class WWANIf(Interface):
'eternal': 'wwan[0-9]+$',
},
}
+
+ def remove(self):
+ """
+ Remove interface from config. Removing the interface deconfigures all
+ assigned IP addresses.
+ Example:
+ >>> from vyos.ifconfig import WWANIf
+ >>> i = WWANIf('wwan0')
+ >>> i.remove()
+ """
+
+ if self.exists(self.ifname):
+ # interface is placed in A/D state when removed from config! It
+ # will remain visible for the operating system.
+ self.set_admin_state('down')
+
+ super().remove()
diff --git a/python/vyos/range_regex.py b/python/vyos/range_regex.py
new file mode 100644
index 000000000..a8190d140
--- /dev/null
+++ b/python/vyos/range_regex.py
@@ -0,0 +1,142 @@
+'''Copyright (c) 2013, Dmitry Voronin
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+this list of conditions and the following disclaimer in the documentation
+and/or other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+'''
+import math
+
+# coding=utf8
+
+# Split range to ranges that has its unique pattern.
+# Example for 12-345:
+#
+# 12- 19: 1[2-9]
+# 20- 99: [2-9]\d
+# 100-299: [1-2]\d{2}
+# 300-339: 3[0-3]\d
+# 340-345: 34[0-5]
+
+def range_to_regex(inpt_range):
+ if isinstance(inpt_range, str):
+ range_list = inpt_range.split('-')
+ # Check input arguments
+ if len(range_list) == 2:
+ # The first element in range must be higher then the second
+ if int(range_list[0]) < int(range_list[1]):
+ return regex_for_range(int(range_list[0]), int(range_list[1]))
+
+ return None
+
+def bounded_regex_for_range(min_, max_):
+ return r'\b({})\b'.format(regex_for_range(min_, max_))
+
+def regex_for_range(min_, max_):
+ """
+ > regex_for_range(12, 345)
+ '1[2-9]|[2-9]\d|[1-2]\d{2}|3[0-3]\d|34[0-5]'
+ """
+ positive_subpatterns = []
+ negative_subpatterns = []
+
+ if min_ < 0:
+ min__ = 1
+ if max_ < 0:
+ min__ = abs(max_)
+ max__ = abs(min_)
+
+ negative_subpatterns = split_to_patterns(min__, max__)
+ min_ = 0
+
+ if max_ >= 0:
+ positive_subpatterns = split_to_patterns(min_, max_)
+
+ negative_only_subpatterns = ['-' + val for val in negative_subpatterns if val not in positive_subpatterns]
+ positive_only_subpatterns = [val for val in positive_subpatterns if val not in negative_subpatterns]
+ intersected_subpatterns = ['-?' + val for val in negative_subpatterns if val in positive_subpatterns]
+
+ subpatterns = negative_only_subpatterns + intersected_subpatterns + positive_only_subpatterns
+ return '|'.join(subpatterns)
+
+
+def split_to_patterns(min_, max_):
+ subpatterns = []
+
+ start = min_
+ for stop in split_to_ranges(min_, max_):
+ subpatterns.append(range_to_pattern(start, stop))
+ start = stop + 1
+
+ return subpatterns
+
+
+def split_to_ranges(min_, max_):
+ stops = {max_}
+
+ nines_count = 1
+ stop = fill_by_nines(min_, nines_count)
+ while min_ <= stop < max_:
+ stops.add(stop)
+
+ nines_count += 1
+ stop = fill_by_nines(min_, nines_count)
+
+ zeros_count = 1
+ stop = fill_by_zeros(max_ + 1, zeros_count) - 1
+ while min_ < stop <= max_:
+ stops.add(stop)
+
+ zeros_count += 1
+ stop = fill_by_zeros(max_ + 1, zeros_count) - 1
+
+ stops = list(stops)
+ stops.sort()
+
+ return stops
+
+
+def fill_by_nines(integer, nines_count):
+ return int(str(integer)[:-nines_count] + '9' * nines_count)
+
+
+def fill_by_zeros(integer, zeros_count):
+ return integer - integer % 10 ** zeros_count
+
+
+def range_to_pattern(start, stop):
+ pattern = ''
+ any_digit_count = 0
+
+ for start_digit, stop_digit in zip(str(start), str(stop)):
+ if start_digit == stop_digit:
+ pattern += start_digit
+ elif start_digit != '0' or stop_digit != '9':
+ pattern += '[{}-{}]'.format(start_digit, stop_digit)
+ else:
+ any_digit_count += 1
+
+ if any_digit_count:
+ pattern += r'\d'
+
+ if any_digit_count > 1:
+ pattern += '{{{}}}'.format(any_digit_count)
+
+ return pattern \ No newline at end of file
diff --git a/python/vyos/remote.py b/python/vyos/remote.py
index e972050b7..66044fa52 100644
--- a/python/vyos/remote.py
+++ b/python/vyos/remote.py
@@ -13,38 +13,40 @@
# 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 ftplib import FTP
import os
import shutil
import socket
+import ssl
import stat
import sys
import tempfile
import urllib.parse
-import urllib.request as urlreq
-from vyos.template import get_ip
-from vyos.template import ip_from_cidr
-from vyos.template import is_interface
-from vyos.template import is_ipv6
-from vyos.util import cmd
+from ftplib import FTP
+from ftplib import FTP_TLS
+
+from paramiko import SSHClient
+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 print_error
-from vyos.util import make_progressbar
+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.version import get_version
-from paramiko import SSHClient
-from paramiko import SSHException
-from paramiko import MissingHostKeyPolicy
-# This is a hardcoded path and no environment variable can change it.
-KNOWN_HOSTS_FILE = os.path.expanduser('~/.ssh/known_hosts')
+
CHUNK_SIZE = 8192
class InteractivePolicy(MissingHostKeyPolicy):
"""
- Policy for interactively querying the user on whether to proceed with
- SSH connections to unknown hosts.
+ Paramiko policy for interactively querying the user on whether to proceed
+ with SSH connections to unknown hosts.
"""
def missing_host_key(self, client, hostname, key):
print_error(f"Host '{hostname}' not found in known hosts.")
@@ -57,347 +59,270 @@ class InteractivePolicy(MissingHostKeyPolicy):
else:
raise SSHException(f"Cannot connect to unknown host '{hostname}'.")
-
-## Helper routines
-def get_authentication_variables(default_username=None, default_password=None):
+class SourceAdapter(HTTPAdapter):
"""
- Return the environment variables `$REMOTE_USERNAME` and `$REMOTE_PASSWORD` and
- return the defaults provided if environment variables are empty or nonexistent.
+ urllib3 transport adapter for setting source addresses per session.
"""
- username, password = os.getenv('REMOTE_USERNAME'), os.getenv('REMOTE_PASSWORD')
- # Fall back to defaults if the username variable doesn't exist or is an empty string.
- # Note that this is different from `os.getenv('REMOTE_USERNAME', default=default_username)`,
- # as we want the username and the password to have the same behaviour.
- if not username:
- return default_username, default_password
- else:
- return username, password
+ def __init__(self, source_pair, *args, **kwargs):
+ # A source pair is a tuple of a source host string and source port respectively.
+ # Supply '' and 0 respectively for default values.
+ self._source_pair = source_pair
+ super(SourceAdapter, self).__init__(*args, **kwargs)
-def get_source_address(source):
- """
- Take a string vaguely indicating an origin source (interface, hostname or IP address),
- return a tuple in the format `(source_pair, address_family)` where
- `source_pair` is `(source_address, source_port)`.
- """
- # TODO: Properly distinguish between IPv4 and IPv6.
- port = 0
- if is_interface(source):
- source = ip_from_cidr(get_ip(source)[0])
- if is_ipv6(source):
- return (source, port), socket.AF_INET6
- else:
- return (socket.gethostbyname(source), port), socket.AF_INET
+ def init_poolmanager(self, connections, maxsize, block=False):
+ self.poolmanager = PoolManager(
+ num_pools=connections, maxsize=maxsize,
+ block=block, source_address=self._source_pair)
-def get_port_from_url(url):
+
+def check_storage(path, size):
"""
- Return the port number from the given `url` named tuple, fall back to
- the default if there isn't one.
+ Check whether `path` has enough storage space for a transfer of `size` bytes.
"""
- defaults = {"http": 80, "https": 443, "ftp": 21, "tftp": 69,\
- "ssh": 22, "scp": 22, "sftp": 22}
- if url.port:
- return url.port
+ path = os.path.abspath(os.path.expanduser(path))
+ directory = path if os.path.isdir(path) else (os.path.dirname(os.path.expanduser(path)) or os.getcwd())
+ # `size` can be None or 0 to indicate unknown size.
+ if not size:
+ print_error('Warning: Cannot determine size of remote file. Bravely continuing regardless.')
+ return
+
+ if size < 1024 * 1024:
+ print_error(f'The file is {size / 1024.0:.3f} KiB.')
else:
- return defaults[url.scheme]
-
-
-## FTP routines
-def upload_ftp(local_path, hostname, remote_path,\
- username='anonymous', password='', port=21,\
- source_pair=None, progressbar=False):
- size = os.path.getsize(local_path)
- with FTP(source_address=source_pair) as conn:
- conn.connect(hostname, port)
- conn.login(username, password)
- with open(local_path, 'rb') as file:
- if progressbar and size:
+ print_error(f'The file is {size / (1024.0 * 1024.0):.3f} MiB.')
+
+ # Will throw `FileNotFoundError' if `directory' is absent.
+ if size > shutil.disk_usage(directory).free:
+ raise OSError(f'Not enough disk space available in "{directory}".')
+
+
+class FtpC:
+ def __init__(self, url, progressbar=False, check_space=False, source_host='', source_port=0):
+ self.secure = url.scheme == 'ftps'
+ self.hostname = url.hostname
+ self.path = url.path
+ self.username = url.username or os.getenv('REMOTE_USERNAME', 'anonymous')
+ self.password = url.password or os.getenv('REMOTE_PASSWORD', '')
+ self.port = url.port or 21
+ self.source = (source_host, source_port)
+ self.progressbar = progressbar
+ self.check_space = check_space
+
+ def _establish(self):
+ if self.secure:
+ return FTP_TLS(source_address=self.source, context=ssl.create_default_context())
+ else:
+ return FTP(source_address=self.source)
+
+ def download(self, location: str):
+ # Open the file upfront before establishing connection.
+ with open(location, 'wb') as f, self._establish() as conn:
+ conn.connect(self.hostname, self.port)
+ conn.login(self.username, self.password)
+ # Set secure connection over TLS.
+ if self.secure:
+ conn.prot_p()
+ # Almost all FTP servers support the `SIZE' command.
+ if self.check_space:
+ check_storage(path, conn.size(self.path))
+ # No progressbar if we can't determine the size or if the file is too small.
+ if self.progressbar and size and size > CHUNK_SIZE:
progress = make_incremental_progressbar(CHUNK_SIZE / size)
next(progress)
- callback = lambda block: next(progress)
+ callback = lambda block: begin(f.write(block), next(progress))
else:
- callback = None
- conn.storbinary(f'STOR {remote_path}', file, CHUNK_SIZE, callback)
-
-def download_ftp(local_path, hostname, remote_path,\
- username='anonymous', password='', port=21,\
- source_pair=None, progressbar=False):
- with FTP(source_address=source_pair) as conn:
- conn.connect(hostname, port)
- conn.login(username, password)
- size = conn.size(remote_path)
- with open(local_path, 'wb') as file:
- # No progressbar if we can't determine the size.
- if progressbar and size:
+ callback = f.write
+ conn.retrbinary('RETR ' + self.path, callback, CHUNK_SIZE)
+
+ def upload(self, location: str):
+ size = os.path.getsize(location)
+ with open(location, 'rb') as f, self._establish() as conn:
+ conn.connect(self.hostname, self.port)
+ conn.login(self.username, self.password)
+ if self.secure:
+ conn.prot_p()
+ if self.progressbar and size and size > CHUNK_SIZE:
progress = make_incremental_progressbar(CHUNK_SIZE / size)
next(progress)
- callback = lambda block: (file.write(block), next(progress))
+ callback = lambda block: next(progress)
else:
- callback = file.write
- conn.retrbinary(f'RETR {remote_path}', callback, CHUNK_SIZE)
-
-def get_ftp_file_size(hostname, remote_path,\
- username='anonymous', password='', port=21,\
- source_pair=None):
- with FTP(source_address=source) as conn:
- conn.connect(hostname, port)
- conn.login(username, password)
- size = conn.size(remote_path)
- if size:
- return size
- else:
- # SIZE is an extension to the FTP specification, although it's extremely common.
- raise ValueError('Failed to receive file size from FTP server. \
- Perhaps the server does not implement the SIZE command?')
-
-
-## SFTP/SCP routines
-def transfer_sftp(mode, local_path, hostname, remote_path,\
- username=None, password=None, port=22,\
- source_tuple=None, progressbar=False):
- sock = None
- if source_tuple:
- (source_address, source_port), address_family = source_tuple
- sock = socket.socket(address_family, socket.SOCK_STREAM)
- sock.bind((source_address, source_port))
- sock.connect((hostname, port))
- callback = make_progressbar() if progressbar else None
- with SSHClient() as ssh:
+ callback = None
+ conn.storbinary('STOR ' + self.path, f, CHUNK_SIZE, callback)
+
+class SshC:
+ known_hosts = os.path.expanduser('~/.ssh/known_hosts')
+ def __init__(self, url, progressbar=False, check_space=False, source_host='', source_port=0):
+ self.hostname = url.hostname
+ self.path = url.path
+ self.username = url.username or os.getenv('REMOTE_USERNAME')
+ self.password = url.password or os.getenv('REMOTE_PASSWORD')
+ self.port = url.port or 22
+ self.source = (source_host, source_port)
+ self.progressbar = progressbar
+ self.check_space = check_space
+
+ def _establish(self):
+ ssh = SSHClient()
ssh.load_system_host_keys()
- if os.path.exists(KNOWN_HOSTS_FILE):
- ssh.load_host_keys(KNOWN_HOSTS_FILE)
+ # Try to load from a user-local known hosts file if one exists.
+ if os.path.exists(self.known_hosts):
+ ssh.load_host_keys(self.known_hosts)
ssh.set_missing_host_key_policy(InteractivePolicy())
- ssh.connect(hostname, port, username, password, sock=sock)
- with ssh.open_sftp() as sftp:
- if mode == 'upload':
+ # `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)
+ ssh.connect(self.hostname, self.port, self.username, self.password, sock=sock)
+ return ssh
+
+ def download(self, location: str):
+ callback = make_progressbar() if self.progressbar else None
+ with self._establish() as ssh, ssh.open_sftp() as sftp:
+ if self.check_space:
+ check_storage(location, sftp.stat(self.path).st_size)
+ sftp.get(self.path, location, callback=callback)
+
+ def upload(self, location: str):
+ callback = make_progressbar() if self.progressbar else None
+ with self._establish() as ssh, ssh.open_sftp() as sftp:
+ try:
+ # If the remote path is a directory, use the original filename.
+ if stat.S_ISDIR(sftp.stat(self.path).st_mode):
+ path = os.path.join(self.path, os.path.basename(location))
+ # A file exists at this destination. We're simply going to clobber it.
+ else:
+ path = self.path
+ # This path doesn't point at any existing file. We can freely use this filename.
+ except IOError:
+ path = self.path
+ finally:
+ sftp.put(location, path, callback=callback)
+
+
+class HttpC:
+ def __init__(self, url, progressbar=False, check_space=False, source_host='', source_port=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')
+
+ def _establish(self):
+ session = Session()
+ session.mount(self.urlstring, SourceAdapter(self.source_pair))
+ session.headers.update({'User-Agent': 'VyOS/' + get_version()})
+ if self.username:
+ session.auth = self.username, self.password
+ return session
+
+ def download(self, location: str):
+ with self._establish() as s:
+ # We ask for uncompressed downloads so that we don't have to deal with decoding.
+ # 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:
+ # Abort early if the destination is inaccessible.
+ r.raise_for_status()
+ # If the request got redirected, keep the last URL we ended up with.
+ final_urlstring = r.url
+ if r.history and self.progressbar:
+ print_error('Redirecting to ' + final_urlstring)
+ # Check for the prospective file size.
try:
- # If the remote path is a directory, use the original filename.
- if stat.S_ISDIR(sftp.stat(remote_path).st_mode):
- path = os.path.join(remote_path, os.path.basename(local_path))
- # A file exists at this destination. We're simply going to clobber it.
- else:
- path = remote_path
- # This path doesn't point at any existing file. We can freely use this filename.
- except IOError:
- path = remote_path
- finally:
- sftp.put(local_path, path, callback=callback)
- elif mode == 'download':
- sftp.get(remote_path, local_path, callback=callback)
- elif mode == 'size':
- return sftp.stat(remote_path).st_size
-
-def upload_sftp(*args, **kwargs):
- transfer_sftp('upload', *args, **kwargs)
-
-def download_sftp(*args, **kwargs):
- transfer_sftp('download', *args, **kwargs)
-
-def get_sftp_file_size(*args, **kwargs):
- return transfer_sftp('size', None, *args, **kwargs)
-
-
-## TFTP routines
-def upload_tftp(local_path, hostname, remote_path, port=69, source=None, progressbar=False):
- source_option = f'--interface {source}' if source else ''
- progress_flag = '--progress-bar' if progressbar else '-s'
- with open(local_path, 'rb') as file:
- cmd(f'curl {source_option} {progress_flag} -T - tftp://{hostname}:{port}/{remote_path}',\
- stderr=None, input=file.read()).encode()
-
-def download_tftp(local_path, hostname, remote_path, port=69, source=None, progressbar=False):
- source_option = f'--interface {source}' if source else ''
- # Not really applicable but we pass it for the sake of uniformity.
- progress_flag = '--progress-bar' if progressbar else '-s'
- with open(local_path, 'wb') as file:
- file.write(cmd(f'curl {source_option} {progress_flag} tftp://{hostname}:{port}/{remote_path}',\
- stderr=None).encode())
-
-# get_tftp_file_size() is unimplemented because there is no way to obtain a file's size through TFTP,
-# as TFTP does not specify a SIZE command.
-
-
-## HTTP(S) routines
-def install_request_opener(urlstring, username, password):
- """
- Take `username` and `password` strings and install the appropriate
- password manager to `urllib.request.urlopen()` for the given `urlstring`.
- """
- manager = urlreq.HTTPPasswordMgrWithDefaultRealm()
- manager.add_password(None, urlstring, username, password)
- urlreq.install_opener(urlreq.build_opener(urlreq.HTTPBasicAuthHandler(manager)))
-
-# upload_http() is unimplemented.
-
-def download_http(local_path, urlstring, username=None, password=None, progressbar=False):
- """
- Download the file from from `urlstring` to `local_path`.
- Optionally takes `username` and `password` for authentication.
- """
- request = urlreq.Request(urlstring, headers={'User-Agent': 'VyOS/' + get_version()})
- if username:
- install_request_opener(urlstring, username, password)
- with open(local_path, 'wb') as file, urlreq.urlopen(request) as response:
- size = response.getheader('Content-Length')
- if progressbar and size:
- progress = make_incremental_progressbar(CHUNK_SIZE / int(size))
- next(progress)
- for chunk in iter(lambda: response.read(CHUNK_SIZE), b''):
- file.write(chunk)
- next(progress)
- next(progress)
- # If we can't determine the size or if a progress bar wasn't requested,
- # we can let `shutil` take care of the copying.
- else:
- shutil.copyfileobj(response, file)
-
-def get_http_file_size(urlstring, username=None, password=None):
- """
- Return the size of the file from `urlstring` in terms of number of bytes.
- Optionally takes `username` and `password` for authentication.
+ size = int(r.headers['Content-Length'])
+ # In case the server does not supply the header.
+ except KeyError:
+ size = None
+ if self.check_space:
+ check_storage(location, size)
+ with s.get(final_urlstring, stream=True) as r, open(location, 'wb') as f:
+ if self.progressbar and size:
+ progress = make_incremental_progressbar(CHUNK_SIZE / size)
+ next(progress)
+ for chunk in iter(lambda: begin(next(progress), r.raw.read(CHUNK_SIZE)), b''):
+ f.write(chunk)
+ else:
+ # We'll try to stream the download directly with `copyfileobj()` so that large
+ # files (like entire VyOS images) don't occupy much memory.
+ shutil.copyfileobj(r.raw, f)
+
+ 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)
+
+
+class TftpC:
+ # We simply allow `curl` to take over because
+ # 1. TFTP is rather simple.
+ # 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):
+ 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.urlstring = urllib.parse.urlunsplit(url)
+
+ def download(self, location: str):
+ with open(location, 'wb') as f:
+ f.write(cmd(f'{self.command} "{self.urlstring}"').encode())
+
+ def upload(self, location: str):
+ with open(location, 'rb') as f:
+ cmd(f'{self.command} -T - "{self.urlstring}"', input=f.read())
+
+
+def urlc(urlstring, *args, **kwargs):
"""
- request = urlreq.Request(urlstring, headers={'User-Agent': 'VyOS/' + get_version()})
- if username:
- install_request_opener(urlstring, username, password)
- with urlreq.urlopen(request) as response:
- size = response.getheader('Content-Length')
- if size:
- return int(size)
- # The server didn't send 'Content-Length' in the response headers.
- else:
- raise ValueError('Failed to receive file size from HTTP server.')
-
-
-## Dynamic dispatchers
-def download(local_path, urlstring, source=None, progressbar=False):
+ Dynamically dispatch the appropriate protocol class.
"""
- Dispatch the appropriate download function for the given `urlstring` and save to `local_path`.
- Optionally takes a `source` address or interface (not valid for HTTP(S)).
- Supports HTTP, HTTPS, FTP, SFTP, SCP (through SFTP) and TFTP.
- Reads `$REMOTE_USERNAME` and `$REMOTE_PASSWORD` environment variables.
- """
- url = urllib.parse.urlparse(urlstring)
- username, password = get_authentication_variables(url.username, url.password)
- port = get_port_from_url(url)
-
- if url.scheme == 'http' or url.scheme == 'https':
- if source:
- print_error('Warning: Custom source address not supported for HTTP connections.')
- download_http(local_path, urlstring, username, password, progressbar)
- elif url.scheme == 'ftp':
- source = get_source_address(source)[0] if source else None
- username = username if username else 'anonymous'
- download_ftp(local_path, url.hostname, url.path, username, password, port, source, progressbar)
- elif url.scheme == 'sftp' or url.scheme == 'scp':
- source = get_source_address(source) if source else None
- download_sftp(local_path, url.hostname, url.path, username, password, port, source, progressbar)
- elif url.scheme == 'tftp':
- download_tftp(local_path, url.hostname, url.path, port, source, progressbar)
- else:
- raise ValueError(f'Unsupported URL scheme: {url.scheme}')
+ url_classes = {'http': HttpC, 'https': HttpC, 'ftp': FtpC, 'ftps': FtpC, \
+ 'sftp': SshC, 'ssh': SshC, 'scp': SshC, 'tftp': TftpC}
+ url = urllib.parse.urlsplit(urlstring)
+ try:
+ return url_classes[url.scheme](url, *args, **kwargs)
+ except KeyError:
+ raise ValueError(f'Unsupported URL scheme: "{url.scheme}"')
-def upload(local_path, urlstring, source=None, progressbar=False):
- """
- Dispatch the appropriate upload function for the given URL and upload from local path.
- Optionally takes a `source` address.
- Supports FTP, SFTP, SCP (through SFTP) and TFTP.
- Reads `$REMOTE_USERNAME` and `$REMOTE_PASSWORD` environment variables.
- """
- url = urllib.parse.urlparse(urlstring)
- username, password = get_authentication_variables(url.username, url.password)
- port = get_port_from_url(url)
-
- if url.scheme == 'ftp':
- username = username if username else 'anonymous'
- source = get_source_address(source)[0] if source else None
- upload_ftp(local_path, url.hostname, url.path, username, password, port, source, progressbar)
- elif url.scheme == 'sftp' or url.scheme == 'scp':
- source = get_source_address(source) if source else None
- upload_sftp(local_path, url.hostname, url.path, username, password, port, source, progressbar)
- elif url.scheme == 'tftp':
- upload_tftp(local_path, url.hostname, url.path, port, source, progressbar)
- else:
- raise ValueError(f'Unsupported URL scheme: {url.scheme}')
+def download(local_path, urlstring, *args, **kwargs):
+ urlc(urlstring, *args, **kwargs).download(local_path)
-def get_remote_file_size(urlstring, source=None):
- """
- Dispatch the appropriate function to return the size of the remote file from `urlstring`
- in terms of number of bytes.
- Optionally takes a `source` address (not valid for HTTP(S)).
- Supports HTTP, HTTPS, FTP and SFTP (through SFTP).
- Reads `$REMOTE_USERNAME` and `$REMOTE_PASSWORD` environment variables.
- """
- url = urllib.parse.urlparse(urlstring)
- username, password = get_authentication_variables(url.username, url.password)
- port = get_port_from_url(url)
-
- if url.scheme == 'http' or url.scheme == 'https':
- if source:
- print_error('Warning: Custom source address not supported for HTTP connections.')
- return get_http_file_size(urlstring, username, password)
- elif url.scheme == 'ftp':
- source = get_source_address(source)[0] if source else None
- username = username if username else 'anonymous'
- return get_ftp_file_size(url.hostname, url.path, username, password, port, source)
- elif url.scheme == 'sftp' or url.scheme == 'scp':
- source = get_source_address(source) if source else None
- return get_sftp_file_size(url.hostname, url.path, username, password, port, source)
- else:
- raise ValueError(f'Unsupported URL scheme: {url.scheme}')
+def upload(local_path, urlstring, *args, **kwargs):
+ urlc(urlstring, *args, **kwargs).upload(local_path)
-def get_remote_config(urlstring, source=None):
+def get_remote_config(urlstring, source_host='', source_port=0):
"""
- Download remote (config) file from `urlstring` and return the contents as a string.
- Args:
- remote file URI:
- tftp://<host>[:<port>]/<file>
- http[s]://<host>[:<port>]/<file>
- [scp|sftp|ftp]://[<user>[:<passwd>]@]<host>[:port]/<file>
- source address (optional):
- <interface>
- <IP address>
+ Quietly download a file and return it as a string.
"""
temp = tempfile.NamedTemporaryFile(delete=False).name
try:
- download(temp, urlstring, source)
- with open(temp, 'r') as file:
- return file.read()
+ download(temp, urlstring, False, False, source_host, source_port)
+ with open(temp, 'r') as f:
+ return f.read()
finally:
os.remove(temp)
-def friendly_download(local_path, urlstring, source=None):
+def friendly_download(local_path, urlstring, source_host='', source_port=0):
"""
- Download from `urlstring` to `local_path` in an informative way.
- Checks the storage space before attempting download.
- Intended to be called from interactive, user-facing scripts.
+ Download with a progress bar, reassuring messages and free space checks.
"""
- destination_directory = os.path.dirname(local_path)
try:
- free_space = shutil.disk_usage(destination_directory).free
- try:
- file_size = get_remote_file_size(urlstring, source)
- if file_size < 1024 * 1024:
- print_error(f'The file is {file_size / 1024.0:.3f} KiB.')
- else:
- print_error(f'The file is {file_size / (1024.0 * 1024.0):.3f} MiB.')
- if file_size > free_space:
- raise OSError(f'Not enough disk space available in "{destination_directory}".')
- except ValueError:
- # Can't do a storage check in this case, so we bravely continue.
- file_size = 0
- print_error('Could not determine the file size in advance.')
- else:
- print_error('Downloading...')
- download(local_path, urlstring, source, progressbar=file_size > 1024 * 1024)
+ print_error('Downloading...')
+ download(local_path, urlstring, True, True, source_host, source_port)
except KeyboardInterrupt:
- print_error('Download aborted by user.')
+ print_error('\nDownload aborted by user.')
sys.exit(1)
except:
import traceback
+ print_error(f'Failed to download {urlstring}.')
# There are a myriad different reasons a download could fail.
# SSH errors, FTP errors, I/O errors, HTTP errors (403, 404...)
# We omit the scary stack trace but print the error nevertheless.
- print_error(f'Failed to download {urlstring}.')
- traceback.print_exception(*sys.exc_info()[:2], None)
+ exc_type, exc_value, exc_traceback = sys.exc_info()
+ traceback.print_exception(exc_type, exc_value, None, 0, None, False)
sys.exit(1)
else:
print_error('Download complete.')
+ sys.exit(0)
diff --git a/python/vyos/template.py b/python/vyos/template.py
index b32cafe74..633b28ade 100644
--- a/python/vyos/template.py
+++ b/python/vyos/template.py
@@ -22,6 +22,7 @@ from jinja2 import FileSystemLoader
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
# Holds template filters registered via register_filter()
@@ -151,6 +152,16 @@ def bracketize_ipv6(address):
return f'[{address}]'
return address
+@register_filter('dot_colon_to_dash')
+def dot_colon_to_dash(text):
+ """ Replace dot and colon to dash for string
+ Example:
+ 192.0.2.1 => 192-0-2-1, 2001:db8::1 => 2001-db8--1
+ """
+ text = text.replace(":", "-")
+ text = text.replace(".", "-")
+ return text
+
@register_filter('netmask_from_cidr')
def netmask_from_cidr(prefix):
""" Take CIDR prefix and convert the prefix length to a "subnet mask".
@@ -479,3 +490,70 @@ def get_openvpn_ncp_ciphers(ciphers):
else:
out.append(cipher)
return ':'.join(out).upper()
+
+@register_filter('snmp_auth_oid')
+def snmp_auth_oid(type):
+ if type not in ['md5', 'sha', 'aes', 'des', 'none']:
+ raise ValueError()
+
+ OIDs = {
+ 'md5' : '.1.3.6.1.6.3.10.1.1.2',
+ 'sha' : '.1.3.6.1.6.3.10.1.1.3',
+ 'aes' : '.1.3.6.1.6.3.10.1.2.4',
+ 'des' : '.1.3.6.1.6.3.10.1.2.2',
+ 'none': '.1.3.6.1.6.3.10.1.2.1'
+ }
+ return OIDs[type]
+
+@register_filter('nft_action')
+def nft_action(vyos_action):
+ if vyos_action == 'accept':
+ return 'return'
+ return vyos_action
+
+@register_filter('nft_rule')
+def nft_rule(rule_conf, fw_name, rule_id, ip_name='ip'):
+ from vyos.firewall import parse_rule
+ return parse_rule(rule_conf, fw_name, rule_id, ip_name)
+
+@register_filter('nft_default_rule')
+def nft_default_rule(fw_conf, fw_name):
+ output = ['counter']
+ default_action = fw_conf.get('default_action', 'accept')
+
+ if 'enable_default_log' in fw_conf:
+ action_suffix = default_action[:1].upper()
+ output.append(f'log prefix "[{fw_name[:19]}-default-{action_suffix}] "')
+
+ output.append(nft_action(default_action))
+ output.append(f'comment "{fw_name} default-action {default_action}"')
+ return " ".join(output)
+
+@register_filter('nft_state_policy')
+def nft_state_policy(conf, state, ipv6=False):
+ out = [f'ct state {state}']
+
+ if 'log' in conf and 'enable' in conf['log']:
+ out.append('log')
+
+ out.append('counter')
+
+ if 'action' in conf:
+ out.append(conf['action'])
+
+ return " ".join(out)
+
+@register_filter('nft_intra_zone_action')
+def nft_intra_zone_action(zone_conf, ipv6=False):
+ if 'intra_zone_filtering' in zone_conf:
+ intra_zone = zone_conf['intra_zone_filtering']
+ fw_name = 'ipv6_name' if ipv6 else 'name'
+
+ if 'action' in intra_zone:
+ if intra_zone['action'] == 'accept':
+ return 'return'
+ return intra_zone['action']
+ elif dict_search_args(intra_zone, 'firewall', fw_name):
+ name = dict_search_args(intra_zone, 'firewall', fw_name)
+ return f'jump {name}'
+ return 'return'
diff --git a/python/vyos/util.py b/python/vyos/util.py
index 9c4c29322..571d43754 100644
--- a/python/vyos/util.py
+++ b/python/vyos/util.py
@@ -794,6 +794,24 @@ def get_interface_address(interface):
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
@@ -856,6 +874,20 @@ def make_incremental_progressbar(increment: float):
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.
@@ -920,14 +952,48 @@ def install_into_config(conf, config_paths, override_prompt=True):
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
- cmd(f'/opt/vyatta/sbin/my_set {path}')
- count += 1
+ 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')
+
+ 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
generated by cgit v1.2.3 (git 2.39.1) at 2025-11-04 00:32:08 +0000