Add TLS support (upgrade-after-INFO)

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.

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,
         );
     }
 }

src/Internal/Connection/Framer.php

@@ -21,18 +21,52 @@
     /** @var ConcurrentIterator<Protocol\Frame> */
     private ConcurrentIterator $iterator;
 
-    public function __construct(Socket $socket)
+    public function __construct(Socket $socket, bool $upgradeTls = false)
     {
         $this->writer = new Writer($socket);
 
         /** @var Queue<Protocol\Frame> $queue */
         $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);
                 }

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();

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);
     }
 

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([