noccy/mercureact
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Extracted middleware and handlers

Browse Source
This commit is contained in:
Chris 2024-03-10 20:22:28 +01:00
parent 83c34f4a47
commit 39869d605c
8 changed files with 592 additions and 416 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/Broker/SubscriptionList.php
View File

@ -2,10 +2,13 @@
namespace NoccyLabs\Mercureact\Broker; namespace NoccyLabs\Mercureact\Broker;
use ArrayIterator;
use Countable; use Countable;
use IteratorAggregate;
use SplObjectStorage; use SplObjectStorage;
use Traversable;
class SubscriptionList implements Countable class SubscriptionList implements Countable, IteratorAggregate
{ {
private array $subscriptions = []; private array $subscriptions = [];
@ -16,5 +19,10 @@ class SubscriptionList implements Countable
{ {
return count($this->subscriptions); return count($this->subscriptions);
} }
public function getIterator(): Traversable
{
return new ArrayIterator($this->subscriptions);
}
} }

110
src/Http/Middleware/ApiHandler.php Normal file
View File

@ -0,0 +1,110 @@
<?php
namespace NoccyLabs\Mercureact\Http\Middleware;
use NoccyLabs\Mercureact\Broker\TopicManager;
use NoccyLabs\Mercureact\Configuration;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\Http\Message\Response;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
class ApiHandler
{
public static string $indexPage;
public function __construct(
private Configuration $config,
private TopicManager $topicManager
)
{
}
/**
*
*
* @param ServerRequestInterface $request
* @param callable $next
* @return PromiseInterface
*/
public function __invoke(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
{
// TODO implement once we can enumerate topics and subscriptions
// mock data
$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);
}
}
ApiHandler::$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;

191
src/Http/Middleware/MercureHandler.php Normal file
View File

@ -0,0 +1,191 @@
<?php
namespace NoccyLabs\Mercureact\Http\Middleware;
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\SimpleJWT\JWTToken;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\Http\Message\Response;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
use React\Stream\ThroughStream;
use SplObjectStorage;
use Symfony\Component\Uid\Uuid;
class MercureHandler
{
private LoopInterface $loop;
public function __construct(
private Configuration $config,
private SplObjectStorage $eventClients,
private TopicManager $topicManager,
?LoopInterface $loop=null
)
{
$this->loop = $loop ?? Loop::get();
}
/**
* Mecure handler middleware
*
* @param ServerRequestInterface $request
* @param callable $next
* @return PromiseInterface
*/
public function __invoke(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);
}
}
} else {
// FIXME reject if access denied
}
// Put an id in there if none already
// TODO add a configurable for this
if (!isset($data['id'])) {
$data['id'] = (string)Uuid::v7();
}
// Attempt to create the message
$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);
}
}
}

27
src/Http/Middleware/NotFoundHandler.php Normal file
View File

@ -0,0 +1,27 @@
<?php
namespace NoccyLabs\Mercureact\Http\Middleware;
use Psr\Http\Message\ServerRequestInterface;
use React\Http\Message\Response;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
class NotFoundHandler
{
/**
* Resolves unhandled requests with a 404 error
*
* @param ServerRequestInterface $request
* @return PromiseInterface
*/
public function __invoke(ServerRequestInterface $request): PromiseInterface
{
return new Promise(
function ($resolve) {
$resolve(Response::plaintext("Not found")->withStatus(404));
}
);
}
}

80
src/Http/Middleware/ResponseMiddleware.php Normal file
View File

@ -0,0 +1,80 @@
<?php
namespace NoccyLabs\Mercureact\Http\Middleware;
use NoccyLabs\Mercureact\Configuration;
use NoccyLabs\Mercureact\Http\Exeption\SecurityException;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\Http\Message\Response;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
use Throwable;
class ResponseMiddleware
{
public function __construct(
private Configuration $config
)
{
}
/**
* 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
*/
public function __invoke(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');
}
);
}
}

68
src/Http/Middleware/SecurityMiddleware.php Normal file
View File

@ -0,0 +1,68 @@
<?php
namespace NoccyLabs\Mercureact\Http\Middleware;
use NoccyLabs\Mercureact\Configuration;
use NoccyLabs\Mercureact\Http\Exeption\SecurityException;
use NoccyLabs\SimpleJWT\JWTToken;
use NoccyLabs\SimpleJWT\Key\JWTPlaintextKey;
use Psr\Http\Message\ServerRequestInterface;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
class SecurityMiddleware
{
public function __construct(
private Configuration $config
)
{
}
/**
*
*
* @param ServerRequestInterface $request
* @param callable $next
* @return PromiseInterface
*/
public function __invoke(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);
}
}
}

71
src/Http/Middleware/WebSocketHandler.php Normal file
View File

@ -0,0 +1,71 @@
<?php
namespace NoccyLabs\Mercureact\Http\Middleware;
use NoccyLabs\Mercureact\Broker\TopicManager;
use NoccyLabs\Mercureact\Configuration;
use NoccyLabs\React\WebSocket\WebSocketConnection;
use NoccyLabs\React\WebSocket\WebSocketMiddleware;
use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
use SplObjectStorage;
class WebSocketHandler
{
private LoopInterface $loop;
private WebSocketMiddleware $webSocket;
public function __construct(
private Configuration $config,
private SplObjectStorage $webSocketClients,
private TopicManager $topicManager,
?LoopInterface $loop=null
)
{
$this->loop = $loop ?? Loop::get();
$this->webSocket = new WebSocketMiddleware();
$this->webSocket->on(WebSocketMiddleware::EVENT_CONNECTION, $this->onWebSocketConnection(...));
}
/**
*
*
* @param ServerRequestInterface $request
* @param callable $next
* @return PromiseInterface
*/
public function __invoke(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);
}
}

451
src/Http/Server.php
View File

@ -2,28 +2,19 @@
namespace NoccyLabs\Mercureact\Http; namespace NoccyLabs\Mercureact\Http;
use NoccyLabs\Mercureact\Broker\Message;
use NoccyLabs\Mercureact\Broker\TopicManager; use NoccyLabs\Mercureact\Broker\TopicManager;
use NoccyLabs\Mercureact\Configuration; use NoccyLabs\Mercureact\Configuration;
use NoccyLabs\Mercureact\Http\Exeption\RequestException; use NoccyLabs\Mercureact\Http\Middleware\ApiHandler;
use NoccyLabs\Mercureact\Http\Exeption\SecurityException; use NoccyLabs\Mercureact\Http\Middleware\MercureHandler;
use NoccyLabs\React\WebSocket\WebSocketConnection; use NoccyLabs\Mercureact\Http\Middleware\NotFoundHandler;
use NoccyLabs\React\WebSocket\WebSocketMiddleware; use NoccyLabs\Mercureact\Http\Middleware\ResponseMiddleware;
use NoccyLabs\SimpleJWT\JWTToken; use NoccyLabs\Mercureact\Http\Middleware\SecurityMiddleware;
use NoccyLabs\SimpleJWT\Key\JWTPlaintextKey; use NoccyLabs\Mercureact\Http\Middleware\WebSocketHandler;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\Loop; use React\EventLoop\Loop;
use React\EventLoop\LoopInterface; use React\EventLoop\LoopInterface;
use React\Http\HttpServer; use React\Http\HttpServer;
use React\Http\Message\Response;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
use React\Socket\ServerInterface; use React\Socket\ServerInterface;
use React\Stream\ThroughStream;
use SplObjectStorage; use SplObjectStorage;
use Symfony\Component\Uid\Uuid;
use Throwable;
class Server class Server
{ {
@ -33,15 +24,19 @@ class Server
private HttpServer $server; private HttpServer $server;
private WebSocketMiddleware $webSocket;
private SplObjectStorage $webSocketClients; private SplObjectStorage $webSocketClients;
private SplObjectStorage $eventClients; private SplObjectStorage $eventClients;
private TopicManager $topicManager; private TopicManager $topicManager;
public static string $indexPage; private ResponseMiddleware $responseMiddleware;
private SecurityMiddleware $securityMiddleware;
private WebSocketHandler $webSocketHandler;
private MercureHandler $mercureHandler;
private ApiHandler $apiRequestHandler;
private NotFoundHandler $notFoundHandler;
/** /**
* *
@ -54,14 +49,11 @@ class Server
$this->config = $config; $this->config = $config;
$this->topicManager = new TopicManager(); $this->topicManager = new TopicManager();
$this->server = $this->createHttpServer($options);
$this->webSocket = new WebSocketMiddleware();
$this->eventClients = new SplObjectStorage(); $this->eventClients = new SplObjectStorage();
$this->webSocketClients = new SplObjectStorage(); $this->webSocketClients = new SplObjectStorage();
$this->webSocket->on(WebSocketMiddleware::EVENT_CONNECTION, $this->onWebSocketConnection(...));
$this->server = $this->createHttpServer($options);
} }
/** /**
@ -80,401 +72,30 @@ class Server
*/ */
private function createHttpServer(array $options): HttpServer private function createHttpServer(array $options): HttpServer
{ {
// TODO break out the middleware to facilitate testing
return new HttpServer( return new HttpServer(
$this->rejectionWrappingMiddleware(...), $this->responseMiddleware = new ResponseMiddleware(
$this->checkRequestSecurityMiddleware(...), config: $this->config
$this->handleWebSocketRequest(...), ),
$this->handleMercureRequest(...), $this->securityMiddleware = new SecurityMiddleware(
$this->handleApiRequest(...), config: $this->config
$this->handleNotFound(...) ),
$this->webSocketHandler = new WebSocketHandler(
config: $this->config,
webSocketClients: $this->webSocketClients,
topicManager: $this->topicManager
),
$this->mercureHandler = new MercureHandler(
config: $this->config,
eventClients: $this->eventClients,
topicManager: $this->topicManager
),
$this->apiRequestHandler = new ApiHandler(
config: $this->config,
topicManager: $this->topicManager
),
$this->notFoundHandler = new NotFoundHandler()
); );
} }
/**
* 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;