control/bin/control_ws.php

<?php
declare(strict_types=1);

use Ratchet\ConnectionInterface;
use Ratchet\MessageComponentInterface;
use Ratchet\Http\HttpServer;
use Ratchet\Server\IoServer;
use Ratchet\WebSocket\WsServer;
use React\EventLoop\Loop;
use React\Socket\SocketServer;

define('CONTROL_API_LIBRARY', true);

require_once __DIR__ . '/../vendor/autoload.php';
require_once __DIR__ . '/../public/api.php';

final class ControlWebSocket implements MessageComponentInterface
{
    /** @var SplObjectStorage<ConnectionInterface, array{dmesg: bool, ip: string}> */
    private SplObjectStorage $clients;

    private bool $statusBusy = false;
    private bool $dmesgBusy = false;

    public function __construct()
    {
        $this->clients = new SplObjectStorage();
    }

    public function onOpen(ConnectionInterface $conn): void
    {
        $request = $conn->httpRequest ?? null;
        $cookies = $this->cookiesFromHeader($request?->getHeaderLine('Cookie') ?? '');
        $ip = $request?->getHeaderLine('X-Real-IP')
            ?: $request?->getHeaderLine('X-Forwarded-For')
            ?: '127.0.0.1';
        $ua = $request?->getHeaderLine('User-Agent') ?? '';

        if (!$this->signedIn($cookies, $ip, $ua)) {
            $conn->send($this->json([
                'type' => 'error',
                'message' => 'login_required',
            ]));
            $conn->close();
            return;
        }

        $this->clients[$conn] = [
            'dmesg' => false,
            'ip' => $ip,
        ];
        $this->sendStatus($conn);
    }

    public function onMessage(ConnectionInterface $from, $msg): void
    {
        $data = json_decode((string)$msg, true);
        if (!is_array($data)) {
            return;
        }

        $state = $this->clients[$from] ?? [
            'dmesg' => false,
            'ip' => '127.0.0.1',
        ];

        if (($data['type'] ?? '') === 'dmesg') {
            $state['dmesg'] = !empty($data['open']);
            $this->clients[$from] = $state;
            if ($state['dmesg']) {
                $this->sendDmesg($from);
            }
            return;
        }

        if (($data['type'] ?? '') === 'status_refresh') {
            $this->sendStatus($from);
        }
    }

    public function onClose(ConnectionInterface $conn): void
    {
        if ($this->clients->contains($conn)) {
            $this->clients->detach($conn);
        }
    }

    public function onError(ConnectionInterface $conn, Exception $e): void
    {
        $conn->send($this->json([
            'type' => 'error',
            'message' => $e->getMessage(),
        ]));
        $conn->close();
    }

    public function broadcastStatus(): void
    {
        if ($this->clients->count() === 0 || $this->statusBusy) {
            return;
        }

        $this->statusBusy = true;
        try {
            $payload = $this->statusPayload();
            foreach ($this->clients as $client) {
                $client->send($payload);
            }
        } catch (Throwable $e) {
            $this->broadcastError($e);
        } finally {
            $this->statusBusy = false;
        }
    }

    public function broadcastDmesg(): void
    {
        if ($this->clients->count() === 0 || $this->dmesgBusy) {
            return;
        }

        $targets = [];
        foreach ($this->clients as $client) {
            $state = $this->clients[$client];
            if (!empty($state['dmesg'])) {
                $targets[] = $client;
            }
        }

        if ($targets === []) {
            return;
        }

        $this->dmesgBusy = true;
        try {
            $payload = $this->json([
                'type' => 'dmesg',
                'data' => dmesg_log(),
            ]);
            foreach ($targets as $client) {
                $client->send($payload);
            }
        } catch (Throwable $e) {
            $this->broadcastError($e);
        } finally {
            $this->dmesgBusy = false;
        }
    }

    private function sendStatus(ConnectionInterface $conn): void
    {
        try {
            $conn->send($this->statusPayload());
        } catch (Throwable $e) {
            $conn->send($this->json([
                'type' => 'error',
                'message' => $e->getMessage(),
            ]));
        }
    }

    private function sendDmesg(ConnectionInterface $conn): void
    {
        try {
            $conn->send($this->json([
                'type' => 'dmesg',
                'data' => dmesg_log(),
            ]));
        } catch (Throwable $e) {
            $conn->send($this->json([
                'type' => 'error',
                'message' => $e->getMessage(),
            ]));
        }
    }

    private function statusPayload(): string
    {
        try {
            $data = collect_snapshot(false);
        } catch (PDOException $e) {
            if (!db_connection_lost($e)) {
                throw $e;
            }

            db(true);
            $data = collect_snapshot(false);
        }

        return $this->json([
            'type' => 'status',
            'data' => $data,
        ]);
    }

    private function broadcastError(Throwable $e): void
    {
        $payload = $this->json([
            'type' => 'error',
            'message' => $e->getMessage(),
        ]);
        foreach ($this->clients as $client) {
            $client->send($payload);
        }
    }

    /** @return array<string, string> */
    private function cookiesFromHeader(string $header): array
    {
        $cookies = [];
        foreach (explode(';', $header) as $chunk) {
            $parts = explode('=', trim($chunk), 2);
            if (count($parts) !== 2 || $parts[0] === '') {
                continue;
            }
            $cookies[$parts[0]] = urldecode($parts[1]);
        }

        return $cookies;
    }

    /** @param array<string, string> $cookies */
    private function signedIn(array $cookies, string $ip, string $ua): bool
    {
        if (
            empty($cookies[session_name()])
            && empty($cookies[REMEMBER_COOKIE])
        ) {
            return false;
        }

        if (session_status() === PHP_SESSION_ACTIVE) {
            session_write_close();
        }

        $_COOKIE = $cookies;
        $_SERVER['REMOTE_ADDR'] = $ip;
        $_SERVER['HTTP_USER_AGENT'] = $ua;
        $_SERVER['HTTPS'] = 'on';

        $sessionId = (string)($cookies[session_name()] ?? '');
        if ($sessionId !== '' && preg_match('/^[A-Za-z0-9,-]{16,128}$/', $sessionId)) {
            session_id($sessionId);
        }

        session_start();
        $ok = signed_in();
        session_write_close();

        return $ok;
    }

    /** @param array<string, mixed> $payload */
    private function json(array $payload): string
    {
        return json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) ?: '{}';
    }
}

$address = getenv('CONTROL_WS_ADDRESS') ?: '127.0.0.1:8088';
$loop = Loop::get();
$app = new ControlWebSocket();

$loop->addPeriodicTimer(1.0, static fn() => $app->broadcastStatus());
$loop->addPeriodicTimer(1.0, static fn() => $app->broadcastDmesg());

$socket = new SocketServer($address, [], $loop);
$server = new IoServer(
    new HttpServer(new WsServer($app)),
    $socket,
    $loop
);

fwrite(STDOUT, 'Control WebSocket listening on ' . $address . PHP_EOL);
$server->run();