Migrate the cyclonedx resolve to native app for 1.5 support #1056 (#1086)

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

Author: tdruez

CHANGELOG.rst

@@ -4,6 +4,9 @@ Changelog
 v34.1.0 (unreleased)
 --------------------
 
+- Add support for importing CycloneDX SBOM 1.2, 1.3, 1.4 and 1.5 spec formats.
+  https://github.com/nexB/scancode.io/issues/1045
+
 - The pipeline help modal is now available from all project views: form, list, details.
   The docstring are converted from markdown to html for proper rendering.
   https://github.com/nexB/scancode.io/pull/1105

scanpipe/pipes/cyclonedx.py

@@ -27,82 +27,28 @@
 
 from django.core.validators import EMPTY_VALUES
 
-import jsonschema
-from hoppr_cyclonedx_models.cyclonedx_1_4 import (
-    CyclonedxSoftwareBillOfMaterialsStandard as Bom_1_4,
-)
-
-SCHEMAS_PATH = Path(__file__).parent / "schemas"
-
-CYCLONEDX_SPEC_VERSION = "1.4"
-CYCLONEDX_SCHEMA_NAME = "bom-1.4.schema.json"
-CYCLONEDX_SCHEMA_PATH = SCHEMAS_PATH / CYCLONEDX_SCHEMA_NAME
-CYCLONEDX_SCHEMA_URL = (
-    "https://raw.githubusercontent.com/"
-    "CycloneDX/specification/master/schema/bom-1.4.schema.json"
-)
-
-SPDX_SCHEMA_NAME = "spdx.schema.json"
-SPDX_SCHEMA_PATH = SCHEMAS_PATH / SPDX_SCHEMA_NAME
-
-JSF_SCHEMA_NAME = "jsf-0.82.schema.json"
-JSF_SCHEMA_PATH = SCHEMAS_PATH / JSF_SCHEMA_NAME
-
-
-def get_bom(cyclonedx_document):
-    """Return CycloneDX BOM object."""
-    return Bom_1_4(**cyclonedx_document)
-
-
-def get_components(bom):
-    """Return list of components from CycloneDX BOM."""
-    return recursive_component_collector(bom.components, [])
-
-
-def bom_attributes_to_dict(cyclonedx_attributes):
-    """Return list of dict from a list of CycloneDX attributes."""
-    if not cyclonedx_attributes:
-        return []
-
-    return [
-        json.loads(attribute.json(exclude_unset=True, by_alias=True))
-        for attribute in cyclonedx_attributes
-    ]
-
-
-def recursive_component_collector(root_component_list, collected):
-    """Return list of components including the nested components."""
-    if not root_component_list:
-        return []
-
-    for component in root_component_list:
-        extra_data = {}
-        if component.components is not None:
-            extra_data = bom_attributes_to_dict(component.components)
-
-        collected.append({"cdx_package": component, "nested_components": extra_data})
-        recursive_component_collector(component.components, collected)
-    return collected
+from cyclonedx.model import license as cdx_license_model
+from cyclonedx.model.bom import Bom
+from cyclonedx.schema import SchemaVersion
+from cyclonedx.validation import ValidationError
+from cyclonedx.validation.json import JsonStrictValidator
+from packageurl import PackageURL
 
 
 def resolve_license(license):
     """Return license expression/id/name from license item."""
-    if "expression" in license:
-        return license["expression"]
-    elif "id" in license["license"]:
-        return license["license"]["id"]
-    else:
-        return license["license"]["name"]
+    if isinstance(license, cdx_license_model.LicenseExpression):
+        return license.value
+    elif isinstance(license, cdx_license_model.License):
+        return license.id or license.name
 
 
 def get_declared_licenses(licenses):
     """Return resolved license from list of LicenseChoice."""
     if not licenses:
         return ""
 
-    resolved_licenses = [
-        resolve_license(license) for license in bom_attributes_to_dict(licenses)
-    ]
+    resolved_licenses = [resolve_license(license) for license in licenses]
     return "\n".join(resolved_licenses)
 
 
@@ -126,14 +72,14 @@ def get_checksums(component):
 
 
 def get_external_references(component):
-    """Return dict of reference urls from list of `component.externalReferences`."""
-    external_references = component.externalReferences
+    """Return dict of reference urls from list of `component.external_references`."""
+    external_references = component.external_references
     if not external_references:
         return {}
 
     references = defaultdict(list)
     for reference in external_references:
-        references[reference.type].append(reference.url)
+        references[reference.type.value].append(reference.url.uri)
 
     return dict(references)
 
@@ -154,38 +100,90 @@ def get_properties_data(component):
     return properties_data
 
 
-def validate_document(document, schema=CYCLONEDX_SCHEMA_PATH):
-    """Check the validity of this CycloneDX document."""
+def validate_document(document):
+    """
+    Check the validity of this CycloneDX document.
+
+    The validator is loaded from the document specVersion property.
+    """
     if isinstance(document, str):
         document = json.loads(document)
 
-    if isinstance(schema, Path):
-        schema = schema.read_text()
-
-    if isinstance(schema, str):
-        schema = json.loads(schema)
+    spec_version = document.get("specVersion")
+    if not spec_version:
+        return ValidationError("'specVersion' is a required property")
 
-    spdx_schema = SPDX_SCHEMA_PATH.read_text()
-    jsf_schema = JSF_SCHEMA_PATH.read_text()
-
-    store = {
-        "http://cyclonedx.org/schema/spdx.schema.json": json.loads(spdx_schema),
-        "http://cyclonedx.org/schema/jsf-0.82.schema.json": json.loads(jsf_schema),
-    }
+    schema_version = SchemaVersion.from_version(spec_version)
 
-    resolver = jsonschema.RefResolver.from_schema(schema, store=store)
-    validator = jsonschema.Draft7Validator(schema=schema, resolver=resolver)
-    validator.validate(instance=document)
+    json_validator = JsonStrictValidator(schema_version)
+    return json_validator._validata_data(document)
 
 
 def is_cyclonedx_bom(input_location):
     """Return True if the file at `input_location` is a CycloneDX BOM."""
     with suppress(Exception):
         data = json.loads(Path(input_location).read_text())
-        conditions = (
-            data.get("$schema", "").endswith(CYCLONEDX_SCHEMA_NAME),
-            data.get("bomFormat") == "CycloneDX",
-        )
-        if any(conditions):
+        if data.get("bomFormat") == "CycloneDX":
             return True
     return False
+
+
+def cyclonedx_component_to_package_data(cdx_component):
+    """Return package_data from CycloneDX component."""
+    extra_data = {}
+
+    package_url_dict = {}
+    if cdx_component.purl:
+        package_url_dict = PackageURL.from_string(str(cdx_component.purl)).to_dict(
+            encode=True
+        )
+
+    declared_license = get_declared_licenses(licenses=cdx_component.licenses)
+
+    if external_references := get_external_references(cdx_component):
+        extra_data["externalReferences"] = external_references
+
+    if nested_components := cdx_component.get_all_nested_components(include_self=False):
+        nested_purls = [component.bom_ref.value for component in nested_components]
+        extra_data["nestedComponents"] = sorted(nested_purls)
+
+    package_data = {
+        "name": cdx_component.name,
+        "extracted_license_statement": declared_license,
+        "copyright": cdx_component.copyright,
+        "version": cdx_component.version,
+        "description": cdx_component.description,
+        "extra_data": extra_data,
+        **package_url_dict,
+        **get_checksums(cdx_component),
+        **get_properties_data(cdx_component),
+    }
+
+    return {
+        key: value for key, value in package_data.items() if value not in EMPTY_VALUES
+    }
+
+
+def get_bom(cyclonedx_document):
+    """Return CycloneDX BOM object."""
+    return Bom.from_json(data=cyclonedx_document)
+
+
+def get_components(bom):
+    """Return list of components from CycloneDX BOM."""
+    return list(bom._get_all_components())
+
+
+def resolve_cyclonedx_packages(input_location):
+    """Resolve the packages from the `input_location` CycloneDX document file."""
+    input_path = Path(input_location)
+    cyclonedx_document = json.loads(input_path.read_text())
+
+    if errors := validate_document(cyclonedx_document):
+        error_msg = f'CycloneDX document "{input_path.name}" is not valid:\n{errors}'
+        raise ValueError(error_msg)
+
+    cyclonedx_bom = get_bom(cyclonedx_document)
+    components = get_components(cyclonedx_bom)
+
+    return [cyclonedx_component_to_package_data(component) for component in components]

scanpipe/pipes/resolve.py

@@ -24,8 +24,6 @@
 import sys
 from pathlib import Path
 
-from django.core.validators import EMPTY_VALUES
-
 from attributecode.model import About
 from packagedcode import APPLICATION_PACKAGE_DATAFILE_HANDLERS
 from packagedcode.licensing import get_license_detections_and_expression
@@ -258,56 +256,6 @@ def resolve_spdx_packages(input_location):
     ]
 
 
-def cyclonedx_component_to_package_data(component_data):
-    """Return package_data from CycloneDX component."""
-    extra_data = {}
-    component = component_data["cdx_package"]
-
-    package_url_dict = {}
-    if component.purl:
-        package_url_dict = PackageURL.from_string(component.purl).to_dict(encode=True)
-
-    declared_license = cyclonedx.get_declared_licenses(licenses=component.licenses)
-
-    if external_references := cyclonedx.get_external_references(component):
-        extra_data["externalReferences"] = external_references
-
-    if nested_components := component_data.get("nested_components"):
-        extra_data["nestedComponents"] = nested_components
-
-    package_data = {
-        "name": component.name,
-        "extracted_license_statement": declared_license,
-        "copyright": component.copyright,
-        "version": component.version,
-        "description": component.description,
-        "extra_data": extra_data,
-        **package_url_dict,
-        **cyclonedx.get_checksums(component),
-        **cyclonedx.get_properties_data(component),
-    }
-
-    return {
-        key: value for key, value in package_data.items() if value not in EMPTY_VALUES
-    }
-
-
-def resolve_cyclonedx_packages(input_location):
-    """Resolve the packages from the `input_location` CycloneDX document file."""
-    input_path = Path(input_location)
-    cyclonedx_document = json.loads(input_path.read_text())
-
-    try:
-        cyclonedx.validate_document(cyclonedx_document)
-    except Exception as e:
-        raise Exception(f'CycloneDX document "{input_path.name}" is not valid: {e}')
-
-    cyclonedx_bom = cyclonedx.get_bom(cyclonedx_document)
-    components = cyclonedx.get_components(cyclonedx_bom)
-
-    return [cyclonedx_component_to_package_data(component) for component in components]
-
-
 def get_default_package_type(input_location):
     """
     Return the package type associated with the provided `input_location`.
@@ -344,7 +292,7 @@ def get_default_package_type(input_location):
 sbom_registry = {
     "about": resolve_about_packages,
     "spdx": resolve_spdx_packages,
-    "cyclonedx": resolve_cyclonedx_packages,
+    "cyclonedx": cyclonedx.resolve_cyclonedx_packages,
 }