August 2016 - calendarserver-changes

[15785] CalendarServer/trunk
by source_changes＠macosforge.org 03 Aug '16

03 Aug '16

Revision: 15785 http://trac.calendarserver.org//changeset/15785 Author: sagen(a)apple.com Date: 2016-08-03 16:21:27 -0700 (Wed, 03 Aug 2016) Log Message: ----------- Get latest twext with new LDAP pool changes Modified Paths: -------------- CalendarServer/trunk/requirements-cs.txt CalendarServer/trunk/txdav/who/util.py Modified: CalendarServer/trunk/requirements-cs.txt =================================================================== --- CalendarServer/trunk/requirements-cs.txt 2016-08-03 23:20:52 UTC (rev 15784) +++ CalendarServer/trunk/requirements-cs.txt 2016-08-03 23:21:27 UTC (rev 15785) @@ -7,7 +7,7 @@ zope.interface==4.1.3 setuptools==18.5 - --editable svn+http://svn.calendarserver.org/repository/calendarserver/twext/trunk@157… + --editable svn+http://svn.calendarserver.org/repository/calendarserver/twext/trunk@157… cffi==1.7.0 pycparser==2.14 #twisted Modified: CalendarServer/trunk/txdav/who/util.py =================================================================== --- CalendarServer/trunk/txdav/who/util.py 2016-08-03 23:20:52 UTC (rev 15784) +++ CalendarServer/trunk/txdav/who/util.py 2016-08-03 23:21:27 UTC (rev 15785) @@ -191,7 +191,8 @@ CalRecordType.address: extraFilters.get("addresses", ""), }, threadPoolMax=params.get("threadPoolMax", 10), - connectionMax=params.get("connectionMax", 10), + authConnectionMax=params.get("authConnectionMax", 5), + queryConnectionMax=params.get("queryConnectionMax", 5), tries=params.get("tries", 3), warningThresholdSeconds=params.get("warningThresholdSeconds", 5), )

1 0

[15784] twext/trunk/twext/who/ldap
by source_changes＠macosforge.org 03 Aug '16

03 Aug '16

Revision: 15784 http://trac.calendarserver.org//changeset/15784 Author: sagen(a)apple.com Date: 2016-08-03 16:20:52 -0700 (Wed, 03 Aug 2016) Log Message: ----------- LDAP connection pool sizes now indepently configurable; pool name is included in log messages and stats keys; an anonymous bind is performed after the credentials-verification bind in case servers limit the number of times a given user can be "logged in" at the same time. Modified Paths: -------------- twext/trunk/twext/who/ldap/_service.py twext/trunk/twext/who/ldap/test/test_service.py Modified: twext/trunk/twext/who/ldap/_service.py =================================================================== --- twext/trunk/twext/who/ldap/_service.py 2016-08-03 18:01:01 UTC (rev 15783) +++ twext/trunk/twext/who/ldap/_service.py 2016-08-03 23:20:52 UTC (rev 15784) @@ -217,6 +217,10 @@ self.activeCount = 0 self.connectionsCreated = 0 self.connectionMax = connectionMax + self.log.debug( + "Created {pool} LDAP connection pool with connectionMax={max}", + pool=self.poolName, max=self.connectionMax + ) def getConnection(self): """ @@ -243,25 +247,25 @@ self.connectionCreateLock.release() else: self.connectionCreateLock.release() - self.ds.poolStats["connection-blocked"] += 1 + self.ds.poolStats["connection-{}-blocked".format(self.poolName)] += 1 connection = self.connectionQueue.get() - connectionID = "connection-{}".format( - self.connections.index(connection) + connectionID = "connection-{}-{}".format( + self.poolName, self.connections.index(connection) ) self.ds.poolStats[connectionID] += 1 self.activeCount = len(self.connections) - self.connectionQueue.qsize() - self.ds.poolStats["connection-active"] = self.activeCount - self.ds.poolStats["connection-max"] = max( - self.ds.poolStats["connection-max"], self.activeCount + self.ds.poolStats["connection-{}-active".format(self.poolName)] = self.activeCount + self.ds.poolStats["connection-{}-max".format(self.poolName)] = max( + self.ds.poolStats["connection-{}-max".format(self.poolName)], self.activeCount ) if self.activeCount > self.connectionMax: self.log.error( - "Active LDAP connections ({active}) exceeds maximum ({max})", - active=self.activeCount, max=self.connectionMax + "[Pool: {pool}] Active LDAP connections ({active}) exceeds maximum ({max})", + pool=self.poolName, active=self.activeCount, max=self.connectionMax ) return connection @@ -272,6 +276,7 @@ """ self.connectionQueue.put(connection) self.activeCount = len(self.connections) - self.connectionQueue.qsize() + self.ds.poolStats["connection-{}-active".format(self.poolName)] = self.activeCount def failedConnection(self, connection): @@ -279,9 +284,10 @@ A connection has failed; remove it from the list of active connections. A new one will be created if needed. """ - self.ds.poolStats["connection-errors"] += 1 + self.ds.poolStats["connection-{}-errors".format(self.poolName)] += 1 self.connections.remove(connection) self.activeCount = len(self.connections) - self.connectionQueue.qsize() + self.ds.poolStats["connection-{}-active".format(self.poolName)] = self.activeCount def _connect(self): @@ -295,7 +301,10 @@ @raises: L{LDAPConnectionError} if unable to connect. """ - self.log.debug("Connecting to LDAP at {log_source.url}") + self.log.debug( + "[Pool: {pool}] Connecting to LDAP at {url}", + pool=self.poolName, url=self.ds.url + ) connection = self._newConnection() if self.credentials is not None: @@ -306,15 +315,15 @@ self.credentials.password, ) self.log.debug( - "Bound to LDAP as {credentials.username}", - credentials=self.credentials + "[Pool: {pool}] Bound to LDAP as {credentials.username}", + pool=self.poolName, credentials=self.credentials ) except ( ldap.INVALID_CREDENTIALS, ldap.INVALID_DN_SYNTAX ) as e: self.log.error( - "Unable to bind to LDAP as {credentials.username}", - credentials=self.credentials + "[Pool: {pool}] Unable to bind to LDAP as {credentials.username}", + pool=self.poolName, credentials=self.credentials ) raise LDAPBindAuthError( self.credentials.username, e @@ -354,7 +363,10 @@ connection.set_option(option, value) if self.ds._useTLS: - self.log.debug("Starting TLS for {log_source.url}") + self.log.debug( + "[Pool: {pool}] Starting TLS for {url}", + pool=self.poolName, url=self.ds.url + ) connection.start_tls_s() self.connectionsCreated += 1 @@ -394,7 +406,8 @@ extraFilters=None, ownThreadpool=True, threadPoolMax=10, - connectionMax=10, + authConnectionMax=5, + queryConnectionMax=5, tries=3, warningThresholdSeconds=5, _debug=False, @@ -509,11 +522,10 @@ max(threadPoolMax, self.threadpool.max) ) - # Separate pools for LDAP queries and LDAP binds. Note, they each get - # half of connectionMax. + # Separate pools for LDAP queries and LDAP binds. self.connectionPools = { - "query": ConnectionPool("query", self, credentials, connectionMax / 2), - "auth": ConnectionPool("auth", self, None, connectionMax / 2), + "query": ConnectionPool("query", self, credentials, queryConnectionMax), + "auth": ConnectionPool("auth", self, None, authConnectionMax), } self.poolStats = collections.defaultdict(int) @@ -625,6 +637,10 @@ except Exception as e: self.log.error("Unexpected error {error} trying to authenticate {dn}", error=str(e), dn=dn) return False + finally: + # Do an unauthenticated bind on this connection at the end in + # case the server limits the number of concurrent auths by a given user. + connection.simple_bind_s("", "") def _recordsFromQueryString( Modified: twext/trunk/twext/who/ldap/test/test_service.py =================================================================== --- twext/trunk/twext/who/ldap/test/test_service.py 2016-08-03 18:01:01 UTC (rev 15783) +++ twext/trunk/twext/who/ldap/test/test_service.py 2016-08-03 23:20:52 UTC (rev 15784) @@ -499,7 +499,7 @@ actually created is what we expect. """ - service = self.service(connectionMax=4) + service = self.service(authConnectionMax=2) pool = service.connectionPools["auth"] self.assertEquals(0, len(pool.connections)) @@ -508,7 +508,11 @@ # Ask for a connection and check the counts with TestService.Connection(service, "auth"): self.assertEquals(1, len(pool.connections)) + self.assertEquals(1, service.poolStats["connection-auth-active"]) + # When the above context exits, the connection is returned + self.assertEquals(0, service.poolStats["connection-auth-active"]) + self.assertEquals(1, len(pool.connections)) self.assertEquals(1, pool.connectionsCreated) @@ -519,33 +523,40 @@ with TestService.Connection(service, "auth"): self.assertEquals(2, len(pool.connections)) self.assertEquals(2, pool.connectionsCreated) + self.assertEquals(2, service.poolStats["connection-auth-active"]) + self.assertEquals(1, service.poolStats["connection-auth-active"]) + self.assertEquals(0, service.poolStats["connection-auth-active"]) - # Ask for three connections (one more than connectionMax/2) and + # Ask for three connections (one more than connectionMax) and # the third will actually block until the returnConnection( ) call with TestService.Connection(service, "auth"): self.assertEquals(2, len(pool.connections)) self.assertEquals(2, pool.connectionsCreated) - with TestService.Connection(service, "auth") as connection: - self.assertEquals(2, len(pool.connections)) - self.assertEquals(2, pool.connectionsCreated) - # schedule a connection to be returned in 1 second - from twisted.internet import reactor - reactor.callLater(1, pool.returnConnection, connection) + # For the 2nd connection, don't use the context manager so we can + # control when returnConnection happens + connection = pool.getConnection() + self.assertEquals(2, len(pool.connections)) + self.assertEquals(2, pool.connectionsCreated) - # For the third connection, I'm using this method so it gets - # requested in a thread, otherwise we'd hang. - yield service._authenticateUsernamePassword( - u"uid=wsanchez,cn=user,{0}".format(self.baseDN), - u"zehcnasw" - ) + # schedule connection 2 to be returned in 1 second + from twisted.internet import reactor + reactor.callLater(1, pool.returnConnection, connection) - # Proof we bumped up against connection-max: - self.assertEquals(1, service.poolStats["connection-blocked"]) + # For the third connection, I'm using this method so it gets + # requested in a thread, otherwise we'd hang. + yield service._authenticateUsernamePassword( + u"uid=wsanchez,cn=user,{0}".format(self.baseDN), + u"zehcnasw" + ) + # Proof we bumped up against connectionMax: + self.assertEquals(1, service.poolStats["connection-auth-blocked"]) + self.assertEquals(2, len(pool.connections)) self.assertEquals(2, pool.connectionsCreated) - self.assertEquals(2, service.poolStats["connection-max"]) + self.assertEquals(2, service.poolStats["connection-auth-max"]) + self.assertEquals(0, service.poolStats["connection-auth-active"])

1 0

[15783] CalendarServer/trunk/txdav/caldav/datastore
by source_changes＠macosforge.org 03 Aug '16

03 Aug '16

Revision: 15783 http://trac.calendarserver.org//changeset/15783 Author: cdaboo(a)apple.com Date: 2016-08-03 11:01:01 -0700 (Wed, 03 Aug 2016) Log Message: ----------- Make sure illegal characters in attachment file names are stripped. Modified Paths: -------------- CalendarServer/trunk/txdav/caldav/datastore/sql.py CalendarServer/trunk/txdav/caldav/datastore/test/test_attachments.py Modified: CalendarServer/trunk/txdav/caldav/datastore/sql.py =================================================================== --- CalendarServer/trunk/txdav/caldav/datastore/sql.py 2016-08-02 23:37:13 UTC (rev 15782) +++ CalendarServer/trunk/txdav/caldav/datastore/sql.py 2016-08-03 18:01:01 UTC (rev 15783) @@ -4895,6 +4895,11 @@ # Check validity of request yield self._checkValidManagedAttachmentChange() + # Protect against invalid file names + if isinstance(filename, unicode): + filename = filename.encode("utf-8") + filename = filename.translate(None, "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x0B\x0C\x0E\x0F\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1A\x1B\x1C\x1D\x1E\x1F") + # First write the data stream # We need to know the resource_ID of the home collection of the owner @@ -4949,6 +4954,11 @@ # Check validity of request yield self._checkValidManagedAttachmentChange() + # Protect against invalid file names + if isinstance(filename, unicode): + filename = filename.encode("utf-8") + filename = filename.translate(None, "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x0B\x0C\x0E\x0F\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1A\x1B\x1C\x1D\x1E\x1F") + # First check the supplied managed-id is associated with this resource cobjs = (yield ManagedAttachment.referencesTo(self._txn, managed_id)) if self._resourceID not in cobjs: Modified: CalendarServer/trunk/txdav/caldav/datastore/test/test_attachments.py =================================================================== --- CalendarServer/trunk/txdav/caldav/datastore/test/test_attachments.py 2016-08-02 23:37:13 UTC (rev 15782) +++ CalendarServer/trunk/txdav/caldav/datastore/test/test_attachments.py 2016-08-03 18:01:01 UTC (rev 15783) @@ -1495,7 +1495,57 @@ self.assertEquals(data, "new attachment text") + @inlineCallbacks + def test_validFilename(self): + """ + L{CalendarObject.addAttachment} will remove any invalid characters from the supplied file name. + """ + # Create attachment + obj = yield self.calendarObjectUnderTest() + attachment, _ignore_location = yield obj.addAttachment(None, MimeType("text", "x-fixture"), "new\x1F.attachment", MemoryStream("new attachment text")) + self.assertEqual(attachment.name(), "new.attachment") + yield self.commit() + + # Verify parameters exist + obj = yield self.calendarObjectUnderTest() + component = yield obj.componentForUser() + attachments = component.getAllPropertiesInAnyComponent("ATTACH", depth=1,) + self.assertEqual(len(attachments), 1) + attach = attachments[0] + managed_id = attach.parameterValue("MANAGED-ID") + fmttype = attach.parameterValue("FMTTYPE") + filename = attach.parameterValue("FILENAME") + size = attach.parameterValue("SIZE") + + self.assertEqual(fmttype, "text/x-fixture") + self.assertEqual(filename, "new.attachment") + self.assertEqual(int(size), 19) + yield self.commit() + + # Update attachment + obj = yield self.calendarObjectUnderTest() + attachment, _ignore_location = yield obj.updateAttachment(managed_id, MimeType("text", "x-fixture"), "updated\x1F.attachment", MemoryStream("updated attachment text")) + self.assertEqual(attachment.name(), "updated.attachment") + yield self.commit() + + # Verify parameters exist + obj = yield self.calendarObjectUnderTest() + component = yield obj.componentForUser() + attachments = component.getAllPropertiesInAnyComponent("ATTACH", depth=1,) + self.assertEqual(len(attachments), 1) + attach = attachments[0] + managed_id = attach.parameterValue("MANAGED-ID") + fmttype = attach.parameterValue("FMTTYPE") + filename = attach.parameterValue("FILENAME") + size = attach.parameterValue("SIZE") + + self.assertEqual(fmttype, "text/x-fixture") + self.assertEqual(filename, "updated.attachment") + self.assertEqual(int(size), 23) + yield self.commit() + + now = DateTime.getToday().getYear() PLAIN_ICS = """BEGIN:VCALENDAR

1 0

[15782] twext/trunk/twext/who/ldap/test/test_service.py
by source_changes＠macosforge.org 02 Aug '16

02 Aug '16

Revision: 15782 http://trac.calendarserver.org//changeset/15782 Author: sagen(a)apple.com Date: 2016-08-02 16:37:13 -0700 (Tue, 02 Aug 2016) Log Message: ----------- Let's finish that thought... Modified Paths: -------------- twext/trunk/twext/who/ldap/test/test_service.py Modified: twext/trunk/twext/who/ldap/test/test_service.py =================================================================== --- twext/trunk/twext/who/ldap/test/test_service.py 2016-08-02 23:24:53 UTC (rev 15781) +++ twext/trunk/twext/who/ldap/test/test_service.py 2016-08-02 23:37:13 UTC (rev 15782) @@ -495,8 +495,8 @@ def test_auth_pool(self): """ Verify acquiring connections from the LDAP connection pool will block - when connectionMax is reached, - and that + when connectionMax is reached, and that the number of connections + actually created is what we expect. """ service = self.service(connectionMax=4)

1 0

[15781] CalendarServer/trunk/requirements-cs.txt
by source_changes＠macosforge.org 02 Aug '16

02 Aug '16

Revision: 15781 http://trac.calendarserver.org//changeset/15781 Author: sagen(a)apple.com Date: 2016-08-02 16:24:53 -0700 (Tue, 02 Aug 2016) Log Message: ----------- Get latest twext with LDAP connection pool for binds Modified Paths: -------------- CalendarServer/trunk/requirements-cs.txt Modified: CalendarServer/trunk/requirements-cs.txt =================================================================== --- CalendarServer/trunk/requirements-cs.txt 2016-08-02 23:24:05 UTC (rev 15780) +++ CalendarServer/trunk/requirements-cs.txt 2016-08-02 23:24:53 UTC (rev 15781) @@ -7,7 +7,7 @@ zope.interface==4.1.3 setuptools==18.5 - --editable svn+http://svn.calendarserver.org/repository/calendarserver/twext/trunk@157… + --editable svn+http://svn.calendarserver.org/repository/calendarserver/twext/trunk@157… cffi==1.7.0 pycparser==2.14 #twisted

1 0

[15780] twext/trunk/twext/who/ldap
by source_changes＠macosforge.org 02 Aug '16

02 Aug '16

Revision: 15780 http://trac.calendarserver.org//changeset/15780 Author: sagen(a)apple.com Date: 2016-08-02 16:24:05 -0700 (Tue, 02 Aug 2016) Log Message: ----------- Use a connection pool for the LDAP bind calls we use for authentication (in addition to the already existing pool used for other LDAP queries) Modified Paths: -------------- twext/trunk/twext/who/ldap/_service.py twext/trunk/twext/who/ldap/test/test_service.py Modified: twext/trunk/twext/who/ldap/_service.py =================================================================== --- twext/trunk/twext/who/ldap/_service.py 2016-07-27 21:50:59 UTC (rev 15779) +++ twext/trunk/twext/who/ldap/_service.py 2016-08-02 23:24:05 UTC (rev 15780) @@ -203,7 +203,165 @@ }) +class ConnectionPool(object): + log = Logger() + + def __init__(self, poolName, ds, credentials, connectionMax): + self.poolName = poolName + self.ds = ds + self.credentials = credentials + self.connectionQueue = Queue() + self.connectionCreateLock = RLock() + self.connections = [] + self.activeCount = 0 + self.connectionsCreated = 0 + self.connectionMax = connectionMax + + def getConnection(self): + """ + Get a connection from the connection pool. + This will retrieve a connection from the connection pool L{Queue} + object. + If the L{Queue} is empty, it will check to see whether a new connection + can be created (based on the connection limit), and if so create that + and use it. + If no new connections can be created, it will block on the L{Queue} + until an existing, in-use, connection is put back. + """ + try: + connection = self.connectionQueue.get(block=False) + except Empty: + # Note we use a lock here to prevent a race condition in which + # multiple requests for a new connection could succeed even though + # the connection counts starts out one less than the maximum. + # This can happen because self._connect() can take a while. + self.connectionCreateLock.acquire() + if len(self.connections) < self.connectionMax: + connection = self._connect() + self.connections.append(connection) + self.connectionCreateLock.release() + else: + self.connectionCreateLock.release() + self.ds.poolStats["connection-blocked"] += 1 + connection = self.connectionQueue.get() + + + connectionID = "connection-{}".format( + self.connections.index(connection) + ) + + self.ds.poolStats[connectionID] += 1 + self.activeCount = len(self.connections) - self.connectionQueue.qsize() + self.ds.poolStats["connection-active"] = self.activeCount + self.ds.poolStats["connection-max"] = max( + self.ds.poolStats["connection-max"], self.activeCount + ) + + if self.activeCount > self.connectionMax: + self.log.error( + "Active LDAP connections ({active}) exceeds maximum ({max})", + active=self.activeCount, max=self.connectionMax + ) + return connection + + + def returnConnection(self, connection): + """ + A connection is no longer needed - return it to the pool. + """ + self.connectionQueue.put(connection) + self.activeCount = len(self.connections) - self.connectionQueue.qsize() + + + def failedConnection(self, connection): + """ + A connection has failed; remove it from the list of active connections. + A new one will be created if needed. + """ + self.ds.poolStats["connection-errors"] += 1 + self.connections.remove(connection) + self.activeCount = len(self.connections) - self.connectionQueue.qsize() + + + def _connect(self): + """ + Connect to the directory server. + This will always be called in a thread to prevent blocking. + + @returns: The connection object. + @rtype: L{ldap.ldapobject.LDAPObject} + + @raises: L{LDAPConnectionError} if unable to connect. + """ + + self.log.debug("Connecting to LDAP at {log_source.url}") + connection = self._newConnection() + + if self.credentials is not None: + if IUsernamePassword.providedBy(self.credentials): + try: + connection.simple_bind_s( + self.credentials.username, + self.credentials.password, + ) + self.log.debug( + "Bound to LDAP as {credentials.username}", + credentials=self.credentials + ) + except ( + ldap.INVALID_CREDENTIALS, ldap.INVALID_DN_SYNTAX + ) as e: + self.log.error( + "Unable to bind to LDAP as {credentials.username}", + credentials=self.credentials + ) + raise LDAPBindAuthError( + self.credentials.username, e + ) + + else: + raise LDAPConnectionError( + "Unknown credentials type: {0}" + .format(self.credentials) + ) + + return connection + + + def _newConnection(self): + """ + Create a new LDAP connection and initialize and start TLS if required. + This will always be called in a thread to prevent blocking. + + @returns: The connection object. + @rtype: L{ldap.ldapobject.LDAPObject} + + @raises: L{LDAPConnectionError} if unable to connect. + """ + connection = ldap.initialize(self.ds.url) + + # FIXME: Use trace_file option to wire up debug logging when + # Twisted adopts the new logging stuff. + + for option, value in ( + (ldap.OPT_TIMEOUT, self.ds._timeout), + (ldap.OPT_X_TLS_CACERTFILE, self.ds._tlsCACertificateFile), + (ldap.OPT_X_TLS_CACERTDIR, self.ds._tlsCACertificateDirectory), + (ldap.OPT_DEBUG_LEVEL, self.ds._debug), + ): + if value is not None: + connection.set_option(option, value) + + if self.ds._useTLS: + self.log.debug("Starting TLS for {log_source.url}") + connection.start_tls_s() + + self.connectionsCreated += 1 + + return connection + + # # Directory Service # @@ -350,12 +508,14 @@ self.threadpool.adjustPoolsize( max(threadPoolMax, self.threadpool.max) ) - self.connectionMax = connectionMax - self.connectionCreateLock = RLock() - self.connections = [] - self.connectionQueue = Queue() + + # Separate pools for LDAP queries and LDAP binds. Note, they each get + # half of connectionMax. + self.connectionPools = { + "query": ConnectionPool("query", self, credentials, connectionMax / 2), + "auth": ConnectionPool("auth", self, None, connectionMax / 2), + } self.poolStats = collections.defaultdict(int) - self.activeCount = 0 reactor.callWhenRunning(self.start) reactor.addSystemEventTrigger("during", "shutdown", self.stop) @@ -396,164 +556,23 @@ needed. """ - def __init__(self, ds): - self.ds = ds + def __init__(self, ds, poolName): + self.pool = ds.connectionPools[poolName] def __enter__(self): - self.connection = self.ds._getConnection() + self.connection = self.pool.getConnection() return self.connection def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: - self.ds._returnConnection(self.connection) + self.pool.returnConnection(self.connection) return True else: - self.ds._failedConnection(self.connection) + self.pool.failedConnection(self.connection) return False - def _getConnection(self): - """ - Get a connection from the connection pool. - This will retrieve a connection from the connection pool L{Queue} - object. - If the L{Queue} is empty, it will check to see whether a new connection - can be created (based on the connection limit), and if so create that - and use it. - If no new connections can be created, it will block on the L{Queue} - until an existing, in-use, connection is put back. - """ - try: - connection = self.connectionQueue.get(block=False) - except Empty: - # Note we use a lock here to prevent a race condition in which - # multiple requests for a new connection could succeed even though - # the connection counts starts out one less than the maximum. - # This can happen because self._connect() can take a while. - self.connectionCreateLock.acquire() - if len(self.connections) < self.connectionMax: - connection = self._connect() - self.connections.append(connection) - self.connectionCreateLock.release() - else: - self.connectionCreateLock.release() - self.poolStats["connection-blocked"] += 1 - connection = self.connectionQueue.get() - - connectionID = "connection-{}".format( - self.connections.index(connection) - ) - - self.poolStats[connectionID] += 1 - self.activeCount = len(self.connections) - self.connectionQueue.qsize() - self.poolStats["connection-active"] = self.activeCount - self.poolStats["connection-max"] = max( - self.poolStats["connection-max"], self.activeCount - ) - - if self.activeCount > self.connectionMax: - self.log.error( - "Active LDAP connections ({active}) exceeds maximum ({max})", - active=self.activeCount, max=self.connectionMax - ) - return connection - - - def _returnConnection(self, connection): - """ - A connection is no longer needed - return it to the pool. - """ - self.connectionQueue.put(connection) - self.activeCount = len(self.connections) - self.connectionQueue.qsize() - - - def _failedConnection(self, connection): - """ - A connection has failed; remove it from the list of active connections. - A new one will be created if needed. - """ - self.poolStats["connection-errors"] += 1 - self.connections.remove(connection) - self.activeCount = len(self.connections) - self.connectionQueue.qsize() - - - def _connect(self): - """ - Connect to the directory server. - This will always be called in a thread to prevent blocking. - - @returns: The connection object. - @rtype: L{ldap.ldapobject.LDAPObject} - - @raises: L{LDAPConnectionError} if unable to connect. - """ - - self.log.debug("Connecting to LDAP at {log_source.url}") - connection = self._newConnection() - - if self._credentials is not None: - if IUsernamePassword.providedBy(self._credentials): - try: - connection.simple_bind_s( - self._credentials.username, - self._credentials.password, - ) - self.log.debug( - "Bound to LDAP as {credentials.username}", - credentials=self._credentials - ) - except ( - ldap.INVALID_CREDENTIALS, ldap.INVALID_DN_SYNTAX - ) as e: - self.log.error( - "Unable to bind to LDAP as {credentials.username}", - credentials=self._credentials - ) - raise LDAPBindAuthError( - self._credentials.username, e - ) - - else: - raise LDAPConnectionError( - "Unknown credentials type: {0}" - .format(self._credentials) - ) - - return connection - - - def _newConnection(self): - """ - Create a new LDAP connection and initialize and start TLS if required. - This will always be called in a thread to prevent blocking. - - @returns: The connection object. - @rtype: L{ldap.ldapobject.LDAPObject} - - @raises: L{LDAPConnectionError} if unable to connect. - """ - connection = ldap.initialize(self.url) - - # FIXME: Use trace_file option to wire up debug logging when - # Twisted adopts the new logging stuff. - - for option, value in ( - (ldap.OPT_TIMEOUT, self._timeout), - (ldap.OPT_X_TLS_CACERTFILE, self._tlsCACertificateFile), - (ldap.OPT_X_TLS_CACERTDIR, self._tlsCACertificateDirectory), - (ldap.OPT_DEBUG_LEVEL, self._debug), - ): - if value is not None: - connection.set_option(option, value) - - if self._useTLS: - self.log.debug("Starting TLS for {log_source.url}") - connection.start_tls_s() - - return connection - - def _authenticateUsernamePassword(self, dn, password): """ Open a secondary connection to the LDAP server and try binding to it @@ -587,30 +606,27 @@ @raises: L{LDAPConnectionError} if unable to connect. """ self.log.debug("Authenticating {dn}", dn=dn) - connection = self._newConnection() + with DirectoryService.Connection(self, "auth") as connection: + try: + connection.simple_bind_s(dn, password) + self.log.debug("Authenticated {dn}", dn=dn) + return True + except ( + ldap.INAPPROPRIATE_AUTH, + ldap.INVALID_CREDENTIALS, + ldap.INVALID_DN_SYNTAX, + ): + self.log.debug("Unable to authenticate {dn}", dn=dn) + return False + except ldap.CONSTRAINT_VIOLATION: + self.log.info("Account locked {dn}", dn=dn) + return False + except Exception as e: + self.log.error("Unexpected error {error} trying to authenticate {dn}", error=str(e), dn=dn) + return False - try: - connection.simple_bind_s(dn, password) - self.log.debug("Authenticated {dn}", dn=dn) - return True - except ( - ldap.INAPPROPRIATE_AUTH, - ldap.INVALID_CREDENTIALS, - ldap.INVALID_DN_SYNTAX, - ): - self.log.debug("Unable to authenticate {dn}", dn=dn) - return False - except ldap.CONSTRAINT_VIOLATION: - self.log.info("Account locked {dn}", dn=dn) - return False - except Exception as e: - self.log.error("Unexpected error {error} trying to authenticate {dn}", error=str(e), dn=dn) - return False - finally: - connection.unbind() - def _recordsFromQueryString( self, queryString, recordTypes=None, limitResults=None, timeoutSeconds=None @@ -654,7 +670,7 @@ try: - with DirectoryService.Connection(self) as connection: + with DirectoryService.Connection(self, "query") as connection: for recordType in recordTypes: @@ -836,7 +852,7 @@ try: - with DirectoryService.Connection(self) as connection: + with DirectoryService.Connection(self, "query") as connection: self.log.debug("Performing LDAP DN query: {dn}", dn=dn) Modified: twext/trunk/twext/who/ldap/test/test_service.py =================================================================== --- twext/trunk/twext/who/ldap/test/test_service.py 2016-07-27 21:50:59 UTC (rev 15779) +++ twext/trunk/twext/who/ldap/test/test_service.py 2016-08-02 23:24:05 UTC (rev 15780) @@ -317,28 +317,27 @@ class DirectoryServiceConnectionTestMixIn(object): - @inlineCallbacks + def test_connect_defaults(self): """ Connect with default arguments. """ service = self.service() - connection = yield service._connect() + with TestService.Connection(service, "query") as connection: - self.assertEquals(connection.methods_called(), ["initialize"]) + self.assertEquals(connection.methods_called(), ["initialize"]) - for option in ( - ldap.OPT_TIMEOUT, - ldap.OPT_X_TLS_CACERTFILE, - ldap.OPT_X_TLS_CACERTDIR, - ldap.OPT_DEBUG_LEVEL, - ): - self.assertRaises(KeyError, connection.get_option, option) + for option in ( + ldap.OPT_TIMEOUT, + ldap.OPT_X_TLS_CACERTFILE, + ldap.OPT_X_TLS_CACERTDIR, + ldap.OPT_DEBUG_LEVEL, + ): + self.assertRaises(KeyError, connection.get_option, option) - self.assertFalse(connection.tls_enabled) + self.assertFalse(connection.tls_enabled) - @inlineCallbacks def test_connect_withUsernamePassword_invalid(self): """ Connect with UsernamePassword credentials. @@ -349,14 +348,14 @@ ) service = self.service(credentials=credentials) try: - yield service._connect() + with TestService.Connection(service, "query"): + pass except LDAPBindAuthError: pass else: self.fail("Should have raised LDAPBindAuthError") - @inlineCallbacks def test_connect_withUsernamePassword_valid(self): """ Connect with UsernamePassword credentials. @@ -366,15 +365,14 @@ u"zehcnasw" ) service = self.service(credentials=credentials) - connection = yield service._connect() - self.assertEquals( - connection.methods_called(), - ["initialize", "simple_bind_s"] - ) + with TestService.Connection(service, "query") as connection: + self.assertEquals( + connection.methods_called(), + ["initialize", "simple_bind_s"] + ) - @inlineCallbacks def test_connect_withOptions(self): """ Connect with default arguments. @@ -385,43 +383,42 @@ tlsCACertificateDirectory=FilePath("/path/to/certdir"), _debug=True, ) - connection = yield service._connect() + with TestService.Connection(service, "query") as connection: - self.assertEquals( - connection.methods_called(), - [ - "initialize", - "set_option", "set_option", "set_option", "set_option", - ] - ) + self.assertEquals( + connection.methods_called(), + [ + "initialize", + "set_option", "set_option", "set_option", "set_option", + ] + ) - opt = lambda k: connection.get_option(k) + opt = lambda k: connection.get_option(k) - self.assertEquals(opt(ldap.OPT_TIMEOUT), 18) - self.assertEquals(opt(ldap.OPT_X_TLS_CACERTFILE), "/path/to/cert") - self.assertEquals(opt(ldap.OPT_X_TLS_CACERTDIR), "/path/to/certdir") - self.assertEquals(opt(ldap.OPT_DEBUG_LEVEL), 255) + self.assertEquals(opt(ldap.OPT_TIMEOUT), 18) + self.assertEquals(opt(ldap.OPT_X_TLS_CACERTFILE), "/path/to/cert") + self.assertEquals(opt(ldap.OPT_X_TLS_CACERTDIR), "/path/to/certdir") + self.assertEquals(opt(ldap.OPT_DEBUG_LEVEL), 255) - # Tested in test_connect_defaults, but test again here since we're - # setting SSL options and we want to be sure they don't somehow enable - # SSL implicitly. - self.assertFalse(connection.tls_enabled) + # Tested in test_connect_defaults, but test again here since we're + # setting SSL options and we want to be sure they don't somehow enable + # SSL implicitly. + self.assertFalse(connection.tls_enabled) - @inlineCallbacks def test_connect_withTLS(self): """ Connect with TLS enabled. """ service = self.service(useTLS=True) - connection = yield service._connect() - self.assertEquals( - connection.methods_called(), - ["initialize", "start_tls_s"] - ) + with TestService.Connection(service, "query") as connection: + self.assertEquals( + connection.methods_called(), + ["initialize", "start_tls_s"] + ) - self.assertTrue(connection.tls_enabled) + self.assertTrue(connection.tls_enabled) @@ -457,11 +454,11 @@ # still have a connection in the pool service._recordsFromQueryString_inThread("(this=that)") self.assertEquals(service._retryNumber, 0) - self.assertEquals(len(service.connections), 1) + self.assertEquals(len(service.connectionPools["query"].connections), 1) service._recordWithDN_inThread("cn=test") self.assertEquals(service._retryNumber, 0) - self.assertEquals(len(service.connections), 1) + self.assertEquals(len(service.connectionPools["query"].connections), 1) # Force a search to raise SERVER_DOWN def raiseServerDown(*args, **kwds): @@ -479,7 +476,7 @@ self.fail("Should have raised LDAPQueryError") # Verify the connections are all closed - self.assertEquals(len(service.connections), 0) + self.assertEquals(len(service.connectionPools["query"].connections), 0) # Now try recordWithDN try: @@ -491,10 +488,67 @@ self.fail("Should have raised LDAPQueryError") # Verify the connections are all closed - self.assertEquals(len(service.connections), 0) + self.assertEquals(len(service.connectionPools["query"].connections), 0) + @inlineCallbacks + def test_auth_pool(self): + """ + Verify acquiring connections from the LDAP connection pool will block + when connectionMax is reached, + and that + """ + service = self.service(connectionMax=4) + pool = service.connectionPools["auth"] + + self.assertEquals(0, len(pool.connections)) + self.assertEquals(0, pool.connectionsCreated) + + # Ask for a connection and check the counts + with TestService.Connection(service, "auth"): + self.assertEquals(1, len(pool.connections)) + + self.assertEquals(1, len(pool.connections)) + self.assertEquals(1, pool.connectionsCreated) + + # Ask for two connections and check the counts + with TestService.Connection(service, "auth"): + self.assertEquals(1, len(pool.connections)) + self.assertEquals(1, pool.connectionsCreated) + with TestService.Connection(service, "auth"): + self.assertEquals(2, len(pool.connections)) + self.assertEquals(2, pool.connectionsCreated) + + # Ask for three connections (one more than connectionMax/2) and + # the third will actually block until the returnConnection( ) call + with TestService.Connection(service, "auth"): + self.assertEquals(2, len(pool.connections)) + self.assertEquals(2, pool.connectionsCreated) + with TestService.Connection(service, "auth") as connection: + self.assertEquals(2, len(pool.connections)) + self.assertEquals(2, pool.connectionsCreated) + + # schedule a connection to be returned in 1 second + from twisted.internet import reactor + reactor.callLater(1, pool.returnConnection, connection) + + # For the third connection, I'm using this method so it gets + # requested in a thread, otherwise we'd hang. + yield service._authenticateUsernamePassword( + u"uid=wsanchez,cn=user,{0}".format(self.baseDN), + u"zehcnasw" + ) + + # Proof we bumped up against connection-max: + self.assertEquals(1, service.poolStats["connection-blocked"]) + + self.assertEquals(2, len(pool.connections)) + self.assertEquals(2, pool.connectionsCreated) + self.assertEquals(2, service.poolStats["connection-max"]) + + + class ExtraFiltersTest(BaseTestCase, unittest.TestCase): def test_extraFilters(self):

1 0