Merge pull request #4 from walkor/master

Sync master

Connection/AsyncTcpConnection.php

@@ -174,13 +174,18 @@ class AsyncTcpConnection extends TcpConnection
         }
         $this->_status           = self::STATUS_CONNECTING;
         $this->_connectStartTime = microtime(true);
-        // Open socket connection asynchronously.
-        if ($this->_contextOption) {
-            $context = stream_context_create($this->_contextOption);
-            $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteHost}:{$this->_remotePort}", $errno, $errstr, 0,
-                STREAM_CLIENT_ASYNC_CONNECT, $context);
+        if ($this->transport !== 'unix') {
+            // Open socket connection asynchronously.
+            if ($this->_contextOption) {
+                $context = stream_context_create($this->_contextOption);
+                $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteHost}:{$this->_remotePort}",
+                    $errno, $errstr, 0, STREAM_CLIENT_ASYNC_CONNECT, $context);
+            } else {
+                $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteHost}:{$this->_remotePort}",
+                    $errno, $errstr, 0, STREAM_CLIENT_ASYNC_CONNECT);
+            }
         } else {
-            $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteHost}:{$this->_remotePort}", $errno, $errstr, 0,
+            $this->_socket = stream_socket_client("{$this->transport}://{$this->_remoteAddress}", $errno, $errstr, 0,
                 STREAM_CLIENT_ASYNC_CONNECT);
         }
         // If failed attempt to emit onError callback.

Connection/ConnectionInterface.php

@@ -81,21 +81,21 @@ abstract class  ConnectionInterface
     abstract public function getRemoteAddress();
 
     /**
-     * Get remote IP.
+     * Get local IP.
      *
      * @return string
      */
     abstract public function getLocalIp();
 
     /**
-     * Get remote port.
+     * Get local port.
      *
      * @return int
      */
     abstract public function getLocalPort();
 
     /**
-     * Get remote address.
+     * Get local address.
      *
      * @return string
      */

Connection/TcpConnection.php

@@ -254,6 +254,31 @@ class TcpConnection extends ConnectionInterface
         self::STATUS_CLOSED      => 'CLOSED',
     );
 
+
+    /**
+     * Adding support of custom functions within protocols
+     *
+     * @param string $name
+     * @param array  $arguments
+     */
+    public function __call($name, $arguments) {
+        // Try to emit custom function within protocol
+        if (method_exists($this->protocol, $name)) {
+            try {
+                return call_user_func(array($this->protocol, $name), $this, $arguments);
+            } catch (\Exception $e) {
+                Worker::log($e);
+                exit(250);
+            } catch (\Error $e) {
+                Worker::log($e);
+                exit(250);
+            }
+	} else {
+	    trigger_error('Call to undefined method '.__CLASS__.'::'.$name.'()', E_USER_ERROR);
+	}
+
+    }
+
     /**
      * Construct.
      *
@@ -538,8 +563,8 @@ class TcpConnection extends ConnectionInterface
     {
         // SSL handshake.
         if ($this->transport === 'ssl' && $this->_sslHandshakeCompleted !== true) {
-            $ret = stream_socket_enable_crypto($socket, true, STREAM_CRYPTO_METHOD_SSLv2_SERVER |
-                STREAM_CRYPTO_METHOD_SSLv3_SERVER | STREAM_CRYPTO_METHOD_SSLv23_SERVER);
+            $ret = stream_socket_enable_crypto($socket, true, STREAM_CRYPTO_METHOD_SSLv2_SERVER | 
+					       STREAM_CRYPTO_METHOD_SSLv23_SERVER);
             // Negotiation has failed.
             if(false === $ret) {
                 if (!feof($socket)) {

Events/Select.php

@@ -265,7 +265,7 @@ class Select implements EventInterface
 
             $read  = $this->_readFds;
             $write = $this->_writeFds;
-            $except = $this->_writeFds;
+            $except = $this->_exceptFds;
 
             // Waiting read/write/signal/timeout events.
             $ret = @stream_select($read, $write, $except, 0, $this->_selectTimeout);

Lib/Timer.php

@@ -54,7 +54,9 @@ class Timer
         if ($event) {
             self::$_event = $event;
         } else {
-            pcntl_signal(SIGALRM, array('\Workerman\Lib\Timer', 'signalHandle'), false);
+            if (function_exists('pcntl_signal')) {
+                pcntl_signal(SIGALRM, array('\Workerman\Lib\Timer', 'signalHandle'), false);
+            }
         }
     }
 
@@ -74,8 +76,8 @@ class Timer
     /**
      * Add a timer.
      *
-     * @param int      $time_interval
-     * @param callback $func
+     * @param float    $time_interval
+     * @param callable $func
      * @param mixed    $args
      * @param bool     $persistent
      * @return int/false

Protocols/Http.php

@@ -168,6 +168,9 @@ class Http
                     case 'multipart/form-data':
                         self::parseUploadFiles($http_body, $http_post_boundary);
                         break;
+                    case 'application/json':
+                        $_POST = json_decode($http_body, true);
+                        break;
                     case 'application/x-www-form-urlencoded':
                         parse_str($http_body, $_POST);
                         break;

Protocols/Websocket.php

@@ -48,7 +48,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
         // Receive length.
         $recv_len = strlen($buffer);
         // We need more data.
-        if ($recv_len < 2) {
+        if ($recv_len < 6) {
             return 0;
         }
 
@@ -70,6 +70,13 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
             $data_len     = $secondbyte & 127;
             $is_fin_frame = $firstbyte >> 7;
             $masked       = $secondbyte >> 7;
+
+            if (!$masked) {
+                echo "frame not masked\n";
+                $connection->close();
+                return 0;
+            }
+
             $opcode       = $firstbyte & 0xf;
             switch ($opcode) {
                 case 0x0:
@@ -118,7 +125,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
 
                     // Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 6;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return static::input(substr($buffer, $head_len), $connection);
@@ -142,7 +149,7 @@ class Websocket implements \Workerman\Protocols\ProtocolInterface
                     }
                     //  Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 6;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return static::input(substr($buffer, $head_len), $connection);

Protocols/Ws.php

@@ -71,6 +71,13 @@ class Ws
             $data_len     = $secondbyte & 127;
             $is_fin_frame = $firstbyte >> 7;
             $masked       = $secondbyte >> 7;
+
+            if ($masked) {
+                echo "frame masked\n";
+                $connection->close();
+                return 0;
+            }
+
             $opcode       = $firstbyte & 0xf;
 
             switch ($opcode) {
@@ -119,7 +126,7 @@ class Ws
                     }
                     // Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 2;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return self::input(substr($buffer, $head_len), $connection);
@@ -143,7 +150,7 @@ class Ws
                     }
                     //  Consume data from receive buffer.
                     if (!$data_len) {
-                        $head_len = $masked ? 6 : 2;
+                        $head_len = 2;
                         $connection->consumeRecvBuffer($head_len);
                         if ($recv_len > $head_len) {
                             return self::input(substr($buffer, $head_len), $connection);
@@ -159,7 +166,7 @@ class Ws
             }
             // Calculate packet length.
             if ($data_len === 126) {
-                if (strlen($buffer) < 6) {
+                if (strlen($buffer) < 4) {
                     return 0;
                 }
                 $pack = unpack('nn/ntotal_len', $buffer);
@@ -289,31 +296,14 @@ class Ws
      */
     public static function decode($bytes, $connection)
     {
-        $masked = ord($bytes[1]) >> 7;
-        $data_length = $masked ? ord($bytes[1]) & 127 : ord($bytes[1]);
-        $decoded_data = '';
-        if ($masked === true) {
-            if ($data_length === 126) {
-                $mask = substr($bytes, 4, 4);
-                $coded_data = substr($bytes, 8);
-            } else if ($data_length === 127) {
-                $mask = substr($bytes, 10, 4);
-                $coded_data = substr($bytes, 14);
-            } else {
-                $mask = substr($bytes, 2, 4);
-                $coded_data = substr($bytes, 6);
-            }
-            for ($i = 0; $i < strlen($coded_data); $i++) {
-                $decoded_data .= $coded_data[$i] ^ $mask[$i % 4];
-            }
+        $data_length = ord($bytes[1]);
+
+        if ($data_length === 126) {
+            $decoded_data = substr($bytes, 4);
+        } else if ($data_length === 127) {
+            $decoded_data = substr($bytes, 10);
         } else {
-            if ($data_length === 126) {
-                $decoded_data = substr($bytes, 4);
-            } else if ($data_length === 127) {
-                $decoded_data = substr($bytes, 10);
-            } else {
-                $decoded_data = substr($bytes, 2);
-            }
+            $decoded_data = substr($bytes, 2);
         }
         if ($connection->websocketCurrentFrameLength) {
             $connection->websocketDataBuffer .= $decoded_data;
@@ -369,13 +359,15 @@ class Ws
         $port = $connection->getRemotePort();
         $host = $port === 80 ? $connection->getRemoteHost() : $connection->getRemoteHost() . ':' . $port;
         // Handshake header.
+        $connection->websocketSecKey = base64_encode(md5(mt_rand(), true));
         $header = 'GET ' . $connection->getRemoteURI() . " HTTP/1.1\r\n".
         "Host: $host\r\n".
         "Connection: Upgrade\r\n".
         "Upgrade: websocket\r\n".
         "Origin: ". (isset($connection->websocketOrigin) ? $connection->websocketOrigin : '*') ."\r\n".
+        (isset($connection->WSClientProtocol)?"Sec-WebSocket-Protocol: ".$connection->WSClientProtocol."\r\n":'').
         "Sec-WebSocket-Version: 13\r\n".
-        "Sec-WebSocket-Key: " . base64_encode(md5(mt_rand(), true)) . "\r\n\r\n";
+        "Sec-WebSocket-Key: " . $connection->websocketSecKey . "\r\n\r\n";
         $connection->send($header, true);
         $connection->handshakeStep               = 1;
         $connection->websocketCurrentFrameLength = 0;
@@ -394,7 +386,26 @@ class Ws
     {
         $pos = strpos($buffer, "\r\n\r\n");
         if ($pos) {
+            //checking Sec-WebSocket-Accept
+            if (preg_match("/Sec-WebSocket-Accept: *(.*?)\r\n/i", $buffer, $match)) {
+                if ($match[1] !== base64_encode(sha1($connection->websocketSecKey . "258EAFA5-E914-47DA-95CA-C5AB0DC85B11", true))) {
+                    echo "Sec-WebSocket-Accept not match. Header:\n" . substr($buffer, 0, $pos) . "\n";
+                    $connection->close();
+                    return 0;
+                }
+            } else {
+                echo "Sec-WebSocket-Accept not found. Header:\n" . substr($buffer, 0, $pos) . "\n";
+                $connection->close();
+                return 0;
+            }
+
             // handshake complete
+
+            // Get WebSocket subprotocol (if specified by server)
+            if (preg_match("/Sec-WebSocket-Protocol: *(.*?)\r\n/i", $buffer, $match)) {
+                $connection->WSServerProtocol = trim($match[1]);
+            }
+
             $connection->handshakeStep = 2;
             $handshake_response_length = $pos + 4;
             // Try to emit onWebSocketConnect callback.
@@ -430,4 +441,13 @@ class Ws
         }
         return 0;
     }
+
+    public static function WSSetProtocol($connection, $params) {
+	$connection->WSClientProtocol = $params[0];
+    }
+
+    public static function WSGetServerProtocol($connection) {
+	return (property_exists($connection, 'WSServerProtocol')?$connection->WSServerProtocol:null);
+    }
+
 }

composer.json

@@ -24,9 +24,7 @@
         "source": "https://github.com/walkor/workerman"
     },
     "require": {
-        "php": ">=5.3",
-        "ext-pcntl": "*",
-        "ext-posix": "*"
+        "php": ">=5.3"
     },
     "suggest": {
         "ext-event": "For better performance. "