0001-auth_fc-make-user-and-sub-relatively-unique-19959.patch

Benjamin Dauvergne, 14 juin 2019 15:52

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

Subject: [PATCH] auth_fc: make user and sub relatively unique (#19959)

As we are not sure they are unique in all deployments, we make them
unique relative to a new order integer field. New federations with
FranceConnect with be forced to have the column order to be 0, making
them unique.

 src/authentic2_auth_fc/backends.py            | 38 +++++---
 .../migrations/0002_fcaccount_order1.py       | 59 +++++++++++++
 .../migrations/0003_fcaccount_order2.py       | 26 ++++++
 src/authentic2_auth_fc/models.py              |  9 ++
 src/authentic2_auth_fc/views.py               | 86 +++++++++----------
 5 files changed, 163 insertions(+), 55 deletions(-)
 create mode 100644 src/authentic2_auth_fc/migrations/0002_fcaccount_order1.py
 create mode 100644 src/authentic2_auth_fc/migrations/0003_fcaccount_order2.py

     from django.contrib.auth import get_user_model
     from django.contrib.auth.backends import ModelBackend
     from django.core.exceptions import PermissionDenied, MultipleObjectsReturned
     from django.db import IntegrityError
     from authentic2.a2_rbac.utils import get_default_ou
     from authentic2 import hooks
-...
             user_info = kwargs.get('user_info')
             user = None
             try:
                 fc_account = models.FcAccount.objects.get(sub=sub, user__is_active=True)
                 logger.debug(u'existing user %s using sub %s', fc_account.user, sub)
                 user = fc_account.user
                 try:
                     account = models.FcAccount.objects.select_related().get(sub=sub)
                 except MultipleObjectsReturned:
                     account = models.FcAccount.objects.select_related().get(sub=sub, order=0)
             except models.FcAccount.DoesNotExist:
                 logger.debug(u'user with the sub %s does not exist.', sub)
             else:
                 user = fc_account.user
                 logger.debug(u'found user %s with sub %s', user, sub)
                 if not user.is_active:
                     logger.info(u'user %s login refused, it is inactive', user)
                     raise PermissionDenied
             if user_info:
                 if not user and app_settings.create:
                     User = get_user_model()
                     user = User.objects.create(ou=get_default_ou())
                     fc_account = models.FcAccount.objects.create(
                         user=user,
                         sub=sub,
                         token=json.dumps(kwargs['token']))
                     logger.debug(u'user creation enabled with fc_account (sub : %s - token : %s)',
                                  sub, json.dumps(kwargs['token']))
                     hooks.call_hooks('event', name='fc-create', user=user, sub=sub)
                     try:
                         fc_account = models.FcAccount.objects.create(
                             user=user,
                             sub=sub,
                             order=0,
                             token=json.dumps(kwargs['token']))
                     except IntegrityError:
                         # uniqueness check failed, as the user is new, it can only means that the sub is not unique
                         # let's try again
                         return self.authenticate(sub, **kwargs)
                     else:
                         logger.debug(u'user creation enabled with fc_account (sub : %s - token : %s)',
                                      sub, json.dumps(kwargs['token']))
                         hooks.call_hooks('event', name='fc-create', user=user, sub=sub)
                 if not user:
                     return None
                 logger.debug(u'updated (given_name : %s - family_name : %s)', user_info['given_name'],
                              user_info['family_name'])
                 user.first_name = user_info['given_name']

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.20 on 2019-06-12 21:27
     from __future__ import unicode_literals
     from collections import Counter
     from django.db import migrations, models
     def set_fcaccount_order(apps, schema_editor):
         FcAccount = apps.get_model('authentic2_auth_fc', 'FcAccount')
         c = Counter()
         FcAccount.objects.update(order=0)
         user_ids = (
             FcAccount.objects
             .values('user_id')
             .annotate(total=models.Count('id'))
             .filter(total__gt=1)
             .values_list('user_id', flat=True)
+        )
         subs = (
             FcAccount.objects
             .values_list('sub')
             .annotate(total=models.Count('id'))
             .filter(total__gt=1)
             .values_list('sub', flat=True)
+        )
         for account in FcAccount.objects.filter(models.Q(user_id__in=user_ids) | models.Q(sub__in=subs)):
             order = max(c[account.user_id], c[account.sub])
             if account.order != order:
                 account.order = order
                 account.save()
             c[account.user_id] = order + 1
             c[account.sub] = order + 1
         assert FcAccount.objects.filter(order__isnull=True).count() == 0
     def noop(apps, schema_editor):
         pass
     class Migration(migrations.Migration):
         # Disembiguate sub and user_id columns using an integer order column, then
         # force sub/order and user_id/order pairs unique to be unique in order
         # to move to a model where user and sub are unique without breaking on existing data
         dependencies = [
             ('authentic2_auth_fc', '0001_initial'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='fcaccount',
                 name='order',
                 field=models.IntegerField(null=True, verbose_name='order'),
             ),
             migrations.RunPython(set_fcaccount_order, noop),
+        ]

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.20 on 2019-06-13 09:44
     from __future__ import unicode_literals
     from django.conf import settings
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             migrations.swappable_dependency(settings.AUTH_USER_MODEL),
             ('authentic2_auth_fc', '0002_fcaccount_order1'),
+        ]
         operations = [
             migrations.AlterField(
                 model_name='fcaccount',
                 name='order',
                 field=models.PositiveIntegerField(default=0, verbose_name='order'),
             ),
             migrations.AlterUniqueTogether(
                 name='fcaccount',
                 unique_together=set([('sub', 'order'), ('user', 'order')]),
             ),
+        ]

         sub = models.TextField(
             verbose_name=_('sub'),
             db_index=True)
         order = models.PositiveIntegerField(
             verbose_name=_('order'),
             default=0)
         token = models.TextField(verbose_name=_('access token'))
         user_info = models.TextField(verbose_name=_('access token'), blank=True, null=True)
-...
             if 'family_name' in user_info:
                 display_name.append(user_info['family_name'])
             return ' '.join(display_name)
         class Meta:
             unique_together = [
                 ('sub', 'order'),
                 ('user', 'order'),
+            ]

     import django
     from django.db import IntegrityError
     from django.views.generic import View, FormView
     from django.http import HttpResponseRedirect, Http404
     from django.contrib.auth import REDIRECT_FIELD_NAME, get_user_model
-...
         redirect_field_name = REDIRECT_FIELD_NAME
         in_popup = False
         token = None
         user_info = None
         def get_in_popup(self):
             return self.in_popup
-...
                     self.scopes.extend(scopes)
                 return ask_authorization(request, self.get_scopes(), self.logger)
         @property
         def fc_display_name(self):
             '''Human representation of the current FC account'''
             if not self.user_info:
                 return u''
             return u'{0} {1}'.format(self.user_info['family_name'], self.user_info['given_name'])
     class PopupViewMixin(object):
         def get_in_popup(self):
-...
         '''
         def update_user_info(self):
             self.fc_account.token = json.dumps(self.token)
             self.fc_account.user_info = json.dumps(self.user_info)
             self.fc_account.save()
             self.fc_account.save(update_fields=['token', 'user_info'])
             utils.apply_user_info_mappings(self.fc_account.user, self.user_info)
             self.logger.debug('updating user_info %s', self.fc_account.user_info)
         def uniqueness_check_failed(self, request):
             if models.FcAccount.objects.filter(user=request.user, order=0).count():
                 messages.error(request,
                                _('Your account is already linked to a FranceConnect account'))
             else:
                 messages.error(request,
                                _('The FranceConnect account {} is already'
                                  ' linked with another account.').format(self.fc_display_name))
             return self.redirect(request)
         def get(self, request, *args, **kwargs):
             registration = True if 'registration' in request.GET else False
             '''Request an access grant code and associate it to the current user'''
             self.service_slug = request.GET.get(constants.SERVICE_FIELD_NAME)
             if request.user.is_authenticated():
                 # Prevent to add a link with an FC account already linked with another user.
                 try:
                     fc_account = models.FcAccount.objects.get(sub=self.sub, user__is_active=True)
                     if fc_account.user is not request.user:
                         msg = 'Attempt to link FC account {} already linked with user {}'
                         self.logger.info(msg.format(self.sub, fc_account.user))
                         messages.error(request,
                                        _('The FranceConnect account {} is already'
                                          ' linked with another account.').format(fc_account))
                         return self.redirect(request)
                 except models.FcAccount.DoesNotExist:
                     pass
                 # Prevent to add a link to an user which is already linked to an FC account
                 if request.user.fc_accounts.exists():
                     self.logger.warning(u'cannot link to sub %s, account is already linked to an '
                                         u'FC account', self.sub)
                     messages.error(request,
                                    _('Your account is already linked to a FranceConnect account'))
                     return self.redirect(request)
                     self.fc_account, created = models.FcAccount.objects.get_or_create(
                         sub=self.sub, user=request.user, order=0,
                         defaults={'token': json.dumps(self.token)})
                 except IntegrityError:
                     # unique index check failed, find why.
                     return self.uniqueness_check_failed(request)
                 json_token = json.dumps(self.token)
                 self.fc_account, created = models.FcAccount.objects.get_or_create(
                     defaults={'token': json_token},
                     sub=self.sub, user=request.user)
                 if created:
                     self.logger.info('fc link created sub %s', self.sub)
                     self.update_user_info()
                     data = utils.get_mapped_attributes_flat(request)
                     if 'email' in data:
                         messages.info(request,
                                       _('Your FranceConnect account {} with '
                                         'email {} has been linked.').format(self.fc_account,
                                                                             data['email']))
                     else:
                         messages.info(request, _('Your FranceConnect account {} '
                                                  'has been linked.').format(self.fc_account))
                     hooks.call_hooks('event', name='fc-link', user=request.user, sub=self.sub,
                                      request=request)
                     messages.info(request,
                                   _('Your FranceConnect account {} has been linked.').format(self.fc_display_name))
                     hooks.call_hooks('event', name='fc-link', user=request.user, sub=self.sub, request=request)
                 else:
                     self.fc_account.token = json_token
                     self.fc_account.save()
                     self.update_user_info()
                     messages.info(request, _('Your local account has been updated.'))
                 self.update_user_info()
                 return self.redirect(request)
             default_ou = get_default_ou()
-...
                 sub=self.sub,
                 user_info=self.user_info,
                 token=self.token)
             if user:
                 self.fc_account = user.fc_accounts.get(order=0)
             if not user and self.user_info.get('email') and email_is_unique:
                 email = self.user_info['email']
                 User = get_user_model()
-...
                         user = qs[0]
                         # but does he have already a link to an FC account ?
                         if not user.fc_accounts.exists():
                             fc_account, created = models.FcAccount.objects.get_or_create(
                                 defaults={'token': json.dumps(self.token)},
                                 sub=self.sub, user=user)
                             try:
                                 self.fc_account, created = models.FcAccount.objects.get_or_create(
                                     defaults={'token': json.dumps(self.token)},
                                     sub=self.sub, user=user, order=0)
                             except IntegrityError:
                                 # unique index check failed, find why.
                                 return self.uniqueness_check_failed(request)
                             if created:
                                 self.logger.info(u'fc link created sub %s user %s', self.sub, user)
                                 hooks.call_hooks('event', name='fc-link', user=user, sub=self.sub,
-...
                             return self.redirect(request)
             if user:
                 a2_utils.login(request, user, 'france-connect', service_slug=self.service_slug)
                 self.fc_account = models.FcAccount.objects.get(sub=self.sub, user=user)
                 self.fc_account.token = json.dumps(self.token)
                 self.fc_account.save(update_fields=['token'])
                 self.update_user_info()
                 self.logger.info('logged in using fc sub %s', self.sub)
                 return self.redirect(request)
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-auth_fc-make-user-and-sub-relatively-unique-19959.patch