0001-ldap-add-useful-output-to-sync-ldap-users-command-54.patch

Valentin Deniaud, 31 août 2021 11:39

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

Subject: [PATCH] ldap: add useful output to sync-ldap-users command (#54078)

 src/authentic2/backends/ldap_backend.py       | 26 ++++++++++-
 .../management/commands/sync-ldap-users.py    | 17 +++++--
 tests/test_ldap.py                            | 45 ++++++++++++++++---
 3 files changed, 75 insertions(+), 13 deletions(-)

         @classmethod
         def get_users(cls):
             for block in cls.get_config():
             blocks = cls.get_config()
             if not blocks:
                 log.info('No LDAP server configured.')
                 return
             for block in blocks:
                 log.info('Synchronising users from realm "%s"', block['realm'])
                 conn = cls.get_connection(block)
                 if conn is None:
                     log.warning('unable to synchronize with LDAP servers %s', force_text(block['url']))
-...
                 user_basedn = force_text(block.get('user_basedn') or block['basedn'])
                 user_filter = cls.get_sync_ldap_user_filter(block)
                 attribute_names = cls.get_ldap_attributes_names(block)
                 log.info('Searching for %s', user_filter)
                 results = cls.paged_search(
                     conn, user_basedn, ldap.SCOPE_SUBTREE, user_filter, attrlist=attribute_names
+                )
                 backend = cls()
                 count = 0
                 for dn, attrs in results:
                     yield backend._return_user(dn, None, conn, block, attrs)
                     count += 1
                     user = backend._return_user(dn, None, conn, block, attrs)
                     if getattr(user, '_changed', False):
                         log.debug(
                             'Updated user %s (username %s, full name %s)',
                             user.uuid,
                             user.get_username(),
                             user.get_full_name(),
+                        )
                     yield user
                 log.info('Server returned %s users.', count)
         @classmethod
         def deactivate_orphaned_users(cls):
-...
                 user_credentials = block['connect_with_user_credentials'] and credentials
                 success, error = cls.bind(block, conn, credentials=user_credentials)
                 if success:
                     log.info('Connected to server %s', url)
                     yield conn
                 else:
                     if block['replicas']:
-...
                     who, password = credentials[0], credentials[1]
                     password = force_text(password)
                     conn.simple_bind_s(who, password)
                     log.info('Binding with %s account: success', who)
                 elif block['bindsasl']:
                     sasl_mech, who, sasl_params = map_text(block['bindsasl'])
                     handler_class = getattr(ldap.sasl, sasl_mech)
                     auth = handler_class(*sasl_params)
                     conn.sasl_interactive_bind_s(who, auth)
                     log.info('Binding with %s account: success', who)
                 elif block['binddn'] and block['bindpw']:
                     who = force_text(block['binddn'])
                     conn.simple_bind_s(who, force_text(block['bindpw']))
                     log.info('Binding with %s binddn: success', who)
                 else:
                     who = 'anonymous'
                     conn.simple_bind_s()
                     log.info('Binding anonymously: success')
                 return True, None
             except ldap.INVALID_CREDENTIALS:
                 return False, 'invalid credentials'

     except ImportError:
         ldap = None
     import logging
     from django.core.management.base import BaseCommand
     from authentic2.backends.ldap_backend import LDAPBackend
     logger = logging.getLogger(__name__)
     class Command(BaseCommand):
         def handle(self, *args, **kwargs):
             verbosity = int(kwargs['verbosity'])
             if verbosity > 1:
                 print('Updated users :')
             root_logger = logging.getLogger()
             if verbosity == 0:
                 root_logger.setLevel(logging.WARNING)
             elif verbosity == 1:
                 root_logger.setLevel(logging.INFO)
             elif verbosity == 2:
                 root_logger.setLevel(logging.DEBUG)
             for user in LDAPBackend.get_users():
                 if getattr(user, '_changed', False) and verbosity > 1:
                     print(' -', user.uuid, user.get_username(), user.get_full_name())
                 continue

     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import json
     import logging
     import os
     import time
     import urllib.parse
-...
         assert '_auth_user_id' in app.session
     def test_sync_ldap_users(slapd, settings, app, db, capsys):
     def test_sync_ldap_users(slapd, settings, app, db, caplog):
         caplog.set_level(logging.DEBUG)  # force pytest to reset log level after test
         management.call_command('sync-ldap-users')
         assert len(caplog.records) == 1
         assert caplog.records[0].message == 'No LDAP server configured.'
         settings.LDAP_AUTH_SETTINGS = [
+            {
                 'url': [slapd.ldap_url],
-...
+        )
         assert User.objects.count() == 0
         capsys.readouterr()
         management.call_command('sync-ldap-users', verbosity=2)
         assert len(capsys.readouterr().out.splitlines()) == 7
         caplog.clear()
         management.call_command('sync-ldap-users')
         assert caplog.messages == [
             'Synchronising users from realm "ldap"',
             'Binding anonymously: success',
             'Connected to server %s' % slapd.ldap_url,
             'Searching for (|(mail=*)(uid=*))',
             'Server returned 6 users.',
+        ]
         assert User.objects.count() == 6
         assert all(user.first_name == 'Étienne' for user in User.objects.all())
         assert all(user.attributes.first_name == 'Étienne' for user in User.objects.all())
-...
                 for user in User.objects.all()
+            ]
+        )
         capsys.readouterr()
         User.objects.update(first_name='John')
         caplog.clear()
         management.call_command('sync-ldap-users', verbosity=2)
         assert len(caplog.records) == 42
         assert (
             caplog.records[10].message
             == 'Updated user %s (username etienne.michu@ldap, full name Étienne Michu)'
             % User.objects.first().uuid
+        )
         caplog.clear()
         management.call_command('sync-ldap-users', verbosity=2)
         assert len(capsys.readouterr().out.splitlines()) == 1
         assert len(caplog.records) == 36  # users have not been updated
         caplog.clear()
         management.call_command('sync-ldap-users', verbosity=0)
         assert len(caplog.records) == 0
     def test_alert_on_wrong_user_filter(slapd, settings, client, db, caplog):
-...
+        }
     def test_switch_user_ldap_user(slapd, settings, app, db):
     def test_switch_user_ldap_user(slapd, settings, app, db, caplog):
         caplog.set_level(logging.DEBUG)  # force pytest to reset log level after test
         settings.LDAP_AUTH_SETTINGS = [
+            {
                 'url': [slapd.ldap_url],
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-ldap-add-useful-output-to-sync-ldap-users-command-54.patch