Merge pull request TooTallNate#193 from Davidiusdadi/master

fixed critical bugs causing random protocol errors and bug that caused server to hang on stop()

Author: Davidiusdadi

src/main/java/org/java_websocket/SSLSocketChannel2.java

@@ -31,6 +31,9 @@
  * Implements the relevant portions of the SocketChannel interface with the SSLEngine wrapper.
  */
 publicclassSSLSocketChannel2implementsByteChannel, WrappedByteChannel{
+/**
+ * This object is used to feed the{@link SSLEngine}'s wrap and unwrap methods during the handshake phase.
+ **/
 protectedstaticByteBufferemptybuffer = ByteBuffer.allocate( 0 );
 
 protectedExecutorServiceexec;
@@ -49,12 +52,9 @@ public class SSLSocketChannel2 implements ByteChannel, WrappedByteChannel{
 /** used to set interestOP SelectionKey.OP_WRITE for the underlying channel */
 protectedSelectionKeyselectionKey;
 
-
-protectedSSLEngineResultengineResult;
 protectedSSLEnginesslEngine;
-
-
-privateStatusengineStatus = Status.BUFFER_UNDERFLOW;
+protectedSSLEngineResultreadEngineResult;
+protectedSSLEngineResultwriteEngineResult;
 
 publicSSLSocketChannel2( SocketChannelchannel , SSLEnginesslEngine , ExecutorServiceexec , SelectionKeykey ) throwsIOException{
 if( channel == null || sslEngine == null || exec == null )
@@ -64,6 +64,8 @@ public SSLSocketChannel2( SocketChannel channel , SSLEngine sslEngine , Executor
 this.sslEngine = sslEngine;
 this.exec = exec;
 
+readEngineResult = writeEngineResult = newSSLEngineResult( Status.BUFFER_UNDERFLOW, sslEngine.getHandshakeStatus(), 0, 0 ); // init to prevent NPEs
+
 tasks = newArrayList<Future<?>>( 3 );
 if( key != null ){
 key.interestOps( key.interestOps() | SelectionKey.OP_WRITE );
@@ -93,8 +95,12 @@ private void consumeFutureUninterruptible( Future<?> f ){
  }
  }
 
+/**
+ * This method will do whatever necessary to process the sslengine handshake.
+ * Thats why it's called both from the{@link #read(ByteBuffer)} and{@link #write(ByteBuffer)}
+ **/
 privatesynchronizedvoidprocessHandshake() throwsIOException{
-if( engineResult.getHandshakeStatus() == HandshakeStatus.NOT_HANDSHAKING )
+if( sslEngine.getHandshakeStatus() == HandshakeStatus.NOT_HANDSHAKING )
 return; // since this may be called either from a reading or a writing thread and because this method is synchronized it is necessary to double check if we are still handshaking.
 if( !tasks.isEmpty() ){
 Iterator<Future<?>> it = tasks.iterator();
@@ -110,8 +116,8 @@ private synchronized void processHandshake() throws IOException{
  }
  }
 
-if( engineResult.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NEED_UNWRAP ){
-if( !isBlocking() || engineStatus == Status.BUFFER_UNDERFLOW ){
+if( sslEngine.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NEED_UNWRAP ){
+if( !isBlocking() || readEngineResult.getStatus() == Status.BUFFER_UNDERFLOW ){
 inCrypt.compact();
 intread = socketChannel.read( inCrypt );
 if( read == -1 ){
@@ -121,36 +127,37 @@ private synchronized void processHandshake() throws IOException{
  }
 inData.compact();
 unwrap();
-if( engineResult.getHandshakeStatus() == HandshakeStatus.FINISHED ){
+if( sslEngine.getHandshakeStatus() == HandshakeStatus.FINISHED ){
 createBuffers( sslEngine.getSession() );
 return;
  }
  }
 consumeDelegatedTasks();
-assert ( engineResult.getHandshakeStatus() != HandshakeStatus.NOT_HANDSHAKING );
-if( tasks.isEmpty() || engineResult.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NEED_WRAP ){
+assert ( sslEngine.getHandshakeStatus() != HandshakeStatus.NOT_HANDSHAKING );
+if( tasks.isEmpty() || sslEngine.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NEED_WRAP ){
 socketChannel.write( wrap( emptybuffer ) );
-if( engineResult.getHandshakeStatus() == HandshakeStatus.FINISHED ){
+if( sslEngine.getHandshakeStatus() == HandshakeStatus.FINISHED ){
 createBuffers( sslEngine.getSession() );
  }
  }
  }
 
 privatesynchronizedByteBufferwrap( ByteBufferb ) throwsSSLException{
 outCrypt.compact();
-engineResult = sslEngine.wrap( b, outCrypt );
+writeEngineResult = sslEngine.wrap( b, outCrypt );
 outCrypt.flip();
 returnoutCrypt;
  }
 
+/**
+ * performs the unwrap operation by unwrapping from{@link #inCrypt} to{@link #inData}
+ **/
 privatesynchronizedByteBufferunwrap() throwsSSLException{
 intrem;
 do{
 rem = inData.remaining();
-engineResult = sslEngine.unwrap( inCrypt, inData );
-engineStatus = engineResult.getStatus();
- } while ( engineStatus == SSLEngineResult.Status.OK && ( rem != inData.remaining() || engineResult.getHandshakeStatus() == HandshakeStatus.NEED_UNWRAP ) );
-
+readEngineResult = sslEngine.unwrap( inCrypt, inData );
+ } while ( readEngineResult.getStatus() == SSLEngineResult.Status.OK && ( rem != inData.remaining() || sslEngine.getHandshakeStatus() == HandshakeStatus.NEED_UNWRAP ) );
 inData.flip();
 returninData;
  }
@@ -200,6 +207,8 @@ public int write( ByteBuffer src ) throws IOException{
 /**
  * Blocks when in blocking mode until at least one byte has been decoded.<br>
  * When not in blocking mode 0 may be returned.
+ * 
+ * @return the number of bytes read.
  **/
 publicintread( ByteBufferdst ) throwsIOException{
 if( !dst.hasRemaining() )
@@ -216,11 +225,16 @@ public int read( ByteBuffer dst ) throws IOException{
  }
  }
  }
-
+/* 1. When "dst" is smaller than "inData" readRemaining will fill "dst" with data decoded in a previous read call.
+ * 2. When "inCrypt" contains more data than "inData" has remaining space, unwrap has to be called on more time(readRemaining)
+ */
 intpurged = readRemaining( dst );
 if( purged != 0 )
 returnpurged;
 
+/* We only continue when we really need more data from the network.
+ * Thats the case if inData is empty or inCrypt holds to less data than necessary for decryption
+ */
 assert ( inData.position() == 0 );
 inData.clear();
 
@@ -229,7 +243,7 @@ public int read( ByteBuffer dst ) throws IOException{
 else
 inCrypt.compact();
 
-if( ( isBlocking() && inCrypt.position() == 0 ) || engineStatus == Status.BUFFER_UNDERFLOW )
+if( isBlocking() || readEngineResult.getStatus() == Status.BUFFER_UNDERFLOW )
 if( socketChannel.read( inCrypt ) == -1 ){
 return -1;
  }
@@ -243,6 +257,9 @@ public int read( ByteBuffer dst ) throws IOException{
 returntransfered;
  }
 
+/**
+ *{@link #read(ByteBuffer)} may not be to leave all buffers(inData, inCrypt)
+ **/
 privateintreadRemaining( ByteBufferdst ) throwsSSLException{
 if( inData.hasRemaining() ){
 returntransfereTo( inData, dst );
@@ -273,7 +290,7 @@ public void close() throws IOException{
  }
 
 privatebooleanisHandShakeComplete(){
-HandshakeStatusstatus = engineResult.getHandshakeStatus();
+HandshakeStatusstatus = sslEngine.getHandshakeStatus();
 returnstatus == SSLEngineResult.HandshakeStatus.FINISHED || status == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING;
  }
 
@@ -304,7 +321,7 @@ public boolean isOpen(){
 
 @Override
 publicbooleanisNeedWrite(){
-returnoutCrypt.hasRemaining() || !isHandShakeComplete();
+returnoutCrypt.hasRemaining() || !isHandShakeComplete();// FIXME this condition can cause high cpu load during handshaking when network is slow
  }
 
 @Override
@@ -314,8 +331,7 @@ public void writeMore() throws IOException{
 
 @Override
 publicbooleanisNeedRead(){
-returninData.hasRemaining() || ( inCrypt.hasRemaining() && engineResult.getStatus() != Status.BUFFER_UNDERFLOW && 
-engineResult.getStatus() != Status.CLOSED );
+returninData.hasRemaining() || ( inCrypt.hasRemaining() && readEngineResult.getStatus() != Status.BUFFER_UNDERFLOW && readEngineResult.getStatus() != Status.CLOSED );
  }
 
 @Override

src/main/java/org/java_websocket/WebSocketImpl.java

@@ -93,7 +93,7 @@ public class WebSocketImpl implements WebSocket{
 privateOpcodecurrent_continuous_frame_opcode = null;
 
 /** the bytes of an incomplete received handshake */
-privateByteBuffertmpHandshakeBytes;
+privateByteBuffertmpHandshakeBytes = ByteBuffer.allocate( 0 );
 
 /** stores the handshake sent by this websocket ( Role.CLIENT only ) */
 privateClientHandshakehandshakerequest = null;
@@ -149,8 +149,11 @@ public WebSocketImpl( WebSocketListener listener , List<Draft> drafts , Socket s
  * 
  */
 publicvoiddecode( ByteBuffersocketBuffer ){
-if( !socketBuffer.hasRemaining() || flushandclosestate )
+assert ( socketBuffer.hasRemaining() );
+
+if( flushandclosestate ){
 return;
+ }
 
 if( DEBUG )
 System.out.println( "process(" + socketBuffer.remaining() + "):{" + ( socketBuffer.remaining() > 1000 ? "too big to display" : newString( socketBuffer.array(), socketBuffer.position(), socketBuffer.remaining() ) ) + "}" );
@@ -159,19 +162,24 @@ public void decode( ByteBuffer socketBuffer ){
 decodeFrames( socketBuffer );
  } else{
 if( decodeHandshake( socketBuffer ) ){
-decodeFrames( socketBuffer );
+assert ( tmpHandshakeBytes.hasRemaining() != socketBuffer.hasRemaining() || !socketBuffer.hasRemaining() ); // the buffers will never have remaining bytes at the same time
+
+if( socketBuffer.hasRemaining() ){
+decodeFrames( socketBuffer );
+ } elseif( tmpHandshakeBytes.hasRemaining() ){
+decodeFrames( tmpHandshakeBytes );
+ }
  }
  }
 assert ( isClosing() || isFlushAndClose() || !socketBuffer.hasRemaining() );
  }
-
 /**
  * Returns whether the handshake phase has is completed.
  * In case of a broken handshake this will be never the case.
  **/
 privatebooleandecodeHandshake( ByteBuffersocketBufferNew ){
 ByteBuffersocketBuffer;
-if( tmpHandshakeBytes == null ){
+if( tmpHandshakeBytes.capacity() == 0 ){
 socketBuffer = socketBufferNew;
  } else{
 if( tmpHandshakeBytes.remaining() < socketBufferNew.remaining() ){
@@ -260,7 +268,7 @@ private boolean decodeHandshake( ByteBuffer socketBufferNew ){
 draft.setParseMode( role );
 Handshakedatatmphandshake = draft.translateHandshake( socketBuffer );
 if( tmphandshakeinstanceofServerHandshake == false ){
-flushAndClose( CloseFrame.PROTOCOL_ERROR, "Wwrong http function", false );
+flushAndClose( CloseFrame.PROTOCOL_ERROR, "wrong http function", false );
 returnfalse;
  }
 ServerHandshakehandshake = (ServerHandshake) tmphandshake;
@@ -286,7 +294,7 @@ private boolean decodeHandshake( ByteBuffer socketBufferNew ){
 close( e );
  }
  } catch ( IncompleteHandshakeExceptione ){
-if( tmpHandshakeBytes == null ){
+if( tmpHandshakeBytes.capacity() == 0 ){
 socketBuffer.reset();
 intnewsize = e.getPreferedSize();
 if( newsize == 0 ){

src/main/java/org/java_websocket/server/WebSocketServer.java

@@ -8,6 +8,7 @@
 importjava.nio.ByteBuffer;
 importjava.nio.channels.ByteChannel;
 importjava.nio.channels.CancelledKeyException;
+importjava.nio.channels.ClosedByInterruptException;
 importjava.nio.channels.SelectableChannel;
 importjava.nio.channels.SelectionKey;
 importjava.nio.channels.Selector;
@@ -325,6 +326,7 @@ public void run(){
 ByteBufferbuf = takeBuffer();
 try{
 if( SocketChannelIOHelper.read( buf, conn, (ByteChannel) conn.channel ) ){
+assert ( buf.hasRemaining() );
 conn.inQueue.put( buf );
 queue( conn );
 i.remove();
@@ -339,9 +341,6 @@ public void run(){
  } catch ( IOExceptione ){
 pushBuffer( buf );
 throwe;
- } catch ( RuntimeExceptione ){
-pushBuffer( buf );
-throwe;
  }
  }
 if( key.isWritable() ){
@@ -359,21 +358,25 @@ public void run(){
 try{
 if( SocketChannelIOHelper.readMore( buf, conn, c ) )
 iqueue.add( conn );
+assert ( buf.hasRemaining() );
 conn.inQueue.put( buf );
 queue( conn );
- } finally{
+ } catch ( IOExceptione ){
 pushBuffer( buf );
+throwe;
  }
 
  }
  } catch ( CancelledKeyExceptione ){
 // an other thread may cancel the key
+ } catch ( ClosedByInterruptExceptione ){
+return; // do the same stuff as when InterruptedException is thrown
  } catch ( IOExceptionex ){
 if( key != null )
 key.cancel();
 handleIOException( key, conn, ex );
  } catch ( InterruptedExceptione ){
-return;// FIXME controlled shutdown
+return;// FIXME controlled shutdown (e.g. take care of buffermanagement)
  }
  }
 
@@ -418,7 +421,7 @@ private void pushBuffer( ByteBuffer buf ) throws InterruptedException{
  }
 
 privatevoidhandleIOException( SelectionKeykey, WebSocketconn, IOExceptionex ){
-//onWebsocketError( conn, ex );// conn may be null here
+//onWebsocketError( conn, ex );// conn may be null here
 if( conn != null ){
 conn.closeConnection( CloseFrame.ABNORMAL_CLOSE, ex.getMessage() );
  } elseif( key != null ){