From 97a5acdff5cbed9b4ad61d0cefbfa30eff12ddf1 Mon Sep 17 00:00:00 2001 From: Marc 'risson' Schmitt Date: Wed, 25 Jun 2025 14:12:05 +0200 Subject: [PATCH] ssf meta task Signed-off-by: Marc 'risson' Schmitt --- authentik/enterprise/providers/ssf/signals.py | 10 +++---- authentik/enterprise/providers/ssf/tasks.py | 28 +++++++++++-------- .../enterprise/providers/ssf/views/stream.py | 4 +-- 3 files changed, 23 insertions(+), 19 deletions(-) diff --git a/authentik/enterprise/providers/ssf/signals.py b/authentik/enterprise/providers/ssf/signals.py index 10b3c40e10..1d4e657790 100644 --- a/authentik/enterprise/providers/ssf/signals.py +++ b/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={ diff --git a/authentik/enterprise/providers/ssf/tasks.py b/authentik/enterprise/providers/ssf/tasks.py index 00c74b4f06..c550308650 100644 --- a/authentik/enterprise/providers/ssf/tasks.py +++ b/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 diff --git a/authentik/enterprise/providers/ssf/views/stream.py b/authentik/enterprise/providers/ssf/views/stream.py index 0b06ec0efb..2b36271739 100644 --- a/authentik/enterprise/providers/ssf/views/stream.py +++ b/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,