Store SBOMs headers in the Project.extra_data field #1253 (#1266)

* Store SBOMs headers in the `Project.extra_data` field #1253

Signed-off-by: tdruez <tdruez@nexb.com>

* Add unit test for the SBOM headers on Project.extra_data #1253

Signed-off-by: tdruez <tdruez@nexb.com>

---------

Signed-off-by: tdruez <tdruez@nexb.com>

Author: tdruez

CHANGELOG.rst

@@ -1,6 +1,13 @@
 Changelog
 =========
 
+v34.6.2 (unreleased)
+--------------------
+
+- Store SBOMs headers in the `Project.extra_data` field during the load_sboms
+  pipeline.
+  https://github.com/nexB/scancode.io/issues/1253
+
 v34.6.1 (2024-06-07)
 --------------------
 

scanpipe/pipes/resolve.py

@@ -58,6 +58,7 @@ def get_packages(project, package_registry, manifest_resources, model=None):
     get package data for resolved packages from package requirements.
     """
     resolved_packages = []
+    sboms_headers = {}
 
     if not manifest_resources.exists():
         project.add_warning(
@@ -69,13 +70,18 @@ def get_packages(project, package_registry, manifest_resources, model=None):
     for resource in manifest_resources:
         if packages := resolve_manifest_resources(resource, package_registry):
             resolved_packages.extend(packages)
+            if headers := get_manifest_headers(resource):
+                sboms_headers[resource.name] = headers
         else:
             project.add_error(
                 description="No packages could be resolved",
                 model=model,
                 resource=resource,
             )
 
+    if sboms_headers:
+        project.update_extra_data({"sboms_headers": sboms_headers})
+
     return resolved_packages
 
 
@@ -320,3 +326,49 @@ def set_license_expression(package_data):
             package_data["declared_license_expression"] = license_expression
 
     return package_data
+
+
+def get_manifest_headers(resource):
+    """Extract headers from a manifest file based on its package type."""
+    input_location = resource.location
+    package_type = get_default_package_type(input_location)
+    extract_fields = []
+
+    if package_type == "cyclonedx":
+        extract_fields = [
+            "bomFormat",
+            "specVersion",
+            "serialNumber",
+            "version",
+            "metadata",
+        ]
+    elif package_type == "spdx":
+        extract_fields = [
+            "spdxVersion",
+            "dataLicense",
+            "SPDXID",
+            "name",
+            "documentNamespace",
+            "creationInfo",
+            "comment",
+        ]
+
+    if extract_fields:
+        return extract_headers(input_location, extract_fields)
+
+
+def extract_headers(input_location, extract_fields):
+    """Read a file from the given location and extracts specified fields."""
+    input_path = Path(input_location)
+    document_data = input_path.read_text()
+
+    if str(input_location).endswith(".json"):
+        cyclonedx_document = json.loads(document_data)
+        extracted_headers = {
+            field: value
+            for field, value in cyclonedx_document.items()
+            if field in extract_fields
+        }
+        return extracted_headers
+
+    return {}

scanpipe/tests/pipes/test_resolve.py

@@ -24,6 +24,8 @@
 
 from django.test import TestCase
 
+import mock
+
 from scanpipe import pipes
 from scanpipe.models import Project
 from scanpipe.pipes import resolve
@@ -201,6 +203,20 @@ def test_scanpipe_resolve_get_packages_from_sbom(self):
         resource1 = project1.codebaseresources.get(name="toml.spdx.json")
         self.assertEqual([resource1], package.get("codebase_resources"))
 
+        self.assertEqual(["sboms_headers"], list(project1.extra_data.keys()))
+        sboms_headers = project1.extra_data["sboms_headers"]
+        self.assertEqual(["toml.spdx.json"], list(sboms_headers.keys()))
+        expected = [
+            "spdxVersion",
+            "dataLicense",
+            "SPDXID",
+            "name",
+            "documentNamespace",
+            "creationInfo",
+            "comment",
+        ]
+        self.assertEqual(expected, list(sboms_headers["toml.spdx.json"].keys()))
+
     def test_scanpipe_resolve_create_packages_and_dependencies(self):
         project1 = Project.objects.create(name="Analysis")
         input_location = self.data_location / "manifests" / "toml.spdx.json"
@@ -221,3 +237,18 @@ def test_scanpipe_resolve_create_packages_and_dependencies(self):
         resource1 = project1.codebaseresources.get(name="toml.spdx.json")
         package = project1.discoveredpackages.get()
         self.assertEqual(resource1, package.codebase_resources.get())
+
+    def test_scanpipe_resolve_get_manifest_headers(self):
+        input_location = self.data_location / "manifests" / "toml.spdx.json"
+        resource = mock.Mock(location=input_location)
+        expected = [
+            "spdxVersion",
+            "dataLicense",
+            "SPDXID",
+            "name",
+            "documentNamespace",
+            "creationInfo",
+            "comment",
+        ]
+        headers = resolve.get_manifest_headers(resource)
+        self.assertEqual(expected, list(headers.keys()))