[13.x] Add a Mercure broadcast driver

composer.json

@@ -92,8 +92,10 @@
         "resend/resend-php": "^1.0",
         "symfony/cache": "^7.4.0 || ^8.0.0",
         "symfony/http-client": "^7.4.0 || ^8.0.0",
+        "symfony/mercure": "^0.8.0",
         "symfony/psr-http-message-bridge": "^7.4.0 || ^8.0.0",
-        "symfony/translation": "^7.4.0 || ^8.0.0"
+        "symfony/translation": "^7.4.0 || ^8.0.0",
+        "web-token/jwt-library": "^4.1"
     },
     "replace": {
         "illuminate/auth": "self.version",
@@ -177,8 +179,10 @@
         "symfony/filesystem": "Required to enable support for relative symbolic links (^7.4 || ^8.0).",
         "symfony/http-client": "Required to enable support for the Symfony API mail transports (^7.4 || ^8.0).",
         "symfony/mailgun-mailer": "Required to enable support for the Mailgun mail transport (^7.4 || ^8.0).",
+        "symfony/mercure": "Required to use the Mercure broadcast driver (^0.8).",
         "symfony/postmark-mailer": "Required to enable support for the Postmark mail transport (^7.4 || ^8.0).",
-        "symfony/psr-http-message-bridge": "Required to use PSR-7 bridging features (^7.4 || ^8.0)."
+        "symfony/psr-http-message-bridge": "Required to use PSR-7 bridging features (^7.4 || ^8.0).",
+        "web-token/jwt-library": "Required to sign Mercure JWTs and use end-to-end encrypted channels (^4.1)."
     },
     "minimum-stability": "dev",
     "prefer-stable": true,

src/Illuminate/Broadcasting/BroadcastManager.php

@@ -7,6 +7,7 @@
 use GuzzleHttp\Client as GuzzleClient;
 use Illuminate\Broadcasting\Broadcasters\AblyBroadcaster;
 use Illuminate\Broadcasting\Broadcasters\LogBroadcaster;
+use Illuminate\Broadcasting\Broadcasters\MercureBroadcaster;
 use Illuminate\Broadcasting\Broadcasters\NullBroadcaster;
 use Illuminate\Broadcasting\Broadcasters\PusherBroadcaster;
 use Illuminate\Broadcasting\Broadcasters\RedisBroadcaster;
@@ -28,6 +29,14 @@
 use Pusher\Pusher;
 use ReflectionException;
 use RuntimeException;
+use Symfony\Component\HttpClient\HttpClient;
+use Symfony\Component\Mercure\FrankenPhpHub;
+use Symfony\Component\Mercure\Hub;
+use Symfony\Component\Mercure\Jwt\DefaultClaimsTokenFactory;
+use Symfony\Component\Mercure\Jwt\FactoryTokenProvider;
+use Symfony\Component\Mercure\Jwt\Grant;
+use Symfony\Component\Mercure\Jwt\WebTokenFactory;
+use Symfony\Component\Mercure\ProtocolVersion;
 use Throwable;
 
 use function Illuminate\Support\enum_value;
@@ -436,6 +445,267 @@ protected function createLogDriver(array $config)
         );
     }
 
+    /**
+     * Create an instance of the driver.
+     *
+     * @param  array  $config
+     * @return \Illuminate\Contracts\Broadcasting\Broadcaster
+     */
+    protected function createMercureDriver(array $config)
+    {
+        $expiration = $this->mercureSubscribeExpiration($config);
+
+        if ($expiration <= 0) {
+            throw new InvalidArgumentException('The Mercure "subscribe_expiration" configuration value must be a positive number of minutes.');
+        }
+
+        $hub = $this->mercure($config);
+
+        if ($hub->getFactory() === null) {
+            throw new InvalidArgumentException('The Mercure broadcasting connection requires a "secret" (or "subscribe_secret") configuration value.');
+        }
+
+        return new MercureBroadcaster(
+            $hub,
+            $expiration,
+            $this->mercureChannelEncrypter($config),
+            (string) (($config['topic_prefix'] ?? null) ?: 'https://laravel.alt/echo/'),
+            (bool) ($config['client_events'] ?? true),
+        );
+    }
+
+    /**
+     * Get a Mercure hub instance for the given configuration.
+     *
+     * The hub's token factory is the subscriber one, feeding the Mercure
+     * Authorization helper; the publish token is minted by the hub's
+     * token provider.
+     *
+     * @param  array  $config
+     * @return \Symfony\Component\Mercure\HubInterface
+     */
+    public function mercure(array $config)
+    {
+        if (empty($config['url'])) {
+            return $this->frankenPhpMercure($config);
+        }
+
+        // 0 (the default) delegates the lifetime to the hub's token factory:
+        // "session.cookie_lifetime", or an hour when that setting is 0.
+        $publishExpiration = (int) (($config['publish_expiration'] ?? 0) * 60);
+
+        if ($publishExpiration < 0) {
+            throw new InvalidArgumentException('The Mercure "publish_expiration" configuration value must be a positive number of minutes, or 0 to use the default lifetime.');
+        }
+
+        $publishTokenFactory = new DefaultClaimsTokenFactory(
+            WebTokenFactory::fromSecret(
+                $this->mercureSecret($config, 'publish'),
+                $config['publish_algorithm'] ?? $config['algorithm'] ?? 'HS256',
+                $publishExpiration,
+                $config['publish_passphrase'] ?? $config['passphrase'] ?? '',
+            ),
+            $this->mercurePublishClaims($config),
+        );
+
+        return new Hub(
+            $config['url'],
+            new CachingTokenProvider(new FactoryTokenProvider($publishTokenFactory, [new Grant([Grant::ACTION_PUBLISH], ['*'])])),
+            $this->mercureSubscribeFactory($config),
+            ($config['public_url'] ?? null) ?: null,
+            empty($config['client_options']) ? null : HttpClient::create($config['client_options']),
+            ($config['cookie_name'] ?? null) ?: null,
+            ProtocolVersion::V1,
+        );
+    }
+
+    /**
+     * Get a Mercure hub instance backed by FrankenPHP's built-in hub.
+     *
+     * FrankenPHP publishes in-process, so no publisher JWT, token provider
+     * or HTTP client is involved; only the subscriber token factory is.
+     *
+     * @param  array  $config
+     * @return \Symfony\Component\Mercure\HubInterface
+     */
+    protected function frankenPhpMercure(array $config)
+    {
+        if (! function_exists('mercure_publish')) {
+            throw new InvalidArgumentException('The Mercure broadcasting connection requires a "url" configuration value, unless the application is served by FrankenPHP with its built-in Mercure hub enabled.');
+        }
+
+        return new FrankenPhpHub(
+            ($config['public_url'] ?? null) ?: '/.well-known/mercure',
+            $this->mercureSubscribeFactory($config),
+            ($config['cookie_name'] ?? null) ?: null,
+            ProtocolVersion::V1,
+        );
+    }
+
+    /**
+     * Get the subscriber token (and cookie) lifetime in whole seconds.
+     *
+     * Shared by the driver and the subscriber factory so the cookie and the
+     * token it carries always expire together. Cast to int up front: a
+     * fractional value would otherwise truncate silently downstream, and a
+     * 0-second lifetime is treated by the hub as one hour.
+     *
+     * @param  array  $config
+     * @return int
+     */
+    protected function mercureSubscribeExpiration(array $config)
+    {
+        return (int) (($config['subscribe_expiration'] ?? 5) * 60);
+    }
+
+    /**
+     * Get the subscriber token factory for the given Mercure configuration,
+     * or null when no subscriber secret is configured. The broadcast driver
+     * itself rejects a factory-less hub; publish-only hubs are only
+     * reachable through the public mercure() helper.
+     *
+     * @param  array  $config
+     * @return \Symfony\Component\Mercure\Jwt\TokenFactoryInterface|null
+     */
+    protected function mercureSubscribeFactory(array $config)
+    {
+        if (empty($config['subscribe_secret']) && empty($config['secret'])) {
+            return null;
+        }
+
+        $claims = $this->mercureClaims($config);
+        // RFC 9068 requires "sub" unconditionally, even for a guest joining
+        // public-only channels; an authenticated user's identity overrides
+        // this default at call time (see MercureBroadcaster::makeAuthorizationCookie()).
+        $claims['sub'] = ($claims['sub'] ?? null) ?: 'anonymous';
+
+        return new DefaultClaimsTokenFactory(
+            WebTokenFactory::fromSecret(
+                $this->mercureSecret($config, 'subscribe'),
+                $config['subscribe_algorithm'] ?? $config['algorithm'] ?? 'HS256',
+                $this->mercureSubscribeExpiration($config),
+                $config['subscribe_passphrase'] ?? $config['passphrase'] ?? '',
+            ),
+            $claims,
+        );
+    }
+
+    /**
+     * Get the end-to-end channel encrypter for the given Mercure
+     * configuration, or null when no "encryption_key" is configured
+     * (end-to-end encrypted channels then throw when used).
+     *
+     * Built eagerly so a missing "web-token/jwt-library" package or a
+     * malformed key surfaces at driver resolution rather than mid-broadcast.
+     *
+     * @param  array  $config
+     * @return \Illuminate\Broadcasting\MercureChannelEncrypter|null
+     */
+    protected function mercureChannelEncrypter(array $config)
+    {
+        if (empty($config['encryption_key'])) {
+            return null;
+        }
+
+        $key = base64_decode($config['encryption_key'], true);
+
+        if ($key === false || strlen($key) !== 32) {
+            throw new InvalidArgumentException('The Mercure "encryption_key" configuration value must be a base64-encoded 32-byte key. You may generate one with: php -r "echo base64_encode(random_bytes(32));"');
+        }
+
+        return new MercureChannelEncrypter($key);
+    }
+
+    /**
+     * Get the additional JWT claims for the publisher side of the given
+     * Mercure configuration.
+     *
+     * The publish token has no request/user to draw a "sub" from, so it
+     * defaults to the app's own "client_id" identity.
+     *
+     * @param  array  $config
+     * @return array
+     */
+    protected function mercurePublishClaims(array $config)
+    {
+        $claims = $this->mercureClaims($config);
+
+        $claims['sub'] = ($claims['sub'] ?? null) ?: (($claims['client_id'] ?? null) ?: $config['url']);
+
+        return $claims;
+    }
+
+    /**
+     * Get the JWT secret for the given side ("subscribe" or "publish") of
+     * the given Mercure configuration.
+     *
+     * @param  array  $config
+     * @param  string  $side
+     * @return string
+     */
+    protected function mercureSecret(array $config, string $side)
+    {
+        $secret = ($config[$side.'_secret'] ?? null) ?: ($config['secret'] ?? null);
+
+        if (empty($secret)) {
+            throw new InvalidArgumentException(sprintf(
+                'The Mercure broadcasting connection requires a "secret" (or "%s_secret") configuration value.', $side
+            ));
+        }
+
+        $algorithm = $config[$side.'_algorithm'] ?? $config['algorithm'] ?? 'HS256';
+
+        // Enforced downstream with a bare "Invalid key length." at the first
+        // token mint; failing here instead points at the configuration.
+        $minimumLength = ['HS256' => 32, 'HS384' => 48, 'HS512' => 64][$algorithm] ?? 0;
+
+        if (strlen($secret) < $minimumLength) {
+            throw new InvalidArgumentException(sprintf(
+                'The Mercure "secret" (or "%s_secret") configuration value must be at least %d bytes long to sign %s tokens.',
+                $side, $minimumLength, $algorithm
+            ));
+        }
+
+        return $secret;
+    }
+
+    /**
+     * Get the additional JWT claims for the given Mercure configuration.
+     *
+     * RFC 9068 access tokens require "iss", "aud", and "client_id", so each
+     * defaults to a sensible identifier when not set explicitly (an empty
+     * string, e.g. from an unset .env value, counts as not set).
+     *
+     * @param  array  $config
+     * @return array
+     */
+    protected function mercureClaims(array $config)
+    {
+        $claims = $config['claims'] ?? [];
+        $appUrl = $this->app['config']['app.url'] ?? null;
+
+        // A FrankenPHP hub has no "url": it falls back to the same public
+        // endpoint the browser subscribes to.
+        $url = ($config['url'] ?? null) ?: (($config['public_url'] ?? null) ?: '/.well-known/mercure');
+
+        if (empty($config['url']) && empty($config['public_url']) && $appUrl) {
+            // The default endpoint is root-relative, but the hub validates
+            // an absolute audience. Do not include the application's path.
+            $origin = parse_url($appUrl);
+
+            if (isset($origin['scheme'], $origin['host'])) {
+                $url = $origin['scheme'].'://'.$origin['host']
+                    .(isset($origin['port']) ? ':'.$origin['port'] : '').$url;
+            }
+        }
+
+        $claims['aud'] = ($claims['aud'] ?? null) ?: (($config['public_url'] ?? null) ?: $url);
+        $claims['iss'] = ($claims['iss'] ?? null) ?: ($appUrl ?: $url);
+        $claims['client_id'] = ($claims['client_id'] ?? null) ?: $claims['iss'];
+
+        return $claims;
+    }
+
     /**
      * Create an instance of the driver.
      *