From 371be24ebf9a9806a4c8a6695bcb3151b439a9c5 Mon Sep 17 00:00:00 2001
From: Yordis Prieto <yordis.prieto@gmail.com>
Date: Sat, 7 Nov 2026 17:27:02 -0500
Subject: [PATCH] feat(natsstore): connect to NATS with the configured
 credentials and TLS

Auth and TLS options map one to one onto the nats CLI variables, and failures must not echo URL userinfo, so connection errors are redacted. Tests run against a real server in Docker so credential handling is exercised for real.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
---
 go.mod                                        |   6 +-
 go.sum                                        |   8 +-
 internal/store/natsstore/connect.go           | 210 ++++++++++++++++++
 internal/store/natsstore/connect_test.go      | 172 ++++++++++++++
 internal/store/natsstore/natstest/natstest.go | 137 ++++++++++++
 5 files changed, 529 insertions(+), 4 deletions(-)
 create mode 100644 internal/store/natsstore/connect.go
 create mode 100644 internal/store/natsstore/connect_test.go
 create mode 100644 internal/store/natsstore/natstest/natstest.go

diff --git a/go.mod b/go.mod
index 5330c35b5f2..ccee8e4514b 100644
--- a/go.mod
+++ b/go.mod
@@ -16,10 +16,12 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/jackc/pgx/v5 v5.9.2
 	github.com/joho/godotenv v1.5.1
-	github.com/klauspost/compress v1.17.4
+	github.com/klauspost/compress v1.20.0
 	github.com/libp2p/zeroconf/v2 v2.2.0
 	github.com/miekg/dns v1.1.43
 	github.com/minio/minio-go/v7 v7.0.66
+	github.com/nats-io/nats.go v1.55.0
+	github.com/nats-io/nkeys v0.4.17
 	github.com/pion/ice/v4 v4.3.0
 	github.com/pion/interceptor v0.1.45
 	github.com/pion/rtp v1.10.4
@@ -46,7 +48,7 @@ replace github.com/MoeForge/moego v0.1.0-alpha.1 => github.com/router-for-me/moe
 require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dlclark/regexp2/v2 v2.5.1 // indirect
-	github.com/nats-io/nkeys v0.4.17 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pion/datachannel v1.6.2 // indirect
 	github.com/pion/dtls/v3 v3.1.5 // indirect
 	github.com/pion/logging v0.2.4 // indirect
diff --git a/go.sum b/go.sum
index b5d9c6b697c..e57d0e9390d 100644
--- a/go.sum
+++ b/go.sum
@@ -110,8 +110,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kevinburke/ssh_config v1.6.0 h1:J1FBfmuVosPHf5GRdltRLhPJtJpTlMdKTBjRgTaQBFY=
 github.com/kevinburke/ssh_config v1.6.0/go.mod h1:q2RIzfka+BXARoNexmF9gkxEX7DmvbW9P4hIVx2Kg4M=
-github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
-github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
+github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
 github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
@@ -155,8 +155,12 @@ github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELU
 github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo=
 github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc=
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
+github.com/nats-io/nats.go v1.55.0 h1:Ng8QKHdvdHIWoaiq0+d/aH1BWkxvcJxWJ0Y/9BPKx3M=
+github.com/nats-io/nats.go v1.55.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
 github.com/nats-io/nkeys v0.4.17 h1:JAJrlDhOyQqOt0a1eg4cbIZ1Kc7+vmkCigAltpwQpVA=
 github.com/nats-io/nkeys v0.4.17/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pelletier/go-toml/v2 v2.4.3-0.20260624044523-35f78d5908f1 h1:gIZQQ2+wbsdQGWLPDSdJFptR25tBCjSC0KbSIQZ1rgE=
 github.com/pelletier/go-toml/v2 v2.4.3-0.20260624044523-35f78d5908f1/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pierrec/xxHash v0.1.5 h1:n/jBpwTHiER4xYvK3/CdPVnLDPchj8eTJFFLUb4QHBo=
diff --git a/internal/store/natsstore/connect.go b/internal/store/natsstore/connect.go
new file mode 100644
index 00000000000..0d8e1b7a9b2
--- /dev/null
+++ b/internal/store/natsstore/connect.go
@@ -0,0 +1,210 @@
+package natsstore
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	// DefaultConnectTimeout bounds one connection attempt and the initial wait.
+	DefaultConnectTimeout = 30 * time.Second
+	// MaxReconnectDelay caps the exponential reconnect backoff.
+	MaxReconnectDelay = 30 * time.Second
+
+	clientName = "cliproxyapi"
+)
+
+// ErrNotConnected is returned when the initial connection is not established in time.
+var ErrNotConnected = errors.New("not connected to NATS")
+
+// reconnectDelay doubles per attempt from one second and stops at MaxReconnectDelay.
+func reconnectDelay(attempts int) time.Duration {
+	if attempts < 1 {
+		attempts = 1
+	}
+	if attempts > 6 {
+		return MaxReconnectDelay
+	}
+	return min(MaxReconnectDelay, time.Duration(1<<uint(attempts-1))*time.Second)
+}
+
+// buildOptions turns the resolved configuration into client options. Secrets are
+// resolved here so a bad reference fails before any network traffic, and no error
+// it returns carries a secret value.
+func buildOptions(cfg Config, env Lookup) ([]nats.Option, error) {
+	var opts []nats.Option
+	switch auth := cfg.Auth.(type) {
+	case CredentialsAuth:
+		path, err := auth.File.Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvCreds, Err: err}
+		}
+		opts = append(opts, nats.UserCredentials(path))
+	case NKeyAuth:
+		seed, err := auth.Seed.resolve()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvNKey, Err: err}
+		}
+		pair, err := nkeys.FromSeed([]byte(seed))
+		if err != nil {
+			return nil, &SettingError{Setting: EnvNKey, Err: fmt.Errorf("invalid NKey seed: %w", err)}
+		}
+		pub, err := pair.PublicKey()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvNKey, Err: err}
+		}
+		opts = append(opts, nats.Nkey(pub, pair.Sign))
+	case UserPasswordAuth:
+		password, err := auth.Password.resolve(env)
+		if err != nil {
+			return nil, &SettingError{Setting: EnvPassword, Err: err}
+		}
+		opts = append(opts, nats.UserInfo(auth.User, password))
+	case JWTAuth:
+		jwt, err := auth.JWT.resolve(env)
+		if err != nil {
+			return nil, &SettingError{Setting: EnvJWT, Err: err}
+		}
+		jwt = strings.TrimSpace(jwt)
+		sign := func([]byte) ([]byte, error) { return nil, nil }
+		if auth.Seed != nil {
+			seed, err := auth.Seed.resolve(env)
+			if err != nil {
+				return nil, &SettingError{Setting: EnvSeed, Err: err}
+			}
+			pair, err := nkeys.FromSeed([]byte(strings.TrimSpace(seed)))
+			if err != nil {
+				return nil, &SettingError{Setting: EnvSeed, Err: fmt.Errorf("invalid NKey seed: %w", err)}
+			}
+			sign = pair.Sign
+		}
+		opts = append(opts, nats.UserJWT(func() (string, error) { return jwt, nil }, sign))
+	case TokenAuth:
+		token, err := auth.Token.resolve(env)
+		if err != nil {
+			return nil, &SettingError{Setting: EnvToken, Err: err}
+		}
+		opts = append(opts, nats.Token(token))
+	}
+
+	if !cfg.TLS.Configured() {
+		return opts, nil
+	}
+	opts = append(opts, nats.Secure())
+	pair, err := cfg.TLS.clientCertificate()
+	if err != nil {
+		return nil, err
+	}
+	if pair != nil {
+		cert, err := pair[0].Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvCert, Err: err}
+		}
+		key, err := pair[1].Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvKey, Err: err}
+		}
+		opts = append(opts, nats.ClientCert(cert, key))
+	}
+	if cfg.TLS.CA != nil {
+		ca, err := cfg.TLS.CA.Path()
+		if err != nil {
+			return nil, &SettingError{Setting: EnvCA, Err: err}
+		}
+		opts = append(opts, nats.RootCAs(ca))
+	}
+	return opts, nil
+}
+
+// Connect opens a client connection that keeps reconnecting with capped backoff,
+// and waits until the first connection is up or the timeout or context ends.
+// Server addresses are logged with any URL userinfo removed, and secrets never are.
+func Connect(ctx context.Context, cfg Config, env Lookup, timeout time.Duration) (*nats.Conn, error) {
+	if timeout <= 0 {
+		timeout = DefaultConnectTimeout
+	}
+	opts, err := buildOptions(cfg, env)
+	if err != nil {
+		return nil, err
+	}
+
+	var lastErr atomic.Value
+	remember := func(err error) {
+		if err != nil {
+			lastErr.Store(err.Error())
+		}
+	}
+	servers := redactServers(cfg.Servers)
+	opts = append(opts,
+		nats.Name(clientName),
+		nats.RetryOnFailedConnect(true),
+		nats.MaxReconnects(-1),
+		nats.Timeout(timeout),
+		nats.CustomReconnectDelay(reconnectDelay),
+		nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
+			remember(err)
+			log.WithError(err).Warn("NATS disconnected, will attempt reconnect")
+		}),
+		nats.ReconnectHandler(func(*nats.Conn) { log.Info("NATS reconnected") }),
+		nats.ClosedHandler(func(*nats.Conn) { log.Info("NATS connection closed") }),
+		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+			remember(err)
+			log.WithError(err).Warn("NATS error")
+		}),
+	)
+
+	log.Infof("connecting to NATS servers %v with %s authentication", servers, cfg.Auth.Description())
+	nc, err := nats.Connect(strings.Join(cfg.Servers, ","), opts...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to NATS servers %v: %w", servers, scrub(err, cfg.Servers))
+	}
+
+	deadline := time.NewTimer(timeout)
+	defer deadline.Stop()
+	tick := time.NewTicker(25 * time.Millisecond)
+	defer tick.Stop()
+	for !nc.IsConnected() {
+		var reason error
+		select {
+		case <-ctx.Done():
+			reason = ctx.Err()
+		case <-deadline.C:
+			reason = ErrNotConnected
+		case <-tick.C:
+			if nc.IsClosed() {
+				reason = ErrNotConnected
+			}
+		}
+		if reason != nil {
+			nc.Close()
+			if last, ok := lastErr.Load().(string); ok {
+				return nil, fmt.Errorf("failed to connect to NATS servers %v: %w (last error: %s)", servers, reason, last)
+			}
+			return nil, fmt.Errorf("failed to connect to NATS servers %v: %w", servers, reason)
+		}
+	}
+	log.Infof("connected to NATS server %s", nc.ConnectedUrlRedacted())
+	return nc, nil
+}
+
+// scrub removes every configured server string from an error message so URL userinfo cannot leak.
+func scrub(err error, servers []string) error {
+	msg := err.Error()
+	for _, server := range servers {
+		if safe := redactServer(server); safe != server {
+			msg = strings.ReplaceAll(msg, server, safe)
+		}
+	}
+	if msg == err.Error() {
+		return err
+	}
+	return errors.New(msg)
+}
diff --git a/internal/store/natsstore/connect_test.go b/internal/store/natsstore/connect_test.go
new file mode 100644
index 00000000000..ec355485003
--- /dev/null
+++ b/internal/store/natsstore/connect_test.go
@@ -0,0 +1,172 @@
+package natsstore
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
+
+	"github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore/natstest"
+)
+
+func testContext(t *testing.T) context.Context {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	t.Cleanup(cancel)
+	return ctx
+}
+
+func connect(t *testing.T, env map[string]string, timeout time.Duration) (*nats.Conn, error) {
+	t.Helper()
+	cfg := SettingsFromEnv(mapEnv(env)).Resolve()
+	nc, err := Connect(testContext(t), cfg, mapEnv(env), timeout)
+	if err == nil {
+		t.Cleanup(nc.Close)
+	}
+	return nc, err
+}
+
+func TestConnectAuthMethods(t *testing.T) {
+	const (
+		user     = "svc"
+		password = "pw-integration-secret"
+		token    = "token-integration-secret"
+	)
+	pair, err := nkeys.CreateUser()
+	if err != nil {
+		t.Fatal(err)
+	}
+	pub, _ := pair.PublicKey()
+	seed, _ := pair.Seed()
+	dir := t.TempDir()
+	conf := filepath.Join(dir, "nats.conf")
+	if err := os.WriteFile(conf, []byte("authorization { users = [ { nkey: "+pub+" } ] }\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	seedFile := filepath.Join(dir, "user.seed")
+	if err := os.WriteFile(seedFile, append(seed, '\n'), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	tokenFile := filepath.Join(dir, "token")
+	if err := os.WriteFile(tokenFile, []byte(token+"\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("token", func(t *testing.T) {
+		srv := natstest.Start(t, natstest.WithArgs("--auth", token))
+		for name, value := range map[string]string{"literal": token, "file": "file://" + tokenFile, "env": "env://INTEGRATION_TOKEN"} {
+			t.Run(name, func(t *testing.T) {
+				if _, err := connect(t, map[string]string{EnvURL: srv.URL, EnvToken: value, "INTEGRATION_TOKEN": token}, 5*time.Second); err != nil {
+					t.Fatal(err)
+				}
+			})
+		}
+	})
+
+	t.Run("user and password", func(t *testing.T) {
+		srv := natstest.Start(t, natstest.WithArgs("--user", user, "--pass", password))
+		if _, err := connect(t, map[string]string{EnvURL: srv.URL, EnvUser: user, EnvPassword: password}, 5*time.Second); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	t.Run("nkey", func(t *testing.T) {
+		srv := natstest.Start(t, natstest.Mount(dir, "/etc/nats-test"), natstest.WithArgs("-c", "/etc/nats-test/nats.conf"))
+		for name, value := range map[string]string{"inline": string(seed), "file": seedFile} {
+			t.Run(name, func(t *testing.T) {
+				if _, err := connect(t, map[string]string{EnvURL: srv.URL, EnvNKey: value}, 5*time.Second); err != nil {
+					t.Fatal(err)
+				}
+			})
+		}
+	})
+
+	t.Run("wrong credentials fail without leaking them", func(t *testing.T) {
+		srv := natstest.Start(t, natstest.WithArgs("--auth", token))
+		const wrong = "wrong-token-integration-secret"
+		_, err := connect(t, map[string]string{EnvURL: srv.URL, EnvToken: wrong}, 2*time.Second)
+		if err == nil {
+			t.Fatal("connected with a wrong token")
+		}
+		if strings.Contains(err.Error(), wrong) || strings.Contains(err.Error(), token) {
+			t.Fatalf("error leaks a secret: %v", err)
+		}
+	})
+
+	t.Run("anonymous is refused by an authenticated server", func(t *testing.T) {
+		srv := natstest.Start(t, natstest.WithArgs("--auth", token))
+		if _, err := connect(t, map[string]string{EnvURL: srv.URL}, 2*time.Second); err == nil {
+			t.Fatal("connected without credentials")
+		}
+	})
+}
+
+func TestConnectErrorsDoNotLeakURLUserinfo(t *testing.T) {
+	const url = "nats://svc:urlpass-secret@127.0.0.1:1"
+	_, err := connect(t, map[string]string{EnvURL: url}, 500*time.Millisecond)
+	if err == nil {
+		t.Fatal("connected to a closed port")
+	}
+	if strings.Contains(err.Error(), "urlpass-secret") {
+		t.Fatalf("error leaks URL userinfo: %v", err)
+	}
+}
+
+func TestTLSClientCertificateNeedsBothHalves(t *testing.T) {
+	tests := []struct {
+		name    string
+		env     map[string]string
+		wantSet string
+		wantMsg string
+	}{
+		{"cert alone", map[string]string{EnvCert: "/c.pem"}, EnvCert, "must be set together with " + EnvKey},
+		{"key alone", map[string]string{EnvKey: "/k.pem"}, EnvKey, "must be set together with " + EnvCert},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := SettingsFromEnv(mapEnv(tt.env)).Resolve()
+			_, err := buildOptions(cfg, OSLookup)
+			var setting *SettingError
+			if !errors.As(err, &setting) {
+				t.Fatalf("err = %v, want SettingError", err)
+			}
+			if setting.Setting != tt.wantSet || !strings.Contains(setting.Error(), tt.wantMsg) {
+				t.Fatalf("err = %v", setting)
+			}
+		})
+	}
+	cfg := SettingsFromEnv(mapEnv(map[string]string{EnvCA: "/ca.pem"})).Resolve()
+	if opts, err := buildOptions(cfg, OSLookup); err != nil || len(opts) == 0 {
+		t.Fatalf("CA alone: %d options, %v", len(opts), err)
+	}
+}
+
+func TestBuildOptionsRejectsUnsupportedReference(t *testing.T) {
+	cfg := SettingsFromEnv(mapEnv(map[string]string{EnvToken: "op://vault/item"})).Resolve()
+	_, err := buildOptions(cfg, OSLookup)
+	if !errors.Is(err, ErrUnsupportedScheme) {
+		t.Fatalf("err = %v", err)
+	}
+	var setting *SettingError
+	if !errors.As(err, &setting) || setting.Setting != EnvToken {
+		t.Fatalf("err = %v, want a %s setting error", err, EnvToken)
+	}
+}
+
+func TestReconnectDelayBacksOffAndCaps(t *testing.T) {
+	want := []int{1, 2, 4, 8, 16, 30, 30}
+	for i, seconds := range want {
+		if got := reconnectDelay(i + 1); got.Seconds() != float64(seconds) {
+			t.Errorf("attempt %d: %v, want %ds", i+1, got, seconds)
+		}
+	}
+	if got := reconnectDelay(0); got.Seconds() != 1 {
+		t.Errorf("attempt 0: %v", got)
+	}
+}
diff --git a/internal/store/natsstore/natstest/natstest.go b/internal/store/natsstore/natstest/natstest.go
new file mode 100644
index 00000000000..8414c598c61
--- /dev/null
+++ b/internal/store/natsstore/natstest/natstest.go
@@ -0,0 +1,137 @@
+// Package natstest runs a real NATS server in a Docker container for tests, so
+// they exercise the same server the backend runs against. Tests skip cleanly
+// when Docker is unavailable.
+package natstest
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+)
+
+// DefaultImage is the server image the tests run against.
+const DefaultImage = "nats:2.14.9"
+
+// EnvImage overrides DefaultImage, for trying another server version.
+const EnvImage = "NATS_TEST_IMAGE"
+
+// Server is a running containerized NATS server.
+type Server struct {
+	// URL is the client address on the host loopback.
+	URL string
+	// Container is the container name.
+	Container string
+}
+
+type settings struct {
+	serverArgs []string
+	mounts     []string
+}
+
+// Option customizes the container.
+type Option func(*settings)
+
+// WithArgs appends raw nats-server arguments after the defaults.
+func WithArgs(args ...string) Option {
+	return func(s *settings) { s.serverArgs = append(s.serverArgs, args...) }
+}
+
+// Mount makes hostPath available read-only inside the container at containerPath.
+func Mount(hostPath, containerPath string) Option {
+	return func(s *settings) { s.mounts = append(s.mounts, hostPath+":"+containerPath+":ro") }
+}
+
+func docker(ctx context.Context, args ...string) (string, error) {
+	cmd := exec.CommandContext(ctx, "docker", args...)
+	var out, errOut bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &out, &errOut
+	if err := cmd.Run(); err != nil {
+		return "", fmt.Errorf("docker %s: %w: %s", args[0], err, strings.TrimSpace(errOut.String()))
+	}
+	return strings.TrimSpace(out.String()), nil
+}
+
+// Start launches a JetStream-enabled server and stops it when the test ends. It
+// skips the test when Docker or the image is unavailable.
+func Start(t testing.TB, opts ...Option) *Server {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	if _, err := exec.LookPath("docker"); err != nil {
+		t.Skip("docker is not installed")
+	}
+	if _, err := docker(ctx, "info", "--format", "{{.ServerVersion}}"); err != nil {
+		t.Skipf("docker is not available: %v", err)
+	}
+	image := DefaultImage
+	if v := os.Getenv(EnvImage); v != "" {
+		image = v
+	}
+	if _, err := docker(ctx, "image", "inspect", image); err != nil {
+		if _, err := docker(ctx, "pull", image); err != nil {
+			t.Skipf("image %s is not available: %v", image, err)
+		}
+	}
+
+	var cfg settings
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+	args := []string{"run", "-d", "--rm", "-p", "127.0.0.1::4222"}
+	for _, mount := range cfg.mounts {
+		args = append(args, "-v", mount)
+	}
+	args = append(args, image, "-js", "-sd", "/data")
+	args = append(args, cfg.serverArgs...)
+
+	id, err := docker(ctx, args...)
+	if err != nil {
+		t.Skipf("cannot start %s: %v", image, err)
+	}
+	t.Cleanup(func() {
+		stopCtx, stop := context.WithTimeout(context.Background(), 30*time.Second)
+		defer stop()
+		_, _ = docker(stopCtx, "rm", "-f", id)
+	})
+
+	hostPort, err := docker(ctx, "port", id, "4222/tcp")
+	if err != nil {
+		t.Fatalf("cannot resolve the published port: %v", err)
+	}
+	// Output may list IPv4 and IPv6 bindings on separate lines; the loopback one is first.
+	addr := strings.SplitN(hostPort, "\n", 2)[0]
+	_, port, err := net.SplitHostPort(strings.TrimSpace(addr))
+	if err != nil {
+		t.Fatalf("unexpected port mapping %q: %v", hostPort, err)
+	}
+	srv := &Server{URL: "nats://127.0.0.1:" + port, Container: id}
+	waitReady(t, srv.URL[len("nats://"):])
+	return srv
+}
+
+// waitReady waits until the server answers the NATS INFO handshake.
+func waitReady(t testing.TB, addr string) {
+	t.Helper()
+	deadline := time.Now().Add(30 * time.Second)
+	for time.Now().Before(deadline) {
+		conn, err := net.DialTimeout("tcp", addr, time.Second)
+		if err == nil {
+			buf := make([]byte, 5)
+			_ = conn.SetReadDeadline(time.Now().Add(time.Second))
+			n, _ := conn.Read(buf)
+			_ = conn.Close()
+			if string(buf[:n]) == "INFO " {
+				return
+			}
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	t.Fatalf("NATS server at %s did not become ready", addr)
+}