refactor: enhance notification manager retrieval and update alert task signature

Author: Maffooch

dojo/celery_dispatch.py

@@ -87,4 +87,9 @@ def dojo_dispatch_task(task_or_sig: _SupportsSi | _SupportsApplyAsync | Signatur
     sig_kwargs.pop("sync", None)
     sig = sig.clone(kwargs=sig_kwargs)
     eager = sig.apply()
-    return eager.get(propagate=True)
+    try:
+        return eager.get(propagate=True)
+    except RuntimeError:
+        # Since we are intentionally running synchronously, we can propagate exceptions directly, and enable sync subtasks
+        # If the requests desires this. Celery docs explain that this is a rare use case, but we support it _just in case_
+        return eager.get(propagate=True, disable_sync_subtasks=False)

dojo/notifications/helper.py

@@ -46,6 +46,26 @@
 labels = get_labels()
 
 
+def get_manager_class_instance():
+    default_manager = NotificationManager
+    notification_manager_class = default_manager
+    if isinstance(
+        (
+            notification_manager := getattr(
+                settings,
+                "NOTIFICATION_MANAGER",
+                default_manager,
+            )
+        ),
+        str,
+    ):
+        with suppress(ModuleNotFoundError):
+            module_name, _separator, class_name = notification_manager.rpartition(".")
+            module = importlib.import_module(module_name)
+            notification_manager_class = getattr(module, class_name)
+    return notification_manager_class()
+
+
 def create_notification(
     event: str | None = None,
     title: str | None = None,
@@ -63,23 +83,7 @@ def create_notification(
     **kwargs: dict,
 ) -> None:
     """Create an instance of a NotificationManager and dispatch the notification."""
-    default_manager = NotificationManager
-    notification_manager_class = default_manager
-    if isinstance(
-        (
-            notification_manager := getattr(
-                settings,
-                "NOTIFICATION_MANAGER",
-                default_manager,
-            )
-        ),
-        str,
-    ):
-        with suppress(ModuleNotFoundError):
-            module_name, _separator, class_name = notification_manager.rpartition(".")
-            module = importlib.import_module(module_name)
-            notification_manager_class = getattr(module, class_name)
-    notification_manager_class().create_notification(
+    get_manager_class_instance().create_notification(
         event=event,
         title=title,
         finding=finding,
@@ -870,30 +874,30 @@ def _process_notifications(
 @app.task
 def send_slack_notification(event: str, user_id: int | None = None, **kwargs: dict) -> None:
     user = Dojo_User.objects.get(pk=user_id) if user_id else None
-    SlackNotificationManger().send_slack_notification(event, user=user, **kwargs)
+    get_manager_class_instance()._get_manager_instance("slack").send_slack_notification(event, user=user, **kwargs)
 
 
 @app.task
 def send_msteams_notification(event: str, user_id: int | None = None, **kwargs: dict) -> None:
     user = Dojo_User.objects.get(pk=user_id) if user_id else None
-    MSTeamsNotificationManger().send_msteams_notification(event, user=user, **kwargs)
+    get_manager_class_instance()._get_manager_instance("msteams").send_msteams_notification(event, user=user, **kwargs)
 
 
 @app.task
 def send_mail_notification(event: str, user_id: int | None = None, **kwargs: dict) -> None:
     user = Dojo_User.objects.get(pk=user_id) if user_id else None
-    EmailNotificationManger().send_mail_notification(event, user=user, **kwargs)
+    get_manager_class_instance()._get_manager_instance("mail").send_mail_notification(event, user=user, **kwargs)
 
 
 @app.task
 def send_webhooks_notification(event: str, user_id: int | None = None, **kwargs: dict) -> None:
     user = Dojo_User.objects.get(pk=user_id) if user_id else None
-    WebhookNotificationManger().send_webhooks_notification(event, user=user, **kwargs)
+    get_manager_class_instance()._get_manager_instance("webhooks").send_webhooks_notification(event, user=user, **kwargs)
 
 
 @app.task(ignore_result=True)
 def webhook_reactivation(endpoint_id: int, **_kwargs: dict) -> None:
-    WebhookNotificationManger()._webhook_reactivation(endpoint_id=endpoint_id)
+    get_manager_class_instance()._get_manager_instance("webhooks")._webhook_reactivation(endpoint_id=endpoint_id)
 
 
 @app.task(ignore_result=True)

dojo/tasks.py

@@ -31,7 +31,7 @@ def log_generic_alert(source, title, description):
 
 
 @app.task(bind=True)
-def add_alerts(self, runinterval):
+def add_alerts(self, runinterval, *args, **kwargs):
     now = timezone.now()
 
     upcoming_engagements = Engagement.objects.filter(target_start__gt=now + timedelta(days=3), target_start__lt=now + timedelta(days=3) + runinterval).order_by("target_start")