Websocket.php 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492
  1. <?php
  2. /**
  3. * This file is part of workerman.
  4. *
  5. * Licensed under The MIT License
  6. * For full copyright and license information, please see the MIT-LICENSE.txt
  7. * Redistributions of files must retain the above copyright notice.
  8. *
  9. * @author walkor<walkor@workerman.net>
  10. * @copyright walkor<walkor@workerman.net>
  11. * @link http://www.workerman.net/
  12. * @license http://www.opensource.org/licenses/mit-license.php MIT License
  13. */
  14. namespace Workerman\Protocols;
  15. use Workerman\Connection\ConnectionInterface;
  16. use Workerman\Connection\TcpConnection;
  17. use Workerman\Worker;
  18. /**
  19. * WebSocket protocol.
  20. */
  21. class Websocket implements \Workerman\Protocols\ProtocolInterface
  22. {
  23. /**
  24. * Websocket blob type.
  25. *
  26. * @var string
  27. */
  28. const BINARY_TYPE_BLOB = "\x81";
  29. /**
  30. * Websocket arraybuffer type.
  31. *
  32. * @var string
  33. */
  34. const BINARY_TYPE_ARRAYBUFFER = "\x82";
  35. /**
  36. * Check the integrity of the package.
  37. *
  38. * @param string $buffer
  39. * @param ConnectionInterface $connection
  40. * @return int
  41. */
  42. public static function input($buffer, ConnectionInterface $connection)
  43. {
  44. // Receive length.
  45. $recv_len = \strlen($buffer);
  46. // We need more data.
  47. if ($recv_len < 6) {
  48. return 0;
  49. }
  50. // Has not yet completed the handshake.
  51. if (empty($connection->websocketHandshake)) {
  52. return static::dealHandshake($buffer, $connection);
  53. }
  54. // Buffer websocket frame data.
  55. if ($connection->websocketCurrentFrameLength) {
  56. // We need more frame data.
  57. if ($connection->websocketCurrentFrameLength > $recv_len) {
  58. // Return 0, because it is not clear the full packet length, waiting for the frame of fin=1.
  59. return 0;
  60. }
  61. } else {
  62. $firstbyte = \ord($buffer[0]);
  63. $secondbyte = \ord($buffer[1]);
  64. $data_len = $secondbyte & 127;
  65. $is_fin_frame = $firstbyte >> 7;
  66. $masked = $secondbyte >> 7;
  67. if (!$masked) {
  68. Worker::safeEcho("frame not masked so close the connection\n");
  69. $connection->close();
  70. return 0;
  71. }
  72. $opcode = $firstbyte & 0xf;
  73. switch ($opcode) {
  74. case 0x0:
  75. break;
  76. // Blob type.
  77. case 0x1:
  78. break;
  79. // Arraybuffer type.
  80. case 0x2:
  81. break;
  82. // Close package.
  83. case 0x8:
  84. // Try to emit onWebSocketClose callback.
  85. if (isset($connection->onWebSocketClose) || isset($connection->worker->onWebSocketClose)) {
  86. try {
  87. \call_user_func(isset($connection->onWebSocketClose)?$connection->onWebSocketClose:$connection->worker->onWebSocketClose, $connection);
  88. } catch (\Exception $e) {
  89. Worker::stopAll(250, $e);
  90. } catch (\Error $e) {
  91. Worker::stopAll(250, $e);
  92. }
  93. } // Close connection.
  94. else {
  95. $connection->close("\x88\x02\x03\xe8", true);
  96. }
  97. return 0;
  98. // Ping package.
  99. case 0x9:
  100. break;
  101. // Pong package.
  102. case 0xa:
  103. break;
  104. // Wrong opcode.
  105. default :
  106. Worker::safeEcho("error opcode $opcode and close websocket connection. Buffer:" . bin2hex($buffer) . "\n");
  107. $connection->close();
  108. return 0;
  109. }
  110. // Calculate packet length.
  111. $head_len = 6;
  112. if ($data_len === 126) {
  113. $head_len = 8;
  114. if ($head_len > $recv_len) {
  115. return 0;
  116. }
  117. $pack = \unpack('nn/ntotal_len', $buffer);
  118. $data_len = $pack['total_len'];
  119. } else {
  120. if ($data_len === 127) {
  121. $head_len = 14;
  122. if ($head_len > $recv_len) {
  123. return 0;
  124. }
  125. $arr = \unpack('n/N2c', $buffer);
  126. $data_len = $arr['c1']*4294967296 + $arr['c2'];
  127. }
  128. }
  129. $current_frame_length = $head_len + $data_len;
  130. $total_package_size = \strlen($connection->websocketDataBuffer) + $current_frame_length;
  131. if ($total_package_size > $connection->maxPackageSize) {
  132. Worker::safeEcho("error package. package_length=$total_package_size\n");
  133. $connection->close();
  134. return 0;
  135. }
  136. if ($is_fin_frame) {
  137. if ($opcode === 0x9) {
  138. if ($recv_len >= $current_frame_length) {
  139. $ping_data = static::decode(\substr($buffer, 0, $current_frame_length), $connection);
  140. $connection->consumeRecvBuffer($current_frame_length);
  141. $tmp_connection_type = isset($connection->websocketType) ? $connection->websocketType : static::BINARY_TYPE_BLOB;
  142. $connection->websocketType = "\x8a";
  143. if (isset($connection->onWebSocketPing) || isset($connection->worker->onWebSocketPing)) {
  144. try {
  145. \call_user_func(isset($connection->onWebSocketPing)?$connection->onWebSocketPing:$connection->worker->onWebSocketPing, $connection, $ping_data);
  146. } catch (\Exception $e) {
  147. Worker::stopAll(250, $e);
  148. } catch (\Error $e) {
  149. Worker::stopAll(250, $e);
  150. }
  151. } else {
  152. $connection->send($ping_data);
  153. }
  154. $connection->websocketType = $tmp_connection_type;
  155. if ($recv_len > $current_frame_length) {
  156. return static::input(\substr($buffer, $current_frame_length), $connection);
  157. }
  158. }
  159. return 0;
  160. } else if ($opcode === 0xa) {
  161. if ($recv_len >= $current_frame_length) {
  162. $pong_data = static::decode(\substr($buffer, 0, $current_frame_length), $connection);
  163. $connection->consumeRecvBuffer($current_frame_length);
  164. $tmp_connection_type = isset($connection->websocketType) ? $connection->websocketType : static::BINARY_TYPE_BLOB;
  165. $connection->websocketType = "\x8a";
  166. // Try to emit onWebSocketPong callback.
  167. if (isset($connection->onWebSocketPong) || isset($connection->worker->onWebSocketPong)) {
  168. try {
  169. \call_user_func(isset($connection->onWebSocketPong)?$connection->onWebSocketPong:$connection->worker->onWebSocketPong, $connection, $pong_data);
  170. } catch (\Exception $e) {
  171. Worker::stopAll(250, $e);
  172. } catch (\Error $e) {
  173. Worker::stopAll(250, $e);
  174. }
  175. }
  176. $connection->websocketType = $tmp_connection_type;
  177. if ($recv_len > $current_frame_length) {
  178. return static::input(\substr($buffer, $current_frame_length), $connection);
  179. }
  180. }
  181. return 0;
  182. }
  183. return $current_frame_length;
  184. } else {
  185. $connection->websocketCurrentFrameLength = $current_frame_length;
  186. }
  187. }
  188. // Received just a frame length data.
  189. if ($connection->websocketCurrentFrameLength === $recv_len) {
  190. static::decode($buffer, $connection);
  191. $connection->consumeRecvBuffer($connection->websocketCurrentFrameLength);
  192. $connection->websocketCurrentFrameLength = 0;
  193. return 0;
  194. } // The length of the received data is greater than the length of a frame.
  195. elseif ($connection->websocketCurrentFrameLength < $recv_len) {
  196. static::decode(\substr($buffer, 0, $connection->websocketCurrentFrameLength), $connection);
  197. $connection->consumeRecvBuffer($connection->websocketCurrentFrameLength);
  198. $current_frame_length = $connection->websocketCurrentFrameLength;
  199. $connection->websocketCurrentFrameLength = 0;
  200. // Continue to read next frame.
  201. return static::input(\substr($buffer, $current_frame_length), $connection);
  202. } // The length of the received data is less than the length of a frame.
  203. else {
  204. return 0;
  205. }
  206. }
  207. /**
  208. * Websocket encode.
  209. *
  210. * @param string $buffer
  211. * @param ConnectionInterface $connection
  212. * @return string
  213. */
  214. public static function encode($buffer, ConnectionInterface $connection)
  215. {
  216. if (!is_scalar($buffer)) {
  217. throw new \Exception("You can't send(" . \gettype($buffer) . ") to client, you need to convert it to a string. ");
  218. }
  219. $len = \strlen($buffer);
  220. if (empty($connection->websocketType)) {
  221. $connection->websocketType = static::BINARY_TYPE_BLOB;
  222. }
  223. $first_byte = $connection->websocketType;
  224. if ($len <= 125) {
  225. $encode_buffer = $first_byte . \chr($len) . $buffer;
  226. } else {
  227. if ($len <= 65535) {
  228. $encode_buffer = $first_byte . \chr(126) . \pack("n", $len) . $buffer;
  229. } else {
  230. $encode_buffer = $first_byte . \chr(127) . \pack("xxxxN", $len) . $buffer;
  231. }
  232. }
  233. // Handshake not completed so temporary buffer websocket data waiting for send.
  234. if (empty($connection->websocketHandshake)) {
  235. if (empty($connection->tmpWebsocketData)) {
  236. $connection->tmpWebsocketData = '';
  237. }
  238. // If buffer has already full then discard the current package.
  239. if (\strlen($connection->tmpWebsocketData) > $connection->maxSendBufferSize) {
  240. if ($connection->onError) {
  241. try {
  242. \call_user_func($connection->onError, $connection, \WORKERMAN_SEND_FAIL, 'send buffer full and drop package');
  243. } catch (\Exception $e) {
  244. Worker::stopAll(250, $e);
  245. } catch (\Error $e) {
  246. Worker::stopAll(250, $e);
  247. }
  248. }
  249. return '';
  250. }
  251. $connection->tmpWebsocketData .= $encode_buffer;
  252. // Check buffer is full.
  253. if ($connection->maxSendBufferSize <= \strlen($connection->tmpWebsocketData)) {
  254. if ($connection->onBufferFull) {
  255. try {
  256. \call_user_func($connection->onBufferFull, $connection);
  257. } catch (\Exception $e) {
  258. Worker::stopAll(250, $e);
  259. } catch (\Error $e) {
  260. Worker::stopAll(250, $e);
  261. }
  262. }
  263. }
  264. // Return empty string.
  265. return '';
  266. }
  267. return $encode_buffer;
  268. }
  269. /**
  270. * Websocket decode.
  271. *
  272. * @param string $buffer
  273. * @param ConnectionInterface $connection
  274. * @return string
  275. */
  276. public static function decode($buffer, ConnectionInterface $connection)
  277. {
  278. $len = \ord($buffer[1]) & 127;
  279. if ($len === 126) {
  280. $masks = \substr($buffer, 4, 4);
  281. $data = \substr($buffer, 8);
  282. } else {
  283. if ($len === 127) {
  284. $masks = \substr($buffer, 10, 4);
  285. $data = \substr($buffer, 14);
  286. } else {
  287. $masks = \substr($buffer, 2, 4);
  288. $data = \substr($buffer, 6);
  289. }
  290. }
  291. $dataLength = \strlen($data);
  292. $masks = \str_repeat($masks, \floor($dataLength / 4)) . \substr($masks, 0, $dataLength % 4);
  293. $decoded = $data ^ $masks;
  294. if ($connection->websocketCurrentFrameLength) {
  295. $connection->websocketDataBuffer .= $decoded;
  296. return $connection->websocketDataBuffer;
  297. } else {
  298. if ($connection->websocketDataBuffer !== '') {
  299. $decoded = $connection->websocketDataBuffer . $decoded;
  300. $connection->websocketDataBuffer = '';
  301. }
  302. return $decoded;
  303. }
  304. }
  305. /**
  306. * Websocket handshake.
  307. *
  308. * @param string $buffer
  309. * @param TcpConnection $connection
  310. * @return int
  311. */
  312. public static function dealHandshake($buffer, TcpConnection $connection)
  313. {
  314. // HTTP protocol.
  315. if (0 === \strpos($buffer, 'GET')) {
  316. // Find \r\n\r\n.
  317. $heder_end_pos = \strpos($buffer, "\r\n\r\n");
  318. if (!$heder_end_pos) {
  319. return 0;
  320. }
  321. $header_length = $heder_end_pos + 4;
  322. // Get Sec-WebSocket-Key.
  323. $Sec_WebSocket_Key = '';
  324. if (\preg_match("/Sec-WebSocket-Key: *(.*?)\r\n/i", $buffer, $match)) {
  325. $Sec_WebSocket_Key = $match[1];
  326. } else {
  327. $connection->close("HTTP/1.1 200 WebSocket\r\nServer: workerman/".Worker::VERSION."\r\n\r\n<div style=\"text-align:center\"><h1>WebSocket</h1><hr>workerman/".Worker::VERSION."</div>",
  328. true);
  329. return 0;
  330. }
  331. // Calculation websocket key.
  332. $new_key = \base64_encode(\sha1($Sec_WebSocket_Key . "258EAFA5-E914-47DA-95CA-C5AB0DC85B11", true));
  333. // Handshake response data.
  334. $handshake_message = "HTTP/1.1 101 Switching Protocols\r\n"
  335. ."Upgrade: websocket\r\n"
  336. ."Sec-WebSocket-Version: 13\r\n"
  337. ."Connection: Upgrade\r\n"
  338. ."Sec-WebSocket-Accept: " . $new_key . "\r\n";
  339. // Websocket data buffer.
  340. $connection->websocketDataBuffer = '';
  341. // Current websocket frame length.
  342. $connection->websocketCurrentFrameLength = 0;
  343. // Current websocket frame data.
  344. $connection->websocketCurrentFrameBuffer = '';
  345. // Consume handshake data.
  346. $connection->consumeRecvBuffer($header_length);
  347. // blob or arraybuffer
  348. if (empty($connection->websocketType)) {
  349. $connection->websocketType = static::BINARY_TYPE_BLOB;
  350. }
  351. $has_server_header = false;
  352. if (isset($connection->headers)) {
  353. if (\is_array($connection->headers)) {
  354. foreach ($connection->headers as $header) {
  355. if (\strpos($header, 'Server:') === 0) {
  356. $has_server_header = true;
  357. }
  358. $handshake_message .= "$header\r\n";
  359. }
  360. } else {
  361. $handshake_message .= "$connection->headers\r\n";
  362. }
  363. }
  364. if (!$has_server_header) {
  365. $handshake_message .= "Server: workerman/".Worker::VERSION."\r\n";
  366. }
  367. $handshake_message .= "\r\n";
  368. // Send handshake response.
  369. $connection->send($handshake_message, true);
  370. // Mark handshake complete..
  371. $connection->websocketHandshake = true;
  372. // Try to emit onWebSocketConnect callback.
  373. $on_websocket_connect = isset($connection->onWebSocketConnect) ? $connection->onWebSocketConnect :
  374. (isset($connection->worker->onWebSocketConnect) ? $connection->worker->onWebSocketConnect : false);
  375. if ($on_websocket_connect) {
  376. static::parseHttpHeader($buffer);
  377. try {
  378. \call_user_func($on_websocket_connect, $connection, $buffer);
  379. } catch (\Exception $e) {
  380. Worker::stopAll(250, $e);
  381. } catch (\Error $e) {
  382. Worker::stopAll(250, $e);
  383. }
  384. if (!empty($_SESSION) && \class_exists('\GatewayWorker\Lib\Context')) {
  385. $connection->session = \GatewayWorker\Lib\Context::sessionEncode($_SESSION);
  386. }
  387. $_GET = $_SERVER = $_SESSION = $_COOKIE = array();
  388. }
  389. // There are data waiting to be sent.
  390. if (!empty($connection->tmpWebsocketData)) {
  391. $connection->send($connection->tmpWebsocketData, true);
  392. $connection->tmpWebsocketData = '';
  393. }
  394. if (\strlen($buffer) > $header_length) {
  395. return static::input(\substr($buffer, $header_length), $connection);
  396. }
  397. return 0;
  398. } // Is flash policy-file-request.
  399. elseif (0 === \strpos($buffer, '<polic')) {
  400. $policy_xml = '<?xml version="1.0"?><cross-domain-policy><site-control permitted-cross-domain-policies="all"/><allow-access-from domain="*" to-ports="*"/></cross-domain-policy>' . "\0";
  401. $connection->send($policy_xml, true);
  402. $connection->consumeRecvBuffer(\strlen($buffer));
  403. return 0;
  404. }
  405. // Bad websocket handshake request.
  406. $connection->close("HTTP/1.1 200 WebSocket\r\nServer: workerman/".Worker::VERSION."\r\n\r\n<div style=\"text-align:center\"><h1>WebSocket</h1><hr>workerman/".Worker::VERSION."</div>",
  407. true);
  408. return 0;
  409. }
  410. /**
  411. * Parse http header.
  412. *
  413. * @param string $buffer
  414. * @return void
  415. */
  416. protected static function parseHttpHeader($buffer)
  417. {
  418. // Parse headers.
  419. list($http_header, ) = \explode("\r\n\r\n", $buffer, 2);
  420. $header_data = \explode("\r\n", $http_header);
  421. if ($_SERVER) {
  422. $_SERVER = array();
  423. }
  424. list($_SERVER['REQUEST_METHOD'], $_SERVER['REQUEST_URI'], $_SERVER['SERVER_PROTOCOL']) = \explode(' ',
  425. $header_data[0]);
  426. unset($header_data[0]);
  427. foreach ($header_data as $content) {
  428. // \r\n\r\n
  429. if (empty($content)) {
  430. continue;
  431. }
  432. list($key, $value) = \explode(':', $content, 2);
  433. $key = \str_replace('-', '_', \strtoupper($key));
  434. $value = \trim($value);
  435. $_SERVER['HTTP_' . $key] = $value;
  436. switch ($key) {
  437. // HTTP_HOST
  438. case 'HOST':
  439. $tmp = \explode(':', $value);
  440. $_SERVER['SERVER_NAME'] = $tmp[0];
  441. if (isset($tmp[1])) {
  442. $_SERVER['SERVER_PORT'] = $tmp[1];
  443. }
  444. break;
  445. // cookie
  446. case 'COOKIE':
  447. \parse_str(\str_replace('; ', '&', $_SERVER['HTTP_COOKIE']), $_COOKIE);
  448. break;
  449. }
  450. }
  451. // QUERY_STRING
  452. $_SERVER['QUERY_STRING'] = \parse_url($_SERVER['REQUEST_URI'], \PHP_URL_QUERY);
  453. if ($_SERVER['QUERY_STRING']) {
  454. // $GET
  455. \parse_str($_SERVER['QUERY_STRING'], $_GET);
  456. } else {
  457. $_SERVER['QUERY_STRING'] = '';
  458. }
  459. }
  460. }