feat(cache): cache authorizations in Celery tasks via cache_for_request_or_task

valentijnscholten · valentijnscholten · commit 487d6cba02e1 · 2026-07-07T12:14:54.000+02:00
@cache_for_request is a no-op in Celery (no request), so per-user authorization
filtering re-runs on every call inside a task -- notably the rules engine, which
runs authorized querysets under impersonate(rule.owner). Add a request-or-task
read-through cache and move the authorized-queryset/permission getters onto it.

- dojo/request_cache: cache_for_request_or_task resolves the store as request
  cache -&gt; task cache -&gt; no caching. The task cache is a per-thread store active
  only inside a DojoAsyncTask (begin/end), so management commands / shells that
  never hit a task boundary do not cache. The key folds get_current_user().pk so
  results keyed on user=None (resolved downstream) cannot leak between users on a
  reused worker thread or within a task that impersonates multiple owners.
- dojo/celery: DojoAsyncTask installs a fresh task cache before each task and
  drops it afterwards (finally), bounding staleness to a single task.
- Migrate all get_authorized_* / authorized_*_id_set / get_*_permissions getters
  from @cache_for_request to @cache_for_request_or_task.
- test_caching: cover task-scope caching, per-user isolation, and reset.
diff --git a/dojo/authorization/query_registrations.py b/dojo/authorization/query_registrations.py
@@ -40,7 +40,7 @@
     Tool_Product_Settings,
     Vulnerability_Id,
 )
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 def _resolve_user(user):
@@ -81,7 +81,7 @@ def _authorized_product_type_ids(user):
     return Product_Type.objects.filter(authorized_users=user).values("id")
 
 
-@cache_for_request
+@cache_for_request_or_task
 def authorized_product_id_set(user_pk):
     """
     Frozen set of product ids the user can access via authorized_users
@@ -101,7 +101,7 @@ def authorized_product_id_set(user_pk):
     )
 
 
-@cache_for_request
+@cache_for_request_or_task
 def authorized_product_type_id_set(user_pk):
     """
     Frozen set of product_type ids the user is a direct member of via
diff --git a/dojo/authorization/template_filters.py b/dojo/authorization/template_filters.py
@@ -2,7 +2,7 @@
 
 from dojo.authorization.authorization import user_has_configuration_permission as configuration_permission
 from dojo.authorization.authorization import user_has_global_permission, user_has_permission
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 def has_object_permission(obj, permission):
@@ -21,7 +21,7 @@ def has_configuration_permission(permission, request):
     return configuration_permission(user, permission)
 
 
-@cache_for_request
+@cache_for_request_or_task
 def get_user_permissions(user):
     return user.user_permissions.all()
 
@@ -31,7 +31,7 @@ def user_has_configuration_permission_without_group(user, codename):
     return any(permission.codename == codename for permission in permissions)
 
 
-@cache_for_request
+@cache_for_request_or_task
 def get_group_permissions(group):
     return group.permissions.all()
 
diff --git a/dojo/celery.py b/dojo/celery.py
@@ -46,19 +46,26 @@ def __call__(self, *args, **kwargs):
         crum.impersonate in tests or request middleware) are not disrupted.
         """
         from dojo.caching import reset_l1_cache  # noqa: PLC0415
+        from dojo.request_cache import begin_task_cache, end_task_cache  # noqa: PLC0415
         reset_l1_cache()
-
-        if "async_user_id" not in kwargs:
-            return super().__call__(*args, **kwargs)
-
-        import crum  # noqa: PLC0415
-
-        from dojo.models import Dojo_User  # noqa: PLC0415 circular import
-
-        user_id = kwargs.pop("async_user_id")
-        user = Dojo_User.objects.filter(pk=user_id).first() if user_id else None
-        with crum.impersonate(user):
-            return super().__call__(*args, **kwargs)
+        # Install a fresh task-scoped request-cache (begin) and drop it afterwards
+        # (end), so cache_for_request_or_task can memoize within this task without a
+        # value leaking to the next task on this reused worker thread.
+        begin_task_cache()
+        try:
+            if "async_user_id" not in kwargs:
+                return super().__call__(*args, **kwargs)
+
+            import crum  # noqa: PLC0415
+
+            from dojo.models import Dojo_User  # noqa: PLC0415 circular import
+
+            user_id = kwargs.pop("async_user_id")
+            user = Dojo_User.objects.filter(pk=user_id).first() if user_id else None
+            with crum.impersonate(user):
+                return super().__call__(*args, **kwargs)
+        finally:
+            end_task_cache()
 
     def apply_async(self, args=None, kwargs=None, **options):
         """Override apply_async to inject user context and track tasks."""
diff --git a/dojo/endpoint/queries.py b/dojo/endpoint/queries.py
@@ -7,11 +7,11 @@ def get_auth_filter(key): return None
     Endpoint,
     Endpoint_Status,
 )
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_endpoints(permission, user=None):
     impl = get_auth_filter("endpoint.get_authorized_endpoints")
     if impl:
@@ -27,7 +27,7 @@ def get_authorized_endpoints_for_queryset(permission, queryset, user=None):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_endpoint_status(permission, user=None):
     impl = get_auth_filter("endpoint.get_authorized_endpoint_status")
     if impl:
diff --git a/dojo/engagement/queries.py b/dojo/engagement/queries.py
@@ -4,11 +4,11 @@
     def get_auth_filter(key): return None
 
 from dojo.models import Engagement
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_engagements(permission):
     impl = get_auth_filter("engagement.get_authorized_engagements")
     if impl:
diff --git a/dojo/finding/queries.py b/dojo/finding/queries.py
@@ -19,13 +19,13 @@ def get_auth_filter(key): return None
     Test_Import_Finding_Action,
     Vulnerability_Id,
 )
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 logger = logging.getLogger(__name__)
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_findings(permission, user=None):
     impl = get_auth_filter("finding.get_authorized_findings")
     if impl:
@@ -41,7 +41,7 @@ def get_authorized_findings_for_queryset(permission, queryset, user=None):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_vulnerability_ids(permission, user=None):
     impl = get_auth_filter("finding.get_authorized_vulnerability_ids")
     if impl:
diff --git a/dojo/finding_group/queries.py b/dojo/finding_group/queries.py
@@ -4,11 +4,11 @@
     def get_auth_filter(key): return None
 
 from dojo.models import Finding_Group
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_finding_groups(permission, user=None):
     impl = get_auth_filter("finding_group.get_authorized_finding_groups")
     if impl:
diff --git a/dojo/jira/queries.py b/dojo/jira/queries.py
@@ -4,11 +4,11 @@
     def get_auth_filter(key): return None
 
 from dojo.models import JIRA_Issue, JIRA_Project
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_jira_projects(permission, user=None):
     impl = get_auth_filter("jira_link.get_authorized_jira_projects")
     if impl:
@@ -17,7 +17,7 @@ def get_authorized_jira_projects(permission, user=None):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_jira_issues(permission):
     impl = get_auth_filter("jira_link.get_authorized_jira_issues")
     if impl:
diff --git a/dojo/product/queries.py b/dojo/product/queries.py
@@ -11,11 +11,11 @@ def get_auth_filter(key): return None
     Product,
     Product_API_Scan_Configuration,
 )
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_products(permission, user=None):
     impl = get_auth_filter("product.get_authorized_products")
     if impl:
@@ -24,7 +24,7 @@ def get_authorized_products(permission, user=None):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_app_analysis(permission):
     impl = get_auth_filter("product.get_authorized_app_analysis")
     if impl:
@@ -33,7 +33,7 @@ def get_authorized_app_analysis(permission):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_dojo_meta(permission):
     impl = get_auth_filter("product.get_authorized_dojo_meta")
     if impl:
@@ -42,7 +42,7 @@ def get_authorized_dojo_meta(permission):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_languages(permission):
     impl = get_auth_filter("product.get_authorized_languages")
     if impl:
@@ -51,7 +51,7 @@ def get_authorized_languages(permission):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_engagement_presets(permission):
     impl = get_auth_filter("product.get_authorized_engagement_presets")
     if impl:
@@ -60,7 +60,7 @@ def get_authorized_engagement_presets(permission):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_product_api_scan_configurations(permission):
     impl = get_auth_filter("product.get_authorized_product_api_scan_configurations")
     if impl:
diff --git a/dojo/product_type/queries.py b/dojo/product_type/queries.py
@@ -4,11 +4,11 @@
     def get_auth_filter(key): return None
 
 from dojo.models import Product_Type
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_product_types(permission):
     impl = get_auth_filter("product_type.get_authorized_product_types")
     if impl:
diff --git a/dojo/request_cache/__init__.py b/dojo/request_cache/__init__.py
@@ -1,4 +1,6 @@
-from crum import get_current_request
+import threading
+
+from crum import get_current_request, get_current_user
 
 # Attribution: This code has been taken from https://github.com/anexia-it/django-request-cache, which has
 # been published under the MIT License. Since this project hasn't been updated for more than a year,
@@ -13,6 +15,44 @@ def get_request_cache():  # noqa: RUF067
     return getattr(get_current_request(), "cache", None)
 
 
+class _TaskCache:  # noqa: RUF067
+
+    """
+    Attribute-bag cache with the same access pattern as ``RequestCache`` (values
+    stored via ``setattr``/``getattr`` on the instance), used inside a Celery task
+    when there is no request. A fresh instance is installed at each task boundary
+    (see ``begin_task_cache``), so it can never outlive one task.
+    """
+
+
+# Per-thread task cache. Only populated between ``begin_task_cache`` and
+# ``end_task_cache`` (called by DojoAsyncTask around each task). ``None`` outside a
+# task, so management commands / shells that never hit a task boundary do NOT cache
+# (a cache with no reset boundary would grow unbounded and serve stale data).
+_task_cache_local = threading.local()  # noqa: RUF067
+
+
+def get_task_cache():  # noqa: RUF067
+    """Return the current task's cache, or ``None`` when not inside a task."""
+    return getattr(_task_cache_local, "cache", None)
+
+
+def begin_task_cache():  # noqa: RUF067
+    """
+    Install a fresh task cache. Called at the START of every task so a task never
+    inherits a cache populated by a prior task on the same (reused) worker thread.
+    """
+    _task_cache_local.cache = _TaskCache()
+
+
+def end_task_cache():  # noqa: RUF067
+    """
+    Drop the task cache. Called at the END of every task so nothing lingers on the
+    thread after the task completes.
+    """
+    _task_cache_local.cache = None
+
+
 cache_args_kwargs_marker = object()  # noqa: RUF067 marker for separating args from kwargs (needs to be global)
 
 
@@ -58,3 +98,41 @@ def wrapper(*args, **kwargs):
 
         return result
     return wrapper
+
+
+def cache_for_request_or_task(fn):  # noqa: RUF067
+    """
+    Like ``cache_for_request``, but also caches inside a Celery task (which has no
+    request). Resolves the store as: request cache if a request exists, else the
+    task cache if inside a task, else no caching (executes directly).
+
+    SECURITY: the cache key folds in the *effective* current user
+    (``get_current_user().pk``) so that user-dependent results (e.g. authorized
+    querysets that resolve ``user`` downstream and would otherwise key on
+    ``user=None``) cannot leak between users. This matters in tasks: a worker
+    thread is reused across tasks and ``DojoAsyncTask`` impersonates a different
+    user per task, and a single task can impersonate multiple users (e.g. the
+    rules engine under ``impersonate(rule.owner)``). The per-task reset
+    (begin/end) bounds staleness (e.g. revoked permissions); the user-aware key
+    guarantees isolation even within a task.
+    """
+    def wrapper(*args, **kwargs):
+        cache = get_request_cache()
+        if cache is None:
+            cache = get_task_cache()          # populated only inside a DojoAsyncTask
+        if cache is None:
+            # neither a request nor a task -> execute without caching
+            return fn(*args, **kwargs)
+
+        user = get_current_user()
+        user_pk = user.pk if user is not None else None
+        key = cache_calculate_key(fn.__name__, user_pk, *args, **kwargs)
+
+        try:
+            result = getattr(cache, key)
+        except AttributeError:
+            result = fn(*args, **kwargs)
+            setattr(cache, key, result)
+
+        return result
+    return wrapper
diff --git a/dojo/risk_acceptance/queries.py b/dojo/risk_acceptance/queries.py
@@ -4,11 +4,11 @@
     def get_auth_filter(key): return None
 
 from dojo.models import Risk_Acceptance
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_risk_acceptances(permission):
     impl = get_auth_filter("risk_acceptance.get_authorized_risk_acceptances")
     if impl:
diff --git a/dojo/test/queries.py b/dojo/test/queries.py
@@ -4,11 +4,11 @@
     def get_auth_filter(key): return None
 
 from dojo.models import Test, Test_Import
-from dojo.request_cache import cache_for_request
+from dojo.request_cache import cache_for_request_or_task
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_tests(permission, product=None):
     impl = get_auth_filter("test.get_authorized_tests")
     if impl:
@@ -17,7 +17,7 @@ def get_authorized_tests(permission, product=None):
 
 
 # Cached: all parameters are hashable, no dynamic queryset filtering
-@cache_for_request
+@cache_for_request_or_task
 def get_authorized_test_imports(permission):
     impl = get_auth_filter("test.get_authorized_test_imports")
     if impl:
diff --git a/dojo/tool_product/queries.py b/dojo/tool_product/queries.py
diff --git a/dojo/user/queries.py b/dojo/user/queries.py
diff --git a/unittests/test_caching.py b/unittests/test_caching.py
