Merge branch 'dev' into valentijnscholten-patch-6

Author: valentijnscholten

.github/workflows/gh-pages.yml

@@ -46,13 +46,12 @@ jobs:
 
       - name: Install dependencies
         run: cd docs && npm ci
-      
+
       - name: Build production website
         env:
           HUGO_ENVIRONMENT: production
           HUGO_ENV: production
         run: cd docs && hugo --minify --gc --config config/production/hugo.toml
-
       - name: Deploy
         uses: peaceiris/actions-gh-pages@4f9cc6602d3f66b9c108549d475ec49e8ef4d45e # v4.0.0
         if: github.repository == 'DefectDojo/django-DefectDojo' # Deploy docs only in core repo, not in forks - it would just fail in fork

docs/content/en/open_source/upgrading/2.54.md

@@ -2,7 +2,7 @@
 title: 'Upgrading to DefectDojo Version 2.54.x'
 toc_hide: true
 weight: -20251226
-description: Removal of django-auditlog & Dropped support for DD_PARSER_EXCLUDE & Reimport performance improvements
+description: Removal of django-auditlog & Dropped support for DD_PARSER_EXCLUDE & Reimport performance improvements & Removal of Finding Template Matching
 ---
 
 ## Breaking Change: Removal of django-auditlog
@@ -55,5 +55,9 @@ DefectDojo 2.54.x includes performance improvements for reimporting scan results
 
 No action is required after upgrading. (Optional tuning knobs exist via `DD_IMPORT_REIMPORT_MATCH_BATCH_SIZE` and `DD_IMPORT_REIMPORT_DEDUPE_BATCH_SIZE`.)
 
+## Finding Template enhancements and removal of CWE matching
+
+As communicated in the [2025Q1 community update](https://github.com/DefectDojo/django-DefectDojo/discussions/12153) the automated matching of Finding Templates based on `CWE` and/or `title` has now been removed.
+
 There are other instructions for upgrading to 2.54.x. Check the Release Notes for the contents of the release: `https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.54.0`
 Check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.54.0) for the contents of the release.

dojo/api_v2/serializers.py

@@ -29,6 +29,7 @@
 from dojo.authorization.roles_permissions import Permissions
 from dojo.endpoint.utils import endpoint_filter, endpoint_meta_import
 from dojo.finding.helper import (
+    save_endpoints_template,
     save_vulnerability_ids,
     save_vulnerability_ids_template,
 )
@@ -111,7 +112,6 @@
     User,
     UserContactInfo,
     Vulnerability_Id,
-    Vulnerability_Id_Template,
     get_current_date,
 )
 from dojo.notifications.helper import create_notification
@@ -2029,57 +2029,80 @@ def validate_severity(self, value: str) -> str:
         return value
 
 
-class VulnerabilityIdTemplateSerializer(serializers.ModelSerializer):
-    class Meta:
-        model = Vulnerability_Id_Template
-        fields = ["vulnerability_id"]
-
-
 class FindingTemplateSerializer(serializers.ModelSerializer):
     tags = TagListSerializerField(required=False)
-    vulnerability_ids = VulnerabilityIdTemplateSerializer(
-        source="vulnerability_id_template_set", many=True, required=False,
-    )
+    vulnerability_ids = serializers.SerializerMethodField()
+    endpoints = serializers.SerializerMethodField()
 
     class Meta:
         model = Finding_Template
-        exclude = ("cve",)
+        exclude = ("cve", "vulnerability_ids_text")
+
+    @extend_schema_field(serializers.ListField(child=serializers.CharField()))
+    def get_vulnerability_ids(self, obj):
+        """Return vulnerability IDs as a list of strings."""
+        return obj.vulnerability_ids
+
+    @extend_schema_field(serializers.ListField(child=serializers.CharField()))
+    def get_endpoints(self, obj):
+        """Return endpoints as a list of URL strings."""
+        return obj.endpoints if hasattr(obj, "endpoints") else []
 
     def create(self, validated_data):
 
-        # Save vulnerability ids and pop them
-        if "vulnerability_id_template_set" in validated_data:
-            vulnerability_id_set = validated_data.pop(
-                "vulnerability_id_template_set",
-            )
-        else:
-            vulnerability_id_set = None
+        # Handle vulnerability_ids if provided as list
+        vulnerability_ids = None
+        if "vulnerability_ids" in self.initial_data:
+            vulnerability_ids = self.initial_data.get("vulnerability_ids", [])
+            if isinstance(vulnerability_ids, str):
+                # If it's a string, split by newlines
+                vulnerability_ids = [vid.strip() for vid in vulnerability_ids.split("\n") if vid.strip()]
+            elif not isinstance(vulnerability_ids, list):
+                vulnerability_ids = []
+
+        # Handle endpoints if provided as list
+        endpoint_urls = None
+        if "endpoints" in self.initial_data:
+            endpoint_urls = self.initial_data.get("endpoints", [])
+            if isinstance(endpoint_urls, str):
+                # If it's a string, split by newlines
+                endpoint_urls = [url.strip() for url in endpoint_urls.split("\n") if url.strip()]
+            elif not isinstance(endpoint_urls, list):
+                endpoint_urls = []
 
         new_finding_template = super().create(
             validated_data,
         )
 
-        if vulnerability_id_set:
-            vulnerability_ids = [vulnerability_id["vulnerability_id"] for vulnerability_id in vulnerability_id_set]
-            validated_data["cve"] = vulnerability_ids[0]
-            save_vulnerability_ids_template(
-                new_finding_template, vulnerability_ids,
-            )
-            new_finding_template.save()
+        # Save vulnerability IDs using helper
+        if vulnerability_ids:
+            save_vulnerability_ids_template(new_finding_template, vulnerability_ids)
+
+        # Save endpoints using helper
+        if endpoint_urls:
+            save_endpoints_template(new_finding_template, endpoint_urls)
 
         return new_finding_template
 
     def update(self, instance, validated_data):
-        # Save vulnerability ids and pop them
-        if "vulnerability_id_template_set" in validated_data:
-            vulnerability_id_set = validated_data.pop(
-                "vulnerability_id_template_set",
-            )
-            vulnerability_ids = []
-            if vulnerability_id_set:
-                vulnerability_ids.extend(vulnerability_id["vulnerability_id"] for vulnerability_id in vulnerability_id_set)
+        # Handle vulnerability_ids if provided
+        if "vulnerability_ids" in self.initial_data:
+            vulnerability_ids = self.initial_data.get("vulnerability_ids", [])
+            if isinstance(vulnerability_ids, str):
+                vulnerability_ids = [vid.strip() for vid in vulnerability_ids.split("\n") if vid.strip()]
+            elif not isinstance(vulnerability_ids, list):
+                vulnerability_ids = []
             save_vulnerability_ids_template(instance, vulnerability_ids)
 
+        # Handle endpoints if provided
+        if "endpoints" in self.initial_data:
+            endpoint_urls = self.initial_data.get("endpoints", [])
+            if isinstance(endpoint_urls, str):
+                endpoint_urls = [url.strip() for url in endpoint_urls.split("\n") if url.strip()]
+            elif not isinstance(endpoint_urls, list):
+                endpoint_urls = []
+            save_endpoints_template(instance, endpoint_urls)
+
         return super().update(instance, validated_data)
 
 

dojo/api_v2/views.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from pathlib import Path
 
+import pghistory
 import tagulous
 from crum import get_current_user
 from dateutil.relativedelta import relativedelta
@@ -2530,7 +2531,17 @@ def perform_create(self, serializer):
             if jira_project := (jira_helper.get_jira_project(jira_driver) if jira_driver else None):
                 push_to_jira = push_to_jira or jira_project.push_all_issues
 
+        # Add pghistory context for audit trail (adds to existing middleware context).
+        # /api/vue is the Pro UI
+        source = "import_vue" if "/api/vue/" in self.request.path else "import_api"
+        pghistory.context(
+            source=source,
+            scan_type=serializer.validated_data.get("scan_type"),
+        )
         serializer.save(push_to_jira=push_to_jira)
+        # Add test_id to pghistory context now that test is created
+        if test_id := serializer.data.get("test"):
+            pghistory.context(test_id=test_id)
 
     def get_queryset(self):
         return get_authorized_tests(Permissions.Import_Scan_Result)
@@ -2678,7 +2689,22 @@ def perform_create(self, serializer):
             if jira_project := (jira_helper.get_jira_project(jira_driver) if jira_driver else None):
                 push_to_jira = push_to_jira or jira_project.push_all_issues
         logger.debug("push_to_jira: %s", push_to_jira)
+        # Add pghistory context for audit trail (adds to existing middleware context)
+        # For reimport, test may already exist or be created during save
+        test_id = test.id if test else serializer.validated_data.get("test", {})
+        if hasattr(test_id, "id"):
+            test_id = test_id.id
+        # /api/vue is the Pro UI
+        source = "reimport_vue" if "/api/vue/" in self.request.path else "reimport_api"
+        pghistory.context(
+            source=source,
+            test_id=test_id if isinstance(test_id, int) else None,
+            scan_type=serializer.validated_data.get("scan_type"),
+        )
         serializer.save(push_to_jira=push_to_jira)
+        # Update test_id if it wasn't available before save
+        if test_id_from_response := serializer.data.get("test"):
+            pghistory.context(test_id=test_id_from_response)
 
 
 # Authorization: configuration

dojo/celery.py

@@ -2,7 +2,7 @@
 import os
 from logging.config import dictConfig
 
-from celery import Celery
+from celery import Celery, Task
 from celery.signals import setup_logging
 from django.conf import settings
 
@@ -11,7 +11,31 @@
 # set the default Django settings module for the 'celery' program.
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "dojo.settings.settings")
 
-app = Celery("dojo")
+
+class PgHistoryTask(Task):
+
+    """
+    Custom Celery base task that automatically applies pghistory context.
+
+    When a task is dispatched via dojo_async_task, the current pghistory
+    context is captured and passed in kwargs as "_pgh_context". This base
+    class extracts that context and applies it before running the task,
+    ensuring all database events share the same context as the original
+    request.
+    """
+
+    def __call__(self, *args, **kwargs):
+        # Import here to avoid circular imports during Celery startup
+        from dojo.pghistory_utils import get_pghistory_context_manager  # noqa: PLC0415
+
+        # Extract context from kwargs (won't be passed to task function)
+        pgh_context = kwargs.pop("_pgh_context", None)
+
+        with get_pghistory_context_manager(pgh_context):
+            return super().__call__(*args, **kwargs)
+
+
+app = Celery("dojo", task_cls=PgHistoryTask)
 
 # Using a string here means the worker will not have to
 # pickle the object when using Windows.

dojo/context_processors.py

@@ -3,6 +3,7 @@
 
 # import the settings file
 from django.conf import settings
+from django.contrib import messages
 
 from dojo.labels import get_labels
 from dojo.models import Alerts, System_Settings, UserAnnouncement
@@ -39,7 +40,26 @@ def globalize_vars(request):
 
 
 def bind_system_settings(request):
-    return {"system_settings": System_Settings.objects.get()}
+    """Load system settings and display warning if there's a database error."""
+    try:
+        system_settings = System_Settings.objects.get()
+        # Check if there was an error stored on the request (from middleware)
+        if hasattr(request, "system_settings_error"):
+            error_msg = request.system_settings_error
+            messages.add_message(
+                request,
+                messages.WARNING,
+                f"Warning: Unable to load system settings from database: {error_msg}. "
+                "Default values are being used. Please check your database configuration and run migrations if needed.",
+                extra_tags="alert-warning",
+            )
+            # Clear after adding message
+            delattr(request, "system_settings_error")
+    except Exception:
+        # If we can't get settings, return empty dict (will cause errors elsewhere, but that's expected)
+        return {}
+
+    return {"system_settings": system_settings}
 
 
 def bind_alert_count(request):