Merge branch 'security-1' into 'main'

Security fix : magic link is not anymore in reset password e-mails unless TRUSTED_HOSTS is configured

See merge request yaal/canaille!240

Author: azmeuk

CHANGES.rst

@@ -6,6 +6,7 @@ Added
 - Instructions in CONTRIBUTING.rst to update the docker image :issue:`59`
 - Instructions in README.md to discover Canaille interface with a docker image :issue:`59`
 - The :ref:`cli dump <cli_dump>` command can dump only some given models.
+- Implement the :class:`canaille.app.configuration.RootSettings.TRUSTED_HOSTS` configuration parameter, to secure password reset e-mails.
 
 Fixed
 ^^^^^

canaille/app/configuration.py

@@ -70,6 +70,12 @@ class RootSettings(BaseSettings):
     This sets domain name on which canaille will be served.
     """
 
+    TRUSTED_HOSTS: list[str] | None = None
+    """The Flask :external:py:data:`TRUSTED_HOSTS` configuration setting.
+
+    This sets trusted values for hosts and validates hosts during requests.
+    """
+
     PREFERRED_URL_SCHEME: str = "https"
     """The Flask :external:py:data:`PREFERRED_URL_SCHEME` configuration
     setting.

canaille/app/features.py

@@ -94,6 +94,14 @@ def has_email_confirmation(self):
             self.app.config["CANAILLE"]["EMAIL_CONFIRMATION"] is None and self.has_smtp
         )
 
+    @property
+    def has_trusted_hosts(self):
+        """Indicate whether the Flask TRUSTED_HOSTS option is enabled.
+
+        It is controlled by the :attr:`TRUSTED_HOSTS <canaille.app.configuration.RootSettings.TRUSTED_HOSTS>` configuration parameter.
+        """
+        return bool(self.app.config["TRUSTED_HOSTS"])
+
     @property
     def has_oidc(self):
         """Indicate whether the OIDC feature is enabled.

canaille/core/endpoints/account.py

@@ -905,4 +905,4 @@ def reset(user):
             )
         )
 
-    return render_template("core/reset-password.html", form=form, user=user, hash=None)
+    return render_template("core/reset-password.html", form=form, user=user, token=None)

canaille/core/endpoints/admin.py

@@ -6,13 +6,13 @@
 from wtforms import StringField
 from wtforms.validators import DataRequired
 
+from canaille.app import build_hash
 from canaille.app import obj_to_b64
 from canaille.app.flask import user_needed
 from canaille.app.forms import Form
 from canaille.app.forms import email_validator
 from canaille.app.i18n import gettext as _
 from canaille.app.templating import render_template
-from canaille.core.mails import build_hash
 from canaille.core.mails import send_test_mail
 
 bp = Blueprint("admin", __name__, url_prefix="/admin")
@@ -79,7 +79,7 @@ def password_init_html(user):
     reset_url = url_for(
         "core.auth.reset",
         user=user,
-        hash=build_hash(user.identifier, user.preferred_email, user.password),
+        token=user.generate_url_safe_token(),
         title=_("Password initialization on {website_name}").format(
             website_name=current_app.config["CANAILLE"]["NAME"]
         ),
@@ -105,7 +105,7 @@ def password_init_txt(user):
     reset_url = url_for(
         "core.auth.reset",
         user=user,
-        hash=build_hash(user.identifier, user.preferred_email, user.password),
+        token=user.generate_url_safe_token(),
         _external=True,
     )
 
@@ -121,10 +121,12 @@ def password_init_txt(user):
 @user_needed("manage_oidc")
 def password_reset_html(user):
     base_url = url_for("core.account.index", _external=True)
+    server_name = current_app.config.get("SERVER_NAME")
+    reset_token = user.generate_url_safe_token()
     reset_url = url_for(
         "core.auth.reset",
         user=user,
-        hash=build_hash(user.identifier, user.preferred_email, user.password),
+        token=reset_token,
         title=_("Password reset on {website_name}").format(
             website_name=current_app.config["CANAILLE"]["NAME"]
         ),
@@ -136,6 +138,9 @@ def password_reset_html(user):
         site_name=current_app.config["CANAILLE"]["NAME"],
         site_url=base_url,
         reset_url=reset_url,
+        server_name=server_name,
+        reset_token=reset_token,
+        reset_code=None,
         logo=current_app.config["CANAILLE"]["LOGO"],
         title=_("Password reset on {website_name}").format(
             website_name=current_app.config["CANAILLE"]["NAME"]
@@ -147,10 +152,12 @@ def password_reset_html(user):
 @user_needed("manage_oidc")
 def password_reset_txt(user):
     base_url = url_for("core.account.index", _external=True)
+    server_name = current_app.config.get("SERVER_NAME")
+    reset_token = user.generate_url_safe_token()
     reset_url = url_for(
         "core.auth.reset",
         user=user,
-        hash=build_hash(user.identifier, user.preferred_email, user.password),
+        token=reset_token,
         _external=True,
     )
 
@@ -159,6 +166,9 @@ def password_reset_txt(user):
         site_name=current_app.config["CANAILLE"]["NAME"],
         site_url=current_app.config.get("SERVER_NAME", base_url),
         reset_url=reset_url,
+        server_name=server_name,
+        reset_token=reset_token,
+        reset_code=None,
     )
 
 

canaille/core/endpoints/auth.py

@@ -28,6 +28,7 @@
 from ..mails import send_password_initialization_mail
 from ..mails import send_password_reset_mail
 from .forms import FirstLoginForm
+from .forms import ForgottenPasswordCodeForm
 from .forms import ForgottenPasswordForm
 from .forms import LoginForm
 from .forms import PasswordForm
@@ -195,13 +196,15 @@ def forgotten():
     if not request.form:
         return render_template("core/forgotten-password.html", form=form)
 
+    item_name = "link" if current_app.features.has_trusted_hosts else "code"
+
     if not form.validate():
-        flash(_("Could not send the password reset link."), "error")
+        flash(_(f"Could not send the password reset {item_name}."), "error")
         return render_template("core/forgotten-password.html", form=form)
 
     user = get_user_from_login(form.login.data)
     success_message = _(
-        "A password reset link has been sent at your email address. "
+        f"A password reset {item_name} has been sent at your email address. "
         "You should receive it within a few minutes."
     )
     if current_app.config["CANAILLE"]["HIDE_INVALID_LOGINS"] and (
@@ -237,32 +240,62 @@ def forgotten():
             "error",
         )
 
-    return render_template("core/forgotten-password.html", form=form)
+    if current_app.features.has_trusted_hosts:
+        return render_template("core/forgotten-password.html", form=form)
+    else:
+        return redirect(url_for(".forgotten_code", user=user))
 
 
-@bp.route("/reset/<user:user>/<hash>", methods=["GET", "POST"])
-def reset(user, hash):
-    if not current_app.config["CANAILLE"]["ENABLE_PASSWORD_RECOVERY"]:
+@bp.route("/reset-code/<user:user>", methods=["GET", "POST"])
+@smtp_needed()
+def forgotten_code(user):
+    if (
+        not current_app.config["CANAILLE"]["ENABLE_PASSWORD_RECOVERY"]
+        or current_app.features.has_trusted_hosts
+    ):
         abort(404)
 
-    form = PasswordResetForm(request.form)
-    hashes = {
-        build_hash(
-            user.identifier,
-            email,
-            user.password if user.has_password() else "",
+    if not user.can_edit_self:
+        flash(
+            _(
+                "The user '%(user)s' does not have permissions to update their password. ",
+                user=user.formatted_name,
+            ),
+            "error",
         )
-        for email in user.emails
-    }
-    if not user or hash not in hashes:
+        return redirect(url_for(".forgotten"))
+
+    form = ForgottenPasswordCodeForm(request.form)
+    if not request.form:
+        return render_template("core/forgotten-password-code.html", form=form)
+
+    if not form.validate() or not user.is_otp_valid(form.code.data, "EMAIL_OTP"):
+        flash(_("Invalid code."), "error")
+        return render_template("core/forgotten-password-code.html", form=form)
+
+    return redirect(url_for(".reset", user=user, token=form.code.data))
+
+
+@bp.route("/reset/<user:user>/<token>", methods=["GET", "POST"])
+def reset(user, token):
+    if not current_app.config["CANAILLE"]["ENABLE_PASSWORD_RECOVERY"]:
+        abort(404)
+    form = PasswordResetForm(request.form)
+
+    if current_app.features.has_trusted_hosts:
+        token = build_hash(token)
+    if not user or not user.is_otp_valid(token, "EMAIL_OTP"):
+        item_name = "link" if current_app.features.has_trusted_hosts else "code"
         flash(
-            _("The password reset link that brought you here was invalid."),
+            _(f"The password reset {item_name} that brought you here was invalid."),
             "error",
         )
         return redirect(url_for("core.account.index"))
 
     if request.form and form.validate():
         Backend.instance.set_user_password(user, form.password.data)
+        user.clear_otp()
+        Backend.instance.save(user)
         login_user(user)
 
         flash(_("Your password has been updated successfully"), "success")
@@ -273,7 +306,9 @@ def reset(user, hash):
             )
         )
 
-    return render_template("core/reset-password.html", form=form, user=user, hash=hash)
+    return render_template(
+        "core/reset-password.html", form=form, user=user, token=token
+    )
 
 
 @bp.route("/setup-mfa")

canaille/core/endpoints/forms.py

@@ -21,6 +21,7 @@
 from canaille.app.i18n import lazy_gettext as _
 from canaille.app.i18n import native_language_name_from_code
 from canaille.backends import Backend
+from canaille.core.mails import RESET_CODE_LENGTH
 from canaille.core.models import OTP_DIGITS
 from canaille.core.validators import existing_group_member
 from canaille.core.validators import existing_login
@@ -65,6 +66,21 @@ class ForgottenPasswordForm(Form):
     )
 
 
+class ForgottenPasswordCodeForm(Form):
+    code = wtforms.StringField(
+        _("Code"),
+        validators=[
+            wtforms.validators.DataRequired(),
+            wtforms.validators.Length(min=RESET_CODE_LENGTH, max=RESET_CODE_LENGTH),
+        ],
+        render_kw={
+            "placeholder": _("123456"),
+            "spellcheck": "false",
+            "autocorrect": "off",
+        },
+    )
+
+
 class PasswordResetForm(Form):
     password = wtforms.PasswordField(
         _("Password"),

canaille/core/mails.py

@@ -1,11 +1,13 @@
 from flask import current_app
 from flask import url_for
 
-from canaille.app import build_hash
 from canaille.app.i18n import gettext as _
 from canaille.app.mails import logo
 from canaille.app.mails import send_email
 from canaille.app.templating import render_template
+from canaille.backends import Backend
+
+RESET_CODE_LENGTH = 6
 
 
 def send_test_mail(email):
@@ -39,32 +41,45 @@ def send_test_mail(email):
 
 def send_password_reset_mail(user, mail):
     base_url = url_for("core.account.index", _external=True)
-    reset_url = url_for(
-        "core.auth.reset",
-        user=user,
-        hash=build_hash(
-            user.identifier,
-            mail,
-            user.password if user.has_password() else "",
-        ),
-        _external=True,
-    )
+    server_name = current_app.config.get("SERVER_NAME")
     logo_cid, logo_filename, logo_raw = logo()
-
     subject = _("Password reset on {website_name}").format(
         website_name=current_app.config["CANAILLE"]["NAME"]
     )
+
+    reset_token = None
+    reset_url = None
+    reset_code = None
+    if current_app.features.has_trusted_hosts:
+        reset_token = user.generate_url_safe_token()
+        Backend.instance.save(user)
+        reset_url = url_for(
+            "core.auth.reset",
+            user=user,
+            token=reset_token,
+            _external=True,
+        )
+    else:
+        reset_code = user.generate_sms_or_mail_otp(length=RESET_CODE_LENGTH)
+        Backend.instance.save(user)
+
     text_body = render_template(
         "core/mails/reset.txt",
         site_name=current_app.config["CANAILLE"]["NAME"],
         site_url=base_url,
         reset_url=reset_url,
+        server_name=server_name,
+        reset_token=reset_token,
+        reset_code=reset_code,
     )
     html_body = render_template(
         "core/mails/reset.html",
         site_name=current_app.config["CANAILLE"]["NAME"],
         site_url=base_url,
         reset_url=reset_url,
+        server_name=server_name,
+        reset_token=reset_token,
+        reset_code=reset_code,
         logo=f"cid:{logo_cid[1:-1]}" if logo_cid else None,
         title=subject,
     )
@@ -80,14 +95,12 @@ def send_password_reset_mail(user, mail):
 
 def send_password_initialization_mail(user, email):
     base_url = url_for("core.account.index", _external=True)
+    reset_token = user.generate_url_safe_token()
+    Backend.instance.save(user)
     reset_url = url_for(
         "core.auth.reset",
         user=user,
-        hash=build_hash(
-            user.identifier,
-            email,
-            user.password if user.has_password() else "",
-        ),
+        token=reset_token,
         _external=True,
     )
     logo_cid, logo_filename, logo_raw = logo()