path: root/python/vyos/configdiff.py

# 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