132 lines
4.9 KiB
Python
132 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
import ldap
|
|
|
|
from allauth.account.utils import user_email, user_field, user_username
|
|
from allauth.socialaccount.adapter import DefaultSocialAccountAdapter
|
|
from allauth.socialaccount.models import SocialAccount
|
|
|
|
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 get_ldap_infos(clipper):
|
|
assert clipper.isalnum()
|
|
data = {'email':'{}@clipper.ens.fr'.format(clipper.strip().lower())}
|
|
try:
|
|
ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT,
|
|
ldap.OPT_X_TLS_NEVER)
|
|
l = ldap.initialize("ldaps://ldap.spi.ens.fr:636")
|
|
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)
|
|
|
|
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.extra_data.get('annee', '-1') != ldap_data.get('annee', '00'):
|
|
# The admission year is different
|
|
# We need a new SocialAccount
|
|
return
|
|
|
|
# The admission year is the same, we can update the model
|
|
a.provider = 'clipper'
|
|
a.save()
|
|
sociallogin.lookup()
|
|
except SocialAccount.DoesNotExist:
|
|
return
|
|
|
|
def create_username(self, clipper, data):
|
|
return "{}:{}".format(clipper, data.get('annee', '00'))
|
|
|
|
def populate_user(self, request, sociallogin, data):
|
|
clipper = sociallogin.account.uid
|
|
ldap_data = self._ldap_data if hasattr(self, '_ldap_data') else get_ldap_infos(clipper)
|
|
# Save extra data (only once)
|
|
sociallogin.account.extra_data = sociallogin.extra_data = ldap_data
|
|
username = self.create_username(clipper, data)
|
|
first_name = data.get('first_name')
|
|
last_name = data.get('last_name')
|
|
email = ldap_data.get('email')
|
|
name = ldap_data.get('name')
|
|
user = sociallogin.user
|
|
user_username(user, username or '')
|
|
user_email(user, email or '')
|
|
name_parts = (name or '').split(' ')
|
|
user_field(user, 'first_name', first_name or name_parts[0])
|
|
user_field(user, 'last_name', last_name or ' '.join(name_parts[1:]))
|
|
print(user.username, user)
|
|
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='old_clipper', uid__in=c_uids).delete()
|
|
|
|
# Deprecate accounts
|
|
clippers.update(provider='old_clipper')
|