2014-09-14 17:24:19 +04:00
|
|
|
import ldap
|
2014-09-23 17:13:25 +04:00
|
|
|
import ldap.filter
|
2014-09-14 17:24:19 +04:00
|
|
|
from ldapprobject import LdapprObject
|
|
|
|
|
|
|
|
|
|
|
|
class Connection(object):
|
2014-11-24 16:12:38 +03:00
|
|
|
"""Initiates connection with handy methods"""
|
2014-09-14 17:24:19 +04:00
|
|
|
def __init__(self, server, protocol='ldap', port='', verify=True,
|
|
|
|
search_base=''):
|
|
|
|
self.search_base = search_base
|
|
|
|
if port == '':
|
|
|
|
port = 389 if protocol == 'ldap' else 636
|
2014-09-25 12:26:09 +04:00
|
|
|
self.ldap_url = '{}://{}:{}'.format(protocol, server, str(port))
|
2014-09-14 17:24:19 +04:00
|
|
|
try:
|
|
|
|
ldap.set_option(ldap.OPT_REFERRALS, 0)
|
|
|
|
if not verify:
|
|
|
|
ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT,
|
|
|
|
ldap.OPT_X_TLS_NEVER)
|
|
|
|
self.conn = ldap.initialize(self.ldap_url)
|
2014-09-16 14:37:21 +04:00
|
|
|
except:
|
2014-09-14 17:24:19 +04:00
|
|
|
raise
|
|
|
|
|
2014-09-25 12:26:09 +04:00
|
|
|
def type(self):
|
|
|
|
result = self.conn.search_s('', ldap.SCOPE_BASE,
|
|
|
|
attrlist=['objectClass', 'vendorName',
|
|
|
|
'supportedCapabilities'])
|
|
|
|
rootDSE = LdapprObject(result[0], self.conn)
|
|
|
|
if rootDSE.attrs['vendorName'] == ['Novell, Inc.']:
|
|
|
|
return 'eDirectory'
|
|
|
|
if '1.2.840.113556.1.4.800' in rootDSE.attrs['supportedCapabilities']:
|
|
|
|
return 'Active Directory'
|
|
|
|
if rootDSE.attrs['vendorName'] == ['Apache Software Foundation']:
|
|
|
|
return 'Apache DS'
|
|
|
|
if 'OpenLDAProotDSE' in rootDSE.attrs['objectClass']:
|
|
|
|
return 'OpenLDAP'
|
|
|
|
return 'Unknown'
|
|
|
|
|
2014-09-14 17:24:19 +04:00
|
|
|
def search(self, search_filter):
|
|
|
|
"""Get list of objects that match the search_filter
|
2014-09-23 18:02:20 +04:00
|
|
|
|
2014-09-14 17:24:19 +04:00
|
|
|
:param search_filter: filter to find the objects
|
2014-09-23 16:51:24 +04:00
|
|
|
:return: list of LdapperObjects (or empty list)
|
2014-09-14 17:24:19 +04:00
|
|
|
"""
|
2014-09-23 17:13:25 +04:00
|
|
|
search_filter = ldap.filter.escape_filter_chars(search_filter)
|
2014-09-14 17:24:19 +04:00
|
|
|
result = self.conn.search_s(self.search_base, ldap.SCOPE_SUBTREE,
|
|
|
|
search_filter)
|
|
|
|
return [LdapprObject(item, self.conn) for item in result]
|
|
|
|
|
|
|
|
def get(self, search_filter):
|
2014-09-16 14:37:21 +04:00
|
|
|
"""Get first object found
|
2014-09-14 17:24:19 +04:00
|
|
|
|
|
|
|
:param search_filter: filter to find the object
|
|
|
|
:return: LdapprObject or None
|
|
|
|
"""
|
2014-09-16 14:37:21 +04:00
|
|
|
# TODO: use sizelimit=1 with proper exception handling
|
2014-09-23 17:13:25 +04:00
|
|
|
search_filter = ldap.filter.escape_filter_chars(search_filter)
|
2014-09-16 14:37:21 +04:00
|
|
|
result = self.conn.search_ext_s(self.search_base,
|
|
|
|
ldap.SCOPE_SUBTREE,
|
|
|
|
search_filter, sizelimit=0)
|
2014-09-14 17:24:19 +04:00
|
|
|
return LdapprObject(result[0], self.conn) if result else None
|
|
|
|
|
|
|
|
def get_by_dn(self, dn):
|
|
|
|
"""Get LdapprObject for known dn
|
|
|
|
|
|
|
|
:param dn: dn of the object we're looking for
|
|
|
|
:return: LdapprObject
|
|
|
|
"""
|
|
|
|
result = self.conn.search_s(dn, ldap.SCOPE_BASE)
|
|
|
|
return LdapprObject(result[0], self.conn)
|
|
|
|
|
|
|
|
def get_dn(self, search_filter):
|
|
|
|
"""Get list of dn's that match the filter
|
|
|
|
|
|
|
|
:param search_filter: filter to find the dn's
|
|
|
|
:return: list of dn's
|
|
|
|
"""
|
2014-09-23 17:13:25 +04:00
|
|
|
search_filter = ldap.filter.escape_filter_chars(search_filter)
|
2014-09-14 17:24:19 +04:00
|
|
|
result = self.conn.search_s(self.search_base, ldap.SCOPE_SUBTREE,
|
|
|
|
search_filter)
|
|
|
|
return [dn for (dn, item) in result]
|
|
|
|
|
|
|
|
def get_values(self, dn, attr):
|
|
|
|
"""Get list of values of given attribute for dn
|
|
|
|
|
|
|
|
:param dn: dn of the object we're looking for
|
|
|
|
:param attr: attribute name (case insensitive)
|
|
|
|
:return: list of values
|
|
|
|
"""
|
|
|
|
result = self.conn.search_s(dn, ldap.SCOPE_BASE)
|
|
|
|
result_object = LdapprObject(result[0], self.conn)
|
|
|
|
return result_object.attrs[attr]
|
|
|
|
|
|
|
|
def get_value(self, dn, attr):
|
|
|
|
"""Get (first) attr value as string
|
|
|
|
|
|
|
|
:param dn: dn of the object we're looking for
|
|
|
|
:param attr: attribute name (case insensitive)
|
|
|
|
:return: value as string
|
|
|
|
"""
|
|
|
|
result = self.get_values(dn, attr)
|
|
|
|
return result[0]
|
|
|
|
|
2014-09-23 18:02:20 +04:00
|
|
|
def verify_password(self, dn, password):
|
|
|
|
try:
|
|
|
|
test_conn = ldap.initialize(self.ldap_url)
|
|
|
|
test_conn.simple_bind_s(dn, password)
|
|
|
|
test_conn.unbind_s()
|
|
|
|
except ldap.LDAPError:
|
|
|
|
return False
|
2014-09-25 12:26:09 +04:00
|
|
|
return True
|
2014-09-23 18:02:20 +04:00
|
|
|
|
2014-09-14 17:24:19 +04:00
|
|
|
def close(self):
|
|
|
|
self.conn.unbind_s()
|
|
|
|
|
|
|
|
|
|
|
|
class AuthConnection(Connection):
|
|
|
|
def __init__(self, server, bind_dn, password, **kwargs):
|
|
|
|
super(AuthConnection, self).__init__(server, **kwargs)
|
|
|
|
try:
|
|
|
|
self.conn.simple_bind_s(bind_dn, password)
|
|
|
|
except ldap.LDAPError:
|
|
|
|
raise
|
|
|
|
|
|
|
|
def add(self, dn, modlist):
|
|
|
|
"""Adds an entry to the LDAP store
|
|
|
|
|
|
|
|
:param dn: dn of the new entry
|
|
|
|
:param modlist: list of attributes made up of two-value tuples, where
|
|
|
|
the first item of each tuple is the attribute name, and the
|
|
|
|
second value is a list of attribute values.
|
|
|
|
"""
|
|
|
|
self.conn.add_s(dn, modlist)
|
|
|
|
|
|
|
|
def modify(self, dn, modlist):
|
|
|
|
self.conn.modify_s(dn, modlist)
|
|
|
|
|
|
|
|
def set_value(self, dn, attr, value):
|
|
|
|
self.conn.modify_s(dn, [(ldap.MOD_REPLACE, attr, value)])
|
|
|
|
|
|
|
|
def add_value(self, dn, attr, value):
|
|
|
|
self.conn.modify_s(dn, [(ldap.MOD_ADD, attr, value)])
|
|
|
|
|
|
|
|
def delete_value(self, dn, attr, value):
|
|
|
|
self.conn.modify_s(dn, [(ldap.MOD_DELETE, attr, value)])
|
|
|
|
|
|
|
|
def delete(self, dn):
|
|
|
|
self.conn.delete_s(dn)
|