feat(cache): guard against saving cache-derived System_Settings snapshots

System_Settings.objects.get() returns a read-only snapshot rebuilt from the
in-process (L1) cache; saving it could overwrite concurrent changes with stale
field values. Tag such instances and make System_Settings.save() fail loudly
instead of silently persisting a stale snapshot.

- dojo/caching.py: add READ_ONLY_CACHE_MARKER, ReadOnlyCachedInstanceError, and
  a read_only= param on cache_dict_to_model (default off, so tuner/features keep
  saving their rebuilt instances for their read-modify-write flows).
- System_Settings.save() raises ReadOnlyCachedInstanceError when the instance is
  tagged; get_cached_system_settings() tags the snapshot read_only=True.
- Migrate get()->modify->save() writer sites (incl. the system_settings() test
  helper) to objects.get(no_cache=True) for a fresh, saveable row.
- Document the singleton manager .get() semantics (ignores filter kwargs, never
  raises DoesNotExist; data migrations use apps.get_model, so get_or_create is
  unaffected). Fix stale L1/L2 comments left by the drop-L2 change.
- test_caching: cover the save-guard and the read_only flag.

Author: valentijnscholten

dojo/caching.py

@@ -150,11 +150,33 @@ def model_to_cache_dict(instance) -> dict:
     return {f.attname: f.value_from_object(instance) for f in instance._meta.concrete_fields}
 
 
-def cache_dict_to_model(model_cls, data: dict):
+# Attribute set on instances rebuilt for read-only cache use (``read_only=True``).
+# A guarded model's ``save()`` checks it and refuses to persist a cache-derived
+# snapshot (see ``System_Settings.save``). Instance-level, so it never leaks to a
+# freshly-fetched (``no_cache=True``) instance, which is a distinct object.
+READ_ONLY_CACHE_MARKER = "_dd_read_only_cache_snapshot"
+
+
+class ReadOnlyCachedInstanceError(RuntimeError):
+
+    """
+    Raised when code tries to save a model instance rebuilt from the read-through
+    cache. Such an instance is a point-in-time snapshot; persisting it can clobber
+    concurrent changes with stale field values. Fetch a fresh DB instance
+    (e.g. ``System_Settings.objects.get(no_cache=True)``) before saving.
+    """
+
+
+def cache_dict_to_model(model_cls, data: dict, *, read_only: bool = False):
     """
     Rebuild an in-memory model instance from a ``model_to_cache_dict`` dict.
 
     For read-only use; callers that persist changes must fetch a fresh DB instance
-    rather than saving a cache-derived one.
+    rather than saving a cache-derived one. Pass ``read_only=True`` to tag the
+    instance with ``READ_ONLY_CACHE_MARKER`` so a guarded model's ``save()`` fails
+    loudly instead of silently writing a stale snapshot.
     """
-    return model_cls(**data)
+    instance = model_cls(**data)
+    if read_only:
+        setattr(instance, READ_ONLY_CACHE_MARKER, True)
+    return instance

dojo/celery.py

@@ -31,8 +31,8 @@ def __call__(self, *args, **kwargs):
         Also resets the request/task-scoped L1 settings cache at the start of every
         task (including eager): a prefork worker reuses its thread across tasks, so an
         L1 value cached during a prior task (e.g. System_Settings) would otherwise be
-        served stale even after another process changed and saved it. The shared L2
-        tier is left intact, so a reset task re-reads each singleton once from L2.
+        served stale even after another process changed and saved it. There is no
+        shared tier, so a reset task re-reads each singleton once from the DB.
 
         The apply_async method injects ``async_user_id`` into kwargs when a task
         is dispatched. Here we pop it, resolve to a user instance, and set it

dojo/middleware.py

@@ -26,10 +26,10 @@
 
 logger = logging.getLogger(__name__)
 
-# Two-tier (in-process L1 + django.core.cache L2) read-through cache for the
-# System_Settings singleton, via the shared dojo_settings_cache decorator. Stored
-# as a plain dict (no pickled model graph) and rebuilt into a read-only instance
-# per call. Write paths use ``objects.get(no_cache=True)`` and are unaffected.
+# In-process (L1) read-through cache for the System_Settings singleton, via the
+# shared dojo_settings_cache decorator. Stored as a plain dict (no pickled model
+# graph) and rebuilt into a read-only instance per call. Write paths use
+# ``objects.get(no_cache=True)`` and are unaffected.
 SYSTEM_SETTINGS_CACHE_KEY = "dojo.system_settings.singleton"
 
 
@@ -50,7 +50,9 @@ def get_cached_system_settings():
     data = _cached_system_settings_dict()
     if not isinstance(data, dict):
         return System_Settings()
-    return cache_dict_to_model(System_Settings, data)
+    # read_only=True tags the rebuilt snapshot so System_Settings.save() refuses
+    # to persist it; writers must use objects.get(no_cache=True) for a fresh row.
+    return cache_dict_to_model(System_Settings, data, read_only=True)
 
 
 @receiver(models.signals.post_save, sender="dojo.System_Settings")
@@ -165,10 +167,17 @@ def get_from_db(self, *args, **kwargs):
         return from_db
 
     def get(self, no_cache=False, *args, **kwargs):  # noqa: FBT002 - this is bit hard to fix nice have this universally fixed
+        # NOTE: this override does not behave like Manager.get(). System_Settings is
+        # a singleton, so the cached path ignores any filter args/kwargs, always
+        # returns the one row, and never raises DoesNotExist/MultipleObjectsReturned
+        # (get_or_create against this manager therefore won't create — data
+        # migrations use apps.get_model(), which keeps Django's default manager).
+        # The cached instance is a read-only snapshot (see get_cached_system_settings);
+        # to modify-and-save, fetch a fresh row with no_cache=True.
         if no_cache:
             # logger.debug('no_cache specified, loading system settings from db')
             return self.get_from_db(*args, **kwargs)
-        # Read through the shared L1/L2 cache (dojo.caching). L1 is request/task
+        # Read through the in-process (L1) cache (dojo.caching). L1 is request/task
         # scoped (reset by the middleware and the Celery task base), so repeated
         # reads within a request are served in-process.
         return get_cached_system_settings()

dojo/system_settings/models.py

@@ -351,6 +351,22 @@ class System_Settings(models.Model):
     from dojo.middleware import System_Settings_Manager  # noqa: PLC0415 circular import
     objects = System_Settings_Manager()
 
+    def save(self, *args, **kwargs):
+        # Guard against persisting an instance handed back by the read-through
+        # cache. ``System_Settings.objects.get()`` returns a snapshot rebuilt from
+        # the cached dict (see dojo.middleware.get_cached_system_settings); saving
+        # it could overwrite concurrent changes with stale values. Writers must
+        # fetch a fresh instance with ``System_Settings.objects.get(no_cache=True)``.
+        from dojo.caching import READ_ONLY_CACHE_MARKER, ReadOnlyCachedInstanceError  # noqa: PLC0415 circular import
+        if getattr(self, READ_ONLY_CACHE_MARKER, False):
+            msg = (
+                "Refusing to save a System_Settings instance obtained from the read-through cache "
+                "(System_Settings.objects.get()); it is a read-only snapshot. Fetch a fresh instance "
+                "with System_Settings.objects.get(no_cache=True) before saving."
+            )
+            raise ReadOnlyCachedInstanceError(msg)
+        super().save(*args, **kwargs)
+
     def clean(self):
         super().clean()
 

unittests/dojo_test_case.py

@@ -138,7 +138,7 @@ def get_test_admin(self, *args, **kwargs):
         return User.objects.get(username="admin")
 
     def system_settings(self, **kwargs):
-        ss = System_Settings.objects.get()
+        ss = System_Settings.objects.get(no_cache=True)
         for key, value in kwargs.items():
             setattr(ss, key, value)
         ss.save()

unittests/test_caching.py

@@ -2,6 +2,8 @@
 
 from dojo.caching import (
     _L1_STORE,  # noqa: PLC2701 test needs the in-process store
+    READ_ONLY_CACHE_MARKER,
+    ReadOnlyCachedInstanceError,
     cache_dict_to_model,
     dojo_settings_cache,
     invalidate_dojo_settings_cache,
@@ -83,3 +85,38 @@ def test_round_trip_preserves_field_values(self):
         rebuilt = cache_dict_to_model(System_Settings, data)
         self.assertEqual(rebuilt.pk, settings_obj.pk)
         self.assertEqual(rebuilt.enable_deduplication, settings_obj.enable_deduplication)
+
+    def test_read_only_flag_only_when_requested(self):
+        data = model_to_cache_dict(System_Settings.objects.get(no_cache=True))
+        # default: saveable (no marker)
+        self.assertFalse(getattr(cache_dict_to_model(System_Settings, data), READ_ONLY_CACHE_MARKER, False))
+        # read_only=True: tagged as a read-only snapshot
+        self.assertTrue(getattr(cache_dict_to_model(System_Settings, data, read_only=True), READ_ONLY_CACHE_MARKER, False))
+
+
+class SystemSettingsSaveGuardTest(DojoTestCase):
+
+    """
+    The read-through cache hands back a read-only snapshot: saving the instance
+    returned by ``System_Settings.objects.get()`` must fail loudly, while a fresh
+    ``no_cache=True`` instance saves normally.
+    """
+
+    def setUp(self):
+        _L1_STORE.clear()
+
+    def tearDown(self):
+        _L1_STORE.clear()
+
+    def test_saving_cached_instance_raises(self):
+        cached = System_Settings.objects.get()  # read-through (cached) path
+        self.assertTrue(getattr(cached, READ_ONLY_CACHE_MARKER, False))
+        cached.enable_deduplication = not cached.enable_deduplication
+        with self.assertRaises(ReadOnlyCachedInstanceError):
+            cached.save()
+
+    def test_no_cache_instance_is_saveable(self):
+        fresh = System_Settings.objects.get(no_cache=True)
+        self.assertFalse(getattr(fresh, READ_ONLY_CACHE_MARKER, False))
+        fresh.enable_deduplication = not fresh.enable_deduplication
+        fresh.save()  # must not raise

unittests/test_deduplication_logic.py

@@ -1894,7 +1894,7 @@ def create_new_test_and_engagment_from_finding(self, finding):
         return test_new, eng_new
 
     def enable_dedupe(self, *, enable=True):
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.enable_deduplication = enable
         system_settings.save()
 

unittests/test_duplication_loops.py

@@ -378,7 +378,7 @@ def test_list_relations_for_three_reverse(self):
     # Test that Delete Duplicate Findings & Maximum Duplicate is correctly deleting olding finding first based off of finding date value
     def test_delete_duplicate_order(self):
         # Turn on delete duplicates and set the maximum dedupe value to 1
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.delete_duplicates = True
         system_settings.max_dupes = 1
         system_settings.save()
@@ -409,7 +409,7 @@ def test_delete_duplicate_order(self):
 
     def test_delete_duplicate_order_same_date_tiebreak_by_id(self):
         """When duplicate findings share the same date, excess deletes use id as tie-break (oldest id first)."""
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.delete_duplicates = True
         system_settings.max_dupes = 1
         system_settings.save()

unittests/test_false_positive_history_logic.py

@@ -2031,26 +2031,26 @@ def create_new_test_and_engagment_and_product_from_finding(self, finding):
         return test_new, eng_new, product_new
 
     def enable_false_positive_history(self):
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.false_positive_history = True
         system_settings.save()
 
     def enable_retroactive_false_positive_history(self):
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.retroactive_false_positive_history = True
         system_settings.save()
 
     def disable_retroactive_false_positive_history(self):
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.retroactive_false_positive_history = False
         system_settings.save()
 
     def enable_dedupe(self):
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.enable_deduplication = True
         system_settings.save()
 
     def disable_dedupe(self):
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.enable_deduplication = False
         system_settings.save()

unittests/test_notifications.py

@@ -941,7 +941,7 @@ def test_system_webhook_timeout(self):
         self.sys_wh.url = f"{self.url_base}/delay/3"
         self.sys_wh.save()
 
-        system_settings = System_Settings.objects.get()
+        system_settings = System_Settings.objects.get(no_cache=True)
         system_settings.webhooks_notifications_timeout = 1
         system_settings.save()