feat: cdx vex

Author: dervoeti

backend/application/vex/migrations/0008_alter_vex_document_type.py

@@ -0,0 +1,20 @@
+# Generated by Django 5.2.4 on 2025-07-30 12:57
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vex", "0007_alter_csaf_tracking_current_release_date_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="vex_document",
+            name="type",
+            field=models.CharField(
+                choices=[("CSAF", "CSAF"), ("OpenVEX", "OpenVEX"), ("CycloneDX", "CycloneDX")], max_length=16
+            ),
+        ),
+    ]

backend/application/vex/services/cyclonedx_parser.py

@@ -0,0 +1,323 @@
+from dataclasses import dataclass
+from typing import Optional
+
+from rest_framework.exceptions import ValidationError
+
+from application.core.api.serializers_helpers import validate_purl
+from application.vex.models import VEX_Document, VEX_Statement
+from application.vex.services.vex_engine import apply_vex_statements_after_import
+from application.vex.types import (
+    CycloneDX_Analysis_State,
+    VEX_Document_Type,
+    VEX_Status,
+)
+
+
+@dataclass
+class CycloneDX_Analysis:
+    state: str = ""
+    justification: str = ""
+    response: list[str] = None
+    detail: str = ""
+    first_issued: str = ""
+    last_updated: str = ""
+
+    def __post_init__(self):
+        if self.response is None:
+            self.response = []
+
+
+def parse_cyclonedx_data(data: dict) -> None:
+    cyclonedx_document = _create_cyclonedx_document(data)
+
+    product_purls, vex_statements = _process_vex_statements(data, cyclonedx_document)
+
+    apply_vex_statements_after_import(product_purls, vex_statements)
+
+
+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))}"
+
+    version = str(data.get("version", 1))
+
+    # 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")
+
+    try:
+        cyclonedx_document = VEX_Document.objects.get(document_id=document_id, author=author)
+        cyclonedx_document.delete()
+    except VEX_Document.DoesNotExist:
+        pass
+
+    cyclonedx_document = VEX_Document.objects.create(
+        type=VEX_Document_Type.VEX_DOCUMENT_TYPE_CYCLONEDX,
+        document_id=document_id,
+        version=version,
+        initial_release_date=timestamp,
+        current_release_date=timestamp,
+        author=author,
+        role="",  # CycloneDX doesn't have explicit roles
+    )
+
+    return cyclonedx_document
+
+
+def _process_vex_statements(data: dict, cyclonedx_document: VEX_Document) -> tuple[set[str], set[VEX_Statement]]:
+    vulnerabilities = data.get("vulnerabilities", [])
+    if not vulnerabilities:
+        raise ValidationError("CycloneDX document doesn't contain any vulnerabilities")
+    if not isinstance(vulnerabilities, list):
+        raise ValidationError("vulnerabilities is not a list")
+
+    # Build components mapping
+    components_map = _build_components_map(data)
+
+    product_purls: set[str] = set()
+    vex_statements: set[VEX_Statement] = set()
+
+    vulnerability_counter = 0
+    for vulnerability in vulnerabilities:
+        if not isinstance(vulnerability, dict):
+            raise ValidationError(f"vulnerability[{vulnerability_counter}] is not a dictionary")
+
+        vulnerability_id = vulnerability.get("id")
+        if not vulnerability_id:
+            raise ValidationError(f"vulnerability[{vulnerability_counter}]/id is missing")
+
+        analysis = vulnerability.get("analysis", {})
+        if not analysis:
+            # Skip vulnerabilities without analysis (not VEX statements)
+            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}")
+
+        description = vulnerability.get("description", "")
+        detail = vulnerability.get("detail", "")
+        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
+                )
+        else:
+            _process_affected_components(
+                cyclonedx_document=cyclonedx_document,
+                product_purls=product_purls,
+                vex_statements=vex_statements,
+                vulnerability_counter=vulnerability_counter,
+                vulnerability_id=vulnerability_id,
+                description=description,
+                vex_status=vex_status,
+                justification=justification,
+                impact=cyclonedx_analysis.detail,
+                remediation=remediation,
+                affects=affects,
+                components_map=components_map,
+            )
+
+        vulnerability_counter += 1
+
+    return product_purls, vex_statements
+
+
+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
+    metadata_component = data.get("metadata", {}).get("component")
+    if metadata_component and metadata_component.get("bom-ref"):
+        components_map[metadata_component["bom-ref"]] = metadata_component
+
+    # Add all components
+    for component in data.get("components", []):
+        if component.get("bom-ref"):
+            components_map[component["bom-ref"]] = component
+
+    return components_map
+
+
+def _parse_analysis(analysis: dict, vulnerability_counter: int) -> CycloneDX_Analysis:
+    state = analysis.get("state", "")
+    if not state:
+        raise ValidationError(f"vulnerability[{vulnerability_counter}]/analysis/state is missing")
+
+    justification = analysis.get("justification", "")
+    response = analysis.get("response", [])
+    if not isinstance(response, list):
+        response = []
+
+    detail = analysis.get("detail", "")
+    first_issued = analysis.get("firstIssued", "")
+    last_updated = analysis.get("lastUpdated", "")
+
+    return CycloneDX_Analysis(
+        state=state,
+        justification=justification,
+        response=response,
+        detail=detail,
+        first_issued=first_issued,
+        last_updated=last_updated,
+    )
+
+
+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,
+        CycloneDX_Analysis_State.CYCLONEDX_STATE_EXPLOITABLE: VEX_Status.VEX_STATUS_AFFECTED,
+        CycloneDX_Analysis_State.CYCLONEDX_STATE_IN_TRIAGE: VEX_Status.VEX_STATUS_UNDER_INVESTIGATION,
+        CycloneDX_Analysis_State.CYCLONEDX_STATE_FALSE_POSITIVE: VEX_Status.VEX_STATUS_NOT_AFFECTED,
+        CycloneDX_Analysis_State.CYCLONEDX_STATE_NOT_AFFECTED: VEX_Status.VEX_STATUS_NOT_AFFECTED,
+    }
+    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:
+        response_text = ", ".join(response)
+        remediation_parts.append(f"Response: {response_text}")
+
+    if recommendation:
+        remediation_parts.append(f"Recommendation: {recommendation}")
+
+    return "; ".join(remediation_parts)
+
+
+def _process_affected_components(
+    *,
+    cyclonedx_document: VEX_Document,
+    product_purls: set,
+    vex_statements: set,
+    vulnerability_counter: int,
+    vulnerability_id: str,
+    description: str,
+    vex_status: str,
+    justification: str,
+    impact: str,
+    remediation: str,
+    affects: list,
+    components_map: dict,
+) -> None:
+    affected_counter = 0
+    for affected in affects:
+        if not isinstance(affected, dict):
+            raise ValidationError(f"affects[{vulnerability_counter}][{affected_counter}] is not a dictionary")
+
+        ref = affected.get("ref")
+        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
+
+        # 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
+            )
+
+        affected_counter += 1
+
+
+def _create_vex_statement(
+    cyclonedx_document: VEX_Document,
+    vulnerability_id: str,
+    description: str,
+    vex_status: str,
+    justification: str,
+    impact: str,
+    remediation: str,
+    product_purl: str,
+    component_purl: str,
+    product_purls: set,
+    vex_statements: set,
+) -> None:
+    """Create and save a VEX statement."""
+    vex_statement = VEX_Statement(
+        document=cyclonedx_document,
+        vulnerability_id=vulnerability_id,
+        description=description,
+        status=vex_status,
+        justification=justification,
+        impact=impact,
+        remediation=remediation,
+        product_purl=product_purl,
+        component_purl=component_purl,
+    )
+    vex_statement.save()
+    vex_statements.add(vex_statement)
+    product_purls.add(product_purl)

backend/application/vex/services/openvex_parser.py

@@ -82,7 +82,7 @@ def _process_vex_statements(data: dict, openvex_document: VEX_Document) -> tuple
         openvex_statement.vulnerability_id = statement.get("vulnerability", {}).get("name")
         if not openvex_statement.vulnerability_id:
             raise ValidationError(f"vulnerability[{statement_counter}]/name is missing")
-        openvex_statement.description = statement.get("vulnerability", {}).get("description")
+        openvex_statement.description = statement.get("vulnerability", {}).get("description", "")
         openvex_statement.status = statement.get("status", "")
         if not openvex_statement.status:
             raise ValidationError(f"status[{statement_counter}] is missing")

backend/application/vex/services/vex_import.py

@@ -5,6 +5,7 @@
 from rest_framework.exceptions import ValidationError
 
 from application.vex.services.csaf_parser import parse_csaf_data
+from application.vex.services.cyclonedx_parser import parse_cyclonedx_data
 from application.vex.services.openvex_parser import parse_openvex_data
 from application.vex.types import VEX_Document_Type
 
@@ -22,6 +23,8 @@ def import_vex(vex_file: File) -> None:
         parse_openvex_data(data)
     elif vex_type == VEX_Document_Type.VEX_DOCUMENT_TYPE_CSAF:
         parse_csaf_data(data)
+    elif vex_type == VEX_Document_Type.VEX_DOCUMENT_TYPE_CYCLONEDX:
+        parse_cyclonedx_data(data)
 
 
 def _get_json_data(vex_file: File) -> Optional[dict]:
@@ -40,4 +43,7 @@ def _get_vex_type(data: dict) -> Optional[str]:
     if data.get("document", {}).get("category") == "csaf_vex" and data.get("document", {}).get("csaf_version") == "2.0":
         return VEX_Document_Type.VEX_DOCUMENT_TYPE_CSAF
 
+    if data.get("bomFormat") == "CycloneDX":
+        return VEX_Document_Type.VEX_DOCUMENT_TYPE_CYCLONEDX
+
     return None