Django AuditLog: Upgrade to 3.x

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

@@ -4,4 +4,17 @@ toc_hide: true
 weight: -20250106
 description: No special instructions.
 ---
-There are no special instructions for upgrading to 2.43.x. Check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.43.0) for the contents of the release.
+
+Audit log migration
+---
+
+As part of the upgrade to django-auditlog 3.x, there is a migration of
+existing records from json-text to json. Depending on the number of
+LogEntry objects in your database, this migration could take a long time
+to fully execute. If you believe this period of time will be disruptive
+to your operations, please consult the [migration guide](https://django-auditlog.readthedocs.io/en/latest/upgrade.html#upgrading-to-version-3)
+for making this migration a two step process.
+
+---
+
+For all other instructions for upgrading to 2.43.x. Check the [Release Notes](https://github.com/DefectDojo/django-DefectDojo/releases/tag/2.43.0) for the contents of the release.

dojo/models.py

@@ -123,6 +123,14 @@ def _manage_inherited_tags(obj, incoming_inherited_tags, potentially_existing_ta
             obj.tags.set(cleaned_tag_list)
 
 
+def _copy_model_util(model_in_database, exclude_fields: list[str] = []):
+    new_model_instance = model_in_database.__class__()
+    for field in model_in_database._meta.fields:
+        if field.name not in ["id", *exclude_fields]:
+            setattr(new_model_instance, field.name, getattr(model_in_database, field.name))
+    return new_model_instance
+
+
 @deconstructible
 class UniqueUploadNameProvider:
 
@@ -670,9 +678,7 @@ class NoteHistory(models.Model):
     current_editor = models.ForeignKey(Dojo_User, editable=False, null=True, on_delete=models.CASCADE)
 
     def copy(self):
-        copy = self
-        copy.pk = None
-        copy.id = None
+        copy = _copy_model_util(self)
         copy.save()
         return copy
 
@@ -698,12 +704,9 @@ def __str__(self):
         return self.entry
 
     def copy(self):
-        copy = self
+        copy = _copy_model_util(self)
         # Save the necessary ManyToMany relationships
         old_history = list(self.history.all())
-        # Wipe the IDs of the new object
-        copy.pk = None
-        copy.id = None
         # Save the object before setting any ManyToMany relationships
         copy.save()
         # Copy the history
@@ -718,10 +721,7 @@ class FileUpload(models.Model):
     file = models.FileField(upload_to=UniqueUploadNameProvider("uploaded_files"))
 
     def copy(self):
-        copy = self
-        # Wipe the IDs of the new object
-        copy.pk = None
-        copy.id = None
+        copy = _copy_model_util(self)
         # Add unique modifier to file name
         copy.title = f"{self.title} - clone-{str(uuid4())[:8]}"
         # Create new unique file name
@@ -1505,16 +1505,13 @@ def get_absolute_url(self):
         return reverse("view_engagement", args=[str(self.id)])
 
     def copy(self):
-        copy = self
+        copy = _copy_model_util(self)
         # Save the necessary ManyToMany relationships
         old_notes = list(self.notes.all())
         old_files = list(self.files.all())
         old_tags = list(self.tags.all())
         old_risk_acceptances = list(self.risk_acceptance.all())
         old_tests = list(Test.objects.filter(engagement=self))
-        # Wipe the IDs of the new object
-        copy.pk = None
-        copy.id = None
         # Save the object before setting any ManyToMany relationships
         copy.save()
         # Copy the notes
@@ -1622,10 +1619,8 @@ def __str__(self):
         return f"'{self.finding}' on '{self.endpoint}'"
 
     def copy(self, finding=None):
-        copy = self
+        copy = _copy_model_util(self)
         current_endpoint = self.endpoint
-        copy.pk = None
-        copy.id = None
         if finding:
             copy.finding = finding
         copy.endpoint = current_endpoint
@@ -2091,15 +2086,12 @@ def get_breadcrumbs(self):
         return bc
 
     def copy(self, engagement=None):
-        copy = self
+        copy = _copy_model_util(self)
         # Save the necessary ManyToMany relationships
         old_notes = list(self.notes.all())
         old_files = list(self.files.all())
         old_tags = list(self.tags.all())
         old_findings = list(Finding.objects.filter(test=self))
-        # Wipe the IDs of the new object
-        copy.pk = None
-        copy.id = None
         if engagement:
             copy.engagement = engagement
         # Save the object before setting any ManyToMany relationships
@@ -2707,7 +2699,7 @@ def __init__(self, *args, **kwargs):
         self.unsaved_vulnerability_ids = None
 
     def copy(self, test=None):
-        copy = self
+        copy = _copy_model_util(self)
         # Save the necessary ManyToMany relationships
         old_notes = list(self.notes.all())
         old_files = list(self.files.all())
@@ -2716,8 +2708,6 @@ def copy(self, test=None):
         old_found_by = list(self.found_by.all())
         old_tags = list(self.tags.all())
         # Wipe the IDs of the new object
-        copy.pk = None
-        copy.id = None
         if test:
             copy.test = test
         # Save the object before setting any ManyToMany relationships
@@ -3700,13 +3690,10 @@ def engagement(self):
         return None
 
     def copy(self, engagement=None):
-        copy = self
+        copy = _copy_model_util(self)
         # Save the necessary ManyToMany relationships
         old_notes = list(self.notes.all())
         old_accepted_findings_hash_codes = [finding.hash_code for finding in self.accepted_findings.all()]
-        # Wipe the IDs of the new object
-        copy.pk = None
-        copy.id = None
         # Save the object before setting any ManyToMany relationships
         copy.save()
         # Copy the notes

dojo/settings/settings.dist.py

@@ -1796,6 +1796,9 @@ def saml2_attrib_map_format(dict):
 # ------------------------------------------------------------------------------
 AUDITLOG_FLUSH_RETENTION_PERIOD = env("DD_AUDITLOG_FLUSH_RETENTION_PERIOD")
 ENABLE_AUDITLOG = env("DD_ENABLE_AUDITLOG")
+AUDITLOG_TWO_STEP_MIGRATION = False
+AUDITLOG_USE_TEXT_CHANGES_IF_JSON_IS_NOT_PRESENT = False
+
 USE_FIRST_SEEN = env("DD_USE_FIRST_SEEN")
 USE_QUALYS_LEGACY_SEVERITY_PARSING = env("DD_QUALYS_LEGACY_SEVERITY_PARSING")
 

dojo/templatetags/display_tags.py

@@ -3,6 +3,7 @@
 import datetime
 import logging
 import mimetypes
+from ast import literal_eval
 from itertools import chain
 
 import bleach
@@ -323,8 +324,7 @@ def display_index(data, index):
 @register.filter(is_safe=True, needs_autoescape=False)
 @stringfilter
 def action_log_entry(value, autoescape=None):
-    import json
-    history = json.loads(value)
+    history = literal_eval(value)
     text = ""
     for k in history:
         if isinstance(history[k], dict):

requirements.txt

@@ -5,7 +5,7 @@ bleach[css]
 celery==5.4.0
 defusedxml==0.7.1
 django_celery_results==2.5.1
-django-auditlog==2.3.0
+django-auditlog==3.0.0
 django-dbbackup==4.2.1
 django-environ==0.12.0
 django-filter==24.3

unittests/test_copy_model.py

@@ -1,9 +1,36 @@
+from django.db import models
+
 from dojo.models import Endpoint, Endpoint_Status, Engagement, Finding, Product, Test, User
 
 from .dojo_test_case import DojoTestCase
 
 
-class TestCopyFindingModel(DojoTestCase):
+class EqualityHelpers:
+    def compare_querysets(self, queryset_1, queryset_2):
+        """
+        Compares two querysets and returns True if all fields (excluding primary key)
+        are equal for each model instance in both querysets. Otherwise, returns False.
+        """
+        # Make sure the querysets are the same length
+        if queryset_1.count() != queryset_2.count():
+            return False
+        # Get the model class from the queryset
+        model_class = queryset_1.model
+        # Create sets of field names excluding the primary key (ID)
+        field_names = [f.name for f in model_class._meta.get_fields() if not isinstance(f, models.AutoField)]
+        # Sort the querysets to compare them in order (if needed)
+        queryset_1 = queryset_1.order_by(*field_names)
+        queryset_2 = queryset_2.order_by(*field_names)
+        # Compare the fields of each model instance in the querysets
+        for obj1, obj2 in zip(queryset_1, queryset_2):
+            # Compare each field in the model, skipping the primary key
+            for field in field_names:
+                if getattr(obj1, field) != getattr(obj2, field):
+                    return False
+        return True
+
+
+class TestCopyFindingModel(DojoTestCase, EqualityHelpers):
 
     def test_duplicate_finding_same_test(self):
         # Set the scene
@@ -78,7 +105,7 @@ def test_duplicate_finding_with_notes(self):
         # Make sure the copy was made without error
         self.assertEqual(current_finding_count + 1, Finding.objects.filter(test=test).count())
         # Do the notes match
-        self.assertEqual(finding.notes, finding_copy.notes)
+        self.assertTrue(self.compare_querysets(finding.notes.all(), finding_copy.notes.all()))
 
     def test_duplicate_finding_with_tags_and_notes(self):
         # Set the scene
@@ -98,9 +125,9 @@ def test_duplicate_finding_with_tags_and_notes(self):
         # Make sure the copy was made without error
         self.assertEqual(current_finding_count + 1, Finding.objects.filter(test=test).count())
         # Do the tags match
-        self.assertEqual(finding.notes, finding_copy.notes)
+        self.assertEqual(finding.tags, finding_copy.tags)
         # Do the notes match
-        self.assertEqual(finding.notes, finding_copy.notes)
+        self.assertTrue(self.compare_querysets(finding.notes.all(), finding_copy.notes.all()))
 
     def test_duplicate_finding_with_endpoints(self):
         # Set the scene
@@ -132,7 +159,7 @@ def test_duplicate_finding_with_endpoints(self):
         self.assertNotEqual(endpoint_status, finding_copy.status_finding.all().first())
 
 
-class TestCopyTestModel(DojoTestCase):
+class TestCopyTestModel(DojoTestCase, EqualityHelpers):
 
     def test_duplicate_test_same_enagagement(self):
         # Set the scene
@@ -173,7 +200,9 @@ def test_duplicate_tests_different_engagements(self):
         # Do the enagements have the same number of findings
         self.assertEqual(Finding.objects.filter(test__engagement=engagement1).count(), Finding.objects.filter(test__engagement=engagement2).count())
         # Are the tests equal
-        self.assertEqual(test, test_copy)
+        self.assertEqual(test.title, test_copy.title)
+        self.assertEqual(test.scan_type, test_copy.scan_type)
+        self.assertEqual(test.test_type, test_copy.test_type)
         # Does the product thave more findings
         self.assertEqual(product_finding_count + 1, Finding.objects.filter(test__engagement__product=product).count())
 
@@ -213,7 +242,7 @@ def test_duplicate_test_with_notes(self):
         # Make sure the copy was made without error
         self.assertEqual(current_test_count + 1, Test.objects.filter(engagement=engagement).count())
         # Do the notes match
-        self.assertEqual(test.notes, test_copy.notes)
+        self.assertTrue(self.compare_querysets(test.notes.all(), test_copy.notes.all()))
 
     def test_duplicate_test_with_tags_and_notes(self):
         # Set the scene
@@ -233,12 +262,12 @@ def test_duplicate_test_with_tags_and_notes(self):
         # Make sure the copy was made without error
         self.assertEqual(current_test_count + 1, Test.objects.filter(engagement=engagement).count())
         # Do the notes match
-        self.assertEqual(test.notes, test_copy.notes)
+        self.assertTrue(self.compare_querysets(test.notes.all(), test_copy.notes.all()))
         # Do the tags match
         self.assertEqual(test.tags, test_copy.tags)
 
 
-class TestCopyEngagementModel(DojoTestCase):
+class TestCopyEngagementModel(DojoTestCase, EqualityHelpers):
 
     def test_duplicate_engagement(self):
         # Set the scene
@@ -297,7 +326,7 @@ def test_duplicate_engagement_with_notes(self):
         # Make sure the copy was made without error
         self.assertEqual(current_engagement_count + 1, Engagement.objects.filter(product=product).count())
         # Do the notes match
-        self.assertEqual(engagement.notes, engagement_copy.notes)
+        self.assertTrue(self.compare_querysets(engagement.notes.all(), engagement_copy.notes.all()))
 
     def test_duplicate_engagement_with_tags_and_notes(self):
         # Set the scene
@@ -317,6 +346,6 @@ def test_duplicate_engagement_with_tags_and_notes(self):
         # Make sure the copy was made without error
         self.assertEqual(current_engagement_count + 1, Engagement.objects.filter(product=product).count())
         # Do the notes match
-        self.assertEqual(engagement.notes, engagement_copy.notes)
+        self.assertTrue(self.compare_querysets(engagement.notes.all(), engagement_copy.notes.all()))
         # Do the tags match
         self.assertEqual(engagement.tags, engagement_copy.tags)

unittests/test_deduplication_logic.py

@@ -4,7 +4,17 @@
 from crum import impersonate
 from django.conf import settings
 
-from dojo.models import Endpoint, Endpoint_Status, Engagement, Finding, Product, System_Settings, Test, User
+from dojo.models import (
+    Endpoint,
+    Endpoint_Status,
+    Engagement,
+    Finding,
+    Product,
+    System_Settings,
+    Test,
+    User,
+    _copy_model_util,
+)
 
 from .dojo_test_case import DojoTestCase
 
@@ -1189,8 +1199,7 @@ def log_summary(self, product=None, engagement=None, test=None):
 
     def copy_and_reset_finding(self, id):
         org = Finding.objects.get(id=id)
-        new = org
-        new.pk = None
+        new = _copy_model_util(org)
         new.duplicate = False
         new.duplicate_finding = None
         new.active = True
@@ -1227,15 +1236,13 @@ def copy_and_reset_finding_add_endpoints(self, id, static=False, dynamic=True):
 
     def copy_and_reset_test(self, id):
         org = Test.objects.get(id=id)
-        new = org
-        new.pk = None
+        new = _copy_model_util(org)
         # return unsaved new finding and reloaded existing finding
         return new, Test.objects.get(id=id)
 
     def copy_and_reset_engagement(self, id):
         org = Engagement.objects.get(id=id)
-        new = org
-        new.pk = None
+        new = _copy_model_util(org)
         # return unsaved new finding and reloaded existing finding
         return new, Engagement.objects.get(id=id)