0001-ldap-optionally-collects-messages-from-ppolicy.patch

Loïc Dachary, 12 février 2021 00:01

Voir les différences: en ligne côte à côte

Subject: [PATCH] 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.

[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 |  71 +++++-
 tests/test_ldap.py                      | 275 ++++++++++++++++++++++++
 2 files changed, 340 insertions(+), 6 deletions(-)

         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:
-...
     import base64
     import os
     import json
     import time
     # code originaly copied from by now merely inspired by
     # http://www.amherst.k12.oh.us/django-ldap.html
-...
     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
-...
         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}').format(error=error)))
             return messages
         if ctrl.timeBeforeExpiration:
             timeBeforeExpiration = time.asctime(time.localtime(time.time() + ctrl.timeBeforeExpiration))
             messages.append(_('The password will expire at {timeBeforeExpiration}.').format(
                 timeBeforeExpiration=timeBeforeExpiration))
         if ctrl.graceAuthNsRemaining:
             messages.append(ngettext(
                 'This password expired: this is the last time it can be used.',
                 'This password expired and can only be used {graceAuthNsRemaining} times, including this one.',
                 ctrl.graceAuthNsRemaining).format(graceAuthNsRemaining=ctrl.graceAuthNsRemaining))
         return messages
     class LDAPUser(User):
         SESSION_LDAP_DATA_KEY = 'ldap-data'
         _changed = False
-...
                 if not conn:
                     log.warning('ldap: set_password failed, could not get a connection')
                     return
                 self.ldap_backend.modify_password(conn, self.block, self.dn, _current_password, new_password)
                 try:
                     self.ldap_backend.modify_password(conn, self.block, self.dn, _current_password, new_password)
                 except ldap.STRONG_AUTH_REQUIRED:
                     log.warning('ldap: set_password failed, STRONG_AUTH_REQUIRED')
                     return
                 self._current_password = new_password
             self.keep_password_in_session(new_password)
             if self.block['keep_password']:
-...
             'can_reset_password': False,
             # mapping from LDAP attributes to User attributes
             'user_attributes': [],
             # https://www.python-ldap.org/en/python-ldap-3.3.0/reference/ldap.html#ldap-controls
             'use_controls': True,
+        }
         _REQUIRED = ('url', 'basedn')
         _TO_ITERABLE = ('url', 'groupsu', 'groupstaff', 'groupactive')
-...
             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))
                     if request is not None:
                         messages.add_message(request, messages.WARNING, message)
                 else:
                     message = str(vars(c))
                 log.info('ldap: bind error with authz_id "%s" -> "%s"', authz_id, message)
         def authenticate(self, request=None, username=None, password=None, realm=None, ou=None):
             if username is None or password is None:
                 return None
-...
                     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 = []
-...
                             if failed:
                                 continue
                             try:
                                 conn.simple_bind_s(authz_id, password)
                                 if block.get('use_controls'):
                                     serverctrls = [ppolicy.PasswordPolicyControl()]
                                 else:
                                     serverctrls = []
                                 results = conn.simple_bind_s(authz_id, password, serverctrls=serverctrls)
                                 self.process_controls(request, authz_id, results[3])
                                 user_login_success(authz_id)
                                 if not block['connect_with_user_credentials']:
                                     try:
-...
                                         log.exception(u'rebind failure after login bind')
                                         raise ldap.SERVER_DOWN
                                 break
                             except ldap.INVALID_CREDENTIALS:
                             except ldap.INVALID_CREDENTIALS as e:
                                 if block.get('use_controls') and 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:

     import pytest
     import mock
     import time
     import ldap
     from ldap.dn import escape_dn_chars
-...
         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')
                     ])
                 ])
             with open('/etc/ldap/schema/ppolicy.ldif') as fd:
                 slapd.add_ldif(fd.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():
-...
         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('''
     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
     '''.format(pwdMaxFailure=pwdMaxFailure))
         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 ppolicy_authenticate_exactly_pwdMaxFailure(slapd_ppolicy, caplog):
         pwdMaxFailure = 2
         slapd_ppolicy.add_ldif('''
     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
     '''.format(pwdMaxFailure=pwdMaxFailure))
         for _ in range(pwdMaxFailure):
             assert authenticate(username=USERNAME, password='incorrect') is None
             assert "failed to login" in caplog.text
     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,
         }]
         ppolicy_authenticate_exactly_pwdMaxFailure(slapd_ppolicy, caplog)
         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_do_not_use_controls(slapd_ppolicy, settings, db, caplog):
         """
         Same as test_authenticate_ppolicy_pwdMaxFailure but with use_controls
         deactivated and therefore not logging when an account is locked.
         """
         settings.LDAP_AUTH_SETTINGS = [{
             'url': [slapd_ppolicy.ldap_url],
             'basedn': u'o=ôrga',
             'use_tls': False,
             'use_controls': False,
         }]
         ppolicy_authenticate_exactly_pwdMaxFailure(slapd_ppolicy, caplog)
         assert 'account is locked' not in caplog.text
         assert authenticate(username=USERNAME, password='incorrect') is None
         # this following line is the difference with test_authenticate_ppolicy_pwdMaxFailure
         assert 'account is locked' not 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('''
     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
     '''.format(pwdMaxAge=pwdMaxAge, pwdGraceAuthnLimit=pwdGraceAuthnLimit))
         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('''
     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
     '''.format(pwdMaxAge=pwdMaxAge))
         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('''
     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)
         assert user.set_password(u'ogutOmyetew4') is None
         assert 'STRONG_AUTH_REQUIRED' in caplog.text
     def test_ou_selector(slapd, settings, app, ou1):
         settings.LDAP_AUTH_SETTINGS = [{
             'url': [slapd.ldap_url],
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-ldap-optionally-collects-messages-from-ppolicy.patch