feat : Added time one-time password (TOTP) authentication

.pre-commit-config.yaml

@@ -15,10 +15,10 @@ repos:
       - id: end-of-file-fixer
         exclude: "\\.svg$|\\.map$|\\.min\\.css$|\\.min\\.js$|\\.po$|\\.pot$"
       - id: check-toml
-  - repo: https://github.com/PyCQA/docformatter
+  # - repo: https://github.com/PyCQA/docformatter
-    rev: v1.7.5
+  #   rev: v1.7.5
-    hooks:
+  #   hooks:
-      - id: docformatter
+  #     - id: docformatter
   - repo: https://github.com/rtts/djhtml
     rev: 3.0.7
     hooks:

canaille/app/__init__.py

@@ -2,7 +2,10 @@ import base64
 import hashlib
 import json
 import re
+from base64 import b64encode
+from io import BytesIO
 
+import qrcode
 from flask import current_app
 from flask import request
 
@@ -66,3 +69,13 @@ class classproperty:
 
     def __get__(self, obj, owner):
         return self.f(owner)
+
+
+def get_b64encoded_qr_image(data):
+    qr = qrcode.QRCode(version=1, box_size=10, border=5)
+    qr.add_data(data)
+    qr.make(fit=True)
+    img = qr.make_image(fill_color="black", back_color="white")
+    buffered = BytesIO()
+    img.save(buffered)
+    return b64encode(buffered.getvalue()).decode("utf-8")

canaille/app/features.py

@@ -14,6 +14,10 @@ class Features:
     def has_password_recovery(self):
         return self.app.config["CANAILLE"]["ENABLE_PASSWORD_RECOVERY"]
 
+    @property
+    def has_totp(self):
+        return self.app.config["CANAILLE"]["ENABLE_TOTP"]
+
     @property
     def has_registration(self):
         return self.app.config["CANAILLE"]["ENABLE_REGISTRATION"]

canaille/backends/ldap/configuration.py

@@ -28,7 +28,7 @@ class LDAPSettings(BaseModel):
     For instance `ou=users,dc=mydomain,dc=tld`.
     """
 
-    USER_CLASS: str = "inetOrgPerson"
+    USER_CLASS: list[str] = ["inetOrgPerson"]
     """The object class to use for creating new users."""
 
     USER_RDN: str = "uid"

canaille/backends/ldap/models.py

@@ -1,4 +1,5 @@
 import ldap.filter
+from flask import current_app
 
 import canaille.core.models
 import canaille.oidc.models
@@ -34,6 +35,8 @@ class User(canaille.core.models.User, LDAPObject):
         "organization": "o",
         "groups": "memberOf",
         "lock_date": "pwdEndTime",
+        "secret_token": "oathSecret",
+        "last_otp_login": "oathLastLogin",
     }
 
     def match_filter(self, filter):
@@ -44,6 +47,9 @@ class User(canaille.core.models.User, LDAPObject):
         return super().match_filter(filter)
 
     def save(self):
+        if current_app.features.has_totp and not self.secret_token:
+            self.generate_otp_token()
+
         group_attr = self.python_attribute_to_ldap("groups")
         if group_attr not in self.changes:
             return

canaille/backends/memory/backend.py

@@ -3,6 +3,9 @@ import datetime
 import uuid
 from typing import Any
 
+from flask import current_app
+
+import canaille.backends.memory.models
 from canaille.backends import Backend
 
 
@@ -124,6 +127,13 @@ class MemoryBackend(Backend):
         return results[0] if results else None
 
     def save(self, instance):
+        if (
+            isinstance(instance, canaille.backends.memory.models.User)
+            and current_app.features.has_totp
+            and not instance.secret_token
+        ):
+            instance.generate_otp_token()
+
         if not instance.id:
             instance.id = str(uuid.uuid4())
 

canaille/backends/sql/backend.py

@@ -110,6 +110,10 @@ class SQLBackend(Backend):
         ).scalar_one_or_none()
 
     def save(self, instance):
+        # run the instance save callback if existing
+        if hasattr(instance, "save"):
+            instance.save()
+
         instance.last_modified = datetime.datetime.now(datetime.timezone.utc).replace(
             microsecond=0
         )

canaille/backends/sql/models.py

@@ -2,6 +2,7 @@ import datetime
 import typing
 import uuid
 
+from flask import current_app
 from sqlalchemy import Boolean
 from sqlalchemy import Column
 from sqlalchemy import ForeignKey
@@ -100,6 +101,14 @@ class User(canaille.core.models.User, Base, SqlAlchemyModel):
     lock_date: Mapped[datetime.datetime] = mapped_column(
         TZDateTime(timezone=True), nullable=True
     )
+    last_otp_login: Mapped[datetime.datetime] = mapped_column(
+        TZDateTime(timezone=True), nullable=True
+    )
+    secret_token: Mapped[str] = mapped_column(String, nullable=True, unique=True)
+
+    def save(self):
+        if current_app.features.has_totp and not self.secret_token:
+            self.generate_otp_token()
 
 
 class Group(canaille.core.models.Group, Base, SqlAlchemyModel):

canaille/config.sample.toml

@@ -62,6 +62,10 @@ SECRET_KEY = "change me before you go in production"
 # recovery link by email. This option is true by default.
 # ENABLE_PASSWORD_RECOVERY = true
 
+# If ENABLE_TOTP is true, then users will need to authenticate themselves
+# using a time one-time password via an authenticator app. This option is false by default.
+# ENABLE_TOTP = false
+
 # The validity duration of registration invitations, in seconds.
 # Defaults to 2 days
 # INVITATION_EXPIRATION = 172800
@@ -115,6 +119,7 @@ SECRET_KEY = "change me before you go in production"
 # USER_BASE = "ou=users,dc=mydomain,dc=tld"
 
 # The object class to use for creating new users
+# Note : If you plan on using TOTP authentication, use : USER_CLASS = ["inetOrgPerson", "oathTOTPToken"]
 # USER_CLASS = "inetOrgPerson"
 
 # The attribute to identify an object in the User dn.

canaille/core/configuration.py

@@ -248,6 +248,10 @@ class CoreSettings(BaseModel):
     """If :py:data:`False`, then users cannot ask for a password recovery link
     by email."""
 
+    ENABLE_TOTP: bool = False
+    """If :py:data:`True`, then users will need to authenticate themselves
+    using a time one-time password via an authenticator app."""
+
     INVITATION_EXPIRATION: int = 172800
     """The validity duration of registration invitations, in seconds.
 

canaille/core/endpoints/account.py

@@ -274,7 +274,8 @@ def registration(data=None, hash=None):
             )
             return redirect(url_for("core.account.index"))
 
-    if current_user():
+    user = current_user()
+    if user:
         flash(
             _("You are already logged in, you cannot create an account."),
             "error",

canaille/core/endpoints/auth.py

@@ -1,28 +1,32 @@
-from flask import Blueprint
+import datetime
-from flask import abort
-from flask import current_app
-from flask import flash
-from flask import redirect
-from flask import request
-from flask import session
-from flask import url_for
 
-from canaille.app import build_hash
+from flask import (
-from canaille.app.flask import smtp_needed
+    Blueprint,
+    abort,
+    current_app,
+    flash,
+    redirect,
+    request,
+    session,
+    url_for,
+)
+
+from canaille.app import build_hash, get_b64encoded_qr_image
+from canaille.app.flask import current_user, login_user, logout_user, smtp_needed
 from canaille.app.i18n import gettext as _
-from canaille.app.session import current_user
+from canaille.app.session import current_user, login_user, logout_user
-from canaille.app.session import login_user
-from canaille.app.session import logout_user
 from canaille.app.themes import render_template
 from canaille.backends import Backend
+from canaille.core.endpoints.forms import TwoFactorForm
 
-from ..mails import send_password_initialization_mail
+from ..mails import send_password_initialization_mail, send_password_reset_mail
-from ..mails import send_password_reset_mail
+from .forms import (
-from .forms import FirstLoginForm
+    FirstLoginForm,
-from .forms import ForgottenPasswordForm
+    ForgottenPasswordForm,
-from .forms import LoginForm
+    LoginForm,
-from .forms import PasswordForm
+    PasswordForm,
-from .forms import PasswordResetForm
+    PasswordResetForm,
+)
 
 bp = Blueprint("auth", __name__)
 
@@ -103,16 +107,28 @@ def password():
             "password.html", form=form, username=session["attempt_login"]
         )
 
-    current_app.logger.security(
+    if not current_app.features.has_totp:
-        f'Succeed login attempt for {session["attempt_login"]} from {request_ip}'
+        current_app.logger.security(
-    )
+            f'Succeed login attempt for {session["attempt_login"]} from {request_ip}'
-    del session["attempt_login"]
+        )
-    login_user(user)
+        del session["attempt_login"]
-    flash(
+        login_user(user)
-        _("Connection successful. Welcome %(user)s", user=user.formatted_name),
+        flash(
-        "success",
+            _("Connection successful. Welcome %(user)s", user=user.formatted_name),
-    )
+            "success",
-    return redirect(session.pop("redirect-after-login", url_for("core.account.index")))
+        )
+        return redirect(
+            session.pop("redirect-after-login", url_for("core.account.index"))
+        )
+    else:
+        session["attempt_login_with_correct_password"] = session.pop("attempt_login")
+        if not user.last_otp_login:
+            flash(
+                "You have not enabled Two-Factor Authentication. Please enable it first to login.",
+                "info",
+            )
+            return redirect(url_for("core.auth.setup_two_factor_auth"))
+        return redirect(url_for("core.auth.verify_two_factor_auth"))
 
 
 @bp.route("/logout")
@@ -251,3 +267,100 @@ def reset(user, hash):
         )
 
     return render_template("reset-password.html", form=form, user=user, hash=hash)
+
+
+@bp.route("/setup-2fa")
+def setup_two_factor_auth():
+    if not current_app.features.has_totp:
+        abort(404)
+
+    if current_user():
+        return redirect(
+            url_for("core.account.profile_edition", edited_user=current_user())
+        )
+
+    if "attempt_login_with_correct_password" not in session:
+        flash(_("Cannot remember the login you attempted to sign in with"), "warning")
+        return redirect(url_for("core.auth.login"))
+
+    user = Backend.instance.get_user_from_login(
+        session["attempt_login_with_correct_password"]
+    )
+
+    uri = user.get_authentication_setup_uri()
+    base64_qr_image = get_b64encoded_qr_image(uri)
+    return render_template(
+        "setup-2fa.html",
+        secret=user.secret_token,
+        qr_image=base64_qr_image,
+        username=user.user_name,
+    )
+
+
+@bp.route("/verify-2fa", methods=["GET", "POST"])
+def verify_two_factor_auth():
+    if not current_app.features.has_totp:
+        abort(404)
+
+    if current_user():
+        return redirect(
+            url_for("core.account.profile_edition", edited_user=current_user())
+        )
+
+    if "attempt_login_with_correct_password" not in session:
+        flash(_("Cannot remember the login you attempted to sign in with"), "warning")
+        return redirect(url_for("core.auth.login"))
+
+    form = TwoFactorForm(request.form or None)
+    form.render_field_macro_file = "partial/login_field.html"
+
+    if not request.form or form.form_control():
+        return render_template(
+            "verify-2fa.html",
+            form=form,
+            username=session["attempt_login_with_correct_password"],
+        )
+
+    user = Backend.instance.get_user_from_login(
+        session["attempt_login_with_correct_password"]
+    )
+
+    if form.validate() and user.is_otp_valid(form.otp.data):
+        welcome_message = (
+            "Connection successful."
+            if user.last_otp_login
+            else "Two-factor authentication setup successful."
+        )
+        try:
+            user.last_otp_login = datetime.datetime.now(datetime.timezone.utc)
+            Backend.instance.save(user)
+            request_ip = request.remote_addr or "unknown IP"
+            current_app.logger.security(
+                f'Succeed login attempt for {session["attempt_login_with_correct_password"]} from {request_ip}'
+            )
+            del session["attempt_login_with_correct_password"]
+            login_user(user)
+            flash(
+                _(
+                    "%(welcome_message)s Welcome %(user)s",
+                    user=user.formatted_name,
+                    welcome_message=welcome_message,
+                ),
+                "success",
+            )
+            return redirect(
+                session.pop("redirect-after-login", url_for("core.account.index"))
+            )
+        except Exception:  # pragma: no cover
+            flash("Two-factor authentication setup failed. Please try again.", "danger")
+            return redirect(url_for("core.auth.verify_two_factor_auth"))
+    else:
+        flash(
+            "The one-time password you entered is invalid. Please try again",
+            "error",
+        )
+        request_ip = request.remote_addr or "unknown IP"
+        current_app.logger.security(
+            f'Failed login attempt (wrong TOTP) for {session["attempt_login_with_correct_password"]} from {request_ip}'
+        )
+        return redirect(url_for("core.auth.verify_two_factor_auth"))

canaille/core/endpoints/forms.py

@@ -23,6 +23,8 @@ from canaille.app.i18n import lazy_gettext as _
 from canaille.app.i18n import native_language_name_from_code
 from canaille.backends import Backend
 
+OTP_LENGTH = 6
+
 
 def unique_user_name(form, field):
     if Backend.instance.get(models.User, user_name=field.data) and (
@@ -480,3 +482,18 @@ class EmailConfirmationForm(Form):
             "autocorrect": "off",
         },
     )
+
+
+class TwoFactorForm(Form):
+    otp = wtforms.StringField(
+        _("One-time password"),
+        validators=[
+            wtforms.validators.DataRequired(),
+            wtforms.validators.Length(min=OTP_LENGTH, max=OTP_LENGTH),
+        ],
+        render_kw={
+            "placeholder": _("123456"),
+            "spellcheck": "false",
+            "autocorrect": "off",
+        },
+    )

canaille/core/models.py

@@ -1,7 +1,9 @@
 import datetime
+import secrets
 from typing import Annotated
 from typing import ClassVar
 
+import otpauth
 from flask import current_app
 
 from canaille.backends.models import Model
@@ -241,6 +243,14 @@ class User(Model):
     lock_date: datetime.datetime | None = None
     """A DateTime indicating when the resource was locked."""
 
+    last_otp_login: datetime.datetime | None = None
+    """A DateTime indicating when the user last logged in with a one-time password.
+    This attribute is currently used to check whether the user has activated multi-factor authentication or not."""
+
+    secret_token: str | None = None
+    """Unique token generated for each user, used for
+    two-factor authentication."""
+
     _readable_fields = None
     _writable_fields = None
     _permissions = None
@@ -318,6 +328,20 @@ class User(Model):
                     self._writable_fields |= set(details["WRITE"])
         return self._writable_fields
 
+    def generate_otp_token(self):
+        self.secret_token = secrets.token_hex(32)
+
+    def generate_otp(self):
+        return otpauth.TOTP(bytes(self.secret_token, "utf-8")).generate()
+
+    def get_authentication_setup_uri(self):
+        return otpauth.TOTP(bytes(self.secret_token, "utf-8")).to_uri(
+            label=self.user_name, issuer=current_app.config["CANAILLE"]["NAME"]
+        )
+
+    def is_otp_valid(self, user_otp):
+        return otpauth.TOTP(bytes(self.secret_token, "utf-8")).verify(user_otp)
+
 
 class Group(Model):
     """User model, based on the `SCIM Group schema

canaille/core/templates/partial/login_field.html

@@ -6,4 +6,7 @@
     {% if field.name == "password" %}
         {{ fui.render_field(field, icon="key", noindicator=true, **kwargs) }}
     {% endif %}
+    {% if field.name == "otp" %}
+        {{ fui.render_field(field, icon="key", noindicator=true, **kwargs) }}
+    {% endif %}
 {% endmacro %}

canaille/core/templates/setup-2fa.html

@@ -0,0 +1,74 @@
+{% extends theme('base.html') %}
+{% import 'macro/flask.html' as flask %}
+{% import 'macro/form.html' as fui %}
+{% import 'partial/login_field.html' as login_field %}
+
+{% block container %}
+    <div class="ui container">
+        <div class="content">
+            <div class="ui clearing segment">
+                {% block header %}
+                    {% if logo_url %}
+                        <a href="{{ url_for('core.account.index') }}">
+                            <img class="ui tiny centered image" src="{{ logo_url }}" alt="{{ website_name }}">
+                        </a>
+                    {% else %}
+                        <i class="massive sign in icon image ui"></i>
+                    {% endif %}
+
+                    <h2 class="ui center aligned header">
+                        <div class="content">
+                            {{ _("Sign in as %(username)s", username=username) }}
+                        </div>
+                        <div class="sub header">{% trans %}Set up Two-Factor Authentication.{% endtrans %}</div>
+                    </h2>
+                {% endblock %}
+
+                {% block messages %}
+                    {{ flask.messages() }}
+                {% endblock %}
+
+                <div class="ui segment">
+                    <h3 class="ui header">Instructions:</h3>
+                    <ol class="ui list">
+                        <li class="item">Download <a href="https://play.google.com/store/apps/details?id=com.google.android.apps.authenticator2" target="_blank">Google Authenticator</a> on your mobile.</li>
+                        <li class="item">Set up a new authenticator.</li>
+                        <li class="item">Scan the QR code below and click "Continue".</li>
+                    </ol>
+                    <p><strong>Note:</strong> If you can't scan the QR code, try entering the secret token in your authenticator app directly.</p>
+                </div>
+                <div>
+                    <img src="data:image/png;base64, {{ qr_image }}" alt="Secret Token" class="ui centered medium image"/>
+                </div>
+                <form class="ui segment grid action input">
+                    <label class="ui large label token-label" for="secret">Secret token:</label>
+                    <input type="text" id="secret" name="secret" value="{{ secret }}" readonly>
+                    <button type="button" class="ui secondary right labeled icon button" onclick="copySecret(this)">
+                        Copy Secret
+                    </button>
+                </form>
+                {% block buttons %}
+                    <div class="ui right aligned container">
+                        <div class="ui stackable buttons">
+                            <a type="button" class="ui right floated primary button" href="{{ url_for('core.auth.verify_two_factor_auth') }}">{{ _("Continue") }}</a>
+                        </div>
+                    </div>
+                {% endblock %}
+            </div>
+        </div>
+    </div>
+{% endblock %}
+{% block script %}
+    <script>
+        function copySecret(clickedButton) {
+            var copyText = document.getElementById("secret");
+            copyText.select();
+            copyText.setSelectionRange(0, 99999); /*For mobile devices*/
+            document.execCommand("copy");
+            clickedButton.textContent = "Successfully copied!";
+            setTimeout(() => {
+                clickedButton.textContent = "Copy Secret";
+            }, 2000);
+        }
+    </script>
+{% endblock %}

canaille/core/templates/verify-2fa.html

@@ -0,0 +1,48 @@
+{% extends theme('base.html') %}
+{% import 'macro/flask.html' as flask %}
+{% import 'macro/form.html' as fui %}
+{% import 'partial/login_field.html' as login_field %}
+
+{% block container %}
+    <div class="ui container">
+        <div class="content">
+            <div class="ui clearing segment">
+                {% block header %}
+                    {% if logo_url %}
+                        <a href="{{ url_for('core.account.index') }}">
+                            <img class="ui tiny centered image" src="{{ logo_url }}" alt="{{ website_name }}">
+                        </a>
+                    {% else %}
+                        <i class="massive sign in icon image ui"></i>
+                    {% endif %}
+
+                    <h2 class="ui center aligned header">
+                        <div class="content">
+                            {{ _("Sign in as %(username)s", username=username) }}
+                        </div>
+                        <div class="sub header">{% trans %}Please enter the one-time password from your authenticator app.{% endtrans %}</div>
+                    </h2>
+                {% endblock %}
+
+                {% block messages %}
+                    {{ flask.messages() }}
+                {% endblock %}
+
+                {% call fui.render_form(form, hx_boost="false") %}
+                    {% block fields %}
+                        {{ login_field.render_field(form.otp, class="autofocus") }}
+                    {% endblock %}
+
+                    {% block buttons %}
+                        <div class="ui right aligned container">
+                            <div class="ui stackable buttons">
+                                <a type="button" class="ui right floated button" href="{{ url_for('core.auth.login') }}">{{ _("I am not %(username)s", username=username) }}</a>
+                                <button type="submit" class="ui right floated primary button">{{ _("Verify") }}</button>
+                            </div>
+                        </div>
+                    {% endblock %}
+                {% endcall %}
+            </div>
+        </div>
+    </div>
+{% endblock %}

canaille/static/css/base.css

@@ -261,4 +261,8 @@ select.ui.multiple.dropdown option[selected] {
     .ui.progress {
         background: #222222;
     }
+
+    .ui.label.token-label {
+        padding-left: 0;
+    }
 }

demo/conf/canaille-ldap.toml

@@ -27,6 +27,7 @@ BIND_PW = "admin"
 TIMEOUT = 10
 USER_BASE = "ou=users,dc=mydomain,dc=tld"
 GROUP_BASE = "ou=groups,dc=mydomain,dc=tld"
+USER_CLASS = ["inetOrgPerson", "oathTOTPToken"]
 
 [CANAILLE.ACL.DEFAULT]
 PERMISSIONS = ["edit_self", "use_oidc"]

demo/ldap-server.py

@@ -17,6 +17,7 @@ schemas = [
     "ldif/memberof-config.ldif",
     "ldif/refint-config.ldif",
     "ldif/ppolicy-config.ldif",
+    "ldif/otp-config.ldif",
     "../canaille/backends/ldap/schemas/oauth2-openldap.ldif",
 ]
 

demo/ldif/otp-config.ldif

@@ -0,0 +1,8 @@
+dn: cn=module,cn=config
+cn: module
+objectClass: olcModuleList
+olcModuleLoad: otp
+
+dn: olcOverlay=otp,olcDatabase={1}mdb,cn=config
+objectClass: olcOverlayConfig
+olcOverlay: otp

doc/features.rst

@@ -174,6 +174,13 @@ Password compromission check
 
 If :attr:`password compromission check feature <canaille.core.configuration.CoreSettings.ENABLE_PASSWORD_COMPROMISSION_CHECK>` is enabled, Canaille will check for password compromise on HIBP (https://haveibeenpwned.com/) every time a new password is register. You will need to set an :attr:`admin email <canaille.core.configuration.CoreSettings.ADMIN_EMAIL>`.
 
+.. _feature_multi_factor_authentication:
+
+Multi-factor authentication
+==============
+
+If the :attr:`time one-time password feature <canaille.core.configuration.CoreSettings.ENABLE_TOTP>` is enabled, then users will need to authenticate themselves using a time one-time password via an authenticator app.
+
 Web interface
 *************
 
@@ -273,7 +280,7 @@ Logging
 
 Canaille writes :attr:`logs <canaille.core.configuration.CoreSettings.LOGGING>` for every important event happening, to help administrators understand what is going on and debug funky situations.
 
-The following security events are logged with the tag [SECURITY] for easy retrieval:
+The following security events are logged with the log level "security" for easy retrieval :
 
 - Authentication attempt
 - Password update

doc/tutorial/databases.rst

@@ -49,10 +49,14 @@ It is used when the ``CANAILLE_LDAP`` configuration parameter is defined. For in
 
     GROUP_BASE = "ou=groups,dc=mydomain,dc=tld"
 
+If you want to use TOTP authentication, you will need to add the ``oathTOTPToken`` class to the user :
+.. code-block:: toml
+   USER_CLASS = ["inetOrgPerson", "oathTOTPToken"]
+
 You can find more details on the LDAP configuration in the :class:`dedicated section <canaille.backends.ldap.configuration.LDAPSettings>`.
 
 .. note ::
-   Currently, only the ``inetOrgPerson`` and ``groupOfNames`` schemas have been tested.
+   Currently, only the ``inetOrgPerson``, ``oathTOTPToken`` and ``groupOfNames`` schemas have been tested.
    If you want to use different schemas or LDAP servers, adaptations may be needed.
    Patches are welcome.
 
@@ -108,3 +112,21 @@ You can adapt and load those configuration files with:
     # Adapt those commands according to your setup
     sudo ldapadd -Q -H ldapi:/// -Y EXTERNAL -f ppolicy-config.ldif
     sudo ldapadd -Q -H ldapi:/// -Y EXTERNAL -f ppolicy.ldif
+
+otp
+~~~~~~~
+
+If the `otp <https://www.openldap.org/software/man.cgi?query=slapo-otp>`_ overlay is configured, you will be able to add one-time password authentication in canaille.
+
+Here is a configuration example compatible with canaille:
+
+.. literalinclude :: ../../demo/ldif/otp-config.ldif
+   :language: ldif
+   :caption: otp-config.ldif
+
+You can adapt and load this configuration file with:
+
+.. code-block:: bash
+
+    # Adapt this command according to your setup
+    sudo ldapadd -Q -H ldapi:/// -Y EXTERNAL -f otp-config.ldif

tests/backends/ldap/fixtures.py

@@ -16,6 +16,7 @@ def slapd_server():
             "demo/ldif/memberof-config.ldif",
             "demo/ldif/ppolicy-config.ldif",
             "demo/ldif/ppolicy.ldif",
+            "demo/ldif/otp-config.ldif",
             "canaille/backends/ldap/schemas/oauth2-openldap.ldif",
             "demo/ldif/bootstrap-users-tree.ldif",
             "demo/ldif/bootstrap-oidc-tree.ldif",
@@ -38,6 +39,7 @@ def ldap_configuration(configuration, slapd_server):
         "USER_FILTER": "(uid={{ login }})",
         "GROUP_BASE": "ou=groups",
         "TIMEOUT": 0.1,
+        "USER_CLASS": ["inetOrgPerson", "oathTOTPToken"],
     }
     yield configuration
     del configuration["CANAILLE_LDAP"]

tests/conftest.py

@@ -1,3 +1,4 @@
+import datetime
 import os
 
 import pytest
@@ -199,6 +200,16 @@ def user(app, backend):
     backend.delete(u)
 
 
+@pytest.fixture
+def user_totp(app, user, backend):
+    user.secret_token = (
+        "fefe9b106b8a033d3fcb4de16ac06b2cae71c7d95a41b158c30380d1bc35b2ba"
+    )
+    user.last_otp_login = datetime.datetime(2020, 1, 1)
+    backend.save(user)
+    yield user
+
+
 @pytest.fixture
 def admin(app, backend):
     u = models.User(
@@ -234,6 +245,13 @@ def logged_user(user, testclient):
     return user
 
 
+@pytest.fixture
+def logged_user_totp(user_totp, testclient):
+    with testclient.session_transaction() as sess:
+        sess["user_id"] = [user_totp.id]
+    return user_totp
+
+
 @pytest.fixture
 def logged_admin(admin, testclient):
     with testclient.session_transaction() as sess:

tests/core/test_multi_factor_authentication.py

@@ -0,0 +1,231 @@
+import datetime
+import logging
+
+import time_machine
+
+from canaille.app import models
+
+
+def test_totp_disabled(testclient):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = False
+
+    testclient.get("/setup-2fa", status=404)
+    testclient.get("/verify-2fa", status=404)
+
+
+def test_signin_and_out_with_totp(testclient, user_totp, caplog):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    with testclient.session_transaction() as session:
+        assert not session.get("user_id")
+
+    res = testclient.get("/login", status=200)
+
+    res.form["login"] = "user"
+    res = res.form.submit(status=302)
+    res = res.follow(status=200)
+
+    with testclient.session_transaction() as session:
+        assert "user" == session.get("attempt_login")
+
+    res.form["password"] = "correct horse battery staple"
+    res = res.form.submit(status=302)
+    res = res.follow(status=200)
+
+    with testclient.session_transaction() as session:
+        assert "attempt_login" not in session
+        assert "user" == session.get("attempt_login_with_correct_password")
+
+    res = testclient.get("/verify-2fa")
+    res.form["otp"] = user_totp.generate_otp()
+    res = res.form.submit(status=302)
+
+    assert (
+        "success",
+        "Connection successful. Welcome John (johnny) Doe",
+    ) in res.flashes
+    assert (
+        "canaille",
+        logging.SECURITY,
+        "Succeed login attempt for user from unknown IP",
+    ) in caplog.record_tuples
+    res = res.follow(status=302)
+    res = res.follow(status=200)
+
+    with testclient.session_transaction() as session:
+        assert [user_totp.id] == session.get("user_id")
+        assert "attempt_login" not in session
+        assert "attempt_login_with_correct_password" not in session
+
+    res = testclient.get("/login", status=302)
+
+    res = testclient.get("/logout")
+    assert (
+        "success",
+        "You have been disconnected. See you next time John (johnny) Doe",
+    ) in res.flashes
+    assert (
+        "canaille",
+        logging.SECURITY,
+        "Logout user from unknown IP",
+    ) in caplog.record_tuples
+    res = res.follow(status=302)
+    res = res.follow(status=200)
+
+
+def test_signin_wrong_totp(testclient, user_totp, caplog):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    with testclient.session_transaction() as session:
+        assert not session.get("user_id")
+
+    res = testclient.get("/login", status=200)
+
+    res.form["login"] = "user"
+    res = res.form.submit(status=302)
+    res = res.follow(status=200)
+
+    res.form["password"] = "correct horse battery staple"
+    res = res.form.submit(status=302)
+    res = res.follow(status=200)
+
+    res.form["otp"] = 111111
+    res = res.form.submit()
+
+    assert (
+        "error",
+        "The one-time password you entered is invalid. Please try again",
+    ) in res.flashes
+    assert (
+        "canaille",
+        logging.SECURITY,
+        "Failed login attempt (wrong TOTP) for user from unknown IP",
+    ) in caplog.record_tuples
+
+
+def test_signin_expired_totp(testclient, user_totp, caplog):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    with time_machine.travel("2020-01-01 01:00:00+00:00", tick=False) as traveller:
+        with testclient.session_transaction() as session:
+            assert not session.get("user_id")
+
+        res = testclient.get("/login", status=200)
+
+        res.form["login"] = "user"
+        res = res.form.submit(status=302)
+        res = res.follow(status=200)
+
+        res.form["password"] = "correct horse battery staple"
+        res = res.form.submit(status=302)
+        res = res.follow(status=200)
+
+        res.form["otp"] = user_totp.generate_otp()
+        traveller.shift(datetime.timedelta(seconds=30))
+        res = res.form.submit()
+
+        assert (
+            "error",
+            "The one-time password you entered is invalid. Please try again",
+        ) in res.flashes
+        assert (
+            "canaille",
+            logging.SECURITY,
+            "Failed login attempt (wrong TOTP) for user from unknown IP",
+        ) in caplog.record_tuples
+
+
+def test_setup_totp(testclient, backend, caplog):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    u = models.User(
+        formatted_name="Totp User",
+        family_name="Totp",
+        user_name="totp",
+        emails=["john@doe.com"],
+        password="correct horse battery staple",
+    )
+    backend.save(u)
+
+    assert u.secret_token is not None
+
+    with testclient.session_transaction() as session:
+        assert not session.get("user_id")
+
+    res = testclient.get("/login", status=200)
+
+    res.form["login"] = "totp"
+    res = res.form.submit(status=302)
+    res = res.follow(status=200)
+
+    res.form["password"] = "correct horse battery staple"
+    res = res.form.submit(status=302)
+
+    assert res.location == "/setup-2fa"
+    assert (
+        "info",
+        "You have not enabled Two-Factor Authentication. Please enable it first to login.",
+    ) in res.flashes
+    res = testclient.get("/setup-2fa", status=200)
+    assert u.secret_token == res.form["secret"].value
+
+    res = testclient.get("/verify-2fa", status=200)
+    res.form["otp"] = u.generate_otp()
+    res = res.form.submit(status=302)
+
+    assert (
+        "success",
+        "Two-factor authentication setup successful. Welcome Totp User",
+    ) in res.flashes
+    assert (
+        "canaille",
+        logging.SECURITY,
+        "Succeed login attempt for totp from unknown IP",
+    ) in caplog.record_tuples
+    res = res.follow(status=302)
+    res = res.follow(status=200)
+
+    with testclient.session_transaction() as session:
+        assert [u.id] == session.get("user_id")
+        assert "attempt_login" not in session
+        assert "attempt_login_with_correct_password" not in session
+
+    res = testclient.get("/login", status=302)
+
+
+def test_verify_totp_page_without_signin_in_redirects_to_login_page(
+    testclient, user_totp
+):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    res = testclient.get("/verify-2fa", status=302)
+    assert res.location == "/login"
+    assert res.flashes == [
+        ("warning", "Cannot remember the login you attempted to sign in with")
+    ]
+
+
+def test_setup_totp_page_without_signin_in_redirects_to_login_page(
+    testclient, user_totp
+):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    res = testclient.get("/setup-2fa", status=302)
+    assert res.location == "/login"
+    assert res.flashes == [
+        ("warning", "Cannot remember the login you attempted to sign in with")
+    ]
+
+
+def test_verify_totp_page_already_logged_in(testclient, logged_user_totp):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    res = testclient.get("/verify-2fa", status=302)
+    assert res.location == "/profile/user"
+
+
+def test_setup_totp_page_already_logged_in(testclient, logged_user_totp):
+    testclient.app.config["CANAILLE"]["ENABLE_TOTP"] = True
+
+    res = testclient.get("/setup-2fa", status=302)
+    assert res.location == "/profile/user"