Update the method to get os version

inbm-lib/inbm_common_lib/constants.py

@@ -47,4 +47,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/utility.py

@@ -15,7 +15,7 @@
 from pathlib import Path
 from typing import Iterable, Optional, Union
 
-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,20 @@ 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
+ """
+ if os.path.exists(OS_RELEASE_PATH):
+ with open(OS_RELEASE_PATH, 'r') as version_file:
+ for line in version_file:
+ if line.startswith('VERSION='):
+ version = line.split('=')[1].replace('\n', ' ')
+ return version
+ logger.error(f"VERSION not found in {OS_RELEASE_PATH}.")
+ else:
+ logger.error(f"{OS_RELEASE_PATH} not exist.")
+ return UNKNOWN

inbm-lib/inbm_lib/detect_os.py

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

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

@@ -72,6 +72,10 @@ def get_os(self, os_type) -> "ISotaOs":
  elif os_type == OsType.Windows.name:
  logger.debug("Windows returned")
  return Windows(self._dispatcher_broker)
+ #TODO: Remove this when confirmed that TiberOS is in use
+ elif os_type == LinuxDistType.Mariner.name:
+ logger.debug("Mariner returned")
+ return TiberOSBasedSotaOs(self._dispatcher_broker)
  elif os_type == LinuxDistType.TiberOS.name:
  logger.debug("TiberOS returned")
  return TiberOSBasedSotaOs(self._dispatcher_broker)

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

@@ -12,6 +12,7 @@
 from inbm_lib.trtl import Trtl
 from typing import Any, Dict, Optional
 from inbm_common_lib.shell_runner import PseudoShellRunner
+from inbm_common_lib.utility import get_os_version
 from .constants import MENDER_FILE_PATH
 from .mender_util import read_current_mender_version
 from .update_tool_util import update_tool_version_command, update_tool_commit_command
@@ -466,7 +467,7 @@ def take_snapshot(self) -> None:
  "SOTA attempting to create a dispatcher state file before SOTA {}...".
  format(self.sota_cmd))
  try:
- content = update_tool_version_command()
+ content = get_os_version()
  state: dispatcher_state.DispatcherState
  if dispatcher_state.is_dispatcher_state_file_exists():
  consumed_state = dispatcher_state.consume_dispatcher_state_file(read=True)
@@ -535,13 +536,13 @@ def update_system(self) -> None:
  state = dispatcher_state.consume_dispatcher_state_file()
  if state is not None and 'tiberos-version' in state:
  logger.debug("got tiberos-version from state: " + str(state['tiberos-version']))
- version = update_tool_version_command()
+ version = get_os_version()
  current_tiberos_version = version
  previous_tiberos_version = state['tiberos-version']
 
  if current_tiberos_version == previous_tiberos_version:
  raise SotaError(
- f"Requested update version is the same as previous version installed. SHA: "
+ f"Requested update version is the same as previous version installed. VERSION: "
  f"{current_tiberos_version}")
  else:
  logger.debug("success; tiberos version changed")

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

@@ -11,7 +11,7 @@
 from typing import Any, List, Optional, Union, Mapping
 
 from inbm_common_lib.exceptions import UrlSecurityException
-from inbm_common_lib.utility import canonicalize_uri, remove_file
+from inbm_common_lib.utility import canonicalize_uri, remove_file, get_os_version
 from inbm_common_lib.request_message_constants import SOTA_FAILURE
 from inbm_common_lib.constants import REMOTE_SOURCE, LOCAL_SOURCE
 from inbm_lib.validate_package_list import parse_and_validate_package_list
@@ -386,7 +386,9 @@ def execute_from_manifest(self,
  # Since there is no installation, there will be no changes in the package status or version.
  # The apt history.log also doesn't record any changes. Therefore we can skip saving granular log.
  # Always save the granular log in TiberOS.
- elif self.sota_mode != 'download-only' or detect_os() == LinuxDistType.TiberOS.name:
+ # TODO: Remove Mariner when confirmed that TiberOS is in use
+ elif self.sota_mode != 'download-only' or detect_os() == LinuxDistType.TiberOS.name \
+ or detect_os() == LinuxDistType.Mariner.name:
  self.save_granular_log()
  if (self.sota_mode == 'download-only') or (not self._is_reboot_device()):
  self._dispatcher_broker.telemetry("No reboot (SOTA pass)")
@@ -400,7 +402,9 @@ def execute_from_manifest(self,
  self._update_logger.status = FAIL
  self._update_logger.save_log()
  # Always save the granular log in TiberOS.
- if self.sota_mode != 'download-only' or detect_os() == LinuxDistType.TiberOS.name:
+ # TODO: Remove Mariner when confirmed that TiberOS is in use
+ if self.sota_mode != 'download-only' or detect_os() == LinuxDistType.TiberOS.name \
+ or detect_os() == LinuxDistType.Mariner.name:
  self.save_granular_log()
  self._dispatcher_broker.telemetry(SOTA_FAILURE)
  self._dispatcher_broker.send_result(SOTA_FAILURE)
@@ -429,7 +433,8 @@ def save_granular_log(self, check_package: bool = True) -> None:
  @param check_package: True if you want to check the package's status and version and record them in Ubuntu.
  """
  log = {}
- if detect_os() == LinuxDistType.TiberOS.name:
+ # TODO: Remove Mariner when confirmed that TiberOS is in use
+ if detect_os() == LinuxDistType.TiberOS.name or detect_os() == LinuxDistType.Mariner.name:
  # Delete the previous log if exist.
  if os.path.exists(GRANULAR_LOG_FILE):
  remove_file(GRANULAR_LOG_FILE)
@@ -442,7 +447,7 @@ def save_granular_log(self, check_package: bool = True) -> None:
  elif self._update_logger.detail_status == OTA_SUCCESS or self._update_logger.detail_status == OTA_PENDING:
  log = {
  "StatusDetail.Status": self._update_logger.detail_status,
- "SHA": update_tool_version_command()
+ "OS Version": get_os_version()
  }
  # In TiberOS, no package level information needed.
  self._update_logger.save_granular_log_file(log=log, check_package=False)

inbm/dispatcher-agent/fpm-template/etc/apparmor.d/usr.bin.inbm-dispatcher

@@ -56,6 +56,7 @@
  /etc/intel-manageability/secret/dispatcher-agent/** r,
  /etc/opt/csl/csl-node/csl-manager r,
  /etc/opt/csl/csl-node/long-lived-token r,
+ /etc/os-release r,
  /etc/apt/sources.list rw,
  /etc/apt/sources.list.bak rw,
  /etc/apt/sources.list.d/* rw,