cachi2: update generate_sbom plugin

Update generate_sbom to process cachi2 results.
SBOM must be combined from cachi2 SBOM stored in build_dir
in this case.

Signed-off-by: Martin Basti <[email protected]>

atomic_reactor/plugins/generate_sbom.py

@@ -13,13 +13,15 @@
 from typing import Any, Dict, List, Optional
 
 from atomic_reactor.constants import (PLUGIN_GENERATE_SBOM,
+                                      PLUGIN_CACHI2_POSTPROCESS,
                                       PLUGIN_RPMQA,
                                       PLUGIN_RESOLVE_REMOTE_SOURCE,
                                       SBOM_SCHEMA_PATH,
                                       PLUGIN_FETCH_MAVEN_KEY,
                                       INSPECT_CONFIG,
                                       KOJI_BTYPE_ICM,
-                                      ICM_JSON_FILENAME)
+                                      ICM_JSON_FILENAME,
+                                      CACHI2_BUILD_DIR)
 from atomic_reactor.config import get_cachito_session, get_koji_session
 from atomic_reactor.utils import retries
 from atomic_reactor.utils.cachito import CachitoAPI
@@ -92,6 +94,8 @@ def __init__(self, workflow):
         remote_source_results = wf_data.plugins_results.get(PLUGIN_RESOLVE_REMOTE_SOURCE) or []
         self.remote_source_ids = [remote_source['id'] for remote_source in remote_source_results]
 
+        self.cachi2_remote_sources = wf_data.plugins_results.get(PLUGIN_CACHI2_POSTPROCESS) or []
+
         self.rpm_components = wf_data.plugins_results.get(PLUGIN_RPMQA) or {}
 
         fetch_maven_results = wf_data.plugins_results.get(PLUGIN_FETCH_MAVEN_KEY) or {}
@@ -131,6 +135,12 @@ def fetch_url_or_koji_check(self) -> None:
         if read_fetch_artifacts_url(self.workflow):
             self.incompleteness_reasons.add("fetch url is used")
 
+    def get_cachi2_sbom(self) -> dict:
+        """Get SBOM from cachi2 results"""
+        global_sbom_path = self.workflow.build_dir.path/CACHI2_BUILD_DIR/"bom.json"
+        with open(global_sbom_path, "r") as f:
+            return json.load(f)
+
     def add_parent_missing_sbom_reason(self, nvr: str) -> None:
         self.incompleteness_reasons.add(f"parent build '{nvr}' is missing SBOM")
 
@@ -331,6 +341,8 @@ def run(self) -> Dict[str, Any]:
         if self.remote_source_ids:
             remote_sources_sbom = self.cachito_session.get_sbom(self.remote_source_ids)
             remote_souces_components = remote_sources_sbom['components']
+        elif self.cachi2_remote_sources:  # Cachi2 and Cachito are not supported to be used together
+            remote_souces_components = self.get_cachi2_sbom()['components']
 
         # add components from cachito, rpms, pnc
         for platform in self.all_platforms:

tests/plugins/test_generate_sbom.py

@@ -32,6 +32,8 @@
     REPO_FETCH_ARTIFACTS_KOJI,
     REPO_FETCH_ARTIFACTS_URL,
     PLUGIN_CHECK_AND_SET_PLATFORMS_KEY,
+    PLUGIN_CACHI2_POSTPROCESS,
+    CACHI2_BUILD_DIR,
 )
 from atomic_reactor.plugin import PluginFailedException
 from atomic_reactor.plugins.generate_sbom import GenerateSbomPlugin
@@ -1065,7 +1067,7 @@ def teardown_function(*args):
     sys.modules.pop(GenerateSbomPlugin.key, None)
 
 
-def mock_env(workflow, df_images):
+def mock_env(workflow, df_images, cachi2=False):
     tmp_dir = tempfile.mkdtemp()
     dockerconfig_contents = {"auths": {LOCALHOST_REGISTRY: {"username": "user",
                                                             "email": "[email protected]",
@@ -1099,12 +1101,20 @@ def mock_env(workflow, df_images):
            .for_plugin(GenerateSbomPlugin.key)
            .set_reactor_config(r_c_m)
            .set_dockerfile_images(df_images)
-           .set_plugin_result(PLUGIN_RESOLVE_REMOTE_SOURCE, deepcopy(REMOTE_SOURCES))
            .set_plugin_result(PLUGIN_FETCH_MAVEN_KEY,
                               {'sbom_components': deepcopy(PNC_SBOM_COMPONENTS)})
            .set_plugin_result(PLUGIN_RPMQA, deepcopy(RPM_SBOM_COMPONENTS))
            )
 
+    if cachi2:
+        # Note: using CACHITO_SBOM_JSON here, as the fields are almost the same as
+        # for Cachi2; I don't want to die from mocking everything again, just to
+        # make test pass for extra property "found_by: cachi2" added by cachi2
+        # this provides good tests
+        mock_cachi2_sbom(workflow, deepcopy(CACHITO_SBOM_JSON))
+    else:
+        env.set_plugin_result(PLUGIN_RESOLVE_REMOTE_SOURCE, deepcopy(REMOTE_SOURCES))
+
     all_inspects = [(EMPTY_SBOM_IMAGE_LABELS, EMPTY_SBOM_IMAGE_NAME),
                     (MISSING_LABEL_IMAGE_LABELS, MISSING_LABEL_IMAGE_NAME),
                     (BUILDING_IMAGE_LABELS, BUILDING_IMAGE_NAME),
@@ -1131,6 +1141,19 @@ def mock_get_sbom_cachito(requests_mock):
     requests_mock.register_uri('GET', CACHITO_SBOM_URL, json=CACHITO_SBOM_JSON)
 
 
+def mock_cachi2_sbom(workflow, cachi2_sbom: dict):
+    workflow.data.plugins_results[PLUGIN_CACHI2_POSTPROCESS] = {
+        "plugin": "did run, real value doesn't matter"
+    }
+
+    # save cachi2 SBOM which is source for ICM
+    path = workflow.build_dir.path/CACHI2_BUILD_DIR/"bom.json"
+    path.parent.mkdir()
+    with open(path, "w") as f:
+        json.dump(cachi2_sbom, f)
+        f.flush()
+
+
 def mock_build_icm_urls(requests_mock):
     all_sboms = [(EMPTY_SBOM_KOJI_BUILD, EMPTY_SBOM_BUILD_SBOM_JSON),
                  (BASE_WITH_SBOM_KOJI_BUILD, BASE_WITH_SBOM_BUILD_SBOM_JSON),
@@ -1311,12 +1334,16 @@ def koji_session():
             INCOMPLETE_CACHE_URL_KOJI,
         ),
     ])
+@pytest.mark.parametrize('cachi2', [True, False])
 def test_sbom(workflow, requests_mock, koji_session, df_images, use_cache, use_fetch_url,
-              use_fetch_koji, expected_components, expected_incomplete):
-    mock_get_sbom_cachito(requests_mock)
+              use_fetch_koji, expected_components, expected_incomplete, cachi2):
+
+    if not cachi2:
+        mock_get_sbom_cachito(requests_mock)
+
     mock_build_icm_urls(requests_mock)
 
-    runner = mock_env(workflow, df_images)
+    runner = mock_env(workflow, df_images, cachi2=cachi2)
     workflow.data.tag_conf.add_unique_image(UNIQUE_IMAGE)
 
     def check_cosign_run(args):