From 956eb9fe22bd5a3d465bf9c48bbf4d1b9d7818c9 Mon Sep 17 00:00:00 2001
From: Benjamin Renard <brenard@easter-eggs.com>
Date: Thu, 22 Sep 2022 18:13:46 +0200
Subject: [PATCH 5/5] ldap: handle ppolicy control changing/reseting password

---
 src/authentic2/backends/ldap_backend.py | 51 ++++++++++++++++++++++---
 tests/test_ldap.py                      |  6 +--
 2 files changed, 48 insertions(+), 9 deletions(-)

diff --git a/src/authentic2/backends/ldap_backend.py b/src/authentic2/backends/ldap_backend.py
index ea7e6d21..76086bf1 100644
--- a/src/authentic2/backends/ldap_backend.py
+++ b/src/authentic2/backends/ldap_backend.py
@@ -592,7 +592,7 @@ class LDAPBackend:
         return blocks
 
     @classmethod
-    def process_controls(cls, request, block, conn, authz_id, ctrls):
+    def process_bind_controls(cls, request, block, conn, authz_id, ctrls):
         attributes = cls.get_ppolicy_attributes(block, conn, authz_id)
         for c in ctrls:
             if c.controlType == ppolicy.PasswordPolicyControl.controlType:
@@ -605,6 +605,20 @@ class LDAPBackend:
                 message = str(vars(c))
             log.info('ldap: bind error with authz_id "%s" -> "%s"', authz_id, message)
 
+    @classmethod
+    def process_modify_password_controls(cls, block, conn, authz_id, ctrls):
+        attributes = cls.get_ppolicy_attributes(block, conn, authz_id)
+        errors = []
+        for c in ctrls:
+            if c.controlType == ppolicy.PasswordPolicyControl.controlType:
+                message = ' '.join(password_policy_control_messages(c, attributes))
+            else:
+                message = str(vars(c))
+            log.info('ldap: fail to modify password of "%s" -> "%s"', authz_id, message)
+            errors.append(message)
+
+        raise PasswordChangeError(' '.join(errors))
+
     @classmethod
     def check_group_to_role_mappings(cls, block):
         group_to_role_mapping = block.get('group_to_role_mapping')
@@ -730,7 +744,7 @@ class LDAPBackend:
                             else:
                                 serverctrls = []
                             results = conn.simple_bind_s(authz_id, password, serverctrls=serverctrls)
-                            self.process_controls(request, block, conn, authz_id, results[3])
+                            self.process_bind_controls(request, block, conn, authz_id, results[3])
                             user_login_success(authz_id)
                             if not block['connect_with_user_credentials']:
                                 try:
@@ -741,7 +755,7 @@ class LDAPBackend:
                             break
                         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(
+                                self.process_bind_controls(
                                     request, block, conn, authz_id, DecodeControlTuples(e.args[0]['ctrls'])
                                 )
                             success, error = self.bind(block, conn)
@@ -1770,7 +1784,19 @@ class LDAPBackend:
     def modify_password(cls, conn, block, dn, old_password, new_password):
         '''Change user password with adaptation for Active Directory'''
         if old_password is not None and (block['use_password_modify'] and not block['active_directory']):
-            conn.passwd_s(dn, old_password, new_password)
+            try:
+                if block.get('use_controls'):
+                    serverctrls = [ppolicy.PasswordPolicyControl()]
+                else:
+                    serverctrls = []
+                results = conn.passwd_s(dn, old_password, new_password, serverctrls=serverctrls)
+                cls.process_modify_password_controls(block, conn, dn, results[3])
+            except ldap.LDAPError as e:
+                if block.get('use_controls') and len(e.args) > 0 and 'ctrls' in e.args[0]:
+                    cls.process_modify_password_controls(
+                        block, conn, dn, DecodeControlTuples(e.args[0]['ctrls'])
+                    )
+                raise
         else:
             modlist = []
             if block['active_directory']:
@@ -1785,8 +1811,21 @@ class LDAPBackend:
                     modlist = [(ldap.MOD_REPLACE, key, [value])]
             else:
                 key = 'userPassword'
-                modlist = [(ldap.MOD_REPLACE, key, [new_password])]
-            conn.modify_s(dn, modlist)
+                modlist = [(ldap.MOD_REPLACE, key, [new_password.encode('utf-8')])]
+            try:
+                if block.get('use_controls'):
+                    serverctrls = [ppolicy.PasswordPolicyControl()]
+                else:
+                    serverctrls = []
+                results = conn.modify_ext_s(dn, modlist, serverctrls=serverctrls)
+                cls.process_modify_password_controls(block, conn, dn, results[3])
+            except ldap.LDAPError as e:
+                if block.get('use_controls') and len(e.args) > 0 and 'ctrls' in e.args[0]:
+                    cls.process_modify_password_controls(
+                        block, conn, dn, DecodeControlTuples(e.args[0]['ctrls'])
+                    )
+                raise
+
         log.debug('modified password for dn %r', dn)
 
     @classmethod
diff --git a/tests/test_ldap.py b/tests/test_ldap.py
index d880cedc..99bd9754 100644
--- a/tests/test_ldap.py
+++ b/tests/test_ldap.py
@@ -2191,13 +2191,13 @@ def test_user_journal_login_failure(slapd, settings, client, db, monkeypatch, ex
         '/login/', {'login-password-submit': '1', 'username': USERNAME, 'password': PASS}, follow=True
     )
 
-    def patched_process_controls(cls, request, block, conn, authz_id, ctrls):
+    def patched_process_bind_controls(cls, request, block, conn, authz_id, ctrls):
         raise exception[0]('oops')
 
     monkeypatch.setattr(
         ldap_backend.LDAPBackend,
-        'process_controls',
-        patched_process_controls,
+        'process_bind_controls',
+        patched_process_bind_controls,
     )
     client.post(
         '/login/', {'login-password-submit': '1', 'username': USERNAME, 'password': PASS}, follow=True
-- 
2.30.2

