diff --git a/python/vyos/configdep.py b/python/vyos/configdep.py
index e0fe1ddac..cf7c9d543 100644
--- a/python/vyos/configdep.py
+++ b/python/vyos/configdep.py
@@ -1,191 +1,211 @@
# Copyright 2023-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import os
import json
import typing
from inspect import stack
from graphlib import TopologicalSorter, CycleError
from vyos.utils.system import load_as_module
from vyos.configdict import dict_merge
from vyos.defaults import directories
from vyos.configsource import VyOSError
from vyos import ConfigError
# https://peps.python.org/pep-0484/#forward-references
# for type 'Config'
if typing.TYPE_CHECKING:
from vyos.config import Config
dependency_dir = os.path.join(directories['data'],
'config-mode-dependencies')
dependency_list: list[typing.Callable] = []
DEBUG = False
def debug_print(s: str):
if DEBUG:
print(s)
def canon_name(name: str) -> str:
return os.path.splitext(name)[0].replace('-', '_')
def canon_name_of_path(path: str) -> str:
script = os.path.basename(path)
return canon_name(script)
def caller_name() -> str:
filename = stack()[2].filename
return canon_name_of_path(filename)
def name_of(f: typing.Callable) -> str:
return f.__name__
def names_of(l: list[typing.Callable]) -> list[str]:
return [name_of(f) for f in l]
def remove_redundant(l: list[typing.Callable]) -> list[typing.Callable]:
names = set()
for e in reversed(l):
_ = l.remove(e) if name_of(e) in names else names.add(name_of(e))
def append_uniq(l: list[typing.Callable], e: typing.Callable):
"""Append an element, removing earlier occurrences
The list of dependencies is generally short and traversing the list on
each append is preferable to the cost of redundant script invocation.
"""
l.append(e)
remove_redundant(l)
def read_dependency_dict(dependency_dir: str = dependency_dir) -> dict:
res = {}
for dep_file in os.listdir(dependency_dir):
if not dep_file.endswith('.json'):
continue
path = os.path.join(dependency_dir, dep_file)
with open(path) as f:
d = json.load(f)
if dep_file == 'vyos-1x.json':
res = dict_merge(res, d)
else:
res = dict_merge(d, res)
return res
def get_dependency_dict(config: 'Config') -> dict:
if hasattr(config, 'cached_dependency_dict'):
d = getattr(config, 'cached_dependency_dict')
else:
d = read_dependency_dict()
setattr(config, 'cached_dependency_dict', d)
return d
-def run_config_mode_script(script: str, config: 'Config'):
+def run_config_mode_script(target: str, config: 'Config'):
+ script = target + '.py'
path = os.path.join(directories['conf_mode'], script)
name = canon_name(script)
mod = load_as_module(name, path)
config.set_level([])
try:
c = mod.get_config(config)
mod.verify(c)
mod.generate(c)
mod.apply(c)
except (VyOSError, ConfigError) as e:
raise ConfigError(str(e)) from e
+def run_conditionally(target: str, tagnode: str, config: 'Config'):
+ tag_ext = f'_{tagnode}' if tagnode else ''
+ script_name = f'{target}{tag_ext}'
+
+ scripts_called = getattr(config, 'scripts_called', [])
+ commit_scripts = getattr(config, 'commit_scripts', [])
+
+ debug_print(f'scripts_called: {scripts_called}')
+ debug_print(f'commit_scripts: {commit_scripts}')
+
+ if script_name in commit_scripts and script_name not in scripts_called:
+ debug_print(f'dependency {script_name} deferred to priority')
+ return
+
+ run_config_mode_script(target, config)
+
def def_closure(target: str, config: 'Config',
tagnode: typing.Optional[str] = None) -> typing.Callable:
- script = target + '.py'
def func_impl():
+ tag_value = ''
if tagnode is not None:
os.environ['VYOS_TAGNODE_VALUE'] = tagnode
- run_config_mode_script(script, config)
+ tag_value = tagnode
+ run_conditionally(target, tag_value, config)
+
tag_ext = f'_{tagnode}' if tagnode is not None else ''
func_impl.__name__ = f'{target}{tag_ext}'
+
return func_impl
def set_dependents(case: str, config: 'Config',
tagnode: typing.Optional[str] = None):
global dependency_list
dependency_list = config.dependency_list
d = get_dependency_dict(config)
k = caller_name()
l = dependency_list
for target in d[k][case]:
func = def_closure(target, config, tagnode)
append_uniq(l, func)
debug_print(f'set_dependents: caller {k}, current dependents {names_of(l)}')
def call_dependents():
k = caller_name()
l = dependency_list
debug_print(f'call_dependents: caller {k}, remaining dependents {names_of(l)}')
while l:
f = l.pop(0)
debug_print(f'calling: {f.__name__}')
try:
f()
except ConfigError as e:
s = f'dependent {f.__name__}: {str(e)}'
raise ConfigError(s) from e
def called_as_dependent() -> bool:
st = stack()[1:]
for f in st:
if f.filename == __file__:
return True
return False
def graph_from_dependency_dict(d: dict) -> dict:
g = {}
for k in list(d):
g[k] = set()
# add the dependencies for every sub-case; should there be cases
# that are mutally exclusive in the future, the graphs will be
# distinguished
for el in list(d[k]):
g[k] |= set(d[k][el])
return g
def is_acyclic(d: dict) -> bool:
g = graph_from_dependency_dict(d)
ts = TopologicalSorter(g)
try:
# get node iterator
order = ts.static_order()
# try iteration
_ = [*order]
except CycleError:
return False
return True
def check_dependency_graph(dependency_dir: str = dependency_dir,
supplement: str = None) -> bool:
d = read_dependency_dict(dependency_dir=dependency_dir)
if supplement is not None:
with open(supplement) as f:
d = dict_merge(json.load(f), d)
return is_acyclic(d)
diff --git a/python/vyos/configdiff.py b/python/vyos/configdiff.py
index f975df45d..b6d4a5558 100644
--- a/python/vyos/configdiff.py
+++ b/python/vyos/configdiff.py
@@ -1,399 +1,436 @@
# Copyright 2020-2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
from enum import IntFlag
from enum import auto
+from itertools import chain
from vyos.config import Config
from vyos.configtree import DiffTree
from vyos.configdict import dict_merge
from vyos.utils.dict import get_sub_dict
from vyos.utils.dict import mangle_dict_keys
from vyos.utils.dict import dict_search_args
+from vyos.utils.dict import dict_to_key_paths
from vyos.xml_ref import get_defaults
+from vyos.xml_ref import owner
+from vyos.xml_ref import priority
class ConfigDiffError(Exception):
"""
Raised on config dict access errors, for example, calling get_value on
a non-leaf node.
"""
pass
def enum_to_key(e):
return e.name.lower()
class Diff(IntFlag):
MERGE = auto()
DELETE = auto()
ADD = auto()
STABLE = auto()
ALL = Diff.MERGE | Diff.DELETE | Diff.ADD | Diff.STABLE
requires_effective = [enum_to_key(Diff.DELETE)]
target_defaults = [enum_to_key(Diff.MERGE)]
def _key_sets_from_dicts(session_dict, effective_dict):
session_keys = list(session_dict)
effective_keys = list(effective_dict)
ret = {}
stable_keys = [k for k in session_keys if k in effective_keys]
ret[enum_to_key(Diff.MERGE)] = session_keys
ret[enum_to_key(Diff.DELETE)] = [k for k in effective_keys if k not in stable_keys]
ret[enum_to_key(Diff.ADD)] = [k for k in session_keys if k not in stable_keys]
ret[enum_to_key(Diff.STABLE)] = stable_keys
return ret
def _dict_from_key_set(key_set, d):
# This will always be applied to a key_set obtained from a get_sub_dict,
# hence there is no possibility of KeyError, as get_sub_dict guarantees
# a return type of dict
ret = {k: d[k] for k in key_set}
return ret
def get_config_diff(config, key_mangling=None):
"""
Check type and return ConfigDiff instance.
"""
if not config or not isinstance(config, Config):
raise TypeError("argument must me a Config instance")
if key_mangling and not (isinstance(key_mangling, tuple) and \
(len(key_mangling) == 2) and \
isinstance(key_mangling[0], str) and \
isinstance(key_mangling[1], str)):
raise ValueError("key_mangling must be a tuple of two strings")
if hasattr(config, 'cached_diff_tree'):
diff_t = getattr(config, 'cached_diff_tree')
else:
diff_t = DiffTree(config._running_config, config._session_config)
setattr(config, 'cached_diff_tree', diff_t)
if hasattr(config, 'cached_diff_dict'):
diff_d = getattr(config, 'cached_diff_dict')
else:
diff_d = diff_t.dict
setattr(config, 'cached_diff_dict', diff_d)
return ConfigDiff(config, key_mangling, diff_tree=diff_t,
diff_dict=diff_d)
+def get_commit_scripts(config) -> list:
+ """Return the list of config scripts to be executed by commit
+
+ Return a list of the scripts to be called by commit for the proposed
+ config. The list is ordered by priority for reference, however, the
+ actual order of execution by the commit algorithm is not reflected
+ (delete vs. add queue), nor needed for current use.
+ """
+ if not config or not isinstance(config, Config):
+ raise TypeError("argument must me a Config instance")
+
+ if hasattr(config, 'commit_scripts'):
+ return getattr(config, 'commit_scripts')
+
+ D = get_config_diff(config)
+ d = D._diff_dict
+ s = set()
+ for p in chain(dict_to_key_paths(d['sub']), dict_to_key_paths(d['add'])):
+ p_owner = owner(p, with_tag=True)
+ if not p_owner:
+ continue
+ p_priority = priority(p)
+ if not p_priority:
+ # default priority in legacy commit-algorithm
+ p_priority = 0
+ p_priority = int(p_priority)
+ s.add((p_priority, p_owner))
+
+ res = [x[1] for x in sorted(s, key=lambda x: x[0])]
+ setattr(config, 'commit_scripts', res)
+
+ return res
+
class ConfigDiff(object):
"""
The class of config changes as represented by comparison between the
session config dict and the effective config dict.
"""
def __init__(self, config, key_mangling=None, diff_tree=None, diff_dict=None):
self._level = config.get_level()
self._session_config_dict = config.get_cached_root_dict(effective=False)
self._effective_config_dict = config.get_cached_root_dict(effective=True)
self._key_mangling = key_mangling
self._diff_tree = diff_tree
self._diff_dict = diff_dict
# mirrored from Config; allow path arguments relative to level
def _make_path(self, path):
if isinstance(path, str):
path = path.split()
elif isinstance(path, list):
pass
else:
raise TypeError("Path must be a whitespace-separated string or a list")
ret = self._level + path
return ret
def set_level(self, path):
"""
Set the *edit level*, that is, a relative config dict path.
Once set, all operations will be relative to this path,
for example, after ``set_level("system")``, calling
``get_value("name-server")`` is equivalent to calling
``get_value("system name-server")`` without ``set_level``.
Args:
path (str|list): relative config path
"""
if isinstance(path, str):
if path:
self._level = path.split()
else:
self._level = []
elif isinstance(path, list):
self._level = path.copy()
else:
raise TypeError("Level path must be either a whitespace-separated string or a list")
def get_level(self):
"""
Gets the current edit level.
Returns:
str: current edit level
"""
ret = self._level.copy()
return ret
def _mangle_dict_keys(self, config_dict):
config_dict = mangle_dict_keys(config_dict, self._key_mangling[0],
self._key_mangling[1])
return config_dict
def is_node_changed(self, path=[]):
if self._diff_tree is None:
raise NotImplementedError("diff_tree class not available")
if (self._diff_tree.add.exists(self._make_path(path)) or
self._diff_tree.sub.exists(self._make_path(path))):
return True
return False
def node_changed_presence(self, path=[]) -> bool:
if self._diff_tree is None:
raise NotImplementedError("diff_tree class not available")
path = self._make_path(path)
before = self._diff_tree.left.exists(path)
after = self._diff_tree.right.exists(path)
return (before and not after) or (not before and after)
def node_changed_children(self, path=[]) -> list:
if self._diff_tree is None:
raise NotImplementedError("diff_tree class not available")
path = self._make_path(path)
add = self._diff_tree.add
sub = self._diff_tree.sub
children = set()
if add.exists(path):
children.update(add.list_nodes(path))
if sub.exists(path):
children.update(sub.list_nodes(path))
return list(children)
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,
recursive=False):
"""
Args:
path (str|list): config path
expand_nodes=Diff(0): bit mask of enum indicating for which nodes
to provide full dict; for example, Diff.MERGE
will expand dict['merge'] into dict under
value
no_detaults=False: if expand_nodes & Diff.MERGE, do not merge default
values to ret['merge']
recursive: if true, use config_tree diff algorithm provided by
diff_tree class
Returns: dict of lists, representing differences between session
and effective config, under path
dict['merge'] = session config values
dict['delete'] = effective config values, not in session
dict['add'] = session config values, not in effective
dict['stable'] = config values in both session and effective
"""
session_dict = get_sub_dict(self._session_config_dict,
self._make_path(path), get_first_key=True)
if recursive:
if self._diff_tree is None:
raise NotImplementedError("diff_tree class not available")
else:
add = get_sub_dict(self._diff_dict, ['add'], get_first_key=True)
sub = get_sub_dict(self._diff_dict, ['sub'], get_first_key=True)
inter = get_sub_dict(self._diff_dict, ['inter'], get_first_key=True)
ret = {}
ret[enum_to_key(Diff.MERGE)] = session_dict
ret[enum_to_key(Diff.DELETE)] = get_sub_dict(sub, self._make_path(path),
get_first_key=True)
ret[enum_to_key(Diff.ADD)] = get_sub_dict(add, self._make_path(path),
get_first_key=True)
ret[enum_to_key(Diff.STABLE)] = get_sub_dict(inter, self._make_path(path),
get_first_key=True)
for e in Diff:
k = enum_to_key(e)
if not (e & expand_nodes):
ret[k] = list(ret[k])
else:
if self._key_mangling:
ret[k] = self._mangle_dict_keys(ret[k])
if k in target_defaults and not no_defaults:
default_values = get_defaults(self._make_path(path),
get_first_key=True,
recursive=True)
ret[k] = dict_merge(default_values, ret[k])
return ret
effective_dict = get_sub_dict(self._effective_config_dict,
self._make_path(path), get_first_key=True)
ret = _key_sets_from_dicts(session_dict, effective_dict)
if not expand_nodes:
return ret
for e in Diff:
if expand_nodes & e:
k = enum_to_key(e)
if k in requires_effective:
ret[k] = _dict_from_key_set(ret[k], effective_dict)
else:
ret[k] = _dict_from_key_set(ret[k], session_dict)
if self._key_mangling:
ret[k] = self._mangle_dict_keys(ret[k])
if k in target_defaults and not no_defaults:
default_values = get_defaults(self._make_path(path),
get_first_key=True,
recursive=True)
ret[k] = dict_merge(default_values, ret[k])
return ret
def get_node_diff(self, path=[], expand_nodes=Diff(0), no_defaults=False,
recursive=False):
"""
Args:
path (str|list): config path
expand_nodes=Diff(0): bit mask of enum indicating for which nodes
to provide full dict; for example, Diff.MERGE
will expand dict['merge'] into dict under
value
no_detaults=False: if expand_nodes & Diff.MERGE, do not merge default
values to ret['merge']
recursive: if true, use config_tree diff algorithm provided by
diff_tree class
Returns: dict of lists, representing differences between session
and effective config, at path
dict['merge'] = session config values
dict['delete'] = effective config values, not in session
dict['add'] = session config values, not in effective
dict['stable'] = config values in both session and effective
"""
session_dict = get_sub_dict(self._session_config_dict, self._make_path(path))
if recursive:
if self._diff_tree is None:
raise NotImplementedError("diff_tree class not available")
else:
add = get_sub_dict(self._diff_dict, ['add'], get_first_key=True)
sub = get_sub_dict(self._diff_dict, ['sub'], get_first_key=True)
inter = get_sub_dict(self._diff_dict, ['inter'], get_first_key=True)
ret = {}
ret[enum_to_key(Diff.MERGE)] = session_dict
ret[enum_to_key(Diff.DELETE)] = get_sub_dict(sub, self._make_path(path))
ret[enum_to_key(Diff.ADD)] = get_sub_dict(add, self._make_path(path))
ret[enum_to_key(Diff.STABLE)] = get_sub_dict(inter, self._make_path(path))
for e in Diff:
k = enum_to_key(e)
if not (e & expand_nodes):
ret[k] = list(ret[k])
else:
if self._key_mangling:
ret[k] = self._mangle_dict_keys(ret[k])
if k in target_defaults and not no_defaults:
default_values = get_defaults(self._make_path(path),
get_first_key=True,
recursive=True)
ret[k] = dict_merge(default_values, ret[k])
return ret
effective_dict = get_sub_dict(self._effective_config_dict, self._make_path(path))
ret = _key_sets_from_dicts(session_dict, effective_dict)
if not expand_nodes:
return ret
for e in Diff:
if expand_nodes & e:
k = enum_to_key(e)
if k in requires_effective:
ret[k] = _dict_from_key_set(ret[k], effective_dict)
else:
ret[k] = _dict_from_key_set(ret[k], session_dict)
if self._key_mangling:
ret[k] = self._mangle_dict_keys(ret[k])
if k in target_defaults and not no_defaults:
default_values = get_defaults(self._make_path(path),
get_first_key=True,
recursive=True)
ret[k] = dict_merge(default_values, ret[k])
return ret
def get_value_diff(self, path=[]):
"""
Args:
path (str|list): config path
Returns: (new, old) tuple of values in session config/effective config
"""
# one should properly use is_leaf as check; for the moment we will
# deduce from type, which will not catch call on non-leaf node if None
new_value_dict = get_sub_dict(self._session_config_dict, self._make_path(path))
old_value_dict = get_sub_dict(self._effective_config_dict, self._make_path(path))
new_value = None
old_value = None
if new_value_dict:
new_value = next(iter(new_value_dict.values()))
if old_value_dict:
old_value = next(iter(old_value_dict.values()))
if new_value and isinstance(new_value, dict):
raise ConfigDiffError("get_value_changed called on non-leaf node")
if old_value and isinstance(old_value, dict):
raise ConfigDiffError("get_value_changed called on non-leaf node")
return new_value, old_value
diff --git a/python/vyos/utils/dict.py b/python/vyos/utils/dict.py
index 062ab9c81..9a4671c5f 100644
--- a/python/vyos/utils/dict.py
+++ b/python/vyos/utils/dict.py
@@ -1,373 +1,374 @@
# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
def colon_separated_to_dict(data_string, uniquekeys=False):
""" Converts a string containing newline-separated entries
of colon-separated key-value pairs into a dict.
Such files are common in Linux /proc filesystem
Args:
data_string (str): data string
uniquekeys (bool): whether to insist that keys are unique or not
Returns: dict
Raises:
ValueError: if uniquekeys=True and the data string has
duplicate keys.
Note:
If uniquekeys=True, then dict entries are always strings,
otherwise they are always lists of strings.
"""
import re
key_value_re = re.compile('([^:]+)\s*\:\s*(.*)')
data_raw = re.split('\n', data_string)
data = {}
for l in data_raw:
l = l.strip()
if l:
match = re.match(key_value_re, l)
if match and (len(match.groups()) == 2):
key = match.groups()[0].strip()
value = match.groups()[1].strip()
else:
raise ValueError(f"""Line "{l}" could not be parsed a colon-separated pair """, l)
if key in data.keys():
if uniquekeys:
raise ValueError("Data string has duplicate keys: {0}".format(key))
else:
data[key].append(value)
else:
if uniquekeys:
data[key] = value
else:
data[key] = [value]
else:
pass
return data
def mangle_dict_keys(data, regex, replacement, abs_path=None, no_tag_node_value_mangle=False):
""" Mangles dict keys according to a regex and replacement character.
Some libraries like Jinja2 do not like certain characters in dict keys.
This function can be used for replacing all offending characters
with something acceptable.
Args:
data (dict): Original dict to mangle
regex, replacement (str): arguments to re.sub(regex, replacement, ...)
abs_path (list): if data is a config dict and no_tag_node_value_mangle is True
then abs_path should be the absolute config path to the first
keys of data, non-inclusive
no_tag_node_value_mangle (bool): do not mangle keys of tag node values
Returns: dict
"""
import re
from vyos.xml_ref import is_tag_value
if abs_path is None:
abs_path = []
new_dict = type(data)()
for k in data.keys():
if no_tag_node_value_mangle and is_tag_value(abs_path + [k]):
new_key = k
else:
new_key = re.sub(regex, replacement, k)
value = data[k]
if isinstance(value, dict):
new_dict[new_key] = mangle_dict_keys(value, regex, replacement,
abs_path=abs_path + [k],
no_tag_node_value_mangle=no_tag_node_value_mangle)
else:
new_dict[new_key] = value
return new_dict
def _get_sub_dict(d, lpath):
k = lpath[0]
if k not in d.keys():
return {}
c = {k: d[k]}
lpath = lpath[1:]
if not lpath:
return c
elif not isinstance(c[k], dict):
return {}
return _get_sub_dict(c[k], lpath)
def get_sub_dict(source, lpath, get_first_key=False):
""" Returns the sub-dict of a nested dict, defined by path of keys.
Args:
source (dict): Source dict to extract from
lpath (list[str]): sequence of keys
Returns: source, if lpath is empty, else
{key : source[..]..[key]} for key the last element of lpath, if exists
{} otherwise
"""
if not isinstance(source, dict):
raise TypeError("source must be of type dict")
if not isinstance(lpath, list):
raise TypeError("path must be of type list")
if not lpath:
return source
ret = _get_sub_dict(source, lpath)
if get_first_key and lpath and ret:
tmp = next(iter(ret.values()))
if not isinstance(tmp, dict):
raise TypeError("Data under node is not of type dict")
ret = tmp
return ret
def dict_search(path, dict_object):
""" Traverse Python dictionary (dict_object) delimited by dot (.).
Return value of key if found, None otherwise.
This is faster implementation then jmespath.search('foo.bar', dict_object)"""
if not isinstance(dict_object, dict) or not path:
return None
parts = path.split('.')
inside = parts[:-1]
if not inside:
if path not in dict_object:
return None
return dict_object[path]
c = dict_object
for p in parts[:-1]:
c = c.get(p, {})
return c.get(parts[-1], None)
def dict_search_args(dict_object, *path):
# Traverse dictionary using variable arguments
# Added due to above function not allowing for '.' in the key names
# Example: dict_search_args(some_dict, 'key', 'subkey', 'subsubkey', ...)
if not isinstance(dict_object, dict) or not path:
return None
for item in path:
if item not in dict_object:
return None
dict_object = dict_object[item]
return dict_object
def dict_search_recursive(dict_object, key, path=[]):
""" Traverse a dictionary recurisvely and return the value of the key
we are looking for.
Thankfully copied from https://stackoverflow.com/a/19871956
Modified to yield optional path to found keys
"""
if isinstance(dict_object, list):
for i in dict_object:
new_path = path + [i]
for x in dict_search_recursive(i, key, new_path):
yield x
elif isinstance(dict_object, dict):
if key in dict_object:
new_path = path + [key]
yield dict_object[key], new_path
for k, j in dict_object.items():
new_path = path + [k]
for x in dict_search_recursive(j, key, new_path):
yield x
def dict_set(key_path, value, dict_object):
""" Set value to Python dictionary (dict_object) using path to key delimited by dot (.).
The key will be added if it does not exist.
"""
path_list = key_path.split(".")
dynamic_dict = dict_object
if len(path_list) > 0:
for i in range(0, len(path_list)-1):
dynamic_dict = dynamic_dict[path_list[i]]
dynamic_dict[path_list[len(path_list)-1]] = value
def dict_delete(key_path, dict_object):
""" Delete key in Python dictionary (dict_object) using path to key delimited by dot (.).
"""
path_dict = dict_object
path_list = key_path.split('.')
inside = path_list[:-1]
if not inside:
del dict_object[path_list]
else:
for key in path_list[:-1]:
path_dict = path_dict[key]
del path_dict[path_list[len(path_list)-1]]
def dict_to_list(d, save_key_to=None):
""" Convert a dict to a list of dicts.
Optionally, save the original key of the dict inside
dicts stores in that list.
"""
def save_key(i, k):
if isinstance(i, dict):
i[save_key_to] = k
return
elif isinstance(i, list):
for _i in i:
save_key(_i, k)
else:
raise ValueError(f"Cannot save the key: the item is {type(i)}, not a dict")
collect = []
for k,_ in d.items():
item = d[k]
if save_key_to is not None:
save_key(item, k)
if isinstance(item, list):
collect += item
else:
collect.append(item)
return collect
def dict_to_paths_values(conf: dict) -> dict:
"""
Convert nested dictionary to simple dictionary, where key is a path is delimited by dot (.).
"""
list_of_paths = []
dict_of_options ={}
for path in dict_to_key_paths(conf):
str_path = '.'.join(path)
list_of_paths.append(str_path)
for path in list_of_paths:
dict_of_options[path] = dict_search(path,conf)
return dict_of_options
+
def dict_to_key_paths(d: dict) -> list:
""" Generator to return list of key paths from dict of list[str]|str
"""
def func(d, path):
if isinstance(d, dict):
if not d:
yield path
for k, v in d.items():
for r in func(v, path + [k]):
yield r
elif isinstance(d, list):
yield path
elif isinstance(d, str):
yield path
else:
raise ValueError('object is not a dict of strings/list of strings')
for r in func(d, []):
yield r
def dict_to_paths(d: dict) -> list:
""" Generator to return list of paths from dict of list[str]|str
"""
def func(d, path):
if isinstance(d, dict):
if not d:
yield path
for k, v in d.items():
for r in func(v, path + [k]):
yield r
elif isinstance(d, list):
for i in d:
for r in func(i, path):
yield r
elif isinstance(d, str):
yield path + [d]
else:
raise ValueError('object is not a dict of strings/list of strings')
for r in func(d, []):
yield r
def embed_dict(p: list[str], d: dict) -> dict:
path = p.copy()
ret = d
while path:
ret = {path.pop(): ret}
return ret
def check_mutually_exclusive_options(d, keys, required=False):
""" Checks if a dict has at most one or only one of
mutually exclusive keys.
"""
present_keys = []
for k in d:
if k in keys:
present_keys.append(k)
# Un-mangle the keys to make them match CLI option syntax
from re import sub
orig_keys = list(map(lambda s: sub(r'_', '-', s), keys))
orig_present_keys = list(map(lambda s: sub(r'_', '-', s), present_keys))
if len(present_keys) > 1:
raise ValueError(f"Options {orig_keys} are mutually-exclusive but more than one of them is present: {orig_present_keys}")
if required and (len(present_keys) < 1):
raise ValueError(f"At least one of the following options is required: {orig_keys}")
class FixedDict(dict):
"""
FixedDict: A dictionnary not allowing new keys to be created after initialisation.
>>> f = FixedDict(**{'count':1})
>>> f['count'] = 2
>>> f['king'] = 3
File "...", line ..., in __setitem__
raise ConfigError(f'Option "{k}" has no defined default')
"""
from vyos import ConfigError
def __init__(self, **options):
self._allowed = options.keys()
super().__init__(**options)
def __setitem__(self, k, v):
"""
__setitem__ is a builtin which is called by python when setting dict values:
>>> d = dict()
>>> d['key'] = 'value'
>>> d
{'key': 'value'}
is syntaxic sugar for
>>> d = dict()
>>> d.__setitem__('key','value')
>>> d
{'key': 'value'}
"""
if k not in self._allowed:
raise ConfigError(f'Option "{k}" has no defined default')
super().__setitem__(k, v)
diff --git a/python/vyos/xml_ref/__init__.py b/python/vyos/xml_ref/__init__.py
index 2ba3da4e8..7e9bf2b43 100644
--- a/python/vyos/xml_ref/__init__.py
+++ b/python/vyos/xml_ref/__init__.py
@@ -1,89 +1,89 @@
-# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
from typing import Optional, Union, TYPE_CHECKING
from vyos.xml_ref import definition
if TYPE_CHECKING:
from vyos.config import ConfigDict
def load_reference(cache=[]):
if cache:
return cache[0]
xml = definition.Xml()
try:
from vyos.xml_ref.cache import reference
except Exception:
raise ImportError('no xml reference cache !!')
if not reference:
raise ValueError('empty xml reference cache !!')
xml.define(reference)
cache.append(xml)
return xml
def is_tag(path: list) -> bool:
return load_reference().is_tag(path)
def is_tag_value(path: list) -> bool:
return load_reference().is_tag_value(path)
def is_multi(path: list) -> bool:
return load_reference().is_multi(path)
def is_valueless(path: list) -> bool:
return load_reference().is_valueless(path)
def is_leaf(path: list) -> bool:
return load_reference().is_leaf(path)
-def owner(path: list) -> str:
- return load_reference().owner(path)
+def owner(path: list, with_tag=False) -> str:
+ return load_reference().owner(path, with_tag=with_tag)
def priority(path: list) -> str:
return load_reference().priority(path)
def cli_defined(path: list, node: str, non_local=False) -> bool:
return load_reference().cli_defined(path, node, non_local=non_local)
def component_version() -> dict:
return load_reference().component_version()
def default_value(path: list) -> Optional[Union[str, list]]:
return load_reference().default_value(path)
def multi_to_list(rpath: list, conf: dict) -> dict:
return load_reference().multi_to_list(rpath, conf)
def get_defaults(path: list, get_first_key=False, recursive=False) -> dict:
return load_reference().get_defaults(path, get_first_key=get_first_key,
recursive=recursive)
def relative_defaults(rpath: list, conf: dict, get_first_key=False,
recursive=False) -> dict:
return load_reference().relative_defaults(rpath, conf,
get_first_key=get_first_key,
recursive=recursive)
def from_source(d: dict, path: list) -> bool:
return definition.from_source(d, path)
def ext_dict_merge(source: dict, destination: Union[dict, 'ConfigDict']):
return definition.ext_dict_merge(source, destination)
diff --git a/python/vyos/xml_ref/definition.py b/python/vyos/xml_ref/definition.py
index c85835ffd..5ff28daed 100644
--- a/python/vyos/xml_ref/definition.py
+++ b/python/vyos/xml_ref/definition.py
@@ -1,329 +1,339 @@
-# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
from typing import Optional, Union, Any, TYPE_CHECKING
# https://peps.python.org/pep-0484/#forward-references
# for type 'ConfigDict'
if TYPE_CHECKING:
from vyos.config import ConfigDict
def set_source_recursive(o: Union[dict, str, list], b: bool):
d = {}
if not isinstance(o, dict):
d = {'_source': b}
else:
for k, v in o.items():
d[k] = set_source_recursive(v, b)
d |= {'_source': b}
return d
def source_dict_merge(src: dict, dest: dict):
from copy import deepcopy
dst = deepcopy(dest)
from_src = {}
for key, value in src.items():
if key not in dst:
dst[key] = value
from_src[key] = set_source_recursive(value, True)
elif isinstance(src[key], dict):
dst[key], f = source_dict_merge(src[key], dst[key])
f |= {'_source': False}
from_src[key] = f
return dst, from_src
def ext_dict_merge(src: dict, dest: Union[dict, 'ConfigDict']):
d, f = source_dict_merge(src, dest)
if hasattr(d, '_from_defaults'):
setattr(d, '_from_defaults', f)
return d
def from_source(d: dict, path: list) -> bool:
for key in path:
d = d[key] if key in d else {}
if not d or not isinstance(d, dict):
return False
return d.get('_source', False)
class Xml:
def __init__(self):
self.ref = {}
def define(self, ref: dict):
self.ref = ref
def _get_ref_node_data(self, node: dict, data: str) -> Union[bool, str]:
res = node.get('node_data', {})
if not res:
raise ValueError("non-existent node data")
if data not in res:
raise ValueError("non-existent data field")
return res.get(data)
def _get_ref_path(self, path: list) -> dict:
ref_path = path.copy()
d = self.ref
while ref_path and d:
d = d.get(ref_path[0], {})
ref_path.pop(0)
if self._is_tag_node(d) and ref_path:
ref_path.pop(0)
return d
def _is_tag_node(self, node: dict) -> bool:
res = self._get_ref_node_data(node, 'node_type')
return res == 'tag'
def is_tag(self, path: list) -> bool:
ref_path = path.copy()
d = self.ref
while ref_path and d:
d = d.get(ref_path[0], {})
ref_path.pop(0)
if self._is_tag_node(d) and ref_path:
if len(ref_path) == 1:
return False
ref_path.pop(0)
return self._is_tag_node(d)
def is_tag_value(self, path: list) -> bool:
if len(path) < 2:
return False
return self.is_tag(path[:-1])
def _is_multi_node(self, node: dict) -> bool:
b = self._get_ref_node_data(node, 'multi')
assert isinstance(b, bool)
return b
def is_multi(self, path: list) -> bool:
d = self._get_ref_path(path)
return self._is_multi_node(d)
def _is_valueless_node(self, node: dict) -> bool:
b = self._get_ref_node_data(node, 'valueless')
assert isinstance(b, bool)
return b
def is_valueless(self, path: list) -> bool:
d = self._get_ref_path(path)
return self._is_valueless_node(d)
def _is_leaf_node(self, node: dict) -> bool:
res = self._get_ref_node_data(node, 'node_type')
return res == 'leaf'
def is_leaf(self, path: list) -> bool:
d = self._get_ref_path(path)
return self._is_leaf_node(d)
def _least_upper_data(self, path: list, name: str) -> str:
ref_path = path.copy()
d = self.ref
data = ''
+ tag = ''
while ref_path and d:
+ tag_val = ''
d = d.get(ref_path[0], {})
ref_path.pop(0)
if self._is_tag_node(d) and ref_path:
+ tag_val = ref_path[0]
ref_path.pop(0)
if self._is_leaf_node(d) and ref_path:
ref_path.pop(0)
res = self._get_ref_node_data(d, name)
if res is not None:
data = res
+ tag = tag_val
- return data
+ return data, tag
- def owner(self, path: list) -> str:
+ def owner(self, path: list, with_tag=False) -> str:
from pathlib import Path
- data = self._least_upper_data(path, 'owner')
+ data, tag = self._least_upper_data(path, 'owner')
+ tag_ext = f'_{tag}' if tag else ''
if data:
- data = Path(data.split()[0]).name
+ if with_tag:
+ data = Path(data.split()[0]).stem
+ data = f'{data}{tag_ext}'
+ else:
+ data = Path(data.split()[0]).name
return data
def priority(self, path: list) -> str:
- return self._least_upper_data(path, 'priority')
+ data, _ = self._least_upper_data(path, 'priority')
+ return data
@staticmethod
def _dict_get(d: dict, path: list) -> dict:
for i in path:
d = d.get(i, {})
if not isinstance(d, dict):
return {}
if not d:
break
return d
def _dict_find(self, d: dict, key: str, non_local=False) -> bool:
for k in list(d):
if k in ('node_data', 'component_version'):
continue
if k == key:
return True
if non_local and isinstance(d[k], dict):
if self._dict_find(d[k], key):
return True
return False
def cli_defined(self, path: list, node: str, non_local=False) -> bool:
d = self._dict_get(self.ref, path)
return self._dict_find(d, node, non_local=non_local)
def component_version(self) -> dict:
d = {}
for k, v in self.ref['component_version'].items():
d[k] = int(v)
return d
def multi_to_list(self, rpath: list, conf: dict) -> dict:
res: Any = {}
for k in list(conf):
d = self._get_ref_path(rpath + [k])
if self._is_leaf_node(d):
if self._is_multi_node(d) and not isinstance(conf[k], list):
res[k] = [conf[k]]
else:
res[k] = conf[k]
else:
res[k] = self.multi_to_list(rpath + [k], conf[k])
return res
def _get_default_value(self, node: dict) -> Optional[str]:
return self._get_ref_node_data(node, "default_value")
def _get_default(self, node: dict) -> Optional[Union[str, list]]:
default = self._get_default_value(node)
if default is None:
return None
if self._is_multi_node(node):
return default.split()
return default
def default_value(self, path: list) -> Optional[Union[str, list]]:
d = self._get_ref_path(path)
default = self._get_default_value(d)
if default is None:
return None
if self._is_multi_node(d) or self._is_tag_node(d):
return default.split()
return default
def get_defaults(self, path: list, get_first_key=False, recursive=False) -> dict:
"""Return dict containing default values below path
Note that descent below path will not proceed beyond an encountered
tag node, as no tag node value is known. For a default dict relative
to an existing config dict containing tag node values, see function:
'relative_defaults'
"""
res: dict = {}
if self.is_tag(path):
return res
d = self._get_ref_path(path)
if self._is_leaf_node(d):
default_value = self._get_default(d)
if default_value is not None:
return {path[-1]: default_value} if path else {}
for k in list(d):
if k in ('node_data', 'component_version') :
continue
if self._is_leaf_node(d[k]):
default_value = self._get_default(d[k])
if default_value is not None:
res |= {k: default_value}
elif self.is_tag(path + [k]):
# tag node defaults are used as suggestion, not default value;
# should this change, append to path and continue if recursive
pass
else:
if recursive:
pos = self.get_defaults(path + [k], recursive=True)
res |= pos
if res:
if get_first_key or not path:
return res
return {path[-1]: res}
return {}
def _well_defined(self, path: list, conf: dict) -> bool:
# test disjoint path + conf for sensible config paths
def step(c):
return [next(iter(c.keys()))] if c else []
try:
tmp = step(conf)
if tmp and self.is_tag_value(path + tmp):
c = conf[tmp[0]]
if not isinstance(c, dict):
raise ValueError
tmp = tmp + step(c)
self._get_ref_path(path + tmp)
else:
self._get_ref_path(path + tmp)
except ValueError:
return False
return True
def _relative_defaults(self, rpath: list, conf: dict, recursive=False) -> dict:
res: dict = {}
res = self.get_defaults(rpath, recursive=recursive,
get_first_key=True)
for k in list(conf):
if isinstance(conf[k], dict):
step = self._relative_defaults(rpath + [k], conf=conf[k],
recursive=recursive)
res |= step
if res:
return {rpath[-1]: res} if rpath else res
return {}
def relative_defaults(self, path: list, conf: dict, get_first_key=False,
recursive=False) -> dict:
"""Return dict containing defaults along paths of a config dict
"""
if not conf:
return self.get_defaults(path, get_first_key=get_first_key,
recursive=recursive)
if not self._well_defined(path, conf):
# adjust for possible overlap:
if path and path[-1] in list(conf):
conf = conf[path[-1]]
conf = {} if not isinstance(conf, dict) else conf
if not self._well_defined(path, conf):
print('path to config dict does not define full config paths')
return {}
res = self._relative_defaults(path, conf, recursive=recursive)
if get_first_key and path:
if res.values():
res = next(iter(res.values()))
else:
res = {}
return res
diff --git a/smoketest/scripts/cli/test_config_dependency.py b/smoketest/scripts/cli/test_config_dependency.py
index 14e88321a..99e807ac5 100755
--- a/smoketest/scripts/cli/test_config_dependency.py
+++ b/smoketest/scripts/cli/test_config_dependency.py
@@ -1,49 +1,130 @@
#!/usr/bin/env python3
# Copyright 2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import unittest
+from time import sleep
-from base_vyostest_shim import VyOSUnitTestSHIM
-
+from vyos.utils.process import is_systemd_service_running
+from vyos.utils.process import cmd
from vyos.configsession import ConfigSessionError
+from base_vyostest_shim import VyOSUnitTestSHIM
+
class TestConfigDep(VyOSUnitTestSHIM.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ # smoketests are run without configd in 1.4; with configd in 1.5
+ # the tests below check behavior under configd:
+ # test_configdep_error checks for regression under configd (T6559)
+ # test_configdep_prio_queue checks resolution under configd (T6671)
+ cls.running_state = is_systemd_service_running('vyos-configd.service')
+
+ if not cls.running_state:
+ cmd('sudo systemctl start vyos-configd.service')
+ # allow time for init
+ sleep(1)
+
+ super(TestConfigDep, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestConfigDep, cls).tearDownClass()
+
+ # return to running_state
+ if not cls.running_state:
+ cmd('sudo systemctl stop vyos-configd.service')
+
def test_configdep_error(self):
address_group = 'AG'
address = '192.168.137.5'
nat_base = ['nat', 'source', 'rule', '10']
interface = 'eth1'
self.cli_set(['firewall', 'group', 'address-group', address_group,
'address', address])
self.cli_set(nat_base + ['outbound-interface', 'name', interface])
self.cli_set(nat_base + ['source', 'group', 'address-group', address_group])
self.cli_set(nat_base + ['translation', 'address', 'masquerade'])
self.cli_commit()
self.cli_delete(['firewall'])
# check error in call to dependent script (nat)
with self.assertRaises(ConfigSessionError):
self.cli_commit()
# clean up remaining
self.cli_delete(['nat'])
self.cli_commit()
+ def test_configdep_prio_queue(self):
+ # confirm that that a dependency (in this case, conntrack ->
+ # conntrack-sync) is not immediately called if the target is
+ # scheduled in the priority queue, indicating that it may require an
+ # intermediate activitation (bond0)
+ bonding_base = ['interfaces', 'bonding']
+ bond_interface = 'bond0'
+ bond_address = '192.0.2.1/24'
+ vrrp_group_base = ['high-availability', 'vrrp', 'group']
+ vrrp_sync_group_base = ['high-availability', 'vrrp', 'sync-group']
+ vrrp_group = 'ETH2'
+ vrrp_sync_group = 'GROUP'
+ conntrack_sync_base = ['service', 'conntrack-sync']
+ conntrack_peer = '192.0.2.77'
+
+ # simple set to trigger in-session conntrack -> conntrack-sync
+ # dependency; note that this is triggered on boot in 1.4 due to
+ # default 'system conntrack modules'
+ self.cli_set(['system', 'conntrack', 'table-size', '524288'])
+
+ self.cli_set(['interfaces', 'ethernet', 'eth2', 'address',
+ '198.51.100.2/24'])
+
+ self.cli_set(bonding_base + [bond_interface, 'address',
+ bond_address])
+ self.cli_set(bonding_base + [bond_interface, 'member', 'interface',
+ 'eth3'])
+
+ self.cli_set(vrrp_group_base + [vrrp_group, 'address',
+ '198.51.100.200/24'])
+ self.cli_set(vrrp_group_base + [vrrp_group, 'hello-source-address',
+ '198.51.100.2'])
+ self.cli_set(vrrp_group_base + [vrrp_group, 'interface', 'eth2'])
+ self.cli_set(vrrp_group_base + [vrrp_group, 'priority', '200'])
+ self.cli_set(vrrp_group_base + [vrrp_group, 'vrid', '22'])
+ self.cli_set(vrrp_sync_group_base + [vrrp_sync_group, 'member',
+ vrrp_group])
+
+ self.cli_set(conntrack_sync_base + ['failover-mechanism', 'vrrp',
+ 'sync-group', vrrp_sync_group])
+
+ self.cli_set(conntrack_sync_base + ['interface', bond_interface,
+ 'peer', conntrack_peer])
+
+ self.cli_commit()
+
+ # clean up
+ self.cli_delete(bonding_base)
+ self.cli_delete(vrrp_group_base)
+ self.cli_delete(vrrp_sync_group_base)
+ self.cli_delete(conntrack_sync_base)
+ self.cli_delete(['interfaces', 'ethernet', 'eth2', 'address'])
+ self.cli_delete(['system', 'conntrack', 'table-size'])
+ self.cli_commit()
+
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/services/vyos-configd b/src/services/vyos-configd
index d797e90cf..3674d9627 100755
--- a/src/services/vyos-configd
+++ b/src/services/vyos-configd
@@ -1,312 +1,327 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-2024 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 os
import sys
import grp
import re
import json
import typing
import logging
import signal
import importlib.util
import zmq
from contextlib import contextmanager
from vyos.defaults import directories
from vyos.utils.boot import boot_configuration_complete
from vyos.configsource import ConfigSourceString
from vyos.configsource import ConfigSourceError
+from vyos.configdiff import get_commit_scripts
from vyos.config import Config
from vyos import ConfigError
CFG_GROUP = 'vyattacfg'
script_stdout_log = '/tmp/vyos-configd-script-stdout'
debug = True
logger = logging.getLogger(__name__)
logs_handler = logging.StreamHandler()
logger.addHandler(logs_handler)
if debug:
logger.setLevel(logging.DEBUG)
else:
logger.setLevel(logging.INFO)
SOCKET_PATH = "ipc:///run/vyos-configd.sock"
# Response error codes
R_SUCCESS = 1
R_ERROR_COMMIT = 2
R_ERROR_DAEMON = 4
R_PASS = 8
vyos_conf_scripts_dir = directories['conf_mode']
configd_include_file = os.path.join(directories['data'], 'configd-include.json')
configd_env_set_file = os.path.join(directories['data'], 'vyos-configd-env-set')
configd_env_unset_file = os.path.join(directories['data'], 'vyos-configd-env-unset')
# sourced on entering config session
configd_env_file = '/etc/default/vyos-configd-env'
session_out = None
session_mode = None
def key_name_from_file_name(f):
return os.path.splitext(f)[0]
def module_name_from_key(k):
return k.replace('-', '_')
def path_from_file_name(f):
return os.path.join(vyos_conf_scripts_dir, f)
# opt-in to be run by daemon
with open(configd_include_file) as f:
try:
include = json.load(f)
except OSError as e:
logger.critical(f"configd include file error: {e}")
sys.exit(1)
except json.JSONDecodeError as e:
logger.critical(f"JSON load error: {e}")
sys.exit(1)
# import conf_mode scripts
(_, _, filenames) = next(iter(os.walk(vyos_conf_scripts_dir)))
filenames.sort()
load_filenames = [f for f in filenames if f in include]
imports = [key_name_from_file_name(f) for f in load_filenames]
module_names = [module_name_from_key(k) for k in imports]
paths = [path_from_file_name(f) for f in load_filenames]
to_load = list(zip(module_names, paths))
modules = []
for x in to_load:
spec = importlib.util.spec_from_file_location(x[0], x[1])
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
modules.append(module)
conf_mode_scripts = dict(zip(imports, modules))
exclude_set = {key_name_from_file_name(f) for f in filenames if f not in include}
include_set = {key_name_from_file_name(f) for f in filenames if f in include}
@contextmanager
def stdout_redirected(filename, mode):
saved_stdout_fd = None
destination_file = None
try:
sys.stdout.flush()
saved_stdout_fd = os.dup(sys.stdout.fileno())
destination_file = open(filename, mode)
os.dup2(destination_file.fileno(), sys.stdout.fileno())
yield
finally:
if saved_stdout_fd is not None:
os.dup2(saved_stdout_fd, sys.stdout.fileno())
os.close(saved_stdout_fd)
if destination_file is not None:
destination_file.close()
def explicit_print(path, mode, msg):
try:
with open(path, mode) as f:
f.write(f"\n{msg}\n\n")
except OSError:
logger.critical("error explicit_print")
def run_script(script_name, config, args) -> int:
script = conf_mode_scripts[script_name]
script.argv = args
config.set_level([])
try:
c = script.get_config(config)
script.verify(c)
script.generate(c)
script.apply(c)
except ConfigError as e:
logger.error(e)
explicit_print(session_out, session_mode, str(e))
return R_ERROR_COMMIT
except Exception as e:
logger.critical(e)
return R_ERROR_DAEMON
return R_SUCCESS
def initialization(socket):
global session_out
global session_mode
# Reset config strings:
active_string = ''
session_string = ''
# check first for resent init msg, in case of client timeout
while True:
msg = socket.recv().decode("utf-8", "ignore")
try:
message = json.loads(msg)
if message["type"] == "init":
resp = "init"
socket.send(resp.encode())
except:
break
# zmq synchronous for ipc from single client:
active_string = msg
resp = "active"
socket.send(resp.encode())
session_string = socket.recv().decode("utf-8", "ignore")
resp = "session"
socket.send(resp.encode())
pid_string = socket.recv().decode("utf-8", "ignore")
resp = "pid"
socket.send(resp.encode())
sudo_user_string = socket.recv().decode("utf-8", "ignore")
resp = "sudo_user"
socket.send(resp.encode())
temp_config_dir_string = socket.recv().decode("utf-8", "ignore")
resp = "temp_config_dir"
socket.send(resp.encode())
changes_only_dir_string = socket.recv().decode("utf-8", "ignore")
resp = "changes_only_dir"
socket.send(resp.encode())
logger.debug(f"config session pid is {pid_string}")
logger.debug(f"config session sudo_user is {sudo_user_string}")
try:
session_out = os.readlink(f"/proc/{pid_string}/fd/1")
session_mode = 'w'
except FileNotFoundError:
session_out = None
# if not a 'live' session, for example on boot, write to file
if not session_out or not boot_configuration_complete():
session_out = script_stdout_log
session_mode = 'a'
os.environ['SUDO_USER'] = sudo_user_string
if temp_config_dir_string:
os.environ['VYATTA_TEMP_CONFIG_DIR'] = temp_config_dir_string
if changes_only_dir_string:
os.environ['VYATTA_CHANGES_ONLY_DIR'] = changes_only_dir_string
try:
configsource = ConfigSourceString(running_config_text=active_string,
session_config_text=session_string)
except ConfigSourceError as e:
logger.debug(e)
return None
config = Config(config_source=configsource)
dependent_func: dict[str, list[typing.Callable]] = {}
setattr(config, 'dependent_func', dependent_func)
+ commit_scripts = get_commit_scripts(config)
+ logger.debug(f'commit_scripts: {commit_scripts}')
+
+ scripts_called = []
+ setattr(config, 'scripts_called', scripts_called)
+
return config
def process_node_data(config, data, last: bool = False) -> int:
if not config:
logger.critical(f"Empty config")
return R_ERROR_DAEMON
script_name = None
+ os.environ['VYOS_TAGNODE_VALUE'] = ''
args = []
config.dependency_list.clear()
res = re.match(r'^(VYOS_TAGNODE_VALUE=[^/]+)?.*\/([^/]+).py(.*)', data)
if res.group(1):
env = res.group(1).split('=')
os.environ[env[0]] = env[1]
if res.group(2):
script_name = res.group(2)
if not script_name:
logger.critical(f"Missing script_name")
return R_ERROR_DAEMON
if res.group(3):
args = res.group(3).split()
args.insert(0, f'{script_name}.py')
+ tag_value = os.getenv('VYOS_TAGNODE_VALUE', '')
+ tag_ext = f'_{tag_value}' if tag_value else ''
+ script_record = f'{script_name}{tag_ext}'
+ scripts_called = getattr(config, 'scripts_called', [])
+ scripts_called.append(script_record)
+
if script_name not in include_set:
return R_PASS
with stdout_redirected(session_out, session_mode):
result = run_script(script_name, config, args)
return result
def remove_if_file(f: str):
try:
os.remove(f)
except FileNotFoundError:
pass
except OSError:
raise
def shutdown():
remove_if_file(configd_env_file)
os.symlink(configd_env_unset_file, configd_env_file)
sys.exit(0)
if __name__ == '__main__':
context = zmq.Context()
socket = context.socket(zmq.REP)
# Set the right permissions on the socket, then change it back
o_mask = os.umask(0)
socket.bind(SOCKET_PATH)
os.umask(o_mask)
cfg_group = grp.getgrnam(CFG_GROUP)
os.setgid(cfg_group.gr_gid)
os.environ['VYOS_CONFIGD'] = 't'
def sig_handler(signum, frame):
shutdown()
signal.signal(signal.SIGTERM, sig_handler)
signal.signal(signal.SIGINT, sig_handler)
# Define the vyshim environment variable
remove_if_file(configd_env_file)
os.symlink(configd_env_set_file, configd_env_file)
config = None
while True:
# Wait for next request from client
msg = socket.recv().decode()
logger.debug(f"Received message: {msg}")
message = json.loads(msg)
if message["type"] == "init":
resp = "init"
socket.send(resp.encode())
config = initialization(socket)
elif message["type"] == "node":
- if message["last"]:
- logger.debug(f'final element of priority queue')
res = process_node_data(config, message["data"], message["last"])
response = res.to_bytes(1, byteorder=sys.byteorder)
logger.debug(f"Sending response {res}")
socket.send(response)
+ if message["last"] and config:
+ scripts_called = getattr(config, 'scripts_called', [])
+ logger.debug(f'scripts_called: {scripts_called}')
else:
logger.critical(f"Unexpected message: {message}")