[FIX] Fixed MFA flow

Author: Paul Amicelli

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

@@ -23,6 +23,7 @@
 import random
 import string
 import json
+import time
 from flask import Blueprint, flash
 from flask import redirect
 from flask import render_template
@@ -354,15 +355,98 @@ def oidc_authorise():
         return wrap_login_user(user, is_oidc=True)
 
 
+# MFA hardening constants. Values are deliberately conservative — a legitimate
+# user mistypes their token occasionally; an attacker brute-forcing 10^6 TOTP
+# codes should not be able to linearly grind them.
+_MFA_MAX_ATTEMPTS = 5
+_MFA_LOCKOUT_SECONDS = 15 * 60
+
+
+def _get_pre_mfa_user():
+    """Return the user this session just passed password auth for, or None.
+
+    The pre_mfa_user_id marker is set exclusively by wrap_login_user after a
+    successful password (or LDAP) check. Any MFA handler that runs without it
+    is being hit directly by an attacker and must be refused.
+    """
+    pre_mfa_user_id = session.get('pre_mfa_user_id')
+    if pre_mfa_user_id is None:
+        return None
+    return get_user(pre_mfa_user_id, id_key='id')
+
+
+def _clear_pre_mfa_state(preserve_lockout=False):
+    """Clear the markers that prove this session just passed password auth.
+
+    preserve_lockout: when True, keep the lockout timestamp and fail counter
+    so that an attacker cannot wipe a brute-force lockout simply by hitting
+    /login again.
+    """
+    session.pop('pre_mfa_user_id', None)
+    session.pop('pending_mfa_secret', None)
+    if not preserve_lockout:
+        session.pop('mfa_fail_count', None)
+        session.pop('mfa_lockout_until', None)
+
+
+def _mfa_is_locked_out():
+    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():
+        # Lockout expired — reset the counter so the user gets a fresh window.
+        session.pop('mfa_lockout_until', None)
+        session['mfa_fail_count'] = 0
+    return False
+
+
+def _register_mfa_failure(user, reason):
+    session['mfa_fail_count'] = session.get('mfa_fail_count', 0) + 1
+    track_activity(
+        f'Failed MFA {reason} for user {user.user} (attempt {session["mfa_fail_count"]}/{_MFA_MAX_ATTEMPTS})',
+        ctx_less=True, display_in_ui=False
+    )
+    if session['mfa_fail_count'] >= _MFA_MAX_ATTEMPTS:
+        session['mfa_lockout_until'] = time.time() + _MFA_LOCKOUT_SECONDS
+        # Drop the pending-MFA marker so the attacker has to go back through
+        # password auth before they get another burst of attempts. Preserve
+        # the lockout timestamp so a fresh /login cannot wipe it.
+        _clear_pre_mfa_state(preserve_lockout=True)
+        session.pop('username', None)
+
+
 @app.route("/auth/mfa-setup", methods=["GET", "POST"])
 def mfa_setup():
-    user = retrieve_user_by_username(username=session["username"])
+    user = _get_pre_mfa_user()
+    if user is None:
+        return redirect(url_for("login.login"))
+
+    # mfa_setup is only for users who haven't completed MFA yet (fresh
+    # accounts, or admin-reset accounts where mfa_setup_complete was cleared).
+    # A user who already has MFA enrolled must go through mfa_verify — this
+    # prevents the "I know the password, let me overwrite the enrolled secret
+    # with one I control" bypass.
+    if user.mfa_setup_complete and user.mfa_secrets:
+        return redirect(url_for("mfa_verify"))
+
+    if _mfa_is_locked_out():
+        flash('Too many attempts. Please try again later.', 'danger')
+        return redirect(url_for("login.login"))
+
     form = MFASetupForm()
 
     if form.submit() and form.validate():
+
         token = form.token.data
-        mfa_secret = form.mfa_secret.data
         user_password = form.user_password.data
+        # The secret MUST come from the server-side session, never from the
+        # submitted form. Trusting form.mfa_secret let an attacker enrol a
+        # secret of their choosing and immediately log in.
+        mfa_secret = session.get('pending_mfa_secret')
+        if not mfa_secret:
+            flash('MFA setup expired. Please restart the login flow.', 'danger')
+            return redirect(url_for("login.login"))
+
         totp = pyotp.TOTP(mfa_secret)
 
         if totp.verify(token):
@@ -379,36 +463,36 @@ def mfa_setup():
                 has_valid_password = True
 
             if not has_valid_password:
-                track_activity(
-                    f"Failed MFA setup for user {user.user}. Invalid password.",
-                    ctx_less=True,
-                    display_in_ui=False,
-                )
+                _register_mfa_failure(user, 'setup (invalid password)')
                 flash("Invalid password. Please try again.", "danger")
                 return render_template("mfa_setup.html", form=form)
 
             user.mfa_secrets = mfa_secret
             user.mfa_setup_complete = True
             db.session.commit()
-            session["mfa_verified"] = False
             track_activity(
                 f"MFA setup successful for user {user.user}",
                 ctx_less=True,
                 display_in_ui=False,
             )
+            # Setup succeeded — promote this session to MFA-verified for this
+            # user and clear the pre-MFA markers. Without this, wrap_login_user
+            # would loop straight back to mfa_verify.
+            session['mfa_verified_for_user_id'] = user.id
+            _clear_pre_mfa_state()
             return wrap_login_user(user)
-        track_activity(
-            f"Failed MFA setup for user {user.user}. Invalid token.",
-            ctx_less=True,
-            display_in_ui=False,
-        )
-        flash("Invalid token or password. Please try again.", "danger")
+        else:
+            _register_mfa_failure(user, 'setup (invalid token)')
+            flash("Invalid token or password. Please try again.", "danger")
 
+    # Generate a fresh secret on every GET and stash it in the session. The
+    # client only sees the QR code and the base32 key for manual entry; it
+    # never sends the secret back.
     temp_otp_secret = pyotp.random_base32()
+    session['pending_mfa_secret'] = temp_otp_secret
     otp_uri = pyotp.TOTP(temp_otp_secret).provisioning_uri(
         user.email, issuer_name="IRIS"
     )
-    form.mfa_secret.data = temp_otp_secret
     img = qrcode.make(otp_uri)
     buf = io.BytesIO()
     img.save(buf, format="PNG")
@@ -421,11 +505,10 @@ def mfa_setup():
 
 @app.route("/auth/mfa-verify", methods=["GET", "POST"])
 def mfa_verify():
-    if "username" not in session:
+    user = _get_pre_mfa_user()
+    if user is None:
         return redirect(url_for("login.login"))
 
-    user = retrieve_user_by_username(username=session["username"])
-
     # Redirect user to MFA setup if MFA is not fully set up
     if not user.mfa_secrets or not user.mfa_setup_complete:
         track_activity(
@@ -435,6 +518,10 @@ def mfa_verify():
         )
         return redirect(url_for("mfa_setup"))
 
+    if _mfa_is_locked_out():
+        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"
 
@@ -446,19 +533,19 @@ def mfa_verify():
 
         totp = pyotp.TOTP(user.mfa_secrets)
         if totp.verify(token):
-            session.pop("username", None)
-            session["mfa_verified"] = True
             track_activity(
                 f"MFA verification successful for user {user.user}",
                 ctx_less=True,
                 display_in_ui=False,
             )
+            # Bind the MFA-verified marker to this specific user id so that a
+            # later login attempt for a different user on the same browser
+            # session cannot reuse it.
+            session['mfa_verified_for_user_id'] = user.id
+            _clear_pre_mfa_state()
             return wrap_login_user(user)
-        track_activity(
-            f"Failed MFA verification for user {user.user}. Invalid token.",
-            ctx_less=True,
-            display_in_ui=False,
-        )
-        flash("Invalid token. Please try again.", "danger")
+        else:
+            _register_mfa_failure(user, 'verification (invalid token)')
+            flash("Invalid token. Please try again.", "danger")
 
     return render_template("mfa_verify.html", form=form)

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

@@ -32,7 +32,6 @@ <h3 class="login__form_title">Your organisation requires to setup MFA</h3>
                           <div class="form-row ml-2">
                             <div class="form-group col-12">
                                 {{ form.hidden_tag() }}
-                                {{ form.mfa_secret(size=32, class="hidden", style="display:None;") }}
                                 <label for="token">Password</label>
                                 {{ form.user_password(class="form-control") }}
 

source/app/business/auth.py

@@ -18,6 +18,7 @@
 
 from urllib.parse import urlsplit
 
+from flask import flash
 from flask import session
 from flask import redirect
 from flask import url_for
@@ -38,6 +39,7 @@
 from app.models.authorization import User
 
 import datetime
+import time
 import jwt
 
 
@@ -139,7 +141,26 @@ def wrap_login_user(user, is_oidc=False):
         app.config['SERVER_SETTINGS'] = get_server_settings_as_dict()
 
     if app.config['SERVER_SETTINGS']['enforce_mfa'] is True and is_oidc is False:
-        if "mfa_verified" not in session or session["mfa_verified"] is False:
+        # MFA state must be bound to the specific user who verified — a flat
+        # boolean would let a prior verified session admit a different user on
+        # the same browser (e.g. shared device, attacker knows user B's
+        # password and reuses user A's mfa_verified=True).
+        verified_for = session.get('mfa_verified_for_user_id')
+        if verified_for != user.id:
+            # If the session is currently MFA-locked out, don't reset state
+            # 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():
+                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.
+            session['pre_mfa_user_id'] = user.id
+            session['mfa_fail_count'] = 0
+            session.pop('mfa_lockout_until', None)
+            session.pop('pending_mfa_secret', None)
             return redirect(url_for('mfa_verify'))
 
     login_user(user)