Merge pull request #445 from warmbly/feat/nats-credentials

feat: support NATS JWT credentials, so a managed bus is an option

docs/content/docs/development/configuration.mdx

@@ -291,6 +291,8 @@ On `filesystem`, a remote worker writes blobs to its own disk rather than a volu
 |---|---|---|---|
 | `EVENTBUS_PROVIDER` | `nats` or `kafka`. Kafka needs images built with `GO_TAGS=kafka` | `nats` under compose, `kafka` for a bare binary | yes |
 | `NATS_URL` | JetStream address. Credentials in the URL are honored by every service, including the Rust tracking publisher: `nats://user:pass@host:4222` for a user, `nats://token@host:4222` for a token, `tls://` for TLS | `nats://nats:4222` | yes |
+| `NATS_CREDS` | Path to a NATS credentials file (user JWT plus nkey seed), for a bus that authenticates with JWT rather than a token. Synadia Cloud and any nsc-managed account issue one | unset | yes |
+| `NATS_CREDS_B64` | The same file, base64 encoded, as a single line. This is the form the fleet uses: a node receives environment variables rather than files, and the env file docker reads cannot express a multi-line value. Takes precedence over `NATS_CREDS` | unset | yes |
 | `NATS_STREAM_NAME`, `NATS_SUBJECT_PREFIX` | Stream and subject naming | `warmbly` | yes |
 | `KAFKA_BOOTSTRAP_SERVERS` | Broker list when `EVENTBUS_PROVIDER=kafka` | unset | yes |
 | `KAFKA_SASL_USERNAME`, `KAFKA_SASL_PASSWORD` | Broker credentials | unset | yes |

go.mod

@@ -32,8 +32,10 @@ require (
 	github.com/meszmate/apple-go v0.0.0-20250828163208-7fea48c91b32
 	github.com/microcosm-cc/bluemonday v1.0.27
 	github.com/mileusna/useragent v1.3.5
+	github.com/nats-io/jwt/v2 v2.8.1
 	github.com/nats-io/nats-server/v2 v2.14.1
 	github.com/nats-io/nats.go v1.52.0
+	github.com/nats-io/nkeys v0.4.15
 	github.com/openai/openai-go/v2 v2.7.1
 	github.com/oschwald/geoip2-golang/v2 v2.0.0
 	github.com/posthog/posthog-go v1.25.1
@@ -227,8 +229,6 @@ require (
 	github.com/moricho/tparallel v0.3.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nakabonne/nestif v0.3.1 // indirect
-	github.com/nats-io/jwt/v2 v2.8.1 // indirect
-	github.com/nats-io/nkeys v0.4.15 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/nishanths/exhaustive v0.12.0 // indirect
 	github.com/nishanths/predeclared v0.2.2 // indirect

internal/api/handler/fleet_nodes.go

@@ -217,6 +217,9 @@ var nodeEnvKeys = []string{
 	"APP_ENV",
 	"EVENTBUS_PROVIDER",
 	"NATS_URL",
+	// Base64, not a path: a node has no file to point at, and the env file
+	// docker reads cannot hold a multi-line value.
+	"NATS_CREDS_B64",
 	"KAFKA_BOOTSTRAP_SERVERS",
 	"KAFKA_SASL_USERNAME",
 	"KAFKA_SASL_PASSWORD",

internal/infrastructure/eventbus/nats.go

@@ -86,6 +86,15 @@ func NewNATS(cfg NATSConfig) (*NATSBus, error) {
 		nats.ReconnectWait(2 * time.Second),
 	}, cfg.Options...)
 
+	// A JWT credential cannot travel in the URL, so it is resolved separately.
+	creds, err := credsOption()
+	if err != nil {
+		return nil, err
+	}
+	if creds != nil {
+		opts = append(opts, creds)
+	}
+
 	nc, err := nats.Connect(cfg.URL, opts...)
 	if err != nil {
 		return nil, fmt.Errorf("eventbus nats: connect: %w", err)

internal/infrastructure/eventbus/nats_creds.go

@@ -0,0 +1,60 @@
+package eventbus
+
+import (
+	"encoding/base64"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nats.go"
+)
+
+// credsOption resolves the NATS user credential from the environment.
+//
+// A managed bus (Synadia Cloud and anything else using JWT auth) authenticates
+// with a .creds file holding a user JWT and an nkey seed, which no URL can
+// carry. Two forms, because the two places this runs cannot use the same one:
+//
+//	NATS_CREDS      path to the file. Containers and local development.
+//	NATS_CREDS_B64  the file, base64 encoded. A worker joins with an env file
+//	                that docker passes via --env-file, which cannot express a
+//	                multi-line value, so the fleet needs a single-line form.
+//
+// Base64 rather than separate JWT and seed variables: it is one value, it is
+// exactly what Synadia hands you, and there is no chance of pasting the two
+// halves the wrong way round. It is the shape NATS's own NEX project uses.
+//
+// Returns nil when neither is set, so token and user-password URLs keep working.
+func credsOption() (nats.Option, error) {
+	if raw := strings.TrimSpace(os.Getenv("NATS_CREDS_B64")); raw != "" {
+		decoded, err := base64.StdEncoding.DecodeString(raw)
+		if err != nil {
+			return nil, fmt.Errorf("eventbus nats: NATS_CREDS_B64 is not valid base64: %w", err)
+		}
+		return credsFromContents(decoded)
+	}
+	if path := strings.TrimSpace(os.Getenv("NATS_CREDS")); path != "" {
+		return nats.UserCredentials(path), nil
+	}
+	return nil, nil
+}
+
+// credsFromContents builds the option without writing the secret to disk.
+// nats.UserCredentials only takes a path, and a temp file would leave the
+// seed readable to anything else on the host for the life of the process.
+func credsFromContents(contents []byte) (nats.Option, error) {
+	userJWT, err := jwt.ParseDecoratedJWT(contents)
+	if err != nil {
+		return nil, fmt.Errorf("eventbus nats: no user JWT in the credentials: %w", err)
+	}
+	kp, err := jwt.ParseDecoratedUserNKey(contents)
+	if err != nil {
+		return nil, fmt.Errorf("eventbus nats: no nkey seed in the credentials: %w", err)
+	}
+	seed, err := kp.Seed()
+	if err != nil {
+		return nil, fmt.Errorf("eventbus nats: credentials carry no private seed: %w", err)
+	}
+	return nats.UserJWTAndSeed(userJWT, string(seed)), nil
+}

internal/infrastructure/eventbus/nats_creds_test.go

@@ -0,0 +1,82 @@
+package eventbus
+
+import (
+	"encoding/base64"
+	"os"
+	"testing"
+
+	"github.com/nats-io/nkeys"
+)
+
+// Built at run time from a real throwaway keypair: an nkey seed carries a
+// checksum, so a hand-written one cannot parse, and embedding a valid seed in
+// the repo would put key-shaped material in git for no reason.
+func sampleCredsFile(t *testing.T) string {
+	t.Helper()
+	kp, err := nkeys.CreateUser()
+	if err != nil {
+		t.Fatal(err)
+	}
+	seed, err := kp.Seed()
+	if err != nil {
+		t.Fatal(err)
+	}
+	return "-----BEGIN NATS USER JWT-----\n" +
+		"eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.eyJzdWIiOiJVQUEifQ.c2ln\n" +
+		"------END NATS USER JWT------\n\n" +
+		"-----BEGIN USER NKEY SEED-----\n" +
+		string(seed) + "\n" +
+		"------END USER NKEY SEED------\n"
+}
+
+// The fleet ships env files through docker --env-file, which cannot express a
+// multi-line value, so the base64 form is the one that has to work.
+func TestCredsOptionFromBase64(t *testing.T) {
+	t.Setenv("NATS_CREDS_B64", base64.StdEncoding.EncodeToString([]byte(sampleCredsFile(t))))
+	t.Setenv("NATS_CREDS", "")
+	opt, err := credsOption()
+	if err != nil {
+		t.Fatalf("credsOption: %v", err)
+	}
+	if opt == nil {
+		t.Fatal("NATS_CREDS_B64 was set but produced no option")
+	}
+}
+
+func TestCredsOptionFromFile(t *testing.T) {
+	f := t.TempDir() + "/u.creds"
+	if err := os.WriteFile(f, []byte(sampleCredsFile(t)), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("NATS_CREDS_B64", "")
+	t.Setenv("NATS_CREDS", f)
+	opt, err := credsOption()
+	if err != nil || opt == nil {
+		t.Fatalf("file form: opt=%v err=%v", opt, err)
+	}
+}
+
+// Neither set must stay silent: token and user-password URLs still work, and
+// erroring here would break every existing deployment.
+func TestCredsOptionAbsent(t *testing.T) {
+	t.Setenv("NATS_CREDS_B64", "")
+	t.Setenv("NATS_CREDS", "")
+	opt, err := credsOption()
+	if err != nil || opt != nil {
+		t.Fatalf("expected no option and no error, got opt=%v err=%v", opt, err)
+	}
+}
+
+// A truncated or mis-pasted value must say so at boot rather than fail as an
+// unexplained authorization error against the bus.
+func TestCredsOptionRejectsGarbage(t *testing.T) {
+	t.Setenv("NATS_CREDS", "")
+	t.Setenv("NATS_CREDS_B64", "not-base64!!")
+	if _, err := credsOption(); err == nil {
+		t.Error("invalid base64 was accepted")
+	}
+	t.Setenv("NATS_CREDS_B64", base64.StdEncoding.EncodeToString([]byte("no jwt here")))
+	if _, err := credsOption(); err == nil {
+		t.Error("credentials with no JWT were accepted")
+	}
+}

tracking/src/nats.rs

@@ -1,4 +1,6 @@
 use async_nats::jetstream;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use base64::Engine as _;
 
 use crate::config::Config;
 use crate::events::TrackingEvent;
@@ -27,6 +29,22 @@ impl NatsProducer {
                 None => opts.token(user.to_string()),
             };
         }
+        // A managed bus authenticates with a user JWT and nkey seed, which no
+        // URL can carry. NATS_CREDS_B64 is the single-line form the fleet needs
+        // (docker --env-file cannot express a multi-line value); NATS_CREDS is
+        // a path, for containers and local development.
+        if let Ok(b64) = std::env::var("NATS_CREDS_B64") {
+            if !b64.trim().is_empty() {
+                let raw = BASE64.decode(b64.trim())?;
+                opts = opts.credentials(std::str::from_utf8(&raw)?)?;
+            }
+        } else if let Ok(path) = std::env::var("NATS_CREDS") {
+            if !path.trim().is_empty() {
+                let contents = std::fs::read_to_string(path.trim())?;
+                opts = opts.credentials(&contents)?;
+            }
+        }
+
         let client = opts.connect(addr).await?;
         let js = jetstream::new(client);
         let subject = format!("{}.{}", config.nats_subject_prefix, config.kafka_topic);