Page MenuHomeVyOS Platform
Files F119062637

No OneTemporary
Actions

Size
20 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/python/vyos/configquery.py b/python/vyos/configquery.py
index 71ad5b4f0..5d6ca9be9 100644
--- a/python/vyos/configquery.py
+++ b/python/vyos/configquery.py
@@ -1,136 +1,192 @@
-# Copyright 2021-2023 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2021-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/>.
'''
A small library that allows querying existence or value(s) of config
settings from op mode, and execution of arbitrary op mode commands.
'''
import os
+import json
+import subprocess
from vyos.utils.process import STDOUT
from vyos.utils.process import popen
from vyos.utils.boot import boot_configuration_complete
from vyos.config import Config
from vyos.configsource import ConfigSourceSession, ConfigSourceString
from vyos.defaults import directories
+from vyos.configtree import ConfigTree
+from vyos.utils.dict import embed_dict
+from vyos.utils.dict import get_sub_dict
+from vyos.utils.dict import mangle_dict_keys
+from vyos.utils.error import cli_shell_api_err
+from vyos.xml_ref import multi_to_list
+from vyos.xml_ref import is_tag
+from vyos.base import Warning
config_file = os.path.join(directories['config'], 'config.boot')
class ConfigQueryError(Exception):
pass
class GenericConfigQuery:
def __init__(self):
pass
def exists(self, path: list):
raise NotImplementedError
def value(self, path: list):
raise NotImplementedError
def values(self, path: list):
raise NotImplementedError
class GenericOpRun:
def __init__(self):
pass
def run(self, path: list, **kwargs):
raise NotImplementedError
class CliShellApiConfigQuery(GenericConfigQuery):
def __init__(self):
super().__init__()
def exists(self, path: list):
cmd = ' '.join(path)
(_, err) = popen(f'cli-shell-api existsActive {cmd}')
if err:
return False
return True
def value(self, path: list):
cmd = ' '.join(path)
(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) = popen(f'cli-shell-api returnActiveValues {cmd}')
if err:
raise ConfigQueryError('No values for given path')
return out
class ConfigTreeQuery(GenericConfigQuery):
def __init__(self):
super().__init__()
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:
config_string = ''
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.config.exists(path)
def value(self, path: list):
return self.config.return_value(path)
def values(self, path: list):
return self.config.return_values(path)
def list_nodes(self, path: list):
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.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)
class VbashOpRun(GenericOpRun):
def __init__(self):
super().__init__()
def run(self, path: list, **kwargs):
cmd = ' '.join(path)
(out, err) = popen(f'/opt/vyatta/bin/vyatta-op-cmd-wrapper {cmd}', stderr=STDOUT, **kwargs)
if err:
raise ConfigQueryError(out)
return out
def query_context(config_query_class=CliShellApiConfigQuery,
op_run_class=VbashOpRun):
query = config_query_class()
run = op_run_class()
return query, run
-
+def verify_mangling(key_mangling):
+ if 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")
+
+def op_mode_run(cmd):
+ """ low-level to avoid overhead """
+ p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
+ out = p.stdout.read()
+ p.wait()
+ return p.returncode, out.decode()
+
+def op_mode_config_dict(path=None, key_mangling=None,
+ no_tag_node_value_mangle=False,
+ no_multi_convert=False, get_first_key=False):
+
+ if path is None:
+ path = []
+ command = ['/bin/cli-shell-api', '--show-active-only', 'showConfig']
+
+ rc, out = op_mode_run(command + path)
+ if rc == cli_shell_api_err.VYOS_EMPTY_CONFIG:
+ out = ''
+ if rc == cli_shell_api_err.VYOS_INVALID_PATH:
+ Warning(out)
+ return {}
+
+ ct = ConfigTree(out)
+ d = json.loads(ct.to_json())
+ # cli-shell-api output includes last path component if tag node
+ if is_tag(path):
+ config_dict = embed_dict(path[:-1], d)
+ else:
+ config_dict = embed_dict(path, d)
+
+ if not no_multi_convert:
+ config_dict = multi_to_list([], config_dict)
+
+ if key_mangling is not None:
+ verify_mangling(key_mangling)
+ config_dict = mangle_dict_keys(config_dict,
+ key_mangling[0], key_mangling[1],
+ no_tag_node_value_mangle=no_tag_node_value_mangle)
+
+ return get_sub_dict(config_dict, path, get_first_key=get_first_key)
diff --git a/python/vyos/utils/dict.py b/python/vyos/utils/dict.py
index d36b6fcfb..062ab9c81 100644
--- a/python/vyos/utils/dict.py
+++ b/python/vyos/utils/dict.py
@@ -1,366 +1,373 @@
# 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/utils/error.py b/python/vyos/utils/error.py
new file mode 100644
index 000000000..8d4709bff
--- /dev/null
+++ b/python/vyos/utils/error.py
@@ -0,0 +1,24 @@
+# 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 enum import IntEnum
+
+class cli_shell_api_err(IntEnum):
+ """ vyatta-cfg/src/vyos-errors.h """
+ VYOS_SUCCESS = 0
+ VYOS_GENERAL_FAILURE = 1
+ VYOS_INVALID_PATH = 2
+ VYOS_EMPTY_CONFIG = 3
+ VYOS_CONFIG_PARSE_ERROR = 4

File Metadata

Mime Type
text/x-diff
Expires
Fri, Nov 6, 7:07 AM (14 h, 35 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
4652998
Default Alt Text
(20 KB)

Event Timeline