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 <noreply@openai.com>
Co-authored-by: Claude <noreply@anthropic.com>

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,