stages/identification: add signal which is sent upon identification failure

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

authentik/policies/reputation/signals.py

@@ -10,6 +10,7 @@ from authentik.policies.reputation.models import (
     CACHE_KEY_IP_PREFIX,
     CACHE_KEY_USER_PREFIX,
 )
+from authentik.stages.identification.signals import identification_failed
 
 LOGGER = get_logger()
 
@@ -36,6 +37,13 @@ def handle_failed_login(sender, request, credentials, **_):
         update_score(request, credentials.get("username"), -1)
 
 
+@receiver(identification_failed)
+# pylint: disable=unused-argument
+def handle_identification_failed(sender, request, uid_field: str, **_):
+    """Lower Score for failed identification attempts"""
+    update_score(request, uid_field, -1)
+
+
 @receiver(user_logged_in)
 # pylint: disable=unused-argument
 def handle_successful_login(sender, request, user, **_):

authentik/stages/identification/signals.py

@@ -0,0 +1,5 @@
+"""authentik identification signals"""
+from django.core.signals import Signal
+
+# Arguments: request: HttpRequest, uid_field: Value entered by user
+identification_failed = Signal()

authentik/stages/identification/stage.py

@@ -21,6 +21,7 @@ from authentik.flows.stage import (
 )
 from authentik.flows.views import SESSION_KEY_APPLICATION_PRE
 from authentik.stages.identification.models import IdentificationStage
+from authentik.stages.identification.signals import identification_failed
 
 LOGGER = get_logger()
 
@@ -53,6 +54,9 @@ class IdentificationChallengeResponse(ChallengeResponse):
         if not pre_user:
             sleep(0.150)
             LOGGER.debug("invalid_login", identifier=value)
+            identification_failed.send(
+                sender=self, request=self.stage.request, uid_field=value
+            )
             raise ValidationError("Failed to authenticate.")
         self.pre_user = pre_user
         return value