Added verification of GPG key URIs against a list of trusted repositories for enhanced security

inbc-program/README.md

@@ -424,6 +424,13 @@ inbc query --option sw
 Optionally Downloads and encrypts GPG key and stores it on the system under <em>/usr/share/keyrings</em>. Creates a file under <em>/etc/apt/sources.list.d</em> to store the update source information.
 This list file is used during 'sudo apt update' to update the application. <em>Deb882</em> format may be used instead of downloading a GPG key.
 
+**NOTE:** Make sure to add gpgKeyUri to trustedrepositories using INBC Config Append command before using Inbc source application add command
+ Step 1: Refer to Inbc Config Append command to set gpgKeyUri to trustedRepositories in intel-manageability.conf file
+ Example: inbc append --path trustedRepositories:https://deb.opera.com/
+ Step 2: Use Inbc source appplication add command 
+```
+
+
 ### Usage
 ```
 inbc source application add
@@ -449,7 +456,6 @@ inbc source application add
 
  - Each blank line has a period in it. -> " ."
  - Each line after the Signed-By: starts with a space -> " gibberish"
-
 ```
 inbc source application add 
  --sources 

inbm/Changelog.md

@@ -10,7 +10,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 ### Added
  - RTC 536601 - Added 'source' command to INBM. This command manages `/etc/apt/sources.list` and `/etc/apt/sources.list.d/*` and associated gpg keys on Ubuntu.
+- RTC 537769 - Added verification of GPG key URIs against a list of trusted repositories for enhanced security
 
+check if sourceApplication Gpg key URL is in trusted repo
 ### Fixed
  - RTC 534426 - Could not write to /var/log/inbm-update-status.log on Yocto due to /var/log being a symlink to /var/volatile/log.
  - RTC 523677 - Improve INBC error logging - invalid child tag not printed

inbm/dispatcher-agent/dispatcher/dispatcher_class.py

@@ -16,6 +16,7 @@
 from threading import Thread, active_count
 from time import sleep
 from typing import Tuple
+from typing import Optional, Any
 
 from dispatcher.config.config_operation import ConfigOperation
 from dispatcher.source.source_command import do_source_command
@@ -293,7 +294,7 @@ def do_install(self, xml: str, schema_location: Optional[str] = None) -> Result:
  elif type_of_manifest == 'source':
  logger.debug('Running source command')
  # FIXME: actually detect OS
- result = do_source_command(parsed_head, source.constants.OsType.Ubuntu)
+ result = do_source_command(parsed_head, source.constants.OsType.Ubuntu, self._dispatcher_broker)
  elif type_of_manifest == 'ota':
  # Parse manifest
  header = parsed_head.get_children('ota/header')

inbm/dispatcher-agent/dispatcher/source/source_command.py

@@ -7,6 +7,8 @@
 import logging
 import json
 from dispatcher.common.result_constants import Result
+from typing import Optional, Any
+from dispatcher.dispatcher_broker import DispatcherBroker
 from dispatcher.source.constants import (
  ApplicationAddSourceParameters,
  ApplicationRemoveSourceParameters,
@@ -22,7 +24,7 @@
 logger = logging.getLogger(__name__)
 
 
-def do_source_command(parsed_head: XmlHandler, os_type: OsType) -> Result:
+def do_source_command(parsed_head: XmlHandler, os_type: OsType, dispatcher_broker: DispatcherBroker) -> Result:
  """
  Run a source command.
 
@@ -42,7 +44,7 @@ def do_source_command(parsed_head: XmlHandler, os_type: OsType) -> Result:
  try:
  app_action = parsed_head.get_children("applicationSource")
  if app_action:
- return _handle_app_source_command(parsed_head, os_type, app_action)
+ return _handle_app_source_command(parsed_head, os_type, app_action, dispatcher_broker)
  except XmlException as e:
  return Result(status=400, message=f"unable to handle source command XML: {e}")
 
@@ -94,16 +96,17 @@ def _handle_os_source_command(parsed_head: XmlHandler, os_type: OsType, os_actio
 
 
 def _handle_app_source_command(
- parsed_head: XmlHandler, os_type: OsType, app_action: dict) -> Result:
+ parsed_head: XmlHandler, os_type: OsType, app_action: dict, dispatcher_broker: DispatcherBroker) -> Result:
  """
  Handle the application source commands.
 
  @param parsed_head: XmlHandler with command information
  @param os_type: OS type
  @param app_action: The action to be performed
+ @param dispatcher_broker: MQTT
  @return Result
  """
- application_source_manager = create_application_source_manager(os_type)
+ application_source_manager = create_application_source_manager(os_type, dispatcher_broker)
 
  if "list" in app_action:
  serialized_list = json.dumps(

inbm/dispatcher-agent/dispatcher/source/source_manager_factory.py

@@ -5,9 +5,11 @@
  SPDX-License-Identifier: Apache-2.0
 """
 import logging
+from dispatcher.dispatcher_broker import DispatcherBroker
 
 from dispatcher.source.constants import OsType
 from dispatcher.source.source_manager import ApplicationSourceManager, OsSourceManager
+from typing import Optional, Any
 from dispatcher.source.ubuntu_source_manager import (
  UbuntuApplicationSourceManager,
  UbuntuOsSourceManager,
@@ -23,8 +25,8 @@ def create_os_source_manager(os_type: OsType) -> OsSourceManager:
  raise ValueError(f"Unsupported OS type: {os_type}.")
 
 
-def create_application_source_manager(os_type: OsType) -> ApplicationSourceManager:
+def create_application_source_manager(os_type: OsType, dispatcher_broker: DispatcherBroker) -> ApplicationSourceManager:
  """Return correct OS application manager based on OS type"""
  if os_type is OsType.Ubuntu:
- return UbuntuApplicationSourceManager()
+ return UbuntuApplicationSourceManager(dispatcher_broker)
  raise ValueError(f"Unsupported OS type: {os_type}.")

inbm/dispatcher-agent/dispatcher/source/ubuntu_source_manager.py

@@ -7,6 +7,9 @@
 import logging
 import os
 
+from dispatcher.packagemanager.package_manager import verify_source
+from dispatcher.dispatcher_broker import DispatcherBroker
+from dispatcher.dispatcher_exception import DispatcherException
 from dispatcher.source.source_exception import SourceError
 from dispatcher.source.constants import (
  UBUNTU_APT_SOURCES_LIST,
@@ -94,16 +97,24 @@ def update(self, parameters: SourceParameters) -> None:
 
 
 class UbuntuApplicationSourceManager(ApplicationSourceManager):
- def __init__(self) -> None:
- pass
+ def __init__(self, broker: DispatcherBroker) -> None:
+ self._dispatcher_broker = broker
 
  def add(self, parameters: ApplicationAddSourceParameters) -> None:
  """Adds a source file and optional GPG key to be used during Ubuntu application updates."""
- # Step 1: Add key (Optional)
+ # Step 1: Verify gpg key uri from trusted repo list 
  if parameters.gpg_key_name and parameters.gpg_key_uri:
+ try:
+ url = parameters.gpg_key_uri
+ #URL slicing to remove the last segment (filename) from the URL 
+ source = url[:-(len(url.split('/')[-1]) + 1)]
+ verify_source(source=source, dispatcher_broker=self._dispatcher_broker)
+ except (DispatcherException, IndexError) as err:
+ raise SourceError(f"Source Gpg key URI verification check failed: {err}")
+ # Step 2: Add key (Optional)
  add_gpg_key(parameters.gpg_key_uri, parameters.gpg_key_name)
 
- # Step 2: Add the source
+ # Step 3: Add the source
  try:
  create_file_with_contents(
  os.path.join(UBUNTU_APT_SOURCES_LIST_D, parameters.source_list_file_name), parameters.sources

inbm/dispatcher-agent/tests/unit/source/test_source_cmd_factory.py

@@ -1,5 +1,6 @@
 import pytest
 from dispatcher.source.constants import OsType
+from ..common.mock_resources import MockDispatcherBroker
 from dispatcher.source.source_manager_factory import (
  create_application_source_manager,
  create_os_source_manager,
@@ -20,13 +21,13 @@ def test_create_os_source_manager_unsupported():
  create_os_source_manager("UnsupportedOS")
  assert "Unsupported OS type" in str(excinfo.value)
 
-
 def test_create_application_source_manager_ubuntu():
- command = create_application_source_manager(OsType.Ubuntu)
+ mock_disp_broker_obj = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = create_application_source_manager(OsType.Ubuntu, mock_disp_broker_obj)
  assert isinstance(command, UbuntuApplicationSourceManager)
 
-
 def test_create_application_source_manager_unsupported():
+ mock_disp_broker_obj = MockDispatcherBroker.build_mock_dispatcher_broker()
  with pytest.raises(ValueError) as excinfo:
- create_application_source_manager("UnsupportedOS")
+ create_application_source_manager("UnsupportedOS", mock_disp_broker_obj)
  assert "Unsupported OS type" in str(excinfo.value)

inbm/dispatcher-agent/tests/unit/source/test_source_command.py

@@ -7,6 +7,7 @@
 
 import pytest
 from dispatcher.common.result_constants import Result
+from ..common.mock_resources import MockDispatcherBroker
 from dispatcher.source.constants import (
  ApplicationAddSourceParameters,
  ApplicationRemoveSourceParameters,
@@ -66,8 +67,8 @@ def test_do_source_command_list(
  mock_source_manager.list.return_value = return_value
 
  mocker.patch(patch_target, return_value=mock_source_manager)
-
- result = do_source_command(xml_handler, OsType.Ubuntu)
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ result = do_source_command(xml_handler, OsType.Ubuntu, broker)
 
  assert result == Result(status=200, message=expected_message)
  mock_source_manager.list.assert_called_once()
@@ -113,8 +114,8 @@ def test_do_source_command_remove(
  mock_manager.remove.return_value = None
 
  mocker.patch(manager_mock, return_value=mock_manager)
-
- result = do_source_command(xml_handler, os_type)
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ result = do_source_command(xml_handler, os_type, broker)
 
  mock_manager.remove.assert_called_once_with(expected_call)
  assert result == Result(status=200, message="SUCCESS")
@@ -180,8 +181,8 @@ def test_do_source_command_add(
  mock_manager.add.return_value = None
 
  mocker.patch(manager_mock, return_value=mock_manager)
-
- result = do_source_command(xml_handler, os_type)
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ result = do_source_command(xml_handler, os_type, broker)
 
  mock_manager.add.assert_called_once_with(expected_call)
  assert result == Result(status=200, message="SUCCESS")
@@ -240,8 +241,8 @@ def test_do_source_command_update(
  mock_manager.update.return_value = None
 
  mocker.patch(manager_mock, return_value=mock_manager)
-
- result = do_source_command(xml_handler, os_type)
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ result = do_source_command(xml_handler, os_type, broker)
 
  mock_manager.update.assert_called_once_with(expected_call)
  assert result == Result(status=200, message="SUCCESS")

inbm/dispatcher-agent/tests/unit/source/test_ubuntu_source_cmd.py

@@ -2,6 +2,8 @@
 import pytest
 from unittest.mock import mock_open, patch
 from dispatcher.source.source_exception import SourceError
+from ..common.mock_resources import MockDispatcherBroker
+from dispatcher.dispatcher_exception import DispatcherException
 from dispatcher.source.constants import (
  UBUNTU_APT_SOURCES_LIST_D,
  UBUNTU_APT_SOURCES_LIST,
@@ -193,22 +195,25 @@ def test_update_sources_os_error(self):
 
 
 class TestUbuntuApplicationSourceManager:
- def test_add_app_with_gpg_key_successfully(self):
+ @patch("dispatcher.source.ubuntu_source_manager.verify_source")
+ def test_add_app_with_gpg_key_successfully(self, mock_verify_source):
  try:
  params = ApplicationAddSourceParameters(
- source_list_file_name="intel-gpu-jammy.list",
+ source_list_file_name="google-chrome.sources",
  sources="deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main",
  gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub",
  gpg_key_name="google-chrome.gpg"
  )
- command = UbuntuApplicationSourceManager()
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
  with (patch("builtins.open", new_callable=mock_open()),
  patch("dispatcher.source.ubuntu_source_manager.add_gpg_key")):
  command.add(params)
  except SourceError as err:
  pytest.fail(f"'UbuntuApplicationSourceManager.add' raised an exception {err}")
 
  def test_add_app_deb_822_format_successfully(self):
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
  try:
  params = ApplicationAddSourceParameters(
  source_list_file_name="google-chrome.sources",
@@ -219,18 +224,19 @@ def test_add_app_deb_822_format_successfully(self):
  "Suites: stable"
  "Components: main",
  )
- command = UbuntuApplicationSourceManager()
+ command = UbuntuApplicationSourceManager(broker)
  with patch("builtins.open", new_callable=mock_open()):
  command.add(params)
  except SourceError as err:
  pytest.fail(f"'UbuntuApplicationSourceManager.add' raised an exception {err}")
 
  def test_update_app_source_successfully(self):
  try:
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
  params = ApplicationUpdateSourceParameters(
  source_list_file_name="intel-gpu-jammy.list", sources=APP_SOURCE
  )
- command = UbuntuApplicationSourceManager()
+ command = UbuntuApplicationSourceManager(broker)
  with patch("builtins.open", new_callable=mock_open()):
  command.update(params)
  except SourceError as err:
@@ -249,7 +255,8 @@ def test_list(self, sources_list_d_content):
  with patch("glob.glob", return_value=["/etc/apt/sources.list.d/example.list"]), patch(
  "builtins.open", mock_open(read_data=sources_list_d_content)
  ):
- command = UbuntuApplicationSourceManager()
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
  sources = command.list()
  assert sources[0].name == "example.list"
  assert sources[0].sources == [
@@ -261,7 +268,8 @@ def test_list_raises_exception(self):
  with patch("glob.glob", return_value=["/etc/apt/sources.list.d/example.list"]), patch(
  "builtins.open", side_effect=OSError
  ):
- command = UbuntuApplicationSourceManager()
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
  with pytest.raises(SourceError) as exc_info:
  command.list()
  assert "Error listing application sources" in str(exc_info.value)
@@ -273,17 +281,49 @@ def test_successfully_remove_gpg_key_and_source_list(
  parameters = ApplicationRemoveSourceParameters(
  gpg_key_name="example_source.gpg", source_list_file_name="example_source.list"
  )
- command = UbuntuApplicationSourceManager()
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
  try:
  command.remove(parameters)
  except SourceError:
  self.fail("Remove GPG key raised DispatcherException unexpectedly!")
 
+ @patch("dispatcher.source.ubuntu_source_manager.verify_source", side_effect=DispatcherException('error'))
+ def test_failed_add_gpg_key_method(self, mock_verify_source):
+ parameters = ApplicationAddSourceParameters(
+ source_list_file_name="example_source.list",
+ sources="deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main",
+ gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub",
+ gpg_key_name="name"
+ )
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
+ with pytest.raises(SourceError) as ex:
+ command.add(parameters)
+ assert str(ex.value) == 'Source Gpg key URI verification check failed: error'
+
+
+ @patch("dispatcher.source.ubuntu_source_manager.verify_source")
+ def test_success_add_gpg_key_method(self, mock_verify_source):
+ mock_verify_source.return_value = True 
+ parameters = ApplicationAddSourceParameters(
+ source_list_file_name="example_source.list",
+ sources="deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main",
+ gpg_key_uri="https://dl-ssl.google.com/linux/linux_signing_key.pub",
+ gpg_key_name="name"
+ )
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
+ with (patch("builtins.open", new_callable=mock_open()),
+ patch("dispatcher.source.ubuntu_source_manager.add_gpg_key")): 
+ command.add(parameters)
+
  def test_raises_when_space_check_fails(self):
  parameters = ApplicationRemoveSourceParameters(
  gpg_key_name="example_source.gpg", source_list_file_name="../example_source.list"
  )
- command = UbuntuApplicationSourceManager()
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
  with pytest.raises(SourceError) as ex:
  command.remove(parameters)
  assert str(ex.value) == "Invalid file name: ../example_source.list"
@@ -293,7 +333,8 @@ def test_raises_when_unable_to_remove_file(self, mock_remove_file):
  parameters = ApplicationRemoveSourceParameters(
  gpg_key_name="example_source.gpg", source_list_file_name="example_source.list"
  )
- command = UbuntuApplicationSourceManager()
+ broker = MockDispatcherBroker.build_mock_dispatcher_broker()
+ command = UbuntuApplicationSourceManager(broker)
  with pytest.raises(SourceError) as ex:
  command.remove(parameters)
- assert str(ex.value) == "Error removing file: example_source.list"
+ assert str(ex.value) == "Error removing file: example_source.list"

inbm/integration-reloaded/test/source/SOURCE.sh

@@ -31,7 +31,10 @@ trap test_failed ERR
 
 echo "Starting source test." | systemd-cat
 
-# OS tests
+# Adding to trusted repo
+inbc append --path trustedRepositories:https://deb.opera.com/
+
+#OS tests
 inbc source os add --sources "$FAKE_SOURCE"
 grep "$FAKE_SOURCE" "$APT_SOURCES"
 inbc source os list 2>&1 | grep "$FAKE_SOURCE"