From 67fbab4f6cce7a8dd4b4f3dfab8db21fc6c6f4eb 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] ldap: log controls on authenticate and enable ppolicy

---
 src/authentic2/backends/ldap_backend.py |  39 +++++-
 tests/test_ldap.py                      | 164 ++++++++++++++++++++++++
 2 files changed, 200 insertions(+), 3 deletions(-)

diff --git a/src/authentic2/backends/ldap_backend.py b/src/authentic2/backends/ldap_backend.py
index 5ffd94903..7855bfd42 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:
@@ -46,6 +48,7 @@ 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_lazy as _, ngettext
 
 from authentic2.a2_rbac.models import Role
 
@@ -232,6 +235,23 @@ def map_text(d):
     raise NotImplementedError
 
 
+def password_policy_control_messages(ctrl):
+    messages = []
+
+    if ctrl.error:
+        if ppolicy.PasswordPolicyError.namedValues[ctrl.error] == 'accountLocked':
+            messages.append(_('The account is locked.'))
+        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 +557,14 @@ class LDAPBackend(object):
         log.debug('got config %r', blocks)
         return blocks
 
+    @staticmethod
+    def process_controls(authz_id, ctrls):
+        for c in ctrls:
+            if c.controlType == ppolicy.PasswordPolicyControl.controlType:
+                log.info('%s: %s', authz_id, password_policy_control_messages(c))
+            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
@@ -628,7 +656,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(authz_id, results[3])
                             user_login_success(authz_id)
                             if not block['connect_with_user_credentials']:
                                 try:
@@ -637,7 +668,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(authz_id, DecodeControlTuples(e.args[0]['ctrls']))
                             user_login_failure(authz_id)
                             pass
                     else:
diff --git a/tests/test_ldap.py b/tests/test_ldap.py
index 8a3e0d338..1aa26a055 100644
--- a/tests/test_ldap.py
+++ b/tests/test_ldap.py
@@ -20,9 +20,11 @@ import os
 
 import pytest
 import mock
+import time
 
 import ldap
 from ldap.dn import escape_dn_chars
+from ldap.controls.ppolicy import PasswordPolicyControl
 
 from ldaptools.slapd import Slapd, has_slapd
 from django.contrib.auth import get_user_model
@@ -77,6 +79,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 +905,137 @@ def test_set_password(slapd, settings, db):
     assert user.pk == user2.pk
 
 
+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
+
+    assert 'account is locked' not in caplog.text
+    assert authenticate(username=USERNAME, password=password) is None
+    assert 'account is locked' 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

