Repository: nats-io/nats.go
Pull request #1613 (2024-04-17 10:26:45)

[FIXED] Validation in jetstream and KV
Signed-off-by: Piotr Piotrowski [piotr@synadia.com](mailto:piotr@synadia.com)


Patch:
diff --git a/jetstream/consumer.go b/jetstream/consumer.go
index 7792ced..aa9003f 100644
--- a/jetstream/consumer.go
+++ b/jetstream/consumer.go
@@ -208,6 +208,9 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 
 	var ccSubj string
 	if cfg.FilterSubject != "" && len(cfg.FilterSubjects) == 0 {
+		if err := validateSubject(cfg.FilterSubject); err != nil {
+			return nil, err
+		}
 		ccSubj = apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject))
 	} else {
 		ccSubj = apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerCreateT, stream, consumerName))
@@ -318,8 +321,11 @@ func deleteConsumer(ctx context.Context, js *jetStream, stream, consumer string)
 }
 
 func validateConsumerName(dur string) error {
-	if strings.Contains(dur, ".") {
-		return fmt.Errorf("%w: %q", ErrInvalidConsumerName, dur)
+	if dur == "" {
+		return fmt.Errorf("%w: '%s'", ErrInvalidConsumerName, "name is required")
+	}
+	if strings.ContainsAny(dur, ">*. /\\") {
+		return fmt.Errorf("%w: '%s'", ErrInvalidConsumerName, dur)
 	}
 	return nil
 }
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 86200de..3e48788 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -773,7 +773,7 @@ func validateStreamName(stream string) error {
 	if stream == "" {
 		return ErrStreamNameRequired
 	}
-	if strings.Contains(stream, ".") {
+	if strings.ContainsAny(stream, ">*. /\\") {
 		return fmt.Errorf("%w: '%s'", ErrInvalidStreamName, stream)
 	}
 	return nil
@@ -783,7 +783,7 @@ func validateSubject(subject string) error {
 	if subject == "" {
 		return fmt.Errorf("%w: %s", ErrInvalidSubject, "subject cannot be empty")
 	}
-	if !subjectRegexp.MatchString(subject) {
+	if subject[0] == '.' || subject[len(subject)-1] == '.' || !subjectRegexp.MatchString(subject) {
 		return fmt.Errorf("%w: %s", ErrInvalidSubject, subject)
 	}
 	return nil
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 3bc6890..7b39afa 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
@@ -447,12 +448,13 @@ const (
 
 // Regex for valid keys and buckets.
 var (
-	validBucketRe = regexp.MustCompile(`\A[a-zA-Z0-9_-]+\z`)
-	validKeyRe    = regexp.MustCompile(`\A[-/_=\.a-zA-Z0-9]+\z`)
+	validBucketRe    = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)
+	validKeyRe       = regexp.MustCompile(`^[-/_=\.a-zA-Z0-9]+$`)
+	validSearchKeyRe = regexp.MustCompile(`^[-/_=\.a-zA-Z0-9*]*[>]?$`)
 )
 
 func (js *jetStream) KeyValue(ctx context.Context, bucket string) (KeyValue, error) {
-	if !validBucketRe.MatchString(bucket) {
+	if !bucketValid(bucket) {
 		return nil, ErrInvalidBucketName
 	}
 	streamName := fmt.Sprintf(kvBucketNameTmpl, bucket)
@@ -488,8 +490,26 @@ func (js *jetStream) CreateKeyValue(ctx context.Context, cfg KeyValueConfig) (Ke
 			// errors are joined so that backwards compatibility is retained
 			// and previous checks for ErrStreamNameAlreadyInUse will still work.
 			err = errors.Join(fmt.Errorf("%w: %s", ErrBucketExists, cfg.Bucket), err)
+
+			// If we have a failure to add, it could be because we have
+			// a config change if the KV was created against before a bug fix
+			// that changed the value of discard policy.
+			// We will check if the stream exists and if the only difference
+			// is the discard policy, we will update the stream.
+			// The same logic applies for KVs created pre 2.9.x and
+			// the AllowDirect setting.
+			if stream, _ = js.Stream(ctx, scfg.Name); stream != nil {
+				cfg := stream.CachedInfo().Config
+				cfg.Discard = scfg.Discard
+				cfg.AllowDirect = scfg.AllowDirect
+				if reflect.DeepEqual(cfg, scfg) {
+					stream, err = js.UpdateStream(ctx, scfg)
+				}
+			}
+		}
+		if err != nil {
+			return nil, err
 		}
-		return nil, err
 	}
 	pushJS, err := js.legacyJetStream()
 	if err != nil {
@@ -539,7 +559,7 @@ func (js *jetStream) CreateOrUpdateKeyValue(ctx context.Context, cfg KeyValueCon
 }
 
 func (js *jetStream) prepareKeyValueConfig(ctx context.Context, cfg KeyValueConfig) (StreamConfig, error) {
-	if !validBucketRe.MatchString(cfg.Bucket) {
+	if !bucketValid(cfg.Bucket) {
 		return StreamConfig{}, ErrInvalidBucketName
 	}
 	if _, err := js.AccountInfo(ctx); err != nil {
@@ -637,7 +657,7 @@ func (js *jetStream) prepareKeyValueConfig(ctx context.Context, cfg KeyValueConf
 
 // DeleteKeyValue will delete this KeyValue store (JetStream stream).
 func (js *jetStream) DeleteKeyValue(ctx context.Context, bucket string) error {
-	if !validBucketRe.MatchString(bucket) {
+	if !bucketValid(bucket) {
 		return ErrInvalidBucketName
 	}
 	stream := fmt.Sprintf(kvBucketNameTmpl, bucket)
@@ -774,6 +794,13 @@ func (js *jetStream) legacyJetStream() (nats.JetStreamContext, error) {
 	return js.conn.JetStream(opts...)
 }
 
+func bucketValid(bucket string) bool {
+	if len(bucket) == 0 {
+		return false
+	}
+	return validBucketRe.MatchString(bucket)
+}
+
 func keyValid(key string) bool {
 	if len(key) == 0 || key[0] == '.' || key[len(key)-1] == '.' {
 		return false
@@ -781,6 +808,13 @@ func keyValid(key string) bool {
 	return validKeyRe.MatchString(key)
 }
 
+func searchKeyValid(key string) bool {
+	if len(key) == 0 || key[0] == '.' || key[len(key)-1] == '.' {
+		return false
+	}
+	return validSearchKeyRe.MatchString(key)
+}
+
 func (kv *kvs) get(ctx context.Context, key string, revision uint64) (KeyValueEntry, error) {
 	if !keyValid(key) {
 		return nil, ErrInvalidKey
@@ -1037,6 +1071,9 @@ func (w *watcher) Stop() error {
 // Watch for any updates to keys that match the keys argument which could include wildcards.
 // Watch will send a nil entry when it has received all initial values.
 func (kv *kvs) Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWatcher, error) {
+	if !searchKeyValid(keys) {
+		return nil, fmt.Errorf("%w: %s", ErrInvalidKey, "keys cannot be empty and must be a valid NATS subject")
+	}
 	var o watchOpts
 	for _, opt := range opts {
 		if opt != nil {
diff --git a/jserrors.go b/jserrors.go
index b9a62bd..f028594 100644
--- a/jserrors.go
+++ b/jserrors.go
@@ -124,6 +124,9 @@ var (
 	// ErrInvalidConsumerName is returned when the provided consumer name is invalid (contains '.' or ' ').
 	ErrInvalidConsumerName JetStreamError = &jsError{message: "invalid consumer name"}
 
+	// ErrInvalidFilterSubject is returned when the provided filter subject is invalid.
+	ErrInvalidFilterSubject JetStreamError = &jsError{message: "invalid filter subject"}
+
 	// ErrNoMatchingStream is returned when stream lookup by subject is unsuccessful.
 	ErrNoMatchingStream JetStreamError = &jsError{message: "no stream matches subject"}
 
diff --git a/jsm.go b/jsm.go
index 78ead5b..dd09b10 100644
--- a/jsm.go
+++ b/jsm.go
@@ -414,6 +414,10 @@ func (js *js) upsertConsumer(stream, consumerName string, cfg *ConsumerConfig, o
 			// if filter subject is empty or ">", use the endpoint without filter subject
 			ccSubj = fmt.Sprintf(apiConsumerCreateT, stream, consumerName)
 		} else {
+			// safeguard against passing invalid filter subject in request subject
+			if cfg.FilterSubject[0] == '.' || cfg.FilterSubject[len(cfg.FilterSubject)-1] == '.' {
+				return nil, fmt.Errorf("%w: %q", ErrInvalidFilterSubject, cfg.FilterSubject)
+			}
 			// if filter subject is not empty, use the endpoint with filter subject
 			ccSubj = fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject)
 		}
diff --git a/kv.go b/kv.go
index 0864f30..8ed84be 100644
--- a/kv.go
+++ b/kv.go
@@ -344,8 +344,9 @@ const (
 
 // Regex for valid keys and buckets.
 var (
-	validBucketRe = regexp.MustCompile(`\A[a-zA-Z0-9_-]+\z`)
-	validKeyRe    = regexp.MustCompile(`\A[-/_=\.a-zA-Z0-9]+\z`)
+	validBucketRe    = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)
+	validKeyRe       = regexp.MustCompile(`^[-/_=\.a-zA-Z0-9]+$`)
+	validSearchKeyRe = regexp.MustCompile(`^[-/_=\.a-zA-Z0-9*]*[>]?$`)
 )
 
 // KeyValue will lookup and bind to an existing KeyValue store.
@@ -353,7 +354,7 @@ func (js *js) KeyValue(bucket string) (KeyValue, error) {
 	if !js.nc.serverMinVersion(2, 6, 2) {
 		return nil, errors.New("nats: key-value requires at least server version 2.6.2")
 	}
-	if !validBucketRe.MatchString(bucket) {
+	if !bucketValid(bucket) {
 		return nil, ErrInvalidBucketName
 	}
 	stream := fmt.Sprintf(kvBucketNameTmpl, bucket)
@@ -381,7 +382,7 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 	if cfg == nil {
 		return nil, ErrKeyValueConfigRequired
 	}
-	if !validBucketRe.MatchString(cfg.Bucket) {
+	if !bucketValid(cfg.Bucket) {
 		return nil, ErrInvalidBucketName
 	}
 	if _, err := js.AccountInfo(); err != nil {
@@ -507,7 +508,7 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 
 // DeleteKeyValue will delete this KeyValue store (JetStream stream).
 func (js *js) DeleteKeyValue(bucket string) error {
-	if !validBucketRe.MatchString(bucket) {
+	if !bucketValid(bucket) {
 		return ErrInvalidBucketName
 	}
 	stream := fmt.Sprintf(kvBucketNameTmpl, bucket)
@@ -547,6 +548,13 @@ func (e *kve) Created() time.Time    { return e.created }
 func (e *kve) Delta() uint64         { return e.delta }
 func (e *kve) Operation() KeyValueOp { return e.op }
 
+func bucketValid(bucket string) bool {
+	if len(bucket) == 0 {
+		return false
+	}
+	return validBucketRe.MatchString(bucket)
+}
+
 func keyValid(key string) bool {
 	if len(key) == 0 || key[0] == '.' || key[len(key)-1] == '.' {
 		return false
@@ -554,6 +562,13 @@ func keyValid(key string) bool {
 	return validKeyRe.MatchString(key)
 }
 
+func searchKeyValid(key string) bool {
+	if len(key) == 0 || key[0] == '.' || key[len(key)-1] == '.' {
+		return false
+	}
+	return validSearchKeyRe.MatchString(key)
+}
+
 // Get returns the latest value for the key.
 func (kv *kvs) Get(key string) (KeyValueEntry, error) {
 	e, err := kv.get(key, kvLatestRevision)
@@ -951,6 +966,9 @@ func (kv *kvs) WatchAll(opts ...WatchOpt) (KeyWatcher, error) {
 // Watch will fire the callback when a key that matches the keys pattern is updated.
 // keys needs to be a valid NATS subject.
 func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
+	if !searchKeyValid(keys) {
+		return nil, fmt.Errorf("%w: %s", ErrInvalidKey, "keys cannot be empty and must be a valid NATS subject")
+	}
 	var o watchOpts
 	for _, opt := range opts {
 		if opt != nil {


Test patch:
diff --git a/jetstream/jetstream_test.go b/jetstream/jetstream_test.go
index 62af4ef..878d361 100644
--- a/jetstream/jetstream_test.go
+++ b/jetstream/jetstream_test.go
@@ -456,3 +456,96 @@ func TestPullConsumer_checkPending(t *testing.T) {
 		})
 	}
 }
+
+func TestKV_keyValid(t *testing.T) {
+	tests := []struct {
+		key string
+		ok  bool
+	}{
+		{key: "foo123", ok: true},
+		{key: "foo.bar", ok: true},
+		{key: "Foo.123=bar_baz-abc", ok: true},
+		{key: "foo.*.bar", ok: false},
+		{key: "foo.>", ok: false},
+		{key: ">", ok: false},
+		{key: "*", ok: false},
+		{key: "foo!", ok: false},
+		{key: "foo bar", ok: false},
+		{key: "", ok: false},
+		{key: " ", ok: false},
+		{key: ".", ok: false},
+		{key: ".foo", ok: false},
+		{key: "foo.", ok: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.key, func(t *testing.T) {
+			res := keyValid(test.key)
+			if res != test.ok {
+				t.Fatalf("Invalid result; want: %v; got: %v", test.ok, res)
+			}
+		})
+	}
+}
+
+func TestKV_searchKeyValid(t *testing.T) {
+	tests := []struct {
+		key string
+		ok  bool
+	}{
+		{key: "foo123", ok: true},
+		{key: "foo.bar", ok: true},
+		{key: "Foo.123=bar_baz-abc", ok: true},
+		{key: "foo.*.bar", ok: true},
+		{key: "foo.>", ok: true},
+		{key: ">", ok: true},
+		{key: "*", ok: true},
+		{key: "foo!", ok: false},
+		{key: "foo bar", ok: false},
+		{key: "", ok: false},
+		{key: " ", ok: false},
+		{key: ".", ok: false},
+		{key: ".foo", ok: false},
+		{key: "foo.", ok: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.key, func(t *testing.T) {
+			res := searchKeyValid(test.key)
+			if res != test.ok {
+				t.Fatalf("Invalid result; want: %v; got: %v", test.ok, res)
+			}
+		})
+	}
+}
+
+func TestKV_bucketValid(t *testing.T) {
+	tests := []struct {
+		key string
+		ok  bool
+	}{
+		{key: "foo123", ok: true},
+		{key: "Foo123-bar_baz", ok: true},
+		{key: "foo.bar", ok: false},
+		{key: "foo.*.bar", ok: false},
+		{key: "foo.>", ok: false},
+		{key: ">", ok: false},
+		{key: "*", ok: false},
+		{key: "foo!", ok: false},
+		{key: "foo bar", ok: false},
+		{key: "", ok: false},
+		{key: " ", ok: false},
+		{key: ".", ok: false},
+		{key: ".foo", ok: false},
+		{key: "foo.", ok: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.key, func(t *testing.T) {
+			res := bucketValid(test.key)
+			if res != test.ok {
+				t.Fatalf("Invalid result; want: %v; got: %v", test.ok, res)
+			}
+		})
+	}
+}
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index b6ad3a6..26a038c 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -514,6 +514,34 @@ func TestKeyValueWatch(t *testing.T) {
 		expectUpdate("age", "22", 3)
 		expectUpdate("name2", "ik", 4)
 	})
+
+	t.Run("invalid watchers", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH"})
+		expectOk(t, err)
+
+		// empty keys
+		_, err = kv.Watch(ctx, "")
+		expectErr(t, err, jetstream.ErrInvalidKey)
+
+		// invalid key
+		_, err = kv.Watch(ctx, "a.>.b")
+		expectErr(t, err, jetstream.ErrInvalidKey)
+
+		_, err = kv.Watch(ctx, "foo.")
+		expectErr(t, err, jetstream.ErrInvalidKey)
+
+		// conflicting options
+		_, err = kv.Watch(ctx, "foo", jetstream.IncludeHistory(), jetstream.UpdatesOnly())
+		expectErr(t, err, jetstream.ErrInvalidOption)
+	})
 }
 
 func TestKeyValueWatchContext(t *testing.T) {
@@ -1634,3 +1662,58 @@ func TestKeyValueCompression(t *testing.T) {
 		t.Fatalf("Expected stream to be compressed with S2")
 	}
 }
+
+func TestKeyValueCreateRepairOldKV(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx := context.Background()
+
+	// create a standard kv
+	_, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
+		Bucket: "A",
+	})
+	if err != nil {
+		t.Fatalf("Error creating kv: %v", err)
+	}
+
+	// get stream config and set discard policy to old and AllowDirect to false
+	stream, err := js.Stream(ctx, "KV_A")
+	if err != nil {
+		t.Fatalf("Error getting stream info: %v", err)
+	}
+	streamCfg := stream.CachedInfo().Config
+	streamCfg.Discard = jetstream.DiscardOld
+	streamCfg.AllowDirect = false
+
+	// create a new kv with the same name - client should fix the config
+	_, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
+		Bucket: "A",
+	})
+	if err != nil {
+		t.Fatalf("Error creating kv: %v", err)
+	}
+
+	// get stream config again and check if the discard policy is set to new
+	stream, err = js.Stream(ctx, "KV_A")
+	if err != nil {
+		t.Fatalf("Error getting stream info: %v", err)
+	}
+	if stream.CachedInfo().Config.Discard != jetstream.DiscardNew {
+		t.Fatalf("Expected stream to have discard policy set to new")
+	}
+	if !stream.CachedInfo().Config.AllowDirect {
+		t.Fatalf("Expected stream to have AllowDirect set to true")
+	}
+
+	// attempting to create a new kv with the same name and different settings should fail
+	_, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
+		Bucket:      "A",
+		Description: "New KV",
+	})
+	if !errors.Is(err, jetstream.ErrBucketExists) {
+		t.Fatalf("Expected error to be ErrBucketExists, got: %v", err)
+	}
+}
diff --git a/jetstream/test/stream_test.go b/jetstream/test/stream_test.go
index ca10090..b7278b3 100644
--- a/jetstream/test/stream_test.go
+++ b/jetstream/test/stream_test.go
@@ -193,6 +193,16 @@ func TestCreateConsumer(t *testing.T) {
 			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", ""}},
 			withError:      jetstream.ErrEmptyFilter,
 		},
+		{
+			name:           "with invalid filter subject, leading dot",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubject: ".foo"},
+			withError:      jetstream.ErrInvalidSubject,
+		},
+		{
+			name:           "with invalid filter subject, trailing dot",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubject: "foo."},
+			withError:      jetstream.ErrInvalidSubject,
+		},
 		{
 			name:           "consumer already exists, error",
 			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", Description: "test consumer"},
diff --git a/test/js_test.go b/test/js_test.go
index a77c399..0363706 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -2604,6 +2604,15 @@ func TestJetStreamManagement(t *testing.T) {
 			}
 		})
 
+		t.Run("with invalid filter subject", func(t *testing.T) {
+			if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Name: "tc", FilterSubject: ".foo"}); !errors.Is(err, nats.ErrInvalidFilterSubject) {
+				t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidFilterSubject, err)
+			}
+			if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Name: "tc", FilterSubject: "foo."}); !errors.Is(err, nats.ErrInvalidFilterSubject) {
+				t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidFilterSubject, err)
+			}
+		})
+
 		t.Run("with invalid consumer name", func(t *testing.T) {
 			if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "test.durable"}); err != nats.ErrInvalidConsumerName {
 				t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidConsumerName, err)
diff --git a/test/kv_test.go b/test/kv_test.go
index 0654ccb..0c89185 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -362,6 +362,28 @@ func TestKeyValueWatch(t *testing.T) {
 		kv.Put("t.age", []byte("66"))
 		expectUpdate("t.age", "66", 12)
 	})
+
+	t.Run("invalid watchers", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH"})
+		expectOk(t, err)
+
+		// empty keys
+		_, err = kv.Watch("")
+		expectErr(t, err, nats.ErrInvalidKey)
+
+		// invalid key
+		_, err = kv.Watch("a.>.b")
+		expectErr(t, err, nats.ErrInvalidKey)
+
+		_, err = kv.Watch("foo.")
+		expectErr(t, err, nats.ErrInvalidKey)
+	})
 }
 
 func TestKeyValueWatchContext(t *testing.T) {
@@ -1010,7 +1032,7 @@ func expectErr(t *testing.T, err error, expected ...error) {
 		return
 	}
 	for _, e := range expected {
-		if err == e || strings.Contains(e.Error(), err.Error()) {
+		if errors.Is(err, e) {
 			return
 		}
 	}