Address comments from AI code review

inbc-program/inbc/parser/source_app_parser.py

@@ -8,12 +8,13 @@
 
 from ..inbc_exception import InbcException
 from ..xml_tag import create_xml_tag
+from inbm_common_lib.utility import clean_input
 
 logger = logging.getLogger(__name__)
 
 
 def application_add(args: argparse.Namespace) -> str:
-    if (args.gpgKeyUri and args.gpgKeyName is None) or (args.gpgKeyName and args.gpgKeyUri is None):
+    if bool(args.gpgKeyUri) != bool(args.gpgKeyName):
         raise InbcException(
             "Source requires either both gpgKeyUri and gpgKeyName to be provided, or neither of them.")
 
@@ -35,8 +36,8 @@ def application_add(args: argparse.Namespace) -> str:
 
     manifest += '<repo><repos>'
 
-    for source in args.sources:
-        manifest += '<source_pkg>' + source + '</source_pkg>'
+    source_tags = (f'<source_pkg>{clean_input(source)}</source_pkg>' for source in args.sources)
+    manifest += ''.join(source_tags)
 
     manifest += ('</repos>'
                  f'{create_xml_tag(arguments, "filename")}</repo>'
@@ -82,7 +83,7 @@ def application_update(args: argparse.Namespace) -> str:
                 '<update><repo><repos>')
 
     for source in args.sources:
-        manifest += '<source_pkg>' + source.strip() + '</source_pkg>'
+        manifest += '<source_pkg>' + clean_input(source.strip()) + '</source_pkg>'
 
     manifest += (f'</repos>{create_xml_tag(arguments, "filename")}' +
                  '</repo></update></applicationSource>' +

inbc-program/inbc/parser/source_os_parser.py

@@ -4,13 +4,14 @@
    SPDX-License-Identifier: Apache-2.0
 """
 import argparse
+from inbm_common_lib.utility import clean_input
 
 
 def os_add(args: argparse.Namespace) -> str:
     manifest = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type>' \
                '<osSource><add><repos>'
     for source in args.sources:
-        manifest += '<source_pkg>' + source.strip() + '</source_pkg>'
+        manifest += '<source_pkg>' + clean_input(source.strip()) + '</source_pkg>'
     manifest += '</repos></add></osSource></manifest>'
 
     print("manifest {0}".format(manifest))
@@ -21,7 +22,7 @@ def os_remove(args: argparse.Namespace) -> str:
     manifest = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type>' \
                '<osSource><remove><repos>'
     for source in args.sources:
-        manifest += '<source_pkg>' + source.strip() + '</source_pkg>'
+        manifest += '<source_pkg>' + clean_input(source.strip()) + '</source_pkg>'
     manifest += '</repos></remove></osSource></manifest>'
 
     print("manifest {0}".format(manifest))
@@ -32,7 +33,7 @@ def os_update(args: argparse.Namespace) -> str:
     manifest = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type>' \
                '<osSource><update><repos>'
     for source in args.sources:
-        manifest += '<source_pkg>' + source.strip() + '</source_pkg>'
+        manifest += '<source_pkg>' + clean_input(source.strip()) + '</source_pkg>'
     manifest += '</repos></update></osSource></manifest>'
 
     print("manifest {0}".format(manifest))

inbc-program/tests/unit/test_source_app_parser.py

@@ -30,11 +30,11 @@ def test_parse_add_arguments_deb822_format_separate_lines_successfully(self):
         f = self.arg_parser.parse_args(
             ['source', 'application', 'add',
              '--sources', 'X-Repolib-Name: Google Chrome',
-                          'Enabled: yes',
-                          'Types: deb',
-                          'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
-                          'Suites: stable',
-                          'Components: main',
+             'Enabled: yes',
+             'Types: deb',
+             'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
+             'Suites: stable',
+             'Components: main',
              '--filename', 'intel-gpu-jammy.list'])
         self.assertEqual(f.gpgKeyUri, None)
         self.assertEqual(f.gpgKeyName, None)
@@ -46,64 +46,64 @@ def test_parse_add_arguments_deb822_format_separate_lines_successfully(self):
                                      'Components: main'])
         self.assertEqual(f.filename, 'intel-gpu-jammy.list')
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_raise_application_add_with_only_one_gpgKeyUri_param(self, m_connect):
+    def test_raise_application_add_with_only_one_gpgKeyUri_param(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'add',
              '--gpgKeyUri', 'https://repositories.intel.com/gpu/intel-graphics.key',
              '--sources', 'deb http://example.com/ focal main restricted universe',
              'deb-src http://example.com/ focal-security main',
              '--filename', 'intel-gpu-jammy.list'])
-        with pytest.raises(InbcException,
-                           match="Source requires either both gpgKeyUri and gpgKeyName "
-                                 "to be provided, or neither of them."):
-            Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            with pytest.raises(InbcException,
+                               match="Source requires either both gpgKeyUri and gpgKeyName "
+                                     "to be provided, or neither of them."):
+                Inbc(p, 'source', False)
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_raise_application_add_with_only_one_gpgKeyName_param(self, m_connect):
+    def test_raise_application_add_with_only_one_gpgKeyName_param(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'add',
              '--gpgKeyName', 'intel-graphics.gpg',
              '--sources', 'deb http://example.com/ focal main restricted universe',
              'deb-src http://example.com/ focal-security main',
              '--filename', 'intel-gpu-jammy.list'])
-        with pytest.raises(InbcException,
-                           match="Source requires either both gpgKeyUri and gpgKeyName "
-                                 "to be provided, or neither of them."):
-            Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            with pytest.raises(InbcException,
+                               match="Source requires either both gpgKeyUri and gpgKeyName "
+                                     "to be provided, or neither of them."):
+                Inbc(p, 'source', False)
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_create_add_deb_822_format_manifest_successfully(self, m_connect):
+    def test_create_add_deb_822_format_manifest_successfully(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'add',
              '--sources', 'X-Repolib-Name: Google Chrome',
-                          'Enabled: yes',
-                          'Types: deb',
-                          'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
-                          'Suites: stable',
-                          'Components: main',
+             'Enabled: yes',
+             'Types: deb',
+             'URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub',
+             'Suites: stable',
+             'Components: main',
              '--filename', 'google-chrome.sources'])
-        Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            Inbc(p, 'source', False)
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type><applicationSource>' \
                    '<add><repo><repos><source_pkg>X-Repolib-Name: Google Chrome</source_pkg>' \
                    '<source_pkg>Enabled: yes</source_pkg>' \
-                   '<source_pkg>Types: deb</source_pkg>'\
+                   '<source_pkg>Types: deb</source_pkg>' \
                    '<source_pkg>URIs: https://dl-ssl.google.com/linux/linux_signing_key.pub</source_pkg>' \
                    '<source_pkg>Suites: stable</source_pkg>' \
                    '<source_pkg>Components: main</source_pkg>' \
                    '</repos><filename>google-chrome.sources</filename></repo></add></applicationSource></manifest>'
         self.assertEqual(p.func(p), expected)
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_create_add_all_param_manifest_successfully(self, m_connect):
+    def test_create_add_all_param_manifest_successfully(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'add',
              '--gpgKeyUri', 'https://repositories.intel.com/gpu/intel-graphics.key',
              '--gpgKeyName', 'intel-graphics.gpg',
              '--sources', 'deb http://example.com/ focal main restricted universe',
              'deb-src http://example.com/ focal-security main',
              '--filename', 'intel-gpu-jammy.list'])
-        Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            Inbc(p, 'source', False)
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type><applicationSource>' \
                    '<add><gpg><uri>https://repositories.intel.com/gpu/intel-graphics.key</uri>' \
                    '<keyname>intel-graphics.gpg</keyname></gpg><repo><repos>' \
@@ -112,14 +112,14 @@ def test_create_add_all_param_manifest_successfully(self, m_connect):
                    '</repos><filename>intel-gpu-jammy.list</filename></repo></add></applicationSource></manifest>'
         self.assertEqual(p.func(p), expected)
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_create_add_minus_gpg_manifest_successfully(self, m_connect):
+    def test_create_add_minus_gpg_manifest_successfully(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'add',
              '--sources', 'deb http://example.com/ focal main restricted universe',
              'deb-src http://example.com/ focal-security main',
              '--filename', 'intel-gpu-jammy.list'])
-        Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            Inbc(p, 'source', False)
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type><applicationSource>' \
                    '<add><repo><repos>' \
                    '<source_pkg>deb http://example.com/ focal main restricted universe</source_pkg>' \
@@ -142,24 +142,24 @@ def test_parse_minus_gpg_remove_arguments_successfully(self):
         self.assertEqual(f.gpgKeyName, None)
         self.assertEqual(f.filename, 'intel-gpu-jammy.list')
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_create_remove_manifest_all_params_successfully(self, m_connect):
+    def test_create_remove_manifest_all_params_successfully(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'remove',
              '--gpgKeyName', 'intel-gpu-jammy.gpg',
              '--filename', 'intel-gpu-jammy.list'])
-        Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            Inbc(p, 'source', False)
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type><applicationSource>' \
                    '<remove><gpg><keyname>intel-gpu-jammy.gpg</keyname></gpg>' \
                    '<repo><filename>intel-gpu-jammy.list</filename></repo></remove></applicationSource></manifest>'
         self.assertEqual(p.func(p), expected)
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_create_remove_manifest_minus_gpg_successfully(self, m_connect):
+    def test_create_remove_manifest_minus_gpg_successfully(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'remove',
              '--filename', 'intel-gpu-jammy.list'])
-        Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            Inbc(p, 'source', False)
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type><applicationSource><remove>' \
                    '<repo><filename>intel-gpu-jammy.list</filename></repo></remove></applicationSource></manifest>'
         self.assertEqual(p.func(p), expected)
@@ -174,25 +174,25 @@ def test_parse_update_arguments_successfully(self):
                                      'deb-src http://example.com/ focal-security main'])
         self.assertEqual(f.filename, 'intel-gpu-jammy.list')
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_create_update_manifest_successfully(self, m_connect):
+    def test_create_update_manifest_successfully(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'update',
              '--sources', 'deb http://example.com/ focal main restricted universe',
              'deb-src http://example.com/ focal-security main',
              '--filename', 'intel-gpu-jammy.list'])
-        Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            Inbc(p, 'source', False)
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type><applicationSource>' \
                    '<update><repo><repos><source_pkg>deb http://example.com/ focal main restricted universe' \
                    '</source_pkg><source_pkg>deb-src http://example.com/ focal-security main</source_pkg>' \
                    '</repos><filename>intel-gpu-jammy.list</filename></repo></update></applicationSource></manifest>'
         self.assertEqual(p.func(p), expected)
 
-    @patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect')
-    def test_create_list_manifest_successfully(self, m_connect):
+    def test_create_list_manifest_successfully(self):
         p = self.arg_parser.parse_args(
             ['source', 'application', 'list'])
-        Inbc(p, 'source', False)
+        with patch('inbm_lib.mqttclient.mqtt.mqtt.Client.connect'):
+            Inbc(p, 'source', False)
         expected = '<?xml version="1.0" encoding="utf-8"?><manifest><type>source</type><applicationSource>' \
                    '<list/></applicationSource></manifest>'
         self.assertEqual(p.func(p), expected)

inbm-lib/inbm_common_lib/utility.py

@@ -180,11 +180,11 @@ def canonicalize_uri(url: str) -> CanonicalUri:
 
     WARNING: a string with no forward slashes will have a scheme added. e.g., 'a' -> 'https://a'
 
-    @param url: the url
+    @param url: URL
     @return canonicalized version"""
 
     if url and URL_NULL_CHAR in url:
-        raise UrlSecurityException("Unsafe characters detected in the url. Cannot process the request.")
+        raise UrlSecurityException("Unsafe characters detected in the URL. Cannot process the request.")
 
     return CanonicalUri(value=url_normalize.url_normalize(url))
 
@@ -205,7 +205,9 @@ def is_within_directory(directory: str, target: str) -> bool:
     return prefix == abs_directory
 
 
-def safe_extract(tarball: tarfile.TarFile, path: str = ".", members: Optional[Iterable[tarfile.TarInfo]] = None, *, numeric_owner: bool = False) -> None:
+def safe_extract(tarball: tarfile.TarFile,
+                 path: str = ".",
+                 members: Optional[Iterable[tarfile.TarInfo]] = None, *, numeric_owner: bool = False) -> None:
     """Avoid path traversal when extracting tarball
 
     @param tarball: tarball to extract

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

@@ -6,7 +6,7 @@
 
 from enum import Enum, unique
 from dataclasses import dataclass, field
-from typing import Optional
+from typing import Optional, List
 
 UBUNTU_APT_SOURCES_LIST = "/etc/apt/sources.list"
 UBUNTU_APT_SOURCES_LIST_D = "/etc/apt/sources.list.d"
@@ -21,27 +21,27 @@ class ApplicationSourceList:
 
 @dataclass(kw_only=True, frozen=True)
 class SourceParameters:
-    sources: list[str] = field(default_factory=lambda: [])
+    sources: List[str] = field(default_factory=list)
 
 
 @dataclass(kw_only=True, frozen=True)
 class ApplicationAddSourceParameters:
-    file_name: str
-    sources: list[str] = field(default_factory=lambda: [])
+    source_list_file_name: str
+    sources: List[str] = field(default_factory=list)
     gpg_key_uri: Optional[str] = field(default=None)
     gpg_key_name: Optional[str] = field(default=None)
 
 
 @dataclass(kw_only=True, frozen=True)
 class ApplicationRemoveSourceParameters:
-    file_name: str
+    source_list_file_name: str
     gpg_key_name: Optional[str] = field(default=None)
 
 
 @dataclass(kw_only=True, frozen=True)
 class ApplicationUpdateSourceParameters:
-    file_name: str
-    sources: list[str] = field(default_factory=lambda: [])
+    source_list_file_name: str
+    sources: List[str] = field(default_factory=list)
 
 
 @unique

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

@@ -13,21 +13,6 @@
 logger = logging.getLogger(__name__)
 
 
-def remove_gpg_key_if_exists(gpg_key_name: str) -> None:
-    """Linux - Removes a GPG key file if it exists
-
-    @param gpg_key_name: name of GPG key file to remove (file under LINUX_GPG_KEY_PATH)
-    """
-    try:
-        key_path = os.path.join(LINUX_GPG_KEY_PATH, gpg_key_name)
-        if os.path.exists(key_path):
-            os.remove(key_path)
-        # it's OK if the key is not there
-
-    except OSError as e:
-        raise SourceError(f"Error checking or deleting GPG key: {gpg_key_name}") from e
-
-
 def add_gpg_key(remote_key_path: str, key_store_name: str) -> None:
     """Linux - Adds a GPG key from a remote source
 
@@ -61,4 +46,4 @@ def add_gpg_key(remote_key_path: str, key_store_name: str) -> None:
         raise SourceError(f"Error getting GPG key from remote source: {e}")
 
     except subprocess.CalledProcessError as e:
-        raise SourceError(f"Error running gpg command to dearmor key: {e}")
+        raise SourceError(f"Error running GPG command to dearmor key: {e}")

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

@@ -99,7 +99,7 @@ def _handle_app_source_command(
     Handle the application source commands.
 
     @param parsed_head: XmlHandler with command information
-    @param os_type: os type
+    @param os_type: OS type
     @param app_action: The action to be performed
     @return Result
     """
@@ -120,7 +120,7 @@ def _handle_app_source_command(
             logger.info(f"Optional GPG key parameters not present in manifest")
         filename = parsed_head.get_children("applicationSource/remove/repo")["filename"]
         application_source_manager.remove(
-            ApplicationRemoveSourceParameters(file_name=filename, gpg_key_name=keyname)
+            ApplicationRemoveSourceParameters(source_list_file_name=filename, gpg_key_name=keyname)
         )
         return Result(status=200, message="SUCCESS")
 
@@ -144,7 +144,7 @@ def _handle_app_source_command(
 
         application_source_manager.add(
             ApplicationAddSourceParameters(
-                file_name=repo_filename,
+                source_list_file_name=repo_filename,
                 gpg_key_name=gpg_key_name,
                 gpg_key_uri=gpg_key_uri,
                 sources=add_source_pkgs,
@@ -161,7 +161,7 @@ def _handle_app_source_command(
 
         application_source_manager.update(
             ApplicationUpdateSourceParameters(
-                file_name=repo_filename,
+                source_list_file_name=repo_filename,
                 sources=update_source_pkgs,
             )
         )