Add 2FA requirement for sysadmin

pyproject.toml

@@ -52,7 +52,6 @@ dependencies = [
     "flask-mail>=0.10.0,<1.0.0",
     "flask-principal>=0.4.0,<1.0.0",
     "flask-restx>=1.3.0,<2.0.0",
-    "flask-security-too>=5.1.2,<6.0.0",
     "flask-sitemap>=0.4.0,<1.0.0",
     "flask-storage>=1.4.0,<2.0.0",
     "flask-wtf>=1.2.2,<2.0.0",
@@ -93,6 +92,7 @@ dependencies = [
     "wtforms[email]>=3.2.1,<4.0.0",
     "wtforms-json>=0.3.5,<1.0.0",
     "qrcode>=8.2,<9.0",
+    "flask-security",
 ]
 
 [project.urls]
@@ -190,3 +190,6 @@ udata = ["templates/*", "translations/*"]
 
 [tool.wheel]
 universal = true
+
+[tool.uv.sources]
+flask-security = { git = "https://github.com/pallets-eco/flask-security.git" }

udata/core/user/models.py

@@ -158,6 +158,43 @@ def organizations(self):
     def sysadmin(self):
         return self.has_role("admin")
 
+    def check_tf_required(
+        self, tf_setup_methods: list[tuple[str, str]], tf_fresh: bool
+    ) -> tuple[bool, list[tuple[str, str]]]:
+        """Check if current user requires two-factor authentication.
+
+        :param tf_setup_methods: A tuple of (two_factor method, label) - methods
+            the user has already set up (from all two-factor implementations)
+        :param tf_fresh: if True then user has recently completed
+            two-factor authentication on the requesting device
+        :return: Whether TFA is required for this user and a possibly augmented
+            list of allowable methods
+
+        Overrides default implementation in Flask-Security to add configurable 2FA requirement for sysadmins.
+        This is called AFTER the user has successfully authenticated.
+        """
+        if (
+            current_app.config["SECURITY_TWO_FACTOR_REQUIRED"]
+            or len(tf_setup_methods) > 0
+            or (current_app.config["SECURITY_TWO_FACTOR_REQUIRED_FOR_ADMIN"] and self.sysadmin)
+        ):
+            if current_app.config["SECURITY_TWO_FACTOR_ALWAYS_VALIDATE"] or not tf_fresh:
+                return True, tf_setup_methods
+        return False, tf_setup_methods
+
+    def check_tf_required_setup(self) -> bool:
+        """Check if current user requires two-factor authentication.
+        This is called as part of two-factor setup to inform the caller
+
+        N.B. this is only called from tf-setup - not from webauthn and
+        is only used to improve UX - the above method check_tf_required is the
+        definitive answer in the authentication path.
+        Overrides default implementation in Flask-Security to add configurable 2FA requirement for sysadmins.
+        """
+        return current_app.config["SECURITY_TWO_FACTOR_REQUIRED"] or (
+            current_app.config["SECURITY_TWO_FACTOR_REQUIRED_FOR_ADMIN"] and self.sysadmin
+        )
+
     def self_web_url(self, **kwargs):
         return cdata_url(f"/users/{self._link_id(**kwargs)}", **kwargs)
 

udata/settings.py

@@ -139,6 +139,7 @@ class Defaults(object):
     # Two-Factor Authentication settings
     SECURITY_TWO_FACTOR = False
     SECURITY_TWO_FACTOR_REQUIRED = False  # Not required by default
+    SECURITY_TWO_FACTOR_REQUIRED_FOR_ADMIN = False
     SECURITY_TWO_FACTOR_ENABLED_METHODS = ["authenticator"]
     SECURITY_TOTP_SECRETS = {"1": "the udata totp secret"}
     SECURITY_TOTP_ISSUER = "udata"

udata/tests/api/test_security_api.py

@@ -5,6 +5,7 @@
 from flask_security.recoverable import generate_reset_password_token
 
 from udata.commands.fixtures import UserFactory
+from udata.core.user.factories import AdminFactory
 from udata.i18n import lazy_gettext as _
 from udata.tests.api import PytestOnlyAPITestCase
 from udata.tests.helpers import capture_mails
@@ -136,7 +137,7 @@ def test_2fa_disabled_by_default(self):
             url_for("security.login"), {"email": user.email, "password": "password123"}
         )
         self.assertStatus(response, 200)
-        assert "tf_required" not in response.json["response"]
+        assert response.json["response"]["tf_required"] is False
         assert "tf_state" not in response.json["response"]
 
         # Should be None by default (2FA not set up)
@@ -187,3 +188,53 @@ def test_reset_password(self):
             },
         )
         self.assertStatus(response, 200)
+
+    @pytest.mark.options(
+        SECURITY_TWO_FACTOR_REQUIRED=False, SECURITY_TWO_FACTOR_REQUIRED_FOR_ADMIN=True
+    )
+    def test_sysadmin_requires_2fa_even_when_globally_disabled(self):
+        today = datetime.now(UTC)
+        admin = AdminFactory(password="password123", confirmed_at=today)
+
+        # Sysadmin should require 2FA even when globally disabled
+        response = self.post(
+            url_for("security.login"), {"email": admin.email, "password": "password123"}
+        )
+        self.assertStatus(response, 200)
+        assert response.json["response"]["tf_required"] is True
+        assert response.json["response"]["tf_state"] == "setup_from_login"
+
+    @pytest.mark.options(
+        SECURITY_TWO_FACTOR_REQUIRED=False, SECURITY_TWO_FACTOR_REQUIRED_FOR_ADMIN=True
+    )
+    def test_regular_user_no_2fa_when_globally_disabled(self):
+        today = datetime.now(UTC)
+        user = UserFactory(password="password123", confirmed_at=today)
+
+        # Regular user should not require 2FA when globally disabled
+        response = self.post(
+            url_for("security.login"), {"email": user.email, "password": "password123"}
+        )
+        self.assertStatus(response, 200)
+        assert response.json["response"]["tf_required"] is False
+        assert "tf_state" not in response.json["response"]
+
+    @pytest.mark.options(
+        SECURITY_TWO_FACTOR_REQUIRED=False, SECURITY_TWO_FACTOR_REQUIRED_FOR_ADMIN=False
+    )
+    def test_user_with_existing_2fa_needs_validation(self):
+        today = datetime.now(UTC)
+        user = UserFactory(
+            password="password123",
+            confirmed_at=today,
+            tf_primary_method="authenticator",
+            tf_totp_secret="test_secret",
+        )
+
+        # Should require 2FA token validation
+        response = self.post(
+            url_for("security.login"), {"email": user.email, "password": "password123"}
+        )
+        assert response.json["response"]["tf_required"] is True
+        assert response.json["response"]["tf_state"] == "ready"
+        assert response.json["response"]["tf_primary_method"] == "authenticator"