ssf meta task

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

authentik/enterprise/providers/ssf/signals.py

@@ -18,7 +18,7 @@ from authentik.enterprise.providers.ssf.models import (
     EventTypes,
     SSFProvider,
 )
-from authentik.enterprise.providers.ssf.tasks import send_ssf_event
+from authentik.enterprise.providers.ssf.tasks import send_ssf_events
 from authentik.events.middleware import audit_ignore
 from authentik.stages.authenticator.models import Device
 from authentik.stages.authenticator_duo.models import DuoDevice
@@ -66,7 +66,7 @@ def ssf_user_session_delete_session_revoked(sender, instance: AuthenticatedSessi
 
     As this signal is also triggered with a regular logout, we can't be sure
     if the session has been deleted by an admin or by the user themselves."""
-    send_ssf_event(
+    send_ssf_events(
         EventTypes.CAEP_SESSION_REVOKED,
         {
             "initiating_entity": "user",
@@ -88,7 +88,7 @@ def ssf_user_session_delete_session_revoked(sender, instance: AuthenticatedSessi
 @receiver(password_changed)
 def ssf_password_changed_cred_change(sender, user: User, password: str | None, **_):
     """Credential change trigger (password changed)"""
-    send_ssf_event(
+    send_ssf_events(
         EventTypes.CAEP_CREDENTIAL_CHANGE,
         {
             "credential_type": "password",
@@ -126,7 +126,7 @@ def ssf_device_post_save(sender: type[Model], instance: Device, created: bool, *
     }
     if isinstance(instance, WebAuthnDevice) and instance.aaguid != UNKNOWN_DEVICE_TYPE_AAGUID:
         data["fido2_aaguid"] = instance.aaguid
-    send_ssf_event(
+    send_ssf_events(
         EventTypes.CAEP_CREDENTIAL_CHANGE,
         data,
         sub_id={
@@ -153,7 +153,7 @@ def ssf_device_post_delete(sender: type[Model], instance: Device, **_):
     }
     if isinstance(instance, WebAuthnDevice) and instance.aaguid != UNKNOWN_DEVICE_TYPE_AAGUID:
         data["fido2_aaguid"] = instance.aaguid
-    send_ssf_event(
+    send_ssf_events(
         EventTypes.CAEP_CREDENTIAL_CHANGE,
         data,
         sub_id={

authentik/enterprise/providers/ssf/tasks.py

@@ -27,7 +27,7 @@ session = get_http_session()
 LOGGER = get_logger()
 
 
-def send_ssf_event(
+def send_ssf_events(
     event_type: EventTypes,
     data: dict,
     stream_filter: dict | None = None,
@@ -35,6 +35,7 @@ def send_ssf_event(
     **extra_data,
 ):
     """Wrapper to send an SSF event to multiple streams"""
+    events_data = {}
     if not stream_filter:
         stream_filter = {}
     stream_filter["events_requested__contains"] = [event_type]
@@ -42,7 +43,17 @@ def send_ssf_event(
         extra_data.setdefault("txn", request.request_id)
     for stream in Stream.objects.filter(**stream_filter):
         event_data = stream.prepare_event_payload(event_type, data, **extra_data)
-        _send_ssf_event.send_with_options(args=(stream.uuid, event_data), rel_obj=stream.provider)
+        events_data[stream.uuid] = event_data
+    ssf_events_dispatch.send(events_data)
+
+
+@actor(description=_("Dispatch SSF events."))
+def ssf_events_dispatch(events_data: dict[str, dict[str, Any]]):
+    for stream_uuid, event_data in events_data.items():
+        stream = Stream.objects.filter(pk=stream_uuid).first()
+        if not stream:
+            continue
+        send_ssf_event.send_with_options(args=(stream_uuid, event_data), rel_obj=stream.provider)
 
 
 def _check_app_access(stream: Stream, event_data: dict) -> bool:
@@ -63,7 +74,7 @@ def _check_app_access(stream: Stream, event_data: dict) -> bool:
 
 
 @actor(description=_("Send an SSF event."))
-def _send_ssf_event(stream_uuid: UUID, event_data: dict[str, Any]):
+def send_ssf_event(stream_uuid: UUID, event_data: dict[str, Any]):
     self: Task = CurrentTask.get_task()
 
     stream = Stream.objects.filter(pk=stream_uuid).first()
@@ -96,15 +107,8 @@ def _send_ssf_event(stream_uuid: UUID, event_data: dict[str, Any]):
                 "content": exc.response.text,
                 "status": exc.response.status_code,
             }
-        self.error(
-            exc,
-            LogEvent(
-                "Failed to send request",
-                log_level="warning",
-                logger=self.uid,
-                attributes=attrs,
-            ),
-        )
+        self.warning(exc)
+        self.warning("Failed to send request", **attrs)
         # Re-up the expiry of the stream event
         event.expires = now() + timedelta_from_string(event.stream.provider.event_retention)
         event.status = SSFEventStatus.PENDING_FAILED

authentik/enterprise/providers/ssf/views/stream.py

@@ -13,7 +13,7 @@ from authentik.enterprise.providers.ssf.models import (
     SSFProvider,
     Stream,
 )
-from authentik.enterprise.providers.ssf.tasks import send_ssf_event
+from authentik.enterprise.providers.ssf.tasks import send_ssf_events
 from authentik.enterprise.providers.ssf.views.base import SSFView
 
 LOGGER = get_logger()
@@ -109,7 +109,7 @@ class StreamView(SSFView):
                 "User does not have permission to create stream for this provider."
             )
         instance: Stream = stream.save(provider=self.provider)
-        send_ssf_event(
+        send_ssf_events(
             EventTypes.SET_VERIFICATION,
             {
                 "state": None,