Fix messages are one-behind when using secure sockets (wss)

test/server.crt

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDazCCAlOgAwIBAgIUHVokn0Hj/tV1U9AE749qXVsncdgwDQYJKoZIhvcNAQEL
+BQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAeFw0yNTA1MTMxNDA3MjFaFw0yNjA1
+MTMxNDA3MjFaMEUxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEw
+HwYDVQQKDBhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQwggEiMA0GCSqGSIb3DQEB
+AQUAA4IBDwAwggEKAoIBAQDyRq07rTRIY8oAyoF2hqzOntANk9ksJ+YCvbykcssd
+kOlLwT795KDfFXBZMYIL9Iz7IBCdoCZSp1YphQh5HsnhkNNFKeUC3F1PDS8VFjPt
+hor84q+emfDO/EP9Pq/ZFgcoTAaucgIkVRkCtUuPKhfWDF8yISnCthmGoJHh4yNu
+rja3LgecBHu3Wj6mft6QTAQ8538LnsltyFg2TxtXuuHN06fHnqeWdlrEs2Dp0gJT
+yOQky/RSxJo4Wb2Wts+eeDJF85dRgdSwjV5FCQcA7hRXX0wAIQu+wbNIjc1mnIC9
+qbLxsa1Y65HDOuub4+XqSTPHYyQD2913JuICDaYR3CV1AgMBAAGjUzBRMB0GA1Ud
+DgQWBBSP5ZeK+PdQaXoKqlYr15ZjOfVDHzAfBgNVHSMEGDAWgBSP5ZeK+PdQaXoK
+qlYr15ZjOfVDHzAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4IBAQDQ
+GqC13PKawdUNaEhXoSgsywsFlSOU5pBIqiQGytpFOFKohHOOySTySb2rHAoCdHQB
+428c9IpwIrCa2ahkXkLaiVL5HVg9ZDduuQBw4vKUXexJUy6muZo5Ov8HtQLHAtUF
+iOEoi/MkMja/xLCDxEA6d4VNwfCFX9vbLcI12oioIZZHzpUdNyTs49Cwf/PECwUL
+FAY87y75H4rjDJvsTeOuxFIe1xRF7Ik9V3C1ef9k1GW341cWM5yQIzHSiJOzKDsb
+CkiF5w74QTujth2t/zdAjNZbLeMNYcgPNB4mr6iaD4RJgZWRwlRHmyQMDPuKxDEc
+Km9w57IdYygjyw5jOx5f
+-----END CERTIFICATE-----

test/server.key

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDyRq07rTRIY8oA
+yoF2hqzOntANk9ksJ+YCvbykcssdkOlLwT795KDfFXBZMYIL9Iz7IBCdoCZSp1Yp
+hQh5HsnhkNNFKeUC3F1PDS8VFjPthor84q+emfDO/EP9Pq/ZFgcoTAaucgIkVRkC
+tUuPKhfWDF8yISnCthmGoJHh4yNurja3LgecBHu3Wj6mft6QTAQ8538LnsltyFg2
+TxtXuuHN06fHnqeWdlrEs2Dp0gJTyOQky/RSxJo4Wb2Wts+eeDJF85dRgdSwjV5F
+CQcA7hRXX0wAIQu+wbNIjc1mnIC9qbLxsa1Y65HDOuub4+XqSTPHYyQD2913JuIC
+DaYR3CV1AgMBAAECggEAHYenzcJKwRgIoxgLt5qqrXSF/2Gp8svaKTNfLtwfDbd/
++A/R0bhwM0C1tOln5HUmSeWaoNvIUAK9acohQkIScT/pwGBe3X5mkSAWQQe3xJfF
+kRVAOqCgzVnKH6/oVxlsPekmV1TmFe+ZYM8gKo8C4MAZSk7ofCcd7V7c6R96Th8J
+LBiMteUCJbUqO8HxWrDX2BuRei74vJNysdACD/21qBnLPqXjqgnp4H5M1rQl8x6G
+OrLdkCQklwb3s+4hioaHbd0i4fvbYeaf/F5ouw6Tqd1zQMZ9bMn/dkeuOR2R1frI
+K3TTQHxpC1RqZJAU5TXrp0xlC3VXEAlpmoNa7eaYAQKBgQD9W30SossLqle5i1cJ
+Zh58e8Ag3xWrz/OVJQ+PlcS/hweEbuw/MOWXQZqYUznPK89Hd9BXtV311G2GF6zQ
+AV6RzTz8UmOmZD343hlite7IJTbUeqYyz6AYPEOLayVHiHZByAMJkr5x2+Wae6TT
+RkdyNLPwIbFijRZk7be1S2r1gQKBgQD0zZlX/OMI5a1GIlQmzY13Dx0Pyta05d6H
+KUMC2Svr/kIYkGyY2xE3EHy7GIzFD/zm3B9eeTIwJYHbniGJ6YAeNzBdPG8nk2R8
+w+1Mj/yliCOwHG89VO830Xovf0EIH+a20+PAtByqvoV2RTbzMS88mTh8URqhtmM/
+jK1zG0Gx9QKBgQDlDH4xh+2LOVAf1YI1ZBYxsmtLDIP6FYGAl8XOqLb79GZuax24
+L0uRiGTsS2mbC19UnFRFxxkQMyFlNigs0OAfbm4xK4cdmciRIrHOlO4wEbzVMaDp
+lN2Gq4zhEVfdqNhItjtQv1Lfes7D7/5eZ04WSOFYOg21LBpP2r3X8DvdgQKBgQCo
+CMxeKhbJD6ZdgsjSjbux4qznHys7pqGVk0wNE3bjmXZTGCeC0LRDYMzNPC+8QJou
++R+LIJPDmqtFTYjl+mJX2zgWd5owxyptvasQJ7GbChS9GPd+WOOPI/nDyoygAA3E
+pzMpHjijNv2zThVG3xb2eJHeO2mVYPVFNNIGNcplVQKBgH6tNsGlVZDzz6SuOPsT
++3ekTghz52X0BcYtCZVetCdMUkGjA6IhIpW/YNGDS07P4MpONlguThEOOVZJziMg
+i0lNFpWThsslsoW+2Os1B3Aim+Tkv7pRPlAVLl10w8Xy0mKW+RU7mtn4kfljSnMt
+NGqZSCK3d2AXzjO/axo/xYjT
+-----END PRIVATE KEY-----

test/test_cherrypy_ssl.py

@@ -0,0 +1,194 @@
+# -*- coding: utf-8 -*-
+"""Tests for the CherryPy and ws4py libraries with SSL support.
+"""
+
+# ruff: noqa: D102,D103,D105
+
+import datetime
+import os
+import time
+import unittest
+from threading import Thread
+
+import cherrypy
+
+from ws4py.client.threadedclient import WebSocketClient
+from ws4py.server.cherrypyserver import WebSocketPlugin, WebSocketTool
+from ws4py.websocket import WebSocket
+
+root = os.path.dirname( os.path.abspath( __file__ ) )
+
+NUMBERS_TO_SEND = 10 # send this amount of numbers back and forth between client and server
+TIMEOUT_LIMIT   = 20 # CherryPy server timeout limit in seconds
+
+global_data = {
+    "sent"     : [],
+    "received" : [],
+    "timedout" : False,
+}
+
+class EchoClient( WebSocketClient ):
+    def opened( self ):
+        self.send( 1 )
+
+    def send( self, payload, binary=False ):
+        global_data[ "sent" ].append( payload )
+        return super().send( str( payload ), binary )
+
+    def closed( self, code, reason=None ):
+        cherrypy.engine.exit()  # close the CherryPy server
+
+    def received_message( self, m ):
+        m = str( m ).strip()
+        try:
+            number = int( m )
+            global_data[ "received" ].append( number )
+            if number < NUMBERS_TO_SEND:
+                # increase number and send it back
+                self.send( number + 1 )
+            else:
+                self.close( 1000, "Done" )
+        except ( TypeError, ValueError ):
+            return
+
+
+class BroadcastWebSocketHandler( WebSocket ):
+    def received_message( self, message ):
+        cherrypy.engine.publish( 'websocket-broadcast', str( message ) )
+
+
+class Root:
+    @cherrypy.expose
+    def ws( self ):
+        pass
+
+
+def wait_for_cherrypy_engine_started():
+    while ( cherrypy.engine.state != cherrypy.engine.states.STARTED ):
+        time.sleep( 0.5 )
+
+
+def run_echo_client( url ):
+    wait_for_cherrypy_engine_started()
+    try:
+        ws = EchoClient( url )
+        ws.connect()
+        ws.run_forever()
+    except KeyboardInterrupt:
+        ws.close()
+
+
+def run_echo_client_thread( host, port, ssl = False ):
+    """Run the EchoClient in a separate thread.
+    """
+    url = "wss://%s:%d/ws" % (host, port) if ssl else "ws://%s:%d/ws" % (host, port)
+    t   = Thread( target=run_echo_client, daemon=True, name="WebSocketClient", args=( url, ) )
+    t.start()
+    return t
+
+
+def run_cherrypy_server( host, port, ssl = False ):
+    config = {
+        "global" : {
+            "server.ssl_module"      : "builtin" if ssl else None,
+            "server.ssl_certificate" : os.path.join( root, "server.crt" ) if ssl else None,
+            "server.ssl_private_key" : os.path.join( root, "server.key" ) if ssl else None,
+            "server.socket_host"     : host,
+            "server.socket_port"     : port,
+            "log.screen"             : False,
+            'engine.autoreload.on'   : False,
+        },
+        "/ws" : {
+            "tools.websocket.on"          : True,
+            "tools.websocket.handler_cls" : BroadcastWebSocketHandler,
+            'tools.websocket.protocols'   : [ 'some-protocol' ],
+            'tools.gzip.on'               : False,
+            'tools.caching.on'            : False,
+            'tools.sessions.on'           : False,
+        },
+    }
+
+    WebSocketPlugin( cherrypy.engine ).subscribe()
+    cherrypy.tools.websocket = WebSocketTool()
+    cherrypy.quickstart( Root(), "/", config=config )
+
+
+class TimeoutSignaller( Thread ):
+    def __init__( self, limit, handler ):
+        Thread.__init__( self, name="TimeoutSignaller" )
+        self.limit   = limit
+        self.running = True
+        self.handler = handler
+        self.daemon  = True
+        assert callable( handler ), "Timeout Handler needs to be a method"
+
+    def run( self ):
+        timeout_limit = datetime.datetime.now() + datetime.timedelta( seconds=self.limit )
+        while self.running:
+            if datetime.datetime.now() >= timeout_limit:
+                self.handler()
+                self.stop_run()
+                break
+
+    def stop_run( self ):
+        self.running = False
+
+
+class CherryPyTimeout:
+    def __init__( self, seconds=0, minutes=0, hours=0 ):
+        self.seconds = ( hours * 3600 ) + ( minutes * 60 ) + seconds
+        self.signal  = TimeoutSignaller( self.seconds, self.signal_handler )
+        self.ok      = True
+
+    def __enter__( self ):
+        self.signal.start()
+        return self
+
+    def __exit__( self, exc_type, exc_val, exc_tb ):
+        self.signal.stop_run()
+
+    def done( self ):
+        self.signal.stop_run()
+
+    def signal_handler( self ):
+        if cherrypy.engine.state == cherrypy.engine.states.STARTED:
+            global_data[ "timedout" ] = True
+            cherrypy.engine.exit()
+            self.ok = False
+
+
+def run( host, port, ssl ):
+    run_echo_client_thread( host, port, ssl=ssl )
+    run_cherrypy_server( host=host, port=port, ssl=ssl )
+
+
+def run_with_timeout( host, port, ssl ):
+    with CherryPyTimeout( seconds=TIMEOUT_LIMIT ) as t:
+        run( host=host, port=port, ssl=ssl )
+        t.done()
+        if not t.ok:
+            global_data[ "timedout" ] = True
+            raise TimeoutError( "CherryPy server timed out" )
+
+class CherryPySSLTest(unittest.TestCase):
+    def test_ssl( self ):
+        global_data[ "sent" ]     = []
+        global_data[ "received" ] = []
+        global_data[ "timedout" ] = False
+
+        run_with_timeout( host="127.0.0.1", port=8877, ssl=True )
+
+        assert not global_data[ "timedout" ], "Test timed out"
+        assert len( global_data[ "sent" ] ) > 0, "No data sent to the server"
+        assert len( global_data[ "received" ] ) > 0, "No data received from the server"
+        assert global_data[ "sent" ] == global_data[ "received" ], "Sent and received data do not match"
+        assert global_data[ "sent" ] == list( range( 1, NUMBERS_TO_SEND + 1 ) ), "Sent data does not match expected data"
+
+
+if __name__ == "__main__":
+    suite = unittest.TestSuite()
+    loader = unittest.TestLoader()
+    for testcase in [CherryPySSLTest]:
+        tests = loader.loadTestsFromTestCase(testcase)
+        suite.addTests(tests)
+    unittest.TextTestRunner(verbosity=2).run(suite)

ws4py/manager.py

@@ -284,6 +284,22 @@ def stop(self):
             self.websockets.clear()
             self.poller.release()
 
+    def get_fds_with_pending_data( self ):
+        """
+        Returns a list of file descriptors that have pending data.
+
+        This is a workaround for a bug where polling returns no file descriptor,
+        even though we have sockets with pending data.
+        """
+        fds = []
+        with self.lock:
+            for ws in ( ws for ws in self.websockets.values() if ws._is_secure ):
+                if hasattr( ws.sock, 'pending' ) and ws.sock.pending():
+                    fds.append( ws.sock.fileno() )
+                    # set special handling flag
+                    ws._force_process_buffer = True
+        return fds
+
     def run(self):
         """
         Manager's mainloop executed from within a thread.
@@ -309,6 +325,10 @@ def run(self):
             if not self.running:
                 break
 
+            if not polled:
+                # workaround WSS bug
+                polled.extend( self.get_fds_with_pending_data() )
+
             for fd in polled:
                 if not self.running:
                     break

ws4py/websocket.py

@@ -113,6 +113,11 @@ def __init__(self, sock, protocols=None, extensions=None, environ=None, heartbea
         Tell us if the socket is secure or not.
         """
 
+        self._force_process_buffer = False
+        """
+        Indicates special handling of the buffer, only relevant for secure sockets.
+        """
+
         self.client_terminated = False
         """
         Indicates if the client has been marked as terminated.
@@ -241,7 +246,7 @@ def close_connection(self):
             except:
                 pass
             self.sock = None
-            
+
 
     def ping(self, message):
         """
@@ -424,6 +429,15 @@ def once(self):
             if not self.process(self.buf[:requested]):
                 return False
             self.buf = self.buf[requested:]
+            if self.buf and self._force_process_buffer:
+                self._force_process_buffer = False
+                # if we are in special handling mode, we need to process
+                # the buffer until it is empty
+                while self.buf:
+                    requested = self.reading_buffer_size
+                    if not self.process(self.buf[:requested]):
+                        return False
+                    self.buf = self.buf[requested:]
 
         return True