Prototype code to loads dependencies from CyclonDX SBOM #1145

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

scanpipe/pipelines/load_sbom.py

@@ -20,6 +20,9 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/nexB/scancode.io for support and download.
 
+import uuid
+
+from scanpipe.models import DiscoveredDependency
 from scanpipe.pipelines.scan_codebase import ScanCodebase
 from scanpipe.pipes import resolve
 
@@ -46,6 +49,7 @@ def steps(cls):
             cls.get_sbom_inputs,
             cls.get_packages_from_sboms,
             cls.create_packages_from_sboms,
+            cls.create_dependencies,
         )
 
     def get_sbom_inputs(self):
@@ -67,3 +71,28 @@ def create_packages_from_sboms(self):
             project=self.project,
             packages=self.packages,
         )
+
+    def create_dependencies(self):
+        packages = self.project.discoveredpackages.all()
+        for package in packages.filter(extra_data__has_key="depends_on"):
+            for bom_ref in package.extra_data.get("depends_on", []):
+                try:
+                    resolved_to_package = packages.get(extra_data__bom_ref=bom_ref)
+                except Exception:
+                    self.project.add_error(
+                        description="Could not find resolved_to package entry.",
+                        model="create_dependencies",
+                    )
+                    continue
+                DiscoveredDependency.objects.create(
+                    project=self.project,
+                    dependency_uid=str(uuid.uuid4()),
+                    for_package=package,
+                    resolved_to_package=resolved_to_package,
+                )
+                # dependency_data = {
+                #     "for_package": package,
+                #     "resolved_to_package": resolved_to_package,
+                # }
+                # raise ValueError("A purl string argument is required.")
+                # update_or_create_dependency(self.project, dependency_data)

scanpipe/pipes/cyclonedx.py

@@ -150,9 +150,15 @@ def is_cyclonedx_bom(input_location):
     return False
 
 
-def cyclonedx_component_to_package_data(cdx_component):
+def cyclonedx_component_to_package_data(cdx_component, dependencies=None):
     """Return package_data from CycloneDX component."""
-    extra_data = {}
+    dependencies = dependencies or {}
+
+    # Store the original bom_ref and dependencies for future processing.
+    bom_ref = str(cdx_component.bom_ref)
+    extra_data = {"bom_ref": bom_ref}
+    if depends_on := dependencies.get(bom_ref):
+        extra_data["depends_on"] = depends_on
 
     package_url_dict = {}
     if cdx_component.purl:
@@ -271,14 +277,15 @@ def is_empty(value):
     return cyclonedx_document_json
 
 
-def resolve_cyclonedx_packages(input_location):
-    """Resolve the packages from the `input_location` CycloneDX document file."""
+def get_bom_instance_from_file(input_location):
+    """Return a Bom instance from the `input_location` CycloneDX document file."""
     input_path = Path(input_location)
     document_data = input_path.read_text()
 
     if str(input_location).endswith(".xml"):
         cyclonedx_document = SafeElementTree.fromstring(document_data)
         cyclonedx_bom = Bom.from_xml(cyclonedx_document)
+        return cyclonedx_bom
 
     elif str(input_location).endswith(".json"):
         cyclonedx_document = json.loads(document_data)
@@ -294,9 +301,23 @@ def resolve_cyclonedx_packages(input_location):
             raise ValueError(error_msg)
 
         cyclonedx_bom = Bom.from_json(data=cyclonedx_document)
+        return cyclonedx_bom
 
-    else:
+
+def resolve_cyclonedx_packages(input_location):
+    """Resolve the packages from the `input_location` CycloneDX document file."""
+    cyclonedx_bom = get_bom_instance_from_file(input_location)
+    if not cyclonedx_bom:
         return []
 
     components = get_components(cyclonedx_bom)
-    return [cyclonedx_component_to_package_data(component) for component in components]
+
+    dependencies = defaultdict(list)
+    for entry in cyclonedx_bom.dependencies:
+        if depends_on := [str(dep.ref) for dep in entry.dependencies]:
+            dependencies[str(entry.ref)].extend(depends_on)
+
+    return [
+        cyclonedx_component_to_package_data(component, dependencies)
+        for component in components
+    ]