Upgrade CycloneDX output to schema v1.5 #807 (#1057)

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

Author: tdruez

CHANGELOG.rst

@@ -17,6 +17,10 @@ Unreleased
   compatible with the old names.
   https://github.com/nexB/scancode.io/issues/1044
 
+- Generate CycloneDX SBOM in 1.5 spec format, migrated from 1.4 previously.
+  The Package vulnerabilities are now included in the CycloneDX SBOM when available.
+  https://github.com/nexB/scancode.io/issues/807
+
 v33.0.0 (2024-01-16)
 --------------------
 

scanpipe/models.py

@@ -66,8 +66,8 @@
 from commoncode.fileutils import parent_directory
 from cyclonedx import model as cyclonedx_model
 from cyclonedx.model import component as cyclonedx_component
+from cyclonedx.model import license as cyclonedx_license
 from extractcode import EXTRACT_SUFFIX
-from formattedcode.output_cyclonedx import CycloneDxExternalRef
 from licensedcode.cache import build_spdx_license_expression
 from licensedcode.cache import get_licensing
 from matchcode_toolkit.fingerprinting import IGNORED_DIRECTORY_FINGERPRINTS
@@ -3062,9 +3062,9 @@ def as_cyclonedx(self):
         """Return this DiscoveredPackage as an CycloneDX Component entry."""
         licenses = []
         if expression_spdx := self.get_declared_license_expression_spdx():
-            licenses = [
-                cyclonedx_model.LicenseChoice(license_expression=expression_spdx),
-            ]
+            # Using the LicenseExpression directly as the make_with_expression method
+            # does not support the "LicenseRef-" keys.
+            licenses = [cyclonedx_license.LicenseExpression(value=expression_spdx)]
 
         hash_fields = {
             "md5": cyclonedx_model.HashAlgorithm.MD5,
@@ -3073,7 +3073,7 @@ def as_cyclonedx(self):
             "sha512": cyclonedx_model.HashAlgorithm.SHA_512,
         }
         hashes = [
-            cyclonedx_model.HashType(algorithm=algorithm, hash_value=hash_value)
+            cyclonedx_model.HashType(alg=algorithm, content=hash_value)
             for field_name, algorithm in hash_fields.items()
             if (hash_value := getattr(self, field_name))
         ]
@@ -3097,25 +3097,53 @@ def as_cyclonedx(self):
             if (value := getattr(self, field_name)) not in EMPTY_VALUES
         ]
 
-        cyclonedx_url_to_type = CycloneDxExternalRef.cdx_url_type_by_scancode_field
+        reference_type = cyclonedx_model.ExternalReferenceType
+        url_field_to_cdx_type = {
+            "api_data_url": reference_type.BOM,
+            "bug_tracking_url": reference_type.ISSUE_TRACKER,
+            "code_view_url": reference_type.OTHER,
+            "download_url": reference_type.DISTRIBUTION,
+            "homepage_url": reference_type.WEBSITE,
+            "repository_download_url": reference_type.DISTRIBUTION,
+            "repository_homepage_url": reference_type.WEBSITE,
+            "vcs_url": reference_type.VCS,
+        }
         external_references = [
-            cyclonedx_model.ExternalReference(reference_type=reference_type, url=url)
-            for field_name, reference_type in cyclonedx_url_to_type.items()
+            cyclonedx_model.ExternalReference(type=reference_type, url=url)
+            for field_name, reference_type in url_field_to_cdx_type.items()
             if (url := getattr(self, field_name)) and field_name not in property_fields
         ]
 
-        purl = self.package_url
+        # Always use the package_uid when available to ensure having unique
+        # package_url in the BOM when several instances of the same DiscoveredPackage
+        # (i.e. same purl) are present in the project.
+        try:
+            package_url = PackageURL.from_string(self.package_uid)
+        except ValueError:
+            package_url = self.get_package_url()
+
+        evidence = None
+        if self.other_license_expression_spdx:
+            evidence = cyclonedx_component.ComponentEvidence(
+                licenses=[
+                    cyclonedx_license.LicenseExpression(
+                        value=self.other_license_expression_spdx
+                    )
+                ],
+            )
+
         return cyclonedx_component.Component(
             name=self.name,
             version=self.version,
-            bom_ref=self.package_uid or str(self.uuid),
-            purl=purl,
+            bom_ref=str(package_url),
+            purl=package_url,
             licenses=licenses,
-            copyright_=self.copyright,
+            copyright=self.copyright,
             description=self.description,
             hashes=hashes,
             properties=properties,
             external_references=external_references,
+            evidence=evidence,
         )
 
 

scanpipe/pipes/cyclonedx.py

@@ -117,10 +117,11 @@ def get_checksums(component):
         "SHA-256": "sha256",
         "SHA-512": "sha512",
     }
+
     return {
-        algorithm_map_cdx_scio[algo_hash.alg.value]: algo_hash.content.__root__
+        algorithm_map_cdx_scio[algo_hash.alg]: algo_hash.content
         for algo_hash in component.hashes
-        if algo_hash.alg.value in algorithm_map_cdx_scio
+        if algo_hash.alg in algorithm_map_cdx_scio
     }
 
 
@@ -132,7 +133,7 @@ def get_external_references(component):
 
     references = defaultdict(list)
     for reference in external_references:
-        references[reference.type.value].append(reference.url)
+        references[reference.type].append(reference.url)
 
     return dict(references)
 

scanpipe/pipes/output.py

@@ -21,6 +21,7 @@
 # Visit https://github.com/nexB/scancode.io for support and download.
 
 import csv
+import decimal
 import json
 import re
 from operator import attrgetter
@@ -34,9 +35,13 @@
 
 import saneyaml
 import xlsxwriter
-from cyclonedx import output as cyclonedx_output
-from cyclonedx.model import bom as cyclonedx_bom
-from cyclonedx.model import component as cyclonedx_component
+from cyclonedx.model import bom as cdx_bom
+from cyclonedx.model import component as cdx_component
+from cyclonedx.model import vulnerability as cdx_vulnerability
+from cyclonedx.output import OutputFormat
+from cyclonedx.output import make_outputter
+from cyclonedx.schema import SchemaVersion
+from cyclonedx.validation.json import JsonStrictValidator
 from license_expression import Licensing
 from license_expression import ordered_unique
 from licensedcode.cache import build_spdx_license_expression
@@ -579,65 +584,156 @@ def to_spdx(project, include_files=False):
     return output_file
 
 
+def vulnerability_as_cyclonedx(vulnerability_data, component_bom_ref):
+    affects = [cdx_vulnerability.BomTarget(ref=f"urn:cdx:{component_bom_ref}")]
+
+    source = cdx_vulnerability.VulnerabilitySource(
+        name="VulnerableCode",
+        url=vulnerability_data.get("url"),
+    )
+
+    references = []
+    ratings = []
+    for reference in vulnerability_data.get("references", []):
+        source = cdx_vulnerability.VulnerabilitySource(
+            url=reference.get("reference_url"),
+        )
+
+        references.append(
+            cdx_vulnerability.VulnerabilityReference(
+                id=reference.get("reference_id"),
+                source=source,
+            )
+        )
+
+        for score_entry in reference.get("scores", []):
+            # CycloneDX only support a float value for the score field,
+            # where on the VulnerableCode data it can be either a score float value
+            # or a severity string value.
+            score_value = score_entry.get("value")
+            try:
+                score = decimal.Decimal(score_value)
+                severity = None
+            except decimal.DecimalException:
+                score = None
+                severity = getattr(
+                    cdx_vulnerability.VulnerabilitySeverity,
+                    score_value.upper(),
+                    None,
+                )
+
+            ratings.append(
+                cdx_vulnerability.VulnerabilityRating(
+                    source=source,
+                    score=score,
+                    severity=severity,
+                    # Providing a value for method raise a AssertionError
+                    # method=score_entry.get("scoring_system"),
+                    vector=score_entry.get("scoring_elements"),
+                )
+            )
+
+    cwes = [
+        weakness.get("cwe_id") for weakness in vulnerability_data.get("weaknesses", [])
+    ]
+
+    return cdx_vulnerability.Vulnerability(
+        id=vulnerability_data.get("vulnerability_id"),
+        source=source,
+        description=vulnerability_data.get("summary"),
+        affects=affects,
+        references=references,
+        cwes=cwes,
+        ratings=ratings,
+    )
+
+
 def get_cyclonedx_bom(project):
     """
     Return a CycloneDX `Bom` object filled with provided `project` data.
     See https://cyclonedx.org/use-cases/#dependency-graph
     """
-    components = [
-        *get_queryset(project, "discoveredpackage"),
-    ]
-
-    cyclonedx_components = [component.as_cyclonedx() for component in components]
-
-    bom = cyclonedx_bom.Bom(components=cyclonedx_components)
-
-    project_as_cyclonedx = cyclonedx_component.Component(
+    project_as_root_component = cdx_component.Component(
         name=project.name,
         bom_ref=str(project.uuid),
     )
 
-    project_as_cyclonedx.dependencies.update(
-        [component.bom_ref for component in cyclonedx_components]
-    )
-
-    bom.metadata = cyclonedx_bom.BomMetaData(
-        component=project_as_cyclonedx,
+    bom = cdx_bom.Bom()
+    bom.metadata = cdx_bom.BomMetaData(
+        component=project_as_root_component,
         tools=[
-            cyclonedx_bom.Tool(
+            cdx_bom.Tool(
                 name="ScanCode.io",
                 version=scancodeio_version,
             )
         ],
         properties=[
-            cyclonedx_bom.Property(
+            cdx_bom.Property(
                 name="notice",
                 value=SCAN_NOTICE,
             )
         ],
     )
 
+    components = []
+    vulnerabilities = []
+    for package in get_queryset(project, "discoveredpackage"):
+        component = package.as_cyclonedx()
+        components.append(component)
+
+        for vulnerability_data in package.affected_by_vulnerabilities:
+            vulnerabilities.append(
+                vulnerability_as_cyclonedx(
+                    vulnerability_data=vulnerability_data,
+                    component_bom_ref=component.bom_ref,
+                )
+            )
+
+    for component in components:
+        bom.components.add(component)
+        bom.register_dependency(project_as_root_component, [component])
+
+    bom.vulnerabilities = vulnerabilities
+
     return bom
 
 
-def to_cyclonedx(project):
+def sort_bom_with_schema_ordering(bom_as_dict, schema_version):
+    """Sort the ``bom_as_dict`` using the ordering from the ``schema_version``."""
+    schema_file = JsonStrictValidator(schema_version)._schema_file
+    with open(schema_file) as sf:
+        schema_dict = json.loads(sf.read())
+
+    order_from_schema = list(schema_dict.get("properties", {}).keys())
+    ordered_dict = {
+        key: bom_as_dict.get(key) for key in order_from_schema if key in bom_as_dict
+    }
+
+    return json.dumps(ordered_dict, indent=2)
+
+
+def to_cyclonedx(project, schema_version=SchemaVersion.V1_5):
     """
     Generate output for the provided ``project`` in CycloneDX BOM format.
     The output file is created in the ``project`` "output/" directory.
     Return the path of the generated output file.
     """
     output_file = project.get_output_file_path("results", "cdx.json")
 
-    cyclonedx_bom = get_cyclonedx_bom(project)
+    bom = get_cyclonedx_bom(project)
+    json_outputter = make_outputter(bom, OutputFormat.JSON, schema_version)
 
-    outputter = cyclonedx_output.get_instance(
-        bom=cyclonedx_bom,
-        output_format=cyclonedx_output.OutputFormat.JSON,
-    )
+    # Using the internal API in place of the output_as_string() method to avoid
+    # a round of deserialization/serialization while fixing the field ordering.
+    json_outputter.generate()
+    bom_as_dict = json_outputter._bom_json
+
+    # The default order out of the outputter is not great, the following sorts the
+    # bom using the order from the schema.
+    sorted_json = sort_bom_with_schema_ordering(bom_as_dict, schema_version)
 
-    bom_json = outputter.output_as_string()
     with output_file.open("w") as file:
-        file.write(bom_json)
+        file.write(sorted_json)
 
     return output_file
 

scanpipe/tests/__init__.py

@@ -137,6 +137,8 @@ def make_resource_directory(project, path, **extra):
     ),
     "declared_license_expression": "gpl-2.0 AND gpl-2.0-plus",
     "declared_license_expression_spdx": "GPL-2.0-only AND GPL-2.0-or-later",
+    "other_license_expression": "apache-2.0 AND (mpl-1.1 OR gpl-2.0)",
+    "other_license_expression_spdx": "Apache-2.0 AND (MPL-1.1 OR GPL-2.0)",
     "extracted_license_statement": "",
     "notice_text": "Notice\nText",
     "root_path": None,