django-allauth-ens/allauth_ens/adapter.py

# -*- coding: utf-8 -*-
import ldap

from allauth.account.utils import user_email, user_field, user_username
from allauth.account.models import EmailAddress
from allauth.socialaccount.adapter import DefaultSocialAccountAdapter, get_account_adapter
from allauth.socialaccount.models import SocialAccount
from django.conf import settings

import six

DEPARTMENTS_LIST = (
    ('phy', u'Physique'),
    ('maths', u'Maths'),
    ('bio', u'Biologie'),
    ('chimie', u'Chimie'),
    ('geol', u'Géosciences'),
    ('dec', u'DEC'),
    ('info', u'Informatique'),
    ('litt', u'Littéraire'),
    ('guests', u'Pensionnaires étrangers'),
    ('pei', u'PEI'),
)

def _init_ldap():
    server = getattr(settings, "LDAP_SERVER", "ldaps://ldap.spi.ens.fr:636")
    ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT,
                    ldap.OPT_X_TLS_NEVER)
    l = ldap.initialize(server)
    l.set_option(ldap.OPT_REFERRALS, 0)
    l.set_option(ldap.OPT_PROTOCOL_VERSION, 3)
    l.set_option(ldap.OPT_X_TLS, ldap.OPT_X_TLS_DEMAND)
    l.set_option(ldap.OPT_X_TLS_DEMAND, True)
    l.set_option(ldap.OPT_DEBUG_LEVEL, 255)
    l.set_option(ldap.OPT_NETWORK_TIMEOUT, 10)
    l.set_option(ldap.OPT_TIMEOUT, 10)
    return l

def get_ldap_infos(clipper):
    assert clipper.isalnum()
    data = {'email':'{}@clipper.ens.fr'.format(clipper.strip().lower())}
    try:
        l = _init_ldap()

        info = l.search_s('dc=spi,dc=ens,dc=fr',
                          ldap.SCOPE_SUBTREE,
                          ('(uid=%s)' % (clipper,)),
                          [str("cn"),
                           str("mailRoutingAddress"),
                           str("homeDirectory") ])

        if len(info) > 0:
            infos = info[0][1]

            # Name
            data['name'] = infos.get('cn', [''])[0].decode("utf-8")

            # Parsing homeDirectory to get entrance year and departments
            annee = '00'
            promotion = 'Inconnue'

            if 'homeDirectory' in infos:
                dirs = infos['homeDirectory'][0].split('/')
                if dirs[1] == 'users':
                    annee = dirs[2]
                    dep = dirs[3]
                    dep = dict(DEPARTMENTS_LIST).get(dep.lower(), '')
                    promotion = u'%s %s' % (dep, annee)
                    data['annee'] = annee
                    data['promotion'] = promotion

            # Mail
            pmail = infos.get('mailRoutingAddress', [])
            if len(pmail) > 0 :
                data['email'] = pmail[0]

    except ldap.LDAPError:
        pass

    return data


class LongTermClipperAccountAdapter(DefaultSocialAccountAdapter):
    """
    A class to manage the fact that people loose their account at the end of 
    their scolarity and that their clipper login might be reused later
    """

    def pre_social_login(self, request, sociallogin):
        clipper = sociallogin.account.uid
        try:
            a = SocialAccount.objects.get(provider='old_clipper',
                                          uid=clipper)
            # An account with that uid was registered, but potentially
            # deprecated at the beginning of the year
            # We need to check that the user is still the same as before
            ldap_data = get_ldap_infos(clipper)
            self._ldap_data = ldap_data
            
            if a.user.username != self.get_username(clipper, ldap_data):
                # The admission year is different
                # We need a new SocialAccount
                # But before that, we need to invalidate the email address of
                # the previous user
                email = ldap_data.get('email')
                print(email, 'deprecate')
                u_mails = EmailAddress.objects.filter(user=a.user)
                try:
                    clipper_mail = u_mails.get(email=email)
                    if clipper_mail.primary:
                        n_mails = u_mails.filter(primary=False)
                        if n_mails.exists():
                            n_mails[0].set_as_primary()
                        else:
                            user_email(a.user, '')
                            a.user.save()
                    clipper_mail.delete()
                except EmailAddress.DoesNotExist:
                    pass
                return
            
            # The admission year is the same, we can update the model and keep
            # the previous SocialAccount instance
            a.provider = 'clipper'
            a.save()
            
            # Redo the thing that had failed just before
            sociallogin.lookup()
            
        except SocialAccount.DoesNotExist:
            return

    def get_username(self, clipper, data):
        """
        Util function to generate a unique username, by default 'clipper@promo'
        This is used to disambiguate and recognize if the person is the same
        """
        return "{}@{}".format(clipper, data.get('annee', '00'))
        
    def save_user(self, request, sociallogin, form=None):
        print("populate user", sociallogin.account.uid)
        user = sociallogin.user
        user.set_unusable_password()
        
        clipper = sociallogin.account.uid
        ldap_data = self._ldap_data if hasattr(self, '_ldap_data') \
                    else get_ldap_infos(clipper)
        
        username = self.get_username(clipper, ldap_data)
        email = ldap_data.get('email')
        name = ldap_data.get('name')
        user_username(user, username or '')
        user_email(user, email or '')
        name_parts = (name or '').split(' ')
        user_field(user, 'first_name', name_parts[0])
        user_field(user, 'last_name', ' '.join(name_parts[1:]))
        print(user.username, user)

        # Ignore form
        get_account_adapter().populate_username(request, user)

        # Save extra data (only once)
        sociallogin.account.extra_data = sociallogin.extra_data = ldap_data
        sociallogin.save(request)
        sociallogin.account.save()
        
        return user

def deprecate_clippers():
    clippers = SocialAccount.objects.filter(provider='clipper')
    c_uids = clippers.values_list('uid', flat=True)
    
    # Clear old clipper accounts that wer replaced by new ones (o avoid conflicts)
    SocialAccount.objects.filter(provider='clipper_inactive', uid__in=c_uids).delete()
    
    # Deprecate accounts
    clippers.update(provider='clipper_inactive')