Support for tracking failed authentication.

jwag956 · jwag956 · commit 05ee541d4cc0 · 2026-03-20T15:02:19.000-07:00
This adds a new signal - user_failed_authn, and a new method to UserMixin: track_failed_authn that sends that signal. All places that accept a credential and that fails call through track_failed_authn(). API errors, and missing data don't trigger the signal. Added a signals fixture for testing which tracks all signals during a test, and a test_util check_signals that can verify precisely which signals were received. closes #1188
diff --git a/CHANGES.rst b/CHANGES.rst
@@ -14,6 +14,8 @@ Features & Improvements
   require two-factor authentication.
 - (:issue:`1178`) Add Cache-Control headers.
 - (:issue:`1165`) Add support for using Social Login (OAuth) for verification.
+- (:issue:`1188`) Add tracking of failed authentication attempts via :py:meth:`.UserMixin.track_failed_authn`
+  and :py:data:`user_failed_authn`
 
 Fixes
 +++++
diff --git a/docs/api.rst b/docs/api.rst
@@ -284,7 +284,7 @@ sends the following signals.
     sender), it is passed `user`, and `authn_via` arguments. The `authn_via` argument
     specifies how the user authenticated - it will be a list with possible values
     of ``password``, ``sms``, ``authenticator``, ``email``, ``confirm``, ``reset``,
-    ``register``.
+    ``register``, ``webauthn``, ``oauth``.
 
     .. versionadded:: 3.4.0
 
@@ -296,6 +296,18 @@ sends the following signals.
 
     .. versionadded:: 5.4.0
 
+.. data:: user_failed_authn
+
+   Sent from :py:meth:`.UserMixin.track_failed_authn` when a user fails authentication.
+   In addition to the app (which is the sender) it is passed:
+
+    * `user` - the user that failed authentication
+    * `endpoint` - the endpoint that was accessed
+    * `auth_type` - `password`, `code`, `passkey`
+    * `tfa` - True if this was a two factor authentication attempt
+
+    .. versionadded:: 5.8.0
+
 .. data:: user_registered
 
    Sent when a user registers on the site. In addition to the app (which is the
diff --git a/flask_security/__init__.py b/flask_security/__init__.py
@@ -86,6 +86,7 @@
     tf_security_token_sent,
     tf_disabled,
     user_authenticated,
+    user_failed_authn,
     user_unauthenticated,
     user_confirmed,
     user_registered,
diff --git a/flask_security/core.py b/flask_security/core.py
@@ -70,6 +70,7 @@
     MfRecoveryCodesForm,
     MfRecoveryCodesUtil,
 )
+from .signals import user_failed_authn
 from .tf_plugin import TfPlugin, TwoFactorSelectForm
 from .twofactor import tf_send_security_token
 from .unified_signin import (
@@ -1146,6 +1147,45 @@ def check_tf_required_setup(self) -> bool:
         """
         return cv("TWO_FACTOR_REQUIRED")
 
+    def track_failed_authn(
+        self, endpoint: str | None, auth_type: str, tfa: bool = False
+    ) -> None:
+        """Called when a user fails to authenticate.
+
+        endpoint - blueprint endpoint name:
+
+            - security.login
+            - security.verify
+            - security.us_signin
+            - security.us_verify
+            - security.us_verify_link
+            - security.wan_signin_response
+            - security.wan_verify_response
+            - security.two_factor_token_validation
+
+        auth_type is what failed:
+
+            - password
+            - passcode (unified signin)
+            - passkey (webauthn
+
+        tfa - True if it was a second factor that failed
+
+        This is called any time a credential is presented but is not verifiable.
+        It is NOT called on API errors or missing data.
+
+        The default implementation sends the user_failed_authn signal.
+
+        """
+        user_failed_authn.send(
+            current_app._get_current_object(),  # type: ignore[attr-defined]
+            _async_wrapper=current_app.ensure_sync,
+            endpoint=endpoint,
+            user=self,
+            auth_type=auth_type,
+            tfa=tfa,
+        )
+
 
 class WebAuthnMixin:
     if t.TYPE_CHECKING:  # pragma: no cover
diff --git a/flask_security/forms.py b/flask_security/forms.py
@@ -6,7 +6,7 @@
 
 :copyright: (c) 2012 by Matt Wright.
 :copyright: (c) 2017 by CERN.
-:copyright: (c) 2019-2025 by J. Christopher Wagner (jwag).
+:copyright: (c) 2019-2026 by J. Christopher Wagner (jwag).
 :license: MIT, see LICENSE for more details.
 """
 
@@ -596,12 +596,14 @@ def validate(self, **kwargs: t.Any) -> bool:
             return False
         if not self.user.password:
             # This is result of PASSWORD_REQUIRED=False and UNIFIED_SIGNIN
+            self.user.track_failed_authn(request.endpoint, "password")
             self.password.errors.append(get_message("INVALID_PASSWORD")[0])
             # Reduce timing variation between existing and non-existing users
             hash_password(self.password.data)
             return False
         self.password.data = _security.password_util.normalize(self.password.data)
         if not self.user.verify_and_update_password(self.password.data):
+            self.user.track_failed_authn(request.endpoint, "password")
             self.password.errors.append(get_message("INVALID_PASSWORD")[0])
             return False
 
@@ -661,6 +663,7 @@ def validate(self, **kwargs: t.Any) -> bool:
         self.password.data = _security.password_util.normalize(self.password.data)
         assert isinstance(self.password.errors, list)
         if not self.user.verify_and_update_password(self.password.data):
+            self.user.track_failed_authn(request.endpoint, "password")
             self.password.errors.append(get_message("INVALID_PASSWORD")[0])
             return False
         return True
@@ -1012,6 +1015,7 @@ def __init__(self, *args: t.Any, **kwargs: t.Any):
         self.primary_method: str = ""
         self.tf_totp_secret: str = ""
         self.user: UserMixin | None = None  # set by view
+        self.is_setup: bool = False
 
     def validate(self, **kwargs: t.Any) -> bool:
         if not super().validate(**kwargs):  # pragma: no cover
@@ -1038,6 +1042,8 @@ def validate(self, **kwargs: t.Any) -> bool:
             user=self.user,
             window=self.window,
         ):
+            if not self.is_setup:
+                self.user.track_failed_authn(request.endpoint, "code", True)
             self.code.errors.append(get_message("TWO_FACTOR_INVALID_TOKEN")[0])
             return False
 
diff --git a/flask_security/oauth_glue.py b/flask_security/oauth_glue.py
@@ -136,7 +136,7 @@ def oauthresponse(name: str) -> ResponseValue:
         if response:
             return response
         # two-factor not required - login user
-        login_user(user)
+        login_user(user, authn_via=["oauth"])
         if cv("REDIRECT_BEHAVIOR") == "spa":
             redirect_url = get_url(
                 cv("POST_OAUTH_LOGIN_VIEW"), qparams=user.get_redirect_qparams()
diff --git a/flask_security/signals.py b/flask_security/signals.py
@@ -5,7 +5,7 @@
 Flask-Security signals module
 
 :copyright: (c) 2012 by Matt Wright.
-:copyright: (c) 2019-2025 by J. Christopher Wagner (jwag).
+:copyright: (c) 2019-2026 by J. Christopher Wagner (jwag).
 :license: MIT, see LICENSE for more details.
 """
 
@@ -17,6 +17,8 @@
 
 user_unauthenticated = signals.signal("user-unauthenticated")
 
+user_failed_authn = signals.signal("user-failed-authn")
+
 user_registered = signals.signal("user-registered")
 
 # For cases of RETURN_GENERIC_RESPONSES with existing email/username
diff --git a/flask_security/unified_signin.py b/flask_security/unified_signin.py
@@ -213,6 +213,7 @@ def validate2(self) -> bool:
                         ok = True
                         break
             if not ok:
+                self.user.track_failed_authn(request.endpoint, "passcode")
                 self.passcode.errors.append(get_message("INVALID_PASSWORD_CODE")[0])
                 return False
 
@@ -721,6 +722,8 @@ def us_verify_link() -> ResponseValue:
         window=cv("US_TOKEN_VALIDITY"),
     ):
         m, c = generic_message("INVALID_CODE", "GENERIC_AUTHN_FAILED")
+        user.track_failed_authn(request.endpoint, "passcode")
+
         if cv("REDIRECT_BEHAVIOR") == "spa":
             return redirect(
                 get_url(
diff --git a/flask_security/views.py b/flask_security/views.py
@@ -951,6 +951,7 @@ def two_factor_setup_validate(token: str) -> ResponseValue:
     form.tf_totp_secret = totp_secret
     form.primary_method = method
     form.user = current_user
+    form.is_setup = True
 
     if form.validate_on_submit():
         tf_clean_session()  # until we completely remove session based setup/state
@@ -1040,6 +1041,7 @@ def two_factor_token_validation():
             logout_user()
             return tf_illegal_state(form, cv("TWO_FACTOR_ERROR_VIEW"))
         form.user = current_user
+        form.is_setup = True
 
     form.primary_method = pm
     form.tf_totp_secret = totp_secret
diff --git a/flask_security/webauthn.py b/flask_security/webauthn.py
@@ -4,7 +4,7 @@
 
 Flask-Security WebAuthn module
 
-:copyright: (c) 2021-2025 by J. Christopher Wagner (jwag).
+:copyright: (c) 2021-2026 by J. Christopher Wagner (jwag).
 :license: MIT, see LICENSE for more details.
 
 This implements support for webauthn/FIDO2 Level 2 using the py_webauthn package.
@@ -342,7 +342,7 @@ def validate(self, **kwargs: t.Any) -> bool:
             credential_public_key=self.cred.public_key,
             credential_current_sign_count=self.cred.sign_count,
         )
-        # Start by verifying requiring user_verification - if that succeeds then
+        # Start by verifying requiring user_verification - if that succeeds, then
         # this authn could be used for both primary and secondary.
         # If it fails, then try to verify with user_verification == False - unless
         # as part of signin the app required user_verification (as stored in the state)
@@ -358,6 +358,7 @@ def validate(self, **kwargs: t.Any) -> bool:
                 self.credential.errors.append(
                     get_message("WEBAUTHN_NO_VERIFY", cause=str(exc))[0]
                 )
+                self.user.track_failed_authn(request.endpoint, "passkey")
                 return False
         return True
 
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -986,6 +986,38 @@ def client_nc(request, app, sqlalchemy_datastore):
     return app.test_client(use_cookies=False)
 
 
+@pytest.fixture()
+def signals():
+    """Tracks signal calls for testing purposes.
+
+    This sets up a receiver for all signals and tracks the data
+    """
+    from flask_security import signals
+    import inspect
+    import blinker
+
+    signal_calls = dict()
+    def_signals = inspect.getmembers(
+        signals, lambda v: isinstance(v, blinker.NamedSignal)
+    )
+
+    for name, sig in def_signals:
+        sc = signal_calls[name] = []
+
+        def _on(app, _sc=sc, **kwargs):
+            assert isinstance(app, Flask)
+            # user argument is an ORM structure so may not be available in a test
+            # outside of the request - capture value(s) here
+            user = kwargs.get("user", None)
+            if user:
+                assert isinstance(kwargs["user"], UserMixin)
+                kwargs["user_email"] = user.email
+            _sc.append(kwargs)
+
+        sig.connect(_on, weak=False)
+    return signal_calls
+
+
 @pytest.fixture(
     params=[
         "cl-fsqlalchemy",
diff --git a/tests/test_basic.py b/tests/test_basic.py
@@ -387,9 +387,13 @@ def test_invalid_user(client, get_message):
     assert get_message("USER_DOES_NOT_EXIST") in response.data
 
 
-def test_bad_password(client, get_message):
+def test_bad_password(client, get_message, signals):
     response = authenticate(client, password="bogus")
     assert get_message("INVALID_PASSWORD") in response.data
+    assert signals["user_failed_authn"][0]["endpoint"] == "security.login"
+    assert signals["user_failed_authn"][0]["user_email"] == "matt@lp.com"
+    assert signals["user_failed_authn"][0]["auth_type"] == "password"
+    assert not signals["user_failed_authn"][0]["tfa"]
 
 
 def test_inactive_user(client, get_message):
diff --git a/tests/test_misc.py b/tests/test_misc.py
@@ -27,6 +27,7 @@
     capture_flashes,
     capture_reset_password_requests,
     check_location,
+    check_signals,
     check_xlation,
     get_csrf_token,
     init_app_with_options,
@@ -1168,7 +1169,7 @@ def test_verify_fresh(app, client, get_message):
     assert b"Fresh Only" in response.data
 
 
-def test_verify_fresh_json(app, client, get_message):
+def test_verify_fresh_json(app, client, get_message, signals):
     # Hit a fresh-required endpoint and walk through verify
     authenticate(client)
     headers = {"Accept": "application/json", "Content-Type": "application/json"}
@@ -1187,6 +1188,10 @@ def test_verify_fresh_json(app, client, get_message):
         "/verify", json=dict(password="not my password"), headers=headers
     )
     assert response.status_code == 400
+    assert signals["user_failed_authn"][0]["endpoint"] == "security.verify"
+    assert signals["user_failed_authn"][0]["user"].email == "matt@lp.com"
+    assert signals["user_failed_authn"][0]["auth_type"] == "password"
+    assert not signals["user_failed_authn"][0]["tfa"]
 
     response = client.post("/verify", json=dict(password="password"), headers=headers)
     assert response.status_code == 200
@@ -1195,10 +1200,11 @@ def test_verify_fresh_json(app, client, get_message):
     response = client.get("/fresh", headers=headers)
     assert response.status_code == 200
     assert response.json["title"] == "Fresh Only"
+    check_signals(signals, "user_failed_authn")
 
 
 @pytest.mark.changeable()
-def test_verify_pwd_json(app, client, get_message):
+def test_verify_pwd_json(app, client, get_message, signals):
     # Make sure verify accepts a normalized and original password.
     authenticate(client)
     headers = {"Accept": "application/json", "Content-Type": "application/json"}
@@ -1227,6 +1233,7 @@ def test_verify_pwd_json(app, client, get_message):
         headers=headers,
     )
     assert response.status_code == 200
+    check_signals(signals, "password_changed")
 
 
 @pytest.mark.settings(verify_url="/auth/")
diff --git a/tests/test_two_factor.py b/tests/test_two_factor.py
diff --git a/tests/test_unified_signin.py b/tests/test_unified_signin.py
diff --git a/tests/test_utils.py b/tests/test_utils.py
diff --git a/tests/test_webauthn.py b/tests/test_webauthn.py
