123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
import urllib
|
|
import logging
|
|
import os
|
|
import json
|
|
import datetime
|
|
|
|
from django.core.urlresolvers import reverse
|
|
|
|
from . import app_settings
|
|
|
|
|
|
def build_logout_url(request):
|
|
"""
|
|
For now fc-id_token in request.session is used as the flag of an
|
|
active session on the OP. It is set in the login view and deleted in the
|
|
logout return view.
|
|
"""
|
|
if 'fc-id_token' in request.session:
|
|
callback = request.build_absolute_uri(reverse('fc-logout'))
|
|
qs = {
|
|
'id_token_hint': request.session.get('fc-id_token_raw'),
|
|
'post_logout_redirect_uri': callback
|
|
}
|
|
return app_settings.logout_url + '?' + urllib.urlencode(qs)
|
|
return None
|
|
|
|
|
|
def get_mapped_attributes(request):
|
|
values = {}
|
|
if 'fc-user_info' in request.session:
|
|
for fc_name, local_name in app_settings.attributes_mapping.items():
|
|
if fc_name in request.session['fc-user_info']:
|
|
values[local_name] = [request.session['fc-user_info'][fc_name]]
|
|
return values
|
|
|
|
|
|
def get_mapped_attributes_flat(request):
|
|
values = {}
|
|
if 'fc-user_info' in request.session:
|
|
for fc_name, local_name in app_settings.attributes_mapping.items():
|
|
if fc_name in request.session['fc-user_info']:
|
|
values[local_name] = request.session['fc-user_info'][fc_name]
|
|
return values
|
|
|
|
|
|
def mapping_to_value(mapping, user_info):
|
|
if isinstance(mapping, basestring):
|
|
value = user_info[mapping]
|
|
elif 'ref' in mapping:
|
|
value = user_info[mapping['ref']]
|
|
elif 'value' in mapping:
|
|
value = mapping['value']
|
|
elif 'compute' in mapping:
|
|
if mapping['compute'] == 'today':
|
|
value = datetime.date.today()
|
|
else:
|
|
raise NotImplementedError
|
|
|
|
if 'translation' in mapping:
|
|
if mapping['translation'] == 'insee-communes':
|
|
value = resolve_insee_commune(value)
|
|
elif mapping['translation'] == 'insee-countries':
|
|
value = resolve_insee_country(value)
|
|
elif mapping['translation'] == 'isodate':
|
|
value = datetime.datetime.strptime(value, '%Y-%m-%d').date()
|
|
elif mapping['translation'] == 'simple':
|
|
value = mapping['translation_simple'].get(
|
|
value, mapping.get('translation_simple_default', ''))
|
|
else:
|
|
raise NotImplementedError
|
|
return value
|
|
|
|
|
|
_insee_communes = None
|
|
|
|
|
|
def resolve_insee_commune(insee_code):
|
|
global _insee_communes
|
|
if not _insee_communes:
|
|
_insee_communes = json.load(
|
|
open(
|
|
os.path.join(
|
|
os.path.dirname(__file__), 'insee-communes.json')))
|
|
return _insee_communes.get(insee_code, 'Code insee inconnu')
|
|
|
|
|
|
_insee_countries = None
|
|
|
|
|
|
def resolve_insee_country(insee_code):
|
|
global _insee_countries
|
|
|
|
if not _insee_countries:
|
|
_insee_countries = json.load(
|
|
open(
|
|
os.path.join(
|
|
os.path.dirname(__file__), 'insee-countries.json')))
|
|
return _insee_countries.get(insee_code, 'Code insee inconnu')
|
|
|
|
|
|
def apply_user_info_mappings(user, user_info):
|
|
assert user
|
|
assert user_info
|
|
|
|
logger = logging.getLogger(__name__)
|
|
mappings = app_settings.user_info_mappings
|
|
|
|
for attribute, mapping in mappings.iteritems():
|
|
try:
|
|
value = mapping_to_value(mapping, user_info)
|
|
except (ValueError, KeyError, NotImplementedError) as e:
|
|
logger.warning(u'auth_fc: cannot apply mapping %s <- %r: %s', attribute, mapping, e)
|
|
continue
|
|
if hasattr(user.attributes, attribute):
|
|
if getattr(mapping, 'verified', False):
|
|
setattr(user.verified_attributes, attribute, value)
|
|
else:
|
|
setattr(user.attributes, attribute, value)
|
|
elif hasattr(user, 'attribute'):
|
|
setattr(user, attribute, value)
|
|
else:
|
|
logger.warning(u'auth_fc: unknown attribut in user_info mapping: %s', attribute)
|