# ***** BEGIN LICENSE BLOCK *****
# Version: MPL 1.1/GPL 2.0/LGPL 2.1
#
# The contents of this file are subject to the Mozilla Public License Version
# 1.1 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
# http://www.mozilla.org/MPL/
#
# Software distributed under the License is distributed on an "AS IS" basis,
# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
# for the specific language governing rights and limitations under the
# License.
#
# The Original Code is Sync Server
#
# The Initial Developer of the Original Code is the Mozilla Foundation.
# Portions created by the Initial Developer are Copyright (C) 2010
# the Initial Developer. All Rights Reserved.
#
# Contributor(s):
# Tarek Ziade (tarek@mozilla.com)
#
# Alternatively, the contents of this file may be used under the terms of
# either the GNU General Public License Version 2 or later (the "GPL"), or
# the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
# in which case the provisions of the GPL or the LGPL are applicable instead
# of those above. If you wish to allow use of your version of this file only
# under the terms of either the GPL or the LGPL, and not to allow others to
# use your version of this file under the terms of the MPL, indicate your
# decision by deleting the provisions above and replace them with the notice
# and other provisions required by the GPL or the LGPL. If you do not delete
# the provisions above, a recipient may use your version of this file under
# the terms of any one of the MPL, the GPL or the LGPL.
#
# ***** END LICENSE BLOCK *****
""" LDAP Connection Pool.
"""
import codecs
from contextlib import contextmanager
import logging
from threading import RLock
import time
import ldap
from ldap.ldapobject import ReconnectLDAPObject
import six
log = logging.getLogger(__name__)
_utf8_encoder = codecs.getencoder('utf-8')
def utf8_encode(value):
"""Encode a basestring to UTF-8.
If the string is unicode encode it to UTF-8, if the string is
str then assume it's already encoded. Otherwise raise a TypeError.
:param value: A basestring
:returns: UTF-8 encoded version of value
:raises TypeError: If value is not basestring
"""
if isinstance(value, six.text_type):
return _utf8_encoder(value)[0]
elif isinstance(value, six.binary_type):
return value
else:
raise TypeError("bytes or Unicode expected, got %s"
% type(value).__name__)
class MaxConnectionReachedError(Exception):
pass
class BackendError(Exception):
def __init__(self, msg, backend):
self.bacend = backend
Exception.__init__(self, msg)
class StateConnector(ReconnectLDAPObject):
"""Just remembers who is connected, and if connected."""
def __init__(self, *args, **kw):
ReconnectLDAPObject.__init__(self, *args, **kw)
self.connected = False
self.who = ''
self.cred = ''
self._connection_time = None
def get_lifetime(self):
"""Returns the lifetime of the connection on the server in seconds."""
if self._connection_time is None:
return 0
return time.time() - self._connection_time
def simple_bind_s(self, who='', cred='', serverctrls=None,
clientctrls=None):
res = ReconnectLDAPObject.simple_bind_s(self, who, cred, serverctrls,
clientctrls)
self.connected = True
self.who = who
self.cred = cred
if self._connection_time is None:
self._connection_time = time.time()
return res
def unbind_ext_s(self, serverctrls=None, clientctrls=None):
try:
return ReconnectLDAPObject.unbind_ext_s(self, serverctrls,
clientctrls)
finally:
self.connected = False
self.who = None
self.cred = None
def add_s(self, *args, **kwargs):
return self._apply_method_s(ReconnectLDAPObject.add_s, *args,
**kwargs)
def modify_s(self, *args, **kwargs):
return self._apply_method_s(ReconnectLDAPObject.modify_s, *args,
**kwargs)
def __str__(self):
res = 'LDAP Connector'
if self.connected:
res += ' (connected)'
else:
res += ' (disconnected)'
if self.who != '':
res += ' - who: %r' % self.who
if self._uri != '':
res += ' - uri: %r' % self._uri
return res
class ConnectionManager(object):
"""LDAP Connection Manager.
Provides a context manager for LDAP connectors.
"""
def __init__(self, uri, bind=None, passwd=None, size=10, retry_max=3,
retry_delay=.1, use_tls=False, timeout=-1,
connector_cls=StateConnector, use_pool=True,
max_lifetime=600):
self._pool = []
self.size = size
self.retry_max = retry_max
self.retry_delay = retry_delay
self.uri = uri
self.bind = bind
self.passwd = passwd
self._pool_lock = RLock()
self.use_tls = use_tls
self.timeout = timeout
self.connector_cls = connector_cls
self.use_pool = use_pool
self.max_lifetime = max_lifetime
def __len__(self):
return len(self._pool)
def _match(self, bind, passwd):
if passwd is not None:
passwd = utf8_encode(passwd)
with self._pool_lock:
inactives = []
for conn in reversed(self._pool):
# already in usage
if conn.active:
continue
# let's check the lifetime
if conn.get_lifetime() > self.max_lifetime:
# this connector has lived for too long,
# we want to unbind it and remove it from the pool
try:
conn.unbind_s()
except Exception:
log.debug('Failure attempting to unbind after '
'timeout; should be harmless', exc_info=True)
self._pool.remove(conn)
continue
# we found a connector for this bind
if conn.who == bind and conn.cred == passwd:
conn.active = True
return conn
inactives.append(conn)
# no connector was available, let's rebind the latest inactive one
if len(inactives) > 0:
for conn in inactives:
try:
self._bind(conn, bind, passwd)
return conn
except Exception:
log.debug('Removing connection from pool after '
'failure to rebind', exc_info=True)
self._pool.remove(conn)
return None
# There are no connector that match
return None
def _bind(self, conn, bind, passwd):
# let's bind
if self.use_tls:
try:
conn.start_tls_s()
except Exception:
raise BackendError('Could not activate TLS on established '
'connection with %s' % self.uri,
backend=conn)
if bind is not None:
conn.simple_bind_s(bind, passwd)
conn.active = True
def _create_connector(self, bind, passwd):
"""Creates a connector, binds it, and returns it.
:param bind: user login
:type bind: string
:param passwd: user password
:type passwd: string
:returns: StateConnector
:raises BackendError: If unable to connect to LDAP
"""
tries = 0
connected = False
if passwd is not None:
passwd = utf8_encode(passwd)
exc = None
conn = None
# trying retry_max times in a row with a fresh connector
while tries < self.retry_max and not connected:
try:
conn = self.connector_cls(self.uri, retry_max=self.retry_max,
retry_delay=self.retry_delay)
conn.timeout = self.timeout
self._bind(conn, bind, passwd)
connected = True
except ldap.LDAPError as error:
exc = error
time.sleep(self.retry_delay)
if tries < self.retry_max:
log.info('Failure attempting to create and bind '
'connector; will retry after %r seconds',
self.retry_delay, exc_info=True)
else:
log.error('Failure attempting to create and bind '
'connector', exc_info=True)
tries += 1
if not connected:
if isinstance(exc, (ldap.NO_SUCH_OBJECT,
ldap.INVALID_CREDENTIALS,
ldap.SERVER_DOWN)):
raise exc
# that's something else
raise BackendError(str(exc), backend=conn)
return conn
def _get_connection(self, bind=None, passwd=None):
if bind is None:
bind = self.bind
if passwd is None:
passwd = self.passwd
if self.use_pool:
# let's try to recycle an existing one
conn = self._match(bind, passwd)
if conn is not None:
return conn
# the pool is full
if len(self._pool) >= self.size:
raise MaxConnectionReachedError(self.uri)
# we need to create a new connector
conn = self._create_connector(bind, passwd)
# adding it to the pool
if self.use_pool:
with self._pool_lock:
self._pool.append(conn)
else:
# with no pool, the connector is always active
conn.active = True
return conn
def _release_connection(self, connection):
if self.use_pool:
with self._pool_lock:
if not connection.connected:
# unconnected connector, let's drop it
self._pool.remove(connection)
else:
# can be reused - let's mark is as not active
connection.active = False
# done.
return
else:
connection.active = False
# let's try to unbind it
try:
connection.unbind_ext_s()
except ldap.LDAPError:
# avoid error on invalid state
log.debug('Failure attempting to unbind on release; '
'should be harmless', exc_info=True)
@contextmanager
def connection(self, bind=None, passwd=None):
"""Creates a context'ed connector, binds it, and returns it.
:param bind: user login
:type bind: string
:param passwd: user password
:type passwd: string
:returns: StateConnector
:raises MaxConnectionReachedError: If unable to connect to LDAP
"""
tries = 0
conn = None
while tries < self.retry_max:
try:
conn = self._get_connection(bind, passwd)
except MaxConnectionReachedError:
tries += 1
time.sleep(0.1)
# removing the first inactive connector going backward
with self._pool_lock:
reversed_list = reversed(list(enumerate(self._pool)))
for index, conn_ in reversed_list:
if not conn_.active:
self._pool.pop(index)
break
else:
break
if conn is None:
raise MaxConnectionReachedError(self.uri)
try:
yield conn
finally:
self._release_connection(conn)
def purge(self, bind, passwd=None):
"""Purge a connector.
:param bind: user login
:type bind: string
:param passwd: user password
:type passwd: string
"""
if self.use_pool:
return
if passwd is not None:
passwd = utf8_encode(passwd)
with self._pool_lock:
for conn in list(self._pool):
if conn.who != bind:
continue
if passwd is not None and conn.cred == passwd:
continue
# let's drop it
try:
conn.unbind_ext_s()
except ldap.LDAPError:
# invalid state
log.debug('Failure attempting to unbind on purge; '
'should be harmless', exc_info=True)
self._pool.remove(conn)