Merge pull request #105 from konflux-ci/ISV-5867

feat(ISV-5867): Update parsing of SBOM inputs

Author: jakubgryc

docs/sboms/generate_oci_image.md

@@ -4,8 +4,9 @@ This utility allows users to generate SBOMs related to Container Images.
 
 ## Architecture
 
-The scripts takes at least one SBOM generated by SYFT and at most one SBOM
-generated by Hermeto (previously known as Cachi2). It combines these SBOMs
+The scripts accepts any number of SBOMs generated by SYFT and at most one SBOM
+generated by Hermeto (previously known as Cachi2), with the requirement that 
+at least one SBOM is provided in total. It combines these SBOMs
 and takes them as a context of the built image.
 
 The script also parses a JSON-ified Dockerfile of the image, parses its
@@ -35,7 +36,7 @@ mobster --verbose  generate oci-image \
 ```
 
 ## List of arguments
-- `--from-syft` -- must be present at least once, points to an SBOM file (in a JSON format) created by Syft
+- `--from-syft` -- points to an SBOM file (in a JSON format) created by Syft, can be used multiple times
 - `--from-hermeto` -- points to an SBOM file (in a JSON format) created by Hermeto
 - `--image-pullspec` -- the pullspec of the image processed in the format `<registry>/<repository>:<tag>`
 - `--image-digest` -- the digest of the image processed in the format `sha256:<digest value>`

src/mobster/cli.py

@@ -94,7 +94,6 @@ def validated_additional_reference(value: str) -> str:
     )
     oci_image_parser.add_argument(
         "--from-syft",
-        required=True,
         type=Path,
         help="Path to the SBOM file generated by Syft",
         action="append",

src/mobster/cmd/generate/oci_image/__init__.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+from argparse import ArgumentError
 from pathlib import Path
 from typing import Any
 
@@ -63,6 +64,39 @@ async def _soft_validate_content(self) -> None:
             except CycloneDxException as e:
                 LOGGER.warning("\n".join(e.args))
 
+    def _handle_bom_inputs(
+        self,
+        syft_boms: list[Path] | None,
+        hermeto_bom: Path | None,
+    ) -> dict[str, Any]:
+        """
+        Handles the input SBOM files, merging them if necessary.
+        Args:
+            syft_boms (list[Path] | None): List of Syft SBOM file paths.
+            hermeto_bom (Path | None): Path to a Hermeto SBOM file.
+        Returns:
+            dict[str, Any]: Merged/loaded SBOM dictionary.
+        Raises:
+            ArgumentError: If neither Syft nor Hermeto SBOMs are provided.
+        """
+        if hermeto_bom is None and syft_boms is None:
+            raise ArgumentError(
+                None,
+                "At least one of --from-syft or --from-hermeto must be provided",
+            )
+
+        def _open_bom(bom: Path) -> dict[str, Any]:
+            with open(bom, encoding="utf-8") as bom_file:
+                return json.load(bom_file)  # type: ignore[no-any-return]
+
+        if syft_boms is not None:
+            # Merging Syft & Hermeto SBOMs
+            if len(syft_boms) > 1 or hermeto_bom:
+                return merge_sboms(syft_boms, hermeto_bom)
+            return _open_bom(syft_boms[0])
+
+        return _open_bom(hermeto_bom)  # type: ignore[arg-type]
+
     async def execute(self) -> Any:
         """
         Generate an SBOM document for OCI image.
@@ -81,13 +115,7 @@ async def execute(self) -> Any:
         # contextualize: bool = self.cli_args.contextualize
         # TODO add contextual SBOM utilities    # pylint: disable=fixme
 
-        # Merging Syft & Hermeto SBOMs
-        if len(syft_boms) > 1 or hermeto_bom:
-            merged_sbom_dict = merge_sboms(syft_boms, hermeto_bom)
-        else:
-            # Just one image provided, nothing to merge
-            with open(syft_boms[0], encoding="utf8") as sbom_file:
-                merged_sbom_dict = json.load(sbom_file)
+        merged_sbom_dict = self._handle_bom_inputs(syft_boms, hermeto_bom)
         sbom: Document | CycloneDX1BomWrapper
 
         # Parsing into objects

tests/cmd/generate/oci_image/test_module_init.py

@@ -1,5 +1,6 @@
 import datetime
 import json
+from argparse import ArgumentError
 from pathlib import Path
 from typing import Any, Literal
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -326,3 +327,71 @@ async def test_GenerateOciImageCommand__soft_validate_content_cdx(
     command._content = CycloneDX1BomWrapper(sbom=cdx_sbom_object)
     await command._soft_validate_content()
     mock_logger.warning.assert_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ["syft_boms", "hermeto_bom", "expected_action"],
+    [
+        # no SBOMs
+        (None, None, "raise_error"),
+        # single syft
+        ([Path("syft.json")], None, "load_syft"),
+        # multiple syft
+        ([Path("syft1.json"), Path("syft2.json")], None, "merge"),
+        # syft + hermeto
+        ([Path("syft.json")], Path("hermeto.json"), "merge"),
+        # multiple syft + hermeto
+        ([Path("syft1.json"), Path("syft2.json")], Path("hermeto.json"), "merge"),
+        # only hermeto
+        (None, Path("hermeto.json"), "load_hermeto"),
+    ],
+)
+@patch("builtins.open")
+@patch("json.load")
+@patch("mobster.cmd.generate.oci_image.merge_sboms")
+async def test_GenerateOciImageCommand__handle_bom_inputs(
+    mock_merge: MagicMock,
+    mock_json_load: MagicMock,
+    mock_open: MagicMock,
+    syft_boms: list[Path] | None,
+    hermeto_bom: Path | None,
+    expected_action: Literal["raise_error", "load_syft", "load_hermeto", "merge"],
+) -> None:
+    command = GenerateOciImageCommand(MagicMock())
+
+    mock_syft_data = {"name": "syft_data"}
+    mock_hermeto_data = {"name": "hermeto_data"}
+    mock_merged_data = {"name": "merged_data"}
+
+    mock_file = MagicMock()
+    mock_open.return_value.__enter__.return_value = mock_file
+    mock_json_load.side_effect = (
+        lambda _: mock_syft_data if hermeto_bom is None else mock_hermeto_data
+    )
+    mock_merge.return_value = mock_merged_data
+
+    if expected_action == "raise_error":
+        with pytest.raises(ArgumentError):
+            command._handle_bom_inputs(syft_boms, hermeto_bom)
+    else:
+        result = command._handle_bom_inputs(syft_boms, hermeto_bom)
+
+        if expected_action == "load_syft":
+            assert syft_boms is not None
+            assert hermeto_bom is None
+            mock_open.assert_called_once()
+            assert result == mock_syft_data
+            mock_merge.assert_not_called()
+
+        elif expected_action == "load_hermeto":
+            assert hermeto_bom is not None
+            assert syft_boms is None
+            mock_open.assert_called_once()
+            assert result == mock_hermeto_data
+            mock_merge.assert_not_called()
+
+        elif expected_action == "merge":
+            mock_merge.assert_called_once_with(syft_boms, hermeto_bom)
+            assert result == mock_merged_data
+            mock_open.assert_not_called()

tests/test_cli.py

@@ -33,7 +33,6 @@ def test_parse_concurrency(value: str, expected: int | type) -> None:
 @pytest.mark.parametrize(
     ["command", "success"],
     [
-        (["generate", "oci-image"], False),
         (
             [
                 "generate",