diff --git a/README.md b/README.md index 7626169..e518425 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,29 @@ Pure non-blocking (fiber based) strictly typed full-featured PHP driver for NATS composer require thesis/nats ``` +## TLS + +For servers that require TLS (for example [Synadia NGS](https://www.synadia.com/), `tls://connect.ngs.global`), use a `tls://` URI — TLS is enabled automatically with the host as the peer name for certificate verification: + +```php +$nc = new Nats\Client( + Nats\Config::fromURI('tls://connect.ngs.global:4222?jwt=...&nkey=...'), +); +``` + +To customise the TLS parameters (CA file, peer verification, minimum version, …), pass an `Amp\Socket\ClientTlsContext` explicitly: + +```php +use Amp\Socket\ClientTlsContext; + +$nc = new Nats\Client(new Nats\Config( + urls: ['nats.internal:4222'], + tls: (new ClientTlsContext('nats.internal'))->withCaFile('/etc/ssl/ca.pem'), +)); +``` + +The client follows the standard NATS upgrade-after-`INFO` flow: it reads the plaintext `INFO`, then performs the TLS handshake before sending `CONNECT`. Servers configured with `tls { handshake_first: true }` are not currently supported. + ## Nats Core The library implements the full functionality of NATS Core, including pub-sub, queues and request–reply. diff --git a/src/Config.php b/src/Config.php index fada1f2..aba7ac8 100644 --- a/src/Config.php +++ b/src/Config.php @@ -4,6 +4,8 @@ namespace Thesis\Nats; +use Amp\Socket\ClientTlsContext; + /** * @api */ @@ -54,6 +56,11 @@ public function __construct( public int $maxPings = self::DEFAULT_MAX_PINGS, public ?string $jetStreamDomain = null, public string $clientName = self::DEFAULT_CLIENT_NAME, + // TLS client context. null keeps the existing plaintext behavior. When + // set, the connection upgrades to TLS after the server's plaintext INFO + // (the standard NATS tls_required flow, e.g. Synadia NGS). The upgrade is + // performed by the socket's single reader fiber — see Framer. + public ?ClientTlsContext $tls = null, ) { $this->version = '0.1.x'; } @@ -75,6 +82,17 @@ public static function fromURI(#[\SensitiveParameter] string $uri): self throw new \InvalidArgumentException("The uri '{$uri}' is invalid."); } + // A tls:// (or nats+tls://) scheme turns on TLS — the same convention the + // Go/JS clients use for connect.ngs.global. Default the SNI/peer name to + // the host so certificate verification works against NGS. + $tls = null; + $scheme = strtolower($components['scheme'] ?? ''); + if (($scheme === 'tls' || $scheme === 'nats+tls' || $scheme === 'ssl') && ($components['host'] ?? '') !== '') { + $firstHost = explode(':', explode(',', $components['host'])[0])[0]; + $tls = new ClientTlsContext($firstHost); + } + + $query = []; if (isset($components['query']) && $components['query'] !== '') { parse_str($components['query'], $query); @@ -177,6 +195,7 @@ public static function fromURI(#[\SensitiveParameter] string $uri): self maxPings: $maxPings, jetStreamDomain: $jetStreamDomain, clientName: $clientName, + tls: $tls, ); } @@ -196,6 +215,7 @@ public static function fromURI(#[\SensitiveParameter] string $uri): self * max_pings?: positive-int, * jetstream_domain?: non-empty-string, * client_name?: non-empty-string, + * tls?: ClientTlsContext, * } $options */ public static function fromArray(#[\SensitiveParameter] array $options): self @@ -215,6 +235,7 @@ public static function fromArray(#[\SensitiveParameter] array $options): self maxPings: $options['max_pings'] ?? self::DEFAULT_MAX_PINGS, jetStreamDomain: $options['jetstream_domain'] ?? null, clientName: $options['client_name'] ?? self::DEFAULT_CLIENT_NAME, + tls: $options['tls'] ?? null, ); } } diff --git a/src/Internal/Connection/Framer.php b/src/Internal/Connection/Framer.php index 53738f5..ed89bcf 100644 --- a/src/Internal/Connection/Framer.php +++ b/src/Internal/Connection/Framer.php @@ -21,7 +21,7 @@ /** @var ConcurrentIterator */ private ConcurrentIterator $iterator; - public function __construct(Socket $socket) + public function __construct(Socket $socket, bool $upgradeTls = false) { $this->writer = new Writer($socket); @@ -29,10 +29,44 @@ public function __construct(Socket $socket) $queue = new Queue(); $this->iterator = $queue->iterate(); - EventLoop::queue(static function () use ($socket, $queue): void { + EventLoop::queue(static function () use ($socket, $queue, $upgradeTls): void { $parser = new Protocol\Parser($queue->push(...)); try { + // TLS upgrade (NATS `tls_required`, e.g. Synadia NGS). The server + // sends a plaintext INFO first, then expects the client to run the + // TLS handshake before CONNECT. We do BOTH here — in the single + // fiber that owns socket reads — so setupTls() never races another + // read (amphp forbids concurrent reads on a socket). We read up to + // the end of the plaintext INFO line, hand those bytes to the + // parser (so startup() still gets the INFO frame + its nonce), then + // upgrade in place. NGS sends nothing between INFO and the upgrade, + // so no post-INFO plaintext bytes are lost. After this the loop + // below reads the encrypted stream exactly as normal. + if ($upgradeTls) { + $preamble = ''; + while (!str_contains($preamble, "\r\n")) { + $bytes = $socket->read(); + if ($bytes === null) { + $parser->cancel(); + $queue->complete(); + $socket->close(); + return; + } + $preamble .= $bytes; + } + // Upgrade BEFORE handing the INFO frame to the parser/queue. + // Order matters: parser->push() makes the INFO frame available + // to startup() (a DIFFERENT fiber), which then immediately + // writes CONNECT. If we pushed before setupTls() completed, that + // CONNECT would be written mid-TLS-handshake and corrupt the + // stream (server resets). setupTls() is synchronous-until-done + // here, so once it returns the channel is encrypted and it's + // safe to release the INFO frame. + $socket->setupTls(); + $parser->push($preamble); + } + while (($bytes = $socket->read()) !== null) { $parser->push($bytes); } diff --git a/src/Internal/Connection/SocketConnection.php b/src/Internal/Connection/SocketConnection.php index c5173e8..e34d69d 100644 --- a/src/Internal/Connection/SocketConnection.php +++ b/src/Internal/Connection/SocketConnection.php @@ -40,7 +40,10 @@ public function __construct( private readonly Config $config, private readonly Socket $socket, ) { - $this->framer = new Framer($this->socket); + // When a TLS context is configured, the Framer upgrades the socket to + // TLS inside its reader fiber (after the plaintext INFO, before its read + // loop) so the single socket reader also owns the handshake. + $this->framer = new Framer($this->socket, $this->config->tls !== null); $this->hooks = new Hooks\ConcurrentProvider(); $this->pingpongs = new PingPongHandler($this); $this->signer = new Signer(); diff --git a/src/Internal/Connection/SocketConnectionFactory.php b/src/Internal/Connection/SocketConnectionFactory.php index e92c14f..a5b9d18 100644 --- a/src/Internal/Connection/SocketConnectionFactory.php +++ b/src/Internal/Connection/SocketConnectionFactory.php @@ -29,6 +29,14 @@ public static function fromConfig(Config $config): self $context = $context->withTcpNoDelay(); } + // Attach the TLS context so the socket can be upgraded later. We do NOT + // connect over TLS immediately: NATS sends a plaintext INFO first, then + // the client upgrades. The upgrade itself runs in the Framer's reader + // fiber; setting the context here just makes setupTls() available to it. + if ($config->tls !== null) { + $context = $context->withTlsContext($config->tls); + } + return new self($config, $context); } diff --git a/tests/ConfigTest.php b/tests/ConfigTest.php index 5ebfe52..06b62d4 100644 --- a/tests/ConfigTest.php +++ b/tests/ConfigTest.php @@ -188,6 +188,32 @@ public function testFromURI(string $uri, Config $config): void self::assertEquals($config, Config::fromURI($uri)); } + /** + * @param non-empty-string $uri + */ + #[TestWith(['tls://connect.ngs.global:4222', 'connect.ngs.global'])] + #[TestWith(['nats+tls://example.com:4222', 'example.com'])] + #[TestWith(['ssl://nats.internal:4222', 'nats.internal'])] + public function testFromURIEnablesTlsFromScheme(string $uri, string $peerName): void + { + $config = Config::fromURI($uri); + + self::assertNotNull($config->tls); + self::assertEquals(new \Amp\Socket\ClientTlsContext($peerName), $config->tls); + } + + public function testFromURIWithoutTlsSchemeLeavesTlsNull(): void + { + self::assertNull(Config::fromURI('tcp://127.0.0.1:4222')->tls); + } + + public function testFromArrayAcceptsTlsContext(): void + { + $tls = new \Amp\Socket\ClientTlsContext('nats.example.com'); + + self::assertSame($tls, Config::fromArray(['tls' => $tls])->tls); + } + public function testFromArrayWithJwtAndNkey(): void { $config = Config::fromArray([