enterprise/providers: SSF (#12327)
* init Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix some other stuff Signed-off-by: Jens Langhammer <jens@goauthentik.io> * more progress Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix missing format Signed-off-by: Jens Langhammer <jens@goauthentik.io> * make it work, send verification event Signed-off-by: Jens Langhammer <jens@goauthentik.io> * progress Signed-off-by: Jens Langhammer <jens@goauthentik.io> * more progress Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix Signed-off-by: Jens Langhammer <jens@goauthentik.io> * save iss Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add signals for MFA devices Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix tests Signed-off-by: Jens Langhammer <jens@goauthentik.io> * refactor more Signed-off-by: Jens Langhammer <jens@goauthentik.io> * re-work auth Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add API to list ssf streams Signed-off-by: Jens Langhammer <jens@goauthentik.io> * start rbac Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add ssf icon Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix web Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix bugs Signed-off-by: Jens Langhammer <jens@goauthentik.io> * make events expire, rewrite sending logic Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add oidc token test Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add stream list Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add jwks tests and fixes Signed-off-by: Jens Langhammer <jens@goauthentik.io> * update web ui Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix configuration endpoint Signed-off-by: Jens Langhammer <jens@goauthentik.io> * replace port number correctly Signed-off-by: Jens Langhammer <jens@goauthentik.io> * better log what went wrong Signed-off-by: Jens Langhammer <jens@goauthentik.io> * linter has opinions Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix messages Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix set status Signed-off-by: Jens Langhammer <jens@goauthentik.io> * more debug logging Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix issuer here too Signed-off-by: Jens Langhammer <jens@goauthentik.io> * remove port :443...removal apparently apple's HTTP logic is wrong and includes the port in the Host header even if the default port is used (80 or 443), which then fails as the URL doesn't exactly match what the admin configured...so instead of trying to add magic about this we'll add it in the docs Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix error when no request in context Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add signal for admin session revoke Signed-off-by: Jens Langhammer <jens@goauthentik.io> * set txn based on request id Signed-off-by: Jens Langhammer <jens@goauthentik.io> * validate method and endpoint url Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix request ID detection Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add timestamp Signed-off-by: Jens Langhammer <jens@goauthentik.io> * temp migration Signed-off-by: Jens Langhammer <jens@goauthentik.io> * fix signal Signed-off-by: Jens Langhammer <jens@goauthentik.io> * add signal tests Signed-off-by: Jens Langhammer <jens@goauthentik.io> * the final commit Signed-off-by: Jens Langhammer <jens@goauthentik.io> * ok actually the last commit Signed-off-by: Jens Langhammer <jens@goauthentik.io> --------- Signed-off-by: Jens Langhammer <jens@goauthentik.io>
This commit is contained in:
Jens L.
112
authentik/enterprise/providers/ssf/tasks.py
Normal file
112
authentik/enterprise/providers/ssf/tasks.py
Normal file
@ -0,0 +1,112 @@
|
||||
from celery import group
|
||||
from django.http import HttpRequest
|
||||
from django.utils.timezone import now
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from requests.exceptions import RequestException
|
||||
from structlog.stdlib import get_logger
|
||||
|
||||
from authentik.enterprise.providers.ssf.models import (
|
||||
DeliveryMethods,
|
||||
EventTypes,
|
||||
SSFEventStatus,
|
||||
Stream,
|
||||
StreamEvent,
|
||||
)
|
||||
from authentik.events.logs import LogEvent
|
||||
from authentik.events.models import TaskStatus
|
||||
from authentik.events.system_tasks import SystemTask
|
||||
from authentik.lib.utils.http import get_http_session
|
||||
from authentik.lib.utils.time import timedelta_from_string
|
||||
from authentik.root.celery import CELERY_APP
|
||||
|
||||
session = get_http_session()
|
||||
LOGGER = get_logger()
|
||||
|
||||
|
||||
def send_ssf_event(
|
||||
event_type: EventTypes,
|
||||
data: dict,
|
||||
stream_filter: dict | None = None,
|
||||
request: HttpRequest | None = None,
|
||||
**extra_data,
|
||||
):
|
||||
"""Wrapper to send an SSF event to multiple streams"""
|
||||
payload = []
|
||||
if not stream_filter:
|
||||
stream_filter = {}
|
||||
stream_filter["events_requested__contains"] = [event_type]
|
||||
if request and hasattr(request, "request_id"):
|
||||
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)
|
||||
payload.append((str(stream.uuid), event_data))
|
||||
return _send_ssf_event.delay(payload)
|
||||
|
||||
|
||||
@CELERY_APP.task()
|
||||
def _send_ssf_event(event_data: list[tuple[str, dict]]):
|
||||
tasks = []
|
||||
for stream, data in event_data:
|
||||
event = StreamEvent.objects.create(**data)
|
||||
tasks.extend(send_single_ssf_event(stream, str(event.uuid)))
|
||||
main_task = group(*tasks)
|
||||
main_task()
|
||||
|
||||
|
||||
def send_single_ssf_event(stream_id: str, evt_id: str):
|
||||
stream = Stream.objects.filter(pk=stream_id).first()
|
||||
if not stream:
|
||||
return
|
||||
event = StreamEvent.objects.filter(pk=evt_id).first()
|
||||
if not event:
|
||||
return
|
||||
if event.status == SSFEventStatus.SENT:
|
||||
return
|
||||
if stream.delivery_method == DeliveryMethods.RISC_PUSH:
|
||||
return [ssf_push_event.si(str(event.pk))]
|
||||
return []
|
||||
|
||||
|
||||
@CELERY_APP.task(bind=True, base=SystemTask)
|
||||
def ssf_push_event(self: SystemTask, event_id: str):
|
||||
self.save_on_success = False
|
||||
event = StreamEvent.objects.filter(pk=event_id).first()
|
||||
if not event:
|
||||
return
|
||||
self.set_uid(event_id)
|
||||
if event.status == SSFEventStatus.SENT:
|
||||
self.set_status(TaskStatus.SUCCESSFUL)
|
||||
return
|
||||
try:
|
||||
response = session.post(
|
||||
event.stream.endpoint_url,
|
||||
data=event.stream.encode(event.payload),
|
||||
headers={"Content-Type": "application/secevent+jwt", "Accept": "application/json"},
|
||||
)
|
||||
response.raise_for_status()
|
||||
event.status = SSFEventStatus.SENT
|
||||
event.save()
|
||||
self.set_status(TaskStatus.SUCCESSFUL)
|
||||
return
|
||||
except RequestException as exc:
|
||||
LOGGER.warning("Failed to send SSF event", exc=exc)
|
||||
self.set_status(TaskStatus.ERROR)
|
||||
attrs = {}
|
||||
if exc.response:
|
||||
attrs["response"] = {
|
||||
"content": exc.response.text,
|
||||
"status": exc.response.status_code,
|
||||
}
|
||||
self.set_error(
|
||||
exc,
|
||||
LogEvent(
|
||||
_("Failed to send request"),
|
||||
log_level="warning",
|
||||
logger=self.__name__,
|
||||
attributes=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
|
||||
event.save()
|
||||
Reference in New Issue
Block a user