Make web-socket pass AutobanhTestSuite.

sdk/lib/io/websocket_impl.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 part of dart.io;
 
 const String _webSocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
 
 class _WebSocketMessageType {
   static const int NONE = 0;
@@ skipping 57 matching lines @@
       if (_state == CLOSED) {
         throw new WebSocketException("Data on closed connection");
       }
       if (_state == FAILURE) {
         throw new WebSocketException("Data on failed connection");
       }
       while ((index < lastIndex) && _state != CLOSED && _state != FAILURE) {
         int byte = buffer[index];
         switch (_state) {
           case START:
             _fin = (byte & 0x80) != 0;

[Søren Gjesse, 2013/05/08 12:15:40]

Add comment

// The RSV1, RSV2 bits RSV3 most be all zero.

[Anders Johnsen, 2013/06/11 12:23:47]

Done.

+            if ((byte & 0x70) != 0) {
+              throw new WebSocketException("Protocol error");
+            }
             _opcode = (byte & 0xF);
             switch (_opcode) {
             case _WebSocketOpcode.CONTINUATION:
               if (_currentMessageType == _WebSocketMessageType.NONE) {
                 throw new WebSocketException("Protocol error");
               }
               break;
 
             case _WebSocketOpcode.TEXT:
               if (_currentMessageType != _WebSocketMessageType.NONE) {
                 throw new WebSocketException("Protocol error");
               }
               _currentMessageType = _WebSocketMessageType.TEXT;
-              _buffer = new StringBuffer();
+              _controller = new StreamController();
+              _controller.stream
+                  .transform(new Utf8DecoderTransformer(null))
+                  .fold(new StringBuffer(), (buffer, str) => buffer..write(str))
+                  .then((buffer) {
+                    sink.add(buffer.toString());
+                  }, onError: (error) {
+                    sink.addError(error);
+                  });
               break;
 
             case _WebSocketOpcode.BINARY:
               if (_currentMessageType != _WebSocketMessageType.NONE) {
                 throw new WebSocketException("Protocol error");
               }
               _currentMessageType = _WebSocketMessageType.BINARY;
-              _buffer = new _BufferList();
+              _controller = new StreamController();
+              _controller.stream
+                  .fold(new _BufferList(), (buffer, data) => buffer..add(data))
+                  .then((buffer) {
+                    sink.add(buffer.readBytes());
+                  }, onError: (error) {
+                    sink.addError(error);
+                  });
               break;
 
             case _WebSocketOpcode.CLOSE:
             case _WebSocketOpcode.PING:
             case _WebSocketOpcode.PONG:
               // Control frames cannot be fragmented.
               if (!_fin) throw new WebSocketException("Protocol error");
               break;
 
             default:
               throw new WebSocketException("Protocol error");
             }
             _state = LEN_FIRST;
             break;
 
           case LEN_FIRST:
             _masked = (byte & 0x80) != 0;
             _len = byte & 0x7F;
-            if (_isControlFrame() && _len > 126) {
+            if (_isControlFrame() && _len > 125) {
               throw new WebSocketException("Protocol error");
             }
             if (_len < 126) {
               _lengthDone(sink);
             } else if (_len == 126) {
               _len = 0;
               _remainingLenBytes = 2;
               _state = LEN_REST;
             } else if (_len == 127) {
               _len = 0;
@@ skipping 46 matching lines @@
                   _controlPayload = new List<int>();
                 }
                 _controlPayload.addAll(buffer.sublist(index, index + payload));
                 index += payload;
               }
 
               if (_remainingPayloadBytes == 0) {
                 _controlFrameEnd(sink);
               }
             } else {
-              switch (_currentMessageType) {
-                case _WebSocketMessageType.NONE:
+              if (_currentMessageType != _WebSocketMessageType.TEXT &&
+                  _currentMessageType != _WebSocketMessageType.BINARY) {
                   throw new WebSocketException("Protocol error");
-
-                case _WebSocketMessageType.TEXT:
-                  _buffer.write(_decodeString(
-                      buffer.sublist(index, index + payload)));
-                  index += payload;
-                  if (_remainingPayloadBytes == 0) {
-                    _messageFrameEnd(sink);
-                  }
-                  break;
-
-                case _WebSocketMessageType.BINARY:
-                  _buffer.write(buffer.sublist(index, index + payload));
-                  index += payload;
-                  if (_remainingPayloadBytes == 0) {
-                    _messageFrameEnd(sink);
-                  }
-                  break;
-
-                default:
-                  throw new WebSocketException("Protocol error");
+              }
+              _controller.add(new Uint8List.view(buffer.buffer, index, payload));
+              index += payload;
+              if (_remainingPayloadBytes == 0) {
+                _messageFrameEnd(sink);
               }
             }
 
             // Hack - as we always do index++ below.
             index--;
             break;
         }
 
         // Move to the next byte.
         index++;
@@ skipping 29 matching lines @@
     // If there is no actual payload perform perform callbacks without
     // going through the PAYLOAD state.
     if (_remainingPayloadBytes == 0) {
       if (_isControlFrame()) {
         switch (_opcode) {
           case _WebSocketOpcode.CLOSE:
             _state = CLOSED;
             sink.close();
             break;
           case _WebSocketOpcode.PING:
-            // TODO(ajohnsen): Handle ping.
+            sink.add(new _WebSocketPing());
             break;
           case _WebSocketOpcode.PONG:
-            // TODO(ajohnsen): Handle pong.
+            sink.add(new _WebSocketPong());
             break;
         }
         _prepareForNextFrame();
       } else {
         _messageFrameEnd(sink);
       }
     } else {
       _state = PAYLOAD;
     }
   }
 
   void _messageFrameEnd(EventSink sink) {
     if (_fin) {
       switch (_currentMessageType) {
         case _WebSocketMessageType.TEXT:
-          sink.add(_buffer.toString());
+          _controller.close();
           break;
         case _WebSocketMessageType.BINARY:
-          sink.add(_buffer.readBytes());
+          _controller.close();
           break;
       }
-      _buffer = null;
+      _controller = null;
       _currentMessageType = _WebSocketMessageType.NONE;
     }
     _prepareForNextFrame();
   }
 
   void _controlFrameEnd(EventSink sink) {
     switch (_opcode) {
       case _WebSocketOpcode.CLOSE:
         closeCode = WebSocketStatus.NO_STATUS_RECEIVED;
         if (_controlPayload.length > 0) {
           if (_controlPayload.length == 1) {
             throw new WebSocketException("Protocol error");
           }
           closeCode = _controlPayload[0] << 8 | _controlPayload[1];
           if (closeCode == WebSocketStatus.NO_STATUS_RECEIVED) {
             throw new WebSocketException("Protocol error");
           }
           if (_controlPayload.length > 2) {
-            closeReason = _decodeString(
-                _controlPayload.sublist(2));
+            closeReason = _decodeUtf8Strict(_controlPayload.sublist(2));
           }
         }
         _state = CLOSED;
         sink.close();
         break;
 
       case _WebSocketOpcode.PING:
-        // TODO(ajohnsen): Handle ping.
+        sink.add(new _WebSocketPing(_controlPayload));
         break;
 
       case _WebSocketOpcode.PONG:
-        // TODO(ajohnsen): Handle pong.
+        sink.add(new _WebSocketPong(_controlPayload));
         break;
     }
     _prepareForNextFrame();
   }
 
   bool _isControlFrame() {
     return _opcode == _WebSocketOpcode.CLOSE ||
            _opcode == _WebSocketOpcode.PING ||
            _opcode == _WebSocketOpcode.PONG;
   }
@@ skipping 18 matching lines @@
   int _len;
   bool _masked;
   int _maskingKey;
   int _remainingLenBytes;
   int _remainingMaskingKeyBytes;
   int _remainingPayloadBytes;
   int _unmaskingIndex;
 
   int _currentMessageType;
   List<int> _controlPayload;
-  var _buffer;  // Either StringBuffer or _BufferList.
+  StreamController _controller;
 
   int closeCode = WebSocketStatus.NO_STATUS_RECEIVED;
   String closeReason = "";
 }
 
 
+class _WebSocketPing {
+  final List<int> payload;
+  _WebSocketPing([this.payload = null]);
+}
+
+
+class _WebSocketPong {
+  final List<int> payload;
+  _WebSocketPong([this.payload = null]);
+}
+
+
 class _WebSocketTransformerImpl implements WebSocketTransformer {
   final StreamController<WebSocket> _controller =
       new StreamController<WebSocket>();
 
   Stream<WebSocket> bind(Stream<HttpRequest> stream) {
     stream.listen((request) {
         _upgrade(request)
             .then((WebSocket webSocket) => _controller.add(webSocket))
             .catchError((error) => _controller.addError(error));
     });
@@ skipping 56 matching lines @@
   }
 }
 
 
 class _WebSocketOutgoingTransformer extends StreamEventTransformer {
   final _WebSocketImpl webSocket;
 
   _WebSocketOutgoingTransformer(_WebSocketImpl this.webSocket);
 
   void handleData(message, EventSink<List<int>> sink) {
+    if (message is _WebSocketPong) {
+      addFrame(_WebSocketOpcode.PONG, message.payload, sink);
+      return;
+    }
+    if (message is _WebSocketPing) {
+      addFrame(_WebSocketOpcode.PONG, message.payload, sink);
+      return;
+    }
     List<int> data;
     int opcode;
     if (message != null) {
       if (message is String) {
         opcode = _WebSocketOpcode.TEXT;
         data = _encodeString(message);
       } else {
         if (message is !List<int>) {
           throw new ArgumentError(message);
         }
@@ skipping 16 matching lines @@
       data.add(code & 0xFF);
       if (reason != null) {
         data.addAll(_encodeString(reason));
       }
     }
     addFrame(_WebSocketOpcode.CLOSE, data, sink);
     sink.close();
   }
 
   void addFrame(int opcode, List<int> data, EventSink<List<int>> sink) {
-    bool mask = !webSocket._serverSide;  // Masking not implemented for server.
+    createFrame(opcode, data, webSocket._serverSide).forEach(sink.add);
+  }
+
+  static Iterator createFrame(int opcode, List<int> data, bool serverSide) {
+    bool mask = !serverSide;  // Masking not implemented for server.
     int dataLength = data == null ? 0 : data.length;
     // Determine the header size.
     int headerSize = (mask) ? 6 : 2;
     if (dataLength > 65535) {
       headerSize += 8;
     } else if (dataLength > 125) {
       headerSize += 2;
     }
     List<int> header = new List<int>(headerSize);
     int index = 0;
@@ skipping 20 matching lines @@
       index += 4;
       if (data != null) {
         var list = new Uint8List(data.length);
         for (int i = 0; i < data.length; i++) {
           list[i] = data[i] ^ maskBytes[i % 4];
         }
         data = list;
       }
     }
     assert(index == headerSize);
-    sink.add(header);
-    if (data != null) {
-      sink.add(data);
+    if (data == null) {
+      return [header];
+    } else {
+      return [header, data];
     }
   }
 }
 
 
 class _WebSocketConsumer implements StreamConsumer {
   final _WebSocketImpl webSocket;
   final Socket socket;
   StreamController _controller;
   StreamSubscription _subscription;
@@ skipping 50 matching lines @@
           _done();
         },
         onError: (error) {
           _done(error);
         },
         cancelOnError: true);
     return _completer.future;
   }
 
   Future close() {
+    _ensureController();
     Future closeSocket() {
       return socket.close().then((_) => webSocket);
     }
-    if (_controller == null) return closeSocket();
     _controller.close();
     return _closeCompleter.future.then((_) => closeSocket());
   }
+
+  void add(data) {
+    _ensureController();
+    _controller.add(data);
+  }
 }
 
 
 class _WebSocketImpl extends Stream implements WebSocket {
   final StreamController _controller = new StreamController();
   StreamSink _sink;
 
   final Socket _socket;
   final bool _serverSide;
   int _readyState = WebSocket.CONNECTING;
@@ skipping 71 matching lines @@
             error("Bad response 'Sec-WebSocket-Accept' header");
           }
         }
         return response.detachSocket()
             .then((socket) => new _WebSocketImpl._fromSocket(socket));
       });
   }
 
   _WebSocketImpl._fromSocket(Socket this._socket,
                              [bool this._serverSide = false]) {
-    _sink = new _StreamSinkImpl(new _WebSocketConsumer(this, _socket));
+    var consumer = new _WebSocketConsumer(this, _socket);
+    _sink = new _StreamSinkImpl(consumer);
     _readyState = WebSocket.OPEN;
 
     var transformer = new _WebSocketProtocolTransformer(_serverSide);
     _socket.transform(transformer).listen(
         (data) {
-          _controller.add(data);
+          if (data is _WebSocketPing) {
+            consumer.add(new _WebSocketPong(data.payload));
+          } else if (data is _WebSocketPong) {
+            // TODO(ajohnsen): Notify pong?
+          } else {
+            _controller.add(data);
+          }
         },
         onError: (error) {
+          if (error is ArgumentError) {
+            close(WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
+          } else {
+            close(WebSocketStatus.PROTOCOL_ERROR);
+          }
           _controller.addError(error);
           _controller.close();
         },
         onDone: () {
           if (_readyState == WebSocket.OPEN) {
             _readyState = WebSocket.CLOSING;
-            if (transformer.closeCode != WebSocketStatus.NO_STATUS_RECEIVED) {
+            if (!_isReservedStatusCode(transformer.closeCode)) {
               close(transformer.closeCode);
             } else {
               close();
             }
             _readyState = WebSocket.CLOSED;
           }
           _closeCode = transformer.closeCode;
           _closeReason = transformer.closeReason;
           _controller.close();
         },
@@ skipping 17 matching lines @@
   int get closeCode => _closeCode;
   String get closeReason => _closeReason;
 
   void add(data) => _sink.add(data);
   void addError(error) => _sink.addError(error);
   Future addStream(Stream stream) => _sink.addStream(stream);
   Future get done => _sink.done;
 
   Future close([int code, String reason]) {
     if (!_writeClosed) {
-      if (code == WebSocketStatus.RESERVED_1004 ||
-          code == WebSocketStatus.NO_STATUS_RECEIVED ||
-          code == WebSocketStatus.RESERVED_1015) {
+      if (_isReservedStatusCode(code)) {
         throw new WebSocketException("Reserved status code $code");
       }
       _outCloseCode = code;
       _outCloseReason = reason;
       _writeClosed = true;
     }
     return _sink.close();
   }
+
+  static bool _isReservedStatusCode(int code) {
+    return code != null &&
+           (code < WebSocketStatus.NORMAL_CLOSURE ||
+            code == WebSocketStatus.RESERVED_1004 ||
+            code == WebSocketStatus.NO_STATUS_RECEIVED ||
+            code == WebSocketStatus.ABNORMAL_CLOSURE ||
+            (code > WebSocketStatus.INTERNAL_SERVER_ERROR &&
+             code < WebSocketStatus.RESERVED_1015) ||
+            (code >= WebSocketStatus.RESERVED_1015 &&
+             code < 3000));
+  }
 }