Fixed under-read of incoming handshake data with an accumulator

mondain · mondain · commit ae18c9e1b7e0 · 2017-06-02T08:06:27.000-07:00
diff --git a/pom.xml b/pom.xml
@@ -9,7 +9,7 @@
     <artifactId>websocket</artifactId>
     <packaging>jar</packaging>
     <name>websocket</name>
-    <version>1.10</version>
+    <version>1.11</version>
     <description>Red5 WebSocket plugin</description>
     <url>https://github.com/Red5/red5-websocket</url>
     <organization>
@@ -224,7 +224,7 @@
         <dependency>
             <groupId>org.glassfish.tyrus.bundles</groupId>
             <artifactId>tyrus-standalone-client-jdk</artifactId>
-            <version>1.12</version>
+            <version>1.13.1</version>
             <scope>test</scope>
         </dependency>
     </dependencies>
diff --git a/src/main/java/org/red5/net/websocket/Constants.java b/src/main/java/org/red5/net/websocket/Constants.java
@@ -33,6 +33,8 @@ public class Constants {
 
     public final static String SESSION = "session";
 
+    public static final Object WS_HANDSHAKE = "ws.handshake";
+
     public final static String WS_HEADER_KEY = "Sec-WebSocket-Key";
 
     public final static String WS_HEADER_VERSION = "Sec-WebSocket-Version";
@@ -58,5 +60,8 @@ public class Constants {
     // magic string for websockets
     public static final String WEBSOCKET_MAGIC_STRING = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
 
-    public static final byte[] CRLF = { 0x0D, 0x0A };
+    public static final byte[] CRLF = { 0x0d, 0x0a };
+
+    public static final byte[] END_OF_REQ = { 0x0d, 0x0a, 0x0d, 0x0a };
+
 }
diff --git a/src/main/java/org/red5/net/websocket/WebSocketConnection.java b/src/main/java/org/red5/net/websocket/WebSocketConnection.java
@@ -132,7 +132,9 @@ public void send(String data) throws UnsupportedEncodingException {
      * @param buf
      */
     public void send(byte[] buf) {
-        log.trace("send binary: {}", Arrays.toString(buf));
+        if (log.isTraceEnabled()) {
+            log.trace("send binary: {}", Arrays.toString(buf));
+        }
         Packet packet = Packet.build(buf);
         session.write(packet);
     }
@@ -143,7 +145,9 @@ public void send(byte[] buf) {
      * @param buf
      */
     public void sendPong(byte[] buf) {
-        log.trace("send pong: {}", buf);
+        if (log.isTraceEnabled()) {
+            log.trace("send pong: {}", buf);
+        }
         Packet packet = Packet.build(buf, MessageType.PONG);
         session.write(packet);
     }
diff --git a/src/main/java/org/red5/net/websocket/WebSocketPlugin.java b/src/main/java/org/red5/net/websocket/WebSocketPlugin.java
@@ -111,9 +111,12 @@ public WebSocketScopeManager getManager(String path) {
         if (log.isTraceEnabled()) {
             log.trace("Path parts: {}", Arrays.toString(parts));
         }
-        if (parts.length > 0) {
+        if (parts.length > 1) {
             // skip default in a path if it exists in slot #1
-            String name = !"default".equals(parts[1]) ? parts[1] : parts[2];
+            String name = !"default".equals(parts[1]) ? parts[1] : ((parts.length >= 3) ? parts[2] : parts[1]);
+            if (log.isDebugEnabled()) {
+                log.debug("Managers: {}", managerMap.entrySet());
+            }
             for (Entry<IScope, WebSocketScopeManager> entry : managerMap.entrySet()) {
                 IScope appScope = entry.getKey();
                 if (appScope.getName().equals(name)) {
diff --git a/src/main/java/org/red5/net/websocket/WebSocketTransport.java b/src/main/java/org/red5/net/websocket/WebSocketTransport.java
@@ -81,6 +81,18 @@ public class WebSocketTransport implements InitializingBean, DisposableBean {
     public void afterPropertiesSet() throws Exception {
         // create the nio acceptor
         acceptor = new NioSocketAcceptor(ioThreads);
+        // configure the acceptor
+        SocketSessionConfig sessionConf = acceptor.getSessionConfig();
+        sessionConf.setReuseAddress(true);
+        sessionConf.setTcpNoDelay(true);
+        sessionConf.setSendBufferSize(sendBufferSize);
+        sessionConf.setReadBufferSize(receiveBufferSize);
+        // close sessions when the acceptor is stopped
+        acceptor.setCloseOnDeactivation(true);
+        acceptor.setHandler(ioHandler);
+        // requested maximum length of the queue of incoming connections
+        acceptor.setBacklog(64);
+        acceptor.setReuseAddress(true);
         // instance the websocket handler
         if (ioHandler == null) {
             ioHandler = new WebSocketHandler();
@@ -102,17 +114,6 @@ public void afterPropertiesSet() throws Exception {
         }
         // add the websocket codec factory
         chain.addLast("protocol", new ProtocolCodecFilter(new WebSocketCodecFactory()));
-        // close sessions when the acceptor is stopped
-        acceptor.setCloseOnDeactivation(true);
-        acceptor.setHandler(ioHandler);
-        // requested maximum length of the queue of incoming connections
-        acceptor.setBacklog(64);
-        SocketSessionConfig sessionConf = acceptor.getSessionConfig();
-        sessionConf.setReuseAddress(true);
-        sessionConf.setTcpNoDelay(true);
-        sessionConf.setReceiveBufferSize(receiveBufferSize);
-        sessionConf.setSendBufferSize(sendBufferSize);
-        acceptor.setReuseAddress(true);
         if (addresses.isEmpty()) {
             if (sslFilter != null) {
                 log.info("WebSocket (wss) will be bound to port {}", port);
@@ -144,6 +145,9 @@ public void afterPropertiesSet() throws Exception {
             }
         }
         log.info("started {} websocket transport", (isSecure() ? "secure" : ""));
+        if (log.isDebugEnabled()) {
+            log.debug("Acceptor sizes - send: {} recv: {}", acceptor.getSessionConfig().getSendBufferSize(), acceptor.getSessionConfig().getReadBufferSize());
+        }
     }
 
     /**
diff --git a/src/main/java/org/red5/net/websocket/codec/WebSocketDecoder.java b/src/main/java/org/red5/net/websocket/codec/WebSocketDecoder.java
@@ -95,11 +95,18 @@ protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput
         IoBuffer resultBuffer;
         WebSocketConnection conn = (WebSocketConnection) session.getAttribute(Constants.CONNECTION);
         if (conn == null) {
+            log.debug("Decode start pos: {}", in.position());
             // first message on a new connection, check if its from a websocket or a native socket
             if (doHandShake(session, in)) {
+                log.debug("Decode end pos: {} limit: {}", in.position(), in.limit());
                 // websocket handshake was successful. Don't write anything to output as we want to abstract the handshake request message from the handler
-                in.position(in.limit());
+                if (in.position() != in.limit()) {
+                    in.position(in.limit());
+                }
                 return true;
+            } else if (session.containsAttribute(Constants.WS_HANDSHAKE)) {
+                // more still expected to come in before HS is completed
+                return false;
             } else {
                 // message is from a native socket. Simply wrap and pass through
                 resultBuffer = IoBuffer.wrap(in.array(), 0, in.limit());
@@ -145,14 +152,40 @@ protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput
      */
     @SuppressWarnings("unchecked")
     private boolean doHandShake(IoSession session, IoBuffer in) {
+        if (log.isDebugEnabled()) {
+            log.debug("Handshake: {}", in);
+        }
+        // incoming data
+        byte[] data = null;
+        // check for existing HS data
+        if (session.containsAttribute(Constants.WS_HANDSHAKE)) {
+            byte[] tmp = (byte[]) session.getAttribute(Constants.WS_HANDSHAKE);
+            // size to hold existing and incoming
+            data = new byte[tmp.length + in.remaining()];
+            System.arraycopy(tmp, 0, data, 0, tmp.length);
+            // get incoming bytes
+            in.get(data, tmp.length, in.remaining());
+        } else {
+            // size for incoming bytes
+            data = new byte[in.remaining()];
+            // get incoming bytes
+            in.get(data, 0, data.length);
+        }
+        // ensure the incoming data is complete (ends with crlfcrlf)
+        byte[] tail = Arrays.copyOfRange(data, data.length - 4, data.length);
+        if (!Arrays.equals(tail, Constants.END_OF_REQ)) {
+            // accumulate the HS data
+            session.setAttribute(Constants.WS_HANDSHAKE, data);
+            return false;
+        }
         // create the connection obj
         WebSocketConnection conn = new WebSocketConnection(session);
         // mark as secure if using ssl
         if (session.getFilterChain().contains("sslFilter")) {
             conn.setSecure(true);
         }
         try {
-            Map<String, Object> headers = parseClientRequest(conn, new String(in.array()));
+            Map<String, Object> headers = parseClientRequest(conn, new String(data));
             if (log.isTraceEnabled()) {
                 log.trace("Header map: {}", headers);
             }
@@ -206,6 +239,8 @@ private boolean doHandShake(IoSession session, IoBuffer in) {
                 // prepare response and write it to the directly to the session
                 HandshakeResponse wsResponse = buildHandshakeResponse(conn, (String) headers.get(Constants.WS_HEADER_KEY));
                 session.write(wsResponse);
+                // remove handshake acculator
+                session.removeAttribute(Constants.WS_HANDSHAKE);
                 log.debug("Handshake complete");
                 return true;
             }
@@ -231,7 +266,7 @@ private Map<String, Object> parseClientRequest(WebSocketConnection conn, String
         if (log.isTraceEnabled()) {
             log.trace("Request: {}", Arrays.toString(request));
         }
-        Map<String, Object> map = new HashMap<String, Object>();
+        Map<String, Object> map = new HashMap<>();
         for (int i = 0; i < request.length; i++) {
             log.trace("Request {}: {}", i, request[i]);
             if (request[i].startsWith("GET ") || request[i].startsWith("POST ") || request[i].startsWith("PUT ")) {
@@ -291,7 +326,7 @@ public void operationComplete(IoFuture future) {
                     });
                     throw new WebSocketException("Handshake failed, missing plugin");
                 }
-            }else if (request[i].contains(Constants.WS_HEADER_KEY)) {
+            } else if (request[i].contains(Constants.WS_HEADER_KEY)) {
                 map.put(Constants.WS_HEADER_KEY, extractHeaderValue(request[i]));
             } else if (request[i].contains(Constants.WS_HEADER_VERSION)) {
                 map.put(Constants.WS_HEADER_VERSION, extractHeaderValue(request[i]));
@@ -307,26 +342,22 @@ public void operationComplete(IoFuture future) {
                 conn.setOrigin(extractHeaderValue(request[i]));
             } else if (request[i].contains(Constants.HTTP_HEADER_USERAGENT)) {
                 map.put(Constants.HTTP_HEADER_USERAGENT, extractHeaderValue(request[i]));
-            }else if (request[i].startsWith(Constants.WS_HEADER_GENERIC_PREFIX)) {
-            	map.put(getHeaderName(request[i]), extractHeaderValue(request[i]));
+            } else if (request[i].startsWith(Constants.WS_HEADER_GENERIC_PREFIX)) {
+                map.put(getHeaderName(request[i]), extractHeaderValue(request[i]));
             }
         }
         return map;
     }
-    
-    
-    
+
     /**
      * Returns the trimmed header name.
      * 
      * @param requestHeader
      * @return value
      */
-    private String getHeaderName(String requestHeader){
-    	return requestHeader.substring(0, requestHeader.indexOf(':')).trim();
+    private String getHeaderName(String requestHeader) {
+        return requestHeader.substring(0, requestHeader.indexOf(':')).trim();
     }
-    
-    
 
     /**
      * Returns the trimmed header value.
diff --git a/src/test/java/org/red5/net/websocket/WebSocketServerTest.java b/src/test/java/org/red5/net/websocket/WebSocketServerTest.java
diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml
