websocket_server.php

<?php
	// CubicleSoft PHP WebSocketServer class.
	// (C) 2021 CubicleSoft.  All Rights Reserved.

	// Make sure PHP doesn't introduce weird limitations.
	ini_set("memory_limit", "-1");
	set_time_limit(0);

	// Requires the CubicleSoft PHP WebSocket class.
	class WebSocketServer
	{
		protected $fp, $clients, $nextclientid, $websocketclass, $origins;
		protected $defaultclosemode, $defaultmaxreadframesize, $defaultmaxreadmessagesize, $defaultkeepalive, $lasttimeoutcheck;

		public function __construct()
		{
			$this->Reset();
		}

		public function Reset()
		{
			if (!class_exists("WebSocket", false))  require_once str_replace("\\", "/", dirname(__FILE__)) . "/websocket.php";

			$this->fp = false;
			$this->clients = array();
			$this->nextclientid = 1;
			$this->websocketclass = "WebSocket";
			$this->origins = false;

			$this->defaultclosemode = WebSocket::CLOSE_IMMEDIATELY;
			$this->defaultmaxreadframesize = 2000000;
			$this->defaultmaxreadmessagesize = 10000000;
			$this->defaultkeepalive = 30;
			$this->lasttimeoutcheck = time();
		}

		public function __destruct()
		{
			$this->Stop();
		}

		public function SetWebSocketClass($newclass)
		{
			if (class_exists($newclass))  $this->websocketclass = $newclass;
		}

		public function SetAllowedOrigins($origins)
		{
			if (is_string($origins))  $origins = array($origins);
			if (!is_array($origins))  $origins = false;
			else if (isset($origins[0]))  $origins = array_flip($origins);

			$this->origins = $origins;
		}

		public function SetDefaultCloseMode($mode)
		{
			$this->defaultclosemode = $mode;
		}

		public function SetDefaultKeepAliveTimeout($keepalive)
		{
			$this->defaultkeepalive = (int)$keepalive;
		}

		public function SetDefaultMaxReadFrameSize($maxsize)
		{
			$this->defaultmaxreadframesize = (is_bool($maxsize) ? false : (int)$maxsize);
		}

		public function SetDefaultMaxReadMessageSize($maxsize)
		{
			$this->defaultmaxreadmessagesize = (is_bool($maxsize) ? false : (int)$maxsize);
		}

		// Starts the server on the host and port.
		// $host is usually 0.0.0.0 or 127.0.0.1 for IPv4 and [::0] or [::1] for IPv6.
		public function Start($host, $port)
		{
			$this->Stop();

			$this->fp = stream_socket_server("tcp://" . $host . ":" . $port, $errornum, $errorstr);
			if ($this->fp === false)  return array("success" => false, "error" => self::WSTranslate("Bind() failed.  Reason:  %s (%d)", $errorstr, $errornum), "errorcode" => "bind_failed");

			// Enable non-blocking mode.
			stream_set_blocking($this->fp, 0);

			return array("success" => true);
		}

		public function Stop()
		{
			foreach ($this->clients as $client)
			{
				if ($client->websocket !== false)  $client->websocket->Disconnect();
				else  fclose($client->fp);
			}

			$this->clients = array();

			if ($this->fp !== false)
			{
				fclose($this->fp);

				$this->fp = false;
			}

			$this->nextclientid = 1;
		}

		// Dangerous but allows for stream_select() calls on multiple, separate stream handles.
		public function GetStream()
		{
			return $this->fp;
		}

		// Return whatever response/headers are needed here.
		protected function ProcessNewConnection($method, $path, $client)
		{
			$result = "";

			if ($method !== "GET")  $result .= "HTTP/1.1 405 Method Not Allowed\r\nConnection: close\r\n\r\n";
			else if (!isset($client->headers["Host"]) || !isset($client->headers["Connection"]) || stripos($client->headers["Connection"], "upgrade") === false || !isset($client->headers["Upgrade"]) || stripos($client->headers["Upgrade"], "websocket") === false || !isset($client->headers["Sec-Websocket-Key"]))
			{
				$result .= "HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n";
			}
			else if (!isset($client->headers["Sec-Websocket-Version"]) || $client->headers["Sec-Websocket-Version"] != 13)
			{
				$result .= "HTTP/1.1 426 Upgrade Required\r\nSec-WebSocket-Version: 13\r\nConnection: close\r\n\r\n";
			}
			else if (!isset($client->headers["Origin"]) || ($this->origins !== false && !isset($this->origins[strtolower($client->headers["Origin"])])))
			{
				$result .= "HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n";
			}

			return $result;
		}

		// Return whatever additional HTTP headers are needed here.
		protected function ProcessAcceptedConnection($method, $path, $client)
		{
			return "";
		}

		protected function InitNewClient($fp)
		{
			$client = new stdClass();

			$client->id = $this->nextclientid;
			$client->readdata = "";
			$client->writedata = "";
			$client->request = false;
			$client->path = "";
			$client->url = "";
			$client->headers = array();
			$client->lastheader = "";
			$client->websocket = false;
			$client->fp = $fp;
			$client->ipaddr = stream_socket_get_name($fp, true);

			// Intended for application storage.
			$client->appdata = false;

			$this->clients[$this->nextclientid] = $client;

			$this->nextclientid++;

			return $client;
		}

		private function ProcessInitialResponse($method, $path, $client)
		{
			// Let a derived class handle the new connection (e.g. processing Origin and Host).
			// Since the 'websocketclass' is instantiated AFTER this function, it is possible to switch classes on the fly.
			$client->writedata .= $this->ProcessNewConnection($method, $path, $client);

			// If an error occurs, the connection will still terminate.
			$client->websocket = new $this->websocketclass();
			$client->websocket->SetCloseMode($this->defaultclosemode);
			$client->websocket->SetKeepAliveTimeout($this->defaultkeepalive);

			// If nothing was output, accept the connection.
			if ($client->writedata === "")
			{
				$client->writedata .= "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n";
				$client->writedata .= "Sec-WebSocket-Accept: " . base64_encode(sha1($client->headers["Sec-Websocket-Key"] . WebSocket::KEY_GUID, true)) . "\r\n";
				$client->writedata .= $this->ProcessAcceptedConnection($method, $path, $client);
				$client->writedata .= "\r\n";

				// Finish class initialization.
				$client->websocket->SetServerMode();
				$client->websocket->SetMaxReadFrameSize($this->defaultmaxreadframesize);
				$client->websocket->SetMaxReadMessageSize($this->defaultmaxreadmessagesize);

				// Set the socket in the WebSocket class.
				$client->websocket->Connect("", "", array("connected_fp" => $client->fp));
			}

			$this->UpdateClientState($client->id);
		}

		public function UpdateStreamsAndTimeout($prefix, &$timeout, &$readfps, &$writefps)
		{
			if ($this->fp !== false)  $readfps[$prefix . "ws_s"] = $this->fp;
			if ($timeout === false || $timeout > $this->defaultkeepalive)  $timeout = $this->defaultkeepalive;

			foreach ($this->clients as $id => $client)
			{
				if ($client->writedata === "")  $readfps[$prefix . "ws_c_" . $id] = $client->fp;

				if ($client->writedata !== "" || ($client->websocket !== false && $client->websocket->NeedsWrite()))  $writefps[$prefix . "ws_c_" . $id] = $client->fp;

				if ($client->websocket !== false)
				{
					$timeout2 = $client->websocket->GetKeepAliveTimeout();
					if ($timeout > $timeout2)  $timeout = $timeout2;
				}
			}
		}

		// Sometimes keyed arrays don't work properly.
		public static function FixedStreamSelect(&$readfps, &$writefps, &$exceptfps, $timeout)
		{
			// In order to correctly detect bad outputs, no '0' integer key is allowed.
			if (isset($readfps[0]) || isset($writefps[0]) || ($exceptfps !== NULL && isset($exceptfps[0])))  return false;

			$origreadfps = $readfps;
			$origwritefps = $writefps;
			$origexceptfps = $exceptfps;

			$result2 = @stream_select($readfps, $writefps, $exceptfps, $timeout);
			if ($result2 === false)  return false;

			if (isset($readfps[0]))
			{
				$fps = array();
				foreach ($origreadfps as $key => $fp)  $fps[(int)$fp] = $key;

				foreach ($readfps as $num => $fp)
				{
					$readfps[$fps[(int)$fp]] = $fp;

					unset($readfps[$num]);
				}
			}

			if (isset($writefps[0]))
			{
				$fps = array();
				foreach ($origwritefps as $key => $fp)  $fps[(int)$fp] = $key;

				foreach ($writefps as $num => $fp)
				{
					$writefps[$fps[(int)$fp]] = $fp;

					unset($writefps[$num]);
				}
			}

			if ($exceptfps !== NULL && isset($exceptfps[0]))
			{
				$fps = array();
				foreach ($origexceptfps as $key => $fp)  $fps[(int)$fp] = $key;

				foreach ($exceptfps as $num => $fp)
				{
					$exceptfps[$fps[(int)$fp]] = $fp;

					unset($exceptfps[$num]);
				}
			}

			return true;
		}

		// Handles new connections, the initial conversation, basic packet management, and timeouts.
		// Can wait on more streams than just sockets and/or more sockets.  Useful for waiting on other resources.
		// 'ws_s' and the 'ws_c_' prefix are reserved.
		// Returns an array of clients that may need more processing.
		public function Wait($timeout = false, $readfps = array(), $writefps = array(), $exceptfps = NULL)
		{
			$this->UpdateStreamsAndTimeout("", $timeout, $readfps, $writefps);

			$result = array("success" => true, "clients" => array(), "removed" => array(), "readfps" => array(), "writefps" => array(), "exceptfps" => array(), "accepted" => array(), "read" => array(), "write" => array());
			if (!count($readfps) && !count($writefps))  return $result;

			$result2 = self::FixedStreamSelect($readfps, $writefps, $exceptfps, $timeout);
			if ($result2 === false)  return array("success" => false, "error" => self::WSTranslate("Wait() failed due to stream_select() failure.  Most likely cause:  Connection failure."), "errorcode" => "stream_select_failed");

			// Return handles that were being waited on.
			$result["readfps"] = $readfps;
			$result["writefps"] = $writefps;
			$result["exceptfps"] = $exceptfps;

			$this->ProcessWaitResult($result);

			return $result;
		}

		protected function ProcessWaitResult(&$result)
		{
			// Handle new connections.
			if (isset($result["readfps"]["ws_s"]))
			{
				while (($fp = @stream_socket_accept($this->fp, 0)) !== false)
				{
					// Enable non-blocking mode.
					stream_set_blocking($fp, 0);

					$client = $this->InitNewClient($fp);

					$result["accepted"][$client->id] = $client;
				}

				unset($result["readfps"]["ws_s"]);
			}

			// Handle clients in the read queue.
			foreach ($result["readfps"] as $cid => $fp)
			{
				if (!is_string($cid) || strlen($cid) < 6 || substr($cid, 0, 5) !== "ws_c_")  continue;

				$id = (int)substr($cid, 5);

				if (!isset($this->clients[$id]))  continue;

				$client = $this->clients[$id];

				$result["read"][$id] = $client;

				if ($client->websocket !== false)
				{
					$this->ProcessClientQueuesAndTimeoutState($result, $id, true, isset($result["writefps"][$cid]));

					// Remove active WebSocket clients from the write queue.
					unset($result["writefps"][$cid]);
				}
				else
				{
					$result2 = @fread($fp, 8192);
					if ($result2 === false || ($result2 === "" && feof($fp)))
					{
						@fclose($fp);

						unset($this->clients[$id]);
					}
					else
					{
						$client->readdata .= $result2;

						if (strlen($client->readdata) > 100000)
						{
							// Bad header size.  Just kill the connection.
							@fclose($fp);

							unset($this->clients[$id]);
						}
						else
						{
							while (($pos = strpos($client->readdata, "\n")) !== false)
							{
								// Retrieve the next line of input.
								$line = rtrim(substr($client->readdata, 0, $pos));
								$client->readdata = (string)substr($client->readdata, $pos + 1);

								if ($client->request === false)  $client->request = trim($line);
								else if ($line !== "")
								{
									// Process the header.
									if ($client->lastheader != "" && (substr($line, 0, 1) == " " || substr($line, 0, 1) == "\t"))  $client->headers[$client->lastheader] .= $header;
									else
									{
										$pos = strpos($line, ":");
										if ($pos === false)  $pos = strlen($line);
										$client->lastheader = self::HeaderNameCleanup(substr($line, 0, $pos));
										$client->headers[$client->lastheader] = ltrim(substr($line, $pos + 1));
									}
								}
								else
								{
									// Headers have all been received.  Process the client request.
									$request = $client->request;
									$pos = strpos($request, " ");
									if ($pos === false)  $pos = strlen($request);
									$method = (string)substr($request, 0, $pos);
									$request = trim(substr($request, $pos));

									$pos = strrpos($request, " ");
									if ($pos === false)  $pos = strlen($request);
									$path = (string)substr($request, 0, $pos);
									if ($path === "")  $path = "/";

									if (isset($client->headers["Host"]))  $client->headers["Host"] = preg_replace('/[^a-z0-9.:\[\]_-]/', "", strtolower($client->headers["Host"]));

									$client->path = $path;
									$client->url = "ws://" . (isset($client->headers["Host"]) ? $client->headers["Host"] : "localhost") . $path;

									$this->ProcessInitialResponse($method, $path, $client);

									break;
								}
							}
						}
					}
				}

				unset($result["readfps"][$cid]);
			}

			// Handle remaining clients in the write queue.
			foreach ($result["writefps"] as $cid => $fp)
			{
				if (!is_string($cid) || strlen($cid) < 6 || substr($cid, 0, 5) !== "ws_c_")  continue;

				$id = (int)substr($cid, 5);

				if (!isset($this->clients[$id]))  continue;

				$client = $this->clients[$id];

				$result["write"][$id] = $client;

				if ($client->writedata === "")  $this->ProcessClientQueuesAndTimeoutState($result, $id, false, true);
				else
				{
					$result2 = @fwrite($fp, $client->writedata);
					if ($result2 === false || ($result2 === "" && feof($fp)))
					{
						@fclose($fp);

						unset($this->clients[$id]);
					}
					else if ($result2 === 0)  $this->ProcessClientQueuesAndTimeoutState($result, $id, true, false, 1);
					else
					{
						$client->writedata = (string)substr($client->writedata, $result2);

						// Let the application know about the new client or close the connection if the WebSocket Upgrade request failed.
						if ($client->writedata === "")
						{
							if ($client->websocket->GetStream() !== false)  $result["clients"][$id] = $client;
							else
							{
								@fclose($fp);

								unset($this->clients[$id]);
							}
						}
					}
				}

				unset($result["writefps"][$cid]);
			}

			// Handle client timeouts.
			$ts = time();
			if ($this->lasttimeoutcheck <= $ts - 5)
			{
				foreach ($this->clients as $id => $client)
				{
					if (!isset($result["clients"][$id]) && $client->writedata === "" && $client->websocket !== false)
					{
						$this->ProcessClientQueuesAndTimeoutState($result, $id, false, false);
					}
				}

				$this->lasttimeoutcheck = $ts;
			}
		}

		protected function ProcessClientQueuesAndTimeoutState(&$result, $id, $read, $write, $readsize = 65536)
		{
			$client = $this->clients[$id];

			$result2 = $client->websocket->ProcessQueuesAndTimeoutState($read, $write, $readsize);
			if ($result2["success"])  $result["clients"][$id] = $client;
			else
			{
				$result["removed"][$id] = array("result" => $result2, "client" => $client);

				$this->RemoveClient($id);
			}
		}

		public function GetClients()
		{
			return $this->clients;
		}

		public function NumClients()
		{
			return count($this->clients);
		}

		public function UpdateClientState($id)
		{
		}

		public function GetClient($id)
		{
			return (isset($this->clients[$id]) ? $this->clients[$id] : false);
		}

		public function RemoveClient($id)
		{
			if (isset($this->clients[$id]))
			{
				$client = $this->clients[$id];

				// Remove the client.
				if ($client->websocket->GetStream() !== false)
				{
					$client->websocket->Disconnect();
					$client->websocket = false;
					$client->fp = false;
				}

				if ($client->fp !== false)  @fclose($client->fp);

				unset($this->clients[$id]);
			}
		}

		public function ProcessWebServerClientUpgrade($webserver, $client)
		{
			if (!($client instanceof WebServer_Client))  return false;

			if (!$client->requestcomplete || $client->mode === "handle_response")  return false;
			if ($client->request["method"] !== "GET" || !isset($client->headers["Connection"]) || stripos($client->headers["Connection"], "upgrade") === false || !isset($client->headers["Upgrade"]) || stripos($client->headers["Upgrade"], "websocket") === false)  return false;

			// Create an equivalent WebSocket server client class.
			$webserver->DetachClient($client->id);

			$method = $client->request["method"];
			$path = $client->request["path"];

			$client2 = $this->InitNewClient($client->fp);
			$client2->request = $client->request["line"];
			$client2->headers = $client->headers;
			$client2->path = $path;
			$client2->url = "ws://" . (isset($client->headers["Host"]) ? $client->headers["Host"] : "localhost") . $path;

			$client2->appdata = $client->appdata;

			$this->ProcessInitialResponse($method, $path, $client2);

			return $client2->id;
		}

		public static function HeaderNameCleanup($name)
		{
			return preg_replace('/\s+/', "-", ucwords(strtolower(trim(preg_replace('/[^A-Za-z0-9 ]/', " ", $name)))));
		}

		public static function WSTranslate()
		{
			$args = func_get_args();
			if (!count($args))  return "";

			return call_user_func_array((defined("CS_TRANSLATE_FUNC") && function_exists(CS_TRANSLATE_FUNC) ? CS_TRANSLATE_FUNC : "sprintf"), $args);
		}
	}
?>