Fix password MFA verification feedback

Author: Marc Schlaeppi

source/app/blueprints/pages/login/login_routes.py

@@ -46,7 +46,7 @@
 from app.datamgmt.manage.manage_users_db import create_user
 from app.datamgmt.manage.manage_users_db import update_user_groups
 from app.datamgmt.manage.manage_users_db import get_user
-from app.forms import LoginForm, MFASetupForm
+from app.forms import LoginForm, MFASetupForm, MFAVerifyForm
 from app.blueprints.iris_user import iris_current_user
 from app.iris_engine.utils.tracker import track_activity
 from app.datamgmt.manage.manage_groups_db import get_groups_list
@@ -154,8 +154,7 @@ def login():
 
         form = LoginForm(request.form)
 
-        # check if both http method is POST and form is valid on submit
-        if not form.is_submitted() and not form.validate():
+        if not form.validate_on_submit():
             return _render_template_login(form, None)
 
         # assign form data to variables
@@ -406,20 +405,27 @@ def _clear_pre_mfa_state(preserve_lockout=False):
     session.pop("pending_mfa_secret", None)
     if not preserve_lockout:
         session.pop("mfa_fail_count", None)
+        session.pop("mfa_fail_user_id", None)
         session.pop("mfa_lockout_until", None)
 
 
-def _mfa_is_locked_out():
+def _mfa_is_locked_out(user):
+    if session.get("mfa_fail_user_id") != user.id:
+        return False
     locked_until = session.get("mfa_lockout_until")
     if locked_until and locked_until > time.time():
         return True
     if locked_until and locked_until <= time.time():
         session.pop("mfa_lockout_until", None)
+        session.pop("mfa_fail_user_id", None)
         session["mfa_fail_count"] = 0
     return False
 
 
 def _register_mfa_failure(user, reason):
+    if session.get("mfa_fail_user_id") != user.id:
+        session["mfa_fail_count"] = 0
+    session["mfa_fail_user_id"] = user.id
     session["mfa_fail_count"] = session.get("mfa_fail_count", 0) + 1
     track_activity(
         f"Failed MFA {reason} for user {user.user} "
@@ -433,6 +439,8 @@ def _register_mfa_failure(user, reason):
         # the lockout timestamp so a fresh /login cannot wipe it.
         _clear_pre_mfa_state(preserve_lockout=True)
         session.pop("username", None)
+        return True
+    return False
 
 
 @app.route("/auth/mfa-setup", methods=["GET", "POST"])
@@ -449,13 +457,13 @@ def mfa_setup():
     if user.mfa_setup_complete and user.mfa_secrets:
         return redirect(url_for("mfa_verify"))
 
-    if _mfa_is_locked_out():
+    if _mfa_is_locked_out(user):
         flash("Too many attempts. Please try again later.", "danger")
         return redirect(url_for("login.login"))
 
     form = MFASetupForm()
 
-    if form.submit() and form.validate():
+    if form.validate_on_submit():
 
         token = form.token.data
         user_password = form.user_password.data
@@ -469,7 +477,7 @@ def mfa_setup():
 
         totp = pyotp.TOTP(mfa_secret)
 
-        if totp.verify(token):
+        if totp.verify(str(token).strip(), valid_window=1):
             has_valid_password = False
             if is_authentication_ldap() is True:
                 if validate_ldap_login(
@@ -483,7 +491,10 @@ def mfa_setup():
                 has_valid_password = True
 
             if not has_valid_password:
-                _register_mfa_failure(user, "setup (invalid password)")
+                locked_out = _register_mfa_failure(user, "setup (invalid password)")
+                if locked_out:
+                    flash("Too many attempts. Please try again later.", "danger")
+                    return redirect(url_for("login.login"))
                 flash("Invalid password. Please try again.", "danger")
                 return render_template("mfa_setup.html", form=form)
 
@@ -501,7 +512,10 @@ def mfa_setup():
             session["mfa_verified_for_user_id"] = user.id
             _clear_pre_mfa_state()
             return wrap_login_user(user)
-        _register_mfa_failure(user, "setup (invalid token)")
+        locked_out = _register_mfa_failure(user, "setup (invalid token)")
+        if locked_out:
+            flash("Too many attempts. Please try again later.", "danger")
+            return redirect(url_for("login.login"))
         flash("Invalid token or password. Please try again.", "danger")
 
     # Generate a fresh secret on every GET and stash it in the session. The
@@ -536,21 +550,16 @@ def mfa_verify():
         )
         return redirect(url_for("mfa_setup"))
 
-    if _mfa_is_locked_out():
+    if _mfa_is_locked_out(user):
         flash("Too many attempts. Please try again later.", "danger")
         return redirect(url_for("login.login"))
 
-    form = MFASetupForm()
-    form.user_password.data = "not required for verification"
+    form = MFAVerifyForm()
 
-    if form.submit() and form.validate():
+    if form.validate_on_submit():
         token = form.token.data
-        if not token:
-            flash("Token is required.", "danger")
-            return render_template("mfa_verify.html", form=form)
-
         totp = pyotp.TOTP(user.mfa_secrets)
-        if totp.verify(token):
+        if totp.verify(str(token).strip(), valid_window=1):
             track_activity(
                 f"MFA verification successful for user {user.user}",
                 ctx_less=True, display_in_ui=False,
@@ -561,7 +570,10 @@ def mfa_verify():
             session["mfa_verified_for_user_id"] = user.id
             _clear_pre_mfa_state()
             return wrap_login_user(user)
-        _register_mfa_failure(user, "verification (invalid token)")
+        locked_out = _register_mfa_failure(user, "verification (invalid token)")
+        if locked_out:
+            flash("Too many attempts. Please try again later.", "danger")
+            return redirect(url_for("login.login"))
         flash("Invalid token. Please try again.", "danger")
 
     return render_template("mfa_verify.html", form=form)

source/app/blueprints/pages/login/templates/_auth_messages.html

@@ -0,0 +1,17 @@
+{% with messages = get_flashed_messages(with_categories=true) %}
+    {% for category, message in messages %}
+        <div class="alert alert-{{ category if category in ['danger', 'warning', 'success', 'info'] else 'info' }}" role="alert">
+            {{ message }}
+        </div>
+    {% endfor %}
+{% endwith %}
+
+{% if form and form.errors %}
+    <div class="alert alert-danger" role="alert">
+        {% for field_errors in form.errors.values() %}
+            {% for error in field_errors %}
+                <div>{{ error }}</div>
+            {% endfor %}
+        {% endfor %}
+    </div>
+{% endif %}

source/app/blueprints/pages/login/templates/login.html

@@ -22,6 +22,8 @@ <h3 class="text-white">{{ organisation_name }}</h3>
         <div class="login__right_part col-xs-12 col-md-4">
             <h3 class="login__form_title">Sign In</h3>
 
+            {% include "_auth_messages.html" %}
+
             {% if auth_type == "oidc" %}
                 <a href="{{ url_for('login.oidc_login') }}" class="btn btn-primary login__submit_button login__form_title">OIDC Sign In</a>
             {% else %}
@@ -63,4 +65,4 @@ <h3 class="login__form_title">Sign In</h3>
     <script type="module" src="/static/assets/js/iris/login.js"></script>
 </body>
 
-{% endblock content %}
+{% endblock content %}

source/app/blueprints/pages/login/templates/mfa_setup.html

@@ -27,6 +27,7 @@ <h1 class="text-white mb-3">Setup MFA</h1>
                 <div class="d-flex flex-column justify-content-center align-items-center" style="height: 100%;">
                     <h3 class="login__form_title">Your organisation requires to setup MFA</h3>
                     <p>Scan the QR code with your authenticator app and enter the token and your password.</p>
+                    {% include "_auth_messages.html" %}
                       <form method="POST" action="">
                         <div class="col-md-12 col-lg-12 col-sm-12">
                           <div class="form-row ml-2">
@@ -54,4 +55,4 @@ <h3 class="login__form_title">Your organisation requires to setup MFA</h3>
 
 {% block javascripts %}
 
-{% endblock javascripts %}
+{% endblock javascripts %}

source/app/blueprints/pages/login/templates/mfa_verify.html

@@ -25,6 +25,7 @@
                 <div class="d-flex flex-column justify-content-center align-items-center" style="height: 100%;">
                     <h3 class="login__form_title">Verify MFA</h3>
                     <p>Enter the token displayed on your authentication app</p>
+                    {% include "_auth_messages.html" %}
                     <form method="POST" action="">
                         <div class="col-md-12 col-lg-12 col-sm-12">
                             <div class="form-row ml-2">
@@ -49,4 +50,4 @@ <h3 class="login__form_title">Verify MFA</h3>
 
 {% block javascripts %}
 
-{% endblock javascripts %}
+{% endblock javascripts %}

source/app/business/auth.py

@@ -147,16 +147,19 @@ def wrap_login_user(user, is_oidc=False):
             # here — an attacker who re-POSTs /login mustn't be able to zero
             # out the fail counter and get a fresh burst of 5 tokens.
             locked_until = session.get('mfa_lockout_until')
-            if locked_until and locked_until > time.time():
+            locked_user_id = session.get('mfa_fail_user_id')
+            if locked_user_id == user.id and locked_until and locked_until > time.time():
                 flash('Too many attempts. Please try again later.', 'danger')
                 return redirect(url_for('login.login'))
 
             # Mark this browser session as the one that just passed password
             # auth for this user. mfa_setup / mfa_verify will refuse to run
             # for any other user id, preventing cross-user MFA handler abuse.
+            same_failed_user = session.get('mfa_fail_user_id') == user.id
             session['pre_mfa_user_id'] = user.id
-            session['mfa_fail_count'] = 0
-            session.pop('mfa_lockout_until', None)
+            if not same_failed_user:
+                session['mfa_fail_count'] = 0
+                session.pop('mfa_lockout_until', None)
             session.pop('pending_mfa_secret', None)
             return redirect(url_for('mfa_verify'))
 

source/app/forms.py

@@ -42,6 +42,11 @@ class MFASetupForm(FlaskForm):
     submit = SubmitField('Verify')
 
 
+class MFAVerifyForm(FlaskForm):
+    token = StringField('Token', validators=[DataRequired()])
+    submit = SubmitField('Verify')
+
+
 class RegisterForm(FlaskForm):
     name = StringField(u'Name', validators=[DataRequired()])
     username = StringField(u'Username', validators=[DataRequired()])

tests/tests_auth_mfa.py

@@ -33,6 +33,7 @@
 
 from unittest import TestCase
 from uuid import uuid4
+from html.parser import HTMLParser
 
 import pyotp
 import requests
@@ -45,6 +46,25 @@
 _PASSWORD = 'aA.1234567890'
 
 
+class _CsrfTokenParser(HTMLParser):
+    def __init__(self):
+        super().__init__()
+        self.token = None
+
+    def handle_starttag(self, tag, attrs):
+        attributes = dict(attrs)
+        if tag == 'input' and attributes.get('name') == 'csrf_token':
+            self.token = attributes.get('value')
+
+
+def _csrf_token(response):
+    parser = _CsrfTokenParser()
+    parser.feed(response.text)
+    if not parser.token:
+        raise AssertionError('CSRF token not found in authentication form')
+    return parser.token
+
+
 def _login(username, password):
     """POST /api/v2/auth/login and return the parsed top-level JSON body.
 
@@ -251,3 +271,63 @@ def test_mfa_verify_should_lock_out_after_repeated_failures(self):
         # surfaces verbatim to the user.
         self.assertIn('Too many MFA attempts',
                       locked.json().get('message', ''))
+
+    def test_legacy_mfa_failures_should_survive_password_reentry(self):
+        user_name, _ = self._provision_mfa_user()
+        browser = requests.Session()
+        login_url = parse.urljoin(API_URL, '/login')
+
+        login_page = browser.get(login_url)
+        password_response = browser.post(
+            login_url,
+            data={
+                'csrf_token': _csrf_token(login_page),
+                'username': user_name,
+                'password': _PASSWORD,
+            },
+            allow_redirects=False,
+        )
+        self.assertEqual(302, password_response.status_code)
+        self.assertIn('/auth/mfa-verify', password_response.headers['Location'])
+
+        challenge_url = parse.urljoin(API_URL, '/auth/mfa-verify')
+        challenge_page = browser.get(challenge_url)
+        for _ in range(3):
+            challenge_page = browser.post(
+                challenge_url,
+                data={
+                    'csrf_token': _csrf_token(challenge_page),
+                    'token': '000000',
+                },
+            )
+            self.assertIn('Invalid token', challenge_page.text)
+
+        login_page = browser.get(login_url)
+        password_response = browser.post(
+            login_url,
+            data={
+                'csrf_token': _csrf_token(login_page),
+                'username': user_name,
+                'password': _PASSWORD,
+            },
+        )
+        self.assertIn('Verify MFA', password_response.text)
+
+        challenge_page = password_response
+        challenge_page = browser.post(
+            challenge_url,
+            data={
+                'csrf_token': _csrf_token(challenge_page),
+                'token': '000000',
+            },
+        )
+        self.assertIn('Invalid token', challenge_page.text)
+
+        locked_page = browser.post(
+            challenge_url,
+            data={
+                'csrf_token': _csrf_token(challenge_page),
+                'token': '000000',
+            },
+        )
+        self.assertIn('Too many attempts. Please try again later.', locked_page.text)