Merge branch 'issue-13-password-policy' into 'main'

Implemented account locking with LDAP ppolicy controls.

See merge request yaal/canaille!118

.gitlab-ci.yml

@@ -7,8 +7,10 @@ stages:
   - release
 
 before_script:
+  - echo deb http://deb.debian.org/debian bullseye-backports main contrib non-free >> /etc/apt/sources.list
   - apt update
-  - env DEBIAN_FRONTEND=noninteractive apt install --yes slapd python3-dev libldap2-dev libsasl2-dev libssl-dev ldap-utils
+  - env apt install --yes python3-dev libldap2-dev libsasl2-dev libssl-dev
+  - env DEBIAN_FRONTEND=noninteractive apt -t bullseye-backports install --yes slapd ldap-utils
   - curl -O https://bootstrap.pypa.io/get-pip.py
   - python get-pip.py
   - pip install --upgrade tox "poetry>1.3.0" coveralls pyyaml tomli

CHANGES.rst

@@ -3,6 +3,12 @@ All notable changes to this project will be documented in this file.
 The format is based on `Keep a Changelog <https://keepachangelog.com/en/1.0.0/>`_,
 and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0.html>`_.
 
+Added
+*****
+
+- Implemented account expiration based on OpenLDAP ppolicy overlay. Needs OpenLDAP 2.5+
+  :issue:`13` :pr:`118`
+
 Fixed
 *****
 

canaille/__init__.py

@@ -149,6 +149,7 @@ def setup_flask(app):
         return {
             "has_smtp": "SMTP" in app.config,
             "has_password_recovery": app.config.get("ENABLE_PASSWORD_RECOVERY", True),
+            "has_account_lockability": app.backend.get().has_account_lockability(),
             "logo_url": app.config.get("LOGO"),
             "favicon_url": app.config.get("FAVICON", app.config.get("LOGO")),
             "website_name": app.config.get("NAME", "Canaille"),

canaille/app/flask.py

@@ -15,7 +15,9 @@ from flask_babel import gettext as _
 def current_user():
     for user_id in session.get("user_id", [])[::-1]:
         user = models.User.get(id=user_id)
-        if user:
+        if user and (
+            not current_app.backend.has_account_lockability() or not user.locked
+        ):
             return user
 
         session["user_id"].remove(user_id)

canaille/backends/__init__.py

@@ -55,3 +55,9 @@ class Backend:
         errors are met.
         """
         raise NotImplementedError()
+
+    def has_account_lockability(self):
+        """
+        Indicates wether the backend supports locking user accounts.
+        """
+        raise NotImplementedError()

canaille/backends/ldap/backend.py

@@ -221,6 +221,11 @@ class LDAPBackend(Backend):
 
         return _(" or ").join(placeholders)
 
+    def has_account_lockability(self):
+        from .ldapobject import LDAPObject
+
+        return "pwdEndTime" in LDAPObject.ldap_object_attributes()
+
 
 def setup_ldap_models(config):
     from .ldapobject import LDAPObject

canaille/backends/ldap/models.py

@@ -2,6 +2,9 @@ import canaille.core.models
 import canaille.oidc.models
 import ldap.filter
 from flask import current_app
+from ldap.controls import DecodeControlTuples
+from ldap.controls.ppolicy import PasswordPolicyControl
+from ldap.controls.ppolicy import PasswordPolicyError
 
 from .backend import LDAPBackend
 from .ldapobject import LDAPObject
@@ -36,6 +39,7 @@ class User(canaille.core.models.User, LDAPObject):
         "organization": "o",
         "last_modified": "modifyTimestamp",
         "groups": "memberOf",
+        "lock_date": "pwdEndTime",
     }
 
     @classmethod
@@ -92,14 +96,37 @@ class User(canaille.core.models.User, LDAPObject):
             current_app.config["BACKENDS"]["LDAP"].get("TIMEOUT"),
         )
 
+        message = None
         try:
-            conn.simple_bind_s(self.id, password)
+            res = conn.simple_bind_s(
-            return True
+                self.id, password, serverctrls=[PasswordPolicyControl()]
-        except ldap.INVALID_CREDENTIALS:
+            )
-            return False
+            controls = res[3]
+            result = True
+        except ldap.INVALID_CREDENTIALS as exc:
+            controls = DecodeControlTuples(exc.args[0]["ctrls"])
+            result = False
         finally:
             conn.unbind_s()
 
+        for control in controls:
+
+            def gettext(x):
+                return x
+
+            if (
+                control.controlType == PasswordPolicyControl.controlType
+                and control.error == PasswordPolicyError.namedValues["accountLocked"]
+            ):
+                message = gettext("Your account has been locked.")
+            elif (
+                control.controlType == PasswordPolicyControl.controlType
+                and control.error == PasswordPolicyError.namedValues["changeAfterReset"]
+            ):
+                message = gettext("You should change your password.")
+
+        return result, message
+
     def set_password(self, password):
         conn = LDAPBackend.get().connection
         conn.passwd_s(

canaille/conf/config.sample.toml

@@ -122,7 +122,11 @@ GROUP_BASE = "ou=groups,dc=mydomain,dc=tld"
 # object that users will be able to read and/or write.
 [ACL.DEFAULT]
 PERMISSIONS = ["edit_self", "use_oidc"]
-READ = ["user_name", "groups"]
+READ = [
+    "user_name",
+    "groups",
+    "lock_date",
+]
 WRITE = [
     "photo",
     "given_name",
@@ -153,7 +157,10 @@ PERMISSIONS = [
     "delete_account",
     "impersonate_users",
 ]
-WRITE = ["groups"]
+WRITE = [
+    "groups",
+    "lock_date",
+]
 
 [OIDC]
 # Wether a token is needed for the RFC7591 dynamical client registration.

canaille/core/account.py

@@ -115,17 +115,21 @@ def password():
         if user and not user.has_password():
             return redirect(url_for("account.firstlogin", user_name=user.user_name[0]))
 
-        if (
+        if not form.validate() or not user:
-            not form.validate()
-            or not user
-            or not user.check_password(form.password.data)
-        ):
             models.User.logout()
             flash(_("Login failed, please check your information"), "error")
             return render_template(
                 "password.html", form=form, username=session["attempt_login"]
             )
 
+        success, message = user.check_password(form.password.data)
+        if not success:
+            models.User.logout()
+            flash(message or _("Login failed, please check your information"), "error")
+            return render_template(
+                "password.html", form=form, username=session["attempt_login"]
+            )
+
         del session["attempt_login"]
         user.login()
         flash(
@@ -547,6 +551,28 @@ def profile_settings(user, username):
 
         return profile_settings_edit(user, edited_user)
 
+    if (
+        request.form.get("action") == "lock"
+        and Backend.get().has_account_lockability()
+        and not edited_user.locked
+    ):
+        flash(_("The account has been locked"), "success")
+        edited_user.lock()
+        edited_user.save()
+
+        return profile_settings_edit(user, edited_user)
+
+    if (
+        request.form.get("action") == "unlock"
+        and Backend.get().has_account_lockability()
+        and edited_user.locked
+    ):
+        flash(_("The account has been unlocked"), "success")
+        edited_user.unlock()
+        edited_user.save()
+
+        return profile_settings_edit(user, edited_user)
+
     abort(400)
 
 
@@ -554,7 +580,7 @@ def profile_settings_edit(editor, edited_user):
     menuitem = "profile" if editor.id == editor.id else "users"
     fields = editor.read | editor.write
 
-    available_fields = {"password", "groups", "user_name"}
+    available_fields = {"password", "groups", "user_name", "lock_date"}
     data = {
         k: getattr(edited_user, k)[0]
         if getattr(edited_user, k) and isinstance(getattr(edited_user, k), list)
@@ -580,6 +606,13 @@ def profile_settings_edit(editor, edited_user):
                 if attribute.name == "groups" and "groups" in editor.write:
                     edited_user.groups = attribute.data
 
+                elif (
+                    attribute.name == "lock_date"
+                    and Backend.get().has_account_lockability()
+                    and form[attribute.name].data
+                ):
+                    edited_user.lock(form[attribute.name].data)
+
             if (
                 "password1" in request.form
                 and form["password1"].data

canaille/core/forms.py

@@ -285,10 +285,22 @@ def profile_form(write_field_names, readonly_field_names, user=None):
     if "groups" in fields and not models.Group.query():
         del fields["groups"]
 
+    fields["lock_date"] = wtforms.DateTimeLocalField(
+        _("Account expiration"),
+        validators=[wtforms.validators.Optional()],
+        format=[
+            "%Y-%m-%d %H:%M",
+            "%Y-%m-%dT%H:%M",
+            "%Y-%m-%d %H:%M:%S",
+            "%Y-%m-%dT%H:%M:%S",
+        ],
+    )
+
     form = HTMXBaseForm(fields)
     form.user = user
     for field in form:
         if field.name in readonly_field_names - write_field_names:
+            field.render_kw = field.render_kw or {}
             field.render_kw["readonly"] = "true"
 
     return form

canaille/core/models.py

@@ -1,3 +1,5 @@
+import datetime
+
 from flask import session
 
 
@@ -72,6 +74,18 @@ class User:
     def can_impersonate_users(self):
         return "impersonate_users" in self.permissions
 
+    def lock(self, lock_datetime=None):
+        self.lock_date = lock_datetime or datetime.datetime.now(datetime.timezone.utc)
+
+    def unlock(self):
+        del self.lock_date
+
+    @property
+    def locked(self):
+        return bool(self.lock_date) and self.lock_date < datetime.datetime.now(
+            datetime.timezone.utc
+        )
+
 
 class Group:
     pass

canaille/oidc/endpoints.py

@@ -77,14 +77,18 @@ def authorize():
             return render_template("login.html", form=form, menu=False)
 
         user = models.User.get_from_login(form.login.data)
-        if (
+        if not form.validate() or not user:
-            not form.validate()
-            or not user
-            or not user.check_password(form.password.data)
-        ):
             flash(_("Login failed, please check your information"), "error")
             return render_template("login.html", form=form, menu=False)
 
+        success, message = user.check_password(form.password.data)
+        if not success:
+            flash(
+                _(message or "Login failed, please check your information"),
+                "error",
+            )
+            return render_template("login.html", form=form, menu=False)
+
         user.login()
 
         return redirect(request.url)

canaille/oidc/oauth.py

@@ -193,8 +193,11 @@ class PasswordGrant(_ResourceOwnerPasswordCredentialsGrant):
 
     def authenticate_user(self, username, password):
         user = models.User.get_from_login(username)
+        if not user:
+            return None
 
-        if not user or not user.check_password(password):
+        success, _ = user.check_password(password)
+        if not success:
             return None
 
         return user

canaille/templates/partial/users.html

@@ -25,7 +25,9 @@
                 {% if user.can_read("photo") %}
                     <td>
                         <a href="{{ url_for('account.profile_edition', username=watched_user.user_name[0]) }}">
-                            {% if watched_user.photo and watched_user.photo[0] %}
+                            {% if user.can_manage_users and watched_user.locked %}
+                                <i class="lock circle big black icon" title="{% trans %}This account is locked{% endtrans %}"></i>
+                            {% elif watched_user.photo and watched_user.photo[0] %}
                                 <img class="ui avatar image" src="{{ url_for("account.photo", user_name=watched_user.user_name[0], field="photo") }}" alt="User photo">
                             {% else %}
                                 <i class="user circle big black icon"></i>

canaille/templates/profile_settings.html

@@ -63,6 +63,28 @@
         </div>
     {% endif %}
 
+    {% if has_account_lockability and user.can_manage_users and not edited_user.locked %}
+        <div id="modal-lock" class="ui basic modal">
+            <div class="ui icon header">
+                <i class="lock icon"></i>
+                {% trans %}Account locking{% endtrans %}
+            </div>
+            <div class="content">
+                <p>
+                    {% if user.user_name != edited_user.user_name %}
+                        {{ _("Are you sure you want to lock this account? The user won't be able to login until their account is unlocked.") }}
+                    {% else %}
+                        {{ _("Are you sure you want to lock your account? You won't be abel to login until your account is unlocked.") }}
+                    {% endif %}
+                </p>
+            </div>
+            <div class="actions">
+                <div class="ui inverted cancel button">{% trans %}Cancel{% endtrans %}</div>
+                <div class="ui inverted red approve button">{% trans %}Lock{% endtrans %}</div>
+            </div>
+        </div>
+    {% endif %}
+
     <div class="ui clearing segment">
         <h2 class="ui center aligned header">
             <div class="content">
@@ -88,6 +110,10 @@
                 {% endblock %}
             {% endif %}
 
+            {% if has_account_lockability and "lock_date" in form and not edited_user.locked %}
+                {% block lock_date_field scoped %}{{ render_field(form.lock_date) }}{% endblock %}
+            {% endif %}
+
             {% if user.can_manage_users %}
 
                 {% if not edited_user.has_password() %}
@@ -127,10 +153,32 @@
 
                 {% endif %}
 
+                {% if has_account_lockability and edited_user.locked %}
+
+                    <div class="ui message warning visible">
+                        <button type="submit" name="action" value="unlock" class="ui right floated button">
+                            {% trans %}Unlock{% endtrans %}
+                        </button>
+                        <div class="header">
+                            {% trans %}This user account is locked{% endtrans %}
+                        </div>
+                        <p>
+                            {% trans %}The user won't be able to connect until their account is unlocked.{% endtrans %}
+                        </p>
+                    </div>
+
+                {% endif %}
+
             {% endif %}
 
             <div class="ui right aligned container">
                 <div class="ui stackable buttons">
+                    {% if has_account_lockability and "lock_date" in user.write and not edited_user.locked %}
+                        <button type="submit" class="ui right floated basic negative button confirm" name="action" value="lock" id="lock">
+                            {% trans %}Lock the account{% endtrans %}
+                        </button>
+                    {% endif %}
+
                     {% if user.can_manage_users or self_deletion %}
                         <button type="submit" class="ui right floated basic negative button confirm" name="action" value="delete" id="delete">
                             {% if user.user_name != edited_user.user_name %}

demo/conf-docker/canaille.toml

@@ -123,7 +123,11 @@ GROUP_BASE = "ou=groups,dc=mydomain,dc=tld"
 # object that users will be able to read and/or write.
 [ACL.DEFAULT]
 PERMISSIONS = ["edit_self", "use_oidc"]
-READ = ["user_name", "groups"]
+READ = [
+    "user_name",
+    "groups",
+    "lock_date",
+]
 WRITE = [
     "photo",
     "given_name",
@@ -154,7 +158,10 @@ PERMISSIONS = [
     "delete_account",
     "impersonate_users",
 ]
-WRITE = ["groups"]
+WRITE = [
+    "groups",
+    "lock_date",
+]
 
 [ACL.HALF_ADMIN]
 FILTER = {groups = "moderators"}

demo/conf/canaille.toml

@@ -123,7 +123,11 @@ GROUP_BASE = "ou=groups,dc=mydomain,dc=tld"
 # object that users will be able to read and/or write.
 [ACL.DEFAULT]
 PERMISSIONS = ["edit_self", "use_oidc"]
-READ = ["user_name", "groups"]
+READ = [
+    "user_name",
+    "groups",
+    "lock_date",
+]
 WRITE = [
     "photo",
     "given_name",
@@ -154,7 +158,10 @@ PERMISSIONS = [
     "delete_account",
     "impersonate_users",
 ]
-WRITE = ["groups"]
+WRITE = [
+    "groups",
+    "lock_date",
+]
 
 [ACL.HALF_ADMIN]
 FILTER = {groups = "moderators"}

demo/docker-compose.yml

@@ -7,11 +7,13 @@ services:
     environment:
       - LDAP_DOMAIN=mydomain.tld
     volumes:
-      - ./ldif/memberof-config.ldif:/container/service/slapd/assets/config/bootstrap/ldif/03-memberOf.ldif:ro
-      - ./ldif/refint-config.ldif:/container/service/slapd/assets/config/bootstrap/ldif/04-refint.ldif:ro
       # memberof overlay is already present in openldap docker image but only for groupOfUniqueNames. We need to overwrite it (until canaille can handle groupOfUniqueNames).
       # https://github.com/osixia/docker-openldap/blob/master/image/service/slapd/assets/config/bootstrap/ldif/03-memberOf.ldif
+      - ./ldif/memberof-config.ldif:/container/service/slapd/assets/config/bootstrap/ldif/03-memberOf.ldif:ro
+      - ./ldif/refint-config.ldif:/container/service/slapd/assets/config/bootstrap/ldif/04-refint.ldif:ro
       - ../canaille/backends/ldap/schemas/oauth2-openldap.ldif:/container/service/slapd/assets/config/bootstrap/ldif/custom/40-oauth2.ldif:ro
+      - ./ldif/ppolicy-config.ldif:/container/service/slapd/assets/config/bootstrap/ldif/custom/30-ppolicy.ldif:ro
+      - ./ldif/ppolicy.ldif:/container/service/slapd/assets/config/bootstrap/ldif/custom/40-ppolicy.ldif:ro
       - ./ldif/bootstrap-users-tree.ldif:/container/service/slapd/assets/config/bootstrap/ldif/custom/50-bootstrap-users-tree.ldif:ro
       - ./ldif/bootstrap-oidc-tree.ldif:/container/service/slapd/assets/config/bootstrap/ldif/custom/50-bootstrap-oidc-tree.ldif:ro
     command: --copy-service --loglevel debug

demo/ldap-server.py

@@ -10,11 +10,13 @@ schemas = [
         "cosine.ldif",
         "nis.ldif",
         "inetorgperson.ldif",
+        "ppolicy.ldif",
     ]
     if os.path.exists(os.path.join(slapd.Slapd.SCHEMADIR, schema))
 ] + [
     "ldif/memberof-config.ldif",
     "ldif/refint-config.ldif",
+    "ldif/ppolicy-config.ldif",
 ]
 
 slapd = slapd.Slapd(
@@ -47,6 +49,7 @@ try:
     )
 
     for ldif in (
+        "ldif/ppolicy.ldif",
         "ldif/bootstrap-users-tree.ldif",
         "ldif/bootstrap-oidc-tree.ldif",
     ):

demo/ldif/ppolicy-config.ldif

@@ -0,0 +1,11 @@
+dn: cn=module,cn=config
+cn: module
+objectClass: olcModuleList
+olcModuleLoad: ppolicy
+
+dn: olcOverlay=ppolicy,olcDatabase={1}mdb,cn=config
+objectClass: olcOverlayConfig
+objectClass: olcPPolicyConfig
+olcOverlay: ppolicy
+olcPPolicyDefault: cn=passwordDefault,dc=mydomain,dc=tld
+olcPPolicyUseLockout: TRUE

demo/ldif/ppolicy.ldif

@@ -0,0 +1,11 @@
+dn: cn=passwordDefault,dc=mydomain,dc=tld
+objectClass: person
+objectClass: top
+objectClass: pwdPolicy
+sn: passwordDefault
+cn: passwordDefault
+pwdAttribute: userPassword
+pwdMustChange: TRUE
+pwdLockout: TRUE
+pwdAllowUserChange: TRUE
+pwdGraceAuthNLimit: 1

tests/backends/test_backends.py

@@ -15,3 +15,6 @@ def test_required_methods(testclient):
 
     with pytest.raises(NotImplementedError):
         backend.teardown()
+
+    with pytest.raises(NotImplementedError):
+        backend.has_account_lockability()

tests/conftest.py

@@ -1,3 +1,5 @@
+import os
+
 import pytest
 import slapd
 from canaille import create_app
@@ -9,14 +11,21 @@ from werkzeug.security import gen_salt
 
 class CustomSlapdObject(slapd.Slapd):
     def __init__(self):
-        super().__init__(
+        schemas = [
-            suffix="dc=mydomain,dc=tld",
+            schema
-            schemas=(
+            for schema in [
                 "core.ldif",
                 "cosine.ldif",
                 "nis.ldif",
                 "inetorgperson.ldif",
-            ),
+                "ppolicy.ldif",
+            ]
+            if os.path.exists(os.path.join(self.SCHEMADIR, schema))
+        ]
+
+        super().__init__(
+            suffix="dc=mydomain,dc=tld",
+            schemas=schemas,
         )
 
     def init_tree(self):
@@ -48,6 +57,8 @@ def slapd_server():
         slapd.init_tree()
         for ldif in (
             "demo/ldif/memberof-config.ldif",
+            "demo/ldif/ppolicy-config.ldif",
+            "demo/ldif/ppolicy.ldif",
             "canaille/backends/ldap/schemas/oauth2-openldap.ldif",
             "demo/ldif/bootstrap-users-tree.ldif",
             "demo/ldif/bootstrap-oidc-tree.ldif",
@@ -80,7 +91,7 @@ def configuration(slapd_server, smtpd):
         "ACL": {
             "DEFAULT": {
                 "READ": ["user_name", "groups"],
-                "PERMISSIONS": ["edit_self", "use_oidc"],
+                "PERMISSIONS": ["edit_self", "use_oidc", "lock_date"],
                 "WRITE": [
                     "email",
                     "given_name",
@@ -112,6 +123,7 @@ def configuration(slapd_server, smtpd):
                 ],
                 "WRITE": [
                     "groups",
+                    "lock_date",
                 ],
             },
             "MODERATOR": {

tests/core/test_account.py

@@ -1,3 +1,4 @@
+import datetime
 from unittest import mock
 
 from canaille.app import models
@@ -89,6 +90,21 @@ def test_signin_wrong_password(testclient, user):
     assert ("error", "Login failed, please check your information") in res.flashes
 
 
+def test_signin_bad_csrf(testclient, user):
+    with testclient.session_transaction() as session:
+        assert not session.get("user_id")
+
+    res = testclient.get("/login", status=200)
+
+    res.form["login"] = "John (johnny) Doe"
+    res = res.form.submit(status=302)
+    res = res.follow(status=200)
+
+    res.form["password"] = ""
+    res = res.form.submit(status=200)
+    assert ("error", "Login failed, please check your information") in res.flashes
+
+
 def test_signin_with_alternate_attribute(testclient, user):
     res = testclient.get("/login", status=200)
 
@@ -336,3 +352,94 @@ def test_user_self_deletion(testclient, backend):
         assert not sess.get("user_id")
 
     testclient.app.config["ACL"]["DEFAULT"]["PERMISSIONS"] = []
+
+
+def test_account_locking(user, backend):
+    assert not user.locked
+    assert not user.lock_date
+    assert user.check_password("correct horse battery staple") == (True, None)
+
+    user.lock()
+    user.save()
+    assert user.locked
+    assert user.lock_date <= datetime.datetime.now(datetime.timezone.utc)
+    assert models.User.get(id=user.id).locked
+    assert models.User.get(id=user.id).lock_date <= datetime.datetime.now(
+        datetime.timezone.utc
+    )
+    assert user.check_password("correct horse battery staple") == (
+        False,
+        "Your account has been locked.",
+    )
+
+    user.unlock()
+    user.save()
+    assert not user.locked
+    assert not user.lock_date
+    assert not models.User.get(id=user.id).locked
+    assert not models.User.get(id=user.id).lock_date
+    assert user.check_password("correct horse battery staple") == (True, None)
+
+
+def test_account_locking_past_date(user, backend):
+    assert not user.locked
+    assert not user.lock_date
+    assert user.check_password("correct horse battery staple") == (True, None)
+
+    lock_datetime = datetime.datetime.now(datetime.timezone.utc).replace(
+        microsecond=0
+    ) - datetime.timedelta(days=30)
+    user.lock(lock_datetime)
+    user.save()
+    assert user.locked
+    assert user.lock_date == lock_datetime
+    assert models.User.get(id=user.id).locked
+    assert models.User.get(id=user.id).lock_date == lock_datetime
+    assert user.check_password("correct horse battery staple") == (
+        False,
+        "Your account has been locked.",
+    )
+
+
+def test_account_locking_future_date(user, backend):
+    assert not user.locked
+    assert not user.lock_date
+    assert user.check_password("correct horse battery staple") == (True, None)
+
+    lock_datetime = datetime.datetime.now(datetime.timezone.utc).replace(
+        microsecond=0
+    ) + datetime.timedelta(days=365 * 4)
+    user.lock(lock_datetime)
+    user.save()
+    assert not user.locked
+    assert user.lock_date == lock_datetime
+    assert not models.User.get(id=user.id).locked
+    assert models.User.get(id=user.id).lock_date == lock_datetime
+    assert user.check_password("correct horse battery staple") == (True, None)
+
+
+def test_signin_locked_account(testclient, user):
+    with testclient.session_transaction() as session:
+        assert not session.get("user_id")
+
+    user.lock()
+    user.save()
+
+    res = testclient.get("/login", status=200)
+    res.form["login"] = "user"
+
+    res = res.form.submit(status=302).follow(status=200)
+    res.form["password"] = "correct horse battery staple"
+
+    res = res.form.submit()
+    res.mustcontain("Your account has been locked.")
+
+    user.unlock()
+    user.save()
+
+
+def test_account_locked_during_session(testclient, logged_user):
+    testclient.get("/profile/user/settings", status=200)
+    logged_user.lock()
+    logged_user.save()
+    testclient.get("/profile/user/settings", status=403)

tests/core/test_invitation.py

@@ -41,7 +41,7 @@ def test_invitation(testclient, logged_admin, foo_group, smtpd):
 
     user = models.User.get_from_login("someone")
     foo_group.reload()
-    assert user.check_password("whatever")
+    assert user.check_password("whatever")[0]
     assert user.groups == [foo_group]
 
     with testclient.session_transaction() as sess:
@@ -91,7 +91,7 @@ def test_invitation_editable_user_name(testclient, logged_admin, foo_group, smtp
 
     user = models.User.get_from_login("djorje")
     foo_group.reload()
-    assert user.check_password("whatever")
+    assert user.check_password("whatever")[0]
     assert user.groups == [foo_group]
 
     with testclient.session_transaction() as sess:
@@ -133,7 +133,7 @@ def test_generate_link(testclient, logged_admin, foo_group, smtpd):
 
     user = models.User.get_from_login("sometwo")
     foo_group.reload()
-    assert user.check_password("whatever")
+    assert user.check_password("whatever")[0]
     assert user.groups == [foo_group]
 
     with testclient.session_transaction() as sess:

tests/core/test_models.py

@@ -44,10 +44,10 @@ def test_user_has_password(testclient, backend):
 
 
 def test_user_set_and_check_password(testclient, user, backend):
-    assert not user.check_password("something else")
+    assert not user.check_password("something else")[0]
-    assert user.check_password("correct horse battery staple")
+    assert user.check_password("correct horse battery staple")[0]
 
     user.set_password("something else")
 
-    assert user.check_password("something else")
+    assert user.check_password("something else")[0]
-    assert not user.check_password("correct horse battery staple")
+    assert not user.check_password("correct horse battery staple")[0]

tests/core/test_password_reset.py

@@ -2,7 +2,7 @@ from canaille.core.account import profile_hash
 
 
 def test_password_reset(testclient, user):
-    assert not user.check_password("foobarbaz")
+    assert not user.check_password("foobarbaz")[0]
     hash = profile_hash("user", user.email[0], user.password[0])
 
     res = testclient.get("/reset/user/" + hash, status=200)
@@ -13,7 +13,7 @@ def test_password_reset(testclient, user):
     assert ("success", "Your password has been updated successfuly") in res.flashes
 
     user.reload()
-    assert user.check_password("foobarbaz")
+    assert user.check_password("foobarbaz")[0]
 
     res = testclient.get("/reset/user/" + hash)
     assert (
@@ -39,7 +39,7 @@ def test_password_reset_bad_password(testclient, user):
     res.form["confirmation"] = "typo"
     res = res.form.submit(status=200)
 
-    assert user.check_password("correct horse battery staple")
+    assert user.check_password("correct horse battery staple")[0]
 
 
 def test_unavailable_if_no_smtp(testclient, user):

tests/core/test_profile_creation.py

@@ -28,7 +28,7 @@ def test_user_creation_edition_and_deletion(
     foo_group.reload()
     assert "George" == george.given_name[0]
     assert george.groups == [foo_group]
-    assert george.check_password("totoyolo")
+    assert george.check_password("totoyolo")[0]
 
     res = testclient.get("/users", status=200)
     res.mustcontain("george")
@@ -47,7 +47,7 @@ def test_user_creation_edition_and_deletion(
 
     george = models.User.get_from_login("george")
     assert "Georgio" == george.given_name[0]
-    assert george.check_password("totoyolo")
+    assert george.check_password("totoyolo")[0]
 
     foo_group.reload()
     bar_group.reload()

tests/core/test_profile_settings.py

@@ -1,3 +1,4 @@
+import datetime
 from unittest import mock
 
 from canaille.app import models
@@ -36,7 +37,7 @@ def test_edition(
     assert foo_group.members == [logged_user]
     assert bar_group.members == [admin]
 
-    assert logged_user.check_password("correct horse battery staple")
+    assert logged_user.check_password("correct horse battery staple")[0]
 
     logged_user.user_name = ["user"]
     logged_user.save()
@@ -73,7 +74,7 @@ def test_edition_without_groups(
     logged_user.reload()
 
     assert logged_user.user_name == ["user"]
-    assert logged_user.check_password("correct horse battery staple")
+    assert logged_user.check_password("correct horse battery staple")[0]
 
     logged_user.user_name = ["user"]
     logged_user.save()
@@ -88,7 +89,7 @@ def test_password_change(testclient, logged_user):
     res = res.form.submit(name="action", value="edit").follow()
 
     logged_user.reload()
-    assert logged_user.check_password("new_password")
+    assert logged_user.check_password("new_password")[0]
 
     res = testclient.get("/profile/user/settings", status=200)
 
@@ -100,7 +101,7 @@ def test_password_change(testclient, logged_user):
     res = res.follow()
 
     logged_user.reload()
-    assert logged_user.check_password("correct horse battery staple")
+    assert logged_user.check_password("correct horse battery staple")[0]
 
 
 def test_password_change_fail(testclient, logged_user):
@@ -112,7 +113,7 @@ def test_password_change_fail(testclient, logged_user):
     res = res.form.submit(name="action", value="edit", status=200)
 
     logged_user.reload()
-    assert logged_user.check_password("correct horse battery staple")
+    assert logged_user.check_password("correct horse battery staple")[0]
 
     res = testclient.get("/profile/user/settings", status=200)
 
@@ -122,7 +123,7 @@ def test_password_change_fail(testclient, logged_user):
     res = res.form.submit(name="action", value="edit", status=200)
 
     logged_user.reload()
-    assert logged_user.check_password("correct horse battery staple")
+    assert logged_user.check_password("correct horse battery staple")[0]
 
 
 def test_password_initialization_mail(smtpd, testclient, backend, logged_admin):
@@ -302,3 +303,87 @@ def test_edition_permission(
 
     testclient.app.config["ACL"]["DEFAULT"]["PERMISSIONS"] = ["edit_self"]
     testclient.get("/profile/user/settings", status=200)
+
+
+def test_account_locking(
+    testclient,
+    backend,
+    logged_admin,
+    user,
+):
+    res = testclient.get("/profile/user/settings")
+    assert not user.lock_date
+    assert not user.locked
+    res.mustcontain("Lock the account")
+    res.mustcontain(no="Unlock")
+
+    res = res.form.submit(name="action", value="lock")
+    user = models.User.get(id=user.id)
+    assert user.lock_date <= datetime.datetime.now(datetime.timezone.utc)
+    assert user.locked
+    res.mustcontain("The account has been locked")
+    res.mustcontain(no="Lock the account")
+    res.mustcontain("Unlock")
+
+    res = res.form.submit(name="action", value="unlock")
+    user = models.User.get(id=user.id)
+    assert not user.lock_date
+    assert not user.locked
+    assert "The account has been unlocked"
+    res.mustcontain("Lock the account")
+    res.mustcontain(no="Unlock")
+
+
+def test_lock_date(
+    testclient,
+    backend,
+    logged_admin,
+    user,
+):
+    res = testclient.get("/profile/user/settings", status=200)
+    assert not user.lock_date
+    assert not user.locked
+
+    expiration_datetime = datetime.datetime.now(datetime.timezone.utc).replace(
+        second=0, microsecond=0
+    ) + datetime.timedelta(days=30)
+    res.form["lock_date"] = expiration_datetime.strftime("%Y-%m-%d %H:%M")
+    res = res.form.submit(name="action", value="edit").follow()
+    user = models.User.get(id=user.id)
+    assert user.lock_date == expiration_datetime
+    assert not user.locked
+    assert res.form["lock_date"].value == expiration_datetime.strftime("%Y-%m-%d %H:%M")
+
+    expiration_datetime = datetime.datetime.now(datetime.timezone.utc).replace(
+        second=0, microsecond=0
+    ) - datetime.timedelta(days=30)
+    res.form["lock_date"] = expiration_datetime.strftime("%Y-%m-%d %H:%M")
+    res = res.form.submit(name="action", value="edit").follow()
+    user = models.User.get(id=user.id)
+    assert user.lock_date == expiration_datetime
+    assert user.locked
+
+    res = res.form.submit(name="action", value="unlock")
+    user = models.User.get(id=user.id)
+    assert not user.lock_date
+    assert not user.locked
+
+
+def test_account_limit_values(
+    testclient,
+    backend,
+    logged_admin,
+    user,
+):
+    res = testclient.get("/profile/user/settings", status=200)
+    assert not user.lock_date
+    assert not user.locked
+
+    expiration_datetime = datetime.datetime.max.replace(
+        microsecond=0, tzinfo=datetime.timezone.utc
+    )
+    res.form["lock_date"] = expiration_datetime.strftime("%Y-%m-%d %H:%M:%S")
+    res = res.form.submit(name="action", value="edit").follow()
+    user = models.User.get(id=user.id)
+    assert user.lock_date == expiration_datetime
+    assert not user.locked

tests/oidc/test_password_flow.py

@@ -63,7 +63,7 @@ def test_password_flow_post(testclient, user, client):
     assert res.json["name"] == "John (johnny) Doe"
 
 
-def test_invalid_credentials(testclient, user, client):
+def test_invalid_user(testclient, user, client):
     res = testclient.post(
         "/oauth/token",
         params=dict(
@@ -80,3 +80,22 @@ def test_invalid_credentials(testclient, user, client):
         "error": "invalid_request",
         "error_description": 'Invalid "username" or "password" in request.',
     }
+
+
+def test_invalid_credentials(testclient, user, client):
+    res = testclient.post(
+        "/oauth/token",
+        params=dict(
+            grant_type="password",
+            username="user",
+            password="invalid",
+            scope="openid profile groups",
+        ),
+        headers={"Authorization": f"Basic {client_credentials(client)}"},
+        status=400,
+    )
+
+    assert res.json == {
+        "error": "invalid_request",
+        "error_description": 'Invalid "username" or "password" in request.',
+    }