Implemented LDAP ppolicy support.

This commit is contained in:
Éloi Rivard 2022-11-01 12:25:21 +01:00 committed by Éloi Rivard
parent d8158d6a72
commit 17c2f7a5cd
30 changed files with 490 additions and 51 deletions

View file

@ -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

View file

@ -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
*****

View file

@ -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"),

View file

@ -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)

View file

@ -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()

View file

@ -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

View file

@ -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)
return True
except ldap.INVALID_CREDENTIALS:
return False
res = conn.simple_bind_s(
self.id, password, serverctrls=[PasswordPolicyControl()]
)
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(

View file

@ -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.

View file

@ -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 (
not form.validate()
or not user
or not user.check_password(form.password.data)
):
if not form.validate() or not user:
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

View file

@ -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

View file

@ -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

View file

@ -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 (
not form.validate()
or not user
or not user.check_password(form.password.data)
):
if not form.validate() or not user:
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)

View file

@ -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

View file

@ -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>

View file

@ -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 %}

View file

@ -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"}

View file

@ -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"}

View file

@ -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

View file

@ -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",
):

View file

@ -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

11
demo/ldif/ppolicy.ldif Normal file
View file

@ -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

View file

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

View file

@ -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__(
suffix="dc=mydomain,dc=tld",
schemas=(
schemas = [
schema
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": {

View file

@ -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)

View file

@ -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:

View file

@ -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 user.check_password("correct horse battery staple")
assert not user.check_password("something else")[0]
assert user.check_password("correct horse battery staple")[0]
user.set_password("something else")
assert user.check_password("something else")
assert not user.check_password("correct horse battery staple")
assert user.check_password("something else")[0]
assert not user.check_password("correct horse battery staple")[0]

View file

@ -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):

View file

@ -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()

View file

@ -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

View file

@ -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.',
}