authentik.enterprise.providers.ssf.signals

  1from hashlib import sha256
  2
  3from django.db.models import Model
  4from django.db.models.signals import post_delete, post_save, pre_delete
  5from django.dispatch import receiver
  6
  7from authentik.core.models import (
  8    USER_PATH_SYSTEM_PREFIX,
  9    AuthenticatedSession,
 10    Token,
 11    TokenIntents,
 12    User,
 13    UserTypes,
 14)
 15from authentik.core.signals import password_changed
 16from authentik.enterprise.providers.ssf.models import (
 17    EventTypes,
 18    SSFProvider,
 19)
 20from authentik.enterprise.providers.ssf.tasks import send_ssf_events
 21from authentik.events.middleware import audit_ignore
 22from authentik.stages.authenticator.models import Device
 23from authentik.stages.authenticator_duo.models import DuoDevice
 24from authentik.stages.authenticator_static.models import StaticDevice
 25from authentik.stages.authenticator_totp.models import TOTPDevice
 26from authentik.stages.authenticator_webauthn.models import (
 27    UNKNOWN_DEVICE_TYPE_AAGUID,
 28    WebAuthnDevice,
 29)
 30
 31USER_PATH_PROVIDERS_SSF = USER_PATH_SYSTEM_PREFIX + "/providers/ssf"
 32
 33
 34@receiver(post_save, sender=SSFProvider)
 35def ssf_providers_post_save(sender: type[Model], instance: SSFProvider, created: bool, **_):
 36    """Create service account before provider is saved"""
 37    identifier = instance.service_account_identifier
 38    user, _ = User.objects.update_or_create(
 39        username=identifier,
 40        defaults={
 41            "name": f"SSF Provider {instance.name} Service-Account",
 42            "type": UserTypes.INTERNAL_SERVICE_ACCOUNT,
 43            "path": USER_PATH_PROVIDERS_SSF,
 44        },
 45    )
 46    user.assign_perms_to_managed_role("authentik_providers_ssf.add_stream", instance)
 47    token, token_created = Token.objects.update_or_create(
 48        identifier=identifier,
 49        defaults={
 50            "user": user,
 51            "intent": TokenIntents.INTENT_API,
 52            "expiring": False,
 53            "managed": f"goauthentik.io/providers/ssf/{instance.pk}",
 54        },
 55    )
 56    if created or token_created:
 57        with audit_ignore():
 58            instance.token = token
 59            instance.save()
 60
 61
 62@receiver(pre_delete, sender=AuthenticatedSession)
 63def ssf_user_session_delete_session_revoked(sender, instance: AuthenticatedSession, **_):
 64    """Session revoked trigger (users' session has been deleted)
 65
 66    As this signal is also triggered with a regular logout, we can't be sure
 67    if the session has been deleted by an admin or by the user themselves."""
 68    send_ssf_events(
 69        EventTypes.CAEP_SESSION_REVOKED,
 70        {
 71            "initiating_entity": "user",
 72        },
 73        sub_id={
 74            "format": "complex",
 75            "session": {
 76                "format": "opaque",
 77                "id": sha256(instance.session.session_key.encode("ascii")).hexdigest(),
 78            },
 79            "user": {
 80                "format": "email",
 81                "email": instance.user.email,
 82            },
 83        },
 84    )
 85
 86
 87@receiver(password_changed)
 88def ssf_password_changed_cred_change(sender, user: User, password: str | None, **_):
 89    """Credential change trigger (password changed)"""
 90    send_ssf_events(
 91        EventTypes.CAEP_CREDENTIAL_CHANGE,
 92        {
 93            "credential_type": "password",
 94            "change_type": "revoke" if password is None else "update",
 95        },
 96        sub_id={
 97            "format": "complex",
 98            "user": {
 99                "format": "email",
100                "email": user.email,
101            },
102        },
103    )
104
105
106device_type_map = {
107    StaticDevice: "pin",
108    TOTPDevice: "pin",
109    WebAuthnDevice: "fido-u2f",
110    DuoDevice: "app",
111}
112
113
114@receiver(post_save)
115def ssf_device_post_save(sender: type[Model], instance: Device, created: bool, **_):
116    if not isinstance(instance, Device):
117        return
118    if not instance.confirmed:
119        return
120    device_type = device_type_map.get(instance.__class__)
121    data = {
122        "credential_type": device_type,
123        "change_type": "create" if created else "update",
124        "friendly_name": instance.name,
125    }
126    if isinstance(instance, WebAuthnDevice) and instance.aaguid != UNKNOWN_DEVICE_TYPE_AAGUID:
127        data["fido2_aaguid"] = instance.aaguid
128    send_ssf_events(
129        EventTypes.CAEP_CREDENTIAL_CHANGE,
130        data,
131        sub_id={
132            "format": "complex",
133            "user": {
134                "format": "email",
135                "email": instance.user.email,
136            },
137        },
138    )
139
140
141@receiver(post_delete)
142def ssf_device_post_delete(sender: type[Model], instance: Device, **_):
143    if not isinstance(instance, Device):
144        return
145    if not instance.confirmed:
146        return
147    device_type = device_type_map.get(instance.__class__)
148    data = {
149        "credential_type": device_type,
150        "change_type": "delete",
151        "friendly_name": instance.name,
152    }
153    if isinstance(instance, WebAuthnDevice) and instance.aaguid != UNKNOWN_DEVICE_TYPE_AAGUID:
154        data["fido2_aaguid"] = instance.aaguid
155    send_ssf_events(
156        EventTypes.CAEP_CREDENTIAL_CHANGE,
157        data,
158        sub_id={
159            "format": "complex",
160            "user": {
161                "format": "email",
162                "email": instance.user.email,
163            },
164        },
165    )