authenticators: attach login failure record to user (#51626)

src/authentic2/authenticators.py

@@ -100,6 +100,7 @@ class LoginPasswordAuthenticator(BaseAuthenticator):
         data = request.POST if is_post else None
         initial = {}
         preferred_ous = []
+        request.failed_logins = set()
 
         # Special handling when the form contains an OU selector
         if app_settings.A2_LOGIN_FORM_OU_SELECTOR:
@@ -139,7 +140,10 @@ class LoginPasswordAuthenticator(BaseAuthenticator):
                 return response
             else:
                 username = form.cleaned_data.get('username', '').strip()
-                if username:
+                if request.failed_logins:
+                    for user in request.failed_logins:
+                        request.journal.record('user.login.failure', user=user, username=username)
+                elif username:
                     request.journal.record('user.login.failure', username=username)
         context['form'] = form
         return render(request, 'authentic2/login_password_form.html', context)

src/authentic2/backends/ldap_backend.py

@@ -720,6 +720,10 @@ class LDAPBackend(object):
                         except ldap.INVALID_CREDENTIALS as e:
                             if block.get('use_controls') and len(e.args) > 0 and 'ctrls' in e.args[0]:
                                 self.process_controls(request, authz_id, DecodeControlTuples(e.args[0]['ctrls']))
+                            attributes = self.get_ldap_attributes(block, conn, authz_id)
+                            user = self.lookup_existing_user(authz_id, block, attributes)
+                            if user and hasattr(request, 'failed_logins'):
+                                request.failed_logins.add(user)
                             user_login_failure(authz_id)
                             pass
                     else:
@@ -1238,7 +1242,7 @@ class LDAPBackend(object):
         for lookup_type in block['lookups']:
             if lookup_type == 'username':
                 return self.lookup_by_username(username)
-            elif lookup_type == 'external_id':
+            elif lookup_type == 'external_id' and attributes:
                 return self.lookup_by_external_id(block, attributes)
 
     def update_user_identifiers(self, user, username, block, attributes):

src/authentic2/backends/models_backend.py

@@ -83,6 +83,8 @@ class ModelBackend(ModelBackend):
                 return user
             else:
                 user_login_failure(user.get_username())
+                if hasattr(request, 'failed_logins'):
+                    request.failed_logins.add(user)
 
     def get_user(self, user_id):
         UserModel = get_user_model()

src/authentic2/journal_event_types.py

@@ -149,8 +149,8 @@ class UserLoginFailure(EventTypeWithService):
     label = _('login failure')
 
     @classmethod
-    def record(cls, service, username):
-        super().record(service=service, data={'username': username})
+    def record(cls, service, username, user):
+        super().record(user=user, service=service, data={'username': username})
 
     @classmethod
     def get_message(cls, event, context):

tests/test_ldap.py

@@ -280,6 +280,26 @@ def test_double_login(slapd, simple_user, settings, app, db):
     utils.login(app, UID, password=PASS, path='/admin/')
 
 
+def test_login_failure(slapd, simple_user, settings, app, db):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd.ldap_url],
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+        'is_superuser': True,
+        'is_staff': True,
+    }]
+    # create ldap user
+    utils.login(app, UID, password=PASS, path='/admin/')
+    utils.logout(app)
+    user = ldap_backend.LDAPUser.objects.get(username='%s@ldap' % UID)
+
+    utils.login(app, simple_user, password='wrong', fail=True)
+    utils.assert_event('user.login.failure', user=simple_user, username=simple_user.username)
+
+    utils.login(app, UID, password='wrong', fail=True)
+    utils.assert_event('user.login.failure', user=user, username=UID)
+
+
 def test_keep_password_in_session(slapd, settings, client, db):
     settings.LDAP_AUTH_SETTINGS = [{
         'url': [slapd.ldap_url],

tests/test_login.py

@@ -36,7 +36,10 @@ def test_success(db, app, simple_user):
 
 def test_failure(db, app, simple_user):
     login(app, simple_user, password='wrong', fail=True)
-    assert_event('user.login.failure', username=simple_user.username)
+    assert_event('user.login.failure', user=simple_user, username=simple_user.username)
+
+    login(app, 'noone', password='wrong', fail=True)
+    assert_event('user.login.failure', username='noone')
 
 
 def test_login_inactive_user(db, app):