From a95738147599c665c1e6a8c701b65982b780c3a9 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Thu, 13 Aug 2026 11:38:51 +0530 Subject: [PATCH] fix(queue): push tenant account claims to the NATS resolver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /queue/new minted a per-tenant account JWT and handed it to the default no-op ResolverPusher in common/queueprovider/nats, so nats-server never learned the account existed. Against an auth_required server every issued credential failed at CONNECT with "Authorization Violation" — issued-looking credentials that do not work. internal/natsresolver is the missing publisher: one long-lived SYS-account connection opened at boot (auto-reconnecting) that request/replies the signed claim on $SYS.REQ.CLAIMS.UPDATE and requires a positive ack. A non-ack, an unparseable reply, a reply naming a different account, and a timeout are all errors — a claim that may not be installed is never reported as installed. The push is bounded (5s) so a NATS outage 503s instead of hanging the synchronous handler. Wiring fails loudly, never open: - buildQueueProvider attaches the pusher after Factory() through a type assertion that simply misses for non-nats backends. - With NATS_OPERATOR_SEED set, a pusher that cannot be built or cannot reach NATS is a hard error; NewQueueHandler then installs a provider that refuses to issue, so /queue/new answers 503 rather than downgrading to a legacy_open URL the server would reject. - A push rejection at request time tears down the backend resource, marks the row failed and returns 503 (CLAUDE.md rule 2). New env, both secrets — never logged, scrubbed from error strings: NATS_SYSTEM_USER_JWT SYS-account user JWT NATS_SYSTEM_USER_SEED matching NKey seed NATS_SYSTEM_URL optional; default nats://$NATS_HOST:4222 Adds github.com/nats-io/nats.go v1.53.1. Co-Authored-By: Claude Opus 5 (1M context) --- go.mod | 2 + go.sum | 4 + internal/config/config.go | 23 +- internal/config/config_test.go | 16 + .../handlers/export_queueresolver_test.go | 51 ++ internal/handlers/queue.go | 67 ++- internal/handlers/queue_provider.go | 145 +++++- .../handlers/queue_provider_provarms_test.go | 44 ++ .../handlers/queue_resolver_wiring_test.go | 444 ++++++++++++++++++ internal/natsresolver/pusher.go | 324 +++++++++++++ internal/natsresolver/pusher_test.go | 440 +++++++++++++++++ 11 files changed, 1544 insertions(+), 16 deletions(-) create mode 100644 internal/handlers/export_queueresolver_test.go create mode 100644 internal/handlers/queue_resolver_wiring_test.go create mode 100644 internal/natsresolver/pusher.go create mode 100644 internal/natsresolver/pusher_test.go diff --git a/go.mod b/go.mod index 5ed2e1bf..a008158d 100644 --- a/go.mod +++ b/go.mod @@ -16,6 +16,7 @@ require ( github.com/lib/pq v1.10.9 github.com/minio/madmin-go/v3 v3.0.110 github.com/minio/minio-go/v7 v7.0.90 + github.com/nats-io/nats.go v1.53.1 github.com/nats-io/nkeys v0.4.15 github.com/newrelic/go-agent/v3 v3.43.3 github.com/oschwald/maxminddb-golang v1.13.0 @@ -103,6 +104,7 @@ require ( github.com/montanaflynn/stats v0.7.1 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect diff --git a/go.sum b/go.sum index b70da279..3a31c44a 100644 --- a/go.sum +++ b/go.sum @@ -182,8 +182,12 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/nats.go v1.53.1 h1:Otsq3uLc/kLdjmkNHkXH0jBqwUquwdKFoe3fq6/3/Xo= +github.com/nats-io/nats.go v1.53.1/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +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/newrelic/go-agent/v3 v3.43.3 h1:0A6DkUBYK2bidV6jJDJ1SD2XkRlg976nl+SiEqkGTUQ= github.com/newrelic/go-agent/v3 v3.43.3/go.mod h1:MFXnCId5xXMIJI6A/kbkg0DO48EVTsKcmNijMYphzTg= github.com/onsi/ginkgo/v2 v2.27.2 h1:LzwLj0b89qtIy6SSASkzlNvX6WktqurSHwkk2ipF/Ns= diff --git a/internal/config/config.go b/internal/config/config.go index f6f04230..543a0928 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -134,9 +134,23 @@ type Config struct { NATSOperatorSeed string // NATS_OPERATOR_SEED — operator NKey seed; empty = legacy_open fallback NATSSystemAccountKey string // NATS_SYSTEM_ACCOUNT_PUBLIC_KEY — system account public key NATSUseTLS bool // NATS_USE_TLS — true → tls:// URLs - R2Endpoint string // R2_ENDPOINT — R2 endpoint hostname (default: r2.instant.dev) - R2BucketName string // R2_BUCKET_NAME — shared R2 bucket name (default: instant-shared) - R2APIToken string // R2_API_TOKEN — Cloudflare API token; if empty, R2 is not used + + // System-account USER credentials (distinct from NATSSystemAccountKey, + // which is only the account's public key). Minting a tenant account JWT + // is useless unless the claim is pushed to the running nats-server on + // $SYS.REQ.CLAIMS.UPDATE, and only a connection authenticated INTO the + // system account may publish there. Both are secrets and are never + // logged. Required whenever NATSOperatorSeed is set; without them + // /queue/new would issue credentials the server rejects. + NATSSystemUserJWT string // NATS_SYSTEM_USER_JWT — SYS-account user JWT (secret) + NATSSystemUserSeed string // NATS_SYSTEM_USER_SEED — SYS-account user NKey seed (secret) + // NATSSystemURL overrides the URL used for that system-account + // connection. Default: nats://:4222 (in-cluster, plaintext). + NATSSystemURL string // NATS_SYSTEM_URL + + R2Endpoint string // R2_ENDPOINT — R2 endpoint hostname (default: r2.instant.dev) + R2BucketName string // R2_BUCKET_NAME — shared R2 bucket name (default: instant-shared) + R2APIToken string // R2_API_TOKEN — Cloudflare API token; if empty, R2 is not used // Object storage backend for /storage/new (provider-agnostic). // // ObjectStoreBackend selects the credential-issuance strategy: @@ -443,6 +457,9 @@ func Load() *Config { cfg.NATSPublicHost = getenv("NATS_PUBLIC_HOST", "nats.instanode.dev") cfg.NATSOperatorSeed = os.Getenv("NATS_OPERATOR_SEED") cfg.NATSSystemAccountKey = os.Getenv("NATS_SYSTEM_ACCOUNT_PUBLIC_KEY") + cfg.NATSSystemUserJWT = strings.TrimSpace(os.Getenv("NATS_SYSTEM_USER_JWT")) + cfg.NATSSystemUserSeed = strings.TrimSpace(os.Getenv("NATS_SYSTEM_USER_SEED")) + cfg.NATSSystemURL = strings.TrimSpace(os.Getenv("NATS_SYSTEM_URL")) cfg.NATSUseTLS = os.Getenv("NATS_USE_TLS") == "true" cfg.R2Endpoint = getenv("R2_ENDPOINT", "r2.instant.dev") cfg.R2BucketName = getenv("R2_BUCKET_NAME", "instant-shared") diff --git a/internal/config/config_test.go b/internal/config/config_test.go index d57d4792..5faa12e4 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -54,6 +54,7 @@ func allKeys() []string { "POSTGRES_CUSTOMERS_URL", "PROVISIONER_ADDR", "PROVISIONER_SECRET", "NATS_HOST", "QUEUE_BACKEND", "NATS_PUBLIC_HOST", "NATS_OPERATOR_SEED", "NATS_SYSTEM_ACCOUNT_PUBLIC_KEY", "NATS_USE_TLS", + "NATS_SYSTEM_USER_JWT", "NATS_SYSTEM_USER_SEED", "NATS_SYSTEM_URL", "R2_ENDPOINT", "R2_BUCKET_NAME", "R2_API_TOKEN", "OBJECT_STORE_MODE", "OBJECT_STORE_BACKEND", "OBJECT_STORE_ENDPOINT", "OBJECT_STORE_PUBLIC_URL", "OBJECT_STORE_ACCESS_KEY", @@ -307,6 +308,9 @@ func TestLoad_OverrideDefaults(t *testing.T) { "NATS_OPERATOR_SEED": "SO_seed", "NATS_SYSTEM_ACCOUNT_PUBLIC_KEY": "ACSYS", "NATS_USE_TLS": "true", + "NATS_SYSTEM_USER_JWT": " eyJ0eXAiOiJKV1QifQ.sys.user\n", + "NATS_SYSTEM_USER_SEED": "SUSYSSEED\n", + "NATS_SYSTEM_URL": " nats://nats.x:4222 ", "R2_ENDPOINT": "r2.x", "R2_BUCKET_NAME": "x-bucket", "R2_API_TOKEN": "r2tok", @@ -333,6 +337,18 @@ func TestLoad_OverrideDefaults(t *testing.T) { if !cfg.NATSUseTLS { t.Error("NATSUseTLS must be true when env=true") } + // SYS-account user credentials + system-URL override. All three are + // whitespace-trimmed: trailing newlines are what `kubectl create secret + // --from-file` leaves behind, and an untrimmed NKey seed fails to parse. + if cfg.NATSSystemUserJWT != "eyJ0eXAiOiJKV1QifQ.sys.user" { + t.Errorf("NATSSystemUserJWT trim: %q", cfg.NATSSystemUserJWT) + } + if cfg.NATSSystemUserSeed != "SUSYSSEED" { + t.Errorf("NATSSystemUserSeed trim: %q", cfg.NATSSystemUserSeed) + } + if cfg.NATSSystemURL != "nats://nats.x:4222" { + t.Errorf("NATSSystemURL trim: %q", cfg.NATSSystemURL) + } // API_PUBLIC_URL — trailing slash must be trimmed. if cfg.APIPublicURL != "https://api.x" { t.Errorf("APIPublicURL trim: %q", cfg.APIPublicURL) diff --git a/internal/handlers/export_queueresolver_test.go b/internal/handlers/export_queueresolver_test.go new file mode 100644 index 00000000..e913de2c --- /dev/null +++ b/internal/handlers/export_queueresolver_test.go @@ -0,0 +1,51 @@ +package handlers + +// export_queueresolver_test.go — test-only re-exports for the NATS +// resolver-pusher wiring slice (queue_provider.go attachResolverPusher / +// natsSystemURL / isolationUnavailable / unavailableCredProvider and +// queue.go failQueueCredIssue). +// +// Kept in its own file (not the shared export_provarms_test.go) so concurrent +// work on other handler slices never collides here. Go only compiles it in +// test builds. + +import ( + "github.com/gofiber/fiber/v2" + + "instant.dev/common/queueprovider" + natsqp "instant.dev/common/queueprovider/nats" + "instant.dev/internal/config" + "instant.dev/internal/models" + "instant.dev/internal/natsresolver" +) + +// SwapResolverPusherFactoryForTest replaces the system-account pusher +// constructor and returns a restore func, so the attach / skip / fail arms of +// attachResolverPusher can be driven without a NATS server. +func SwapResolverPusherFactoryForTest( + fn func(natsresolver.Config) (natsqp.ResolverPusher, error), +) (restore func()) { + prev := newResolverPusher + newResolverPusher = fn + return func() { newResolverPusher = prev } +} + +// NATSSystemURLForTest re-exports natsSystemURL. +func NATSSystemURLForTest(cfg *config.Config) string { return natsSystemURL(cfg) } + +// IsolationUnavailableForTest re-exports isolationUnavailable. +func IsolationUnavailableForTest(err error) bool { return isolationUnavailable(err) } + +// NewUnavailableCredProviderForTest re-exports the stand-in provider installed +// when isolation is configured but could not be initialised. +func NewUnavailableCredProviderForTest(cause error) queueprovider.QueueCredentialProvider { + return unavailableCredProvider{cause: cause} +} + +// FailQueueCredIssueForTest re-exports QueueHandler.failQueueCredIssue so the +// mark-failed-error log branch can be driven with a closed DB. +func (h *QueueHandler) FailQueueCredIssueForTest( + c *fiber.Ctx, resource *models.Resource, prid, token, logPrefix string, cause error, +) error { + return h.failQueueCredIssue(c, resource, prid, token, logPrefix, cause) +} diff --git a/internal/handlers/queue.go b/internal/handlers/queue.go index ca443f12..7847b1e6 100644 --- a/internal/handlers/queue.go +++ b/internal/handlers/queue.go @@ -65,21 +65,40 @@ func NewQueueHandler(db *sql.DB, rdb *redis.Client, cfg *config.Config, provClie // does not yet have a ProvisionQueue RPC. When it does, wire it here like // CacheHandler.provisionCache does. h.queueProvider = queueprovider.New(cfg.NATSHost) - // Build the credential issuer. Falls back to legacy_open when no operator - // seed is configured so api can deploy before the operator-key generation. - if cp, err := buildQueueProvider(cfg); err == nil { + // Build the credential issuer. Three outcomes: a working provider; a + // legacy_open fallback when NO operator seed is configured (so api can + // deploy before the operator-key generation); or — when the operator seed + // IS configured but the isolation path could not be initialised — a + // provider that refuses to issue, so /queue/new 503s instead of returning + // credentials nats-server would reject. + cp, err := buildQueueProvider(cfg) + switch { + case err == nil: h.credProvider = cp - } else { + case cfg.NATSOperatorSeed != "": + // Isolation is CONFIGURED (operator seed present) but could not be + // initialised — typically the $SYS resolver publisher failed to + // connect. Falling back to legacy_open here would hand every caller a + // connection URL with no credentials against an auth_required server: + // the exact "issued but dead" failure this path exists to prevent. + // Fail the queue credential path loudly instead; /queue/new answers + // 503 until the operator fixes the NATS wiring. + slog.Error("queue.cred_provider_init_failed_isolation_unavailable", + "error", err, + "backend", cfg.QueueBackend, + "detail", "operator seed is set — refusing to downgrade to legacy_open; /queue/new will 503") + h.credProvider = unavailableCredProvider{cause: err} + default: slog.Error("queue.cred_provider_init_failed_fallback_legacy_open", "error", err, "backend", cfg.QueueBackend) // Defensive: never leave h.credProvider nil. The legacyopen provider // is always registered so this fallback always succeeds. fallback, _ := commonqp.Factory(commonqp.Config{ - Backend: "legacy_open", + Backend: queueBackendLegacyOpen, Host: cfg.NATSHost, PublicHost: cfg.NATSPublicHost, - Port: 4222, + Port: natsClientPort, UseTLS: cfg.NATSUseTLS, }) h.credProvider = fallback @@ -150,6 +169,32 @@ func (h *QueueHandler) issueTenantCreds(ctx context.Context, token, subjectPrefi return creds, nil } +// failQueueCredIssue aborts a provision whose per-tenant credentials could not +// be issued because the isolation path is broken (the account claim never +// reached the nats-server resolver). +// +// CLAUDE.md rule 2: provisioning is synchronous and a backend failure is a +// 503 — never a 201 carrying credentials for something the backend does not +// know about. Returning the legacy_open response shape here would be exactly +// that: a connection URL with no credentials against an auth_required server. +// So the backend resource is torn down, the row is marked failed (failed rows +// never count against quota) and the caller gets 503. +func (h *QueueHandler) failQueueCredIssue( + c *fiber.Ctx, resource *models.Resource, prid, token, logPrefix string, cause error, +) error { + ctx := c.UserContext() + metrics.ProvisionFailures.WithLabelValues("queue", "cred_issue_error").Inc() + middleware.RecordProvisionFail("queue", middleware.ProvisionFailBackendUnavailable) + slog.Error(logPrefix+".cred_issue_failed_isolation_unavailable", + "error", cause, "token", token, "resource_id", resource.ID) + deprovisionBestEffort(ctx, h.provClient, token, prid, "queue", logPrefix) + if delErr := models.MarkResourceFailed(ctx, h.db, resource.ID); delErr != nil { + slog.Error(logPrefix+".soft_delete_failed_cred_issue", + "error", delErr, "resource_id", resource.ID) + } + return respondProvisionFailed(c, cause, "Failed to issue isolated NATS credentials") +} + // NewQueue handles POST /queue/new. func (h *QueueHandler) NewQueue(c *fiber.Ctx) error { if !h.cfg.IsServiceEnabled("queue") { @@ -308,7 +353,10 @@ func (h *QueueHandler) NewQueue(c *fiber.Ctx) error { // MR-P0-5: issue per-tenant credentials via the queueprovider abstraction. // May return AuthMode=isolated (real per-tenant account JWT) or // AuthMode=legacy_open (no auth — staged-cutover fallback). - tenantCreds, _ := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix) + tenantCreds, credErr := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix) + if isolationUnavailable(credErr) { + return h.failQueueCredIssue(c, resource, creds.ProviderResourceID, tokenStr, "queue.new", credErr) + } authMode := commonqp.AuthModeLegacyOpen if tenantCreds != nil && tenantCreds.AuthMode != "" { authMode = tenantCreds.AuthMode @@ -531,7 +579,10 @@ func (h *QueueHandler) newQueueAuthenticated( } // MR-P0-5: issue per-tenant credentials via the queueprovider abstraction. - tenantCreds, _ := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix) + tenantCreds, credErr := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix) + if isolationUnavailable(credErr) { + return h.failQueueCredIssue(c, resource, creds.ProviderResourceID, tokenStr, "queue.new.auth", credErr) + } authMode := commonqp.AuthModeLegacyOpen if tenantCreds != nil && tenantCreds.AuthMode != "" { authMode = tenantCreds.AuthMode diff --git a/internal/handlers/queue_provider.go b/internal/handlers/queue_provider.go index 8cb6d402..79c15921 100644 --- a/internal/handlers/queue_provider.go +++ b/internal/handlers/queue_provider.go @@ -13,23 +13,68 @@ package handlers // - once the operator seed is configured, every new /queue/new mints a real // per-tenant account JWT + user NKey via the provider. // +// Minting is only half the job. A tenant account JWT the running nats-server +// has never seen is rejected at CONNECT with "Authorization Violation", so the +// nats provider's ResolverPusher seam must be filled with a real +// system-account publisher — internal/natsresolver — or every isolated +// credential is dead on arrival. attachResolverPusher below is that wiring, +// and it fails LOUDLY: when the operator seed says "issue isolated creds" but +// the resolver push cannot be set up, buildQueueProvider errors instead of +// quietly degrading to legacy_open. A silent degrade is what shipped the bug. +// // The provider lives at handler-scope (one per process); IssueTenantCredentials // is concurrency-safe per the queueprovider contract. import ( + "context" + "errors" + "fmt" "log/slog" + "strings" "instant.dev/common/queueprovider" // register every backend by side-effect import — same pattern as - // storageprovider wiring in router.go. + // storageprovider wiring in router.go. The nats backend is imported by + // name (not blank) because attachResolverPusher needs its ResolverPusher + // interface type for the setter type-assertion. _ "instant.dev/common/queueprovider/kafka" _ "instant.dev/common/queueprovider/legacyopen" - _ "instant.dev/common/queueprovider/nats" + natsqp "instant.dev/common/queueprovider/nats" _ "instant.dev/common/queueprovider/rabbitmq" "instant.dev/internal/config" + "instant.dev/internal/natsresolver" +) + +const ( + // queueBackendNATS is the operator-mode backend that mints per-tenant + // account JWTs. + queueBackendNATS = "nats" + // queueBackendLegacyOpen is the pre-cutover unauthenticated shim. + queueBackendLegacyOpen = "legacy_open" + // natsClientPort is the broker port for both tenant and system-account + // connections. + natsClientPort = 4222 + // natsPlainScheme is the URL scheme used for the in-cluster + // system-account connection. Operators needing TLS set NATS_SYSTEM_URL. + natsPlainScheme = "nats" ) +// resolverPusherSetter is the sub-interface of *natsqp.Provider that accepts a +// ResolverPusher. queueprovider.Factory returns an interface, and only the +// nats backend implements this method — legacy_open / rabbitmq / kafka have no +// resolver, so the type-assertion below simply misses for them (never panics). +type resolverPusherSetter interface { + SetResolverPusher(natsqp.ResolverPusher) +} + +// newResolverPusher is the construction seam for the system-account pusher. +// Production aliases natsresolver.New; tests substitute a stub so the +// attach/skip/fail arms can be driven without a NATS server. +var newResolverPusher = func(cfg natsresolver.Config) (natsqp.ResolverPusher, error) { + return natsresolver.New(cfg) +} + // buildQueueProvider constructs the queueprovider.QueueCredentialProvider from // cfg. Falls back to the legacy_open shim when QUEUE_BACKEND is unset AND no // operator seed is configured, so deploys before the operator-key generation @@ -43,16 +88,16 @@ func buildQueueProvider(cfg *config.Config) (queueprovider.QueueCredentialProvid // (un-isolated) traffic until the operator keys are generated. After // the operator seed is wired, the same code mints isolated creds. if cfg.NATSOperatorSeed == "" { - backend = "legacy_open" + backend = queueBackendLegacyOpen } else { - backend = "nats" + backend = queueBackendNATS } } qpCfg := queueprovider.Config{ Backend: backend, Host: cfg.NATSHost, PublicHost: cfg.NATSPublicHost, - Port: 4222, + Port: natsClientPort, UseTLS: cfg.NATSUseTLS, NATSOperatorSeed: cfg.NATSOperatorSeed, NATSSystemAccountPublicKey: cfg.NATSSystemAccountKey, @@ -62,6 +107,9 @@ func buildQueueProvider(cfg *config.Config) (queueprovider.QueueCredentialProvid if err != nil { return nil, err } + if err := attachResolverPusher(cfg, qp); err != nil { + return nil, err + } caps := qp.Capabilities() slog.Info("queue.provider_initialised", "backend", qp.Name(), @@ -72,3 +120,90 @@ func buildQueueProvider(cfg *config.Config) (queueprovider.QueueCredentialProvid ) return qp, nil } + +// attachResolverPusher gives the nats provider a real +// $SYS.REQ.CLAIMS.UPDATE publisher. +// +// Three outcomes, and only one of them is an error: +// - backend has no resolver (legacy_open / rabbitmq / kafka) → nothing to do. +// - nats backend with no operator seed → the provider returns legacy_open +// creds and never mints an account claim, so there is nothing to push. +// - nats backend WITH an operator seed → isolated credentials will be +// minted, and they only work if the claim reaches the server. A pusher +// that cannot be built or cannot connect is a hard error: returning nil +// here would leave the provider on its no-op pusher and hand customers +// credentials that fail at CONNECT. +func attachResolverPusher(cfg *config.Config, qp queueprovider.QueueCredentialProvider) error { + setter, ok := qp.(resolverPusherSetter) + if !ok { + return nil + } + if cfg.NATSOperatorSeed == "" { + slog.Warn("queue.resolver_pusher_skipped_no_operator_seed", + "backend", qp.Name(), + "detail", "NATS_OPERATOR_SEED unset — /queue/new serves legacy_open credentials") + return nil + } + url := natsSystemURL(cfg) + pusher, err := newResolverPusher(natsresolver.Config{ + URL: url, + UserJWT: cfg.NATSSystemUserJWT, + UserSeed: cfg.NATSSystemUserSeed, + }) + if err != nil { + return fmt.Errorf("queue: NATS_OPERATOR_SEED is set so /queue/new mints per-tenant "+ + "account JWTs, but the %s publisher could not be started — those credentials "+ + "would be rejected by nats-server. Set NATS_SYSTEM_USER_JWT + NATS_SYSTEM_USER_SEED "+ + "and make %s reachable: %w", natsresolver.ClaimsUpdateSubject, url, err) + } + setter.SetResolverPusher(pusher) + slog.Info("queue.resolver_pusher_attached", + "backend", qp.Name(), + "url", url, + "subject", natsresolver.ClaimsUpdateSubject) + return nil +} + +// natsSystemURL resolves the URL for the system-account connection. +// NATS_SYSTEM_URL wins when set (that is the escape hatch for TLS or an +// out-of-cluster endpoint); otherwise it is derived from NATS_HOST. +func natsSystemURL(cfg *config.Config) string { + if u := strings.TrimSpace(cfg.NATSSystemURL); u != "" { + return u + } + return fmt.Sprintf("%s://%s:%d", natsPlainScheme, cfg.NATSHost, natsClientPort) +} + +// isolationUnavailable reports whether a credential-issuance error means the +// per-tenant isolation path is broken — i.e. the account claim never reached +// the resolver. Such an error must fail the provision (503) rather than +// degrade to a legacy_open response, because the connection URL in that +// response is unusable against an auth_required server. +func isolationUnavailable(err error) bool { + return err != nil && errors.Is(err, natsresolver.ErrPushFailed) +} + +// unavailableCredProvider stands in for the nats provider when isolation is +// CONFIGURED but could not be initialised. It never issues a credential; every +// call returns an ErrPushFailed-wrapped error so /queue/new answers 503 +// instead of returning an unauthenticated URL that the server will reject. +type unavailableCredProvider struct{ cause error } + +func (u unavailableCredProvider) IssueTenantCredentials(_ context.Context, _ queueprovider.IssueRequest) (*queueprovider.TenantCreds, error) { + return nil, fmt.Errorf("%w: queue isolation is configured but unavailable: %v", + natsresolver.ErrPushFailed, u.cause) +} + +func (u unavailableCredProvider) RevokeTenantCredentials(_ context.Context, _ string) error { + return nil +} + +func (u unavailableCredProvider) Capabilities() queueprovider.Capabilities { + return queueprovider.Capabilities{ + PerTenantAccounts: true, + SubjectScopedAuth: true, + StreamIsolation: true, + } +} + +func (u unavailableCredProvider) Name() string { return "nats-unavailable" } diff --git a/internal/handlers/queue_provider_provarms_test.go b/internal/handlers/queue_provider_provarms_test.go index ce4def66..0a73caa0 100644 --- a/internal/handlers/queue_provider_provarms_test.go +++ b/internal/handlers/queue_provider_provarms_test.go @@ -10,14 +10,17 @@ package handlers_test // direct calls cover each arm. import ( + "context" "testing" "github.com/nats-io/nkeys" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + natsqp "instant.dev/common/queueprovider/nats" "instant.dev/internal/config" "instant.dev/internal/handlers" + "instant.dev/internal/natsresolver" ) // TestBuildQueueProvider_DefaultNoSeed_FallsBackToLegacyOpen — empty @@ -41,18 +44,28 @@ func TestBuildQueueProvider_DefaultNoSeed_FallsBackToLegacyOpen(t *testing.T) { // TestBuildQueueProvider_DefaultWithSeed_SelectsNATS — empty QueueBackend but // a valid operator seed present → the "nats" backend is selected and builds. +// +// An operator seed also means real account claims will be minted, so the +// system-account resolver publisher must come up too; the constructor seam is +// stubbed here so no NATS server is needed. func TestBuildQueueProvider_DefaultWithSeed_SelectsNATS(t *testing.T) { kp, err := nkeys.CreateOperator() require.NoError(t, err) seed, err := kp.Seed() require.NoError(t, err) + restore := handlers.SwapResolverPusherFactoryForTest( + func(natsresolver.Config) (natsqp.ResolverPusher, error) { return stubResolverPusher{}, nil }) + defer restore() + cfg := &config.Config{ QueueBackend: "", NATSOperatorSeed: string(seed), NATSHost: "nats.test", NATSPublicHost: "nats.instanode.dev", NATSSystemAccountKey: "", + NATSSystemUserJWT: "eyJ0eXAiOiJKV1QifQ.sys.user", + NATSSystemUserSeed: "SUAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", } qp, err := handlers.BuildQueueProviderForTest(cfg) require.NoError(t, err) @@ -60,6 +73,37 @@ func TestBuildQueueProvider_DefaultWithSeed_SelectsNATS(t *testing.T) { assert.Equal(t, "nats", qp.Name()) } +// stubResolverPusher is an accept-everything resolver publisher for the +// backend-selection tests, which are not about push behaviour. +type stubResolverPusher struct{} + +func (stubResolverPusher) PushAccountClaim(context.Context, string, string) error { return nil } + +// TestBuildQueueProvider_SeedWithoutSystemCreds_Errors — the regression guard +// for the bug this wiring fixes. An operator seed with no SYS user credentials +// means account claims can never be pushed to nats-server, so every issued +// credential would be rejected at CONNECT. buildQueueProvider must refuse to +// return a provider rather than let the caller degrade to legacy_open. +func TestBuildQueueProvider_SeedWithoutSystemCreds_Errors(t *testing.T) { + kp, err := nkeys.CreateOperator() + require.NoError(t, err) + seed, err := kp.Seed() + require.NoError(t, err) + + cfg := &config.Config{ + QueueBackend: "nats", + NATSOperatorSeed: string(seed), + NATSHost: "nats.test", + NATSPublicHost: "nats.instanode.dev", + // NATSSystemUserJWT / NATSSystemUserSeed deliberately unset. + } + qp, err := handlers.BuildQueueProviderForTest(cfg) + require.Error(t, err) + assert.Nil(t, qp) + assert.ErrorIs(t, err, natsresolver.ErrPushFailed) + assert.Contains(t, err.Error(), "NATS_SYSTEM_USER_JWT") +} + // TestBuildQueueProvider_ExplicitLegacyOpen — explicit backend overrides the // seed-based default selection. func TestBuildQueueProvider_ExplicitLegacyOpen(t *testing.T) { diff --git a/internal/handlers/queue_resolver_wiring_test.go b/internal/handlers/queue_resolver_wiring_test.go new file mode 100644 index 00000000..54aec90f --- /dev/null +++ b/internal/handlers/queue_resolver_wiring_test.go @@ -0,0 +1,444 @@ +package handlers_test + +// queue_resolver_wiring_test.go — covers the NATS resolver-pusher wiring +// added to queue_provider.go / queue.go. +// +// The bug being locked down: common/queueprovider/nats minted a per-tenant +// account JWT and pushed it to a NO-OP ResolverPusher, so nats-server never +// learned the account existed and every /queue/new credential failed at +// CONNECT with "Authorization Violation". These tests assert that +// +// 1. a nats backend with an operator seed gets a REAL pusher, and the minted +// account claim actually reaches it; +// 2. a pusher that cannot be built is a hard error, never a quiet downgrade +// to legacy_open (that downgrade is what shipped dead credentials); +// 3. a push rejection fails the provision with 503 instead of returning an +// unusable connection URL; +// 4. a non-nats backend never panics on the SetResolverPusher type assertion. +// +// No NATS server is involved: the pusher-construction seam is swapped. + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/gofiber/fiber/v2" + "github.com/google/uuid" + "github.com/nats-io/nkeys" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + commonqp "instant.dev/common/queueprovider" + natsqp "instant.dev/common/queueprovider/nats" + "instant.dev/internal/config" + "instant.dev/internal/handlers" + "instant.dev/internal/middleware" + "instant.dev/internal/models" + "instant.dev/internal/natsresolver" + "instant.dev/internal/plans" + "instant.dev/internal/testhelpers" +) + +// ── fixtures ───────────────────────────────────────────────────────────────── + +// recordingPusher captures every account claim handed to the resolver and can +// be programmed to reject, exactly as a real nats-server non-ack does. +type recordingPusher struct { + mu sync.Mutex + pushed []pushedClaim + pushErr error +} + +type pushedClaim struct{ accountPub, accountJWT string } + +func (r *recordingPusher) PushAccountClaim(_ context.Context, accountPub, accountJWT string) error { + r.mu.Lock() + defer r.mu.Unlock() + r.pushed = append(r.pushed, pushedClaim{accountPub, accountJWT}) + return r.pushErr +} + +func (r *recordingPusher) claims() []pushedClaim { + r.mu.Lock() + defer r.mu.Unlock() + return append([]pushedClaim(nil), r.pushed...) +} + +// mustOperatorSeed returns a real operator NKey seed so the nats provider +// reaches its isolated-credential path (a fake seed fails at builder time). +func mustOperatorSeed(t *testing.T) string { + t.Helper() + kp, err := nkeys.CreateOperator() + require.NoError(t, err) + seed, err := kp.Seed() + require.NoError(t, err) + return string(seed) +} + +func sysCredCfg(t *testing.T, backend string) *config.Config { + t.Helper() + return &config.Config{ + QueueBackend: backend, + NATSOperatorSeed: mustOperatorSeed(t), + NATSHost: "nats.instant-data.svc.cluster.local", + NATSPublicHost: "nats.instanode.dev", + NATSSystemUserJWT: "eyJ0eXAiOiJKV1QifQ.system.user", + NATSSystemUserSeed: "SUAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + } +} + +// ── attachResolverPusher ───────────────────────────────────────────────────── + +// TestAttachResolverPusher_Arms is the wiring table: which configurations get +// a pusher, which skip it, and which fail loudly. +func TestAttachResolverPusher_Arms(t *testing.T) { + tests := []struct { + name string + // cfgFn builds the config under test. + cfgFn func(t *testing.T) *config.Config + // factoryErr, when set, makes the pusher constructor fail. + factoryErr error + wantErr bool + wantErrHas string + // wantFactoryCalls is how often the pusher constructor should run. + wantFactoryCalls int + wantBackend string + }{ + { + name: "nats backend with operator seed attaches a real pusher", + cfgFn: func(t *testing.T) *config.Config { return sysCredCfg(t, "nats") }, + wantFactoryCalls: 1, + wantBackend: "nats", + }, + { + name: "nats backend without an operator seed skips the pusher", + cfgFn: func(*testing.T) *config.Config { + return &config.Config{ + QueueBackend: "nats", + NATSHost: "nats.test", + NATSPublicHost: "nats.instanode.dev", + } + }, + wantFactoryCalls: 0, + wantBackend: "nats", + }, + { + name: "non-nats backend has no SetResolverPusher — type assertion misses, no panic", + cfgFn: func(t *testing.T) *config.Config { + // Operator seed present but the backend is legacy_open: the + // returned provider does not implement the setter at all. + return sysCredCfg(t, "legacy_open") + }, + wantFactoryCalls: 0, + wantBackend: "legacy_open", + }, + { + name: "pusher construction failure is a hard error, never a legacy_open downgrade", + cfgFn: func(t *testing.T) *config.Config { return sysCredCfg(t, "nats") }, + factoryErr: fmt.Errorf("%w: connect refused", natsresolver.ErrPushFailed), + wantErr: true, + wantErrHas: natsresolver.ClaimsUpdateSubject, + wantFactoryCalls: 1, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var calls int + var gotCfg natsresolver.Config + restore := handlers.SwapResolverPusherFactoryForTest( + func(c natsresolver.Config) (natsqp.ResolverPusher, error) { + calls++ + gotCfg = c + if tc.factoryErr != nil { + return nil, tc.factoryErr + } + return &recordingPusher{}, nil + }) + defer restore() + + cfg := tc.cfgFn(t) + qp, err := handlers.BuildQueueProviderForTest(cfg) + + assert.Equal(t, tc.wantFactoryCalls, calls, "resolver-pusher constructor call count") + if tc.wantErr { + require.Error(t, err) + assert.Nil(t, qp, "a provider that cannot push claims must not be returned") + assert.Contains(t, err.Error(), tc.wantErrHas) + assert.ErrorIs(t, err, natsresolver.ErrPushFailed) + return + } + require.NoError(t, err) + require.NotNil(t, qp) + assert.Equal(t, tc.wantBackend, qp.Name()) + if tc.wantFactoryCalls > 0 { + assert.Equal(t, "nats://nats.instant-data.svc.cluster.local:4222", gotCfg.URL) + assert.Equal(t, cfg.NATSSystemUserJWT, gotCfg.UserJWT) + assert.Equal(t, cfg.NATSSystemUserSeed, gotCfg.UserSeed) + } + }) + } +} + +// TestAttachResolverPusher_ClaimReachesTheResolver is the end-to-end assertion +// the original bug failed: minting a tenant account must PUBLISH that account's +// claim, otherwise the credential is dead on arrival. +func TestAttachResolverPusher_ClaimReachesTheResolver(t *testing.T) { + rec := &recordingPusher{} + restore := handlers.SwapResolverPusherFactoryForTest( + func(natsresolver.Config) (natsqp.ResolverPusher, error) { return rec, nil }) + defer restore() + + qp, err := handlers.BuildQueueProviderForTest(sysCredCfg(t, "nats")) + require.NoError(t, err) + + token := uuid.NewString() + creds, err := qp.IssueTenantCredentials(context.Background(), commonqp.IssueRequest{ResourceToken: token}) + require.NoError(t, err) + require.NotNil(t, creds) + assert.Equal(t, commonqp.AuthModeIsolated, creds.AuthMode) + + claims := rec.claims() + require.Len(t, claims, 1, "exactly one account claim must be pushed per issued credential") + assert.Equal(t, creds.KeyID, claims[0].accountPub, + "the pushed account must be the one the returned user JWT belongs to") + assert.NotEmpty(t, claims[0].accountJWT) +} + +// TestAttachResolverPusher_RejectedClaimFailsIssuance — a resolver that does +// not ack must abort issuance, and the error must be recognisable as an +// isolation failure so the handler can 503. +func TestAttachResolverPusher_RejectedClaimFailsIssuance(t *testing.T) { + rec := &recordingPusher{pushErr: fmt.Errorf("%w: resolver did not ack", natsresolver.ErrPushFailed)} + restore := handlers.SwapResolverPusherFactoryForTest( + func(natsresolver.Config) (natsqp.ResolverPusher, error) { return rec, nil }) + defer restore() + + qp, err := handlers.BuildQueueProviderForTest(sysCredCfg(t, "nats")) + require.NoError(t, err) + + creds, err := qp.IssueTenantCredentials(context.Background(), + commonqp.IssueRequest{ResourceToken: uuid.NewString()}) + require.Error(t, err) + assert.Nil(t, creds, "no credential may be returned when the claim was not installed") + assert.True(t, handlers.IsolationUnavailableForTest(err), + "a push rejection must be classified as isolation-unavailable so /queue/new 503s") +} + +// ── natsSystemURL ──────────────────────────────────────────────────────────── + +func TestNATSSystemURL(t *testing.T) { + tests := []struct { + name string + cfg *config.Config + want string + }{ + { + name: "derived from NATS_HOST", + cfg: &config.Config{NATSHost: "nats.instant-data.svc.cluster.local"}, + want: "nats://nats.instant-data.svc.cluster.local:4222", + }, + { + name: "explicit NATS_SYSTEM_URL wins", + cfg: &config.Config{NATSHost: "ignored", NATSSystemURL: "tls://nats.example:4443"}, + want: "tls://nats.example:4443", + }, + { + name: "whitespace-only override falls back to the derived URL", + cfg: &config.Config{NATSHost: "nats.test", NATSSystemURL: " "}, + want: "nats://nats.test:4222", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, handlers.NATSSystemURLForTest(tc.cfg)) + }) + } +} + +// ── isolationUnavailable / unavailableCredProvider ─────────────────────────── + +func TestIsolationUnavailable(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "nil error", err: nil, want: false}, + {name: "unrelated error", err: errors.New("operator seed unavailable"), want: false}, + {name: "sentinel", err: natsresolver.ErrPushFailed, want: true}, + { + name: "wrapped sentinel (the shape the nats provider returns)", + err: fmt.Errorf("queueprovider.nats: push account claim to resolver: %w", + fmt.Errorf("%w: timeout", natsresolver.ErrPushFailed)), + want: true, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, handlers.IsolationUnavailableForTest(tc.err)) + }) + } +} + +func TestUnavailableCredProvider(t *testing.T) { + cause := errors.New("resolver pusher could not connect") + p := handlers.NewUnavailableCredProviderForTest(cause) + + assert.Equal(t, "nats-unavailable", p.Name()) + caps := p.Capabilities() + assert.True(t, caps.PerTenantAccounts, "isolation is configured — the capability is still advertised") + assert.True(t, caps.SubjectScopedAuth) + assert.True(t, caps.StreamIsolation) + assert.NoError(t, p.RevokeTenantCredentials(context.Background(), "AKEY")) + + creds, err := p.IssueTenantCredentials(context.Background(), + commonqp.IssueRequest{ResourceToken: uuid.NewString()}) + require.Error(t, err) + assert.Nil(t, creds) + assert.True(t, handlers.IsolationUnavailableForTest(err)) + assert.Contains(t, err.Error(), cause.Error()) +} + +// ── handler behaviour: 503 instead of a dead connection URL ────────────────── + +// queueResolverApp builds a /queue/new app whose QueueHandler is constructed +// from cfg — i.e. the real NewQueueHandler branch selection runs, including +// the "isolation configured but broken" arm. +func queueResolverApp(t *testing.T, db *sql.DB, rdb *redis.Client, cfg *config.Config) *fiber.App { + t.Helper() + cfg.JWTSecret = testhelpers.TestJWTSecret + cfg.AESKey = testhelpers.TestAESKeyHex + cfg.EnabledServices = "queue" + cfg.Environment = "test" + // The local queue provider health-checks http://:8222/healthz + // before returning a URL; an empty host resolves to localhost, where both + // CI (ci.yml / deploy.yml run nats-server with -m 8222) and the local gate + // have a NATS. Without this the provision 503s before the credential step + // under test is ever reached. + cfg.NATSHost = "" + app := fiber.New(fiber.Config{ + ErrorHandler: func(c *fiber.Ctx, err error) error { + if errors.Is(err, handlers.ErrResponseWritten) { + return nil + } + return c.Status(fiber.StatusInternalServerError). + JSON(fiber.Map{"ok": false, "error": "internal_error", "message": err.Error()}) + }, + }) + app.Use(middleware.RequestID(), middleware.Fingerprint()) + h := handlers.NewQueueHandler(db, rdb, cfg, nil, plans.Default()) + app.Post("/queue/new", middleware.OptionalAuth(cfg), h.NewQueue) + return app +} + +type queueErrBody struct { + OK bool `json:"ok"` + Error string `json:"error"` + Message string `json:"message"` +} + +func postQueueNew(t *testing.T, app *fiber.App, ip, bearer string) (int, queueErrBody) { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/queue/new", strings.NewReader(`{"name":"events"}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Forwarded-For", ip) + if bearer != "" { + req.Header.Set("Authorization", "Bearer "+bearer) + } + resp, err := app.Test(req, 10000) + require.NoError(t, err) + defer resp.Body.Close() + var body queueErrBody + require.NoError(t, json.NewDecoder(resp.Body).Decode(&body)) + return resp.StatusCode, body +} + +// TestQueueNew_IsolationBroken_Returns503_Anonymous — with the operator seed +// configured and the resolver pusher unbuildable, the anonymous path must NOT +// answer 201 with a legacy_open URL (which no client could connect to against +// an auth_required server). It must fail the provision. +func TestQueueNew_IsolationBroken_Returns503_Anonymous(t *testing.T) { + requireTestDB(t) + db, cleanDB := testhelpers.SetupTestDB(t) + defer cleanDB() + rdb, cleanR := testhelpers.SetupTestRedis(t) + defer cleanR() + + restore := handlers.SwapResolverPusherFactoryForTest( + func(natsresolver.Config) (natsqp.ResolverPusher, error) { + return nil, fmt.Errorf("%w: dial tcp: connection refused", natsresolver.ErrPushFailed) + }) + defer restore() + + app := queueResolverApp(t, db, rdb, sysCredCfg(t, "nats")) + status, body := postQueueNew(t, app, "10.71.0.1", "") + + require.Equal(t, http.StatusServiceUnavailable, status) + assert.False(t, body.OK) + assert.Equal(t, "provision_failed", body.Error) + assert.Contains(t, body.Message, "isolated NATS credentials", + "the 503 must name the credential-issuance failure, not a generic provision failure") +} + +// TestQueueNew_IsolationBroken_Returns503_Authenticated — same guarantee on +// the authenticated path (the second issueTenantCreds call site). +func TestQueueNew_IsolationBroken_Returns503_Authenticated(t *testing.T) { + requireTestDB(t) + db, cleanDB := testhelpers.SetupTestDB(t) + defer cleanDB() + rdb, cleanR := testhelpers.SetupTestRedis(t) + defer cleanR() + + restore := handlers.SwapResolverPusherFactoryForTest( + func(natsresolver.Config) (natsqp.ResolverPusher, error) { + return nil, fmt.Errorf("%w: dial tcp: connection refused", natsresolver.ErrPushFailed) + }) + defer restore() + + teamID := testhelpers.MustCreateTeamDB(t, db, "hobby") + sessionJWT := testhelpers.MustSignSessionJWT(t, "user-queue-iso", teamID, "queue-iso@example.com") + + app := queueResolverApp(t, db, rdb, sysCredCfg(t, "nats")) + status, body := postQueueNew(t, app, "10.71.0.2", sessionJWT) + + require.Equal(t, http.StatusServiceUnavailable, status) + assert.Equal(t, "provision_failed", body.Error) + assert.Contains(t, body.Message, "isolated NATS credentials") +} + +// TestFailQueueCredIssue_MarkFailedError_IsLogged drives the mark-failed +// error branch of the shared teardown helper with a closed DB. +func TestFailQueueCredIssue_MarkFailedError_IsLogged(t *testing.T) { + cfg := &config.Config{AESKey: testhelpers.TestAESKeyHex, EnabledServices: "queue"} + h := handlers.NewQueueHandler(closedPlatformDB(t), nil, cfg, nil, plans.Default()) + + app := fiber.New(fiber.Config{ + ErrorHandler: func(c *fiber.Ctx, err error) error { + if errors.Is(err, handlers.ErrResponseWritten) { + return nil + } + return c.SendStatus(fiber.StatusTeapot) + }, + }) + res := &models.Resource{ID: uuid.New()} + app.Post("/fail", func(c *fiber.Ctx) error { + return h.FailQueueCredIssueForTest(c, res, "prid-1", "tok-1", "queue.test", + fmt.Errorf("%w: rejected", natsresolver.ErrPushFailed)) + }) + + resp, err := app.Test(httptest.NewRequest(http.MethodPost, "/fail", nil), 5000) + require.NoError(t, err) + defer resp.Body.Close() + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode, + "a broken isolation path is a 503, never a partially-issued 201") +} diff --git a/internal/natsresolver/pusher.go b/internal/natsresolver/pusher.go new file mode 100644 index 00000000..82467e5b --- /dev/null +++ b/internal/natsresolver/pusher.go @@ -0,0 +1,324 @@ +// Package natsresolver implements the "push the signed account claim to the +// running nats-server" half of NATS operator-mode credential issuance. +// +// # Why this package exists +// +// common/queueprovider/nats mints a per-tenant account NKey, signs an account +// JWT with the operator seed, and then signs a user JWT inside that account. +// None of that reaches the running nats-server on its own: the server only +// learns an account exists when the signed account claim is published to +// $SYS.REQ.CLAIMS.UPDATE by a connection authenticated into the SYSTEM +// account. common/queueprovider/nats abstracts that step behind its +// ResolverPusher interface and defaults it to a no-op, so before this package +// existed every /queue/new credential was cryptographically valid and +// operationally dead — nats-server answered the tenant's CONNECT with +// "Authorization Violation" because it had never heard of the account. +// +// Pusher is that missing implementation. It holds ONE long-lived SYS-account +// connection (established at api boot, auto-reconnecting) and does a bounded +// request/reply per claim, because it is called synchronously on the +// /queue/new request path (CLAUDE.md rule 2 — provisioning is synchronous, a +// backend failure is a 503 and never a half-issued credential). +// +// # Failure semantics +// +// Every error returned by this package wraps ErrPushFailed. Callers use +// errors.Is(err, ErrPushFailed) to distinguish "the isolation path is broken, +// fail the request" from softer credential-issuance problems. A non-ack reply +// from the server is an error exactly like a transport failure is: the account +// is not installed either way, so the credential must never be handed out. +// +// # Secrets +// +// Config.UserJWT and Config.UserSeed are credentials. They are never logged +// and every error string produced here is passed through redact() so a +// library error that echoed the credential back cannot leak it into logs. +package natsresolver + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/nats-io/nats.go" +) + +const ( + // ClaimsUpdateSubject is the nats-server system subject that accepts a + // signed account JWT and installs/updates it in the account resolver. + // Only connections in the system account may publish to it. + ClaimsUpdateSubject = "$SYS.REQ.CLAIMS.UPDATE" + + // ConnectionName labels this connection in `nats server report + // connections` so an operator can tell it apart from tenant traffic. + ConnectionName = "instant-api-resolver-pusher" + + // DefaultPushTimeout bounds a single claims-update request/reply. It is + // deliberately short: /queue/new is synchronous, so a NATS outage must + // surface as a fast 503 rather than a hung request. + DefaultPushTimeout = 5 * time.Second + + // DefaultDialTimeout bounds the boot-time connect attempt. + DefaultDialTimeout = 5 * time.Second + + // reconnectWait is the pause between reconnect attempts after the + // long-lived connection drops. + reconnectWait = 2 * time.Second + + // reconnectForever keeps the client retrying for the life of the + // process (nats.go treats a negative value as unlimited). + reconnectForever = -1 + + // redactedPlaceholder replaces any credential material found in an + // error string before it reaches a log line. + redactedPlaceholder = "[redacted]" + + // minRedactLen guards against replacing incidental short strings; real + // JWTs and NKey seeds are far longer than this. + minRedactLen = 8 +) + +// ErrPushFailed is the sentinel wrapped by every error this package returns. +// A caller seeing errors.Is(err, ErrPushFailed) knows the tenant's account +// claim did NOT reach the resolver, so any credential minted alongside it is +// dead and must not be returned to the customer. +var ErrPushFailed = errors.New("nats resolver: account claim push failed") + +// Requester is the minimal slice of *nats.Conn this package needs. It exists +// so tests can drive the ack / non-ack / timeout paths without a real server. +type Requester interface { + RequestWithContext(ctx context.Context, subj string, data []byte) (*nats.Msg, error) + Close() +} + +// connectFn is the dial seam. Production aliases nats.Connect 1:1; tests +// substitute a stub. Package-level var mirrors the test-seam convention in +// common/queueprovider/nats. +var connectFn = dialNATS + +// dialNATS returns nats.Connect's result pair unchanged. Callers MUST check +// the error before touching the Requester: a failed nats.Connect yields a nil +// *nats.Conn boxed in a non-nil interface value. +func dialNATS(url string, opts ...nats.Option) (Requester, error) { + return nats.Connect(url, opts...) +} + +// Config describes the system-account connection used to push claims. +type Config struct { + // URL is the server URL to connect to, e.g. nats://nats.instant-data.svc.cluster.local:4222. + URL string + + // UserJWT is the SYS-account user JWT. SECRET — never logged. + UserJWT string + + // UserSeed is the NKey seed matching UserJWT. SECRET — never logged. + UserSeed string + + // Timeout bounds one claims-update request/reply. Zero → DefaultPushTimeout. + Timeout time.Duration +} + +// Pusher is a long-lived system-account connection that installs account +// claims in the resolver. Safe for concurrent use — *nats.Conn is. +type Pusher struct { + conn Requester + timeout time.Duration + // secrets are scrubbed out of every error string this Pusher produces. + secrets []string +} + +// New dials NATS as the system user and returns a ready Pusher. It fails +// loudly: a missing credential or an unreachable server is an error, never a +// silently degraded no-op pusher — a no-op pusher is precisely what made +// /queue/new hand out unusable credentials. +func New(cfg Config) (*Pusher, error) { + if strings.TrimSpace(cfg.URL) == "" { + return nil, fmt.Errorf("%w: no NATS URL configured", ErrPushFailed) + } + if strings.TrimSpace(cfg.UserJWT) == "" { + return nil, fmt.Errorf("%w: no system-account user JWT configured", ErrPushFailed) + } + if strings.TrimSpace(cfg.UserSeed) == "" { + return nil, fmt.Errorf("%w: no system-account user NKey seed configured", ErrPushFailed) + } + timeout := cfg.Timeout + if timeout <= 0 { + timeout = DefaultPushTimeout + } + conn, err := connectFn(cfg.URL, natsOptions(cfg)...) + if err != nil { + // %v (not %w) on the library error: it is rendered through redact + // first so no credential can survive into the error chain. + return nil, fmt.Errorf("%w: connect to %s as system user: %v", + ErrPushFailed, cfg.URL, redact(err.Error(), cfg.UserJWT, cfg.UserSeed)) + } + slog.Info("nats.resolver_pusher_connected", + "url", cfg.URL, + "subject", ClaimsUpdateSubject, + "push_timeout", timeout.String()) + return &Pusher{ + conn: conn, + timeout: timeout, + secrets: []string{cfg.UserJWT, cfg.UserSeed}, + }, nil +} + +// natsOptions builds the connection options: named connection, SYS-account +// credentials, bounded dial, and unlimited reconnect so a NATS restart does +// not permanently break queue provisioning. +func natsOptions(cfg Config) []nats.Option { + return []nats.Option{ + nats.Name(ConnectionName), + nats.UserJWTAndSeed(cfg.UserJWT, cfg.UserSeed), + nats.Timeout(DefaultDialTimeout), + nats.MaxReconnects(reconnectForever), + nats.ReconnectWait(reconnectWait), + nats.DisconnectErrHandler(onDisconnect), + nats.ReconnectHandler(onReconnect), + } +} + +// onDisconnect / onReconnect log connection-lifecycle transitions. They +// deliberately ignore the *nats.Conn argument: the only fields worth logging +// are static, and touching the connection from a callback risks a nil deref +// during shutdown. +func onDisconnect(_ *nats.Conn, err error) { + slog.Warn("nats.resolver_pusher_disconnected", + "connection", ConnectionName, + "error", errString(err)) +} + +func onReconnect(_ *nats.Conn) { + slog.Info("nats.resolver_pusher_reconnected", "connection", ConnectionName) +} + +// PushAccountClaim publishes the signed account JWT to the resolver and +// requires a positive acknowledgement. Implements +// queueprovider/nats.ResolverPusher. +func (p *Pusher) PushAccountClaim(ctx context.Context, accountPublicKey, accountJWT string) error { + if p == nil || p.conn == nil { + return fmt.Errorf("%w: pusher is not initialised", ErrPushFailed) + } + if accountJWT == "" { + return fmt.Errorf("%w: empty account JWT for account %s", ErrPushFailed, accountPublicKey) + } + if ctx == nil { + ctx = context.Background() + } + ctx, cancel := context.WithTimeout(ctx, p.timeout) + defer cancel() + + msg, err := p.conn.RequestWithContext(ctx, ClaimsUpdateSubject, []byte(accountJWT)) + if err != nil { + return fmt.Errorf("%w: %s request for account %s: %v", + ErrPushFailed, ClaimsUpdateSubject, accountPublicKey, p.redact(err.Error())) + } + return verifyAck(accountPublicKey, msg) +} + +// Close tears down the long-lived connection. Safe on a nil/zero Pusher. +func (p *Pusher) Close() { + if p == nil || p.conn == nil { + return + } + p.conn.Close() +} + +// claimsUpdateReply models the nats-server $SYS response envelope. +// +// nats-server's respondToUpdate() answers with +// +// {"server":{...},"data":{"account":"A…","code":200,"message":"jwt updated"}} +// +// on success and the same envelope with code 500 + "description" on failure. +// The top-level "error" member is used by other $SYS endpoints; it is parsed +// too so a server-version difference surfaces as a rejection rather than +// being mistaken for an ack. +type claimsUpdateReply struct { + Data *claimsUpdateData `json:"data"` + Error *claimsUpdateError `json:"error"` +} + +type claimsUpdateData struct { + Account string `json:"account"` + Code int `json:"code"` + Message string `json:"message"` + Description string `json:"description"` +} + +type claimsUpdateError struct { + Code int `json:"code"` + Description string `json:"description"` +} + +// verifyAck turns the resolver reply into an error unless it is an +// unambiguous success. Anything unrecognised is treated as a rejection: a +// claim that may not have been installed must never be reported as installed. +func verifyAck(accountPublicKey string, msg *nats.Msg) error { + if msg == nil || len(msg.Data) == 0 { + return fmt.Errorf("%w: empty reply from resolver for account %s", + ErrPushFailed, accountPublicKey) + } + var reply claimsUpdateReply + if err := json.Unmarshal(msg.Data, &reply); err != nil { + return fmt.Errorf("%w: unparseable resolver reply for account %s: %v", + ErrPushFailed, accountPublicKey, err) + } + if reply.Error != nil { + return fmt.Errorf("%w: resolver returned an error for account %s: code=%d %s", + ErrPushFailed, accountPublicKey, reply.Error.Code, reply.Error.Description) + } + if reply.Data == nil { + return fmt.Errorf("%w: resolver reply for account %s carried no data envelope", + ErrPushFailed, accountPublicKey) + } + if reply.Data.Code < http.StatusOK || reply.Data.Code >= http.StatusMultipleChoices { + return fmt.Errorf("%w: resolver did not ack account %s: code=%d %s", + ErrPushFailed, accountPublicKey, reply.Data.Code, ackDetail(reply.Data)) + } + // A reply naming a different account means we cannot claim THIS account + // was installed. + if reply.Data.Account != "" && reply.Data.Account != accountPublicKey { + return fmt.Errorf("%w: resolver acked a different account (want %s, got %s)", + ErrPushFailed, accountPublicKey, reply.Data.Account) + } + return nil +} + +// ackDetail picks whichever human-readable field the server populated. +func ackDetail(d *claimsUpdateData) string { + if d.Description != "" { + return d.Description + } + return d.Message +} + +// redact scrubs this Pusher's credentials out of a string bound for a log or +// an error. +func (p *Pusher) redact(s string) string { + return redact(s, p.secrets...) +} + +func redact(s string, secrets ...string) string { + for _, secret := range secrets { + if len(secret) < minRedactLen { + continue + } + s = strings.ReplaceAll(s, secret, redactedPlaceholder) + } + return s +} + +// errString renders an error for a log field without a nil check at every +// call site. +func errString(err error) string { + if err == nil { + return "" + } + return err.Error() +} diff --git a/internal/natsresolver/pusher_test.go b/internal/natsresolver/pusher_test.go new file mode 100644 index 00000000..2a737e99 --- /dev/null +++ b/internal/natsresolver/pusher_test.go @@ -0,0 +1,440 @@ +package natsresolver + +// pusher_test.go — in-package tests for the SYS-account resolver pusher. +// +// In-package (not natsresolver_test) so the connectFn dial seam can be +// substituted: every test here runs without a NATS server, which is the point +// — the ack / non-ack / timeout arms are exactly the ones a live-server test +// cannot force on demand. + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + testAccountPub = "ACCOUNTPUBLICKEY123456789" + testAccountJWT = "eyJ0eXAiOiJKV1QifQ.account.claim" + testSysJWT = "eyJ0eXAiOiJKV1QifQ.system.user" + testSysSeed = "SUAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" +) + +// fakeRequester is a programmable stand-in for *nats.Conn. +type fakeRequester struct { + mu sync.Mutex + reply *nats.Msg + err error + block bool // block until the context deadline fires + gotSubj string + gotData []byte + calls int + closed bool +} + +func (f *fakeRequester) RequestWithContext(ctx context.Context, subj string, data []byte) (*nats.Msg, error) { + f.mu.Lock() + f.calls++ + f.gotSubj = subj + f.gotData = data + block, reply, err := f.block, f.reply, f.err + f.mu.Unlock() + + if block { + <-ctx.Done() + return nil, ctx.Err() + } + return reply, err +} + +func (f *fakeRequester) Close() { + f.mu.Lock() + defer f.mu.Unlock() + f.closed = true +} + +func (f *fakeRequester) snapshot() (subj string, data []byte, calls int, closed bool) { + f.mu.Lock() + defer f.mu.Unlock() + return f.gotSubj, f.gotData, f.calls, f.closed +} + +// withConnect swaps the dial seam for the duration of a test. +func withConnect(t *testing.T, fn func(url string, opts ...nats.Option) (Requester, error)) { + t.Helper() + prev := connectFn + connectFn = fn + t.Cleanup(func() { connectFn = prev }) +} + +func msgOf(payload string) *nats.Msg { + return &nats.Msg{Data: []byte(payload)} +} + +func validConfig() Config { + return Config{URL: "nats://nats.test:4222", UserJWT: testSysJWT, UserSeed: testSysSeed} +} + +// ── New ────────────────────────────────────────────────────────────────────── + +// TestNew_ValidationAndDial covers every constructor arm: each missing +// credential, an unreachable server, and the two success shapes (default and +// explicit push timeout). The "pusher constructed / not constructed" split +// lives here. +func TestNew_ValidationAndDial(t *testing.T) { + dialErr := errors.New("nats: no servers available for connection, seed " + testSysSeed) + + tests := []struct { + name string + cfg Config + dial func(url string, opts ...nats.Option) (Requester, error) + wantErr bool + wantErrHas string + wantTimeout time.Duration + }{ + { + name: "missing url", + cfg: Config{UserJWT: testSysJWT, UserSeed: testSysSeed}, + wantErr: true, + wantErrHas: "no NATS URL configured", + }, + { + name: "missing user jwt", + cfg: Config{URL: "nats://nats.test:4222", UserSeed: testSysSeed}, + wantErr: true, + wantErrHas: "no system-account user JWT configured", + }, + { + name: "blank user seed", + cfg: Config{URL: "nats://nats.test:4222", UserJWT: testSysJWT, UserSeed: " "}, + wantErr: true, + wantErrHas: "no system-account user NKey seed configured", + }, + { + name: "dial failure surfaces as an error, never a no-op pusher", + cfg: validConfig(), + dial: func(string, ...nats.Option) (Requester, error) { + return nil, dialErr + }, + wantErr: true, + wantErrHas: "connect to nats://nats.test:4222 as system user", + }, + { + name: "constructed with default timeout", + cfg: validConfig(), + wantTimeout: DefaultPushTimeout, + }, + { + name: "constructed with explicit timeout", + cfg: Config{ + URL: "nats://nats.test:4222", UserJWT: testSysJWT, + UserSeed: testSysSeed, Timeout: 250 * time.Millisecond, + }, + wantTimeout: 250 * time.Millisecond, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var gotURL string + var gotOpts int + dial := tc.dial + if dial == nil { + dial = func(url string, opts ...nats.Option) (Requester, error) { + gotURL, gotOpts = url, len(opts) + return &fakeRequester{}, nil + } + } + withConnect(t, dial) + + p, err := New(tc.cfg) + if tc.wantErr { + require.Error(t, err) + assert.Nil(t, p, "a failed New must not return a usable pusher") + assert.ErrorIs(t, err, ErrPushFailed, "every failure wraps the sentinel") + assert.Contains(t, err.Error(), tc.wantErrHas) + assert.NotContains(t, err.Error(), testSysSeed, "the NKey seed must never reach an error string") + return + } + require.NoError(t, err) + require.NotNil(t, p) + assert.Equal(t, tc.wantTimeout, p.timeout) + assert.Equal(t, tc.cfg.URL, gotURL) + assert.Positive(t, gotOpts, "connection options (creds, reconnect, handlers) must be passed") + assert.Equal(t, []string{testSysJWT, testSysSeed}, p.secrets) + }) + } +} + +// TestNatsOptions_CarriesIdentityAndReconnect asserts the option set is the +// production one — a real nats.Options is materialised from it so a renamed +// or dropped option fails here rather than at 3am in the cluster. +func TestNatsOptions_CarriesIdentityAndReconnect(t *testing.T) { + var opts nats.Options + for _, apply := range natsOptions(validConfig()) { + require.NoError(t, apply(&opts)) + } + assert.Equal(t, ConnectionName, opts.Name) + assert.Equal(t, DefaultDialTimeout, opts.Timeout) + assert.Equal(t, reconnectForever, opts.MaxReconnect) + assert.Equal(t, reconnectWait, opts.ReconnectWait) + assert.NotNil(t, opts.DisconnectedErrCB) + assert.NotNil(t, opts.ReconnectedCB) + assert.NotEmpty(t, opts.UserJWT, "SYS user JWT callback must be installed") +} + +// TestLifecycleHandlers_DoNotPanic — the reconnect callbacks run on a +// nats.go goroutine; they must never touch the connection. +func TestLifecycleHandlers_DoNotPanic(t *testing.T) { + assert.NotPanics(t, func() { onDisconnect(nil, errors.New("connection reset")) }) + assert.NotPanics(t, func() { onDisconnect(nil, nil) }) + assert.NotPanics(t, func() { onReconnect(nil) }) +} + +// ── PushAccountClaim ───────────────────────────────────────────────────────── + +// TestPushAccountClaim_ReplyHandling is the core table: an ack is the ONLY +// outcome that returns nil. Every other reply shape must error, because a +// claim that may not be installed must never be reported as installed. +func TestPushAccountClaim_ReplyHandling(t *testing.T) { + tests := []struct { + name string + reply *nats.Msg + reqErr error + wantErr bool + wantErrHas string + }{ + { + name: "ack — code 200 with matching account", + reply: msgOf(`{"server":{"name":"n1"},"data":{"account":"` + testAccountPub + `","code":200,"message":"jwt updated"}}`), + }, + { + name: "ack — 2xx with no account echoed back", + reply: msgOf(`{"data":{"code":204,"message":"jwt updated"}}`), + }, + { + name: "non-ack — server-side 500 in the data envelope", + reply: msgOf(`{"data":{"account":"` + testAccountPub + `","code":500,"description":"jwt update skipped - memory resolver"}}`), + wantErr: true, + wantErrHas: "resolver did not ack", + }, + { + name: "non-ack — 4xx with only a message field", + reply: msgOf(`{"data":{"code":400,"message":"bad claim"}}`), + wantErr: true, + wantErrHas: "bad claim", + }, + { + name: "non-ack — top-level error envelope", + reply: msgOf(`{"error":{"code":401,"description":"permissions violation"}}`), + wantErr: true, + wantErrHas: "resolver returned an error", + }, + { + name: "non-ack — no data envelope at all", + reply: msgOf(`{"server":{"name":"n1"}}`), + wantErr: true, + wantErrHas: "carried no data envelope", + }, + { + name: "non-ack — reply names a different account", + reply: msgOf(`{"data":{"account":"AOTHERACCOUNT","code":200,"message":"jwt updated"}}`), + wantErr: true, + wantErrHas: "acked a different account", + }, + { + name: "non-ack — unparseable reply", + reply: msgOf(`not-json`), + wantErr: true, + wantErrHas: "unparseable resolver reply", + }, + { + name: "non-ack — empty reply body", + reply: msgOf(``), + wantErr: true, + wantErrHas: "empty reply from resolver", + }, + { + name: "non-ack — nil message", + reply: nil, + wantErr: true, + wantErrHas: "empty reply from resolver", + }, + { + name: "transport error is redacted", + reqErr: errors.New("write failed for user " + testSysJWT), + wantErr: true, + wantErrHas: ClaimsUpdateSubject, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + fake := &fakeRequester{reply: tc.reply, err: tc.reqErr} + p := &Pusher{conn: fake, timeout: time.Second, secrets: []string{testSysJWT, testSysSeed}} + + err := p.PushAccountClaim(context.Background(), testAccountPub, testAccountJWT) + + subj, data, calls, _ := fake.snapshot() + assert.Equal(t, ClaimsUpdateSubject, subj) + assert.Equal(t, testAccountJWT, string(data), "the raw account JWT is the request payload") + assert.Equal(t, 1, calls) + + if !tc.wantErr { + require.NoError(t, err) + return + } + require.Error(t, err) + assert.ErrorIs(t, err, ErrPushFailed) + assert.Contains(t, err.Error(), tc.wantErrHas) + assert.NotContains(t, err.Error(), testSysJWT, "credentials must be redacted from error strings") + }) + } +} + +// TestPushAccountClaim_Timeout — a NATS server that never replies must fail +// inside the push budget, so /queue/new can return 503 instead of hanging. +func TestPushAccountClaim_Timeout(t *testing.T) { + fake := &fakeRequester{block: true} + p := &Pusher{conn: fake, timeout: 30 * time.Millisecond} + + start := time.Now() + err := p.PushAccountClaim(context.Background(), testAccountPub, testAccountJWT) + elapsed := time.Since(start) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrPushFailed) + assert.Contains(t, err.Error(), context.DeadlineExceeded.Error()) + assert.Less(t, elapsed, 2*time.Second, "the push must be bounded by Pusher.timeout") +} + +// TestPushAccountClaim_Guards covers the arms that never reach the wire. +func TestPushAccountClaim_Guards(t *testing.T) { + t.Run("nil pusher", func(t *testing.T) { + var p *Pusher + err := p.PushAccountClaim(context.Background(), testAccountPub, testAccountJWT) + require.Error(t, err) + assert.ErrorIs(t, err, ErrPushFailed) + assert.Contains(t, err.Error(), "not initialised") + }) + + t.Run("pusher with no connection", func(t *testing.T) { + err := (&Pusher{}).PushAccountClaim(context.Background(), testAccountPub, testAccountJWT) + require.Error(t, err) + assert.ErrorIs(t, err, ErrPushFailed) + }) + + t.Run("empty account jwt", func(t *testing.T) { + fake := &fakeRequester{} + p := &Pusher{conn: fake, timeout: time.Second} + err := p.PushAccountClaim(context.Background(), testAccountPub, "") + require.Error(t, err) + assert.ErrorIs(t, err, ErrPushFailed) + _, _, calls, _ := fake.snapshot() + assert.Zero(t, calls, "an empty claim must not be published") + }) + + t.Run("nil context is tolerated", func(t *testing.T) { + // A nil ctx would panic inside context.WithTimeout; the guard turns + // it into a background context. Assigned to a variable so staticcheck + // does not flag the literal-nil call. + var nilCtx context.Context + fake := &fakeRequester{reply: msgOf(`{"data":{"code":200,"message":"jwt updated"}}`)} + p := &Pusher{conn: fake, timeout: time.Second} + require.NoError(t, p.PushAccountClaim(nilCtx, testAccountPub, testAccountJWT)) + }) + + t.Run("caller cancellation propagates", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + p := &Pusher{conn: &fakeRequester{block: true}, timeout: time.Minute} + err := p.PushAccountClaim(ctx, testAccountPub, testAccountJWT) + require.Error(t, err) + assert.ErrorIs(t, err, ErrPushFailed) + }) +} + +// ── Close / helpers ────────────────────────────────────────────────────────── + +func TestClose(t *testing.T) { + t.Run("nil pusher is a no-op", func(t *testing.T) { + var p *Pusher + assert.NotPanics(t, p.Close) + }) + t.Run("pusher with no connection is a no-op", func(t *testing.T) { + assert.NotPanics(t, (&Pusher{}).Close) + }) + t.Run("closes the underlying connection", func(t *testing.T) { + fake := &fakeRequester{} + (&Pusher{conn: fake}).Close() + _, _, _, closed := fake.snapshot() + assert.True(t, closed) + }) +} + +func TestRedact(t *testing.T) { + tests := []struct { + name string + in string + secrets []string + want string + }{ + { + name: "replaces every occurrence", + in: "auth error for " + testSysSeed + " (" + testSysSeed + ")", + secrets: []string{testSysSeed}, + want: "auth error for " + redactedPlaceholder + " (" + redactedPlaceholder + ")", + }, + { + name: "short secrets are skipped so common substrings survive", + in: "connection refused", + secrets: []string{"con"}, + want: "connection refused", + }, + { + name: "no secrets configured", + in: "plain error", + want: "plain error", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, redact(tc.in, tc.secrets...)) + }) + } + + t.Run("method form uses the pusher's own secrets", func(t *testing.T) { + p := &Pusher{secrets: []string{testSysJWT}} + assert.Equal(t, "boom "+redactedPlaceholder, p.redact("boom "+testSysJWT)) + }) +} + +func TestAckDetail(t *testing.T) { + assert.Equal(t, "why it failed", ackDetail(&claimsUpdateData{Description: "why it failed", Message: "m"})) + assert.Equal(t, "jwt updated", ackDetail(&claimsUpdateData{Message: "jwt updated"})) +} + +func TestErrString(t *testing.T) { + assert.Empty(t, errString(nil)) + assert.Equal(t, "boom", errString(errors.New("boom"))) +} + +// TestConnectFn_DefaultDialsRealNATS exercises the production dial seam +// itself (the one every other test replaces) against an address with no +// listener: it must return an error and no Requester, never a half-built +// connection. Keeps the seam's own two lines honest. +func TestConnectFn_DefaultDialsRealNATS(t *testing.T) { + conn, err := connectFn("nats://127.0.0.1:1", nats.Timeout(200*time.Millisecond), nats.MaxReconnects(0)) + require.Error(t, err) + assert.Nil(t, conn) + assert.True(t, + strings.Contains(err.Error(), "connection refused") || strings.Contains(err.Error(), "no servers"), + "unexpected dial error: %v", err) +}