summaryrefslogtreecommitdiff
path: root/python/vyos
diff options
context:
space:
mode:
Diffstat (limited to 'python/vyos')
-rw-r--r--python/vyos/configdict.py95
-rw-r--r--python/vyos/configquery.py46
-rw-r--r--python/vyos/configsource.py3
-rw-r--r--python/vyos/configverify.py7
-rw-r--r--python/vyos/defaults.py2
-rw-r--r--python/vyos/ethtool.py3
-rw-r--r--python/vyos/frr.py44
-rw-r--r--python/vyos/ifconfig/ethernet.py17
-rwxr-xr-xpython/vyos/ifconfig/interface.py61
-rw-r--r--python/vyos/ifconfig/section.py10
-rw-r--r--python/vyos/ifconfig/wwan.py17
-rw-r--r--python/vyos/range_regex.py142
-rw-r--r--python/vyos/remote.py581
-rw-r--r--python/vyos/template.py1
-rw-r--r--python/vyos/util.py121
15 files changed, 753 insertions, 397 deletions
diff --git a/python/vyos/configdict.py b/python/vyos/configdict.py
index 5c6836e97..425a2e416 100644
--- a/python/vyos/configdict.py
+++ b/python/vyos/configdict.py
@@ -155,18 +155,15 @@ def get_removed_vlans(conf, dict):
D.set_level(conf.get_level())
# get_child_nodes() will return dict_keys(), mangle this into a list with PEP448
keys = D.get_child_nodes_diff(['vif'], expand_nodes=Diff.DELETE)['delete'].keys()
- if keys:
- dict.update({'vif_remove': [*keys]})
+ if keys: dict['vif_remove'] = [*keys]
# get_child_nodes() will return dict_keys(), mangle this into a list with PEP448
keys = D.get_child_nodes_diff(['vif-s'], expand_nodes=Diff.DELETE)['delete'].keys()
- if keys:
- dict.update({'vif_s_remove': [*keys]})
+ if keys: dict['vif_s_remove'] = [*keys]
for vif in dict.get('vif_s', {}).keys():
keys = D.get_child_nodes_diff(['vif-s', vif, 'vif-c'], expand_nodes=Diff.DELETE)['delete'].keys()
- if keys:
- dict.update({'vif_s': { vif : {'vif_c_remove': [*keys]}}})
+ if keys: dict['vif_s'][vif]['vif_c_remove'] = [*keys]
return dict
@@ -319,6 +316,40 @@ def is_source_interface(conf, interface, intftype=None):
old_level = conf.set_level(old_level)
return ret_val
+def get_dhcp_interfaces(conf, vrf=None):
+ """ Common helper functions to retrieve all interfaces from current CLI
+ sessions that have DHCP configured. """
+ dhcp_interfaces = []
+ dict = conf.get_config_dict(['interfaces'], get_first_key=True)
+ if not dict:
+ return dhcp_interfaces
+
+ def check_dhcp(config, ifname):
+ out = []
+ if 'address' in config and 'dhcp' in config['address']:
+ if 'vrf' in config:
+ if vrf is config['vrf']: out.append(ifname)
+ else: out.append(ifname)
+ return out
+
+ for section, interface in dict.items():
+ for ifname, ifconfig in interface.items():
+ tmp = check_dhcp(ifconfig, ifname)
+ dhcp_interfaces.extend(tmp)
+ # check per VLAN interfaces
+ for vif, vif_config in ifconfig.get('vif', {}).items():
+ tmp = check_dhcp(vif_config, f'{ifname}.{vif}')
+ dhcp_interfaces.extend(tmp)
+ # check QinQ VLAN interfaces
+ for vif_s, vif_s_config in ifconfig.get('vif-s', {}).items():
+ tmp = check_dhcp(vif_s_config, f'{ifname}.{vif_s}')
+ dhcp_interfaces.extend(tmp)
+ for vif_c, vif_c_config in vif_s_config.get('vif-c', {}).items():
+ tmp = check_dhcp(vif_c_config, f'{ifname}.{vif_s}.{vif_c}')
+ dhcp_interfaces.extend(tmp)
+
+ return dhcp_interfaces
+
def get_interface_dict(config, base, ifname=''):
"""
Common utility function to retrieve and mangle the interfaces configuration
@@ -368,9 +399,11 @@ def get_interface_dict(config, base, ifname=''):
del default_values['dhcpv6_options']
# We have gathered the dict representation of the CLI, but there are
- # default options which we need to update into the dictionary
- # retrived.
- dict = dict_merge(default_values, dict)
+ # default options which we need to update into the dictionary retrived.
+ # But we should only add them when interface is not deleted - as this might
+ # confuse parsers
+ if 'deleted' not in dict:
+ dict = dict_merge(default_values, dict)
# XXX: T2665: blend in proper DHCPv6-PD default values
dict = T2665_set_dhcpv6pd_defaults(dict)
@@ -423,9 +456,12 @@ def get_interface_dict(config, base, ifname=''):
if not 'dhcpv6_options' in vif_config:
del default_vif_values['dhcpv6_options']
- dict['vif'][vif] = dict_merge(default_vif_values, vif_config)
- # XXX: T2665: blend in proper DHCPv6-PD default values
- dict['vif'][vif] = T2665_set_dhcpv6pd_defaults(dict['vif'][vif])
+ # 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)
+ # XXX: T2665: blend in proper DHCPv6-PD default values
+ dict['vif'][vif] = T2665_set_dhcpv6pd_defaults(dict['vif'][vif])
# Check if we are a member of a bridge device
bridge = is_member(config, f'{ifname}.{vif}', 'bridge')
@@ -441,10 +477,12 @@ def get_interface_dict(config, base, ifname=''):
if not 'dhcpv6_options' in vif_s_config:
del default_vif_s_values['dhcpv6_options']
- dict['vif_s'][vif_s] = dict_merge(default_vif_s_values, vif_s_config)
- # XXX: T2665: blend in proper DHCPv6-PD default values
- dict['vif_s'][vif_s] = T2665_set_dhcpv6pd_defaults(
- dict['vif_s'][vif_s])
+ # 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)
+ # XXX: T2665: blend in proper DHCPv6-PD default values
+ dict['vif_s'][vif_s] = T2665_set_dhcpv6pd_defaults(dict['vif_s'][vif_s])
# Check if we are a member of a bridge device
bridge = is_member(config, f'{ifname}.{vif_s}', 'bridge')
@@ -458,11 +496,14 @@ def get_interface_dict(config, base, ifname=''):
if not 'dhcpv6_options' in vif_c_config:
del default_vif_c_values['dhcpv6_options']
- dict['vif_s'][vif_s]['vif_c'][vif_c] = dict_merge(
+ # 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]['vif_c'][vif_c] = dict_merge(
default_vif_c_values, vif_c_config)
- # 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])
+ # 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])
# Check if we are a member of a bridge device
bridge = is_member(config, f'{ifname}.{vif_s}.{vif_c}', 'bridge')
@@ -522,6 +563,11 @@ def get_accel_dict(config, base, chap_secrets):
if dict_search('authentication.local_users.username', default_values):
del default_values['authentication']['local_users']['username']
+ # T2665: defaults include IPv6 client-pool mask per TAG node which need to be
+ # added to individual local users instead - so we can simply delete them
+ if dict_search('client_ipv6_pool.prefix.mask', default_values):
+ del default_values['client_ipv6_pool']['prefix']['mask']
+
dict = dict_merge(default_values, dict)
# set CPUs cores to process requests
@@ -565,4 +611,13 @@ def get_accel_dict(config, base, chap_secrets):
dict['authentication']['local_users']['username'][username] = dict_merge(
default_values, dict['authentication']['local_users']['username'][username])
+ # Add individual IPv6 client-pool default mask if required
+ if dict_search('client_ipv6_pool.prefix', dict):
+ # T2665
+ default_values = defaults(base + ['client-ipv6-pool', 'prefix'])
+
+ for prefix in dict_search('client_ipv6_pool.prefix', dict):
+ dict['client_ipv6_pool']['prefix'][prefix] = dict_merge(
+ default_values, dict['client_ipv6_pool']['prefix'][prefix])
+
return dict
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..a0f6a46b5 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:
diff --git a/python/vyos/configverify.py b/python/vyos/configverify.py
index 8aca76568..365a28feb 100644
--- a/python/vyos/configverify.py
+++ b/python/vyos/configverify.py
@@ -110,15 +110,12 @@ def verify_tunnel(config):
raise ConfigError('Must configure the tunnel encapsulation for '\
'{ifname}!'.format(**config))
- if 'source_address' not in config and 'dhcp_interface' not in config:
- raise ConfigError('source-address is mandatory for tunnel')
+ if 'source_address' not in config and 'source_interface' not in config:
+ raise ConfigError('source-address or source-interface required for tunnel!')
if 'remote' not in config and config['encapsulation'] != 'gre':
raise ConfigError('remote ip address is mandatory for tunnel')
- if {'source_address', 'dhcp_interface'} <= set(config):
- raise ConfigError('Can not use both source-address and dhcp-interface')
-
if config['encapsulation'] in ['ipip6', 'ip6ip6', 'ip6gre', 'ip6gretap', 'ip6erspan']:
error_ipv6 = 'Encapsulation mode requires IPv6'
if 'source_address' in config and not is_ipv6(config['source_address']):
diff --git a/python/vyos/defaults.py b/python/vyos/defaults.py
index 00b14a985..f355c4919 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'
diff --git a/python/vyos/ethtool.py b/python/vyos/ethtool.py
index eb5b0a456..e45b0f041 100644
--- a/python/vyos/ethtool.py
+++ b/python/vyos/ethtool.py
@@ -56,9 +56,6 @@ class Ethtool:
link = os.readlink(sysfs_file)
self._driver_name = os.path.basename(link)
- if not self._driver_name:
- raise ValueError(f'Could not determine driver for interface {ifname}!')
-
# Build a dictinary of supported link-speed and dupley settings.
out, err = popen(f'ethtool {ifname}')
reading = False
diff --git a/python/vyos/frr.py b/python/vyos/frr.py
index df6849472..a8f115d9a 100644
--- a/python/vyos/frr.py
+++ b/python/vyos/frr.py
@@ -84,12 +84,14 @@ if DEBUG:
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
@@ -214,13 +216,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):
@@ -448,16 +445,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/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 e6dbd861b..bcb692697 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}',
@@ -461,15 +465,19 @@ class Interface(Control):
# Get processor ID number
cpu_id = self._cmd('sudo dmidecode -t 4 | grep ID | head -n1 | sed "s/.*ID://;s/ //g"')
- # Get system eth0 base MAC address - every system has eth0
- eth0_mac = Interface('eth0').get_mac()
+
+ # XXX: T3894 - it seems not all systems have eth0 - get a list of all
+ # available Ethernet interfaces on the system (without VLAN subinterfaces)
+ # and then take the first one.
+ all_eth_ifs = [x for x in Section.interfaces('ethernet') if '.' not in x]
+ first_mac = Interface(all_eth_ifs[0]).get_mac()
sha = sha256()
# Calculate SHA256 sum based on the CPU ID number, eth0 mac address and
# this interface identifier - this is as predictable as an interface
# MAC address and thus can be used in the same way
sha.update(cpu_id.encode())
- sha.update(eth0_mac.encode())
+ sha.update(first_mac.encode())
sha.update(self.ifname.encode())
# take the most significant 48 bits from the SHA256 string
tmp = sha.hexdigest()[:12]
@@ -508,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.
@@ -1052,14 +1089,6 @@ class Interface(Control):
addr_is_v4 = is_ipv4(addr)
- # we can't have both DHCP and static IPv4 addresses assigned
- for a in self._addr:
- if ( ( addr == 'dhcp' and a != 'dhcpv6' and is_ipv4(a) ) or
- ( a == 'dhcp' and addr != 'dhcpv6' and addr_is_v4 ) ):
- raise ConfigError((
- "Can't configure both static IPv4 and DHCP address "
- "on the same interface"))
-
# add to interface
if addr == 'dhcp':
self.set_dhcp(True)
@@ -1357,6 +1386,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/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..732ef76b7 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,339 +59,279 @@ 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 get_source_address(source):
+ 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 init_poolmanager(self, connections, maxsize, block=False):
+ self.poolmanager = PoolManager(
+ num_pools=connections, maxsize=maxsize,
+ block=block, source_address=self._source_pair)
+
+class WrappedFile:
+ def __init__(self, obj, size=None, chunk_size=CHUNK_SIZE):
+ self._obj = obj
+ self._progress = size and make_incremental_progressbar(chunk_size / size)
+ def read(self, size=-1):
+ if self._progress:
+ next(self._progress)
+ self._obj.read(size)
+ def write(self, size=-1):
+ if self._progress:
+ next(self._progress)
+ self._obj.write(size)
+ def __getattr__(self, attr):
+ return getattr(self._obj, attr)
+
+
+def check_storage(path, size):
"""
- 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)`.
+ Check whether `path` has enough storage space for a transfer of `size` bytes.
"""
- # 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
+ 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.')
+ print_error('Bravely continuing regardless.')
+ return
+
+ if size < 1024 * 1024:
+ print_error(f'The file is {size / 1024.0:.3f} KiB.')
else:
- return (socket.gethostbyname(source), port), socket.AF_INET
-
-def get_port_from_url(url):
- """
- Return the port number from the given `url` named tuple, fall back to
- the default if there isn't one.
- """
- defaults = {"http": 80, "https": 443, "ftp": 21, "tftp": 69,\
- "ssh": 22, "scp": 22, "sftp": 22}
- if url.port:
- return url.port
- 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.
+ if r.history:
+ final_urlstring = r.history[-1].url
+ print_error('Redirecting to ' + final_urlstring)
+ else:
+ final_urlstring = self.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.
- """
- 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):
+ 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):
+ size = os.path.getsize(location) if self.progressbar else None
+ # Keep in mind that `data` can be a file-like or iterable object.
+ with self._establish() as s, file(location, 'rb') as f:
+ s.post(self.urlstring, data=WrappedFile(f, size), 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):
"""
- 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.
+ Dynamically dispatch the appropriate protocol class.
"""
- 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
@@ -401,3 +343,4 @@ def friendly_download(local_path, urlstring, source=None):
sys.exit(1)
else:
print_error('Download complete.')
+ sys.exit(0)
diff --git a/python/vyos/template.py b/python/vyos/template.py
index d13915766..b32cafe74 100644
--- a/python/vyos/template.py
+++ b/python/vyos/template.py
@@ -349,7 +349,6 @@ def get_dhcp_router(interface):
Returns False of no router is found, returns the IP address as string if
a router is found.
"""
- interface = interface.replace('.', '_')
lease_file = f'/var/lib/dhcp/dhclient_{interface}.leases'
if not os.path.exists(lease_file):
return None
diff --git a/python/vyos/util.py b/python/vyos/util.py
index 849b27d3b..954c6670d 100644
--- a/python/vyos/util.py
+++ b/python/vyos/util.py
@@ -489,6 +489,40 @@ def seconds_to_human(s, separator=""):
return result
+def bytes_to_human(bytes, initial_exponent=0):
+ """ Converts a value in bytes to a human-readable size string like 640 KB
+
+ The initial_exponent parameter is the exponent of 2,
+ e.g. 10 (1024) for kilobytes, 20 (1024 * 1024) for megabytes.
+ """
+
+ from math import log2
+
+ bytes = bytes * (2**initial_exponent)
+
+ # log2 is a float, while range checking requires an int
+ exponent = int(log2(bytes))
+
+ if exponent < 10:
+ value = bytes
+ suffix = "B"
+ elif exponent in range(10, 20):
+ value = bytes / 1024
+ suffix = "KB"
+ elif exponent in range(20, 30):
+ value = bytes / 1024**2
+ suffix = "MB"
+ elif exponent in range(30, 40):
+ value = bytes / 1024**3
+ suffix = "GB"
+ else:
+ value = bytes / 1024**4
+ suffix = "TB"
+ # Add a new case when the first machine with petabyte RAM
+ # hits the market.
+
+ size_string = "{0:.2f} {1}".format(value, suffix)
+ return size_string
def get_cfg_group_id():
from grp import getgrnam
@@ -630,7 +664,6 @@ def ask_yes_no(question, default=False) -> bool:
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
@@ -761,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
@@ -823,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.
@@ -869,3 +934,57 @@ def check_port_availability(ipaddress, port, protocol):
return True
except:
return False
+
+def install_into_config(conf, config_paths, override_prompt=True):
+ # Allows op-mode scripts to install values if called from an active config session
+ # config_paths: dict of config paths
+ # override_prompt: if True, user will be prompted before existing nodes are overwritten
+
+ if not config_paths:
+ return None
+
+ from vyos.config import Config
+
+ if not Config().in_session():
+ print('You are not in configure mode, commands to install manually from configure mode:')
+ for path in config_paths:
+ print(f'set {path}')
+ return None
+
+ count = 0
+
+ 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
+
+ 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 2024-11-25 17:56:15 +0000