security-and-privacy/ldapcherry
security-and-privacy/ldapcherry
1
0
Fork 0
mirror of https://github.com/kakwa/ldapcherry synced 2024-11-22 09:24:21 +01:00
Code Releases Activity

multiple modification in backend ldap

Browse Source 
* better handling of which attributes is recovered
* fix set_attrs method
* change api of set_attrs method for constancy
* change calls to _get_user and _search methods (internal api change)
This commit is contained in:
kakwa 2015-06-16 21:29:40 +02:00
parent beb6d96d90
commit a2fe74539b
1 changed files with 39 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
ldapcherry/backend/backendLdap.py
View File

@ -17,6 +17,11 @@ class DelUserDontExists(Exception):
self.user = user self.user = user
self.log = "cannot remove user, user <%(user)s> does not exist" % { 'user' : user} self.log = "cannot remove user, user <%(user)s> does not exist" % { 'user' : user}
NO_ATTR = 0
DISPLAYED_ATTRS = 1
LISTED_ATTRS = 2
ALL_ATTRS = 3
class Backend(ldapcherry.backend.Backend): class Backend(ldapcherry.backend.Backend):
@ -100,12 +105,24 @@ class Backend(ldapcherry.backend.Backend):
return ldap_client return ldap_client
def _search(self, searchfilter, attrs, basedn): def _search(self, searchfilter, attrs, basedn):
if attrs == NO_ATTR:
attrlist = []
elif attrs == DISPLAYED_ATTRS:
# fix me later (to much attributes)
attrlist = self.attrlist
elif attrs == LISTED_ATTRS:
attrlist = self.attrlist
elif attrs == ALL_ATTRS:
attrlist = None
else:
attrlist = None
ldap_client = self._bind() ldap_client = self._bind()
try: try:
r = ldap_client.search_s(basedn, r = ldap_client.search_s(basedn,
ldap.SCOPE_SUBTREE, ldap.SCOPE_SUBTREE,
searchfilter, searchfilter,
attrlist=attrs attrlist=attrlist
) )
except ldap.FILTER_ERROR as e: except ldap.FILTER_ERROR as e:
self._logger( self._logger(
@ -128,25 +145,21 @@ class Backend(ldapcherry.backend.Backend):
ldap_client.unbind_s() ldap_client.unbind_s()
return r return r
def _get_user(self, username, attrs=True): def _get_user(self, username, attrs=ALL_ATTRS):
if attrs:
a = self.attrlist
else:
a = None
user_filter = self.user_filter_tmpl % { user_filter = self.user_filter_tmpl % {
'username': username 'username': username
} }
r = self._search(user_filter, a, self.userdn) r = self._search(user_filter, attrs, self.userdn)
if len(r) == 0: if len(r) == 0:
return None return None
if attrs: if attrs == NO_ATTR:
dn_entry = r[0]
else:
dn_entry = r[0][0] dn_entry = r[0][0]
else:
dn_entry = r[0]
return dn_entry return dn_entry
def _str(self, s): def _str(self, s):
@ -157,7 +170,7 @@ class Backend(ldapcherry.backend.Backend):
def auth(self, username, password): def auth(self, username, password):
binddn = self._get_user(username, False) binddn = self._get_user(username, NO_ATTR)
if not binddn is None: if not binddn is None:
ldap_client = self._connect() ldap_client = self._connect()
try: try:
@ -210,33 +223,34 @@ class Backend(ldapcherry.backend.Backend):
def del_user(self, username): def del_user(self, username):
ldap_client = self._bind() ldap_client = self._bind()
dn = self._get_user(username, False) dn = self._get_user(username, NO_ATTR)
if not dn is None: if not dn is None:
ldap_client.delete_s(dn) ldap_client.delete_s(dn)
else: else:
raise DelUserDontExists(username) raise DelUserDontExists(username)
ldap_client.unbind_s() ldap_client.unbind_s()
def set_attrs(self, attrs, username): def set_attrs(self, username, attrs):
ldap_client = self._bind() ldap_client = self._bind()
tmp = self._get_user(username, True) tmp = self._get_user(username, ALL_ATTRS)
dn = tmp[0] dn = tmp[0]
old_attrs = tmp[1] old_attrs = tmp[1]
for attr in attrs: for attr in attrs:
content = attrs[attr] content = self._str(attrs[attr])
attr = self._str(attr)
new = { attr : content } new = { attr : content }
if attr in old_attrs: if attr in old_attrs:
old = { attr: old_attrs[attr]} old = { attr: old_attrs[attr]}
else:
old = {}
ldif = modlist.modifyModlist(old, new) ldif = modlist.modifyModlist(old, new)
ldap_client.modify_s(dn, ldif) ldap_client.modify_s(dn, ldif)
else:
ldif = modlist.addModlist({ attr : content })
ldap_client.add_s(dn,ldif)
ldap_client.unbind_s() ldap_client.unbind_s()
def add_to_group(self, username, groups): def add_to_group(self, username, groups):
ldap_client = self._bind() ldap_client = self._bind()
tmp = self._get_user(username, True) tmp = self._get_user(username, NO_ATTR)
dn = tmp[0] dn = tmp[0]
attrs = tmp[1] attrs = tmp[1]
attrs['dn'] = dn attrs['dn'] = dn
@ -249,7 +263,7 @@ class Backend(ldapcherry.backend.Backend):
def rm_from_group(self, username): def rm_from_group(self, username):
ldap_client = self._bind() ldap_client = self._bind()
tmp = self._get_user(username, True) tmp = self._get_user(username, NO_ATTR)
dn = tmp[0] dn = tmp[0]
attrs = tmp[1] attrs = tmp[1]
attrs['dn'] = dn attrs['dn'] = dn
@ -266,7 +280,7 @@ class Backend(ldapcherry.backend.Backend):
searchfilter = self.search_filter_tmpl % { searchfilter = self.search_filter_tmpl % {
'searchstring': searchstring 'searchstring': searchstring
} }
for u in self._search(searchfilter, None, self.userdn): for u in self._search(searchfilter, DISPLAYED_ATTRS, self.userdn):
attrs = {} attrs = {}
attrs_tmp = u[1] attrs_tmp = u[1]
for attr in attrs_tmp: for attr in attrs_tmp:
@ -282,7 +296,7 @@ class Backend(ldapcherry.backend.Backend):
def get_user(self, username): def get_user(self, username):
ret = {} ret = {}
attrs_tmp = self._get_user(username)[1] attrs_tmp = self._get_user(username, ALL_ATTRS)[1]
for attr in attrs_tmp: for attr in attrs_tmp:
value_tmp = attrs_tmp[attr] value_tmp = attrs_tmp[attr]
if len(value_tmp) == 1: if len(value_tmp) == 1:
@ -292,14 +306,14 @@ class Backend(ldapcherry.backend.Backend):
return ret return ret
def get_groups(self, username): def get_groups(self, username):
userdn = self._get_user(username, False) userdn = self._get_user(username, NO_ATTR)
searchfilter = self.group_filter_tmpl % { searchfilter = self.group_filter_tmpl % {
'userdn': userdn, 'userdn': userdn,
'username': username 'username': username
} }
groups = self._search(searchfilter, None, self.groupdn) groups = self._search(searchfilter, NO_ATTR, self.groupdn)
ret = [] ret = []
for entry in groups: for entry in groups:
ret.append(entry[0]) ret.append(entry[0])