From f74fa0372a2e829bf1c91901019140ec4385ea62 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Dachary?= <ldachary@easter-eggs.com>
Date: Thu, 18 Feb 2021 18:12:22 +0100
Subject: [PATCH] ldap: redirect /password/change if it is about to expire
 (#51268)

Fixes: #51268

License: MIT
---
 src/authentic2/backends/ldap_backend.py |  4 +-
 src/authentic2/constants.py             |  1 +
 src/authentic2/middleware.py            | 13 ++++-
 src/authentic2/settings.py              |  1 +
 tests/test_ldap.py                      | 73 ++++++++++++++++++++++---
 5 files changed, 82 insertions(+), 10 deletions(-)

diff --git a/src/authentic2/backends/ldap_backend.py b/src/authentic2/backends/ldap_backend.py
index 28f212e31..43c707ac7 100644
--- a/src/authentic2/backends/ldap_backend.py
+++ b/src/authentic2/backends/ldap_backend.py
@@ -56,7 +56,7 @@ from authentic2.a2_rbac.models import Role
 
 from authentic2.compat_lasso import lasso
 
-from authentic2 import crypto, app_settings
+from authentic2 import crypto, app_settings, constants
 from authentic2.models import UserExternalId
 from authentic2.middleware import StoreRequestMiddleware
 from authentic2.user_login_failure import user_login_failure, user_login_success
@@ -585,6 +585,8 @@ class LDAPBackend(object):
                 message = ' '.join(password_policy_control_messages(c))
                 if request is not None:
                     messages.add_message(request, messages.WARNING, message)
+                    if c.graceAuthNsRemaining or c.timeBeforeExpiration:
+                        request.session[constants.PASSWORD_EXPIRATION_SESSION_KEY] = True
             else:
                 message = str(vars(c))
             log.info('ldap: bind error with authz_id "%s" -> "%s"', authz_id, message)
diff --git a/src/authentic2/constants.py b/src/authentic2/constants.py
index f32c226cb..85c098619 100644
--- a/src/authentic2/constants.py
+++ b/src/authentic2/constants.py
@@ -19,6 +19,7 @@ NONCE_FIELD_NAME = 'nonce'
 CANCEL_FIELD_NAME = 'cancel'
 AUTHENTICATION_EVENTS_SESSION_KEY = 'authentication-events'
 SWITCH_USER_SESSION_KEY = '_switch_user'
+PASSWORD_EXPIRATION_SESSION_KEY = '_password_expiration'
 LAST_LOGIN_SESSION_KEY = '_last_login'
 SERVICE_FIELD_NAME = 'service'
 NEXT_URL_SIGNATURE = 'next-signature'
diff --git a/src/authentic2/middleware.py b/src/authentic2/middleware.py
index e5aeb4f02..afb8b0edb 100644
--- a/src/authentic2/middleware.py
+++ b/src/authentic2/middleware.py
@@ -28,7 +28,7 @@ from django.utils.translation import ugettext as _
 from django.utils.six.moves.urllib import parse as urlparse
 from django import http
 
-from . import app_settings, utils, plugins
+from . import app_settings, utils, plugins, constants
 from .utils.service import get_service_from_request, get_service_from_session
 
 
@@ -167,6 +167,17 @@ class DisplayMessageBeforeRedirectMiddleware(MiddlewareMixin):
         return utils.redirect(request, 'continue', resolve=True, params={'next': url})
 
 
+class ChangePasswordAboutToExpireMiddleware(MiddlewareMixin):
+    '''If the password is about to expire, redirect the user to the password
+       modification page.
+    '''
+    def process_response(self, request, response):
+        if request.session.get(constants.PASSWORD_EXPIRATION_SESSION_KEY) is None:
+            return response
+        del request.session[constants.PASSWORD_EXPIRATION_SESSION_KEY]
+        return utils.redirect(request, 'password_change', resolve=True)
+
+
 class ServiceAccessControlMiddleware(MiddlewareMixin):
     def process_exception(self, request, exception):
         if not isinstance(exception, (utils.ServiceAccessDenied,)):
diff --git a/src/authentic2/settings.py b/src/authentic2/settings.py
index cce65c88a..c05bbafd9 100644
--- a/src/authentic2/settings.py
+++ b/src/authentic2/settings.py
@@ -92,6 +92,7 @@ MIDDLEWARE = (
     'authentic2.middleware.StoreRequestMiddleware',
     'authentic2.middleware.RequestIdMiddleware',
     'authentic2.middleware.ServiceAccessControlMiddleware',
+    'authentic2.middleware.ChangePasswordAboutToExpireMiddleware',
     'authentic2.middleware.CookieTestMiddleware',
     'django.middleware.common.CommonMiddleware',
     'django.middleware.http.ConditionalGetMiddleware',
diff --git a/tests/test_ldap.py b/tests/test_ldap.py
index c02898a30..7ed6fe841 100644
--- a/tests/test_ldap.py
+++ b/tests/test_ldap.py
@@ -742,14 +742,7 @@ def test_no_connect_with_user_credentials(slapd_strict_acl, db, settings, app):
     assert force_bytes('Étienne Michu') in response.body
 
 
-def test_reset_password_ldap_user(slapd, settings, app, db):
-    settings.LDAP_AUTH_SETTINGS = [{
-        'url': [slapd.ldap_url],
-        'binddn': force_text(slapd.root_bind_dn),
-        'bindpw': force_text(slapd.root_bind_password),
-        'basedn': u'o=ôrga',
-        'use_tls': False,
-    }]
+def reset_password_ldap_user(settings, app):
     assert User.objects.count() == 0
     # first login
     response = app.get('/login/')
@@ -781,6 +774,24 @@ def test_reset_password_ldap_user(slapd, settings, app, db):
     response.form['new_password1'] = new_password
     response.form['new_password2'] = new_password
     response = response.form.submit(status=302).maybe_follow()
+    # logout
+    response = response.click('Logout')
+    if response.status_code == 200:  # Django 1.7, same_origin is bugged; testserver != localhost:80
+        response = response.form.submit().maybe_follow()
+    else:
+        response = response.maybe_follow()
+    return new_password
+
+
+def test_reset_password_ldap_user(slapd, settings, app, db):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd.ldap_url],
+        'binddn': force_text(slapd.root_bind_dn),
+        'bindpw': force_text(slapd.root_bind_password),
+        'basedn': u'o=ôrga',
+        'use_tls': False,
+    }]
+    new_password = reset_password_ldap_user(settings, app)
     # verify password has changed
     slapd.get_connection().bind_s(DN, new_password)
     with pytest.raises(ldap.INVALID_CREDENTIALS):
@@ -1130,6 +1141,52 @@ pwdSafeModify: FALSE
     assert 'password will expire' in caplog.text
 
 
+def test_login_ppolicy_pwdExpireWarning(slapd_ppolicy, settings, app, db, caplog):
+    settings.LDAP_AUTH_SETTINGS = [{
+        'url': [slapd_ppolicy.ldap_url],
+        'binddn': force_text(slapd_ppolicy.root_bind_dn),
+        'bindpw': force_text(slapd_ppolicy.root_bind_password),
+        '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))
+
+    password = reset_password_ldap_user(settings, app)
+
+    time.sleep(2)
+
+    response = app.get('/login/')
+    response.form['username'] = USERNAME
+    response.form['password'] = password
+    response = response.form.submit('login-password-submit')
+    assert response['Location'].endswith('/password/change/')
+
+
 def test_authenticate_ppolicy_pwdAllowUserChange(slapd_ppolicy, settings, db, caplog):
     settings.LDAP_AUTH_SETTINGS = [{
         'url': [slapd_ppolicy.ldap_url],
-- 
2.25.1

