Fix login: support email in user lookup

Author: Tome Petrovski

ckanext/auth/logic/action.py

@@ -4,7 +4,7 @@
 from typing import TypedDict
 
 import ckan.plugins.toolkit as tk
-from ckan import model, types
+from ckan import types
 from ckan.lib import captcha
 from ckan.logic import validate
 
@@ -29,8 +29,7 @@ def auth_2fa_user_login(
 ) -> LoginResponse:
     tk.check_access("auth_2fa_user_login", context, data_dict)
 
-    user = model.User.by_name(data_dict["login"])
-
+    user = utils.get_user_by_username_or_email(data_dict["login"])
     if not user:
         raise tk.ObjectNotFound("User not found")
 
@@ -65,7 +64,7 @@ def auth_2fa_check_credentials(
 ) -> LoginResponse:
     tk.check_access("auth_2fa_user_login", context, data_dict)
 
-    user = model.User.by_name(data_dict["login"])
+    user = utils.get_user_by_username_or_email(data_dict["login"])
 
     try:
         captcha.check_recaptcha(tk.request)

ckanext/auth/model.py

@@ -18,6 +18,7 @@
 from ckan.model.types import make_uuid
 
 import ckanext.auth.config as auth_config
+from ckanext.auth import utils
 from ckanext.auth.exceptions import ReplayAttackError
 
 log = logging.getLogger(__name__)
@@ -68,7 +69,7 @@ def get_for_user(cls, user_name: str) -> Self | None:
 
         :raises ValueError if the user_name is not provided
         """
-        user = model.User.get(user_name)
+        user = utils.get_user_by_username_or_email(user_name)
 
         if not user:
             raise tk.ObjectNotFound("User not found")

ckanext/auth/tests/test_utils.py

@@ -16,13 +16,17 @@
 @pytest.mark.usefixtures("with_plugins", "clean_db", "with_request_context")
 class TestSendVerificationCodeToUser:
     @mock.patch("ckan.lib.mailer.mail_user")
-    def test_send_verification_code_to_user(self, mocker, app, user):
+    def test_send_verification_code_to_user(self, mocker, user):
         mocker.return_value = True
         assert utils.send_verification_email_to_user(user["id"])
 
     def test_user_does_not_exist(self):
         assert not utils.send_verification_email_to_user("xxx")
 
+    @mock.patch("ckan.lib.mailer.mail_user")
+    def test_send_verification_code_to_user_with_email(self, mocker, user):
+        mocker.return_value = True
+        assert utils.send_verification_email_to_user(user["email"])
 
 @pytest.mark.usefixtures("with_plugins", "clean_db")
 class TestGetEmailVerificationCode:

ckanext/auth/utils.py

@@ -16,7 +16,6 @@
 from ckan.lib.redis import connect_to_redis
 from ckan.views.user import next_page_or_default, rotate_token
 
-from ckanext.auth import config
 from ckanext.auth import config as auth_config
 from ckanext.auth.exceptions import ReplayAttackError
 from ckanext.auth.model import UserSecret
@@ -84,8 +83,8 @@ def reset_all(cls) -> None:
             redis.delete(key)
 
 
-def send_verification_email_to_user(user_id: str) -> bool:
-    user = model.User.get(user_id)
+def send_verification_email_to_user(user_reference: str) -> bool:
+    user = get_user_by_username_or_email(user_reference)
 
     if not user or not user.email:
         return False
@@ -96,7 +95,7 @@ def send_verification_email_to_user(user_id: str) -> bool:
         "site_url": tk.config["ckan.site_url"],
         "site_title": tk.config["ckan.site_title"],
         "user_name": user.display_name,
-        "subject": tk._(config.get_2fa_subject()),
+        "subject": tk._(auth_config.get_2fa_subject()),
         "body": f"Your verification code is: {code}",
     }
 
@@ -138,23 +137,23 @@ def get_email_verification_code(user: model.User) -> str:
     return user_secret.get_code()
 
 
-def regenerate_user_secret(user_id: str) -> str:
+def regenerate_user_secret(user_reference: str) -> str:
     """Regenerate the secret for a user.
 
     Args:
-        user_id (str): The id of the user
+        user_reference (str): The user’s ID or email.
 
     Returns:
         str: The new secret
     """
-    user = model.User.get(user_id)
+    user = get_user_by_username_or_email(user_reference)
 
     if not user:
         raise tk.ObjectNotFound("User not found")
 
     user_secret = UserSecret.create_for_user(user.name)
 
-    log.debug("2FA: Rotated the 2fa secret for user %s", user_id)
+    log.debug("2FA: Rotated the 2fa secret for user %s", user.id)
 
     return cast(str, user_secret.secret)
 
@@ -205,13 +204,16 @@ def authenticate(identity: IdentityDict) -> model.User | model.AnonymousUser | N
     if LoginManager.is_login_blocked(identity["login"]):
         return None
 
-    if LoginManager.get_user_login_attempts(identity["login"]) > config.get_2fa_max_attempts():
+    if (
+        LoginManager.get_user_login_attempts(identity["login"])
+        > auth_config.get_2fa_max_attempts()
+    ):
         LoginManager.block_user_login(identity["login"])
 
     if not ckan_auth_result:
         return LoginManager.log_user_login_attempt(identity["login"])
 
-    if not config.is_2fa_enabled():
+    if not auth_config.is_2fa_enabled():
         LoginManager.reset_for_user(identity["login"])
         return ckan_auth_result
 
@@ -254,3 +256,7 @@ def authenticate_totp(user_name: str) -> str | None:
         )
     else:
         return user_name if result else None
+
+
+def get_user_by_username_or_email(user_reference: str) -> model.User | None:
+    return model.User.get(user_reference) or model.User.by_email(user_reference)

ckanext/auth/views.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import contextlib
 from collections.abc import Callable
 from functools import wraps
 from typing import Any, cast
@@ -9,7 +10,6 @@
 from flask.views import MethodView
 
 from ckan import model, types
-from ckan.lib import helpers
 from ckan.logic import parse_params
 from ckan.plugins import plugin_loaded
 from ckan.plugins import toolkit as tk
@@ -113,15 +113,17 @@ def _setup_totp_extra_vars(self, user_id: str) -> dict[str, Any]:
 def regenerate_secret(user_id: str):
     utils.regenerate_user_secret(user_id)
     tk.h.flash_success(tk._("Your 2FA secret has been regenerated."))
-    return helpers.redirect_to("auth.configure_2fa", user_id=user_id)
+    return tk.redirect_to("auth.configure_2fa", user_id=user_id)
 
 
 @auth.route("/send_verification_code", methods=["POST"])
 def send_verification_code() -> Response:
-    user_name: str = tk.get_or_bust(dict(tk.request.form), "login")
+    login: str = tk.get_or_bust(dict(tk.request.form), "login")
+
+    with contextlib.suppress(tk.ObjectNotFound):
+        utils.regenerate_user_secret(login)
 
-    utils.regenerate_user_secret(user_name)
-    success = utils.send_verification_email_to_user(user_name)
+    success = utils.send_verification_email_to_user(login)
 
     return jsonify(
         {
@@ -137,7 +139,6 @@ def init_qr_code() -> Response:
     user_name: str = tk.get_or_bust(dict(tk.request.form), "login")
 
     secret = UserSecret.get_for_user(user_name)
-
     if not secret:
         secret = UserSecret.create_for_user(user_name)
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ckanext-auth"
-version = "0.4.4"
+version = "0.4.5"
 description = "2FA authentication for CKAN"
 authors = [
     {name = "DataShades", email = "[email protected]"},