summaryrefslogtreecommitdiff
path: root/python
diff options
context:
space:
mode:
Diffstat (limited to 'python')
-rw-r--r--python/vyos/defaults.py4
-rw-r--r--python/vyos/ifconfig/bridge.py5
-rw-r--r--python/vyos/ifconfig/interface.py17
-rw-r--r--python/vyos/kea.py63
-rw-r--r--python/vyos/utils/activate.py126
-rw-r--r--python/vyos/utils/func.py28
-rw-r--r--python/vyos/utils/system.py34
-rw-r--r--python/vyos/vpp/control_vpp.py60
8 files changed, 258 insertions, 79 deletions
diff --git a/python/vyos/defaults.py b/python/vyos/defaults.py
index 6aa93ebed..78721c0d2 100644
--- a/python/vyos/defaults.py
+++ b/python/vyos/defaults.py
@@ -102,3 +102,7 @@ SSH_DSA_DEPRECATION_WARNING: str = \
'ED25519) to avoid authentication failures after the upgrade.'
reference_tree_cache = '/usr/share/vyos/reftree.cache'
+
+activation_list = os.path.join(directories['config'], 'activation-list')
+activation_init = os.path.join(directories['data'], 'activation-init')
+activation_hint = os.path.join(directories['data'], '.activation_hint')
diff --git a/python/vyos/ifconfig/bridge.py b/python/vyos/ifconfig/bridge.py
index 6b5f49778..5f184f221 100644
--- a/python/vyos/ifconfig/bridge.py
+++ b/python/vyos/ifconfig/bridge.py
@@ -386,6 +386,11 @@ class BridgeIf(Interface):
value = '1' if (tmp != None) else '0'
lower.set_root_guard(value)
+ # set learning
+ disable_learning = dict_search('disable_learning', interface_config)
+ value = '1' if disable_learning is None else '0'
+ lower.set_learning(value)
+
if 'enable_vlan' in config:
add_vlan = []
native_vlan_id = None
diff --git a/python/vyos/ifconfig/interface.py b/python/vyos/ifconfig/interface.py
index febd68f44..136adb289 100644
--- a/python/vyos/ifconfig/interface.py
+++ b/python/vyos/ifconfig/interface.py
@@ -226,6 +226,11 @@ class Interface(Control):
'location': '/sys/class/net/{ifname}/brport/root_block',
'errormsg': '{ifname} is not a bridge port member'
},
+ 'learning': {
+ 'validate': assert_boolean,
+ 'location': '/sys/class/net/{ifname}/brport/learning',
+ 'errormsg': '{ifname} is not a bridge port member'
+ },
'proxy_arp': {
'validate': assert_boolean,
'location': '/proc/sys/net/ipv4/conf/{ifname}/proxy_arp',
@@ -1159,6 +1164,18 @@ class Interface(Control):
"""
self.set_interface('bridge_port_isolation', on_or_off)
+ def set_learning(self, state):
+ """
+ Set MAC address learning state on a bridge port. When disabled,
+ the bridge will not learn source MAC addresses from incoming frames on
+ this port, causing all unknown unicast traffic to be flooded.
+
+ Example:
+ >>> from vyos.ifconfig import Interface
+ >>> Interface('eth0').set_learning(0)
+ """
+ self.set_interface('learning', state)
+
def set_proxy_arp(self, enable):
"""
Set per interface proxy ARP configuration
diff --git a/python/vyos/kea.py b/python/vyos/kea.py
index 703dd8b49..8dcd0d6eb 100644
--- a/python/vyos/kea.py
+++ b/python/vyos/kea.py
@@ -88,6 +88,7 @@ def _find_list_of_dict_index(lst, key='ip', value=''):
idx = next((index for (index, d) in enumerate(lst) if d[key] == value), None)
return idx
+
def kea_test_config(process: str, config_path: str) -> tuple[bool, str]:
result, output = rc_cmd(f'{process} -t {config_path}')
@@ -97,6 +98,7 @@ def kea_test_config(process: str, config_path: str) -> tuple[bool, str]:
find = re.search(r'Error encountered:\s([^\n$]+)', output)
return (False, find[1] if find else None)
+
def kea_parse_options(config):
options = []
@@ -159,7 +161,11 @@ def kea_parse_options(config):
def kea_parse_subnet(subnet, config):
- out = {'subnet': subnet, 'id': int(config['subnet_id']), 'user-context': {'enable-ping-check': False}}
+ out = {
+ 'subnet': subnet,
+ 'id': int(config['subnet_id']),
+ 'user-context': {'enable-ping-check': False},
+ }
if 'option' in config:
out['option-data'] = kea_parse_options(config['option'])
@@ -371,6 +377,7 @@ def kea6_parse_subnet(subnet, config):
return out
+
def kea_parse_tsig_algo(algo_spec):
translate = {
'md5': 'HMAC-MD5',
@@ -378,15 +385,17 @@ def kea_parse_tsig_algo(algo_spec):
'sha224': 'HMAC-SHA224',
'sha256': 'HMAC-SHA256',
'sha384': 'HMAC-SHA384',
- 'sha512': 'HMAC-SHA512'
+ 'sha512': 'HMAC-SHA512',
}
if algo_spec not in translate:
raise ConfigError(f'Unsupported TSIG algorithm: {algo_spec}')
return translate[algo_spec]
+
def kea_parse_enable_disable(value):
return True if value == 'enable' else False
+
def kea_parse_ddns_settings(config):
data = {}
@@ -394,7 +403,9 @@ def kea_parse_ddns_settings(config):
data['ddns-send-updates'] = kea_parse_enable_disable(send_updates)
if override_client_update := config.get('override_client_update'):
- data['ddns-override-client-update'] = kea_parse_enable_disable(override_client_update)
+ data['ddns-override-client-update'] = kea_parse_enable_disable(
+ override_client_update
+ )
if override_no_update := config.get('override_no_update'):
data['ddns-override-no-update'] = kea_parse_enable_disable(override_no_update)
@@ -403,7 +414,9 @@ def kea_parse_ddns_settings(config):
data['ddns-update-on-renew'] = kea_parse_enable_disable(update_on_renew)
if conflict_resolution := config.get('conflict_resolution'):
- data['ddns-use-conflict-resolution'] = kea_parse_enable_disable(conflict_resolution)
+ data['ddns-use-conflict-resolution'] = kea_parse_enable_disable(
+ conflict_resolution
+ )
if 'replace_client_name' in config:
data['ddns-replace-client-name'] = config['replace_client_name']
@@ -420,6 +433,7 @@ def kea_parse_ddns_settings(config):
return data
+
def _ctrl_socket_command(inet, vrf_name, command, args=None):
if vrf_name:
vrf_append = f'-{vrf_name}'
@@ -603,7 +617,9 @@ def kea_get_static_mappings(config, inet, pools=[]) -> list:
return mappings
-def kea_get_server_leases(config, inet, vrf_name, pools=[], state=[], origin=None) -> list:
+def kea_get_server_leases(
+ config, inet, vrf_name, pools=[], state=[], origin=None
+) -> list:
"""
Get DHCP server leases from active Kea DHCPv4 or DHCPv6 configuration
:return list
@@ -659,9 +675,10 @@ def kea_get_server_leases(config, inet, vrf_name, pools=[], state=[], origin=Non
now = datetime.now(timezone.utc)
if lease['valid-lft'] > 0 and lease['expire_time'] > now:
- # subtraction gives us a timedelta object which can't be formatted
- # with strftime so we use str(), split gets rid of the microseconds
- data_lease['remaining'] = str(lease['expire_time'] - now).split('.')[0]
+ # clear the microseconds before subtraction for visual clarity
+ data_lease['remaining'] = lease['expire_time'].replace(
+ microsecond=0
+ ) - now.replace(microsecond=0)
# Do not add old leases
if (
@@ -685,21 +702,31 @@ def kea_get_server_leases(config, inet, vrf_name, pools=[], state=[], origin=Non
return data
+
def _build_relay_hex_condition(sub_option_index, value):
- if value.startswith("0x"):
- return f"relay4[{sub_option_index}].hex == {value}"
+ if value.startswith('0x'):
+ return f'relay4[{sub_option_index}].hex == {value}'
else:
- return f"relay4[{sub_option_index}].hex == 0x{value.encode().hex().lower()}"
+ return f'relay4[{sub_option_index}].hex == 0x{value.encode().hex().lower()}'
+
def kea_build_client_class_test(config):
conditions = []
- if "relay_agent_information" in config:
- if "circuit_id" in config["relay_agent_information"]:
- conditions.append(_build_relay_hex_condition(1, config["relay_agent_information"]["circuit_id"]))
- if "remote_id" in config["relay_agent_information"]:
- conditions.append(_build_relay_hex_condition(2, config["relay_agent_information"]["remote_id"]))
-
- test = " and ".join(conditions)
+ if 'relay_agent_information' in config:
+ if 'circuit_id' in config['relay_agent_information']:
+ conditions.append(
+ _build_relay_hex_condition(
+ 1, config['relay_agent_information']['circuit_id']
+ )
+ )
+ if 'remote_id' in config['relay_agent_information']:
+ conditions.append(
+ _build_relay_hex_condition(
+ 2, config['relay_agent_information']['remote_id']
+ )
+ )
+
+ test = ' and '.join(conditions)
return test
diff --git a/python/vyos/utils/activate.py b/python/vyos/utils/activate.py
new file mode 100644
index 000000000..51b92a4aa
--- /dev/null
+++ b/python/vyos/utils/activate.py
@@ -0,0 +1,126 @@
+# Copyright (C) VyOS Inc.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library. If not, see <http://www.gnu.org/licenses/>.
+
+
+import json
+import typing
+from pathlib import Path
+
+from vyos.base import Warning as Warn
+from vyos.defaults import activation_list
+from vyos.defaults import activation_init
+from vyos.defaults import activation_hint
+from vyos.defaults import directories
+
+
+ActiveOpt = typing.Literal['enabled', 'once', 'off', 'never']
+
+
+def get_activation_scripts() -> dict:
+ list_path = Path(activation_list)
+ return json.loads(list_path.read_text())
+
+
+def set_activation(file_name: str, value: ActiveOpt):
+ script_dict = get_activation_scripts()
+ file_key = Path(file_name).stem
+ script_dict[file_key] = value
+ list_path = Path(activation_list)
+ list_path.write_text(json.dumps(script_dict))
+
+
+def get_activation(file_name: str) -> ActiveOpt:
+ script_dict = get_activation_scripts()
+ file_key = Path(file_name).stem
+ return script_dict[file_key]
+
+
+def is_active(file_name: str) -> bool:
+ script_dict = get_activation_scripts()
+ file_key = Path(file_name).stem
+ if script_dict[file_key] in ('enabled', 'once'):
+ return True
+ return False
+
+
+def stable_update(new: dict, old: dict):
+ res = {}
+ for key in new.keys():
+ res[key] = old[key] if key in old.keys() else new[key]
+ return res
+
+
+def init_activation_list() -> bool:
+ """Init if activation_hint exists, left on install_image or if image was
+ built as raw_image"""
+
+ init_hint = Path(activation_hint)
+ if not init_hint.exists():
+ return False
+
+ init_hint.unlink()
+ init_list = Path(activation_init)
+ data_list = Path(activation_list)
+ data_obj = json.loads(init_list.read_text())
+ data_list.write_text(json.dumps(data_obj))
+
+ return True
+
+
+def refresh_activation_list():
+ """Refresh activation list, as will be needed after image update"""
+
+ if init_activation_list():
+ return
+
+ new_list_path = Path(directories['data']).joinpath(Path(activation_list).name)
+ if not new_list_path.exists():
+ return
+
+ new_obj = json.loads(new_list_path.read_text())
+
+ orig_list_path = Path(activation_list)
+ if orig_list_path.exists():
+ orig_obj = json.loads(orig_list_path.read_text())
+ if orig_obj == new_obj:
+ return
+
+ obj = stable_update(new_obj, orig_obj)
+ else:
+ obj = new_obj
+
+ orig_list_path.write_text(json.dumps(obj))
+
+
+first_installed_boot_file = '/run/first_installed_boot'
+
+
+def set_first_installed_boot():
+ try:
+ Path(first_installed_boot_file).touch(exist_ok=False)
+ except FileExistsError:
+ Warn('redundant set of first_installed_boot')
+
+
+def is_first_installed_boot():
+ return Path(first_installed_boot_file).exists()
+
+
+def set_config_path_hint():
+ """The config hint allows subsequent installs to find previous disk
+ resident config data. It is traditionally added as part of the image
+ install procedure, however, for raw image builds an alternative is
+ needed."""
+ Path(directories['config']).joinpath('.vyatta_config').touch(exist_ok=True)
diff --git a/python/vyos/utils/func.py b/python/vyos/utils/func.py
new file mode 100644
index 000000000..830b76c94
--- /dev/null
+++ b/python/vyos/utils/func.py
@@ -0,0 +1,28 @@
+# Copyright (C) VyOS Inc.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library. If not, see <http://www.gnu.org/licenses/>.
+
+
+"""Module for functors and higher order functions."""
+
+
+class FalseCallable:
+ """Define falsy callable for use as default value for getattr of a
+ function from a module defined by vyos.utils.system.load_as_module"""
+
+ def __call__(self, *args, **kwargs):
+ pass
+
+ def __bool__(self):
+ return False
diff --git a/python/vyos/utils/system.py b/python/vyos/utils/system.py
index 578537544..fd5f49645 100644
--- a/python/vyos/utils/system.py
+++ b/python/vyos/utils/system.py
@@ -16,28 +16,34 @@
import os
from subprocess import run
-def sysctl_read(name: str) -> str:
+def _sysctl_key(parts: list[str]) -> str:
+ normalized = (p.replace('.', '/') for p in parts)
+ return '.'.join(normalized)
+
+def sysctl_read(name: list[str]) -> str:
"""Read and return current value of sysctl() option
Args:
- name (str): sysctl key name
+ name (list[str]): sysctl key name components
Returns:
str: sysctl key value
"""
- tmp = run(['sysctl', '-nb', name], capture_output=True)
- return tmp.stdout.decode()
+ key = _sysctl_key(name)
+ tmp = run(['sysctl', '-nb', key], capture_output=True)
+ return tmp.stdout.decode().strip()
-def sysctl_write(name: str, value: str | int) -> bool:
+def sysctl_write(name: list[str], value: str | int) -> bool:
"""Change value via sysctl()
Args:
- name (str): sysctl key name
+ name (list[str]): sysctl key name components
value (str | int): sysctl key value
Returns:
bool: True if changed, False otherwise
"""
+ key = _sysctl_key(name)
# convert other types to string before comparison
if not isinstance(value, str):
value = str(value)
@@ -45,7 +51,7 @@ def sysctl_write(name: str, value: str | int) -> bool:
if sysctl_read(name) == value:
return True
# return False if sysctl call failed
- if run(['sysctl', '-wq', f'{name}={value}']).returncode != 0:
+ if run(['sysctl', '-wq', f'{key}={value}']).returncode != 0:
return False
# compare old and new values
# sysctl may apply value, but its actual value will be
@@ -55,11 +61,11 @@ def sysctl_write(name: str, value: str | int) -> bool:
# False in other cases
return False
-def sysctl_apply(sysctl_dict: dict[str, str], revert: bool = True) -> bool:
+def sysctl_apply(sysctl_dict: dict[tuple[str, ...], str], revert: bool = True) -> bool:
"""Apply sysctl values.
Args:
- sysctl_dict (dict[str, str]): dictionary with sysctl keys with values
+ sysctl_dict (dict[tuple[str, ...], str]): dictionary with sysctl key components (tuple) with values
revert (bool, optional): Revert to original values if new were not
applied. Defaults to True.
@@ -67,12 +73,12 @@ def sysctl_apply(sysctl_dict: dict[str, str], revert: bool = True) -> bool:
bool: True if all params configured properly, False in other cases
"""
# get current values
- sysctl_original: dict[str, str] = {}
- for key_name in sysctl_dict.keys():
- sysctl_original[key_name] = sysctl_read(key_name)
+ sysctl_original: dict[tuple[str, ...], str] = {}
+ for key_parts in sysctl_dict.keys():
+ sysctl_original[key_parts] = sysctl_read(list(key_parts))
# apply new values and revert in case one of them was not applied
- for key_name, value in sysctl_dict.items():
- if not sysctl_write(key_name, value):
+ for key_parts, value in sysctl_dict.items():
+ if not sysctl_write(list(key_parts), value):
if revert:
sysctl_apply(sysctl_original, revert=False)
return False
diff --git a/python/vyos/vpp/control_vpp.py b/python/vyos/vpp/control_vpp.py
index 5f3a18848..1ad098b71 100644
--- a/python/vyos/vpp/control_vpp.py
+++ b/python/vyos/vpp/control_vpp.py
@@ -472,73 +472,39 @@ class VPPControl:
"""
vpp_pair = self.lcp_pair_find(kernel_name=ifname)
if vpp_pair:
- vpp_iface_name = vpp_pair['vpp_name_hw']
- vpp_pair_name = vpp_pair['vpp_name_kernel']
+ vpp_iface_index = vpp_pair['vpp_index_hw']
+ vpp_pair_index = vpp_pair['vpp_index_kernel']
self.__vpp_api_client.api.pppoe_add_del_cp(
- dp_sw_if_index=self.get_sw_if_index(vpp_iface_name),
- cp_sw_if_index=self.get_sw_if_index(vpp_pair_name),
+ dp_sw_if_index=vpp_iface_index,
+ cp_sw_if_index=vpp_pair_index,
is_add=True,
)
@_Decorators.api_call
def get_pppoe_interface_mapping(self) -> dict:
"""
- Parse the output of 'show pppoe control-plane binding' to build a mapping
- between data-plane and control-plane interfaces.
+ Build a mapping between data-plane and control-plane interfaces.
Returns:
- dict: Mapping of data-plane interface names/indices to control-plane interface names/indices.
+ dict: Mapping of data-plane interface indices to control-plane interface indices.
"""
- reply = self.cli_cmd('show pppoe control-plane binding').reply
result = {}
-
- deleted_re = re.compile(r'DELETED\s*\((\d+)\)')
-
- lines = reply.strip().splitlines()
-
- # If there are less than 2 lines, assume no PPPoE configured
- if len(lines) < 2:
- return result
-
- # Skip header lines (first 2 lines)
- for line in lines[2:]:
- line = line.strip()
- if not line:
- continue
-
- # Split into two columns by 2+ spaces
- cols = re.split(r'\s{2,}', line, maxsplit=1)
- if len(cols) != 2:
- continue
-
- # Determine key/value depending on which column has "DELETED (index)"
- col_a, col_b = cols
-
- match_a = deleted_re.search(col_a)
- match_b = deleted_re.search(col_b)
-
- key = int(match_a.group(1)) if match_a else col_a
- value = int(match_b.group(1)) if match_b else col_b
-
- result[key] = value
+ for binding in self.__vpp_api_client.api.pppoe_cp_binding_dump():
+ dp_index = binding.dp_sw_if_index
+ cp_index = binding.cp_sw_if_index
+ result[dp_index] = cp_index
return result
@_Decorators.api_call
- def delete_pppoe_mapping(self, dp_iface: str | int, cp_iface: str | int) -> None:
+ def delete_pppoe_mapping(self, dp_index: int, cp_index: int) -> None:
"""
Delete PPPoE mapping between data-plane and control-plane interfaces.
Args:
- dp_iface (str|int): Name or index of an interface in data-plane.
- cp_iface (str|int): Name or index of an interface in control plane.
+ dp_index (int): Index of an interface in data-plane.
+ cp_index (int): Index of an interface in control plane.
"""
- dp_index = dp_iface
- if isinstance(dp_iface, str):
- dp_index = self.get_sw_if_index(dp_iface)
- cp_index = cp_iface
- if isinstance(cp_iface, str):
- cp_index = self.get_sw_if_index(cp_iface)
self.__vpp_api_client.api.pppoe_add_del_cp(
dp_sw_if_index=dp_index,
cp_sw_if_index=cp_index,
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-16 04:19:12 +0000