|
'''
|
|
Module to provide the kind of interaction needed for a eZ publish website
|
|
acting as reverse proxy and passing in an HTTP header a pointer to an LDAP
|
|
entry.
|
|
'''
|
|
|
|
try:
|
|
import ldap
|
|
import ldap.modlist
|
|
except ImportError:
|
|
ldap = None
|
|
|
|
LDAP_EMAIL = 'courriel'
|
|
|
|
import time
|
|
import random
|
|
|
|
from quixote import get_publisher, get_request, get_session
|
|
from qommon import get_cfg, get_logger
|
|
from qommon.form import *
|
|
from qommon.admin.texts import TextsDirectory
|
|
from qommon import misc
|
|
|
|
from wcs.users import User
|
|
|
|
|
|
def get_ldap_conn():
|
|
if hasattr(get_request(), 'ldap_conn'):
|
|
return get_request().ldap_conn
|
|
|
|
misc_cfg = get_cfg('misc', {})
|
|
ldap_url = misc_cfg.get('aq-ezldap-url')
|
|
if not ldap_url:
|
|
return
|
|
ldap_binddn = misc_cfg.get('aq-ezldap-binddn')
|
|
ldap_bindpassword = misc_cfg.get('aq-ezldap-bindpassword')
|
|
|
|
ldap_conn = ldap.initialize(ldap_url)
|
|
ldap_conn.protocol_version = ldap.VERSION3
|
|
ldap_conn.simple_bind(ldap_binddn, ldap_bindpassword)
|
|
get_request().ldap_conn = ldap_conn
|
|
return ldap_conn
|
|
|
|
|
|
class EzEmailWidget(EmailWidget):
|
|
extra_css_class = 'EmailWidget'
|
|
def _parse(self, request):
|
|
EmailWidget._parse(self, request)
|
|
if self.value and not self.error:
|
|
# check if an account already exists
|
|
misc_cfg = get_cfg('misc', {})
|
|
ldap_basedn = misc_cfg.get('aq-ezldap-basedn')
|
|
ldap_conn = get_ldap_conn()
|
|
ldap_result_id = ldap_conn.search(ldap_basedn, ldap.SCOPE_SUBTREE,
|
|
filterstr='%s=%s' % (LDAP_EMAIL, self.value))
|
|
result_type, result_data = ldap_conn.result(ldap_result_id, all=0)
|
|
if result_type == ldap.RES_SEARCH_ENTRY:
|
|
self.error = _('This email already has an account.')
|
|
|
|
|
|
class EzLdapUser(object):
|
|
anonymous = False
|
|
is_admin = False
|
|
roles = []
|
|
|
|
def __init__(self, dn):
|
|
self.id = dn
|
|
self.dn = dn
|
|
|
|
_ldap_dict = None
|
|
def get_ldap_dict(self):
|
|
if self._ldap_dict:
|
|
return self._ldap_dict
|
|
else:
|
|
t = get_ldap_conn().search_s(self.dn, ldap.SCOPE_BASE)[0][1]
|
|
self._ldap_dict = {}
|
|
for k, v in t.items():
|
|
if v:
|
|
if k.startswith('date'):
|
|
date = datetime.datetime.strptime(v[0][:14], '%Y%m%d%H%M%S')
|
|
v[0] = date.strftime(misc.date_format())
|
|
self._ldap_dict[k] = v[0]
|
|
return self._ldap_dict
|
|
|
|
_form_data = None
|
|
def get_form_data(self):
|
|
if self._form_data:
|
|
return self._form_data
|
|
else:
|
|
self._form_data = {}
|
|
ldap_dict = self.get_ldap_dict()
|
|
for k, v in ldap_dict.items():
|
|
# get the corresponding id from wcs user
|
|
for admin_field in User.get_formdef().fields:
|
|
if admin_field.varname == k:
|
|
self._form_data[admin_field.id] = v
|
|
break
|
|
return self._form_data
|
|
|
|
def set_form_data(self, data):
|
|
pass
|
|
form_data = property(get_form_data, set_form_data)
|
|
|
|
_name = None
|
|
def get_display_name(self):
|
|
if self._name is None:
|
|
parts = []
|
|
user = self.get_ldap_dict()
|
|
users_cfg = get_cfg('users', {})
|
|
# get the corresponding ldap attribute
|
|
parts = [user.get(admin_field.varname, '')
|
|
for admin_field in User.get_formdef().fields
|
|
if admin_field.id in users_cfg.get('field_name')]
|
|
self._name = ' '.join(parts)
|
|
return self._name
|
|
display_name = property(get_display_name)
|
|
name = property(get_display_name)
|
|
|
|
def get_email(self):
|
|
return self.get_ldap_dict().get(LDAP_EMAIL)
|
|
email = property(get_email)
|
|
|
|
def can_go_in_admin(self):
|
|
return False
|
|
|
|
def can_go_in_backoffice(self):
|
|
return False
|
|
|
|
def store(self):
|
|
pass
|
|
|
|
def get_formdef(cls):
|
|
formdef = User.get_formdef()
|
|
# use only fields with a varname (= ldap attribute)
|
|
formdef.fields = [f for f in User.get_formdef().fields if f.varname ]
|
|
return formdef
|
|
get_formdef = classmethod(get_formdef)
|
|
|
|
def get_substitution_variables(self, prefix='session_'):
|
|
d = {
|
|
prefix+'user': self,
|
|
prefix+'user_display_name': self.display_name,
|
|
prefix+'user_email': self.email,
|
|
}
|
|
formdef = self.get_formdef()
|
|
if formdef:
|
|
from formdata import get_dict_with_varnames
|
|
data = get_dict_with_varnames(formdef.fields, self.form_data)
|
|
for k, v in data.items():
|
|
d[prefix+'user_'+k] = v
|
|
return d
|
|
|
|
def get_substitution_variables_list(self):
|
|
return User.get_substitution_variables_list()
|
|
|
|
def set_attributes_from_formdata(self, formdata):
|
|
new_entry = self.get_ldap_dict().copy()
|
|
for k, v in formdata.items():
|
|
# for each field, the ldap attribute is the varname
|
|
for admin_field in User.get_formdef().fields:
|
|
if k == admin_field.id:
|
|
if isinstance(v, time.struct_time):
|
|
v = time.strftime('%Y%m%d%H%M%SZ',v)
|
|
new_entry[admin_field.varname] = v or ''
|
|
break
|
|
|
|
mod_list = ldap.modlist.modifyModlist(self.get_ldap_dict(), new_entry)
|
|
if mod_list:
|
|
get_ldap_conn().modify_s(get_session().user, mod_list)
|
|
|
|
|
|
class EzMagicUser(User):
|
|
def get(cls, id, ignore_errors=False, ignore_migration=False):
|
|
if '=' in str(id):
|
|
try:
|
|
# FIXME : do a ldap search on each get is not very efficient,
|
|
# perhaps add a little cache ? (ttl 5 min ?)
|
|
get_ldap_conn().search_s(str(id), ldap.SCOPE_BASE)
|
|
except ldap.NO_SUCH_OBJECT:
|
|
if ignore_errors:
|
|
return None
|
|
raise KeyError
|
|
return EzLdapUser(id)
|
|
else:
|
|
return User.get(id, ignore_errors, ignore_migration)
|
|
get = classmethod(get)
|
|
|
|
def search(cls, q):
|
|
'''search in names and email (from wcs user configuration)'''
|
|
result = []
|
|
|
|
# build the ldap filter string : (|(attr1=*q*)(attr2=*q*)...)
|
|
users_cfg = get_cfg('users', {})
|
|
search_fields = [ LDAP_EMAIL ] + users_cfg.get('field_name')
|
|
attrs = [f.varname for f in User.get_formdef().fields if f.id in search_fields]
|
|
filterstr = '(|' + ''.join(['(%%s=*%s*)' % q % a for a in attrs]) + ')'
|
|
|
|
ldap_basedn = get_cfg('misc', {}).get('aq-ezldap-basedn')
|
|
ldap_conn = get_ldap_conn()
|
|
try:
|
|
ldap_result_id = ldap_conn.search(ldap_basedn, ldap.SCOPE_SUBTREE,
|
|
filterstr=filterstr)
|
|
except ldap.FILTER_ERROR:
|
|
return []
|
|
while True:
|
|
result_type, result_data = ldap_conn.result(ldap_result_id, all=0)
|
|
if not result_data:
|
|
break
|
|
if result_type == ldap.RES_SEARCH_ENTRY:
|
|
result.append(result_data[0][0])
|
|
return [EzLdapUser(dn) for dn in result]
|
|
search = classmethod(search)
|
|
|