diff --git a/Lib/test/test_ftplib.py b/Lib/test/test_ftplib.py
index 29f7f7d..ab03faf 100644
--- a/Lib/test/test_ftplib.py
+++ b/Lib/test/test_ftplib.py
@@ -28,6 +28,7 @@
 
 
 class DummyDTPHandler(asynchat.async_chat):
+    dtp_conn_closed = False
 
     def __init__(self, conn, baseclass):
         asynchat.async_chat.__init__(self, conn)
@@ -38,8 +39,13 @@
         self.baseclass.last_received_data += self.recv(1024).decode('ascii')
 
     def handle_close(self):
-        self.baseclass.push('226 transfer complete')
-        self.close()
+        # XXX: this method can be called many times in a row for a single
+        # connection, including in clear-text (non-TLS) mode.
+        # (behaviour witnessed with test_data_connection)
+        if not self.dtp_conn_closed:
+            self.baseclass.push('226 transfer complete')
+            self.close()
+            self.dtp_conn_closed = True
 
     def push(self, what):
         super(DummyDTPHandler, self).push(what.encode('ascii'))
@@ -254,6 +260,7 @@
         """An asyncore.dispatcher subclass supporting TLS/SSL."""
 
         _ssl_accepting = False
+        _ssl_closing = False
 
         def secure_connection(self):
             self.del_channel()
@@ -280,15 +287,36 @@
             else:
                 self._ssl_accepting = False
 
+        def _do_ssl_shutdown(self):
+            self._ssl_closing = True
+            try:
+                self.socket = self.socket.unwrap()
+            except ssl.SSLError as err:
+                if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
+                                   ssl.SSL_ERROR_WANT_WRITE):
+                    return
+            except socket.error as err:
+                # Any "socket error" corresponds to a SSL_ERROR_SYSCALL return
+                # from OpenSSL's SSL_shutdown(), corresponding to a
+                # closed socket condition. See also:
+                # http://www.mail-archive.com/openssl-users@openssl.org/msg60710.html
+                pass
+            self._ssl_closing = False
+            super(SSLConnection, self).close()
+
         def handle_read_event(self):
             if self._ssl_accepting:
                 self._do_ssl_handshake()
+            elif self._ssl_closing:
+                self._do_ssl_shutdown()
             else:
                 super(SSLConnection, self).handle_read_event()
 
         def handle_write_event(self):
             if self._ssl_accepting:
                 self._do_ssl_handshake()
+            elif self._ssl_closing:
+                self._do_ssl_shutdown()
             else:
                 super(SSLConnection, self).handle_write_event()
 
@@ -308,7 +336,7 @@
             except ssl.SSLError as err:
                 if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
                                    ssl.SSL_ERROR_WANT_WRITE):
-                    return ''
+                    return b''
                 if err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN):
                     self.handle_close()
                     return b''
@@ -318,12 +346,9 @@
             raise
 
         def close(self):
-            try:
-                if isinstance(self.socket, ssl.SSLSocket):
-                    if self.socket._sslobj is not None:
-                        self.socket.unwrap()
-            finally:
-                super(SSLConnection, self).close()
+            if (isinstance(self.socket, ssl.SSLSocket) and
+                self.socket._sslobj is not None):
+                self._do_ssl_shutdown()
 
 
     class DummyTLS_DTPHandler(SSLConnection, DummyDTPHandler):
@@ -606,21 +631,21 @@
         sock = self.client.transfercmd('list')
         self.assertNotIsInstance(sock, ssl.SSLSocket)
         sock.close()
-        self.client.voidresp()
+        self.assertEqual(self.client.voidresp(), "226 transfer complete")
 
         # secured, after PROT P
         self.client.prot_p()
         sock = self.client.transfercmd('list')
         self.assertIsInstance(sock, ssl.SSLSocket)
         sock.close()
-        self.client.voidresp()
+        self.assertEqual(self.client.voidresp(), "226 transfer complete")
 
         # PROT C is issued, the connection must be in cleartext again
         self.client.prot_c()
         sock = self.client.transfercmd('list')
         self.assertNotIsInstance(sock, ssl.SSLSocket)
         sock.close()
-        self.client.voidresp()
+        self.assertEqual(self.client.voidresp(), "226 transfer complete")
 
     def test_login(self):
         # login() is supposed to implicitly secure the control connection
diff --git a/Misc/NEWS b/Misc/NEWS
index a8cc594..c65f8ab 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -329,6 +329,11 @@
 Library
 -------
 
+- Issue #8108: Fix the unwrap() method of SSL objects when the socket has
+  a non-infinite timeout.  Also make that method friendlier with applications
+  wanting to continue using the socket in clear-text mode, by disabling
+  OpenSSL's internal readahead.  Thanks to Darryl Miles for guidance.
+
 - Issue #8496: make mailcap.lookup() always return a list, rather than an
   iterator.  Patch by Gregory Nofi.
 
@@ -1120,6 +1125,9 @@
 Tests
 -----
 
+- Issue #8108: test_ftplib's non-blocking SSL server now has proper handling
+  of SSL shutdowns.
+
 - Issues #8279, #8330, #8437, #8480: Fix test_gdb failures, patch written by
   Dave Malcolm
 
diff --git a/Modules/_ssl.c b/Modules/_ssl.c
index 3232df5..e06b7e0 100644
--- a/Modules/_ssl.c
+++ b/Modules/_ssl.c
@@ -9,6 +9,9 @@
    directly.
 
    XXX should partial writes be enabled, SSL_MODE_ENABLE_PARTIAL_WRITE?
+
+   XXX integrate several "shutdown modes" as suggested in
+       http://bugs.python.org/issue8108#msg102867 ?
 */
 
 #include "Python.h"
@@ -116,6 +119,7 @@
 	SSL_CTX*	ctx;
 	SSL*		ssl;
 	X509*		peer_cert;
+	int		shutdown_seen_zero;
 
 } PySSLObject;
 
@@ -1392,7 +1396,8 @@
 
 static PyObject *PySSL_SSLshutdown(PySSLObject *self)
 {
-	int err;
+	int err, ssl_err, sockstate, nonblocking;
+	int zeros = 0;
         PySocketSockObject *sock
           = (PySocketSockObject *) PyWeakref_GetObject(self->Socket);
 
@@ -1403,13 +1408,65 @@
                 return NULL;
         }
 
-	PySSL_BEGIN_ALLOW_THREADS
-	err = SSL_shutdown(self->ssl);
-	if (err == 0) {
-		/* we need to call it again to finish the shutdown */
+        /* Just in case the blocking state of the socket has been changed */
+	nonblocking = (sock->sock_timeout >= 0.0);
+	BIO_set_nbio(SSL_get_rbio(self->ssl), nonblocking);
+	BIO_set_nbio(SSL_get_wbio(self->ssl), nonblocking);
+
+	while (1) {
+		PySSL_BEGIN_ALLOW_THREADS
+		/* Disable read-ahead so that unwrap can work correctly.
+		 * Otherwise OpenSSL might read in too much data,
+		 * eating clear text data that happens to be
+		 * transmitted after the SSL shutdown.
+		 * Should be safe to call repeatedly everytime this
+		 * function is used and the shutdown_seen_zero != 0
+		 * condition is met.
+		 */
+		if (self->shutdown_seen_zero)
+			SSL_set_read_ahead(self->ssl, 0);
 		err = SSL_shutdown(self->ssl);
+		PySSL_END_ALLOW_THREADS
+		/* If err == 1, a secure shutdown with SSL_shutdown() is complete */
+		if (err > 0)
+			break;
+		if (err == 0) {
+			/* Don't loop endlessly; instead preserve legacy
+			   behaviour of trying SSL_shutdown() only twice.
+			   This looks necessary for OpenSSL < 0.9.8m */
+			if (++zeros > 1)
+				break;
+			/* Shutdown was sent, now try receiving */
+			self->shutdown_seen_zero = 1;
+			continue;
+		}
+
+		/* Possibly retry shutdown until timeout or failure */
+		ssl_err = SSL_get_error(self->ssl, err);
+		if (ssl_err == SSL_ERROR_WANT_READ)
+			sockstate = check_socket_and_wait_for_timeout(sock, 0);
+		else if (ssl_err == SSL_ERROR_WANT_WRITE)
+			sockstate = check_socket_and_wait_for_timeout(sock, 1);
+		else
+			break;
+		if (sockstate == SOCKET_HAS_TIMED_OUT) {
+			if (ssl_err == SSL_ERROR_WANT_READ)
+				PyErr_SetString(PySSLErrorObject,
+		                                "The read operation timed out");
+			else
+				PyErr_SetString(PySSLErrorObject,
+		                                "The write operation timed out");
+			return NULL;
+		}
+		else if (sockstate == SOCKET_TOO_LARGE_FOR_SELECT) {
+			PyErr_SetString(PySSLErrorObject,
+	                                "Underlying socket too large for select().");
+			return NULL;
+		}
+		else if (sockstate != SOCKET_OPERATION_OK)
+			/* Retain the SSL error code */
+			break;
 	}
-	PySSL_END_ALLOW_THREADS
 
 	if (err < 0)
 		return PySSL_SetError(self, err, __FILE__, __LINE__);