# Copyright 2020 VyOS maintainers and contributors # # 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 . from enum import IntFlag, auto from vyos.config import Config from vyos.configtree import DiffTree from vyos.configdict import dict_merge from vyos.configdict import list_diff 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.xml_ref import get_defaults 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) 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