Merge branch '47-multiple-factor-authentication' into 'main'

Implement multiple factors authentication

Closes #47

See merge request yaal/canaille!193

.gitignore

@@ -1,5 +1,6 @@
 .env
 *.sqlite
+*.sqlite-journal
 *.pyc
 *.mo
 *.prof

.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
-    rev: v1.7.5
-    hooks:
-      - id: docformatter
+  # - repo: https://github.com/PyCQA/docformatter
+  #   rev: v1.7.5
+  #   hooks:
+  #     - id: docformatter
   - repo: https://github.com/rtts/djhtml
     rev: 3.0.7
     hooks:

CHANGES.rst

@@ -3,6 +3,12 @@
 
 Added
 ^^^^^
+- Multi-factor authentication :issue:`47`
+- :attr:`~canaille.core.configuration.CoreSettings.OTP_METHOD` and
+  :attr:`~canaille.core.configuration.CoreSettings.EMAIL_OTP` and
+  :attr:`~canaille.core.configuration.CoreSettings.SMS_OTP` and
+  :attr:`~canaille.core.configuration.CoreSettings.SMPP`
+  :issue:`47`
 - Password compromission check :issue:`179`
 - :attr:`~canaille.core.configuration.CoreSettings.ADMIN_EMAIL` and
   :attr:`~canaille.core.configuration.CoreSettings.ENABLE_PASSWORD_COMPROMISSION_CHECK` and

canaille/app/__init__.py

@@ -2,6 +2,8 @@
 import hashlib
 import json
 import re
+from base64 import b64encode
+from io import BytesIO
 
 from flask import current_app
 from flask import request
@@ -66,3 +68,29 @@ def __init__(self, f):
 
     def __get__(self, obj, owner):
         return self.f(owner)
+
+
+def get_b64encoded_qr_image(data):
+    try:
+        import qrcode
+    except ImportError:
+        return None
+
+    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")
+
+
+def mask_email(email):
+    atpos = email.find("@")
+    if atpos > 0:
+        return email[0] + "#####" + email[atpos - 1 :]
+    return None
+
+
+def mask_phone(phone):
+    return phone[0:3] + "#####" + phone[-2:]

canaille/app/configuration.py

@@ -1,3 +1,4 @@
+import importlib.util
 import os
 import smtplib
 import socket
@@ -175,7 +176,7 @@ def validate(config, validate_remote=False):
     validate_keypair(config.get("CANAILLE_OIDC"))
     validate_theme(config["CANAILLE"])
     validate_admin_email(config["CANAILLE"])
-
+    validate_otp_config(config["CANAILLE"])
     if not validate_remote:
         return
 
@@ -184,6 +185,8 @@ def validate(config, validate_remote=False):
     Backend.instance.validate(config)
     if smtp_config := config["CANAILLE"]["SMTP"]:
         validate_smtp_configuration(smtp_config)
+    if smpp_config := config["CANAILLE"]["SMPP"]:
+        validate_smpp_configuration(smpp_config)
 
 
 def validate_keypair(config):
@@ -231,6 +234,31 @@ def validate_smtp_configuration(config):
         raise ConfigurationException(exc) from exc
 
 
+def validate_smpp_configuration(config):
+    try:
+        import smpplib
+    except ImportError as exc:
+        raise ConfigurationException(
+            "You have configured a SMPP server but the 'sms' extra is not installed."
+        ) from exc
+
+    host = config["HOST"]
+    port = config["PORT"]
+    try:
+        with smpplib.client.Client(host, port, allow_unknown_opt_params=True) as client:
+            client.connect()
+            if config["LOGIN"]:
+                client.bind_transmitter(
+                    system_id=config["LOGIN"], password=config["PASSWORD"]
+                )
+    except smpplib.exceptions.ConnectionError as exc:
+        raise ConfigurationException(
+            f"Could not connect to the SMPP server '{host}' on port '{port}'"
+        ) from exc
+    except smpplib.exceptions.UnknownCommandError as exc:  # pragma: no cover
+        raise ConfigurationException(exc) from exc
+
+
 def validate_theme(config):
     if not os.path.exists(config["THEME"]) and not os.path.exists(
         os.path.join(ROOT, "themes", config["THEME"])
@@ -243,3 +271,25 @@ def validate_admin_email(config):
         raise ConfigurationException(
             "You must set an administration email if you want to check if users' passwords are compromised."
         )
+
+
+def validate_otp_config(config):
+    if (
+        config["OTP_METHOD"] or config["EMAIL_OTP"] or config["SMS_OTP"]
+    ) and not importlib.util.find_spec("otpauth"):  # pragma: no cover
+        raise ConfigurationException(
+            "You are trying to use OTP but the 'otp' extra is not installed."
+        )
+
+    if config["OTP_METHOD"] not in [None, "TOTP", "HOTP"]:
+        raise ConfigurationException("Invalid OTP method")
+
+    if config["EMAIL_OTP"] and not config["SMTP"]:
+        raise ConfigurationException(
+            "Cannot activate email one-time password authentication without SMTP"
+        )
+
+    if config["SMS_OTP"] and not config["SMPP"]:
+        raise ConfigurationException(
+            "Cannot activate sms one-time password authentication without SMPP"
+        )

canaille/app/features.py

@@ -14,6 +14,22 @@ def has_oidc(self):
     def has_password_recovery(self):
         return self.app.config["CANAILLE"]["ENABLE_PASSWORD_RECOVERY"]
 
+    @property
+    def otp_method(self):
+        return self.app.config["CANAILLE"]["OTP_METHOD"]
+
+    @property
+    def has_otp(self):
+        return bool(self.app.config["CANAILLE"]["OTP_METHOD"])
+
+    @property
+    def has_email_otp(self):
+        return bool(self.app.config["CANAILLE"]["EMAIL_OTP"])
+
+    @property
+    def has_sms_otp(self):
+        return self.app.config["CANAILLE"]["SMS_OTP"]
+
     @property
     def has_registration(self):
         return self.app.config["CANAILLE"]["ENABLE_REGISTRATION"]

canaille/app/sms.py

@@ -0,0 +1,51 @@
+from flask import current_app
+
+
+def send_sms(recipient, sender, text):
+    try:
+        import smpplib.client
+        import smpplib.consts
+        import smpplib.gsm
+    except ImportError as exc:
+        raise RuntimeError(
+            "You are trying to send a sms but the 'sms' extra is not installed."
+        ) from exc
+
+    port = current_app.config["CANAILLE"]["SMPP"]["PORT"]
+    host = current_app.config["CANAILLE"]["SMPP"]["HOST"]
+    login = current_app.config["CANAILLE"]["SMPP"]["LOGIN"]
+    password = current_app.config["CANAILLE"]["SMPP"]["PASSWORD"]
+
+    try:
+        client = smpplib.client.Client(host, port, allow_unknown_opt_params=True)
+        client.connect()
+        try:
+            client.bind_transmitter(system_id=login, password=password)
+            pdu = client.send_message(
+                source_addr_ton=smpplib.consts.SMPP_TON_INTL,
+                source_addr=sender,
+                dest_addr_ton=smpplib.consts.SMPP_TON_INTL,
+                destination_addr=recipient,
+                short_message=bytes(text, "utf-8"),
+            )
+            current_app.logger.debug(pdu.generate())
+        finally:
+            if client.state in [
+                smpplib.consts.SMPP_CLIENT_STATE_BOUND_TX
+            ]:  # pragma: no cover
+                # if bound to transmitter
+                try:
+                    client.unbind()
+                except smpplib.exceptions.UnknownCommandError:
+                    try:
+                        client.unbind()
+                    except smpplib.exceptions.PDUError:
+                        pass
+    except Exception as exc:
+        current_app.logger.warning(f"Could not send sms: {exc}")
+        return False
+    finally:
+        if client:  # pragma: no branch
+            client.disconnect()
+
+    return True

canaille/backends/commands.py

@@ -4,9 +4,11 @@
 import typing
 
 import click
+from flask import current_app
 from flask.cli import AppGroup
 from flask.cli import with_appcontext
 
+from canaille.app import models
 from canaille.app.commands import with_backendcontext
 from canaille.app.models import MODELS
 from canaille.backends import Backend
@@ -77,6 +79,8 @@ def register(cli):
         @cli.command(cls=ModelCommand, factory=factory, name=name, help=command_help)
         def factory_command(): ...
 
+    cli.add_command(reset_otp)
+
 
 def serialize(instance):
     """Quick and dirty serialization method.
@@ -281,3 +285,32 @@ def command(*args, identifier, **kwargs):
             raise click.ClickException(exc) from exc
 
     return command
+
+
+@click.command()
+@with_appcontext
+@with_backendcontext
+@click.argument("identifier")
+def reset_otp(identifier):
+    """Reset one-time password authentication for a user and display the
+    edited user in JSON format in the standard output.
+
+    IDENTIFIER should be a user id or user_name
+    """
+
+    user = Backend.instance.get(models.User, identifier)
+    if not user:
+        raise click.ClickException(f"No user with id '{identifier}'")
+
+    user.initialize_otp()
+    current_app.logger.security(
+        f"Reset one-time password authentication from CLI for {user.user_name}"
+    )
+
+    try:
+        Backend.instance.save(user)
+    except Exception as exc:  # pragma: no cover
+        raise click.ClickException(exc) from exc
+
+    output = json.dumps(serialize(user))
+    click.echo(output)

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,11 @@ class User(canaille.core.models.User, LDAPObject):
         "organization": "o",
         "groups": "memberOf",
         "lock_date": "pwdEndTime",
+        "secret_token": "oathSecret",
+        "last_otp_login": "oathLastLogin",
+        "hotp_counter": "oathHOTPCounter",
+        "one_time_password": "oathTokenPIN",
+        "one_time_password_emission_date": "oathSecretTime",
     }
 
     def match_filter(self, filter):
@@ -44,6 +50,9 @@ def match_filter(self, filter):
         return super().match_filter(filter)
 
     def save(self):
+        if current_app.features.has_otp and not self.secret_token:
+            self.initialize_otp()
+
         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 uuid
 from typing import Any
 
+from flask import current_app
+
+import canaille.backends.memory.models
 from canaille.backends import Backend
 
 
@@ -124,6 +127,13 @@ def get(self, model, identifier=None, /, **kwargs):
         return results[0] if results else None
 
     def save(self, instance):
+        if (
+            isinstance(instance, canaille.backends.memory.models.User)
+            and current_app.features.has_otp
+            and not instance.secret_token
+        ):
+            instance.initialize_otp()
+
         if not instance.id:
             instance.id = str(uuid.uuid4())
 

canaille/backends/sql/backend.py

@@ -110,6 +110,10 @@ def get(self, model, identifier=None, /, **kwargs):
         ).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 typing
 import uuid
 
+from flask import current_app
 from sqlalchemy import Boolean
 from sqlalchemy import Column
 from sqlalchemy import ForeignKey
@@ -100,6 +101,19 @@ 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)
+    hotp_counter: Mapped[int] = mapped_column(Integer, nullable=True)
+    one_time_password: Mapped[str] = mapped_column(String, nullable=True)
+    one_time_password_emission_date: Mapped[datetime.datetime] = mapped_column(
+        TZDateTime(timezone=True), nullable=True
+    )
+
+    def save(self):
+        if current_app.features.has_otp and not self.secret_token:
+            self.initialize_otp()
 
 
 class Group(canaille.core.models.Group, Base, SqlAlchemyModel):