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,355 @@
+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 = []
+
+
+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 = _extract_vulnerabilities(data)
+    components_map = _build_components_map(data)
+    product_purl = _extract_product_purl(data)
+
+    product_purls: set[str] = set()
+    vex_statements: set[VEX_Statement] = set()
+
+    for vulnerability_counter, vulnerability in enumerate(vulnerabilities):
+        context = _prepare_vulnerability_context(vulnerability, vulnerability_counter)
+        if context is None:
+            # skip vulnerabilities without analysis
+            continue
+
+        if not context["affects"]:
+            statement = _create_general_vex_statement(
+                cyclonedx_document=cyclonedx_document,
+                product_purl=product_purl,
+                context=context,
+            )
+            vex_statements.add(statement)
+            product_purls.add(product_purl)
+            continue
+
+        if not isinstance(context["affects"], list):
+            raise ValidationError(f"affects[{vulnerability_counter}] is not a list")
+
+        statements = _create_component_vex_statements(
+            cyclonedx_document=cyclonedx_document,
+            context=context,
+            components_map=components_map,
+            product_purl=product_purl,
+            vulnerability_counter=vulnerability_counter,
+        )
+        vex_statements.update(statements)
+        if statements:
+            product_purls.add(product_purl)
+
+    return product_purls, vex_statements
+
+
+def _extract_vulnerabilities(data: dict) -> list[dict]:
+    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")
+    return vulnerabilities
+
+
+def _extract_product_purl(data: dict) -> str:
+    product_purl = data.get("metadata", {}).get("component", {}).get("purl", "")
+    if not product_purl:
+        raise ValidationError("metadata/component/purl is missing")
+    validate_purl(product_purl)
+    return product_purl
+
+
+def _prepare_vulnerability_context(vulnerability: dict, vulnerability_counter: int) -> Optional[dict]:
+    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:
+        return None
+
+    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", [])
+
+    return {
+        "vulnerability_id": vulnerability_id,
+        "cyclonedx_analysis": cyclonedx_analysis,
+        "vex_status": vex_status,
+        "description": description,
+        "remediation": remediation,
+        "affects": affects,
+    }
+
+
+def _create_general_vex_statement(
+    *, cyclonedx_document: VEX_Document, product_purl: str, context: dict
+) -> VEX_Statement:
+    statement = VEX_Statement(
+        document=cyclonedx_document,
+        vulnerability_id=context["vulnerability_id"],
+        description=context["description"],
+        status=context["vex_status"],
+        justification=context["cyclonedx_analysis"].justification,
+        detail=context["cyclonedx_analysis"].detail,
+        remediation=context["remediation"],
+        product_purl=product_purl,
+        component_purl="",
+    )
+    statement.save()
+    return statement
+
+
+def _create_component_vex_statements(
+    *,
+    cyclonedx_document: VEX_Document,
+    context: dict,
+    components_map: dict[str, dict],
+    product_purl: str,
+    vulnerability_counter: int,
+) -> set[VEX_Statement]:
+    statements: set[VEX_Statement] = set()
+    for affected_counter, affected in enumerate(context["affects"]):
+        component_purl = _get_component_purl_from_affect(
+            affected=affected,
+            components_map=components_map,
+            vulnerability_counter=vulnerability_counter,
+            affected_counter=affected_counter,
+        )
+
+        statement = VEX_Statement(
+            document=cyclonedx_document,
+            vulnerability_id=context["vulnerability_id"],
+            description=context["description"],
+            status=context["vex_status"],
+            justification=context["cyclonedx_analysis"].justification,
+            detail=context["cyclonedx_analysis"].detail,
+            remediation=context["remediation"],
+            product_purl=product_purl,
+            component_purl=component_purl,
+        )
+        statement.save()
+        statements.add(statement)
+    return statements
+
+
+def _get_component_purl_from_affect(
+    *,
+    affected: dict,
+    components_map: dict[str, dict],
+    vulnerability_counter: int,
+    affected_counter: int,
+) -> str:
+    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)
+    return component_purl
+
+
+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)