5 files changed, 556 insertions, 0 deletions
diff --git a/python/vyos/xml_ref/__init__.py b/python/vyos/xml_ref/__init__.py
new file mode 100644
index 000000000..bf434865d
--- /dev/null
+++ b/python/vyos/xml_ref/__init__.py
@@ -0,0 +1,83 @@
+# 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/>.
+
+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 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
new file mode 100644
index 000000000..c90c5ddbc
--- /dev/null
+++ b/python/vyos/xml_ref/definition.py
@@ -0,0 +1,302 @@
+# 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/>.
+
+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)
+
+    @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/python/vyos/xml_ref/generate_cache.py b/python/vyos/xml_ref/generate_cache.py
new file mode 100755
index 000000000..6a05d4608
--- /dev/null
+++ b/python/vyos/xml_ref/generate_cache.py
@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2023 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 sys
+import json
+from argparse import ArgumentParser
+from argparse import ArgumentTypeError
+from os import getcwd
+from os import makedirs
+from os.path import join
+from os.path import abspath
+from os.path import dirname
+from os.path import basename
+from xmltodict import parse
+
+_here = dirname(__file__)
+
+sys.path.append(join(_here, '..'))
+from configtree import reference_tree_to_json, ConfigTreeError
+
+xml_cache_json = 'xml_cache.json'
+xml_tmp = join('/tmp', xml_cache_json)
+pkg_cache = abspath(join(_here, 'pkg_cache'))
+ref_cache = abspath(join(_here, 'cache.py'))
+
+node_data_fields = ("node_type", "multi", "valueless", "default_value")
+
+def trim_node_data(cache: dict):
+    for k in list(cache):
+        if k == "node_data":
+            for l in list(cache[k]):
+                if l not in node_data_fields:
+                    del cache[k][l]
+        else:
+            if isinstance(cache[k], dict):
+                trim_node_data(cache[k])
+
+def non_trivial(s):
+    if not s:
+        raise ArgumentTypeError("Argument must be non empty string")
+    return s
+
+def main():
+    parser = ArgumentParser(description='generate and save dict from xml defintions')
+    parser.add_argument('--xml-dir', type=str, required=True,
+                        help='transcluded xml interface-definition directory')
+    parser.add_argument('--package-name', type=non_trivial, default='vyos-1x',
+                        help='name of current package')
+    parser.add_argument('--output-path', help='path to generated cache')
+    args = vars(parser.parse_args())
+
+    xml_dir = abspath(args['xml_dir'])
+    pkg_name = args['package_name'].replace('-','_')
+    cache_name = pkg_name + '_cache.py'
+    out_path = args['output_path']
+    path = out_path if out_path is not None else pkg_cache
+    xml_cache = abspath(join(path, cache_name))
+
+    try:
+        reference_tree_to_json(xml_dir, xml_tmp)
+    except ConfigTreeError as e:
+        print(e)
+        sys.exit(1)
+
+    with open(xml_tmp) as f:
+        d = json.loads(f.read())
+
+    trim_node_data(d)
+
+    syntax_version = join(xml_dir, 'xml-component-version.xml')
+    try:
+        with open(syntax_version) as f:
+            component = f.read()
+    except FileNotFoundError:
+        if pkg_name != 'vyos_1x':
+            component = ''
+        else:
+            print("\nWARNING: missing xml-component-version.xml\n")
+            sys.exit(1)
+
+    if component:
+        parsed = parse(component)
+    else:
+        parsed = None
+    version = {}
+    # addon package definitions may have empty (== 0) version info
+    if parsed is not None and parsed['interfaceDefinition'] is not None:
+        converted = parsed['interfaceDefinition']['syntaxVersion']
+        if not isinstance(converted, list):
+            converted = [converted]
+        for i in converted:
+            tmp = {i['@component']: i['@version']}
+            version |= tmp
+
+    version = {"component_version": version}
+
+    d |= version
+
+    with open(xml_cache, 'w') as f:
+        f.write(f'reference = {str(d)}')
+
+    print(cache_name)
+
+if __name__ == '__main__':
+    main()
diff --git a/python/vyos/xml_ref/pkg_cache/__init__.py b/python/vyos/xml_ref/pkg_cache/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/python/vyos/xml_ref/pkg_cache/__init__.py
diff --git a/python/vyos/xml_ref/update_cache.py b/python/vyos/xml_ref/update_cache.py
new file mode 100755
index 000000000..0842bcbe9
--- /dev/null
+++ b/python/vyos/xml_ref/update_cache.py
@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2023 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
+from copy import deepcopy
+from generate_cache import pkg_cache
+from generate_cache import ref_cache
+
+def dict_merge(source, destination):
+    dest = deepcopy(destination)
+
+    for key, value in source.items():
+        if key not in dest:
+            dest[key] = value
+        elif isinstance(source[key], dict):
+            dest[key] = dict_merge(source[key], dest[key])
+
+    return dest
+
+def main():
+    res = {}
+    cache_dir = os.path.basename(pkg_cache)
+    for mod in os.listdir(pkg_cache):
+        mod = os.path.splitext(mod)[0]
+        if not mod.endswith('_cache'):
+            continue
+        d = getattr(__import__(f'{cache_dir}.{mod}', fromlist=[mod]), 'reference')
+        if mod == 'vyos_1x_cache':
+            res = dict_merge(res, d)
+        else:
+            res = dict_merge(d, res)
+
+    with open(ref_cache, 'w') as f:
+        f.write(f'reference = {str(res)}')
+
+if __name__ == '__main__':
+    main()