clean
Бородин Роман
parent
a38da31983
commit
be7a0a914c
|
|
@ -1,3 +0,0 @@
|
|||
from api import *
|
||||
from connection import *
|
||||
from ldapprobject import *
|
||||
|
|
@ -1,13 +0,0 @@
|
|||
import sys
|
||||
import ldap
|
||||
from connection import Connection, AuthConnection
|
||||
|
||||
|
||||
def connect_to(server, *args, **kwargs):
|
||||
try:
|
||||
if args or 'bind_dn' and 'password' in kwargs:
|
||||
return AuthConnection(server, *args, **kwargs)
|
||||
return Connection(server, **kwargs)
|
||||
except Exception as e:
|
||||
print str(e)
|
||||
|
||||
|
|
@ -1,147 +0,0 @@
|
|||
import ldap
|
||||
import ldap.filter
|
||||
from ldapprobject import LdapprObject
|
||||
|
||||
|
||||
class Connection(object):
|
||||
"""Initiates connection with handy methods"""
|
||||
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
|
||||
self.ldap_url = '{}://{}:{}'.format(protocol, server, str(port))
|
||||
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)
|
||||
except:
|
||||
raise
|
||||
|
||||
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'
|
||||
|
||||
def search(self, search_filter):
|
||||
"""Get list of objects that match the search_filter
|
||||
|
||||
:param search_filter: filter to find the objects
|
||||
:return: list of LdapperObjects (or empty list)
|
||||
"""
|
||||
search_filter = ldap.filter.escape_filter_chars(search_filter)
|
||||
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):
|
||||
"""Get first object found
|
||||
|
||||
:param search_filter: filter to find the object
|
||||
:return: LdapprObject or None
|
||||
"""
|
||||
# TODO: use sizelimit=1 with proper exception handling
|
||||
search_filter = ldap.filter.escape_filter_chars(search_filter)
|
||||
result = self.conn.search_ext_s(self.search_base,
|
||||
ldap.SCOPE_SUBTREE,
|
||||
search_filter, sizelimit=0)
|
||||
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
|
||||
"""
|
||||
search_filter = ldap.filter.escape_filter_chars(search_filter)
|
||||
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]
|
||||
|
||||
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
|
||||
return True
|
||||
|
||||
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)
|
||||
|
|
@ -1,79 +0,0 @@
|
|||
import ldap
|
||||
import ldif
|
||||
from ldap.cidict import cidict
|
||||
from StringIO import StringIO
|
||||
from string import lower
|
||||
|
||||
|
||||
class CustomCidict(cidict):
|
||||
def __getitem__(self, key):
|
||||
"""Override of the __getitem__ method to return an empty list if a key
|
||||
does not exist (instead of raising an exception)
|
||||
"""
|
||||
if lower(key) in self.data:
|
||||
return self.data[lower(key)]
|
||||
return []
|
||||
|
||||
|
||||
class LdapprObject(object):
|
||||
"""\
|
||||
The LdapprObject is used to handle search results from the Connection
|
||||
class. It's a representation of a single object in the LDAP Directory.
|
||||
"""
|
||||
def __init__(self, result, conn):
|
||||
"""The class is initialized with a tuple: (dn, {attributes}), and the
|
||||
existing connection
|
||||
"""
|
||||
(self.dn, self.attributes) = result
|
||||
self.attrs = CustomCidict(self.attributes)
|
||||
self.conn = conn
|
||||
|
||||
def __str__(self):
|
||||
"""Pretty prints all attributes with values."""
|
||||
col_width = max(len(key) for key in self.attrs.keys())
|
||||
pretty_string = '{attr:{width}} : {value}\n'.format(
|
||||
attr='dn', width=col_width, value=self.dn)
|
||||
for key, value in self.attrs.iteritems():
|
||||
if len(str(value[0])) > 80: # hack to 'detect' binary attrs
|
||||
value = ['binary']
|
||||
for single_value in value:
|
||||
pretty_string += '{attr:{width}} : {value}\n'.format(
|
||||
attr=self._case(key), width=col_width, value=single_value)
|
||||
key = ''
|
||||
return pretty_string
|
||||
|
||||
def _case(self, attr):
|
||||
"""Transforms an attribute to correct case (e.g. gIvEnNaMe becomes
|
||||
givenName). If attr is unknown nothing is transformed.
|
||||
|
||||
:param attr: may be incorrectly cased
|
||||
:return: attr in proper case
|
||||
"""
|
||||
try:
|
||||
index = [x.lower() for x in self.attrs.keys()].index(attr.lower())
|
||||
return self.attrs.keys()[index]
|
||||
except:
|
||||
return attr
|
||||
|
||||
def to_ldif(self):
|
||||
"""Makes LDIF of ldappr object."""
|
||||
out = StringIO()
|
||||
ldif_out = ldif.LDIFWriter(out)
|
||||
ldif_out.unparse(self.dn, self.attributes)
|
||||
return out.getvalue()
|
||||
|
||||
def set_value(self, attr, value):
|
||||
attr = self._case(attr)
|
||||
self.conn.modify_s(self.dn, [(ldap.MOD_REPLACE, attr, value)])
|
||||
self.attrs[attr] = [value]
|
||||
|
||||
def add_value(self, attr, value):
|
||||
attr = self._case(attr)
|
||||
self.conn.modify_s(self.dn, [(ldap.MOD_ADD, attr, value)])
|
||||
self.attrs[attr].append(value)
|
||||
|
||||
def remove_value(self, attr, value):
|
||||
attr = self._case(attr)
|
||||
self.conn.modify_s(self.dn, [(ldap.MOD_DELETE, attr, value)])
|
||||
if value in self.attrs[attr]:
|
||||
self.attrs[attr].remove(value)
|
||||
Loading…
Reference in New Issue