feat: NATS mTLS pubsub implementation (#26902)

coderd/x/nats/cluster.go

@@ -11,6 +11,7 @@ import (
 	"golang.org/x/xerrors"
 
 	"cdr.dev/slog/v3"
+	"github.com/coder/coder/v2/coderd/cryptokeys"
 )
 
 const defaultClusterTokenUsername = "coder"
@@ -49,6 +50,21 @@ func (p *Pubsub) SetPeerFetcher(fetcher PeerFetcher) {
 	p.RefreshPeers()
 }
 
+// SetCACache swaps the cluster mTLS CA cache, then triggers a peer refresh so
+// any route blocked by the previous (for example noop) cache is retried. It is
+// a no-op unless the pubsub was started with cluster TLS enabled
+// (Options.ClusterCA set, which installs the TLS callbacks). Passing a noop
+// cache reverts to no mTLS: new route handshakes can no longer mint a leaf and
+// will not form. The leaf IP SAN is fixed at construction from ClusterHost, so
+// it is not passed here. It logs the resulting mTLS state.
+func (p *Pubsub) SetCACache(ca cryptokeys.SigningKeycache) {
+	if p.clusterTLS == nil {
+		return
+	}
+	p.clusterTLS.setCACache(ca)
+	p.RefreshPeers()
+}
+
 // RefreshPeers signals the peer refresh worker to fetch and apply the latest
 // peer route addresses. Multiple pending refreshes are coalesced.
 func (p *Pubsub) RefreshPeers() {

coderd/x/nats/pubsub.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"hash/fnv"
+	"net"
 	"net/url"
 	"sync"
 	"time"
@@ -15,7 +16,9 @@ import (
 	"golang.org/x/xerrors"
 
 	"cdr.dev/slog/v3"
+	"github.com/coder/coder/v2/coderd/cryptokeys"
 	"github.com/coder/coder/v2/coderd/database/pubsub"
+	"github.com/coder/quartz"
 )
 
 // DefaultServerMaxPendingBytes caps how many bytes the embedded NATS server will
@@ -121,6 +124,24 @@ type Options struct {
 	// clustered embedded NATS servers. Empty disables route auth.
 	ClusterAuthToken string
 
+	// ClusterCA enables mutual TLS on the cluster route listener. When set
+	// (and cluster mode is enabled), each replica mints an ephemeral leaf
+	// certificate from the active nats_ca CA and verifies peers against the
+	// CA fetched from this cache on each handshake. Nil keeps routes
+	// plaintext (token auth only). cryptokeys.SigningKeycache satisfies this.
+	//
+	// The leaf's IP SAN (and the accept-side source binding) is this replica's
+	// ClusterHost, so ClusterHost must be an IP for mTLS to activate.
+	ClusterCA cryptokeys.SigningKeycache
+
+	// clock overrides the cluster TLS clock, for tests.
+	clock quartz.Clock
+
+	// clusterTLSTimeout overrides the cluster route TLS handshake timeout, for
+	// tests. Zero leaves the NATS default (2s). Tests use a longer timeout
+	// because handshakes are flaky under load and in CI.
+	clusterTLSTimeout time.Duration
+
 	// PeerFetcher provides the current set of peer route addresses.
 	// RefreshPeers uses it to update the configured cluster routes.
 	PeerFetcher PeerFetcher
@@ -182,6 +203,8 @@ type Pubsub struct {
 	clustered     bool
 	serverOpts    *natsserver.Options
 	currentRoutes []*url.URL
+	// clusterTLS is non-nil when the cluster route listener runs mutual TLS.
+	clusterTLS *clusterTLS
 
 	peerFetcher PeerFetcher
 	peerRefresh chan struct{}
@@ -304,6 +327,25 @@ func New(ctx context.Context, logger slog.Logger, opts Options) (pubSub *Pubsub,
 		return nil, err
 	}
 
+	// When ClusterCA is set, install the cluster TLS callbacks at boot so the
+	// route listener can negotiate mTLS. The callbacks read the CA cache on
+	// each handshake, so the default noop cache keeps routes inert (no leaf can
+	// be minted) until SetCACache swaps in a real cache. The leaf IP SAN is this
+	// replica's ClusterHost, fixed here at construction; leaf minting enforces
+	// that it is an IP. ClusterCA == nil keeps routes plaintext (token auth
+	// only).
+	var ct *clusterTLS
+	if !opts.disableCluster && opts.ClusterCA != nil {
+		selfIP := net.ParseIP(opts.ClusterHost)
+		ct = newClusterTLS(ctx, logger, opts.clock, opts.ClusterCA, selfIP)
+		sopts.Cluster.TLSConfig = ct.tlsConfig()
+		// Leave TLSTimeout unset (NATS defaults to 2s) unless a test overrides
+		// it; the default has not shown a need to change in production.
+		if opts.clusterTLSTimeout > 0 {
+			sopts.Cluster.TLSTimeout = opts.clusterTLSTimeout.Seconds()
+		}
+	}
+
 	ns, err := startEmbeddedServer(sopts)
 	if err != nil {
 		return nil, err
@@ -333,6 +375,7 @@ func New(ctx context.Context, logger slog.Logger, opts Options) (pubSub *Pubsub,
 	p.clustered = !opts.disableCluster
 	p.serverOpts = sopts.Clone()
 	p.currentRoutes = cloneRouteURLs(sopts.Routes)
+	p.clusterTLS = ct
 	handlers := p.buildConnHandlers()
 
 	publishPool, err := newConnPool(ns, opts, handlers, opts.PublishConns, "coder-pubsub-pub")

coderd/x/nats/pubsub_internal_test.go

@@ -613,13 +613,19 @@ func defaultTestOptions() Options {
 	return Options{disableCluster: true}
 }
 
+// testClusterTLSTimeout relaxes the cluster route TLS handshake timeout in
+// tests. NATS defaults to a tight 2s, which is flaky under load and in CI;
+// production keeps the default until it is shown to need changing.
+const testClusterTLSTimeout = 10 * time.Second
+
 func clusterTestOptions(t *testing.T) Options {
 	t.Helper()
 	return Options{
-		ClusterHost:      "127.0.0.1",
-		ClusterPort:      natsserver.RANDOM_PORT,
-		disableCluster:   false,
-		ClusterAuthToken: fmt.Sprintf("shared-token-%d", time.Now().UnixNano()),
+		ClusterHost:       "127.0.0.1",
+		ClusterPort:       natsserver.RANDOM_PORT,
+		disableCluster:    false,
+		ClusterAuthToken:  fmt.Sprintf("shared-token-%d", time.Now().UnixNano()),
+		clusterTLSTimeout: testClusterTLSTimeout,
 	}
 }
 

coderd/x/nats/tls.go

@@ -0,0 +1,429 @@
+package nats
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"math/big"
+	"net"
+	"slices"
+	"strconv"
+	"sync"
+	"time"
+
+	"golang.org/x/xerrors"
+
+	"cdr.dev/slog/v3"
+	"github.com/coder/coder/v2/coderd/cryptokeys"
+	"github.com/coder/quartz"
+)
+
+const (
+	// leafSerialBits is the entropy of a leaf certificate serial number.
+	leafSerialBits = 128
+	// clockSkewToleranceTLS backdates a leaf's NotBefore so a peer with a
+	// mildly skewed clock still accepts a freshly minted leaf.
+	clockSkewToleranceTLS = time.Hour
+)
+
+// clusterTLS builds the cluster route *tls.Config. Certificate selection and
+// peer verification are tls.Config callbacks that consult the CA cache on each
+// use, so a CA rotation is tracked without restarting or reloading the server.
+type clusterTLS struct {
+	ctx    context.Context
+	logger slog.Logger
+	clock  quartz.Clock
+
+	mu sync.Mutex
+	// ca is swapped by setCACache: the default noop cache mints no leaf (so no
+	// route forms) until the real cache is installed once cluster mTLS is
+	// enabled. ip is this replica's cluster host, fixed at construction and
+	// embedded as the leaf IP SAN.
+	ca cryptokeys.SigningKeycache
+	ip net.IP
+	// leaf is the cached leaf certificate, reused until it expires (its NotAfter
+	// equals the signing CA's) or setCACache clears it on a cache swap.
+	leaf *tls.Certificate
+	// verifyPools caches the root pool used to verify a peer leaf, keyed by the
+	// CA sequence stamped in the leaf. A CA cert is immutable for a given
+	// sequence, so the pool is built once and reused across handshakes. Expired
+	// entries are pruned on insert to bound the map across rotations.
+	verifyPools map[string]cachedVerifyPool
+}
+
+// cachedVerifyPool is a verify root pool plus the NotAfter of the CA cert it
+// holds; the entry is dropped once the clock passes notAfter.
+type cachedVerifyPool struct {
+	pool     *x509.CertPool
+	notAfter time.Time
+}
+
+func newClusterTLS(ctx context.Context, logger slog.Logger, clock quartz.Clock, ca cryptokeys.SigningKeycache, ip net.IP) *clusterTLS {
+	if clock == nil {
+		clock = quartz.NewReal()
+	}
+	return &clusterTLS{
+		ctx:    ctx,
+		logger: logger.Named("cluster_tls"),
+		clock:  clock,
+		ca:     ca,
+		ip:     ip,
+	}
+}
+
+// setCACache swaps the CA cache. Because the tls.Config callbacks read it on
+// each handshake, the swap takes effect without a server restart or route
+// reload: installing the real cache lets routes negotiate mTLS, and reverting
+// to a noop cache makes leaf minting fail so no new route can form. The leaf IP
+// SAN is fixed at construction (this replica's cluster host does not change), so
+// it is not touched here. A swap clears the cached leaf so the next handshake
+// re-mints under the new CA.
+func (t *clusterTLS) setCACache(ca cryptokeys.SigningKeycache) {
+	t.mu.Lock()
+	t.ca = ca
+	t.leaf = nil
+	// Verify pools follow the CA source: drop them so stale roots are not
+	// reused after a swap to a noop or different CA.
+	t.verifyPools = nil
+	ip := t.ip
+	t.mu.Unlock()
+
+	// Log the resulting mTLS state. A noop cache disables mTLS (no leaf can be
+	// minted); a real cache with a valid self IP enables it; a real cache
+	// without an IP cluster host leaves routes plaintext (token auth only).
+	switch {
+	case isNoopSigningCache(ca):
+		t.logger.Info(t.ctx, "nats cluster mTLS disabled")
+	case len(ip) == 0:
+		t.logger.Warn(t.ctx, "nats cluster mTLS inactive: cluster host is not an IP; cluster routes use token auth only")
+	default:
+		t.logger.Info(t.ctx, "nats cluster mTLS enabled")
+	}
+}
+
+// isNoopSigningCache reports whether ca is the no-op cache used to disable
+// cluster mTLS.
+func isNoopSigningCache(ca cryptokeys.SigningKeycache) bool {
+	_, ok := ca.(cryptokeys.NoopSigningKeycache)
+	return ok
+}
+
+// caCache returns the current CA cache under lock so callers do not hold the
+// lock across cache I/O.
+func (t *clusterTLS) caCache() cryptokeys.SigningKeycache {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.ca
+}
+
+// tlsConfig returns the *tls.Config for the embedded server's cluster route
+// listener. The same config is used by NATS for both accepting inbound routes
+// (TLS server) and soliciting outbound routes (TLS client), so it sets both
+// GetCertificate and GetClientCertificate.
+//
+// Verification is done in VerifyConnection against the CA fetched fresh from
+// the cache, not against a static RootCAs/ClientCAs pool that cannot follow a
+// rotating CA. InsecureSkipVerify disables Go's default static-root check on
+// the dialing side ONLY so verifyConnection can run instead; it does not make
+// the connection unauthenticated. Every connection is still mutually verified
+// (ClientAuth requires a peer certificate) against live CA material.
+//
+// GetConfigForClient runs only when accepting a route (TLS server side), where
+// the dialing peer's source IP is available on the underlying connection. It
+// returns a per-connection config whose VerifyConnection additionally requires
+// the peer leaf's IP SAN to match that source IP, binding the certificate to
+// the network origin. The dialing side has no equivalent hook (Go does not
+// expose the connection in client-certificate callbacks), so it relies on the
+// base VerifyConnection: chain + membership against the known peer set.
+func (t *clusterTLS) tlsConfig() *tls.Config {
+	return &tls.Config{
+		MinVersion: tls.VersionTLS13,
+		GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
+			leaf, err := t.currentLeaf()
+			if err != nil {
+				t.logger.Warn(t.ctx, "get nats cluster leaf for GetCertificate", slog.Error(err))
+			}
+			return leaf, err
+		},
+		GetClientCertificate: func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
+			leaf, err := t.currentLeaf()
+			if err != nil {
+				t.logger.Warn(t.ctx, "get nats cluster leaf for GetClientCertificate", slog.Error(err))
+			}
+			return leaf, err
+		},
+		ClientAuth: tls.RequireAnyClientCert,
+		//nolint:gosec // Not insecure: verify performs full chain verification
+		// against the live CA cache. Go's static RootCAs cannot track a rotating
+		// CA, so default verification is replaced, not removed.
+		InsecureSkipVerify: true,
+		VerifyConnection: func(cs tls.ConnectionState) error {
+			err := t.verify(cs, nil)
+			if err != nil {
+				t.logger.Warn(t.ctx, "verify nats cluster peer for VerifyConnection", slog.Error(err))
+			}
+			return err
+		},
+		GetConfigForClient: t.configForClient,
+	}
+}
+
+// configForClient builds the per-connection config used when accepting a route.
+// It captures the dialing peer's source IP from the underlying connection so
+// VerifyConnection can require the peer leaf's IP SAN to match it. NATS calls
+// this on each inbound handshake, so a fresh config is allocated per accepted
+// connection; that is fine at cluster-route cardinality (a handful of peers).
+func (t *clusterTLS) configForClient(chi *tls.ClientHelloInfo) (*tls.Config, error) {
+	// The accept side must bind the peer leaf to the address it connected from,
+	// so a source IP is required. Fail closed if it cannot be determined rather
+	// than silently skipping the binding in verify.
+	sourceIP, err := clientSourceIP(chi)
+	if err != nil {
+		t.logger.Warn(t.ctx, "reject nats cluster route: no source IP", slog.Error(err))
+		return nil, err
+	}
+
+	cfg := &tls.Config{
+		MinVersion: tls.VersionTLS13,
+		GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
+			leaf, err := t.currentLeaf()
+			if err != nil {
+				t.logger.Warn(t.ctx, "get nats cluster leaf for GetCertificate", slog.Error(err))
+			}
+			return leaf, err
+		},
+		ClientAuth: tls.RequireAnyClientCert,
+		//nolint:gosec // See tlsConfig: verification is performed in VerifyConnection.
+		InsecureSkipVerify: true,
+		VerifyConnection: func(cs tls.ConnectionState) error {
+			err := t.verify(cs, sourceIP)
+			if err != nil {
+				t.logger.Warn(t.ctx, "verify nats cluster peer for VerifyConnection", slog.Error(err))
+			}
+			return err
+		},
+	}
+	return cfg, nil
+}
+
+// clientSourceIP extracts the dialing peer's source IP from the accepted
+// connection. The accept side requires it, so every failure is an error rather
+// than a nil that would bypass source binding in verify.
+func clientSourceIP(chi *tls.ClientHelloInfo) (net.IP, error) {
+	if chi.Conn == nil {
+		return nil, xerrors.New("no underlying connection")
+	}
+	remote := chi.Conn.RemoteAddr()
+	if remote == nil {
+		return nil, xerrors.New("no remote address")
+	}
+	host, _, err := net.SplitHostPort(remote.String())
+	if err != nil {
+		return nil, xerrors.Errorf("split remote address %q: %w", remote.String(), err)
+	}
+	ip := net.ParseIP(host)
+	if ip == nil {
+		return nil, xerrors.Errorf("remote host %q is not an IP", host)
+	}
+	return ip, nil
+}
+
+// currentLeaf returns the cached leaf, re-minting it when it is missing or
+// expired. A leaf carries no independent lifetime: its NotAfter equals its
+// signing CA's (see mintLeaf), so re-minting is driven purely by CA rotation.
+//
+// The whole method holds t.mu so the CA cache, IP, and cached leaf are read as
+// a consistent set: a concurrent setCACache cannot swap the CA out from under
+// the IP we mint with. The lock is held across the SigningKey lookup and the
+// (rare) mint; both are cheap (an in-memory cache hit and, only on a miss, a
+// keygen+sign).
+func (t *clusterTLS) currentLeaf() (*tls.Certificate, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	// Reuse the cached leaf while it is still within its validity window,
+	// before consulting the signing cache. A leaf's NotAfter equals its signing
+	// CA's, and the previous CA stays trusted by peers through the rotation
+	// overlap, so a still-valid cached leaf always chains to a CA peers accept.
+	// A new CA is picked up when the leaf expires (forcing a re-mint) or when
+	// setCACache swaps the cache and clears the leaf.
+	now := t.clock.Now()
+	if t.leaf != nil && now.Before(t.leaf.Leaf.NotAfter) {
+		return t.leaf, nil
+	}
+
+	id, key, err := t.ca.SigningKey(t.ctx)
+	if err != nil {
+		return nil, xerrors.Errorf("get signing CA: %w", err)
+	}
+	ca, ok := key.(*cryptokeys.NATSCA)
+	if !ok {
+		return nil, xerrors.Errorf("unexpected signing key type %T", key)
+	}
+
+	leaf, err := mintLeaf(ca, t.ip, now)
+	if err != nil {
+		return nil, xerrors.Errorf("mint leaf: %w", err)
+	}
+	t.leaf = leaf
+	t.logger.Debug(t.ctx, "minted nats cluster leaf", slog.F("ca_sequence", id))
+	return leaf, nil
+}
+
+// mintLeaf creates an ephemeral leaf certificate signed by the active CA. The
+// signing CA's sequence is stamped into the leaf's Subject SerialNumber so a
+// verifying peer can look up exactly that CA (see verifyConnection), and the
+// replica's relay IP is embedded as an IP SAN so a dialing peer can confirm it
+// reached the host it intended. The leaf is usable as both a TLS server and
+// client certificate because each replica both accepts and dials cluster
+// routes.
+func mintLeaf(ca *cryptokeys.NATSCA, ip net.IP, now time.Time) (*tls.Certificate, error) {
+	if len(ip) == 0 {
+		return nil, xerrors.New("leaf IP SAN is required")
+	}
+
+	leafKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return nil, xerrors.Errorf("generate leaf key: %w", err)
+	}
+
+	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), leafSerialBits))
+	if err != nil {
+		return nil, xerrors.Errorf("generate serial: %w", err)
+	}
+
+	// A leaf is only ever used to authenticate a handshake, so it need only be
+	// valid as long as the CA that signed it. Tie the leaf's NotAfter to the
+	// CA's so a leaf never outlives its CA and carries no independent lifetime.
+	// An expired active CA means a fully-dead rotator; fail loud rather than
+	// mint a dead leaf.
+	if !ca.Cert.NotAfter.After(now) {
+		return nil, xerrors.Errorf("signing CA (seq %d) is expired: NotAfter %s",
+			ca.Sequence, ca.Cert.NotAfter)
+	}
+	notAfter := ca.Cert.NotAfter
+
+	template := &x509.Certificate{
+		SerialNumber: serial,
+		Subject: pkix.Name{
+			CommonName: "coder-nats-cluster-leaf",
+			// SerialNumber carries the sequence of the CA that signed this
+			// leaf, letting a verifier fetch exactly that CA from its cache.
+			SerialNumber: strconv.FormatInt(int64(ca.Sequence), 10),
+		},
+		IPAddresses:           []net.IP{ip},
+		NotBefore:             now.Add(-clockSkewToleranceTLS),
+		NotAfter:              notAfter,
+		KeyUsage:              x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
+		BasicConstraintsValid: true,
+	}
+
+	leafDER, err := x509.CreateCertificate(rand.Reader, template, ca.Cert, &leafKey.PublicKey, ca.Key)
+	if err != nil {
+		return nil, xerrors.Errorf("create leaf certificate: %w", err)
+	}
+	leaf, err := x509.ParseCertificate(leafDER)
+	if err != nil {
+		return nil, xerrors.Errorf("parse leaf certificate: %w", err)
+	}
+
+	return &tls.Certificate{
+		Certificate: [][]byte{leafDER},
+		PrivateKey:  leafKey,
+		Leaf:        leaf,
+	}, nil
+}
+
+// verify verifies a peer's leaf certificate. It reads the signing CA sequence
+// the peer stamped into its leaf, fetches that exact CA from the cache, and
+// confirms the leaf chains to it. Using the stamped sequence is not a trust
+// decision: the leaf must still chain to OUR trusted copy of that CA, and a CA
+// that has been retired is no longer returned by the cache, so leaves from a
+// deleted CA are rejected.
+//
+// It then enforces source binding: when sourceIP is set (the accept side, where
+// the dialing peer's connection address is available), the leaf must carry that
+// source IP as an IP SAN, binding the certificate to the network origin. Go's
+// default hostname verification, which InsecureSkipVerify disables, cannot do
+// this because Go does not populate cs.ServerName for IP-based routes. On the
+// dial side sourceIP is nil (Go does not expose the connection in the
+// client-certificate callbacks), so only the chain is verified there.
+func (t *clusterTLS) verify(cs tls.ConnectionState, sourceIP net.IP) error {
+	if len(cs.PeerCertificates) == 0 {
+		return xerrors.New("no peer certificate presented")
+	}
+	leaf := cs.PeerCertificates[0]
+
+	seq := leaf.Subject.SerialNumber
+	if seq == "" {
+		return xerrors.New("peer leaf missing signing CA sequence")
+	}
+
+	key, err := t.caCache().VerifyingKey(t.ctx, seq)
+	if err != nil {
+		return xerrors.Errorf("get CA for sequence %q: %w", seq, err)
+	}
+	ca, ok := key.(*cryptokeys.NATSCA)
+	if !ok {
+		return xerrors.Errorf("unexpected verifying key type %T", key)
+	}
+
+	// Leaves carry both ServerAuth and ClientAuth, since each replica is both a
+	// route server and client. Requiring those specific usages rejects a leaf
+	// with some unexpected EKU rather than accepting any usage.
+	if _, err := leaf.Verify(x509.VerifyOptions{
+		Roots:       t.verifyPool(seq, ca.Cert),
+		KeyUsages:   []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
+		CurrentTime: t.clock.Now(),
+	}); err != nil {
+		return xerrors.Errorf("verify peer leaf against CA sequence %q: %w", seq, err)
+	}
+
+	// On the accept side, confirm the leaf's IP SAN matches the address the
+	// peer actually connected from.
+	if len(sourceIP) != 0 && !slices.ContainsFunc(leaf.IPAddresses, sourceIP.Equal) {
+		return xerrors.Errorf("peer leaf IP SANs %v do not match source IP %s", leaf.IPAddresses, sourceIP)
+	}
+	return nil
+}
+
+// verifyPool returns the root pool used to verify a peer leaf minted under the
+// given CA sequence, building it once and caching it for reuse. It is called
+// from verify on every route handshake; cluster routes are long-lived, so a
+// handshake is a rare event, and the common case here is a cache hit (a single
+// map lookup).
+//
+// A miss occurs only the first time a sequence is seen (startup, and once per
+// CA rotation), which is the only moment the map can grow, so pruning of expired
+// entries is attached to the miss path rather than run on every handshake. An
+// entry is dropped once the clock passes the CA cert's NotAfter: no valid leaf
+// can chain to an expired CA, and the CA outlives every leaf it signed, so this
+// is always safe and bounds the map across rotations.
+func (t *clusterTLS) verifyPool(seq string, cert *x509.Certificate) *x509.CertPool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	if cp, ok := t.verifyPools[seq]; ok {
+		return cp.pool
+	}
+
+	now := t.clock.Now()
+	for s, cp := range t.verifyPools {
+		if now.After(cp.notAfter) {
+			delete(t.verifyPools, s)
+		}
+	}
+
+	pool := x509.NewCertPool()
+	pool.AddCert(cert)
+	if t.verifyPools == nil {
+		t.verifyPools = map[string]cachedVerifyPool{}
+	}
+	t.verifyPools[seq] = cachedVerifyPool{pool: pool, notAfter: cert.NotAfter}
+	return pool
+}

coderd/x/nats/tls_internal_test.go

@@ -0,0 +1,530 @@
+package nats
+
+import (
+	"context"
+	"crypto"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"math/big"
+	"net"
+	"strconv"
+	"testing"
+	"time"
+
+	natsserver "github.com/nats-io/nats-server/v2/server"
+	"github.com/stretchr/testify/require"
+
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	"github.com/coder/coder/v2/coderd/cryptokeys"
+	"github.com/coder/coder/v2/coderd/database"
+	"github.com/coder/coder/v2/coderd/database/dbgen"
+	"github.com/coder/coder/v2/coderd/database/dbtestutil"
+	"github.com/coder/coder/v2/codersdk"
+	"github.com/coder/coder/v2/testutil"
+	"github.com/coder/quartz"
+)
+
+// fakeCACache is an in-memory cryptokeys.SigningKeycache for tests. active is returned
+// by SigningKey (the CA this replica mints leaves under); byID is consulted by
+// VerifyingKey (the CAs this replica trusts when verifying peers).
+type fakeCACache struct {
+	active *cryptokeys.NATSCA
+	byID   map[string]*cryptokeys.NATSCA
+}
+
+func (f *fakeCACache) SigningKey(context.Context) (string, interface{}, error) {
+	if f.active == nil {
+		return "", nil, cryptokeys.ErrKeyNotFound
+	}
+	return strconv.FormatInt(int64(f.active.Sequence), 10), f.active, nil
+}
+
+func (f *fakeCACache) VerifyingKey(_ context.Context, id string) (interface{}, error) {
+	ca, ok := f.byID[id]
+	if !ok {
+		return nil, cryptokeys.ErrKeyNotFound
+	}
+	return ca, nil
+}
+
+func (*fakeCACache) Close() error { return nil }
+
+func generateTestCA(t *testing.T, sequence int32) *cryptokeys.NATSCA {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	require.NoError(t, err)
+
+	now := time.Now()
+	template := &x509.Certificate{
+		SerialNumber:          big.NewInt(int64(sequence)),
+		Subject:               pkix.Name{CommonName: "coder-nats-ca-test"},
+		NotBefore:             now.Add(-time.Hour),
+		NotAfter:              now.Add(72 * time.Hour),
+		KeyUsage:              x509.KeyUsageCertSign,
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+	}
+	der, err := x509.CreateCertificate(rand.Reader, template, template, key.Public(), key)
+	require.NoError(t, err)
+	cert, err := x509.ParseCertificate(der)
+	require.NoError(t, err)
+
+	return &cryptokeys.NATSCA{Sequence: sequence, Cert: cert, Key: crypto.Signer(key)}
+}
+
+// newTLSPubsub builds a clustered pubsub whose route listener requires mTLS,
+// using the supplied CA cache. ip is this node's cluster host: the route
+// listener bind host and the leaf IP SAN. Peers dial each other on that host
+// (clusterRouteAddress), so ip must be 127.0.0.1 for routes to form.
+func newTLSPubsub(t *testing.T, ca cryptokeys.SigningKeycache, ip net.IP) *Pubsub {
+	t.Helper()
+	logger := slogtest.Make(t, nil)
+	ctx := testutil.Context(t, testutil.WaitLong)
+	ps, err := New(ctx, logger, Options{
+		ClusterHost:       ip.String(),
+		ClusterPort:       natsserver.RANDOM_PORT,
+		disableCluster:    false,
+		ClusterCA:         ca,
+		clusterTLSTimeout: testClusterTLSTimeout,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = ps.Close() })
+	return ps
+}
+
+// setLeafSAN overrides a node's leaf IP SAN after construction, for tests that
+// need the minted SAN to differ from the loopback address the node binds and
+// connects on (which are otherwise both the node's ClusterHost). Clearing the
+// cached leaf forces the next handshake to re-mint under the new SAN.
+func setLeafSAN(ps *Pubsub, ip net.IP) {
+	ps.clusterTLS.mu.Lock()
+	defer ps.clusterTLS.mu.Unlock()
+	ps.clusterTLS.ip = ip
+	ps.clusterTLS.leaf = nil
+}
+
+func numRoutes(t *testing.T, ps *Pubsub) int {
+	t.Helper()
+	routes, err := ps.Server.Routez(&natsserver.RoutezOptions{})
+	require.NoError(t, err)
+	return routes.NumRoutes
+}
+
+// TestPubsub_ClusterTLS validates that the embedded NATS server honors the
+// tls.Config callbacks on cluster routes: leaves minted from the CA cache form
+// a verified mesh, peers under unrelated CAs are rejected, and peers on either
+// side of a CA rotation still verify each other.
+func TestPubsub_ClusterTLS(t *testing.T) {
+	t.Parallel()
+
+	t.Run("Mesh", func(t *testing.T) {
+		t.Parallel()
+
+		ca := generateTestCA(t, 1)
+		cache := func() *fakeCACache {
+			return &fakeCACache{active: ca, byID: map[string]*cryptokeys.NATSCA{"1": ca}}
+		}
+		a := newTLSPubsub(t, cache(), net.IPv4(127, 0, 0, 1))
+		b := newTLSPubsub(t, cache(), net.IPv4(127, 0, 0, 1))
+		c := newTLSPubsub(t, cache(), net.IPv4(127, 0, 0, 1))
+
+		addrA := clusterRouteAddress(t, a)
+		addrB := clusterRouteAddress(t, b)
+		addrC := clusterRouteAddress(t, c)
+		// Full symmetric mesh: every node must know a peer to accept a route
+		// from it (accept-side membership), so each is given the other two.
+		// Drive peers through fetchers, as production does: a fetcher re-applies
+		// the same peers on every refresh, so the startup refresh (which runs
+		// with the boot-time noop fetcher) cannot race a manual call and wipe
+		// the route/known-peer set.
+		a.SetPeerFetcher(&testPeerFetcher{addresses: []string{addrB, addrC}})
+		b.SetPeerFetcher(&testPeerFetcher{addresses: []string{addrA, addrC}})
+		c.SetPeerFetcher(&testPeerFetcher{addresses: []string{addrA, addrB}})
+
+		event := "tls-mesh"
+		got := make(chan []byte, 8)
+		cancel, err := c.Subscribe(event, func(_ context.Context, msg []byte) { got <- msg })
+		require.NoError(t, err)
+		defer cancel()
+
+		// Retry publishes until the route subscription has propagated.
+		require.Eventually(t, func() bool {
+			if err := b.Publish(event, []byte("hello")); err != nil {
+				return false
+			}
+			if err := b.Flush(); err != nil {
+				return false
+			}
+			select {
+			case msg := <-got:
+				return string(msg) == "hello"
+			case <-time.After(testutil.IntervalMedium):
+				return false
+			}
+		}, testutil.WaitLong, testutil.IntervalFast)
+	})
+
+	t.Run("WrongCARejected", func(t *testing.T) {
+		t.Parallel()
+
+		caX := generateTestCA(t, 1)
+		caY := generateTestCA(t, 1)
+		a := newTLSPubsub(t, &fakeCACache{active: caX, byID: map[string]*cryptokeys.NATSCA{"1": caX}}, net.IPv4(127, 0, 0, 1))
+		b := newTLSPubsub(t, &fakeCACache{active: caY, byID: map[string]*cryptokeys.NATSCA{"1": caY}}, net.IPv4(127, 0, 0, 1))
+
+		require.NoError(t, a.setPeerAddresses([]string{clusterRouteAddress(t, b)}))
+
+		// Each side only trusts its own CA, so the route handshake never
+		// completes and no route is established.
+		require.Never(t, func() bool {
+			return numRoutes(t, a) > 0 || numRoutes(t, b) > 0
+		}, testutil.WaitShort, testutil.IntervalFast)
+	})
+
+	t.Run("RotationOverlap", func(t *testing.T) {
+		t.Parallel()
+
+		ca1 := generateTestCA(t, 1)
+		ca2 := generateTestCA(t, 2)
+		bundle := map[string]*cryptokeys.NATSCA{"1": ca1, "2": ca2}
+		// a still mints under the old CA; b has already rotated to the new CA.
+		// Both trust both CAs, so the mesh forms across the rotation overlap.
+		a := newTLSPubsub(t, &fakeCACache{active: ca1, byID: bundle}, net.IPv4(127, 0, 0, 1))
+		b := newTLSPubsub(t, &fakeCACache{active: ca2, byID: bundle}, net.IPv4(127, 0, 0, 1))
+
+		// Symmetric peers so each side accepts a route from the other, driven
+		// through fetchers (see Mesh) so the startup noop refresh cannot race a
+		// manual call and wipe the route/known-peer set.
+		a.SetPeerFetcher(&testPeerFetcher{addresses: []string{clusterRouteAddress(t, b)}})
+		b.SetPeerFetcher(&testPeerFetcher{addresses: []string{clusterRouteAddress(t, a)}})
+
+		require.Eventually(t, func() bool {
+			return numRoutes(t, a) > 0 && numRoutes(t, b) > 0
+		}, testutil.WaitLong, testutil.IntervalFast)
+	})
+
+	t.Run("SANMismatch", func(t *testing.T) {
+		t.Parallel()
+
+		ca := generateTestCA(t, 1)
+		cache := func() *fakeCACache {
+			return &fakeCACache{active: ca, byID: map[string]*cryptokeys.NATSCA{"1": ca}}
+		}
+		// Both nodes bind and connect on loopback and know each other as peers,
+		// so the CA and source-membership checks pass. But both mint their leaf
+		// with a SAN that does not match the loopback address they connect from,
+		// so every handshake is rejected on the SAN binding alone and no route
+		// forms. This isolates the SAN check: a valid CA-signed leaf presented
+		// from a known replica is still rejected when the cert is not bound to
+		// the address it connects from (e.g. a stolen or mis-minted leaf).
+		a := newTLSPubsub(t, cache(), net.IPv4(127, 0, 0, 1))
+		b := newTLSPubsub(t, cache(), net.IPv4(127, 0, 0, 1))
+		setLeafSAN(a, net.IPv4(10, 99, 99, 99))
+		setLeafSAN(b, net.IPv4(10, 99, 99, 99))
+
+		require.NoError(t, a.setPeerAddresses([]string{clusterRouteAddress(t, b)}))
+		require.NoError(t, b.setPeerAddresses([]string{clusterRouteAddress(t, a)}))
+
+		require.Never(t, func() bool {
+			return numRoutes(t, a) > 0 || numRoutes(t, b) > 0
+		}, testutil.WaitShort, testutil.IntervalFast)
+	})
+
+	t.Run("MixedTLSAndPlaintext", func(t *testing.T) {
+		t.Parallel()
+
+		ca := generateTestCA(t, 1)
+		// a requires mTLS on its route listener; b is a plaintext node
+		// (newTestPubsub leaves ClusterCA nil). Routes must not form in either
+		// direction: a rollout has to enable TLS on every replica at once.
+		a := newTLSPubsub(t, &fakeCACache{active: ca, byID: map[string]*cryptokeys.NATSCA{"1": ca}}, net.IPv4(127, 0, 0, 1))
+		b := newTestPubsub(t, clusterTestOptions(t))
+
+		require.NoError(t, a.setPeerAddresses([]string{clusterRouteAddress(t, b)}))
+		require.NoError(t, b.setPeerAddresses([]string{clusterRouteAddress(t, a)}))
+
+		require.Never(t, func() bool {
+			return numRoutes(t, a) > 0 || numRoutes(t, b) > 0
+		}, testutil.WaitShort, testutil.IntervalFast)
+	})
+}
+
+// TestPubsub_ClusterTLS_CacheSwap covers the Part C optional-mTLS model: a node
+// that boots with the noop CA cache forms no route, and swapping in a real cache
+// via SetCACache lets routes form over mTLS with no server restart.
+func TestPubsub_ClusterTLS_CacheSwap(t *testing.T) {
+	t.Parallel()
+
+	t.Run("NoopFormsNoRoute", func(t *testing.T) {
+		t.Parallel()
+
+		ca := generateTestCA(t, 1)
+		// a boots with the noop cache (production default); b has a real cache.
+		// a cannot mint a leaf, so its route handshakes fail and no route forms.
+		a := newTLSPubsub(t, cryptokeys.NoopSigningKeycache{}, net.IPv4(127, 0, 0, 1))
+		b := newTLSPubsub(t, &fakeCACache{active: ca, byID: map[string]*cryptokeys.NATSCA{"1": ca}}, net.IPv4(127, 0, 0, 1))
+
+		require.NoError(t, a.setPeerAddresses([]string{clusterRouteAddress(t, b)}))
+		require.NoError(t, b.setPeerAddresses([]string{clusterRouteAddress(t, a)}))
+
+		require.Never(t, func() bool {
+			return numRoutes(t, a) > 0 || numRoutes(t, b) > 0
+		}, testutil.WaitShort, testutil.IntervalFast)
+	})
+
+	t.Run("SwapToRealFormsRoute", func(t *testing.T) {
+		t.Parallel()
+
+		ca := generateTestCA(t, 1)
+		realCache := func() *fakeCACache {
+			return &fakeCACache{active: ca, byID: map[string]*cryptokeys.NATSCA{"1": ca}}
+		}
+		// Both boot with the noop cache, then both get the real cache swapped in
+		// (mirroring the enterprise HA enable path) without a server restart.
+		a := newTLSPubsub(t, cryptokeys.NoopSigningKeycache{}, net.IPv4(127, 0, 0, 1))
+		b := newTLSPubsub(t, cryptokeys.NoopSigningKeycache{}, net.IPv4(127, 0, 0, 1))
+
+		// Drive peers through fetchers, as production does, rather than calling
+		// setPeerAddresses directly: SetCACache and SetPeerFetcher both trigger
+		// a peer refresh that reads the current fetcher, so routes converge on
+		// the fetcher's addresses without racing a manual call.
+		a.SetCACache(realCache())
+		b.SetCACache(realCache())
+
+		a.SetPeerFetcher(&testPeerFetcher{addresses: []string{clusterRouteAddress(t, b)}})
+		b.SetPeerFetcher(&testPeerFetcher{addresses: []string{clusterRouteAddress(t, a)}})
+
+		require.Eventually(t, func() bool {
+			return numRoutes(t, a) > 0 && numRoutes(t, b) > 0
+		}, testutil.WaitLong, testutil.IntervalFast)
+	})
+}
+
+// TestClusterTLS_configForClient_RequiresSourceIP asserts the accept side fails
+// closed when it cannot determine the peer's source IP, rather than skipping the
+// source-binding check.
+func TestClusterTLS_configForClient_RequiresSourceIP(t *testing.T) {
+	t.Parallel()
+
+	ctx := testutil.Context(t, testutil.WaitShort)
+	ca := generateTestCA(t, 1)
+	cache := &fakeCACache{active: ca, byID: map[string]*cryptokeys.NATSCA{"1": ca}}
+	ct := newClusterTLS(ctx, slogtest.Make(t, nil), nil, cache, net.IPv4(127, 0, 0, 1))
+
+	// No underlying connection: the source IP cannot be determined, so the
+	// accept-side config is refused and the handshake aborts.
+	_, err := ct.configForClient(&tls.ClientHelloInfo{})
+	require.Error(t, err)
+}
+
+// TestClusterTLS_verify unit-tests the verifier directly, isolating chain
+// verification and source-IP binding that the mesh tests exercise only
+// indirectly.
+func TestClusterTLS_verify(t *testing.T) {
+	t.Parallel()
+
+	ctx := testutil.Context(t, testutil.WaitShort)
+	ca := generateTestCA(t, 1)
+	cache := &fakeCACache{active: ca, byID: map[string]*cryptokeys.NATSCA{"1": ca}}
+
+	leafIP := net.IPv4(10, 0, 0, 5)
+	ct := newClusterTLS(ctx, slogtest.Make(t, nil), nil, cache, net.IPv4(10, 0, 0, 1))
+
+	// A leaf bound to leafIP, signed by the trusted CA.
+	leafCert, err := mintLeaf(ca, leafIP, time.Now())
+	require.NoError(t, err)
+	leaf, err := x509.ParseCertificate(leafCert.Certificate[0])
+	require.NoError(t, err)
+	cs := tls.ConnectionState{PeerCertificates: []*x509.Certificate{leaf}}
+
+	t.Run("DialSideChainOnly", func(t *testing.T) {
+		t.Parallel()
+		// No source IP (dial side): only the chain is verified.
+		require.NoError(t, ct.verify(cs, nil))
+	})
+
+	t.Run("AcceptSideSourceMatches", func(t *testing.T) {
+		t.Parallel()
+		// Source IP equals the leaf SAN: accepted.
+		require.NoError(t, ct.verify(cs, leafIP))
+	})
+
+	t.Run("AcceptSideSourceMismatch", func(t *testing.T) {
+		t.Parallel()
+		// The leaf is bound to leafIP, so a connection from a different source
+		// is rejected even though the chain is valid.
+		err := ct.verify(cs, net.IPv4(10, 0, 0, 1))
+		require.ErrorContains(t, err, "do not match source IP")
+	})
+
+	t.Run("UntrustedCARejected", func(t *testing.T) {
+		t.Parallel()
+		otherCA := generateTestCA(t, 9)
+		strangerCert, err := mintLeaf(otherCA, leafIP, time.Now())
+		require.NoError(t, err)
+		stranger, err := x509.ParseCertificate(strangerCert.Certificate[0])
+		require.NoError(t, err)
+		// The stamped sequence (9) is not in the cache, so the CA lookup fails.
+		err = ct.verify(tls.ConnectionState{PeerCertificates: []*x509.Certificate{stranger}}, nil)
+		require.Error(t, err)
+	})
+}
+
+// TestClusterTLS_verifyPool asserts the verify-pool cache reuses a pool for a
+// given CA sequence and prunes entries whose CA cert has expired, so the map
+// does not grow unbounded across rotations.
+func TestClusterTLS_verifyPool(t *testing.T) {
+	t.Parallel()
+
+	ctx := testutil.Context(t, testutil.WaitShort)
+	clock := quartz.NewMock(t)
+	clock.Set(time.Now())
+
+	ca1 := generateTestCA(t, 1)
+	ca2 := generateTestCA(t, 2)
+	cache := &fakeCACache{byID: map[string]*cryptokeys.NATSCA{"1": ca1, "2": ca2}}
+	ct := newClusterTLS(ctx, slogtest.Make(t, nil), clock, cache, net.IPv4(10, 0, 0, 1))
+
+	// First build for seq 1 caches the pool; a second call returns the same one.
+	p1 := ct.verifyPool("1", ca1.Cert)
+	require.Same(t, p1, ct.verifyPool("1", ca1.Cert))
+	require.Len(t, ct.verifyPools, 1)
+
+	// Advance past ca1's NotAfter. Building a pool for a new sequence prunes the
+	// now-expired seq 1 entry, leaving only seq 2.
+	clock.Set(ca1.Cert.NotAfter.Add(time.Minute))
+	ct.verifyPool("2", ca2.Cert)
+	require.Len(t, ct.verifyPools, 1)
+	_, ok := ct.verifyPools["1"]
+	require.False(t, ok, "expired seq 1 pool should be pruned")
+	_, ok = ct.verifyPools["2"]
+	require.True(t, ok)
+}
+
+// generateTestCAWithValidity is like generateTestCA but lets a test control the
+// CA certificate's NotAfter, so leaf-clamp behavior near CA expiry is testable.
+func generateTestCAWithValidity(t *testing.T, sequence int32, notAfter time.Time) *cryptokeys.NATSCA {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	require.NoError(t, err)
+
+	template := &x509.Certificate{
+		SerialNumber:          big.NewInt(int64(sequence)),
+		Subject:               pkix.Name{CommonName: "coder-nats-ca-test"},
+		NotBefore:             notAfter.Add(-90 * 24 * time.Hour),
+		NotAfter:              notAfter,
+		KeyUsage:              x509.KeyUsageCertSign,
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+	}
+	der, err := x509.CreateCertificate(rand.Reader, template, template, key.Public(), key)
+	require.NoError(t, err)
+	cert, err := x509.ParseCertificate(der)
+	require.NoError(t, err)
+
+	return &cryptokeys.NATSCA{Sequence: sequence, Cert: cert, Key: crypto.Signer(key)}
+}
+
+// TestMintLeaf asserts a leaf's NotAfter is exactly its signing CA's NotAfter
+// (a leaf carries no independent lifetime), and that minting against an
+// already-expired CA fails rather than emitting a dead leaf.
+func TestMintLeaf(t *testing.T) {
+	t.Parallel()
+
+	ip := net.IPv4(127, 0, 0, 1)
+
+	t.Run("MatchesCAValidity", func(t *testing.T) {
+		t.Parallel()
+		now := time.Now()
+		// generateTestCA mints a 72h CA; the leaf's NotAfter tracks it exactly.
+		ca := generateTestCA(t, 1)
+		leaf, err := mintLeaf(ca, ip, now)
+		require.NoError(t, err)
+		require.WithinDuration(t, ca.Cert.NotAfter, leaf.Leaf.NotAfter, time.Second)
+		require.WithinDuration(t, now.Add(-clockSkewToleranceTLS), leaf.Leaf.NotBefore, time.Second)
+	})
+
+	t.Run("ErrorsWhenCAExpired", func(t *testing.T) {
+		t.Parallel()
+		now := time.Now()
+		// CA's NotAfter is already in the past: no usable leaf can be minted.
+		ca := generateTestCAWithValidity(t, 1, now.Add(-time.Minute))
+		_, err := mintLeaf(ca, ip, now)
+		require.Error(t, err)
+		require.ErrorContains(t, err, "expired")
+	})
+}
+
+// TestPubsub_ClusterTLS_RealCA stands up a three-node TLS mesh whose trust root
+// is a real CA served by the cryptokeys signing cache against a real DB, then
+// verifies a cross-route publish/subscribe round-trip. This exercises the
+// integration seam between the cryptokeys CA cache and the x/nats cluster TLS
+// callbacks, including the real PEM/x509 round-trip that the synthetic
+// generateTestCA helper does not cover. Nodes form a direct full mesh to avoid
+// depending on multi-hop route gossip.
+func TestPubsub_ClusterTLS_RealCA(t *testing.T) {
+	t.Parallel()
+
+	db, _ := dbtestutil.NewDB(t)
+	ctx := testutil.Context(t, testutil.WaitLong)
+
+	// Seed an active nats_ca crypto key, mirroring the row the key rotator
+	// mints in production. The signing cache decodes the PEM secret into a
+	// *cryptokeys.NATSCA the same way production reads it.
+	dbgen.CryptoKey(t, db, database.CryptoKey{
+		Feature:  database.CryptoKeyFeatureNATSCA,
+		Sequence: 1,
+		StartsAt: time.Now().UTC().Add(-time.Hour),
+	})
+
+	newNode := func() *Pubsub {
+		// A real signing cache per node, as each replica builds in coderd.New.
+		cache, err := cryptokeys.NewSigningCache(ctx, slogtest.Make(t, nil), &cryptokeys.DBFetcher{DB: db}, codersdk.CryptoKeyFeatureNATSCA)
+		require.NoError(t, err)
+		t.Cleanup(func() { _ = cache.Close() })
+		// Nodes mesh on loopback, so the leaf IP SAN must be 127.0.0.1.
+		return newTLSPubsub(t, cache, net.IPv4(127, 0, 0, 1))
+	}
+
+	a := newNode()
+	b := newNode()
+	c := newNode()
+
+	addrA := clusterRouteAddress(t, a)
+	addrB := clusterRouteAddress(t, b)
+	addrC := clusterRouteAddress(t, c)
+	// Drive peers through fetchers, as production does, so the startup noop
+	// refresh cannot race a manual call and wipe the route/known-peer set.
+	a.SetPeerFetcher(&testPeerFetcher{addresses: []string{addrB, addrC}})
+	b.SetPeerFetcher(&testPeerFetcher{addresses: []string{addrA, addrC}})
+	c.SetPeerFetcher(&testPeerFetcher{addresses: []string{addrA, addrB}})
+
+	received := make(chan string, 4)
+	cancelSub, err := c.Subscribe("tls-realca", func(_ context.Context, msg []byte) {
+		select {
+		case received <- string(msg):
+		default:
+		}
+	})
+	require.NoError(t, err)
+	defer cancelSub()
+
+	// Routes and subscription interest propagate asynchronously after the
+	// servers report ready, so retry rather than gate on a one-shot check.
+	require.Eventually(t, func() bool {
+		if err := b.Publish("tls-realca", []byte("hello")); err != nil {
+			return false
+		}
+		select {
+		case msg := <-received:
+			require.Equal(t, "hello", msg)
+			return true
+		case <-time.After(testutil.IntervalMedium):
+			return false
+		}
+	}, testutil.WaitLong, testutil.IntervalFast)
+}