https://github.com/redpanda-data/connect/issues/3010

initial

user
password
token

docs/modules/components/pages/inputs/nats_jetstream.adoc

@@ -80,6 +80,7 @@ input:
       user_credentials_file: ./user.creds # No default (optional)
       user_jwt: "" # No default (optional)
       user_nkey_seed: "" # No default (optional)
+      token: "" # No default (optional)
     extract_tracing_map: root = @ # No default (optional)
 ```
 

internal/impl/nats/auth.go

@@ -60,6 +60,10 @@ generated with the https://docs.nats.io/nats-tools/nsc[nsc tool^].
 Alternatively, the ` + "`user_jwt`" + ` field can contain a plain text JWT and the ` + "`user_nkey_seed`" + `can contain
 the plain text NKey Seed.
 
+Alternatively, the ` + "`token`" + ` field can contain a plain text random string.
+
+Alternatively, the ` + "`user`" + ` and ` + "`password`" + ` fields can contain plain text user and password.
+
 https://docs.nats.io/using-nats/developer/connecting/creds[More details^].`
 }
 
@@ -83,6 +87,18 @@ func authFieldSpec() *service.ConfigField {
 			Description("An optional plain text user JWT (given along with the corresponding user NKey Seed).").
 			Secret().
 			Optional(),
+		service.NewStringField("user").
+			Description("An optional plain text user name (given along with the corresponding user password).").
+			Secret().
+			Optional(),
+		service.NewStringField("password").
+			Description("An optional plain text password (given along with the corresponding user name).").
+			Secret().
+			Optional(),
+		service.NewStringField("token").
+			Description("An optional plain text token.").
+			Secret().
+			Optional(),
 		service.NewStringField("user_nkey_seed").
 			Description("An optional plain text user NKey Seed (given along with the corresponding user JWT).").
 			Secret().
@@ -97,6 +113,9 @@ type authConfig struct {
 	UserCredentialsFile string
 	UserJWT             string
 	UserNkeySeed        string
+	Token               string
+	User                string
+	Password            string
 }
 
 //------------------------------------------------------------------------------
@@ -137,6 +156,15 @@ func authConfToOptions(auth authConfig, fs *service.FS) []nats.Option {
 		))
 	}
 
+	if auth.Token != "" {
+		opts = append(opts, nats.Token(
+			auth.Token,
+		))
+	}
+
+	if auth.User != "" && auth.Password != "" {
+		opts = append(opts, nats.UserInfo(auth.User, auth.Password))
+	}
 	return opts
 }
 
@@ -173,6 +201,20 @@ func AuthFromParsedConfig(p *service.ParsedConfig) (c authConfig, err error) {
 			return
 		}
 	}
+	if p.Contains("token") {
+		if c.Token, err = p.FieldString("token"); err != nil {
+			return
+		}
+	}
+
+	if p.Contains("user") && p.Contains("password") {
+		if c.User, err = p.FieldString("user"); err != nil {
+			return
+		}
+		if c.Password, err = p.FieldString("password"); err != nil {
+			return
+		}
+	}
 	return
 }
 

internal/impl/nats/input_jetstream_test.go

@@ -35,6 +35,9 @@ auth:
   nkey_file: test auth n key file
   user_credentials_file: test auth user creds file
   user_jwt: test auth inline user JWT
+  token: test auth inline user token
+  user: test auth inline user name
+  password: test auth inline user password
   user_nkey_seed: test auth inline user NKey Seed
 `
 
@@ -49,9 +52,41 @@ auth:
 		assert.Equal(t, "test auth n key file", e.connDetails.authConf.NKeyFile)
 		assert.Equal(t, "test auth user creds file", e.connDetails.authConf.UserCredentialsFile)
 		assert.Equal(t, "test auth inline user JWT", e.connDetails.authConf.UserJWT)
+		assert.Equal(t, "test auth inline user token", e.connDetails.authConf.Token)
+		assert.Equal(t, "test auth inline user name", e.connDetails.authConf.User)
+		assert.Equal(t, "test auth inline user password", e.connDetails.authConf.Password)
 		assert.Equal(t, "test auth inline user NKey Seed", e.connDetails.authConf.UserNkeySeed)
 	})
 
+	t.Run("Missing password", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1, url2 ]
+subject: testsubject
+auth:
+  user: test auth inline user name
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.Error(t, err)
+	})
+	t.Run("Missing user", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1, url2 ]
+subject: testsubject
+auth:
+  password: test auth inline user password
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.Error(t, err)
+	})
+
 	t.Run("Missing user_nkey_seed", func(t *testing.T) {
 		inputConfig := `
 urls: [ url1, url2 ]

internal/impl/nats/input_kv_test.go

@@ -39,6 +39,9 @@ auth:
   nkey_file: test auth n key file
   user_credentials_file: test auth user creds file
   user_jwt: test auth inline user JWT
+  token: test auth inline user token
+  user: test auth inline user name
+  password: test auth inline user password
   user_nkey_seed: test auth inline user NKey Seed
 `
 
@@ -57,9 +60,40 @@ auth:
 		assert.Equal(t, "test auth n key file", e.connDetails.authConf.NKeyFile)
 		assert.Equal(t, "test auth user creds file", e.connDetails.authConf.UserCredentialsFile)
 		assert.Equal(t, "test auth inline user JWT", e.connDetails.authConf.UserJWT)
+		assert.Equal(t, "test auth inline user token", e.connDetails.authConf.Token)
+		assert.Equal(t, "test auth inline user name", e.connDetails.authConf.User)
+		assert.Equal(t, "test auth inline user password", e.connDetails.authConf.Password)
 		assert.Equal(t, "test auth inline user NKey Seed", e.connDetails.authConf.UserNkeySeed)
 	})
 
+	t.Run("Missing password", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1, url2 ]
+subject: testsubject
+auth:
+  user: test auth inline user name
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.Error(t, err)
+	})
+	t.Run("Missing user", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1, url2 ]
+subject: testsubject
+auth:
+  password: test auth inline user password
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.Error(t, err)
+	})
 	t.Run("Missing user_nkey_seed", func(t *testing.T) {
 		inputConfig := `
 urls: [ url1, url2 ]

internal/impl/nats/output_jetstream_test.go

@@ -38,6 +38,9 @@ auth:
   nkey_file: test auth n key file
   user_credentials_file: test auth user creds file
   user_jwt: test auth inline user JWT
+  token: test auth inline user token
+  user: test auth inline user name
+  password: test auth inline user password
   user_nkey_seed: test auth inline user NKey Seed
 `
 
@@ -66,9 +69,41 @@ auth:
 		assert.Equal(t, "test auth n key file", e.connDetails.authConf.NKeyFile)
 		assert.Equal(t, "test auth user creds file", e.connDetails.authConf.UserCredentialsFile)
 		assert.Equal(t, "test auth inline user JWT", e.connDetails.authConf.UserJWT)
+		assert.Equal(t, "test auth inline user token", e.connDetails.authConf.Token)
+		assert.Equal(t, "test auth inline user name", e.connDetails.authConf.User)
+		assert.Equal(t, "test auth inline user password", e.connDetails.authConf.Password)
 		assert.Equal(t, "test auth inline user NKey Seed", e.connDetails.authConf.UserNkeySeed)
 	})
 
+	t.Run("Missing password", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1, url2 ]
+subject: testsubject
+auth:
+  user: test auth inline user name
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.Error(t, err)
+	})
+	t.Run("Missing user", func(t *testing.T) {
+		inputConfig := `
+urls: [ url1, url2 ]
+subject: testsubject
+auth:
+  password: test auth inline user password
+`
+
+		conf, err := spec.ParseYAML(inputConfig, env)
+		require.NoError(t, err)
+
+		_, err = newJetStreamReaderFromConfig(conf, service.MockResources())
+		require.Error(t, err)
+	})
+
 	t.Run("Missing user_nkey_seed", func(t *testing.T) {
 		inputConfig := `
 urls: [ url1, url2 ]