0001-ldap-use-guid-attributes-as-global-external-id-63646.patch

Benjamin Dauvergne, 07 avril 2022 10:20

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

Subject: [PATCH 1/2] ldap: use guid attributes as global external id (#63646)

 src/authentic2/backends/ldap_backend.py       | 167 ++++++++++++++----
 .../migrations/0038_add_external_guid.py      |  36 ++++
 src/authentic2/models.py                      |  18 +-
 tests/test_ldap.py                            | 108 ++++++++---
 4 files changed, 267 insertions(+), 62 deletions(-)
 create mode 100644 src/authentic2/migrations/0038_add_external_guid.py

     import ssl
     import time
     import urllib.parse
     import uuid
     import ldap
     import ldap.modlist
-...
     from django.contrib.auth.models import Group
     from django.core.cache import cache
     from django.core.exceptions import ImproperlyConfigured
     from django.db.models import Q
     from django.db.transaction import atomic
     from django.utils.encoding import force_bytes, force_text
     from django.utils.translation import ngettext
-...
         '/var/lib/ca-certificates/ca-bundle.pem',  # OpenSuse
+    ]
     USUAL_GUID_ATTRIBUTES = ['entryuuid', 'objectguid', 'nsuniqueid']
     class UserCreationError(Exception):
         pass
-...
         @to_list
         def _convert_results_to_unicode(self, result_list):
             for dn, attrs in result_list:
                 if dn is not None:
                     # tuple is a real entry with a DN not a search reference
                     attrs = {attribute: filter_non_unicode_values(attrs[attribute]) for attribute in attrs}
                     yield dn, attrs
                 if dn is None:
                     continue
                 new_attrs = {}
                 for attribute in attrs:
                     values = attrs[attribute]
                     # specialize for GUID attributes
                     if attribute.lower() in USUAL_GUID_ATTRIBUTES and len(values[0]) == 16:
                         try:
                             values = [str(uuid.UUID(bytes=values[0])).encode()]
                         except ValueError:
                             values = []
                     values = filter_non_unicode_values(values)
                     if not values:
                         continue
                     new_attrs[attribute] = values
                 yield dn, new_attrs
         def modify_s(self, dn, modlist):
             new_modlist = []
-...
             # generated username are unique
             'update_username': False,
             # lookup existing user with an external id build with attributes
             'lookups': ('external_id', 'username', 'email'),
             'lookups': ('guid', 'external_id', 'username', 'email'),
             'external_id_tuples': (
                 ('uid',),
                 ('dn:noquote',),
-...
                     try:
                         return self._return_user(authz_id, password, conn, block)
                     except UserCreationError as e:
                         messages.error(request, str(e))
                         if request:
                             messages.error(request, str(e))
                         return None
                 except ldap.CONNECT_ERROR:
                     log.error(
-...
                 for key in at_mapping:
                     if at_mapping[key] != 'dn':
                         attributes.add(at_mapping[key])
             # add usual GUID attributes
             attributes.update(USUAL_GUID_ATTRIBUTES)
             return list({attribute.lower() for attribute in attributes})
         @classmethod
-...
                 return None
         def _lookup_by_external_id(self, block, attributes):
             realm = block['realm']
             for eid_tuple in map_text(block['external_id_tuples']):
                 external_id = self.build_external_id(eid_tuple, attributes)
                 if not external_id:
-...
                     LDAPUser.objects.prefetch_related('groups')
                     .filter(
                         userexternalid__external_id__iexact=external_id,
                         userexternalid__source=force_text(block['realm']),
                         userexternalid__source=realm,
+                    )
                     .order_by('-last_login')
+                )
-...
                             len(users),
+                        )
                         for other in users[1:]:
                             for r in other.roles.all():
                                 user.roles.add(r)
                             user.roles.add(*other.roles.all())
                             other.delete()
                     return user
             return None
         def _lookup_by_external_guid(self, block, attribute, guid):
             if not guid:
                 return None
             log.debug('ldap: lookup by external_guid %s=%s', attribute, guid)
             try:
                 return LDAPUser.objects.get(
                     userexternalid__source=block['realm'], userexternalid__external_guid=guid
+                )
             except LDAPUser.DoesNotExist:
                 return None
         # entryuuid is encoded as the string representation of the UUID, as an octet string
         @classmethod
         def _decode_entryuuid_guid(cls, value):
             try:
                 return uuid.UUID(value)
             except (ValueError, UnicodeDecodeError):
                 return None
         # objectid is encoded as the byte representation of the UUID
         _decode_objectguid_guid = _decode_entryuuid_guid
         # nsuniqueid is encoded as the byte representation of the UUID
         _decode_nsuniqueid_guid = _decode_objectguid_guid
         def _lookup_by_guid(self, block, attributes):
             attribute, guid = self._get_guid(attributes)
             return self._lookup_by_external_guid(block=block, attribute=attribute, guid=guid)
         @classmethod
         def _get_guid(cls, attributes):
             for attribute in USUAL_GUID_ATTRIBUTES:
                 if attribute not in attributes:
                     continue
                 value = attributes[attribute][0]
                 guid = getattr(cls, f'_decode_{attribute}_guid')(value)
                 if guid:
                     return attribute, guid
             return None, None
         def _lookup_existing_user(self, username, block, attributes):
             user = None
             ou = self._get_target_ou(block)
-...
                     user = self._lookup_by_external_id(block=block, attributes=attributes)
                 elif lookup_type == 'email' and attributes:
                     user = self._lookup_by_email(ou=ou, block=block, attributes=attributes)
                 elif lookup_type == 'guid' and attributes:
                     user = self._lookup_by_guid(block=block, attributes=attributes)
                 if user:
                     return user
         def update_user_identifiers(self, user, username, block, attributes):
             realm = block['realm']
             # if username has changed and we propagate those changes, update it
             if block['update_username']:
                 if user.username != username:
-...
                     log_msg = 'updating username from %r to %r'
                     log.debug(log_msg, old_username, user.username)
             # if external_id lookup is used, update it
             userexternalid = None
             use_guid = False
             use_external_id = False
             guid = None
             external_id = None
             if 'guid' in block['lookups']:
                 use_guid = True
                 _attribute, guid = self._get_guid(attributes)
             if guid and user.pk:
                 if guid:
                     try:
                         userexternalid = UserExternalId.objects.get(user=user, external_guid=guid, source=realm)
                     except UserExternalId.DoesNotExist:
                         pass
             if (
                 'external_id' in block['lookups']
                 and block.get('external_id_tuples')
                 and block['external_id_tuples'][0]
             ):
                 if not user.pk:
                     user.save()
                     user._changed = False
                 use_external_id = True
                 external_id = self.build_external_id(map_text(block['external_id_tuples'][0]), attributes)
                 if external_id:
                     new, dummy = UserExternalId.objects.get_or_create(
                         user=user, external_id=external_id, source=force_text(block['realm'])
             if external_id and user.pk:
                 try:
                     userexternalid = UserExternalId.objects.get(
                         user=user, external_id__iexact=external_id, source=realm
+                    )
                     if block['clean_external_id_on_update']:
                         UserExternalId.objects.exclude(id=new.id).filter(
                             user=user, source=force_text(block['realm'])
                         ).delete()
                 elif user._created:
                 except UserExternalId.DoesNotExist:
                     pass
             if userexternalid:
                 changed = False
                 if userexternalid.external_guid != guid:
                     userexternalid.external_guid = guid
                     changed = True
                 if userexternalid.external_id != external_id:
                     userexternalid.external_id = external_id
                     changed = True
                 if changed:
                     userexternalid.save()
             elif use_guid or use_external_id:
                 if not guid and not external_id:
                     log.error(
                         'ldap: unable to build an external_id (%r) with attributes: %r',
                         block['external_id_tuples'][0],
                         'ldap: unable to build an user_external_id (%r) with attributes: %r',
                         block['external_id_tuples'],
                         attributes,
+                    )
                     raise UserCreationError(_('LDAP configuration is broken, please contact your administrator'))
                 else:
                     log.error(
                         'ldap: unable to update an external_id (%r) with attributes: %r',
                         block['external_id_tuples'][0],
                         attributes,
+                    )
                 if not user.pk:
                     user._changed = False
                     user.save()
                 UserExternalId.objects.create(
                     user=user, source=realm, external_id=external_id, external_guid=guid
+                )
         def _record_failure_for_user(self, request, reason, user_id, block, conn, attributes=None):
             user = None
-...
                         'external_id', flat=True
+                    )
+                )
                 guids = set(
                     UserExternalId.objects.filter(user__is_active=True, source=block['realm']).values_list(
                         'external_guid', flat=True
+                    )
+                )
                 basedn = force_text(block.get('user_basedn') or block['basedn'])
                 attribute_names = [
                     a[0] for a in cls.attribute_name_from_external_id_tuple(block['external_id_tuples'])
+                ]
                 attribute_names = list(
                     {a[0] for a in cls.attribute_name_from_external_id_tuple(block['external_id_tuples'])}
                     | set(USUAL_GUID_ATTRIBUTES)
+                )
                 user_filter = cls.get_sync_ldap_user_filter(block)
                 results = cls.paged_search(
                     conn, basedn, ldap.SCOPE_SUBTREE, user_filter, attrlist=attribute_names
-...
                 for dn, attrs in results:
                     data = attrs.copy()
                     data['dn'] = dn
                     _attribute, guid = cls._get_guid(data)
                     if guid and guid in guids:
                         guids.discard(guid)
                     for eid_tuple in map_text(block['external_id_tuples']):
                         backend = cls()
                         external_id = backend.build_external_id(eid_tuple, data)
-...
                                 eids.remove(external_id)
                             except ValueError:
                                 pass
                 for eid in UserExternalId.objects.filter(
                     external_id__in=eids, user__is_active=True, source=block['realm']
                 for eid in UserExternalId.objects.filter(user__is_active=True, source=block['realm']).filter(
                     Q(external_id__in=eids, external_guid__isnull=True)
                     | Q(external_guid__in=guids, external_id__isnull=True)
                     | Q(external_guid__in=guids, external_id__in=eids)
                 ):
                     if eid.user.is_active:
                         eid.user.mark_as_inactive(reason=LDAP_DEACTIVATION_REASON_NOT_PRESENT)
-...
                 return
             for user_external_id in user.userexternalid_set.all():
                 external_id = user_external_id.external_id
                 if not external_id:
                     continue
                 for block in config:
                     if user_external_id.source != force_text(block['realm']):
                         continue

     # Generated by Django 2.2.27 on 2022-04-06 21:11
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('authentic2', '0037_auto_20220331_1513'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='userexternalid',
                 name='external_guid',
                 field=models.UUIDField(null=True, verbose_name='External GUID'),
             ),
             migrations.AlterField(
                 model_name='userexternalid',
                 name='external_id',
                 field=models.CharField(max_length=256, null=True, verbose_name='external id'),
             ),
             migrations.AlterUniqueTogether(
                 name='userexternalid',
                 unique_together={('source', 'external_guid'), ('source', 'external_id')},
             ),
             migrations.AddConstraint(
                 model_name='userexternalid',
                 constraint=models.CheckConstraint(
                     check=models.Q(
                         ('external_id__isnull', False), ('external_guid__isnull', False), _connector='OR'
                     ),
                     name='at_least_one_id',
                 ),
             ),
+        ]

     class UserExternalId(models.Model):
         user = models.ForeignKey(settings.AUTH_USER_MODEL, verbose_name=_('user'), on_delete=models.CASCADE)
         source = models.CharField(max_length=256, verbose_name=_('source'))
         external_id = models.CharField(max_length=256, verbose_name=_('external id'))
         external_id = models.CharField(max_length=256, verbose_name=_('external id'), null=True)
         external_guid = models.UUIDField(verbose_name=_('External GUID'), null=True)
         created = models.DateTimeField(auto_now_add=True, verbose_name=_('creation date'))
         updated = models.DateTimeField(auto_now=True, verbose_name=_('last update date'))
         def __str__(self):
             return f'{self.user} is {self.external_id} on {self.source}'
             return f'{self.user} is {self.external_id or self.external_guid} on {self.source}'
         def __repr__(self):
             return '<UserExternalId user: {!r} source: {!r} external_id: {!r} created: {} updated: {}'.format(
                 self.user_id, self.source, self.external_id, self.created, self.updated
             return (
                 f'<UserExternalId user: {self.user!r} source: {self.source!r}'
                 f"{f' external_id: {self.external_id!r}' if self.external_id else ''}"
                 f"{f' external_guid: {self.external_guid!r}' if self.external_guid else ''}"
                 f' created: {self.created} updated: {self.updated}'
+            )
         class Meta:
-...
             verbose_name_plural = _('user external ids')
             unique_together = [
                 ('source', 'external_id'),
                 ('source', 'external_guid'),
+            ]
             constraints = [
                 models.CheckConstraint(
                     check=Q(external_id__isnull=False) | Q(external_guid__isnull=False), name='at_least_one_id'
                 ),
+            ]

     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import base64
     import json
     import logging
     import os
-...
     UPASS = 'passé'
     EMAIL = 'etienne.michu@example.net'
     CARLICENSE = '123445ABC'
     UUID = '8ff2f34a-4a36-103c-8d0a-e3a0333484d3'
     OBJECTGUID_RAW = b'\xab' * 16
     OBJECTGUID_B64 = base64.b64encode(OBJECTGUID_RAW).decode()
     CN_INCOMPLETE = 'Jean Dupond'
     DN_INCOMPLETE = 'cn=%s,o=ôrga' % escape_dn_chars(CN_INCOMPLETE)
-...
             yield s
     OBJECTGUID_SCHEMA = '''\
     dn: cn=objectguid,cn=schema,cn=config
     objectClass: olcSchemaConfig
     cn: objectguid
     olcAttributeTypes: ( 1.3.6.1.4.1.36560.1.1.3.2 NAME 'objectGUID'
       SYNTAX 1.3.6.1.4.1.1466.115.121.1.40 )
     olcObjectClasses: ( 1.3.6.1.4.1.36560.1.1.3.1 NAME 'objectWithObjectGuid'
       MAY ( objectGUID )
       SUP top AUXILIARY )
     '''
     @pytest.fixture
     def slapd_ppolicy():
         with create_slapd() as slapd:
-...
                 'group_to_role_mapping': [
                     ['cn=unknown,o=dn', ['Role2']],
                 ],
                 'lookups': ['external_id', 'username'],
+            }
+        ]
         save = mock.Mock(wraps=ldap_backend.LDAPUser.save)
-...
     def test_sync_ldap_users(slapd, settings, app, db, caplog):
         caplog.set_level('INFO')
         conn = slapd.get_connection_admin()
         entryuuid = conn.search_s('o=ôrga', ldap.SCOPE_SUBTREE, f'(uid={UID})', ['entryUUID'])[0][1]['entryUUID'][
         ].decode()
         management.call_command('sync-ldap-users')
         assert caplog.records[0].message == 'No LDAP server configured.'
-...
         assert caplog.records[2].message == (
+            (
                 "Created user etienne.michu@ldap (uuid %s) from dn=cn=Étienne Michu,o=ôrga, uid=['etienne.michu'], "
                 "sn=['Michu'], givenname=['Étienne'], l=['Paris'], mail=['etienne.michu@example.net']"
                 "sn=['Michu'], givenname=['Étienne'], l=['Paris'], mail=['etienne.michu@example.net'], entryuuid=['%s']"
+            )
             % User.objects.first().uuid
             % (User.objects.first().uuid, entryuuid)
+        )
         assert caplog.records[-1].message == 'Search for (|(mail=*)(uid=*)) returned 6 users.'
-...
         caplog.clear()
         User.objects.update(first_name='John')
         management.call_command('sync-ldap-users', verbosity=3)
         assert (
             caplog.records[2].message
             == (
                 "Updated user etienne.michu@ldap (uuid %s) from dn=cn=Étienne Michu,o=ôrga, uid=['etienne.michu'], "
                 "sn=['Michu'], givenname=['Étienne'], l=['Paris'], mail=['etienne.michu@example.net']"
+            )
             % User.objects.first().uuid
+        )
         assert caplog.records[2].message == (
             "Updated user etienne.michu@ldap (uuid %s) from dn=cn=Étienne Michu,o=ôrga, uid=['etienne.michu'], "
             "sn=['Michu'], givenname=['Étienne'], l=['Paris'], mail=['etienne.michu@example.net'], entryuuid=['%s']"
         ) % (User.objects.first().uuid, entryuuid)
     def test_get_users_select_realm(slapd, settings, db, caplog):
-...
+        ]
         user = authenticate(username=USERNAME, password=UPASS)
         assert user
         assert user.get_attributes(object(), {}) == {
         assert dict(user.get_attributes(object(), {}), entryuuid=None) == {
             'dn': 'cn=étienne michu,o=\xf4rga',
             'givenname': ['Étienne'],
             'mail': ['etienne.michu@example.net'],
             'sn': ['Michu'],
             'uid': ['etienne.michu'],
             'carlicense': ['123445ABC'],
             'entryuuid': None,
+        }
         # simulate LDAP down
         slapd.stop()
         assert user.get_attributes(object(), {}) == {
         assert dict(user.get_attributes(object(), {}), entryuuid=None) == {
             'dn': 'cn=étienne michu,o=\xf4rga',
             'givenname': ['\xc9tienne'],
             'mail': ['etienne.michu@example.net'],
             'sn': ['Michu'],
             'uid': ['etienne.michu'],
             'carlicense': ['123445ABC'],
             'entryuuid': None,
+        }
         assert not user.check_password(UPASS)
         # simulate LDAP come back up
-...
         conn = slapd.get_connection_admin()
         ldif = [(ldap.MOD_REPLACE, 'sn', [b'Micho'])]
         conn.modify_s(DN, ldif)
         assert user.get_attributes(object(), {}) == {
         assert dict(user.get_attributes(object(), {}), entryuuid=None) == {
             'dn': 'cn=étienne michu,o=\xf4rga',
             'givenname': ['\xc9tienne'],
             'mail': ['etienne.michu@example.net'],
             'sn': ['Micho'],
             'uid': ['etienne.michu'],
             'carlicense': ['123445ABC'],
             'entryuuid': None,
+        }
-...
             assert auth_user == user
             assert auth_user.username == f'{UID}@ldap'
         def test_by_username_only(self, backend, slapd, settings, client, db):
             settings.LDAP_AUTH_SETTINGS[0]['lookups'] = ['username']
             user = User.objects.create(username=UID, ou=get_default_ou())
             assert backend.authenticate(None, username=EMAIL, password=PASS) == user
             assert not models.UserExternalId.objects.exists()
             user.username = ''
             user.save()
             new_user = backend.authenticate(None, username=EMAIL, password=PASS)
             assert new_user and new_user != user
             assert new_user.username == f'{UID}@ldap'
         def test_by_guid_migration(self, backend, slapd, settings, client, db):
             settings.LDAP_AUTH_SETTINGS[0]['lookups'] = ['external_id']
             assert backend.authenticate(None, username=USERNAME, password=PASS)
             assert User.objects.count() == 1
             user_external_id = models.UserExternalId.objects.get()
             assert user_external_id.external_id
             assert not user_external_id.external_guid
             settings.LDAP_AUTH_SETTINGS[0]['lookups'] = ['guid', 'external_id']
             assert backend.authenticate(None, username=USERNAME, password=PASS)
             assert User.objects.count() == 1
             user_external_id = models.UserExternalId.objects.get()
             assert user_external_id.external_id
             assert user_external_id.external_guid
         def test_by_guid_only(self, backend, slapd, settings, client, db):
             settings.LDAP_AUTH_SETTINGS[0]['lookups'] = ['guid']
             assert backend.authenticate(None, username=USERNAME, password=PASS)
             assert User.objects.count() == 1
             user_external_id = models.UserExternalId.objects.get()
             assert not user_external_id.external_id
             assert user_external_id.external_guid
             assert backend.authenticate(None, username=USERNAME, password=PASS)
             assert User.objects.count() == 1
             user_external_id = models.UserExternalId.objects.get()
             assert not user_external_id.external_id
             assert user_external_id.external_guid
         def test_by_guid_only_objectguid(self, backend, slapd, settings, client, db, monkeypatch):
             slapd.add_ldif(OBJECTGUID_SCHEMA)
             conn = slapd.get_connection_admin()
             ldif = [
                 (ldap.MOD_ADD, 'objectClass', b'objectWithObjectGuid'),
                 (ldap.MOD_ADD, 'objectguid', OBJECTGUID_RAW),
+            ]
             conn.modify_s(DN, ldif)
             conn.unbind_s()
             settings.LDAP_AUTH_SETTINGS[0]['lookups'] = ['guid']
             monkeypatch.setattr(ldap_backend, 'USUAL_GUID_ATTRIBUTES', ['objectguid'])
             assert backend.authenticate(None, username=USERNAME, password=PASS)
             assert User.objects.count() == 1
             user_external_id = models.UserExternalId.objects.get()
             assert not user_external_id.external_id
             assert user_external_id.external_guid.bytes == OBJECTGUID_RAW
     def test_build_external_id_failure_authenticate(db, rf, slapd, settings, caplog):
-...
                 'external_id_tuples': [
                     ['missing'],
                 ],
                 'lookups': ['external_id', 'username'],
+            }
+        ]
         request = rf.get('/login/')
-...
+        )
         assert len(caplog.records) == 1
         assert caplog.records[0].levelname == 'ERROR'
         assert 'unable to build an external_id' in caplog.records[0].message
         assert 'unable to build an user_external_id' in caplog.records[0].message
     def test_build_external_id_failure_get_users(db, rf, slapd, settings, caplog):
-...
                 'external_id_tuples': [
                     ['missing'],
                 ],
                 'lookups': ['external_id', 'username'],
+            }
+        ]
         backend = ldap_backend.LDAPBackend()
-...
         assert not users
         assert len(caplog.records) == 6
         assert all(record.levelname == 'ERROR' for record in caplog.records)
         assert all('unable to build an external_id' in record.message for record in caplog.records)
         assert all('unable to build an user_external_id' in record.message for record in caplog.records)
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-ldap-use-guid-attributes-as-global-external-id-63646.patch