From f1f524915343adf7c80f3f00391f640beff5eadd Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 12 Sep 2026 05:29:09 +0200 Subject: [PATCH] feat: authenticate to NATS with a user JWT and nkey seed so a managed bus like Synadia Cloud can replace a self-run one, taking the credential from a file path for containers and from a single-line base64 value for the fleet, because a node receives environment variables rather than files and the env file docker reads cannot express the multi-line credentials format, in both the Go event bus and the Rust tracking publisher --- .../docs/development/configuration.mdx | 2 + go.mod | 4 +- internal/api/handler/fleet_nodes.go | 3 + internal/infrastructure/eventbus/nats.go | 9 ++ .../infrastructure/eventbus/nats_creds.go | 60 ++++++++++++++ .../eventbus/nats_creds_test.go | 82 +++++++++++++++++++ tracking/src/nats.rs | 18 ++++ 7 files changed, 176 insertions(+), 2 deletions(-) create mode 100644 internal/infrastructure/eventbus/nats_creds.go create mode 100644 internal/infrastructure/eventbus/nats_creds_test.go diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index 70439be5..e7ef394e 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/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 | diff --git a/go.mod b/go.mod index a8542128..76d5252d 100644 --- a/go.mod +++ b/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 diff --git a/internal/api/handler/fleet_nodes.go b/internal/api/handler/fleet_nodes.go index 753ab35c..6cbb86ff 100644 --- a/internal/api/handler/fleet_nodes.go +++ b/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", diff --git a/internal/infrastructure/eventbus/nats.go b/internal/infrastructure/eventbus/nats.go index 87a8e1f3..bcfa740f 100644 --- a/internal/infrastructure/eventbus/nats.go +++ b/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) diff --git a/internal/infrastructure/eventbus/nats_creds.go b/internal/infrastructure/eventbus/nats_creds.go new file mode 100644 index 00000000..a34d1a7d --- /dev/null +++ b/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 +} diff --git a/internal/infrastructure/eventbus/nats_creds_test.go b/internal/infrastructure/eventbus/nats_creds_test.go new file mode 100644 index 00000000..068d9ea9 --- /dev/null +++ b/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") + } +} diff --git a/tracking/src/nats.rs b/tracking/src/nats.rs index 5b42e97d..df79148e 100644 --- a/tracking/src/nats.rs +++ b/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);