mercureact/src/Http/Server.php

<?php

namespace NoccyLabs\Mercureact\Http;

use NoccyLabs\Mercureact\Broker\Message;
use NoccyLabs\Mercureact\Broker\TopicManager;
use NoccyLabs\Mercureact\Configuration;
use NoccyLabs\Mercureact\Http\Exeption\RequestException;
use NoccyLabs\Mercureact\Http\Exeption\SecurityException;
use NoccyLabs\React\WebSocket\WebSocketConnection;
use NoccyLabs\React\WebSocket\WebSocketMiddleware;
use NoccyLabs\SimpleJWT\JWTToken;
use NoccyLabs\SimpleJWT\Key\JWTPlaintextKey;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\Http\HttpServer;
use React\Http\Message\Response;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
use React\Socket\ServerInterface;
use React\Stream\ThroughStream;
use SplObjectStorage;
use Symfony\Component\Uid\Uuid;
use Throwable;

class Server
{
    private Configuration $config;

    private LoopInterface $loop;

    private HttpServer $server;

    private WebSocketMiddleware $webSocket;

    private SplObjectStorage $webSocketClients;

    private SplObjectStorage $eventClients;

    private TopicManager $topicManager;

    public static string $indexPage;

    /**
     * 
     * 
     */
    public function __construct(Configuration $config, array $options=[], ?LoopInterface $loop=null)
    {
        $this->loop = $loop??Loop::get();

        $this->config = $config;

        $this->topicManager = new TopicManager();

        $this->server = $this->createHttpServer($options);
        
        $this->webSocket = new WebSocketMiddleware();

        $this->eventClients = new SplObjectStorage();
        $this->webSocketClients = new SplObjectStorage();
        $this->webSocket->on(WebSocketMiddleware::EVENT_CONNECTION, $this->onWebSocketConnection(...));
    }

    /**
     * 
     * 
     * @return void
     */
    public function listen(ServerInterface $socket): void
    {
        $this->server->listen($socket);
    }

    /**
     * 
     * @return HttpServer
     */
    private function createHttpServer(array $options): HttpServer
    {
        // TODO break out the middleware to facilitate testing
        return new HttpServer(
            $this->rejectionWrappingMiddleware(...),
            $this->checkRequestSecurityMiddleware(...),
            $this->handleWebSocketRequest(...),
            $this->handleMercureRequest(...),
            $this->handleApiRequest(...),
            $this->handleNotFound(...)
        );
    }

    /**
     * Resolves unhandled requests with a 404 error
     * 
     * @param ServerRequestInterface $request
     * @return PromiseInterface
     */
    private function handleNotFound(ServerRequestInterface $request): PromiseInterface
    {
        return new Promise(
            function ($resolve) { 
                $resolve(Response::plaintext("Not found")->withStatus(404));
            }
        );
    }

    /**
     * Wraps rejections into error messages, and also does some sanity checks on the returned
     * data, making sure it is a response.
     * 
     * @param ServerRequestInterface $request
     * @param callable $next
     * @return PromiseInterface
     */
    private function rejectionWrappingMiddleware(ServerRequestInterface $request, callable $next): PromiseInterface
    {
        $promise = new Promise(
            function (callable $resolve) use ($request, $next) {
                $resolve($next($request));
            }
        );
        return $promise->then(
            function ($response) {
                if ($response instanceof ResponseInterface) {
                    return $response;
                }
                if (is_array($response)) {
                    return Response::json($response);
                }
                if (is_string($response)) {
                    return Response::plaintext($response);
                }
                return Response::plaintext((string)$response);
            },
            function (Throwable $t) {
                if ($t instanceof SecurityException) {
                    return Response::plaintext("Access Denied")->withStatus(Response::STATUS_UNAUTHORIZED);
                }
                return Response::plaintext("500: Internal Server Error (".$t->getMessage().")\n")->withStatus(500);
            }
        )->then(
            function ($response) use ($request) {
                assert("\$response instanceof ResponseInterface");
                $host = ($request->getServerParams()['SERVER_ADDR']??"");
                        //. ":" . ($request->getServerParams()['SERVER_PORT']??"80");
                fprintf(STDOUT, "%s %3d %s %s %d\n", 
                    $request->getServerParams()['REMOTE_ADDR'], 
                    $response->getStatusCode(), 
                    $request->getMethod(), 
                    $request->getUri()->getPath(), 
                    strlen($response->getBody())
                );
                return $response
                    ->withAddedHeader('Link', '<https://'.$host.'/.well-known/mercure>; rel="mercure"')
                    ->withAddedHeader('Link', '<wss://'.$host.'/.well-known/mercure>; rel="mercure+ws"')
                    ->withHeader('Access-Control-Allow-Origin', '*')
                    ->withHeader('Content-Security-Policy', "default-src * 'self' http: 'unsafe-eval' 'unsafe-inline'; connect-src * 'self'")
                    ->withHeader('Cache-Control', 'must-revalidate')
                    ->withHeader('Server', 'Mercureact/0.1.0');
            }
        );
    }

    /**
     * 
     * 
     * @param ServerRequestInterface $request
     * @param callable $next
     * @return PromiseInterface
     */
    private function checkRequestSecurityMiddleware(ServerRequestInterface $request, callable $next): PromiseInterface
    {
        return new Promise(
            function (callable $resolve, callable $reject) use ($request, $next) {
                // Check JWT in authorization header or authorization query param
                $request = $this->checkAuthorization($request);

                $resolve($next($request));
            }
        );
    }

    /**
     * 
     * 
     * @param ServerRequestInterface $request
     * @return ServerRequestInterface
     */
    private function checkAuthorization(ServerRequestInterface $request): ServerRequestInterface
    {
        $authorization = $request->getHeaderLine('authorization');
        if (str_starts_with(strtolower($authorization), "bearer ")) {
            $jwt = substr($authorization, strpos($authorization, " ")+1);
            $key = new JWTPlaintextKey($this->config->getJwtSecret());
            $tok = new JWTToken($key, $jwt);
            if (!$tok->isValid()) {
                throw new SecurityException(message:"Invalid token", code:SecurityException::ERR_ACCESS_DENIED);
            }
            $mercureClaims = $tok->claims->get('mercure');
            return $request
                ->withAttribute('authorization', $tok);
        } else {
            return $request
                ->withAttribute('authorization', null);

        }
    }

    /**
     * 
     * 
     * @param ServerRequestInterface $request
     * @param callable $next
     * @return PromiseInterface
     */
    private function handleMercureRequest(ServerRequestInterface $request, callable $next): PromiseInterface
    {
        return new Promise(
            function (callable $resolve, callable $reject) use ($next, $request) {
                if ($request->getUri()->getPath() == "/.well-known/mercure") {
                    if ($request->getMethod() == 'POST') {
                        $resolve($this->handleMercurePublish($request));
                        return;
                    }
                    $resolve($this->handleMercureClient($request));
                } else {
                    $resolve($next($request));
                }
            }
        );
    }

    /**
     * 
     * 
     * @param ServerRequestInterface $request
     * @return ResponseInterface
     */
    private function handleMercureClient(ServerRequestInterface $request): ResponseInterface
    {
        $tok = $request->getAttribute('authorization');
        if ($tok instanceof JWTToken) {
            $claims = $tok->claims->getAll();
            if (isset($claims['mercure']['subscribe'])) {
                $subscribeClaims = $claims['mercure']['subscribe'];
                // TODO check topic against subscribeClaims
            }            
        }

        $responseStream = new ThroughStream();

        $response = new Response(
            body: $responseStream
        );

        $this->eventClients->attach($responseStream, $request);
        $responseStream->on('close', function () use ($responseStream) {
            $this->eventClients->detach($responseStream);;
        });

        return $response
            ->withHeader("Cache-Control", "no-store")
            ->withHeader("Content-Type", "text/event-stream");
    }

    /**
     * 
     * 
     * @param ServerRequestInterface $request
     * @return ResponseInterface
     */
    private function handleMercurePublish(ServerRequestInterface $request): ResponseInterface
    {
        if ($request->getHeaderLine('content-type') !== 'application/x-www-form-urlencoded') {
            throw new \Exception("Invalid request");
        }

        // Parse out the urlencoded body. Pretty sure there is a better way to do this?
        $body = explode("&", (string)$request->getBody());
        $data = [];
        foreach ($body as $param) {
            if (!str_contains($param, "="))
                throw new RequestException("Invalid request data", RequestException::ERR_INVALID_REQUEST_DATA);
            [ $name, $value ] = array_map('urldecode', explode("=", $param, 2));
            if (in_array($name, [ 'topic' ])) {
                if (!isset($data[$name]))
                    $data[$name] = [];
                $data[$name][] = $value;
            } else {
                $data[$name] = $value;
            }
        }

        // Grab the JWT token from the requests authorization attribute
        $tok = $request->getAttribute('authorization');
        if ($tok instanceof JWTToken) {
            $claims = $tok->claims->getAll();
            if (isset($claims['mercure']['publish'])) {
                $publishClaims = $claims['mercure']['publish'];
                // TODO check topic against publishClaims
                if (!$this->checkTopicClaims($data['topic']??[], $publishClaims)) {
                    throw new SecurityException("Insufficient permissions for publish", SecurityException::ERR_NO_PERMISSION);
                }
            }
        }

        // Put an id in there if none already
        // TODO add a configurable for this
        if (!isset($data['id'])) {
            $data['id'] = (string)Uuid::v7();
        }

        $message = Message::fromData($data);

        $this->loop->futureTick(function () use ($message) {
            $this->publishMercureMessage($message);
        });

        return Response::plaintext("urn:uuid:".$message->id."\n");
    }

    private function checkTopicClaims(string|array $topic, array $claims): bool
    {
        foreach ((array)$topic as $match) {
            foreach ($claims as $claim) {
                if ($claim === "*") return true;
                if ($claim === $match) return true;
                // TODO implement full matching
            }
        }
        return false;
    }

    /**
     * 
     * 
     * @param Message $message
     * @return void
     */
    private function publishMercureMessage(Message $message): void
    {
        foreach ($this->webSocketClients as $webSocket) {
            $webSocket->write(json_encode([
                'type' => $message->type,
                'topic' => $message->topic,
                'data' => (@json_decode($message->data))??$message->data
            ]));
        }

        $sseMessage = "";
        if ($message->type) {
            $sseMessage .= "event: ".$message->type."\n";
        }
        $sseMessage .= "data: ".$message->data."\n\n";

        foreach ($this->eventClients as $client) {
            $client->write($sseMessage);
        }
    }

    /**
     * 
     * 
     * @param ServerRequestInterface $request
     * @param callable $next
     * @return PromiseInterface
     */
    private function handleWebSocketRequest(ServerRequestInterface $request, callable $next): PromiseInterface
    {
        return new Promise(
            function (callable $resolve, callable $reject) use ($next, $request) {
                if ($request->getUri()->getPath() == "/.well-known/mercure")
                    $resolve(call_user_func($this->webSocket, $request, $next));
                else 
                    $resolve($next($request));
            }
        );
    }

    /**
     * 
     * 
     */
    private function onWebSocketConnection(WebSocketConnection $connection)
    {
        $this->webSocketClients->attach($connection);

        $connection->on('close', function () use ($connection) {
            $this->webSocketClients->detach($connection);
        });

        $request = $connection->getServerRequest();
        $topic = $request->getQueryParams()['topic'][0]??'';
        $connection->setGroup($topic);
    }

    /**
     * 
     * 
     * @param ServerRequestInterface $request
     * @param callable $next
     * @return PromiseInterface
     */
    private function handleApiRequest(ServerRequestInterface $request, callable $next): PromiseInterface
    {
        return new Promise(
            function (callable $resolve, callable $reject) use ($next, $request) {

                $path = $request->getUri()->getPath();

                if ($path === "/index.html") {
                    $resolve(Response::html(self::$indexPage));
                }

                switch (true) {
                    case preg_match('<^/.well-known/mercure/subscriptions(/.+?)$>', $path, $m):
                        $query = explode("/", trim($m[1]??null, "/"));
                        $topic = array_shift($query);
                        $subscription = array_shift($query);
                        $resolve($this->apiGetSubscriptions($topic, $subscription));
                        return;
                
                    case preg_match('<^/.well-known/mercureact/status$>', $path):
                        $resolve([
                            'server' => 'Mercureact/1.0',
                            'topics' => $this->topicManager->getTopicCount(),
                            'subscriptions' => $this->topicManager->getSubscriberCount(),
                            'memoryPeak' => memory_get_peak_usage(true),
                            'memoryUsage' => memory_get_usage(true)
                        ]);
                        return;
    
                    case preg_match('<^/.well-known/mercureact/status$>', $path):
                        $resolve([ 'version' => '1.0' ]);
                        return;
                }

                $resolve($next($request));
            }
        );
    }

    /**
     * 
     * 
     * @return ResponseInterface
     */
    private function apiGetSubscriptions(string|null $topic, string|null $subscription): ResponseInterface
    {
        $lastEventId = "urn:uuid:5e94c686-2c0b-4f9b-958c-92ccc3bbb4eb";

        $data = [
            "@context" => "https://mercure.rocks/",
            "id" => "/.well-known/mercure/subscriptions",
            "type" => "Subscriptions",
            "lastEventID" => $lastEventId,
            "subscriptions" => []
        ];

        return Response::json($data)
            ->withHeader('Content-Type', 'application/ld+json')
            ->withHeader('ETag', $lastEventId);
    }
}

Server::$indexPage = <<<ENDHTML
<html>
<head>
<meta http-equiv="Content-Security-Policy" content="default-src 'self' http: 'unsafe-eval' 'unsafe-inline'; style-src 'self';">
</head>
<body>
<script type="text/javascript">
  const events = new EventSource("http://127.0.0.1:9000/.well-known/mercure");
  events.onmessage = msg => console.log(msg);
</script>
</body>
</html>
ENDHTML;