From 5fd0ec9aa83b0fca8c9031f9685785f572ed8890 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Dachary?= <ldachary@easter-eggs.com>
Date: Tue, 9 Feb 2021 08:46:22 +0100
Subject: [PATCH 1/4] ldap: optionally collects messages from ppolicy

Enable PasswordPolicyControl[0] in authenticate() and log the
information it returns, on success or error. In the context of a
request, this information is also set as a message[1] to be displayed
to the user.

All messages are translated into French.

[0] https://github.com/python-ldap/python-ldap/blob/python-ldap-3.3.1/Lib/ldap/controls/ppolicy.py
[1] https://docs.djangoproject.com/en/3.1/ref/contrib/messages/

Fixes: #50959

License: MIT
---
 src/authentic2/backends/ldap_backend.py       |  58 +++-
 .../locale/fr/LC_MESSAGES/django.po           |  50 ++++
 tests/test_ldap.py                            | 251 ++++++++++++++++++
 3 files changed, 354 insertions(+), 5 deletions(-)

diff --git a/src/authentic2/backends/ldap_backend.py b/src/authentic2/backends/ldap_backend.py
index 5ffd94903..81ba48263 100644
--- a/src/authentic2/backends/ldap_backend.py
+++ b/src/authentic2/backends/ldap_backend.py
@@ -23,7 +23,9 @@ try:
     from ldap.filter import filter_format
     from ldap.dn import escape_dn_chars
     from ldap.ldapobject import ReconnectLDAPObject as NativeLDAPObject
-    from ldap.controls import SimplePagedResultsControl
+    from ldap.controls import SimplePagedResultsControl, DecodeControlTuples
+    from ldap.controls import ppolicy
+    from pyasn1.codec.der import decoder
     PYTHON_LDAP3 = [int(x) for x in ldap.__version__.split('.')] >= [3]
     LDAPObject = NativeLDAPObject
 except ImportError:
@@ -41,11 +43,13 @@ import json
 from django.core.cache import cache
 from django.core.exceptions import ImproperlyConfigured
 from django.conf import settings
+from django.contrib import messages
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Group
 from django.utils.encoding import force_bytes, force_text
 from django.utils import six
 from django.utils.six.moves.urllib import parse as urlparse
+from django.utils.translation import ugettext as _, ngettext
 
 from authentic2.a2_rbac.models import Role
 
@@ -232,6 +236,34 @@ def map_text(d):
     raise NotImplementedError
 
 
+def password_policy_control_messages(ctrl):
+    messages = []
+
+    if ctrl.error:
+        error = ppolicy.PasswordPolicyError.namedValues[ctrl.error]
+        error2message = {
+            'passwordExpired': _('The password expired'),
+            'accountLocked': _('The account is locked.'),
+            'changeAfterReset': _('The password was reset and must be changed.'),
+            'passwordModNotAllowed': _('It is not possible to modify the password.'),
+            'mustSupplyOldPassword': _('The old password must be supplied.'),
+            'insufficientPasswordQuality': _('The password does not meet the quality requirements.'),
+            'passwordTooShort': _('The password is too short.'),
+            'passwordTooYoung': _('It is too soon to change the password.'),
+            'passwordInHistory': _('This password was recently used and cannot be used again.'),
+        }
+        messages.append(error2message.get(error, _('Unexpected error {error}')))
+        return messages
+
+    if ctrl.timeBeforeExpiration:
+        messages.append(_(f'The password will expire in {ctrl.timeBeforeExpiration} seconds.'))
+    if ctrl.graceAuthNsRemaining:
+        messages.append(ngettext(
+            f'This is the last time this password can be used.',
+            f'This password can only be used {ctrl.graceAuthNsRemaining} times, including this one.',
+            ctrl.graceAuthNsRemaining))
+    return messages
+
 class LDAPUser(User):
     SESSION_LDAP_DATA_KEY = 'ldap-data'
     _changed = False
@@ -537,6 +569,17 @@ class LDAPBackend(object):
         log.debug('got config %r', blocks)
         return blocks
 
+    @staticmethod
+    def process_controls(request, authz_id, ctrls):
+        for c in ctrls:
+            if c.controlType == ppolicy.PasswordPolicyControl.controlType:
+                message = ' '.join(password_policy_control_messages(c))
+                log.info('%s: %s', authz_id, message)
+                if request is not None:
+                    messages.add_message(request, messages.WARNING, message)
+            else:
+                log.info('%s: %s', authz_id, vars(c))
+
     def authenticate(self, request=None, username=None, password=None, realm=None, ou=None):
         if username is None or password is None:
             return None
@@ -566,11 +609,11 @@ class LDAPBackend(object):
                 log.error(
                     "account name authentication filter doesn't contain '%s'")
                 continue
-            user = self.authenticate_block(block, uid, password)
+            user = self.authenticate_block(request, block, uid, password)
             if user is not None:
                 return user
 
-    def authenticate_block(self, block, username, password):
+    def authenticate_block(self, request, block, username, password):
         for conn in self.get_connections(block):
             ldap_uri = conn.get_option(ldap.OPT_URI)
             authz_ids = []
@@ -628,7 +671,10 @@ class LDAPBackend(object):
                         if failed:
                             continue
                         try:
-                            conn.simple_bind_s(authz_id, password)
+                            results = conn.simple_bind_s(authz_id, password, serverctrls=[
+                                ppolicy.PasswordPolicyControl()
+                            ])
+                            self.process_controls(request, authz_id, results[3])
                             user_login_success(authz_id)
                             if not block['connect_with_user_credentials']:
                                 try:
@@ -637,7 +683,9 @@ class LDAPBackend(object):
                                     log.exception(u'rebind failure after login bind')
                                     raise ldap.SERVER_DOWN
                             break
-                        except ldap.INVALID_CREDENTIALS:
+                        except ldap.INVALID_CREDENTIALS as e:
+                            if len(e.args) > 0 and 'ctrls' in e.args[0]:
+                                self.process_controls(request, authz_id, DecodeControlTuples(e.args[0]['ctrls']))
                             user_login_failure(authz_id)
                             pass
                     else:
diff --git a/src/authentic2/locale/fr/LC_MESSAGES/django.po b/src/authentic2/locale/fr/LC_MESSAGES/django.po
index 253d52bea..af179fe1c 100644
--- a/src/authentic2/locale/fr/LC_MESSAGES/django.po
+++ b/src/authentic2/locale/fr/LC_MESSAGES/django.po
@@ -824,6 +824,56 @@ msgstr "attribut de rôle"
 msgid "Username or email"
 msgstr "Identifiant ou courriel"
 
+#: src/authentic2/backends/ldap_backend.py:244
+msgid "The password expired"
+msgstr "Le mot de passe a expiré."
+
+#: src/authentic2/backends/ldap_backend.py:245
+msgid "The account is locked."
+msgstr "Le compte est bloqué."
+
+#: src/authentic2/backends/ldap_backend.py:246
+msgid "The password was reset and must be changed."
+msgstr "Le mot de passe a été re-initialisé et doit être changé."
+
+#: src/authentic2/backends/ldap_backend.py:247
+msgid "It is not possible to modify the password."
+msgstr "Il n'est pas possible de modifier le mot de passe."
+
+#: src/authentic2/backends/ldap_backend.py:248
+msgid "The old password must be supplied."
+msgstr "L'ancien mot de passe doit être fourni."
+
+#: src/authentic2/backends/ldap_backend.py:249
+msgid "The password does not meet the quality requirements."
+msgstr "Le mot de passe n'a pas le niveau de qualité requis."
+
+#: src/authentic2/backends/ldap_backend.py:250
+msgid "The password is too short."
+msgstr "Le mot de passe est trop court."
+
+#: src/authentic2/backends/ldap_backend.py:251
+msgid "It is too soon to change the password."
+msgstr "Il est trop tôt pour changer ce mot de passe."
+
+#: src/authentic2/backends/ldap_backend.py:252
+msgid "This password was recently used and cannot be used again."
+msgstr "Ce mot de passe a été utilisé récemment et ne peut pas être utilisé de nouveau."
+
+#: src/authentic2/backends/ldap_backend.py:258
+#, python-brace-format
+msgid "The password will expire in {ctrl.timeBeforeExpiration} seconds."
+msgstr "Le mot de passe expirera dans {ctrl.timeBeforeExpiration} secondes."
+
+#: src/authentic2/backends/ldap_backend.py:261
+#, python-brace-format
+msgid "This is the last time this password can be used."
+msgid_plural ""
+"This password can only be used {ctrl.graceAuthNsRemaining} times, including "
+"this one."
+msgstr[0] "C'est la dernière fois qu'il est possible d'utiliser ce mot de passe."
+msgstr[1] "Ce mot de passe ne pourra être utilisé que {ctrl.graceAuthNsRemaining} fois, celle ci incluse."
+
 #: src/authentic2/csv_import.py:157
 msgid "Cannot detect encoding"
 msgstr "Impossible de détecter l’encodage"
diff --git a/tests/test_ldap.py b/tests/test_ldap.py
index 8a3e0d338..8d5dc77da 100644
--- a/tests/test_ldap.py
+++ b/tests/test_ldap.py
@@ -20,6 +20,7 @@ import os
 
 import pytest
 import mock
+import time
 
 import ldap
 from ldap.dn import escape_dn_chars
@@ -77,6 +78,37 @@ def slapd():
     with create_slapd() as s:
         yield s
 
+@pytest.fixture
+def slapd_ppolicy():
+    with create_slapd() as slapd:
+        conn = slapd.get_connection_admin()
+        assert conn.protocol_version == ldap.VERSION3
+        conn.modify_s(
+            'cn=module{0},cn=config',
+            [
+                (ldap.MOD_ADD, 'olcModuleLoad', [
+                    force_bytes('ppolicy')
+                ])
+            ])
+        slapd.add_ldif(open('/etc/ldap/schema/ppolicy.ldif').read())
+        slapd.add_ldif('''
+dn: olcOverlay={0}ppolicy,olcDatabase={2}mdb,cn=config
+objectclass: olcOverlayConfig
+objectclass: olcPPolicyConfig
+olcoverlay: {0}ppolicy
+olcppolicydefault: cn=default,ou=ppolicies,o=ôrga
+olcppolicyforwardupdates: FALSE
+olcppolicyhashcleartext: TRUE
+olcppolicyuselockout: TRUE
+''')
+
+        slapd.add_ldif('''
+dn: ou=ppolicies,o=ôrga
+objectclass: organizationalUnit
+ou: ppolicies
+''')
+        yield slapd
+
 
 @pytest.fixture
 def tls_slapd():
@@ -872,6 +904,225 @@ def test_set_password(slapd, settings, db):
     assert user.pk == user2.pk
 
 
+def test_login_ppolicy_pwdMaxFailure(slapd_ppolicy, settings, db, app):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd_ppolicy.ldap_url],
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+    }]
+
+    pwdMaxFailure = 2
+    slapd_ppolicy.add_ldif(f'''
+dn: cn=default,ou=ppolicies,o=ôrga
+cn: default
+objectclass: top
+objectclass: device
+objectclass: pwdPolicy
+objectclass: pwdPolicyChecker
+pwdAttribute: userPassword
+pwdMinAge: 0
+pwdMaxAge: 0
+pwdInHistory: 0
+pwdCheckQuality: 0
+pwdMinLength: 0
+pwdExpireWarning: 0
+pwdGraceAuthnLimit: 0
+pwdLockout: TRUE
+pwdLockoutDuration: 0
+pwdMaxFailure: {pwdMaxFailure}
+pwdMaxRecordedFailure: 0
+pwdFailureCountInterval: 0
+pwdMustChange: FALSE
+pwdAllowUserChange: FALSE
+pwdSafeModify: FALSE
+''')
+
+    for _ in range(pwdMaxFailure):
+        response = app.get('/login/')
+        response.form.set('username', USERNAME)
+        response.form.set('password', 'invalid')
+        response = response.form.submit(name='login-password-submit')
+        assert 'Incorrect Username or password' in str(response.pyquery('.errornotice'))
+        assert 'account is locked' not in str(response.pyquery('.messages'))
+    response = app.get('/login/')
+    response.form.set('username', USERNAME)
+    response.form.set('password', 'invalid')
+    response = response.form.submit(name='login-password-submit')
+    assert 'account is locked' in str(response.pyquery('.messages'))
+
+
+def test_authenticate_ppolicy_pwdMaxFailure(slapd_ppolicy, settings, db, caplog):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd_ppolicy.ldap_url],
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+    }]
+
+    pwdMaxFailure = 2
+    slapd_ppolicy.add_ldif(f'''
+dn: cn=default,ou=ppolicies,o=ôrga
+cn: default
+objectclass: top
+objectclass: device
+objectclass: pwdPolicy
+objectclass: pwdPolicyChecker
+pwdAttribute: userPassword
+pwdMinAge: 0
+pwdMaxAge: 0
+pwdInHistory: 0
+pwdCheckQuality: 0
+pwdMinLength: 0
+pwdExpireWarning: 0
+pwdGraceAuthnLimit: 0
+pwdLockout: TRUE
+pwdLockoutDuration: 0
+pwdMaxFailure: {pwdMaxFailure}
+pwdMaxRecordedFailure: 0
+pwdFailureCountInterval: 0
+pwdMustChange: FALSE
+pwdAllowUserChange: FALSE
+pwdSafeModify: FALSE
+''')
+
+    for _ in range(pwdMaxFailure):
+        assert authenticate(username=USERNAME, password='incorrect') is None
+        assert "failed to login" in caplog.text
+    assert 'account is locked' not in caplog.text
+    assert authenticate(username=USERNAME, password='incorrect') is None
+    assert 'account is locked' in caplog.text
+
+
+def test_authenticate_ppolicy_pwdGraceAuthnLimit(slapd_ppolicy, settings, db, caplog):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd_ppolicy.ldap_url],
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+    }]
+
+    pwdMaxAge = 1
+    pwdGraceAuthnLimit = 2
+    slapd_ppolicy.add_ldif(f'''
+dn: cn=default,ou=ppolicies,o=ôrga
+cn: default
+objectclass: top
+objectclass: device
+objectclass: pwdPolicy
+objectclass: pwdPolicyChecker
+pwdAttribute: userPassword
+pwdMinAge: 0
+pwdMaxAge: {pwdMaxAge}
+pwdInHistory: 1
+pwdCheckQuality: 0
+pwdMinLength: 0
+pwdExpireWarning: 0
+pwdGraceAuthnLimit: {pwdGraceAuthnLimit}
+pwdLockout: TRUE
+pwdLockoutDuration: 0
+pwdMaxFailure: 0
+pwdMaxRecordedFailure: 0
+pwdFailureCountInterval: 0
+pwdMustChange: FALSE
+pwdAllowUserChange: TRUE
+pwdSafeModify: FALSE
+''')
+
+    user = authenticate(username=USERNAME, password=UPASS)
+    assert user.check_password(UPASS)
+    password = u'ogutOmyetew4'
+    user.set_password(password)
+
+    time.sleep(pwdMaxAge * 3)
+
+    assert 'used 2 time' not in caplog.text
+    assert authenticate(username=USERNAME, password=password) is not None
+    assert 'used 2 times' in caplog.text
+
+    assert 'last time' not in caplog.text
+    assert authenticate(username=USERNAME, password=password) is not None
+    assert 'last time' in caplog.text
+
+
+def test_authenticate_ppolicy_pwdExpireWarning(slapd_ppolicy, settings, db, caplog):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd_ppolicy.ldap_url],
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+    }]
+
+    pwdMaxAge = 3600
+    slapd_ppolicy.add_ldif(f'''
+dn: cn=default,ou=ppolicies,o=ôrga
+cn: default
+objectclass: top
+objectclass: device
+objectclass: pwdPolicy
+objectclass: pwdPolicyChecker
+pwdAttribute: userPassword
+pwdMinAge: 0
+pwdMaxAge: {pwdMaxAge}
+pwdInHistory: 1
+pwdCheckQuality: 0
+pwdMinLength: 0
+pwdExpireWarning: {pwdMaxAge}
+pwdGraceAuthnLimit: 0
+pwdLockout: TRUE
+pwdLockoutDuration: 0
+pwdMaxFailure: 0
+pwdMaxRecordedFailure: 0
+pwdFailureCountInterval: 0
+pwdMustChange: FALSE
+pwdAllowUserChange: TRUE
+pwdSafeModify: FALSE
+''')
+
+    user = authenticate(username=USERNAME, password=UPASS)
+    assert user.check_password(UPASS)
+    password = u'ogutOmyetew4'
+    user.set_password(password)
+
+    time.sleep(2)
+
+    assert 'password will expire' not in caplog.text
+    assert authenticate(username=USERNAME, password=password) is not None
+    assert 'password will expire' in caplog.text
+
+
+def test_authenticate_ppolicy_pwdAllowUserChange(slapd_ppolicy, settings, db, caplog):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd_ppolicy.ldap_url],
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+    }]
+
+    slapd_ppolicy.add_ldif(f'''
+dn: cn=default,ou=ppolicies,o=ôrga
+cn: default
+objectclass: top
+objectclass: device
+objectclass: pwdPolicy
+pwdAttribute: userPassword
+pwdMinAge: 0
+pwdMaxAge: 0
+pwdInHistory: 0
+pwdCheckQuality: 0
+pwdMinLength: 0
+pwdExpireWarning: 0
+pwdGraceAuthnLimit: 0
+pwdLockout: TRUE
+pwdLockoutDuration: 0
+pwdMaxFailure: 0
+pwdMaxRecordedFailure: 0
+pwdFailureCountInterval: 0
+pwdMustChange: FALSE
+pwdAllowUserChange: FALSE
+pwdSafeModify: FALSE
+''')
+
+    user = authenticate(username=USERNAME, password=UPASS)
+    with pytest.raises(ldap.STRONG_AUTH_REQUIRED):
+        user.set_password(u'ogutOmyetew4')
+
+
 def test_ou_selector(slapd, settings, app, ou1):
     settings.LDAP_AUTH_SETTINGS = [{
         'url': [slapd.ldap_url],
-- 
2.25.1

