From 162ded3438f45b94a2195ff7996989ee7036b3c9 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Mon, 3 Aug 2026 23:39:06 +0800 Subject: [PATCH] fix: register NATS/AMQP notify config keys read by parsers GetNotifyNATS reads user_credentials, nkey_seed and tls_handshake_first and GetNotifyAMQP reads immediate, but none of them were registered in DefaultNATSKVS/DefaultAMQPKVS or the help schema, so CheckValidKeys rejected any enable=on target carrying them. Worse, the legacy config migration wrote exactly these keys - including the env var name MINIO_NOTIFY_NATS_USER_CREDENTIALS used as a config key, because the NATSUserCredentials constant doubled as both - so a migrated NATS config failed validation on every load, and the FetchEnabledTargets fail-fast then silently disabled all bucket notification targets. - Register user_credentials/nkey_seed/tls_handshake_first (NATS) and immediate (AMQP) in the default KVS and help schema; split NATSUserCredentials into a real config key plus EnvNATSUserCredentials (all env var names byte-stable) - Fix legacy migration: SetNotifyNATS writes the proper key; SetNotifyAMQP no longer writes cfg.Immediate under the internal key and now carries both immediate and internal - Tolerate the legacy MINIO_NOTIFY_NATS_USER_CREDENTIALS key written by pre-fix migrations (NATS-scoped, load path only) with fallback read; env > user_credentials > legacy key - Print key names only, never values, in the invalid-keys error of both CheckValidKeys forms; rejected values can carry credentials - Add an AST-based audit test asserting parser reads, migration writes and help entries stay within the registered key set for all ten notify subsystems, with floor assertions so collector drift fails loudly - Document (unchanged) FetchEnabledTargets fail-fast and pin it with a characterization test Known same-class gap left in place and pinned by the audit's allowlist: SetNotifyPostgres/SetNotifyMySQL write five unregistered DSN-era keys; tracked for a follow-up issue. Closes #39 Co-authored-by: ChatGPT Co-authored-by: Claude --- internal/config/config.go | 17 +- internal/config/config_test.go | 54 ++ internal/config/notify/help.go | 26 + internal/config/notify/legacy.go | 6 +- internal/config/notify/legacy_test.go | 115 ++++ internal/config/notify/parse.go | 58 +- internal/config/notify/parse_test.go | 757 ++++++++++++++++++++++++++ internal/event/target/nats.go | 3 +- 8 files changed, 1029 insertions(+), 7 deletions(-) create mode 100644 internal/config/notify/legacy_test.go create mode 100644 internal/config/notify/parse_test.go diff --git a/internal/config/config.go b/internal/config/config.go index 20a9f3024..c254d50d1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -624,6 +624,19 @@ func LookupSite(siteKV KVS, regionKV KVS) (s Site, err error) { return s, err } +// invalidKeyNames returns a comma separated list of the key names in kvs. +// +// Only names are returned, never values: this list is embedded in errors that +// are written to the server log and printed by `mc`, and a rejected key may +// well be carrying a credential. +func invalidKeyNames(kvs KVS) string { + names := make([]string, 0, len(kvs)) + for _, kv := range kvs { + names = append(names, kv.Key) + } + return strings.Join(names, ", ") +} + // CheckValidKeys - checks if inputs KVS has the necessary keys, // returns error if it find extra or superfluous keys. func CheckValidKeys(subSys string, kv KVS, validKVS KVS, deprecatedKeys ...string) error { @@ -648,7 +661,7 @@ func CheckValidKeys(subSys string, kv KVS, validKVS KVS, deprecatedKeys ...strin } if len(nkv) > 0 { return Errorf( - "found invalid keys (%s) for '%s' sub-system, use 'mc admin config reset myminio %s' to fix invalid keys", nkv.String(), subSys, subSys) + "found invalid keys (%s) for '%s' sub-system, use 'mc admin config reset myminio %s' to fix invalid keys", invalidKeyNames(nkv), subSys, subSys) } return nil } @@ -1093,7 +1106,7 @@ func (c Config) CheckValidKeys(subSys string, deprecatedKeys []string) error { if len(invalidKV) > 0 { return Errorf( "found invalid keys (%s) for '%s:%s' sub-system, use 'mc admin config reset myminio %s:%s' to fix invalid keys", - invalidKV.String(), subSys, tgt, subSys, tgt) + invalidKeyNames(invalidKV), subSys, tgt, subSys, tgt) } } return nil diff --git a/internal/config/config_test.go b/internal/config/config_test.go index e55d446b3..5034cd20b 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -18,6 +18,7 @@ package config import ( + "strings" "testing" ) @@ -128,3 +129,56 @@ func TestValidRegion(t *testing.T) { }) } } + +// The invalid-keys error is logged by the server and printed by `mc`. A +// rejected key may carry a credential, so only key names may appear in it. +func TestCheckValidKeysDoesNotLeakValues(t *testing.T) { + const ( + badKey = "unknown_key" + badSecret = "s3cr3t-must-not-appear" + ) + + assertRedacted := func(t *testing.T, err error) { + t.Helper() + if err == nil { + t.Fatal("expected an error for an unregistered key") + } + msg := err.Error() + if strings.Contains(msg, badSecret) { + t.Errorf("error leaks the rejected value: %s", msg) + } + if !strings.Contains(msg, badKey) { + t.Errorf("error does not name the rejected key: %s", msg) + } + if !strings.Contains(msg, "mc admin config reset") { + t.Errorf("error lost the remediation hint: %s", msg) + } + } + + t.Run("func", func(t *testing.T) { + kv := KVS{ + KV{Key: Enable, Value: EnableOn}, + KV{Key: badKey, Value: badSecret}, + } + validKVS := KVS{KV{Key: Enable, Value: EnableOff}} + assertRedacted(t, CheckValidKeys("test_subsys", kv, validKVS)) + }) + + t.Run("method", func(t *testing.T) { + const subSys = "test_subsys_method" + RegisterDefaultKVS(map[string]KVS{ + subSys: {KV{Key: Enable, Value: EnableOff}}, + }) + t.Cleanup(func() { delete(DefaultKVS, subSys) }) + + c := Config{ + subSys: map[string]KVS{ + Default: { + KV{Key: Enable, Value: EnableOn}, + KV{Key: badKey, Value: badSecret}, + }, + }, + } + assertRedacted(t, c.CheckValidKeys(subSys, nil)) + }) +} diff --git a/internal/config/notify/help.go b/internal/config/notify/help.go index 343f46c7b..5836b1c6e 100644 --- a/internal/config/notify/help.go +++ b/internal/config/notify/help.go @@ -111,6 +111,12 @@ var ( Optional: true, Type: "on|off", }, + config.HelpKV{ + Key: target.AmqpImmediate, + Description: "return messages that cannot be delivered to a consumer straight away when set to 'on', default is 'off'", + Optional: true, + Type: "on|off", + }, config.HelpKV{ Key: target.AmqpDurable, Description: "persist queue across broker restarts when set to 'on', default is 'off'", @@ -459,6 +465,13 @@ var ( Type: "string", Sensitive: true, }, + config.HelpKV{ + Key: target.NATSUserCredentials, + Description: "path to NATS user credentials (.creds) file for JWT auth", + Optional: true, + Type: "string", + Sensitive: true, + }, config.HelpKV{ Key: target.NATSPassword, Description: "NATS password", @@ -475,6 +488,13 @@ var ( Sensitive: true, Secret: true, }, + config.HelpKV{ + Key: target.NATSNKeySeed, + Description: "path to NATS NKey seed file", + Optional: true, + Type: "string", + Sensitive: true, + }, config.HelpKV{ Key: target.NATSTLS, Description: "set to 'on' to enable TLS", @@ -487,6 +507,12 @@ var ( Optional: true, Type: "on|off", }, + config.HelpKV{ + Key: target.NATSTLSHandshakeFirst, + Description: "set to 'on' to perform TLS handshake before waiting for server INFO", + Optional: true, + Type: "on|off", + }, config.HelpKV{ Key: target.NATSPingInterval, Description: "client ping commands interval in s,m,h,d. Disabled by default", diff --git a/internal/config/notify/legacy.go b/internal/config/notify/legacy.go index c72aff126..d2a1d67f3 100644 --- a/internal/config/notify/legacy.go +++ b/internal/config/notify/legacy.go @@ -147,9 +147,13 @@ func SetNotifyAMQP(s config.Config, amqpName string, cfg target.AMQPArgs) error Value: config.FormatBool(cfg.Mandatory), }, config.KV{ - Key: target.AmqpInternal, + Key: target.AmqpImmediate, Value: config.FormatBool(cfg.Immediate), }, + config.KV{ + Key: target.AmqpInternal, + Value: config.FormatBool(cfg.Internal), + }, config.KV{ Key: target.AmqpDurable, Value: config.FormatBool(cfg.Durable), diff --git a/internal/config/notify/legacy_test.go b/internal/config/notify/legacy_test.go new file mode 100644 index 000000000..50f298d7f --- /dev/null +++ b/internal/config/notify/legacy_test.go @@ -0,0 +1,115 @@ +// Copyright (c) 2015-2026 MinIO, Inc. +// +// This file is part of MinIO Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package notify + +import ( + "testing" + + "github.com/minio/minio/internal/config" + "github.com/minio/minio/internal/event/target" + xnet "github.com/minio/pkg/v3/net" + "github.com/rabbitmq/amqp091-go" +) + +// T5 (NATS): a config produced by the legacy migration must survive validation +// and round-trip back through the parser unchanged. Before the fix the +// migration wrote an env var name as a config key, so every migrated NATS +// target failed CheckValidKeys on the next config load. +func TestSetNotifyNATSRoundTrip(t *testing.T) { + addr, err := xnet.ParseHost(testNATSAddr) + if err != nil { + t.Fatalf("ParseHost: %v", err) + } + + args := target.NATSArgs{ + Enable: true, + Address: *addr, + Subject: testNATSSubj, + UserCredentials: testCredsPath, + NKeySeed: testNKeyPath, + TLSHandshakeFirst: true, + } + + s := config.Config{config.NotifyNATSSubSys: map[string]config.KVS{}} + if err := SetNotifyNATS(s, testTargetName, args); err != nil { + t.Fatalf("SetNotifyNATS: %v", err) + } + + if err := checkValidNotificationKeysForSubSys(config.NotifyNATSSubSys, s[config.NotifyNATSSubSys]); err != nil { + t.Fatalf("migrated NATS config must pass key validation, got: %v", err) + } + + targets, err := GetNotifyNATS(s[config.NotifyNATSSubSys], nil) + if err != nil { + t.Fatalf("GetNotifyNATS: %v", err) + } + got, ok := targets[testTargetName] + if !ok { + t.Fatalf("target %q missing after round trip: %v", testTargetName, targets) + } + if got.UserCredentials != args.UserCredentials { + t.Errorf("UserCredentials = %q, want %q", got.UserCredentials, args.UserCredentials) + } + if got.NKeySeed != args.NKeySeed { + t.Errorf("NKeySeed = %q, want %q", got.NKeySeed, args.NKeySeed) + } + if got.TLSHandshakeFirst != args.TLSHandshakeFirst { + t.Errorf("TLSHandshakeFirst = %v, want %v", got.TLSHandshakeFirst, args.TLSHandshakeFirst) + } +} + +// T5 (AMQP): the migration mapped cfg.Immediate onto the `internal` key and +// dropped cfg.Internal entirely, so a migrated target came back with both +// fields wrong. +func TestSetNotifyAMQPRoundTrip(t *testing.T) { + uri, err := amqp091.ParseURI(testAMQPURL) + if err != nil { + t.Fatalf("ParseURI: %v", err) + } + + args := target.AMQPArgs{ + Enable: true, + URL: uri, + Immediate: true, + Internal: false, + } + + s := config.Config{config.NotifyAMQPSubSys: map[string]config.KVS{}} + if err := SetNotifyAMQP(s, testTargetName, args); err != nil { + t.Fatalf("SetNotifyAMQP: %v", err) + } + + if err := checkValidNotificationKeysForSubSys(config.NotifyAMQPSubSys, s[config.NotifyAMQPSubSys]); err != nil { + t.Fatalf("migrated AMQP config must pass key validation, got: %v", err) + } + + targets, err := GetNotifyAMQP(s[config.NotifyAMQPSubSys]) + if err != nil { + t.Fatalf("GetNotifyAMQP: %v", err) + } + got, ok := targets[testTargetName] + if !ok { + t.Fatalf("target %q missing after round trip: %v", testTargetName, targets) + } + if !got.Immediate { + t.Errorf("Immediate = false, want true") + } + if got.Internal { + t.Errorf("Internal = true, want false (immediate must not be written to the internal key)") + } +} diff --git a/internal/config/notify/parse.go b/internal/config/notify/parse.go index b479d0d4d..c4c7f7c93 100644 --- a/internal/config/notify/parse.go +++ b/internal/config/notify/parse.go @@ -255,6 +255,14 @@ func fetchSubSysTargets(ctx context.Context, cfg config.Config, subSys string, t } // FetchEnabledTargets - Returns a set of configured TargetList +// +// This fails fast: the first sub-system that fails to validate or parse aborts +// the whole call and no target list is returned. A single malformed notify +// sub-system therefore disables bucket notifications for every other target as +// well, since the caller only logs the error and leaves the global target list +// nil. That is the long-standing behavior and is kept deliberately; changing +// it to skip only the broken sub-system would silently degrade a config that +// operators currently expect to fail loudly. func FetchEnabledTargets(ctx context.Context, cfg config.Config, transport *http.Transport) (_ *event.TargetList, err error) { targetList := event.NewTargetList(ctx) for _, subSys := range config.NotifySubSystems.ToSlice() { @@ -287,18 +295,37 @@ var ( } ) +// legacyNATSUserCredentialsKey is the NATS user credentials env var name, which +// pre-fix migration code wrote into the config store as if it were a config +// key. It is tolerated so that already-migrated stores keep loading; the proper +// user_credentials key takes precedence when both are present. +// +// Spelled as a literal on purpose: it names what is already written on disk, so +// it must not follow any later rename of target.EnvNATSUserCredentials. +// +// The tolerance below relies on config.CheckValidKeys, the free function, whose +// variadic deprecatedKeys means "accept these anyway". The same-named method +// config.Config.CheckValidKeys takes deprecatedKeys with the opposite meaning: +// it subtracts them from the valid set, making them rejected. Switching this +// call to the method form would therefore invert the tolerance into a ban. +const legacyNATSUserCredentialsKey = "MINIO_NOTIFY_NATS_USER_CREDENTIALS" + func checkValidNotificationKeysForSubSys(subSys string, tgt map[string]config.KVS) error { validKVS, ok := DefaultNotificationKVS[subSys] if !ok { return nil } + var deprecatedKeys []string + if subSys == config.NotifyNATSSubSys { + deprecatedKeys = []string{legacyNATSUserCredentialsKey} + } for tname, kv := range tgt { subSysTarget := subSys if tname != config.Default { subSysTarget = subSys + config.SubSystemSeparator + tname } if v, ok := kv.Lookup(config.Enable); ok && v == config.EnableOn { - if err := config.CheckValidKeys(subSysTarget, kv, validKVS); err != nil { + if err := config.CheckValidKeys(subSysTarget, kv, validKVS, deprecatedKeys...); err != nil { return err } } @@ -836,6 +863,10 @@ var ( Key: target.NATSUsername, Value: "", }, + config.KV{ + Key: target.NATSUserCredentials, + Value: "", + }, config.KV{ Key: target.NATSPassword, Value: "", @@ -844,6 +875,10 @@ var ( Key: target.NATSToken, Value: "", }, + config.KV{ + Key: target.NATSNKeySeed, + Value: "", + }, config.KV{ Key: target.NATSTLS, Value: config.EnableOff, @@ -852,6 +887,10 @@ var ( Key: target.NATSTLSSkipVerify, Value: config.EnableOff, }, + config.KV{ + Key: target.NATSTLSHandshakeFirst, + Value: config.EnableOff, + }, config.KV{ Key: target.NATSCertAuthority, Value: "", @@ -975,7 +1014,7 @@ func GetNotifyNATS(natsKVS map[string]config.KVS, rootCAs *x509.CertPool) (map[s usernameEnv = usernameEnv + config.Default + k } - userCredentialsEnv := target.NATSUserCredentials + userCredentialsEnv := target.EnvNATSUserCredentials if k != config.Default { userCredentialsEnv = userCredentialsEnv + config.Default + k } @@ -1020,12 +1059,21 @@ func GetNotifyNATS(natsKVS map[string]config.KVS, rootCAs *x509.CertPool) (map[s jetStreamEnableEnv = jetStreamEnableEnv + config.Default + k } + userCredentials := kv.Get(target.NATSUserCredentials) + if userCredentials == "" { + // Fall back to the legacy key written by pre-fix migration code; + // tolerated for compatibility. The proper user_credentials key + // wins when both are present, and the environment still overrides + // both. + userCredentials = kv.Get(legacyNATSUserCredentialsKey) + } + natsArgs := target.NATSArgs{ Enable: true, Address: *address, Subject: env.Get(subjectEnv, kv.Get(target.NATSSubject)), Username: env.Get(usernameEnv, kv.Get(target.NATSUsername)), - UserCredentials: env.Get(userCredentialsEnv, kv.Get(target.NATSUserCredentials)), + UserCredentials: env.Get(userCredentialsEnv, userCredentials), Password: env.Get(passwordEnv, kv.Get(target.NATSPassword)), CertAuthority: env.Get(certAuthorityEnv, kv.Get(target.NATSCertAuthority)), ClientCert: env.Get(clientCertEnv, kv.Get(target.NATSClientCert)), @@ -1652,6 +1700,10 @@ var ( Key: target.AmqpMandatory, Value: config.EnableOff, }, + config.KV{ + Key: target.AmqpImmediate, + Value: config.EnableOff, + }, config.KV{ Key: target.AmqpDurable, Value: config.EnableOff, diff --git a/internal/config/notify/parse_test.go b/internal/config/notify/parse_test.go new file mode 100644 index 000000000..51e99747f --- /dev/null +++ b/internal/config/notify/parse_test.go @@ -0,0 +1,757 @@ +// Copyright (c) 2015-2026 MinIO, Inc. +// +// This file is part of MinIO Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package notify + +import ( + "context" + "crypto/tls" + "go/ast" + "go/parser" + "go/token" + "io/fs" + "net/http" + "path/filepath" + "slices" + "sort" + "strconv" + "strings" + "testing" + + "github.com/minio/minio/internal/config" + "github.com/minio/minio/internal/event/target" +) + +// Test fixtures. Every value here is deliberately non-functional: the paths do +// not exist and the credentials are placeholders, so a leaked fixture is inert. +const ( + testCredsPath = "/nonexistent/test.creds" + testNKeyPath = "/nonexistent/test.nk" + testNATSAddr = "127.0.0.1:4222" + testNATSSubj = "test-subject" + testAMQPURL = "amqp://guest:guest@127.0.0.1:5672" + testTargetName = "FITCHECK" // named target from issue #39 + + // The config key an operator types into `mc admin config set`. Spelled as a + // literal so these tests pin the on-disk schema rather than tracking + // whatever the Go constant happens to say. + natsCredsKey = "user_credentials" + // The env var name that the pre-fix migration wrote into the config store + // as if it were a config key. + legacyNATSCredsKey = "MINIO_NOTIFY_NATS_USER_CREDENTIALS" +) + +// natsKVSFor builds a stored (on-disk) NATS target KVS carrying the three keys +// that issue #39 reported as rejected. config.Merge passes a stored target +// through verbatim rather than layering it over the defaults, so the numeric +// keys the parser reads must be present here too. +func natsKVSFor(credsKey string) config.KVS { + return config.KVS{ + config.KV{Key: config.Enable, Value: config.EnableOn}, + config.KV{Key: target.NATSAddress, Value: testNATSAddr}, + config.KV{Key: target.NATSSubject, Value: testNATSSubj}, + config.KV{Key: credsKey, Value: testCredsPath}, + config.KV{Key: target.NATSNKeySeed, Value: testNKeyPath}, + config.KV{Key: target.NATSTLSHandshakeFirst, Value: config.EnableOn}, + config.KV{Key: target.NATSPingInterval, Value: "0"}, + config.KV{Key: target.NATSQueueLimit, Value: "0"}, + } +} + +// T1: an enable=on NATS target carrying user_credentials / nkey_seed / +// tls_handshake_first must pass key validation. This is the exact shape from +// issue #39 (`mc admin config set us notify_nats:FITCHECK ...`). +func TestCheckValidNotificationKeysNATSJWTKeys(t *testing.T) { + for _, tgtName := range []string{config.Default, testTargetName} { + t.Run(tgtName, func(t *testing.T) { + tgt := map[string]config.KVS{ + tgtName: natsKVSFor(natsCredsKey), + } + if err := checkValidNotificationKeysForSubSys(config.NotifyNATSSubSys, tgt); err != nil { + t.Fatalf("expected NATS JWT/nkey/handshake keys to validate, got: %v", err) + } + }) + } +} + +// T2: the parser must surface the three values from stored config (no env). +func TestGetNotifyNATSFromStoredKVS(t *testing.T) { + for _, tgtName := range []string{config.Default, testTargetName} { + t.Run(tgtName, func(t *testing.T) { + natsKVS := map[string]config.KVS{ + tgtName: natsKVSFor(natsCredsKey), + } + targets, err := GetNotifyNATS(natsKVS, nil) + if err != nil { + t.Fatalf("GetNotifyNATS: %v", err) + } + args, ok := targets[tgtName] + if !ok { + t.Fatalf("target %q missing from result %v", tgtName, targets) + } + if args.UserCredentials != testCredsPath { + t.Errorf("UserCredentials = %q, want %q", args.UserCredentials, testCredsPath) + } + if args.NKeySeed != testNKeyPath { + t.Errorf("NKeySeed = %q, want %q", args.NKeySeed, testNKeyPath) + } + if !args.TLSHandshakeFirst { + t.Errorf("TLSHandshakeFirst = false, want true") + } + }) + } +} + +// T3: AMQP `immediate` must validate and parse. +func TestNotifyAMQPImmediate(t *testing.T) { + amqpKVS := map[string]config.KVS{ + config.Default: { + config.KV{Key: config.Enable, Value: config.EnableOn}, + config.KV{Key: target.AmqpURL, Value: testAMQPURL}, + config.KV{Key: target.AmqpImmediate, Value: config.EnableOn}, + config.KV{Key: target.AmqpDeliveryMode, Value: "0"}, + config.KV{Key: target.AmqpQueueLimit, Value: "0"}, + }, + } + if err := checkValidNotificationKeysForSubSys(config.NotifyAMQPSubSys, amqpKVS); err != nil { + t.Fatalf("expected amqp immediate key to validate, got: %v", err) + } + targets, err := GetNotifyAMQP(amqpKVS) + if err != nil { + t.Fatalf("GetNotifyAMQP: %v", err) + } + args, ok := targets[config.Default] + if !ok { + t.Fatalf("default target missing from result %v", targets) + } + if !args.Immediate { + t.Errorf("Immediate = false, want true") + } + if args.Internal { + t.Errorf("Internal = true, want false (immediate must not bleed into internal)") + } +} + +// T4 (guard, must hold before and after the fix): environment variables win +// over stored config, for both the default and the `_` suffixed form. +// Env var names are asserted as raw literals so that a rename of the Go +// constant cannot silently change the public interface. +func TestNotifyEnvOverridesStoredKVS(t *testing.T) { + const ( + envCreds = "MINIO_NOTIFY_NATS_USER_CREDENTIALS" + envNKeySeed = "MINIO_NOTIFY_NATS_NKEY_SEED" + envHandshakeFirst = "MINIO_NOTIFY_NATS_TLS_HANDSHAKE_FIRST" + envAMQPImmediate = "MINIO_NOTIFY_AMQP_IMMEDIATE" + envCredsOverride = "/nonexistent/env-override.creds" + envNKeySeedOverride = "/nonexistent/env-override.nk" + ) + + t.Run("nats-default", func(t *testing.T) { + t.Setenv(envCreds, envCredsOverride) + t.Setenv(envNKeySeed, envNKeySeedOverride) + t.Setenv(envHandshakeFirst, config.EnableOff) + + natsKVS := map[string]config.KVS{config.Default: natsKVSFor(natsCredsKey)} + targets, err := GetNotifyNATS(natsKVS, nil) + if err != nil { + t.Fatalf("GetNotifyNATS: %v", err) + } + args := targets[config.Default] + if args.UserCredentials != envCredsOverride { + t.Errorf("UserCredentials = %q, want env value %q", args.UserCredentials, envCredsOverride) + } + if args.NKeySeed != envNKeySeedOverride { + t.Errorf("NKeySeed = %q, want env value %q", args.NKeySeed, envNKeySeedOverride) + } + if args.TLSHandshakeFirst { + t.Errorf("TLSHandshakeFirst = true, want env value false") + } + }) + + t.Run("nats-named-target", func(t *testing.T) { + t.Setenv(envCreds+config.Default+testTargetName, envCredsOverride) + t.Setenv(envNKeySeed+config.Default+testTargetName, envNKeySeedOverride) + t.Setenv(envHandshakeFirst+config.Default+testTargetName, config.EnableOff) + + natsKVS := map[string]config.KVS{testTargetName: natsKVSFor(natsCredsKey)} + targets, err := GetNotifyNATS(natsKVS, nil) + if err != nil { + t.Fatalf("GetNotifyNATS: %v", err) + } + args := targets[testTargetName] + if args.UserCredentials != envCredsOverride { + t.Errorf("UserCredentials = %q, want env value %q", args.UserCredentials, envCredsOverride) + } + if args.NKeySeed != envNKeySeedOverride { + t.Errorf("NKeySeed = %q, want env value %q", args.NKeySeed, envNKeySeedOverride) + } + if args.TLSHandshakeFirst { + t.Errorf("TLSHandshakeFirst = true, want env value false") + } + }) + + t.Run("amqp-immediate", func(t *testing.T) { + t.Setenv(envAMQPImmediate, config.EnableOn) + amqpKVS := map[string]config.KVS{ + config.Default: { + config.KV{Key: config.Enable, Value: config.EnableOn}, + config.KV{Key: target.AmqpURL, Value: testAMQPURL}, + config.KV{Key: target.AmqpDeliveryMode, Value: "0"}, + config.KV{Key: target.AmqpQueueLimit, Value: "0"}, + }, + } + targets, err := GetNotifyAMQP(amqpKVS) + if err != nil { + t.Fatalf("GetNotifyAMQP: %v", err) + } + if !targets[config.Default].Immediate { + t.Errorf("Immediate = false, want env value true") + } + }) + + // Pin the env var names that the constants must keep producing. These are + // public interface: renaming one silently breaks every deployment that + // sets it. + t.Run("env-names-are-stable", func(t *testing.T) { + for _, tc := range []struct{ got, want string }{ + {target.EnvNATSUserCredentials, envCreds}, + {target.EnvNATSNKeySeed, envNKeySeed}, + {target.EnvNatsTLSHandshakeFirst, envHandshakeFirst}, + {target.EnvAMQPImmediate, envAMQPImmediate}, + } { + if tc.got != tc.want { + t.Errorf("env var name = %q, want %q", tc.got, tc.want) + } + } + }) + + // The config key must be the snake_case name, not the env var name. + t.Run("config-key-names-are-snake-case", func(t *testing.T) { + for _, tc := range []struct{ got, want string }{ + {target.NATSUserCredentials, natsCredsKey}, + {target.NATSNKeySeed, "nkey_seed"}, + {target.NATSTLSHandshakeFirst, "tls_handshake_first"}, + {target.AmqpImmediate, "immediate"}, + } { + if tc.got != tc.want { + t.Errorf("config key = %q, want %q", tc.got, tc.want) + } + } + }) +} + +// T6: a store written by the pre-fix migration carries the literal env-var name +// as a config key. Such a store must keep loading, and the value must still +// reach the parser. When both the legacy and the current key are present, the +// current key wins. +func TestNotifyNATSLegacyUserCredentialsKey(t *testing.T) { + const legacyKey = legacyNATSCredsKey + const newValue = "/nonexistent/new-key.creds" + + t.Run("legacy-key-alone-validates-and-is-read", func(t *testing.T) { + natsKVS := map[string]config.KVS{ + config.Default: natsKVSFor(legacyKey), + } + if err := checkValidNotificationKeysForSubSys(config.NotifyNATSSubSys, natsKVS); err != nil { + t.Fatalf("legacy-migrated store must keep validating, got: %v", err) + } + targets, err := GetNotifyNATS(natsKVS, nil) + if err != nil { + t.Fatalf("GetNotifyNATS: %v", err) + } + if got := targets[config.Default].UserCredentials; got != testCredsPath { + t.Errorf("UserCredentials = %q, want fallback to legacy key value %q", got, testCredsPath) + } + }) + + t.Run("new-key-wins-over-legacy", func(t *testing.T) { + kvs := natsKVSFor(legacyKey) + kvs = append(kvs, config.KV{Key: natsCredsKey, Value: newValue}) + natsKVS := map[string]config.KVS{config.Default: kvs} + + if err := checkValidNotificationKeysForSubSys(config.NotifyNATSSubSys, natsKVS); err != nil { + t.Fatalf("mixed old/new store must validate, got: %v", err) + } + targets, err := GetNotifyNATS(natsKVS, nil) + if err != nil { + t.Fatalf("GetNotifyNATS: %v", err) + } + if got := targets[config.Default].UserCredentials; got != newValue { + t.Errorf("UserCredentials = %q, want new key value %q", got, newValue) + } + }) + + t.Run("env-wins-over-legacy-key", func(t *testing.T) { + const envOverride = "/nonexistent/env-wins.creds" + t.Setenv(legacyNATSCredsKey, envOverride) + + natsKVS := map[string]config.KVS{config.Default: natsKVSFor(legacyKey)} + targets, err := GetNotifyNATS(natsKVS, nil) + if err != nil { + t.Fatalf("GetNotifyNATS: %v", err) + } + if got := targets[config.Default].UserCredentials; got != envOverride { + t.Errorf("UserCredentials = %q, want env value %q", got, envOverride) + } + }) + + t.Run("legacy-key-rejected-for-other-subsystems", func(t *testing.T) { + // The tolerance is NATS-scoped; it must not become a global escape hatch. + amqpKVS := map[string]config.KVS{ + config.Default: { + config.KV{Key: config.Enable, Value: config.EnableOn}, + config.KV{Key: target.AmqpURL, Value: testAMQPURL}, + config.KV{Key: legacyKey, Value: testCredsPath}, + }, + } + if err := checkValidNotificationKeysForSubSys(config.NotifyAMQPSubSys, amqpKVS); err == nil { + t.Fatal("expected the NATS legacy key to be rejected for the AMQP sub-system") + } + }) +} + +// T9 (characterization, not an endorsement): FetchEnabledTargets fails fast. +// A single sub-system with an invalid key aborts the whole target list, which +// is why one bad notify config disables every other notification target. This +// test pins the current behavior so that a future change to it is deliberate. +func TestFetchEnabledTargetsFailsFastAcrossSubSystems(t *testing.T) { + cfg := config.Config{ + config.NotifyNATSSubSys: map[string]config.KVS{ + config.Default: { + config.KV{Key: config.Enable, Value: config.EnableOn}, + config.KV{Key: target.NATSAddress, Value: testNATSAddr}, + config.KV{Key: target.NATSSubject, Value: testNATSSubj}, + config.KV{Key: target.NATSPingInterval, Value: "0"}, + config.KV{Key: target.NATSQueueLimit, Value: "0"}, + config.KV{Key: "this_key_does_not_exist", Value: "junk"}, + }, + }, + // A healthy sub-system that would yield a working target if the call + // got that far. NotifySubSystems.ToSlice() is sorted, so notify_nats + // is reached before notify_webhook and the error aborts the loop + // before this target is ever constructed -- the healthy config is lost + // without anything being built or dialed. + config.NotifyWebhookSubSys: map[string]config.KVS{ + config.Default: { + config.KV{Key: config.Enable, Value: config.EnableOn}, + config.KV{Key: target.WebhookEndpoint, Value: "http://127.0.0.1:65535/"}, + config.KV{Key: target.WebhookQueueLimit, Value: "0"}, + }, + }, + } + + // FetchEnabledTargets dereferences transport.TLSClientConfig unconditionally + // for some sub-systems, so a real transport is required even though no + // target here performs I/O. + transport := &http.Transport{TLSClientConfig: &tls.Config{}} + + targetList, err := FetchEnabledTargets(context.Background(), cfg, transport) + if err == nil { + t.Fatal("expected FetchEnabledTargets to fail fast on the invalid sub-system") + } + if targetList != nil { + t.Errorf("target list = %v, want nil (fail-fast discards healthy targets)", targetList) + } +} + +// --------------------------------------------------------------------------- +// T7: source-level consistency guard. +// +// Every config key the parser reads or the legacy migration writes must be +// registered in the sub-system's default KVS, otherwise config.CheckValidKeys +// rejects a config that the parser would happily consume (issue #39). +// --------------------------------------------------------------------------- + +// notifySubSysAudit enumerates the notify sub-systems and the symbols that make +// up their config surface. Keep in sync with DefaultNotificationKVS; the audit +// itself asserts that every registered sub-system appears here. +// tolerated lists keys that are not registered but are still accepted by +// checkValidNotificationKeysForSubSys as deprecated keys. The parser may read +// them; the migration must never write them and they stay undocumented. +var notifySubSysAudit = []struct { + subSys string + getFn string + setFn string + help config.HelpKVS + tolerated []string +}{ + {config.NotifyAMQPSubSys, "GetNotifyAMQP", "SetNotifyAMQP", HelpAMQP, nil}, + {config.NotifyESSubSys, "GetNotifyES", "SetNotifyES", HelpES, nil}, + {config.NotifyKafkaSubSys, "GetNotifyKafka", "SetNotifyKafka", HelpKafka, nil}, + {config.NotifyMQTTSubSys, "GetNotifyMQTT", "SetNotifyMQTT", HelpMQTT, nil}, + {config.NotifyMySQLSubSys, "GetNotifyMySQL", "SetNotifyMySQL", HelpMySQL, nil}, + {config.NotifyNATSSubSys, "GetNotifyNATS", "SetNotifyNATS", HelpNATS, []string{legacyNATSUserCredentialsKey}}, + {config.NotifyNSQSubSys, "GetNotifyNSQ", "SetNotifyNSQ", HelpNSQ, nil}, + {config.NotifyPostgresSubSys, "GetNotifyPostgres", "SetNotifyPostgres", HelpPostgres, nil}, + {config.NotifyRedisSubSys, "GetNotifyRedis", "SetNotifyRedis", HelpRedis, nil}, + {config.NotifyWebhookSubSys, "GetNotifyWebhook", "SetNotifyWebhook", HelpWebhook, nil}, +} + +// notifyPkgConsts resolves bare identifiers used as config keys in this +// package. Compiler-resolved, so they cannot drift. +var notifyPkgConsts = map[string]string{ + "legacyNATSUserCredentialsKey": legacyNATSUserCredentialsKey, +} + +// configPkgConsts resolves the `config.X` selectors that appear as config keys +// in parse.go / legacy.go. These are compiler-resolved, so they cannot drift. +var configPkgConsts = map[string]string{ + "Enable": config.Enable, + "Comment": config.Comment, +} + +// knownUnregisteredWrites records pre-existing instances of the exact defect +// this audit exists to catch: a legacy migration writing config keys that no +// default KVS registers, so the migrated config is rejected on the next load. +// +// These are inherited from upstream and are the same class as issue #39, but +// they are NOT part of the issue #39 fix and were left untouched deliberately. +// The Postgres/MySQL keys below are the pre-connection-string DSN fields; the +// migration still writes them and `password` carries a plaintext database +// password. +// +// This list must only ever shrink. Do not add entries to silence a new gap. +var knownUnregisteredWrites = map[string][]string{ + "SetNotifyPostgres": {"host", "port", "username", "password", "database"}, + "SetNotifyMySQL": {"host", "port", "username", "password", "database"}, +} + +func TestNotifyConfigKeysAreRegistered(t *testing.T) { + targetConsts, err := parseTargetPkgStringConsts("../../event/target") + if err != nil { + t.Fatalf("resolving internal/event/target constants: %v", err) + } + + fset := token.NewFileSet() + readKeys := map[string]map[string]token.Position{} // fn name -> key -> pos + writtenKeys := map[string]map[string]token.Position{} // fn name -> key -> pos + + for _, src := range []string{"parse.go", "legacy.go"} { + f, err := parser.ParseFile(fset, src, nil, 0) + if err != nil { + t.Fatalf("parsing %s: %v", src, err) + } + resolve := func(e ast.Expr) (string, bool) { + return resolveKeyExpr(t, fset, src, e, targetConsts) + } + for _, decl := range f.Decls { + fn, ok := decl.(*ast.FuncDecl) + if !ok || fn.Body == nil { + continue + } + name := fn.Name.Name + switch { + case strings.HasPrefix(name, "GetNotify"): + readKeys[name] = collectKVGetKeys(fset, fn, resolve) + case strings.HasPrefix(name, "SetNotify"): + writtenKeys[name] = collectKVWriteKeys(fset, fn, resolve) + } + } + } + + // Guard against a new sub-system slipping past the audit table. + audited := map[string]bool{} + for _, a := range notifySubSysAudit { + audited[a.subSys] = true + } + for subSys := range DefaultNotificationKVS { + if !audited[subSys] { + t.Errorf("sub-system %q is registered in DefaultNotificationKVS but missing from notifySubSysAudit", subSys) + } + } + + for _, a := range notifySubSysAudit { + t.Run(a.subSys, func(t *testing.T) { + defaults := DefaultNotificationKVS[a.subSys] + registered := map[string]bool{} + for _, kv := range defaults { + registered[kv.Key] = true + } + + read, ok := readKeys[a.getFn] + if !ok { + t.Fatalf("%s not found in parse.go/legacy.go", a.getFn) + } + for _, key := range sortedKeys(read) { + if !registered[key] && !slices.Contains(a.tolerated, key) { + t.Errorf("%s reads key %q (%s) which is not registered in the default KVS for %s", + a.getFn, key, read[key], a.subSys) + } + } + // Floor assertion, in the opposite direction. Every registered key + // is in fact read by every GetNotifyX today, so this holds. Its + // real job is to fail loudly if the read collector ever stops + // matching the source — a collector that silently returns nothing + // would otherwise make the check above pass vacuously. + for _, kv := range defaults { + if _, ok := read[kv.Key]; !ok { + t.Errorf("registered key %q for %s is never read by %s; either the key is dead or this audit has gone blind", + kv.Key, a.subSys, a.getFn) + } + } + // A tolerated key must genuinely be accepted by validation, + // otherwise reading it is pointless. + for _, key := range a.tolerated { + kvs := map[string]config.KVS{config.Default: { + config.KV{Key: config.Enable, Value: config.EnableOn}, + config.KV{Key: key, Value: "x"}, + }} + if err := checkValidNotificationKeysForSubSys(a.subSys, kvs); err != nil { + t.Errorf("tolerated key %q is still rejected for %s: %v", key, a.subSys, err) + } + } + + written, ok := writtenKeys[a.setFn] + if !ok { + t.Fatalf("%s not found in parse.go/legacy.go", a.setFn) + } + allowed := knownUnregisteredWrites[a.setFn] + for _, key := range sortedKeys(written) { + if !registered[key] && !slices.Contains(allowed, key) { + t.Errorf("%s writes key %q (%s) which is not registered in the default KVS for %s", + a.setFn, key, written[key], a.subSys) + } + } + // Keep the allowlist honest: an entry that no longer corresponds to + // a real gap must be deleted, not left to rot. + for _, key := range allowed { + if registered[key] { + t.Errorf("%s: %q is registered now; remove it from knownUnregisteredWrites", a.setFn, key) + } else if _, ok := written[key]; !ok { + t.Errorf("%s no longer writes %q; remove it from knownUnregisteredWrites", a.setFn, key) + } + } + + // Help entries must describe registered keys only. `comment` is + // accepted for every sub-system by CheckValidKeys and is therefore + // documented without being registered. + for _, hkv := range a.help { + if hkv.Key == config.Comment { + continue + } + if !registered[hkv.Key] { + t.Errorf("Help entry %q for %s is not registered in the default KVS", hkv.Key, a.subSys) + } + } + + // Conversely: every registered key a user can set should be + // documented, so `mc admin config get` explains it. `enable` is + // implicit for every target and HiddenIfEmpty keys are deprecated + // leftovers that are deliberately undocumented. + for _, kv := range defaults { + if kv.Key == config.Enable || kv.HiddenIfEmpty { + continue + } + if _, ok := a.help.Lookup(kv.Key); !ok { + t.Errorf("registered key %q for %s has no Help entry", kv.Key, a.subSys) + } + } + }) + } +} + +func sortedKeys(m map[string]token.Position) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} + +// resolveKeyExpr turns a config-key expression into its string value. Only the +// shapes actually used in this package are supported; anything else fails the +// test loudly rather than being skipped, so the audit cannot silently go blind. +func resolveKeyExpr(t *testing.T, fset *token.FileSet, src string, e ast.Expr, targetConsts map[string]string) (string, bool) { + t.Helper() + switch v := e.(type) { + case *ast.Ident: + val, ok := notifyPkgConsts[v.Name] + if !ok { + t.Errorf("%s: %s used as a config key but not listed in notifyPkgConsts; add it", + fset.Position(v.Pos()), v.Name) + return "", false + } + return val, true + case *ast.BasicLit: + if v.Kind == token.STRING { + s, err := strconv.Unquote(v.Value) + if err != nil { + t.Errorf("%s: unquoting %s: %v", fset.Position(v.Pos()), v.Value, err) + return "", false + } + return s, true + } + case *ast.SelectorExpr: + pkg, ok := v.X.(*ast.Ident) + if !ok { + break + } + switch pkg.Name { + case "target": + val, ok := targetConsts[v.Sel.Name] + if !ok { + t.Errorf("%s: cannot resolve target.%s to a string constant", fset.Position(v.Pos()), v.Sel.Name) + return "", false + } + return val, true + case "config": + val, ok := configPkgConsts[v.Sel.Name] + if !ok { + t.Errorf("%s: config.%s used as a config key but not listed in configPkgConsts; add it", + fset.Position(v.Pos()), v.Sel.Name) + return "", false + } + return val, true + } + } + t.Errorf("%s: unsupported config-key expression %T in %s; extend resolveKeyExpr", fset.Position(e.Pos()), e, src) + return "", false +} + +// collectKVGetKeys finds every `.Get()` call inside fn. This is how +// GetNotifyX reads a stored config value. +func collectKVGetKeys(fset *token.FileSet, fn *ast.FuncDecl, resolve func(ast.Expr) (string, bool)) map[string]token.Position { + keys := map[string]token.Position{} + ast.Inspect(fn.Body, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok || len(call.Args) != 1 { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok || sel.Sel.Name != "Get" { + return true + } + recv, ok := sel.X.(*ast.Ident) + if !ok || recv.Name != "kv" { + return true + } + if key, ok := resolve(call.Args[0]); ok { + keys[key] = fset.Position(call.Pos()) + } + return true + }) + return keys +} + +// collectKVWriteKeys finds every `{Key: , ...}` composite literal inside +// fn. This is how SetNotifyX writes a migrated config. +// +// Both the explicit `config.KV{...}` form and the elided `{...}` form that Go +// permits inside a `config.KVS{...}` literal must be matched: the elided form +// is idiomatic and would otherwise slip past this audit silently. +func collectKVWriteKeys(fset *token.FileSet, fn *ast.FuncDecl, resolve func(ast.Expr) (string, bool)) map[string]token.Position { + keys := map[string]token.Position{} + ast.Inspect(fn.Body, func(n ast.Node) bool { + lit, ok := n.(*ast.CompositeLit) + if !ok { + return true + } + // A typed literal that is neither config.KV nor config.KVS cannot hold + // or contain a config key. An untyped (elided) literal has no type to + // check, so it must be inspected rather than skipped. + if sel, ok := lit.Type.(*ast.SelectorExpr); ok && sel.Sel.Name != "KV" && sel.Sel.Name != "KVS" { + return true + } + for _, elt := range lit.Elts { + kve, ok := elt.(*ast.KeyValueExpr) + if !ok { + continue + } + ident, ok := kve.Key.(*ast.Ident) + if !ok || ident.Name != "Key" { + continue + } + if key, ok := resolve(kve.Value); ok { + keys[key] = fset.Position(lit.Pos()) + } + } + return true + }) + return keys +} + +// parseTargetPkgStringConsts reads every `Name = "literal"` constant from the +// internal/event/target sources. Those constants are the single source of truth +// for both config keys and env var names. +func parseTargetPkgStringConsts(dir string) (map[string]string, error) { + fset := token.NewFileSet() + pkgs, err := parser.ParseDir(fset, filepath.Clean(dir), func(fi fs.FileInfo) bool { + return !strings.HasSuffix(fi.Name(), "_test.go") + }, 0) + if err != nil { + return nil, err + } + out := map[string]string{} + for _, pkg := range pkgs { + for _, f := range pkg.Files { + for _, decl := range f.Decls { + gd, ok := decl.(*ast.GenDecl) + if !ok || gd.Tok != token.CONST { + continue + } + for _, spec := range gd.Specs { + vs, ok := spec.(*ast.ValueSpec) + if !ok || len(vs.Names) != 1 || len(vs.Values) != 1 { + continue + } + lit, ok := vs.Values[0].(*ast.BasicLit) + if !ok || lit.Kind != token.STRING { + continue + } + s, err := strconv.Unquote(lit.Value) + if err != nil { + continue + } + out[vs.Names[0].Name] = s + } + } + } + } + if len(out) == 0 { + return nil, errNoTargetConsts + } + return out, nil +} + +var errNoTargetConsts = errConst("no string constants found in internal/event/target; the audit would be vacuous") + +type errConst string + +func (e errConst) Error() string { return string(e) } + +// Sanity check for the resolver itself: if this ever stops finding known +// constants the audit above would pass vacuously. +func TestParseTargetPkgStringConsts(t *testing.T) { + consts, err := parseTargetPkgStringConsts("../../event/target") + if err != nil { + t.Fatal(err) + } + for name, want := range map[string]string{ + "NATSAddress": "address", + "NATSNKeySeed": "nkey_seed", + "AmqpImmediate": "immediate", + "EnvAMQPImmediate": "MINIO_NOTIFY_AMQP_IMMEDIATE", + } { + if got := consts[name]; got != want { + t.Errorf("target.%s = %q, want %q", name, got, want) + } + } + if slices.Contains([]string{""}, consts["NATSSubject"]) { + t.Error("target.NATSSubject resolved to empty string") + } +} diff --git a/internal/event/target/nats.go b/internal/event/target/nats.go index c96833bc4..9011de5eb 100644 --- a/internal/event/target/nats.go +++ b/internal/event/target/nats.go @@ -43,6 +43,7 @@ const ( NATSAddress = "address" NATSSubject = "subject" NATSUsername = "username" + NATSUserCredentials = "user_credentials" NATSPassword = "password" NATSToken = "token" NATSNKeySeed = "nkey_seed" @@ -69,7 +70,7 @@ const ( EnvNATSAddress = "MINIO_NOTIFY_NATS_ADDRESS" EnvNATSSubject = "MINIO_NOTIFY_NATS_SUBJECT" EnvNATSUsername = "MINIO_NOTIFY_NATS_USERNAME" - NATSUserCredentials = "MINIO_NOTIFY_NATS_USER_CREDENTIALS" + EnvNATSUserCredentials = "MINIO_NOTIFY_NATS_USER_CREDENTIALS" EnvNATSPassword = "MINIO_NOTIFY_NATS_PASSWORD" EnvNATSToken = "MINIO_NOTIFY_NATS_TOKEN" EnvNATSNKeySeed = "MINIO_NOTIFY_NATS_NKEY_SEED"