feat(finding): multiple CWEs per finding

docs/content/releases/os_upgrading/3.2.md

@@ -2,33 +2,58 @@
 title: 'Upgrading to DefectDojo Version 3.2.x'
 toc_hide: true
 weight: -20260701
-description: Vulnerability ids gain an autodetected type. A migration adds the type column, de-duplicates vulnerability-id rows, and adds a uniqueness constraint. Existing hash codes are unaffected.
+description: Vulnerability ids gain an autodetected type and a uniqueness constraint; findings can now carry multiple CWEs via a new Finding_CWE relationship. Migrations add the type column, de-duplicate vulnerability-id rows, add the uniqueness constraint, create the CWE table, and backfill it. Existing hash codes are unaffected.
 ---
 
 ## Vulnerability id type
 
 Each `Vulnerability_Id` gains an autodetected `vulnerability_id_type` — the identifier's leading
 prefix (`CVE-2024-1234` → `CVE`, `GHSA-…` → `GHSA`, `RUSTSEC-…` → `RUSTSEC`). It is derived
 structurally (no registry) and stored (indexed) so identifiers can be filtered and grouped by type
-efficiently. It is `NULL` when there is no non-numeric prefix.
+efficiently. It is `NULL` when there is no non-numeric prefix. It is populated automatically on
+import and on `save()`; existing rows are backfilled by migration. It does not participate in
+`hash_code`, so **existing hash codes and deduplication are unaffected**.
 
-The type is populated automatically: on import (bulk paths) and on `save()`. Existing rows are
-backfilled by the migration below. This is a denormalized, derived attribute — it does not
-participate in `hash_code`, so **existing hash codes and deduplication are unaffected**.
+A unique constraint is also added on `(finding, vulnerability_id)`; pre-existing duplicate rows
+(unintended) are consolidated first.
+
+## Multiple CWEs per finding
+
+A finding could previously store only one CWE (the integer `cwe` field). This release adds a
+dedicated `Finding_CWE` relationship so a finding can carry **multiple CWEs**, using the same
+approach as vulnerability ids: the primary CWE stays on `Finding.cwe` (unchanged — legacy
+deduplication and hash codes still use it), and additional CWEs live in the relationship.
+
+CWE is modeled separately from vulnerability identifiers on purpose: a CWE is a weakness *class*,
+not a vulnerability *instance* identifier, so it must not participate in `hash_code`,
+vulnerability-id deduplication, or the `cve` field. Because of this separation, **existing hash
+codes and deduplication are unaffected**.
+
+CWEs are populated automatically on import and when a finding is created or edited (from the
+finding's CWE field, plus any additional CWEs a parser supplies). The finding exposes them via
+`finding.cwes` (primary first, deduplicated).
 
 ## Database migration
 
-Three migrations run automatically on upgrade:
+Five migrations run automatically on upgrade:
 
-- `0276_vulnerability_id_type` — adds the indexed `vulnerability_id_type` column and a leading
+- `0280_vulnerability_id_type` — adds the indexed `vulnerability_id_type` column and a leading
   index on `vulnerability_id`.
-- `0277_backfill_vulnerability_id_type` — backfills `vulnerability_id_type` for existing rows and
-  removes duplicate `(finding, vulnerability_id)` rows, keeping the earliest of each — such
-  duplicates are unintended, and consolidating them allows a uniqueness constraint to be added.
-- `0278_unique_finding_vulnerability_id` — adds a unique constraint on `(finding, vulnerability_id)`.
+- `0281_backfill_vulnerability_id_type` — backfills `vulnerability_id_type` and removes duplicate
+  `(finding, vulnerability_id)` rows (keeping the earliest).
+- `0282_unique_finding_vulnerability_id` — adds the unique constraint on `(finding, vulnerability_id)`.
+- `0283_finding_cwe` — creates the `Finding_CWE` table (unique per `(finding, cwe)`).
+- `0284_backfill_finding_cwe` — seeds `Finding_CWE` rows for all existing findings from their
+  legacy `Finding.cwe` values.
 
 ### What you need to do
 
-The migrations are applied automatically. No manual steps are required.
+Nothing — the migrations are applied automatically, `0284_backfill_finding_cwe` backfills every
+existing finding, and new/edited findings populate their CWE relationship on save and import.
+
+The `manage.py migrate_cwe` management command is **optional** and is *not* required on upgrade
+(it would repeat the work `0284` already did). It exists only as an idempotent re-scan you can run
+later — for example after a parser upgrade changes how a scan's CWEs are derived — to reconcile
+`Finding_CWE` rows from the current `Finding.cwe` values.
 
 For more information, check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/3.2.0).

dojo/db_migrations/0283_finding_cwe.py

@@ -0,0 +1,28 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    """Schema only: create the Finding_CWE relationship (multiple CWEs per finding). The data
+    backfill from the legacy Finding.cwe field is kept in a separate migration (0280) so data
+    migrations are never mixed with schema migrations."""
+
+    dependencies = [
+        ("dojo", "0282_unique_finding_vulnerability_id"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Finding_CWE",
+            fields=[
+                ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("cwe", models.CharField(db_index=True, max_length=11)),
+                ("finding", models.ForeignKey(editable=False, on_delete=django.db.models.deletion.CASCADE, to="dojo.finding")),
+            ],
+        ),
+        migrations.AddConstraint(
+            model_name="finding_cwe",
+            constraint=models.UniqueConstraint(fields=("finding", "cwe"), name="unique_finding_cwe"),
+        ),
+    ]

dojo/db_migrations/0284_backfill_finding_cwe.py

@@ -0,0 +1,36 @@
+from django.db import migrations
+
+from dojo.finding.cwe import cwe_label
+
+BATCH_SIZE = 1000
+
+
+def create_finding_cwe_records(apps, schema_editor):
+    """Backfill Finding_CWE rows (canonical CWE-<n>) from the legacy int Finding.cwe field."""
+    Finding = apps.get_model("dojo", "Finding")
+    Finding_CWE = apps.get_model("dojo", "Finding_CWE")
+    batch = []
+    for finding in Finding.objects.filter(cwe__gt=0).only("id", "cwe").iterator(chunk_size=BATCH_SIZE):
+        label = cwe_label(finding.cwe)
+        if label is None:
+            continue
+        batch.append(Finding_CWE(finding_id=finding.id, cwe=label))
+        if len(batch) >= BATCH_SIZE:
+            Finding_CWE.objects.bulk_create(batch, batch_size=BATCH_SIZE, ignore_conflicts=True)
+            batch = []
+    if batch:
+        Finding_CWE.objects.bulk_create(batch, batch_size=BATCH_SIZE, ignore_conflicts=True)
+
+
+class Migration(migrations.Migration):
+
+    """Data only (no schema changes): create the initial Finding_CWE rows from the legacy
+    Finding.cwe field."""
+
+    dependencies = [
+        ("dojo", "0283_finding_cwe"),
+    ]
+
+    operations = [
+        migrations.RunPython(create_finding_cwe_records, migrations.RunPython.noop),
+    ]

dojo/finding/api/serializer.py

@@ -15,7 +15,9 @@
 import dojo.finding.helper as finding_helper
 from dojo.authorization.authorization import user_has_permission
 from dojo.celery_dispatch import dojo_dispatch_task
+from dojo.finding.cwe import cwe_label, cwe_number
 from dojo.finding.helper import (
+    save_cwes,
     save_endpoints_template,
     save_vulnerability_ids,
     save_vulnerability_ids_template,
@@ -32,6 +34,7 @@
     Endpoint,
     Engagement,
     Finding,
+    Finding_CWE,
     Finding_Group,
     Finding_Template,
     Note_Type,
@@ -298,6 +301,30 @@ class Meta:
         fields = ["vulnerability_id"]
 
 
+@extend_schema_field(serializers.CharField())
+class CweField(serializers.Field):
+
+    """Serialize a CWE as the canonical ``CWE-<n>`` string; accept ``"CWE-79"`` or ``"79"`` on write."""
+
+    def to_representation(self, value):
+        return cwe_label(value) or value
+
+    def to_internal_value(self, data):
+        label = cwe_label(data)
+        if label is None:
+            msg = "Enter a CWE number, e.g. 89 or CWE-89."
+            raise serializers.ValidationError(msg)
+        return label
+
+
+class FindingCweSerializer(serializers.ModelSerializer):
+    cwe = CweField()
+
+    class Meta:
+        model = Finding_CWE
+        fields = ["cwe"]
+
+
 class FindingSerializer(serializers.ModelSerializer):
     mitigated = serializers.DateTimeField(required=False, allow_null=True)
     mitigated_by = serializers.PrimaryKeyRelatedField(required=False, allow_null=True, queryset=User.objects.all())
@@ -321,6 +348,9 @@ class FindingSerializer(serializers.ModelSerializer):
     vulnerability_ids = VulnerabilityIdSerializer(
         source="vulnerability_id_set", many=True, required=False,
     )
+    cwes = FindingCweSerializer(
+        source="finding_cwe_set", many=True, required=False,
+    )
     reporter = serializers.PrimaryKeyRelatedField(
         required=False, queryset=User.objects.all(),
     )
@@ -417,6 +447,13 @@ def update(self, instance, validated_data):
             logger.debug("SETTING CVE FROM VULNERABILITY_ID_SET: %s", parsed_vulnerability_ids[0])
             validated_data["cve"] = parsed_vulnerability_ids[0]
 
+        # CWEs (mirror vulnerability_ids): the first entry is the primary Finding.cwe; the rest
+        # become Finding_CWE rows via save_cwes() below.
+        parsed_cwes = None
+        if (cwes := validated_data.pop("finding_cwe_set", None)) is not None:
+            parsed_cwes = [entry["cwe"] for entry in cwes]
+            validated_data["cwe"] = cwe_number(parsed_cwes[0]) if parsed_cwes else 0
+
         # Save the reporter on the finding
         if reporter_id := validated_data.get("reporter"):
             instance.reporter = reporter_id
@@ -445,6 +482,14 @@ def update(self, instance, validated_data):
             instance, validated_data,
         )
 
+        # Sync the CWE relation (separate from vulnerability ids) after the new cwe is applied.
+        # Only touch the CWE rows when the request actually carried `cwes`; calling save_cwes()
+        # unconditionally would wipe a finding's extra Finding_CWE rows on any partial PATCH that
+        # omits the field (mirrors the guarded vulnerability-id path above).
+        if parsed_cwes is not None:
+            instance.unsaved_cwes = parsed_cwes[1:]
+            save_cwes(instance)
+
         if settings.V3_FEATURE_LOCATIONS and locations is not None:
             for location_ref in instance.locations.all():
                 location_ref.location.disassociate_from_finding(instance)
@@ -561,6 +606,9 @@ class FindingCreateSerializer(serializers.ModelSerializer):
     vulnerability_ids = VulnerabilityIdSerializer(
         source="vulnerability_id_set", many=True, required=False,
     )
+    cwes = FindingCweSerializer(
+        source="finding_cwe_set", many=True, required=False,
+    )
     reporter = serializers.PrimaryKeyRelatedField(
         required=False, queryset=User.objects.all(),
     )
@@ -601,6 +649,12 @@ def create(self, validated_data):
             validated_data["cve"] = parsed_vulnerability_ids[0]
             # validated_data["unsaved_vulnerability_ids"] = parsed_vulnerability_ids
 
+        # CWEs (mirror vulnerability_ids): first entry is the primary cwe, the rest are extras.
+        parsed_cwes = None
+        if (cwes := validated_data.pop("finding_cwe_set", None)) is not None:
+            parsed_cwes = [entry["cwe"] for entry in cwes]
+            validated_data["cwe"] = cwe_number(parsed_cwes[0]) if parsed_cwes else 0
+
         # super.create() doesn't accept unsaved_vulnerability_ids or dedupe_option=False, so call save directly.
         new_finding = Finding(**validated_data)
         new_finding.unsaved_vulnerability_ids = parsed_vulnerability_ids or []
@@ -617,6 +671,9 @@ def create(self, validated_data):
             new_finding.reviewers.set(reviewers)
         if parsed_vulnerability_ids:
             save_vulnerability_ids(new_finding, parsed_vulnerability_ids)
+        if parsed_cwes is not None:
+            new_finding.unsaved_cwes = parsed_cwes[1:]
+        save_cwes(new_finding)
 
         if push_to_jira:
             jira_services.push(new_finding)

dojo/finding/cwe.py

@@ -0,0 +1,53 @@
+"""Pure helpers for CWE identifiers (no model imports, safe to import anywhere)."""
+
+
+def cwe_number(value) -> int | None:
+    """
+    ``"CWE-79"`` / ``"79"`` / ``79`` -> ``79`` (positive int, case-insensitive); anything else -> ``None``.
+
+    CWE numbers are positive, so ``0`` (Finding.cwe's "unset" sentinel) and non-numeric input -> ``None``.
+    """
+    if value is None:
+        return None
+    token = str(value).strip().upper().removeprefix("CWE-")
+    if not token.isdigit():
+        return None
+    number = int(token)
+    # Finding_CWE.cwe stores the "CWE-<n>" label in a varchar(11), so n is bounded to 7 digits.
+    # Reject anything larger here (clean None -> 400/skip) instead of 500-ing at insert.
+    if number <= 0 or number > 9_999_999:
+        return None
+    return number
+
+
+def cwe_label(value) -> str | None:
+    """``79`` / ``"79"`` / ``"CWE-79"`` -> ``"CWE-79"`` (canonical, case-insensitive); invalid -> ``None``."""
+    number = cwe_number(value)
+    return f"CWE-{number}" if number is not None else None
+
+
+def parse_cwes(text: str | None) -> list[str]:
+    """
+    Parse CWEs from user text (one per line or comma-separated) into canonical ``CWE-<n>`` labels.
+
+    Accepts ``89`` or ``CWE-89`` (case-insensitive); ignores anything non-numeric; deduplicates.
+    """
+    result: list[str] = []
+    for token in (text or "").replace(",", "\n").split():
+        label = cwe_label(token)
+        if label is not None and label not in result:
+            result.append(label)
+    return result
+
+
+def finding_cwe_labels(cwe, unsaved_cwes=None) -> list[str]:
+    """Canonical ``CWE-<n>`` labels for a finding: the primary ``cwe`` first, then extras, deduplicated."""
+    labels: list[str] = []
+    primary = cwe_label(cwe)
+    if primary is not None:
+        labels.append(primary)
+    for extra in unsaved_cwes or []:
+        label = cwe_label(extra)
+        if label is not None:
+            labels.append(label)
+    return list(dict.fromkeys(labels))

dojo/finding/deduplication.py

@@ -340,11 +340,11 @@ def build_candidate_scope_queryset(test, mode="deduplication", service=None):
         queryset = Finding.objects.filter(scope_q)
 
     if settings.V3_FEATURE_LOCATIONS:
-        prefetch_list = ["locations__location__url", "vulnerability_id_set", "found_by"]
+        prefetch_list = ["locations__location__url", "vulnerability_id_set", "finding_cwe_set", "found_by"]
     else:
         # TODO: Delete this after the move to Locations
         # Base prefetches for both modes
-        prefetch_list = ["endpoints", "vulnerability_id_set", "found_by"]
+        prefetch_list = ["endpoints", "vulnerability_id_set", "finding_cwe_set", "found_by"]
 
         # Prefetch all endpoint statuses with their endpoint for reimport mode.
         # The non-special filtering (excluding false_positive, out_of_scope, risk_accepted)

dojo/finding/helper.py

@@ -19,6 +19,7 @@
 from dojo.celery import app
 from dojo.endpoint.utils import endpoint_get_or_create, save_endpoints_to_add
 from dojo.file_uploads.helper import delete_related_files
+from dojo.finding.cwe import finding_cwe_labels
 from dojo.finding.deduplication import (
     dedupe_batch_of_findings,
     do_dedupe_finding_task_internal,
@@ -37,6 +38,7 @@
     Engagement,
     FileUpload,
     Finding,
+    Finding_CWE,
     Finding_Group,
     JIRA_Instance,
     Notes,
@@ -1080,6 +1082,24 @@ def save_vulnerability_ids(finding, vulnerability_ids, *, delete_existing: bool
         finding.cve = None
 
 
+def save_cwes(finding, *, delete_existing: bool = True):
+    """
+    Persist the finding's CWEs as Finding_CWE rows.
+
+    The primary Finding.cwe plus any parser-supplied unsaved_cwes, stored as canonical CWE-<n>
+    strings. CWE is a weakness class, kept separate from vulnerability ids.
+    """
+    cwe_values = finding_cwe_labels(finding.cwe, getattr(finding, "unsaved_cwes", None))
+
+    if delete_existing:
+        Finding_CWE.objects.filter(finding=finding).delete()
+
+    Finding_CWE.objects.bulk_create(
+        [Finding_CWE(finding=finding, cwe=cwe) for cwe in cwe_values],
+        ignore_conflicts=True,
+    )
+
+
 def save_vulnerability_ids_template(finding_template, vulnerability_ids):
     """Save vulnerability IDs as newline-separated string in TextField."""
     # Remove duplicates and empty strings