wip

Author: dervoeti

backend/application/vex/services/cyclonedx_parser.py

@@ -36,41 +36,35 @@ def parse_cyclonedx_data(data: dict) -> None:
 
 
 def _create_cyclonedx_document(data: dict) -> VEX_Document:
-    # Use serial number as document ID, fallback to a generated one
     document_id = data.get("serialNumber")
     if not document_id:
-        bom_format = data.get("bomFormat", "CycloneDX")
-        spec_version = data.get("specVersion", "1.0")
-        document_id = f"{bom_format}-{spec_version}-{hash(str(data))}"
+        raise ValidationError("serialNumber is missing")
 
-    version = str(data.get("version", 1))
+    version_value = data.get("version")
+    if version_value is None:
+        raise ValidationError("version is missing")
+    version = str(version_value)
 
-    # Extract metadata for author and timestamps
     metadata = data.get("metadata", {})
 
-    # Get timestamp from metadata or use current time
     timestamp = metadata.get("timestamp")
     if not timestamp:
-        # If no timestamp, use a default ISO format string
-        from datetime import datetime
-        timestamp = datetime.now().isoformat() + "Z"
-
-    # Extract author from tools or component
-    author = "Unknown"
-    tools = metadata.get("tools", [])
-    if tools:
-        if isinstance(tools, list) and tools:
-            author = tools[0].get("name", "Unknown")
-        elif isinstance(tools, dict):
-            components = tools.get("components", [])
-            if components:
-                author = components[0].get("name", "Unknown")
-
-    # If no author from tools, try to get from component
-    if author == "Unknown":
-        component = metadata.get("component", {})
-        if component:
-            author = component.get("name", "Unknown")
+        raise ValidationError("metadata/timestamp is missing")
+
+    author = None
+    # Prefer authors list if available
+    authors = metadata.get("authors")
+    if authors and isinstance(authors, list) and len(authors) > 0:
+        # Find the first author with a name set
+        author = next((item.get("name") for item in authors
+                  if isinstance(item, dict) and item.get("name")), None)
+
+    # Fall back to manufacturer or supplier if no authors
+    if not author:
+        author = metadata.get("manufacturer", {}).get("name") or metadata.get("supplier", {}).get("name")
+
+    if not author:
+        raise ValidationError("author is missing")
 
     try:
         cyclonedx_document = VEX_Document.objects.get(document_id=document_id, author=author)
@@ -98,9 +92,13 @@ def _process_vex_statements(data: dict, cyclonedx_document: VEX_Document) -> tup
     if not isinstance(vulnerabilities, list):
         raise ValidationError("vulnerabilities is not a list")
 
-    # Build components mapping
     components_map = _build_components_map(data)
 
+    product_purl = data.get("metadata", {}).get("component", {}).get("purl", "")
+    if not product_purl:
+        raise ValidationError("metadata/component/purl is missing")
+    validate_purl(product_purl)
+
     product_purls: set[str] = set()
     vex_statements: set[VEX_Statement] = set()
 
@@ -115,13 +113,12 @@ def _process_vex_statements(data: dict, cyclonedx_document: VEX_Document) -> tup
 
         analysis = vulnerability.get("analysis", {})
         if not analysis:
-            # Skip vulnerabilities without analysis (not VEX statements)
+            # Skip vulnerabilities without analysis
             vulnerability_counter += 1
             continue
 
         cyclonedx_analysis = _parse_analysis(analysis, vulnerability_counter)
 
-        # Map CycloneDX state to VEX status
         vex_status = _map_cyclonedx_state_to_vex_status(cyclonedx_analysis.state)
         if not vex_status:
             raise ValidationError(f"vulnerability[{vulnerability_counter}]/analysis/state is not valid: {cyclonedx_analysis.state}")
@@ -131,26 +128,26 @@ def _process_vex_statements(data: dict, cyclonedx_document: VEX_Document) -> tup
         if detail:
             description += f"\n\n{detail}"
 
-        # Build justification from CycloneDX justification
-        justification = _map_cyclonedx_justification_to_vex_justification(cyclonedx_analysis.justification)
-
-        # Build remediation from response and recommendation
         remediation = _build_remediation_text(cyclonedx_analysis.response, vulnerability.get("recommendation", ""))
 
-        # Process affected components
         affects = vulnerability.get("affects", [])
         if not affects:
-            # If no affects, this is a general statement - we'll need a product PURL
-            # Try to extract from metadata component
-            component = data.get("metadata", {}).get("component", {})
-            product_purl = component.get("purl", "")
-            if product_purl:
-                validate_purl(product_purl)
-                _create_vex_statement(
-                    cyclonedx_document, vulnerability_id, description, vex_status,
-                    justification, cyclonedx_analysis.detail, remediation,
-                    product_purl, "", product_purls, vex_statements
-                )
+            # General statement for the product
+            _create_vex_statement(
+                cyclonedx_document,
+                vulnerability_id,
+                description,
+                vex_status,
+                cyclonedx_analysis.justification,
+                cyclonedx_analysis.detail,
+                remediation,
+                product_purl,
+                "",
+                product_purls,
+                vex_statements,
+            )
+        elif not isinstance(affects, list):
+            raise ValidationError(f"affects[{vulnerability_counter}] is not a list")
         else:
             _process_affected_components(
                 cyclonedx_document=cyclonedx_document,
@@ -160,11 +157,12 @@ def _process_vex_statements(data: dict, cyclonedx_document: VEX_Document) -> tup
                 vulnerability_id=vulnerability_id,
                 description=description,
                 vex_status=vex_status,
-                justification=justification,
+                justification=cyclonedx_analysis.justification,
                 impact=cyclonedx_analysis.detail,
                 remediation=remediation,
                 affects=affects,
                 components_map=components_map,
+                product_purl=product_purl,
             )
 
         vulnerability_counter += 1
@@ -173,7 +171,6 @@ def _process_vex_statements(data: dict, cyclonedx_document: VEX_Document) -> tup
 
 
 def _build_components_map(data: dict) -> dict[str, dict]:
-    """Build a mapping of bom-ref to component data for quick lookup."""
     components_map = {}
 
     # Add root component from metadata
@@ -214,7 +211,6 @@ def _parse_analysis(analysis: dict, vulnerability_counter: int) -> CycloneDX_Ana
 
 
 def _map_cyclonedx_state_to_vex_status(state: str) -> Optional[str]:
-    """Map CycloneDX analysis state to VEX status."""
     mapping = {
         CycloneDX_Analysis_State.CYCLONEDX_STATE_RESOLVED: VEX_Status.VEX_STATUS_FIXED,
         CycloneDX_Analysis_State.CYCLONEDX_STATE_RESOLVED_WITH_PEDIGREE: VEX_Status.VEX_STATUS_FIXED,
@@ -226,15 +222,7 @@ def _map_cyclonedx_state_to_vex_status(state: str) -> Optional[str]:
     return mapping.get(state)
 
 
-def _map_cyclonedx_justification_to_vex_justification(justification: str) -> str:
-    """Map CycloneDX justification to VEX justification if possible."""
-    # CycloneDX doesn't have standardized justification values like OpenVEX
-    # We'll pass through the justification as is, or return empty string
-    return justification if justification else ""
-
-
 def _build_remediation_text(response: list[str], recommendation: str) -> str:
-    """Build remediation text from response actions and recommendation."""
     remediation_parts = []
 
     if response:
@@ -261,6 +249,7 @@ def _process_affected_components(
     remediation: str,
     affects: list,
     components_map: dict,
+    product_purl: str,
 ) -> None:
     affected_counter = 0
     for affected in affects:
@@ -271,24 +260,32 @@ def _process_affected_components(
         if not ref:
             raise ValidationError(f"affects[{vulnerability_counter}][{affected_counter}]/ref is missing")
 
-        # Look up component by bom-ref
         component = components_map.get(ref)
         if not component:
-            # Skip if we can't find the component
-            affected_counter += 1
-            continue
+            raise ValidationError(
+                f"affects[{vulnerability_counter}][{affected_counter}]/ref '{ref}' not found in components"
+            )
 
-        # Extract PURL from component
         component_purl = component.get("purl", "")
-        if component_purl:
-            validate_purl(component_purl)
-
-            # For affected components, we'll use the component PURL as both product and component
-            _create_vex_statement(
-                cyclonedx_document, vulnerability_id, description, vex_status,
-                justification, impact, remediation,
-                component_purl, component_purl, product_purls, vex_statements
+        if not component_purl:
+            raise ValidationError(
+                f"affects[{vulnerability_counter}][{affected_counter}]/ref '{ref}' component is missing purl"
             )
+        validate_purl(component_purl)
+
+        _create_vex_statement(
+            cyclonedx_document,
+            vulnerability_id,
+            description,
+            vex_status,
+            justification,
+            impact,
+            remediation,
+            product_purl,
+            component_purl,
+            product_purls,
+            vex_statements,
+        )
 
         affected_counter += 1
 
@@ -306,7 +303,6 @@ def _create_vex_statement(
     product_purls: set,
     vex_statements: set,
 ) -> None:
-    """Create and save a VEX statement."""
     vex_statement = VEX_Statement(
         document=cyclonedx_document,
         vulnerability_id=vulnerability_id,