feat: fold the TLS-first handshake into the NATS pub/sub tlsEnabled flag

Author: igor-kupczynski

frontend/docs/content/docs/self-hosting/configuration-options.mdx

@@ -318,9 +318,8 @@ Variables marked with ⚠️ are conditionally required when specific features a
 | `SERVER_MSGQUEUE_PUBSUB_NATS_URL`                   | Pub/sub NATS seed URL(s), comma-separated for a cluster                                                    |                  |
 | `SERVER_MSGQUEUE_PUBSUB_NATS_USERNAME`              | Pub/sub NATS username (sent as a connect option, so reconnects also authenticate)                          |                  |
 | `SERVER_MSGQUEUE_PUBSUB_NATS_PASSWORD`              | Pub/sub NATS password (see username)                                                                       |                  |
-| `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED`           | Pub/sub NATS: when `true`, require verified TLS regardless of the URL scheme                               | `false`          |
+| `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED`           | Pub/sub NATS: verified TLS with a TLS-first handshake (server must set `handshake_first`)                  | `false`          |
 | `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ROOT_CA_FILE`      | Pub/sub NATS PEM CA bundle for a private CA; requires `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED=true`       |                  |
-| `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_HANDSHAKE_FIRST`   | Pub/sub NATS: TLS handshake before the server's `INFO` (server must set `handshake_first`); requires TLS   | `false`          |
 | `SERVER_MSGQUEUE_PUBSUB_NATS_SUBJECT_PREFIX`        | Pub/sub NATS subject prefix, joined to topic names with `.`                                                | `hatchet.pubsub` |
 | `SERVER_SINGLE_QUEUE_LIMIT`                         | Single queue limit                                                                                         | `100`            |
 
@@ -340,17 +339,16 @@ password separately. Credentials embedded in the URL are not reapplied when the
 client reconnects to a cluster peer it learned about through gossip.
 
 Set `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED=true` to require verified TLS
-regardless of the URL scheme; without a CA file the server certificate is
-verified against the system roots. If the server's certificate is signed by a
-private CA, also point `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ROOT_CA_FILE` at a PEM
-CA bundle, which also applies when the client reconnects to rediscovered
-cluster peers. The CA file only takes effect with TLS enabled, so Hatchet
-refuses to start if it is set while `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED`
-is false. A `tls://` URL enables TLS with the system roots via the scheme
-alone. If the server sets `handshake_first` in its tls block (the TLS
-handshake happens before the server's `INFO` message), also set
-`SERVER_MSGQUEUE_PUBSUB_NATS_TLS_HANDSHAKE_FIRST=true`; both sides must agree
-on the mode or the connect fails.
+with a TLS-first handshake: the handshake happens before the server's `INFO`
+message, so the server must set `handshake_first` in its tls block, and a
+client with this flag fails to connect to an INFO-first TLS or plaintext
+server. Without a CA file the server certificate is verified against the
+system roots. If the server's certificate is signed by a private CA, also
+point `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ROOT_CA_FILE` at a PEM CA bundle, which
+also applies when the client reconnects to rediscovered cluster peers. The CA
+file only takes effect with TLS enabled, so Hatchet refuses to start if it is
+set while `SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED` is false. A `tls://` URL
+does standard INFO-first TLS with the system roots via the scheme alone.
 
 Unlike the other backends, which isolate installations at the connection level
 (a RabbitMQ vhost, a Postgres database), NATS isolates by subject. Set

internal/msgqueue/nats/pubsub.go

@@ -29,14 +29,13 @@ type PubSub struct {
 type PubSubOpt func(*PubSubOpts)
 
 type PubSubOpts struct {
-	l                 *zerolog.Logger
-	url               string
-	username          string
-	password          string
-	tlsEnabled        bool
-	tlsRootCAFile     string
-	tlsHandshakeFirst bool
-	subjectPrefix     string
+	l             *zerolog.Logger
+	url           string
+	username      string
+	password      string
+	tlsEnabled    bool
+	tlsRootCAFile string
+	subjectPrefix string
 }
 
 func defaultPubSubOpts() *PubSubOpts {
@@ -70,9 +69,12 @@ func WithPubSubPassword(password string) PubSubOpt {
 	}
 }
 
-// WithPubSubTLSEnabled requires verified TLS regardless of the URL scheme.
-// Without a CA file the server certificate is verified against the system
-// roots. False leaves TLS to the URL scheme (tls:// uses system roots).
+// WithPubSubTLSEnabled requires verified TLS with a TLS-first handshake: the
+// handshake happens before the server's INFO message, so the server must set
+// handshake_first in its tls block. A client with this flag fails to connect
+// to an INFO-first TLS or plaintext server. Without a CA file the server
+// certificate is verified against the system roots. False leaves TLS to the
+// URL scheme (tls:// does standard INFO-first TLS with system roots).
 func WithPubSubTLSEnabled(enabled bool) PubSubOpt {
 	return func(opts *PubSubOpts) {
 		opts.tlsEnabled = enabled
@@ -89,18 +91,6 @@ func WithPubSubTLSRootCAFile(path string) PubSubOpt {
 	}
 }
 
-// WithPubSubTLSHandshakeFirst performs the TLS handshake before the server's
-// INFO message, for servers with handshake_first in their tls block. Against
-// a server that sends INFO first, the client's TLS handshake reads the
-// plaintext INFO as a malformed handshake reply and fails the connect within
-// the connect timeout. Requires WithPubSubTLSEnabled(true); NewPubSub fails
-// fast otherwise.
-func WithPubSubTLSHandshakeFirst(handshakeFirst bool) PubSubOpt {
-	return func(opts *PubSubOpts) {
-		opts.tlsHandshakeFirst = handshakeFirst
-	}
-}
-
 // WithPubSubSubjectPrefix sets the NATS subject prefix (default
 // "hatchet.pubsub"). Empty falls back to the default. No trimming or
 // validation: a bad prefix fails loudly via nats ErrBadSubject at startup.
@@ -133,10 +123,6 @@ func NewPubSub(fs ...PubSubOpt) (func() error, *PubSub, error) {
 		return nil, nil, fmt.Errorf("nats pubsub tlsRootCAFile is set but tlsEnabled is false; a private CA bundle only takes effect with tlsEnabled: true (SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED)")
 	}
 
-	if opts.tlsHandshakeFirst && !opts.tlsEnabled {
-		return nil, nil, fmt.Errorf("nats pubsub tlsHandshakeFirst is set but tlsEnabled is false; a TLS-first handshake only takes effect with tlsEnabled: true (SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED)")
-	}
-
 	l := opts.l
 
 	connectOpts := []natsgo.Option{
@@ -187,6 +173,13 @@ func NewPubSub(fs ...PubSubOpt) (func() error, *PubSub, error) {
 		// the server is verified against the system roots.
 		connectOpts = append(connectOpts, natsgo.Secure(&tls.Config{MinVersion: tls.VersionTLS12}))
 
+		// tlsEnabled always means a TLS-first handshake (before the server's
+		// INFO message); the server must set handshake_first in its tls
+		// block. Against an INFO-first server the client's TLS handshake
+		// reads the plaintext INFO as a malformed handshake reply and fails
+		// the connect within the connect timeout.
+		connectOpts = append(connectOpts, natsgo.TLSHandshakeFirst())
+
 		if opts.tlsRootCAFile != "" {
 			// RootCAs keeps the TLS config set by Secure above and attaches
 			// the CA pool via a callback, which also applies to rediscovered
@@ -195,10 +188,6 @@ func NewPubSub(fs ...PubSubOpt) (func() error, *PubSub, error) {
 			// needed here.
 			connectOpts = append(connectOpts, natsgo.RootCAs(opts.tlsRootCAFile))
 		}
-
-		if opts.tlsHandshakeFirst {
-			connectOpts = append(connectOpts, natsgo.TLSHandshakeFirst())
-		}
 	}
 
 	nc, err := natsgo.Connect(opts.url, connectOpts...)

internal/msgqueue/nats/pubsub_tls_test.go

@@ -26,19 +26,6 @@ func TestNewPubSubTLSRootCAFileRequiresTLSEnabled(t *testing.T) {
 	assert.Nil(t, ps)
 }
 
-func TestNewPubSubTLSHandshakeFirstRequiresTLSEnabled(t *testing.T) {
-	cleanup, ps, err := NewPubSub(
-		WithPubSubURL("nats://127.0.0.1:4222"),
-		WithPubSubTLSHandshakeFirst(true),
-	)
-
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "tlsHandshakeFirst")
-	assert.Contains(t, err.Error(), "tlsEnabled")
-	assert.Nil(t, cleanup)
-	assert.Nil(t, ps)
-}
-
 func TestNewPubSubTLSRootCAFileMissing(t *testing.T) {
 	cleanup, ps, err := NewPubSub(
 		WithPubSubURL("nats://127.0.0.1:4222"),

pkg/config/loader/loader.go

@@ -1051,7 +1051,6 @@ func createPubSubV1(dc *database.Layer, cf *server.ServerConfigFile, l *zerolog.
 			natsmq.WithPubSubPassword(cf.MessageQueue.PubSub.NATS.Password),
 			natsmq.WithPubSubTLSEnabled(cf.MessageQueue.PubSub.NATS.TLSEnabled),
 			natsmq.WithPubSubTLSRootCAFile(cf.MessageQueue.PubSub.NATS.TLSRootCAFile),
-			natsmq.WithPubSubTLSHandshakeFirst(cf.MessageQueue.PubSub.NATS.TLSHandshakeFirst),
 			natsmq.WithPubSubSubjectPrefix(cf.MessageQueue.PubSub.NATS.SubjectPrefix),
 			natsmq.WithPubSubLogger(l),
 		)

pkg/config/loader/loader_pubsub_test.go

@@ -21,9 +21,8 @@ func TestPubSubSettingsInheritance(t *testing.T) {
 		wantNatsURL               string
 		wantNatsUsername          string
 		wantNatsPassword          string
-		wantNatsTLSEnabled        bool
-		wantNatsTLSRootCAFile     string
-		wantNatsTLSHandshakeFirst bool
+		wantNatsTLSEnabled    bool
+		wantNatsTLSRootCAFile string
 		wantNatsSubjectPrefix     string
 		wantMaxPub                int32
 		wantMaxSub                int32
@@ -120,24 +119,22 @@ func TestPubSubSettingsInheritance(t *testing.T) {
 			wantMaxSub:       20,
 		},
 		{
-			name: "nats pubsub with tls enabled, root ca file, and handshake first",
+			name: "nats pubsub with tls enabled and root ca file",
 			env: map[string]string{
-				"SERVER_MSGQUEUE_KIND":                            "rabbitmq",
-				"SERVER_MSGQUEUE_RABBITMQ_URL":                    "amqp://user:password@rabbit:5672/",
-				"SERVER_MSGQUEUE_PUBSUB_KIND":                     "nats",
-				"SERVER_MSGQUEUE_PUBSUB_NATS_URL":                 "nats://nats:4222",
-				"SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED":         "true",
-				"SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ROOT_CA_FILE":    "/etc/hatchet/nats-ca.pem",
-				"SERVER_MSGQUEUE_PUBSUB_NATS_TLS_HANDSHAKE_FIRST": "true",
+				"SERVER_MSGQUEUE_KIND":                         "rabbitmq",
+				"SERVER_MSGQUEUE_RABBITMQ_URL":                 "amqp://user:password@rabbit:5672/",
+				"SERVER_MSGQUEUE_PUBSUB_KIND":                  "nats",
+				"SERVER_MSGQUEUE_PUBSUB_NATS_URL":              "nats://nats:4222",
+				"SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED":      "true",
+				"SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ROOT_CA_FILE": "/etc/hatchet/nats-ca.pem",
 			},
-			wantKind:                  "nats",
-			wantURL:                   "amqp://user:password@rabbit:5672/",
-			wantNatsURL:               "nats://nats:4222",
-			wantNatsTLSEnabled:        true,
-			wantNatsTLSRootCAFile:     "/etc/hatchet/nats-ca.pem",
-			wantNatsTLSHandshakeFirst: true,
-			wantMaxPub:                10,
-			wantMaxSub:                20,
+			wantKind:              "nats",
+			wantURL:               "amqp://user:password@rabbit:5672/",
+			wantNatsURL:           "nats://nats:4222",
+			wantNatsTLSEnabled:    true,
+			wantNatsTLSRootCAFile: "/etc/hatchet/nats-ca.pem",
+			wantMaxPub:            10,
+			wantMaxSub:            20,
 		},
 		{
 			name: "nats pubsub with subject prefix",
@@ -177,7 +174,6 @@ func TestPubSubSettingsInheritance(t *testing.T) {
 			assert.Equal(t, tc.wantNatsPassword, cf.MessageQueue.PubSub.NATS.Password)
 			assert.Equal(t, tc.wantNatsTLSEnabled, cf.MessageQueue.PubSub.NATS.TLSEnabled)
 			assert.Equal(t, tc.wantNatsTLSRootCAFile, cf.MessageQueue.PubSub.NATS.TLSRootCAFile)
-			assert.Equal(t, tc.wantNatsTLSHandshakeFirst, cf.MessageQueue.PubSub.NATS.TLSHandshakeFirst)
 			assert.Equal(t, tc.wantNatsSubjectPrefix, cf.MessageQueue.PubSub.NATS.SubjectPrefix)
 		})
 	}

pkg/config/server/server.go

@@ -584,24 +584,20 @@ type PubSubNATSConfigFile struct {
 	Username string `mapstructure:"username" json:"username,omitempty"`
 	Password string `mapstructure:"password" json:"password,omitempty"`
 
-	// TLSEnabled requires verified TLS regardless of the URL scheme. Without
-	// TLSRootCAFile the server certificate is verified against the system
-	// roots. False leaves TLS to the URL scheme alone.
+	// TLSEnabled requires verified TLS with a TLS-first handshake: the
+	// handshake happens before the server's INFO message, so the server must
+	// set handshake_first in its tls block. A client with this flag fails to
+	// connect to an INFO-first TLS or plaintext server. Without TLSRootCAFile
+	// the server certificate is verified against the system roots. False
+	// leaves TLS to the URL scheme alone (tls:// does standard INFO-first
+	// TLS with system roots).
 	TLSEnabled bool `mapstructure:"tlsEnabled" json:"tlsEnabled,omitempty"`
 
 	// TLSRootCAFile is a PEM CA bundle used to verify a NATS server whose
 	// certificate is signed by a private CA (it also applies to rediscovered
 	// cluster peers). Requires TLSEnabled: true; startup fails otherwise.
 	TLSRootCAFile string `mapstructure:"tlsRootCAFile" json:"tlsRootCAFile,omitempty"`
 
-	// TLSHandshakeFirst performs the TLS handshake before the server's INFO
-	// message; the server must enable handshake_first in its tls block.
-	// Against a server that sends INFO first, the client's TLS handshake
-	// reads the plaintext INFO as a malformed handshake reply and fails the
-	// connect within the connect timeout. Requires TLSEnabled: true; startup
-	// fails otherwise.
-	TLSHandshakeFirst bool `mapstructure:"tlsHandshakeFirst" json:"tlsHandshakeFirst,omitempty"`
-
 	// SubjectPrefix is prepended (with a trailing ".") to topic names.
 	// Empty defaults to "hatchet.pubsub".
 	SubjectPrefix string `mapstructure:"subjectPrefix" json:"subjectPrefix,omitempty"`
@@ -970,7 +966,6 @@ func BindAllEnv(v *viper.Viper) {
 	_ = v.BindEnv("msgQueue.pubSub.nats.subjectPrefix", "SERVER_MSGQUEUE_PUBSUB_NATS_SUBJECT_PREFIX")
 	_ = v.BindEnv("msgQueue.pubSub.nats.tlsEnabled", "SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ENABLED")
 	_ = v.BindEnv("msgQueue.pubSub.nats.tlsRootCAFile", "SERVER_MSGQUEUE_PUBSUB_NATS_TLS_ROOT_CA_FILE")
-	_ = v.BindEnv("msgQueue.pubSub.nats.tlsHandshakeFirst", "SERVER_MSGQUEUE_PUBSUB_NATS_TLS_HANDSHAKE_FIRST")
 	_ = v.BindEnv("runtime.singleQueueLimit", "SERVER_SINGLE_QUEUE_LIMIT")
 	_ = v.BindEnv("runtime.optimisticSchedulingEnabled", "SERVER_OPTIMISTIC_SCHEDULING_ENABLED")
 	_ = v.BindEnv("runtime.optimisticSchedulingSlots", "SERVER_OPTIMISTIC_SCHEDULING_SLOTS")