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,356 @@
+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_Justification,
+    VEX_Status,
+)
+
+
+@dataclass
+class CycloneDX_Analysis:
+    state: str = ""
+    justification: str = ""
+    response: Optional[list[str]] = None
+    detail: str = ""
+    first_issued: str = ""
+    last_updated: str = ""
+
+    def __post_init__(self) -> None:
+        if self.response is None:
+            self.response = []
+
+
+@dataclass
+class VexStatementData:
+    vulnerability_id: str
+    description: str
+    status: str
+    justification: str
+    impact: str
+    remediation: str
+    product_purl: str
+    component_purl: str = ""
+
+
+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:
+    document_id = data.get("serialNumber")
+    if not document_id:
+        raise ValidationError("serialNumber is missing")
+
+    version_value = data.get("version")
+    if version_value is None:
+        raise ValidationError("version is missing")
+    version = str(version_value)
+
+    metadata = data.get("metadata", {})
+
+    timestamp = metadata.get("timestamp")
+    if not timestamp:
+        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")
+
+    # Fall back to tools name if available
+    if not author:
+        tools = metadata.get("tools")
+        if isinstance(tools, list) and len(tools) > 0:
+            first_tool = tools[0]
+            if isinstance(first_tool, dict):
+                author = first_tool.get("name") or author
+        elif isinstance(tools, dict):
+            author = tools.get("name") or author
+
+    if not author:
+        raise ValidationError("author is missing")
+
+    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="",
+    )
+
+    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")
+
+    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()
+
+    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
+            vulnerability_counter += 1
+            continue
+
+        cyclonedx_analysis = _parse_analysis(analysis, vulnerability_counter)
+
+        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}"
+
+        remediation = _build_remediation_text(cyclonedx_analysis.response, vulnerability.get("recommendation", ""))
+
+        affects = vulnerability.get("affects", [])
+        if not affects:
+            # General statement for the product
+            _create_vex_statement(
+                cyclonedx_document,
+                product_purls,
+                vex_statements,
+                data=VexStatementData(
+                    vulnerability_id=vulnerability_id,
+                    description=description,
+                    status=vex_status,
+                    justification=cyclonedx_analysis.justification,
+                    impact=cyclonedx_analysis.detail,
+                    remediation=remediation,
+                    product_purl=product_purl,
+                    component_purl="",
+                ),
+            )
+        elif not isinstance(affects, list):
+            raise ValidationError(f"affects[{vulnerability_counter}] is not a list")
+        else:
+            _process_affected_components(
+                document=cyclonedx_document,
+                product_purls=product_purls,
+                vex_statements=vex_statements,
+                vulnerability_counter=vulnerability_counter,
+                base_data=VexStatementData(
+                    vulnerability_id=vulnerability_id,
+                    description=description,
+                    status=vex_status,
+                    justification=cyclonedx_analysis.justification,
+                    impact=cyclonedx_analysis.detail,
+                    remediation=remediation,
+                    product_purl=product_purl,
+                ),
+                affects=affects,
+                components_map=components_map,
+            )
+
+        vulnerability_counter += 1
+
+    return product_purls, vex_statements
+
+
+def _build_components_map(data: dict) -> dict[str, dict]:
+    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", "")
+    if justification:
+        justification = _map_cyclonedx_justification_to_vex_justification(justification) or ""
+    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]:
+    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 _build_remediation_text(response: Optional[list[str]], recommendation: str) -> str:
+    remediation_parts = []
+
+    if response:
+        response_text = ", ".join(response)
+        remediation_parts.append(f"Response: {response_text}")
+
+    if recommendation:
+        remediation_parts.append(recommendation)
+
+    return "; ".join(remediation_parts)
+
+
+def _map_cyclonedx_justification_to_vex_justification(justification: str) -> Optional[str]:
+    mapping = {
+        "code_not_present": VEX_Justification.STATUS_VULNERABLE_CODE_NOT_PRESENT,
+        "code_not_reachable": VEX_Justification.STATUS_VULNERABLE_CODE_NOT_IN_EXECUTE_PATH,
+        "requires_configuration": VEX_Justification.STATUS_VULNERABLE_CODE_CANNOT_BE_CONTROLLED_BY_ADVERSARY,
+        "requires_dependency": VEX_Justification.STATUS_VULNERABLE_CODE_CANNOT_BE_CONTROLLED_BY_ADVERSARY,
+        "requires_environment": VEX_Justification.STATUS_VULNERABLE_CODE_CANNOT_BE_CONTROLLED_BY_ADVERSARY,
+        "protected_by_compiler": VEX_Justification.STATUS_INLINE_MITIGATIONS_ALREADY_EXIST,
+        "protected_at_runtime": VEX_Justification.STATUS_INLINE_MITIGATIONS_ALREADY_EXIST,
+        "protected_at_perimeter": VEX_Justification.STATUS_INLINE_MITIGATIONS_ALREADY_EXIST,
+        "protected_by_mitigating_control": VEX_Justification.STATUS_INLINE_MITIGATIONS_ALREADY_EXIST,
+    }
+    return mapping.get(justification)
+
+
+def _process_affected_components(
+    *,
+    document: VEX_Document,
+    product_purls: set[str],
+    vex_statements: set[VEX_Statement],
+    vulnerability_counter: int,
+    base_data: VexStatementData,
+    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")
+
+        component = components_map.get(ref)
+        if not component:
+            raise ValidationError(
+                f"affects[{vulnerability_counter}][{affected_counter}]/ref '{ref}' not found in components"
+            )
+
+        component_purl = component.get("purl", "")
+        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(
+            document,
+            product_purls,
+            vex_statements,
+            data=VexStatementData(
+                vulnerability_id=base_data.vulnerability_id,
+                description=base_data.description,
+                status=base_data.status,
+                justification=base_data.justification,
+                impact=base_data.impact,
+                remediation=base_data.remediation,
+                product_purl=base_data.product_purl,
+                component_purl=component_purl,
+            ),
+        )
+
+        affected_counter += 1
+
+
+def _create_vex_statement(
+    document: VEX_Document,
+    product_purls: set[str],
+    vex_statements: set[VEX_Statement],
+    data: VexStatementData,
+) -> None:
+    vex_statement = VEX_Statement(
+        document=document,
+        vulnerability_id=data.vulnerability_id,
+        description=data.description,
+        status=data.status,
+        justification=data.justification,
+        impact=data.impact,
+        remediation=data.remediation,
+        product_purl=data.product_purl,
+        component_purl=data.component_purl,
+    )
+    vex_statement.save()
+    vex_statements.add(vex_statement)
+    product_purls.add(data.product_purl)