image: T4516: support for interoperability of legacy/new image tools

This commit allows management of system images with either new or legacy
tools: 'add/delete/rename system image' and 'set default' are translated
appropriately on booting between images with the old and new tools.
Consequently, the warning of the initial commit of T4516 is dropped.

(cherry picked from commit 96b65e90fbfa1fe63d97929ac86fc910abb0caa9)

1 files changed, 307 insertions, 0 deletions

diff --git a/python/vyos/system/compat.py b/python/vyos/system/compat.py
new file mode 100644
index 000000000..aa9b0b4b5
--- /dev/null
+++ b/python/vyos/system/compat.py
@@ -0,0 +1,307 @@
+# 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 pathlib import Path
+from re import compile, MULTILINE, DOTALL
+from functools import wraps
+from copy import deepcopy
+from typing import Union
+
+from vyos.system import disk, grub, image, SYSTEM_CFG_VER
+from vyos.template import render
+
+TMPL_GRUB_COMPAT: str = 'grub/grub_compat.j2'
+
+# define regexes and variables
+REGEX_VERSION = r'^menuentry "[^\n]*{\n[^}]*\s+linux /boot/(?P<version>\S+)/[^}]*}'
+REGEX_MENUENTRY = r'^menuentry "[^\n]*{\n[^}]*\s+linux /boot/(?P<version>\S+)/vmlinuz (?P<options>[^\n]+)\n[^}]*}'
+REGEX_CONSOLE = r'^.*console=(?P<console_type>[^\s\d]+)(?P<console_num>[\d]+).*$'
+REGEX_SANIT_CONSOLE = r'\ ?console=[^\s\d]+[\d]+(,\d+)?\ ?'
+REGEX_SANIT_INIT = r'\ ?init=\S*\ ?'
+REGEX_SANIT_QUIET = r'\ ?quiet\ ?'
+PW_RESET_OPTION = 'init=/opt/vyatta/sbin/standalone_root_pw_reset'
+
+
+class DowngradingImageTools(Exception):
+    """Raised when attempting to add an image with an earlier version
+    of image-tools than the current system, as indicated by the value
+    of SYSTEM_CFG_VER or absence thereof."""
+    pass
+
+
+def mode():
+    if grub.get_cfg_ver() >= SYSTEM_CFG_VER:
+        return False
+
+    return True
+
+
+def find_versions(menu_entries: list) -> list:
+    """Find unique VyOS versions from menu entries
+
+    Args:
+        menu_entries (list): a list with menu entries
+
+    Returns:
+        list: List of installed versions
+    """
+    versions = []
+    for vyos_ver in menu_entries:
+        versions.append(vyos_ver.get('version'))
+    # remove duplicates
+    versions = list(set(versions))
+    return versions
+
+
+def filter_unparsed(grub_path: str) -> str:
+    """Find currently installed VyOS version
+
+    Args:
+        grub_path (str): a path to the grub.cfg file
+
+    Returns:
+        str: unparsed grub.cfg items
+    """
+    config_text = Path(grub_path).read_text()
+    regex_filter = compile(REGEX_VERSION, MULTILINE | DOTALL)
+    filtered = regex_filter.sub('', config_text)
+    regex_filter = compile(grub.REGEX_GRUB_VARS, MULTILINE)
+    filtered = regex_filter.sub('', filtered)
+    regex_filter = compile(grub.REGEX_GRUB_MODULES, MULTILINE)
+    filtered = regex_filter.sub('', filtered)
+    # strip extra new lines
+    filtered = filtered.strip()
+    return filtered
+
+
+def sanitize_boot_opts(boot_opts: str) -> str:
+    """Sanitize boot options from console and init
+
+    Args:
+        boot_opts (str): boot options
+
+    Returns:
+        str: sanitized boot options
+    """
+    regex_filter = compile(REGEX_SANIT_CONSOLE)
+    boot_opts = regex_filter.sub('', boot_opts)
+    regex_filter = compile(REGEX_SANIT_INIT)
+    boot_opts = regex_filter.sub('', boot_opts)
+    # legacy tools add 'quiet' on add system image; this is not desired
+    regex_filter = compile(REGEX_SANIT_QUIET)
+    boot_opts = regex_filter.sub(' ', boot_opts)
+
+    return boot_opts
+
+
+def parse_entry(entry: tuple) -> dict:
+    """Parse GRUB menuentry
+
+    Args:
+        entry (tuple): tuple of (version, options)
+
+    Returns:
+        dict: dictionary with parsed options
+    """
+    # save version to dict
+    entry_dict = {'version': entry[0]}
+    # detect boot mode type
+    if PW_RESET_OPTION in entry[1]:
+        entry_dict['bootmode'] = 'pw_reset'
+    else:
+        entry_dict['bootmode'] = 'normal'
+    # find console type and number
+    regex_filter = compile(REGEX_CONSOLE)
+    entry_dict.update(regex_filter.match(entry[1]).groupdict())
+    entry_dict['boot_opts'] = sanitize_boot_opts(entry[1])
+
+    return entry_dict
+
+
+def parse_menuentries(grub_path: str) -> list:
+    """Parse all GRUB menuentries
+
+    Args:
+        grub_path (str): a path to GRUB config file
+
+    Returns:
+        list: list with menu items (each item is a dict)
+    """
+    menuentries = []
+    # read configuration file
+    config_text = Path(grub_path).read_text()
+    # parse menuentries to tuples (version, options)
+    regex_filter = compile(REGEX_MENUENTRY, MULTILINE)
+    filter_results = regex_filter.findall(config_text)
+    # parse each entry
+    for entry in filter_results:
+        menuentries.append(parse_entry(entry))
+
+    return menuentries
+
+
+def prune_vyos_versions(root_dir: str = '') -> None:
+    """Delete vyos-versions files of registered images subsequently deleted
+    or renamed by legacy image-tools
+
+    Args:
+        root_dir (str): an optional path to the root directory
+    """
+    if not root_dir:
+        root_dir = disk.find_persistence()
+
+    for version in grub.version_list():
+        if not Path(f'{root_dir}/boot/{version}').is_dir():
+            grub.version_del(version)
+
+
+def update_cfg_ver(root_dir:str = '') -> int:
+    """Get minumum version of image-tools across all installed images
+
+    Args:
+        root_dir (str): an optional path to the root directory
+
+    Returns:
+        int: minimum version of image-tools
+    """
+    if not root_dir:
+        root_dir = disk.find_persistence()
+
+    prune_vyos_versions(root_dir)
+
+    images_details = image.get_images_details()
+    cfg_version = min(d['tools_version'] for d in images_details)
+
+    return cfg_version
+
+
+def get_default(menu_entries: list, root_dir: str = '') -> Union[int, None]:
+    """Translate default version to menuentry index
+
+    Args:
+        menu_entries (list): list of dicts of installed version boot data
+        root_dir (str): an optional path to the root directory
+
+    Returns:
+        int: index of default version in menu_entries or None
+    """
+    if not root_dir:
+        root_dir = disk.find_persistence()
+
+    grub_cfg_main = f'{root_dir}/{grub.GRUB_CFG_MAIN}'
+
+    image_name = image.get_default_image()
+
+    sublist = list(filter(lambda x: x.get('version') == image_name,
+                        menu_entries))
+    if sublist:
+        return menu_entries.index(sublist[0])
+
+    return None
+
+
+def update_version_list(root_dir: str = '') -> list[dict]:
+    """Update list of dicts of installed version boot data
+
+    Args:
+        root_dir (str): an optional path to the root directory
+
+    Returns:
+        list: list of dicts of installed version boot data
+    """
+    if not root_dir:
+        root_dir = disk.find_persistence()
+
+    grub_cfg_main = f'{root_dir}/{grub.GRUB_CFG_MAIN}'
+
+    # get list of versions in menuentries
+    menu_entries = parse_menuentries(grub_cfg_main)
+    menu_versions = find_versions(menu_entries)
+
+    # get list of versions added/removed by image-tools
+    current_versions = grub.version_list(root_dir)
+
+    remove = list(set(menu_versions) - set(current_versions))
+    for ver in remove:
+        menu_entries = list(filter(lambda x: x.get('version') != ver,
+                                   menu_entries))
+
+    add = list(set(current_versions) - set(menu_versions))
+    for ver in add:
+        last = menu_entries[0].get('version')
+        new = deepcopy(list(filter(lambda x: x.get('version') == last,
+                                   menu_entries)))
+        for e in new:
+            boot_opts = e.get('boot_opts').replace(last, ver)
+            e.update({'version': ver, 'boot_opts': boot_opts})
+
+        menu_entries = new + menu_entries
+
+    return menu_entries
+
+
+def grub_cfg_fields(root_dir: str = '') -> dict:
+    """Gather fields for rendering grub.cfg
+
+    Args:
+        root_dir (str): an optional path to the root directory
+
+    Returns:
+        dict: dictionary for rendering TMPL_GRUB_COMPAT
+    """
+    if not root_dir:
+        root_dir = disk.find_persistence()
+
+    grub_cfg_main = f'{root_dir}/{grub.GRUB_CFG_MAIN}'
+
+    fields = {'default': 0, 'timeout': 5}
+    # 'default' and 'timeout' from legacy grub.cfg
+    fields |= grub.vars_read(grub_cfg_main)
+    fields['tools_version'] = SYSTEM_CFG_VER
+    menu_entries = update_version_list(root_dir)
+    fields['versions'] = menu_entries
+    default = get_default(menu_entries, root_dir)
+    if default is not None:
+        fields['default'] = default
+
+    p = Path('/sys/firmware/efi')
+    if p.is_dir():
+        fields['efi'] = True
+    else:
+        fields['efi'] = False
+
+    return fields
+
+
+def render_grub_cfg(root_dir: str = '') -> None:
+    """Render grub.cfg for legacy compatibility"""
+    if not root_dir:
+        root_dir = disk.find_persistence()
+
+    grub_cfg_main = f'{root_dir}/{grub.GRUB_CFG_MAIN}'
+
+    fields = grub_cfg_fields(root_dir)
+    render(grub_cfg_main, TMPL_GRUB_COMPAT, fields)
+
+
+def grub_cfg_update(func):
+    """Decorator to update grub.cfg after function call"""
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        ret = func(*args, **kwargs)
+        if mode():
+            render_grub_cfg()
+        return ret
+    return wrapper