[NEXMANAGE-515][NEXMANAGE-598] Support TiberOS (#543)

This PR updates INBM to support TiberOS SOTA.

Signed-off-by: yengliong <[email protected]>
Co-authored-by: Gavin Lewis <[email protected]>

inbc-program/inbc/parser/ota_parser.py

@@ -46,6 +46,7 @@ def sota(args: argparse.Namespace) -> str:
         'mode': args.mode,
         'release_date': release_date,
         'fetch': fetch_location,
+        'signature': args.signature,
         'username': args.username,
         'password': _get_password(args.username, "Please provide the password: "),
         'deviceReboot': "no" if args.mode == "download-only" else args.reboot,
@@ -71,6 +72,7 @@ def sota(args: argparse.Namespace) -> str:
                         "mode",
                         "package_list",
                         "fetch",
+                        "signature",
                         "username",
                         "password",
                         "release_date",

inbc-program/inbc/parser/parser.py

@@ -215,6 +215,8 @@ def parse_sota_args(self) -> None:
                                  type=lambda x: validate_string_less_than_n_characters(
                                      x, 'URL', 1000),
                                  help='Remote URI from where to retrieve package')
+        parser_sota.add_argument('--signature', '-s', default='',
+                                 required=False, help='Signature file')
         parser_sota.add_argument('--releasedate', '-r', default='2026-12-31', required=False, type=validate_date,
                                  help='Release date of the applying package - format YYYY-MM-DD')
         parser_sota.add_argument('--username', '-un', required=False, help='Username on the remote server',

inbc-program/tests/unit/test_ota_parser.py

@@ -203,7 +203,7 @@ def test_create_ubuntu_update_manifest(self) -> None:
         s = self.arg_parser.parse_args(['sota'])
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>ota</type><ota><header><type>sota</type' \
                    '><repo>remote</repo></header><type><sota><cmd logtofile="y">update</cmd><mode>full</mode>' \
-                   '<package_list></package_list><deviceReboot>yes</deviceReboot></sota></type>' \
+                   '<package_list></package_list><signature></signature><deviceReboot>yes</deviceReboot></sota></type>' \
                    '</ota></manifest>'
         self.assertEqual(s.func(s), expected)
 
@@ -212,7 +212,7 @@ def test_create_ubuntu_update_manifest_with_package_list(self) -> None:
             ['sota', '--package-list', 'hello,cowsay', '--reboot', 'no', '--mode', 'download-only'])
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>ota</type><ota><header><type>sota</type' \
                    '><repo>remote</repo></header><type><sota><cmd logtofile="y">update</cmd><mode>download-only</mode>' \
-                   '<package_list>hello,cowsay</package_list><deviceReboot>no</deviceReboot></sota></type>' \
+                   '<package_list>hello,cowsay</package_list><signature></signature><deviceReboot>no</deviceReboot></sota></type>' \
                    '</ota></manifest>'
         self.assertEqual(s.func(s), expected)
 
@@ -235,7 +235,21 @@ def test_create_sota_manifest(self, mock_pass, mock_reconnect) -> None:
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>ota</type><ota><header><type>sota</type' \
                    '><repo>remote</repo></header><type><sota><cmd ' \
                    'logtofile="y">update</cmd><mode>full</mode><package_list></package_list>' \
-                   '<fetch>https://abc.com/test.tar</fetch><username>Frank</username><password>123abc</password>' \
+                   '<fetch>https://abc.com/test.tar</fetch><signature></signature><username>Frank</username><password>123abc</password>' \
+                   '<release_date>2026-12-31</release_date><deviceReboot>yes</deviceReboot>' \
+                   '</sota></type></ota></manifest>'
+        self.assertEqual(s.func(s), expected)
+
+    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.reconnect')
+    @patch('inbc.utility.getpass.getpass', return_value='123abc')
+    def test_create_sota_manifest_with_signature(self, mock_pass, mock_reconnect) -> None:
+        s = self.arg_parser.parse_args(
+            ['sota', '-u', 'https://abc.com/test.tar', '-un', 'Frank', '-s', '64a37255b4eb18ae858768c0c06d2875124e3111081cdade737196ec7502c53e'])
+        expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>ota</type><ota><header><type>sota</type' \
+                   '><repo>remote</repo></header><type><sota><cmd ' \
+                   'logtofile="y">update</cmd><mode>full</mode><package_list></package_list>' \
+                   '<fetch>https://abc.com/test.tar</fetch><signature>64a37255b4eb18ae858768c0c06d2875124e3111081cdade737196ec7502c53e</signature>' \
+                   '<username>Frank</username><password>123abc</password>' \
                    '<release_date>2026-12-31</release_date><deviceReboot>yes</deviceReboot>' \
                    '</sota></type></ota></manifest>'
         self.assertEqual(s.func(s), expected)
@@ -248,7 +262,8 @@ def test_create_sota_mode_manifest(self, mock_pass, mock_reconnect) -> None:
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>ota</type><ota><header><type>sota</type' \
                    '><repo>remote</repo></header><type><sota><cmd ' \
                    'logtofile="y">update</cmd><mode>full</mode><package_list></package_list>' \
-                   '<fetch>https://abc.com/test.tar</fetch><username>Frank</username><password>123abc</password>' \
+                   '<fetch>https://abc.com/test.tar</fetch><signature></signature>' \
+                   '<username>Frank</username><password>123abc</password>' \
                    '<release_date>2026-12-31</release_date><deviceReboot>yes</deviceReboot></sota>' \
                    '</type></ota></manifest>'
         self.assertEqual(s.func(s), expected)

inbm-lib/inbm_common_lib/constants.py

@@ -49,4 +49,7 @@
 AFULNX_64 = 'afulnx_64'
 
 # Default signature version
-DEFAULT_HASH_ALGORITHM = 384
+DEFAULT_HASH_ALGORITHM = 384
+
+# Os release path
+OS_RELEASE_PATH = '/etc/os-release'

inbm-lib/inbm_common_lib/shell_runner.py

@@ -80,7 +80,8 @@ def _sanitize(self, filename: str) -> str:
     def run_with_log_path(self,
                           cmd: str,
                           log_path: Optional[str],
-                          cwd: Optional[str] = None) -> Tuple[str, Optional[str], int, Optional[str]]:
+                          cwd: Optional[str] = None,
+                          stdin: Optional[str] = None,) -> Tuple[str, Optional[str], int, Optional[str]]:
         """Run/Invoke system commands
 
         NOTE: on Windows, stderr will appear in stdout instead, alongside stdout,
@@ -89,6 +90,7 @@ def run_with_log_path(self,
         @param cmd: Shell cmd to execute
         @param log_path: string format of log file's absolute path
         @param cwd: if not None, run process from this working directory
+        @param stdin: if stdin provided, it will be passed as stdin input
         @return: Result of subprocess along with output, error (possibly None), exit status, and absolute log path
         """
         shlex_split_cmd = PseudoShellRunner().interpret_shell_like_command(cmd)
@@ -126,7 +128,12 @@ def run_with_log_path(self,
             abs_log_path = None
 
         logger.debug("")
-        (out, err) = proc.communicate(b'yes\n') if AFULNX_64 in cmd else proc.communicate()
+        if AFULNX_64 in cmd:
+            (out, err) = proc.communicate(b'yes\n')
+        elif stdin:
+            (out, err) = proc.communicate(input=stdin.encode())
+        else:
+            (out, err) = proc.communicate()
 
         # we filter out bad characters but still accept the rest of the string
         # here based on experience running the underlying command
@@ -139,17 +146,19 @@ def run_with_log_path(self,
 
         return decoded_out, decoded_err, proc.returncode, abs_log_path
 
-    def run(self, cmd: str, cwd: Optional[str] = None) -> Tuple[str, Optional[str], int]:
+    def run(self, cmd: str, cwd: Optional[str] = None,
+            stdin: Optional[str] = None) -> Tuple[str, Optional[str], int]:
         """Run/Invoke system commands
 
         NOTE: on Windows, stderr will appear in stdout instead, alongside stdout,
         due to limitations with Windows services
 
         @param cmd: Shell cmd to execute
         @param cwd: if not None, run process from this working directory
+        @param stdin: if stdin provided, it will be passed as stdin input
         @return: Result of subprocess along with output, error (possibly None) & exit status
         """
-        (out, err, code, _) = PseudoShellRunner().run_with_log_path(cmd, log_path=None, cwd=cwd)
+        (out, err, code, _) = PseudoShellRunner().run_with_log_path(cmd, log_path=None, cwd=cwd, stdin=stdin)
         return out, err, code
 
     def interpret_shell_like_command(self, cmd: str) -> List[str]:
@@ -189,4 +198,4 @@ def is_exe(fpath: str) -> bool:
         which_cmd = which(shlex_split_cmd[0])
         if which_cmd:
             shlex_split_cmd[0] = which_cmd
-        return shlex_split_cmd
+        return shlex_split_cmd

inbm-lib/inbm_common_lib/utility.py

@@ -10,12 +10,12 @@
 import shutil
 import tarfile
 import logging
-
+import ast
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Iterable, Optional, Union
+from typing import Iterable, Optional, Union, Dict, Tuple
 
-from inbm_common_lib.constants import VALID_MAGIC_FILE_TYPE_PREFIXES, TEMP_EXT_FOLDER
+from inbm_common_lib.constants import VALID_MAGIC_FILE_TYPE_PREFIXES, TEMP_EXT_FOLDER, OS_RELEASE_PATH, UNKNOWN
 from inbm_common_lib.shell_runner import PseudoShellRunner
 
 from .constants import URL_NULL_CHAR
@@ -271,3 +271,64 @@ def validate_file_type(path: list[str]) -> None:
     if os.path.exists(TEMP_EXT_FOLDER):
         shutil.rmtree(TEMP_EXT_FOLDER, ignore_errors=True)
     remove_file_list(extracted_file_list)
+
+
+def get_os_version() -> str:
+    """Get os version from os release file.
+
+    @return value of the VERSION
+    """
+    try:
+        if os.path.exists(OS_RELEASE_PATH):
+            with open(OS_RELEASE_PATH, 'r') as version_file:
+                content = version_file.read()
+
+            content_dict = parse_os_release(content)
+            version = content_dict.get("VERSION")
+            if version:
+                return version
+            logger.error(f"VERSION not found in {OS_RELEASE_PATH}.")
+        else:
+            logger.error(f"{OS_RELEASE_PATH} not exist.")
+
+        return UNKNOWN
+    except OSError as err:
+        raise OSError(f"Error while reading the os version: {err}")
+
+
+def parse_os_release(file_content: str) -> Dict[str, str]:
+    """
+    Parses the content of an os-release file and returns a dictionary of key-value pairs.
+
+    :param file_content: The content of the os-release file as a string.
+    :return: A dictionary containing key-value pairs from the file.
+    """
+    result: Dict[str, str] = {}
+    for line in file_content.splitlines():
+        parsed = parse_line(line)
+        if parsed:
+            key, value = parsed
+            result[key] = value
+    return result
+
+
+def parse_line(line: str) -> Optional[Tuple[str, str]]:
+    """
+    Parses a line of the os-release file and returns a key-value pair.
+    Returns None if the line is empty or a comment.
+
+    :param line: A line from the os-release file.
+    :return: A tuple of (key, value) if the line contains a key-value pair, else None.
+    """
+    line = line.strip()
+    if not line or line.startswith('#'):
+        return None
+    if '=' not in line:
+        return None
+    key, value = line.split('=', 1)
+    key = key.strip()
+    value = value.strip()
+    if (value.startswith('"') and value.endswith('"')) or \
+       (value.startswith("'") and value.endswith("'")):
+           value = value[1:-1]
+    return key, value

inbm-lib/inbm_lib/constants.py

@@ -88,5 +88,6 @@
 FAIL = "FAIL"
 OTA_PENDING = "PENDING"
 OTA_NO_UPDATE = "NO_UPDATE_AVAILABLE"
+ROLLBACK = "ROLLBACK"
 
 FORMAT_VERSION = "v1"

inbm-lib/inbm_lib/detect_os.py

@@ -34,6 +34,8 @@ class LinuxDistType(Enum):
     Deby = 3
     Debian = 4
     CentOS = 5
+    Mariner = 6  # TODO: Remove this when confirmed that TiberOS is in use
+    TiberOS = 7
 
 
 def verify_os_supported() -> str:

inbm-lib/tests/unit/inbm_common_lib/test_utility.py

@@ -1,10 +1,12 @@
 import shutil
+import tempfile
 from unittest.mock import patch, Mock, mock_open
 from unittest import TestCase
 
 from inbm_common_lib.exceptions import UrlSecurityException
 from inbm_common_lib.utility import clean_input, get_canonical_representation_of_path, canonicalize_uri, \
-    validate_file_type, remove_file, copy_file, move_file, create_file_with_contents
+    validate_file_type, remove_file, copy_file, move_file, create_file_with_contents, get_os_version
+from inbm_common_lib.constants import UNKNOWN
 
 
 class TestUtility(TestCase):
@@ -109,3 +111,19 @@ def test_create_file_with_contents_successfully(self) -> None:
 
         except IOError as e:
             self.fail(f"Unexpected exception raised during test: {e}")
+
+    @patch('builtins.open', new_callable=mock_open, read_data='VERSION="2.0.20240802.0213"')
+    def test_get_os_version_successfully(self, mock_open: Mock) -> None:
+        try:
+            self.assertEqual(get_os_version(), "2.0.20240802.0213")
+        except IOError as e:
+            self.fail(f"Unexpected exception raised during test: {e}")
+        mock_open.assert_called_once_with('/etc/os-release', 'r')
+
+    @patch('builtins.open', new_callable=mock_open, read_data='')
+    def test_get_os_version_with_no_version_found(self, mock_open: Mock) -> None:
+        try:
+            self.assertEqual(get_os_version(), UNKNOWN)
+        except IOError as e:
+            self.fail(f"Unexpected exception raised during test: {e}")
+        mock_open.assert_called_once_with('/etc/os-release', 'r')

inbm-lib/tests/unit/inbm_lib/test_runner.py

@@ -12,7 +12,7 @@ def __init__(self, output: str, err: str, return_code: int) -> None:
         self.__return_code = return_code
         self.__err = err
 
-    def run(self, cmd: str, cwd: str | None = None) -> Tuple[str, str | None, int]:
+    def run(self, cmd: str, cwd: str | None = None, stdin: str | None = None) -> Tuple[str, str | None, int]:
         self.__last_commands.append(cmd)
         return self.__output, self.__err, self.__return_code
 

inbm/Changelog.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 ## NEXT - YYYY-MM-DD
 ### Added
+ - (NEXMANAGE-515) Update dispatcher SOTA related classes for supporting TiberOS
+ - (NEXMANAGE-598) Expanding INBC for handling TiberOS update cmd
  - Updated proto files to add new RPC calls to allow edge node to update
    its status with INBS.
  - (NEXMANAGE- 610) Add functionality to INBM Cloudadapter-agent to support OOB AMT RPC command requests from INBS

inbm/dispatcher-agent/dispatcher/common/dispatcher_state.py

@@ -60,7 +60,8 @@ def is_dispatcher_state_file_exists() -> bool:
     'snapshot_num': str,
     'bios_version': str,
     'release_date': datetime,
-    'mender-version': str
+    'mender-version': str,
+    'tiberos-version': str
 }, total=False)
 
 

inbm/dispatcher-agent/dispatcher/sota/constants.py

@@ -14,6 +14,13 @@
 # Mender artifact path
 MENDER_ARTIFACT_PATH = get_canonical_representation_of_path("/etc/mender/artifact_info")
 
+# Tiber Update Tool file path
+TIBER_UPDATE_TOOL_PATH = get_canonical_representation_of_path('/usr/bin/os-update-tool.sh')
+
+# ORAS release server access token path
+ORAS_TOKEN_PATH = get_canonical_representation_of_path('/etc/intel_edge_node/tokens/platform-update-agent/'
+                                                       'rs_access_token')
+
 SOTA_STATE = 'normal'
 
 LOGPATH = '/var/lib/dispatcher/upload'

inbm/dispatcher-agent/dispatcher/sota/downloader.py

@@ -12,6 +12,7 @@
 from typing import Optional
 from .mender_util import read_current_mender_version
 from .sota_error import SotaError
+from .oras_util import oras_download, read_oras_token
 from ..constants import UMASK_OTA
 from ..downloader import download
 from ..packagemanager.irepo import IRepo
@@ -165,3 +166,43 @@ def download(self,
 
     def check_release_date(self, release_date: Optional[str]) -> bool:
         return self.is_valid_release_date(release_date)
+
+
+class TiberOSDownloader(Downloader):
+    """TiberOSDownloader class, child of Downloader"""
+
+    def __init__(self) -> None:
+        super().__init__()
+
+    def download(self,
+                 dispatcher_broker: DispatcherBroker,
+                 uri: Optional[CanonicalUri],
+                 repo: IRepo,
+                 username: Optional[str],
+                 password: Optional[str],
+                 release_date: Optional[str]) -> None:
+        """Downloads files and places image in local cache
+
+        @param dispatcher_broker: DispatcherBroker object used to communicate with other INBM services
+        @param uri: URI of the source location
+        @param repo: repository for holding the download
+        @param username: username to use for download
+        @param password: password to use for download
+        @param release_date: manifest release date
+        @raises SotaError: release date is not valid
+        """
+
+        if uri is None:
+            raise SotaError("URI is None while performing TiberOS download")
+
+        password = read_oras_token()
+
+        oras_download(dispatcher_broker=dispatcher_broker,
+                      uri=uri,
+                      repo=repo,
+                      umask=UMASK_OTA,
+                      username=username,
+                      password=password)
+
+    def check_release_date(self, release_date: Optional[str]) -> bool:
+        raise NotImplementedError()