filer.redis3: accept username and keyPrefix options (#12192)

* filer.redis3: prefix every key via getKey, honoring keyPrefix

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.redis3: read username and keyPrefix options into client config

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.redis3: reject a username configured without a password

go-redis only sends AUTH when a password is set, so a configured username
with an empty password silently connects as the default user instead of
the intended ACL user. Fail loudly at initialize instead.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.redis3: cover keyPrefix scoping in the store tests

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.redis3: derive the test prefix and kv key from the per-run directory

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/filer/redis3/redis_cluster_store.go

@@ -36,7 +36,9 @@ func (store *RedisCluster3Store) Initialize(configuration util.Configuration, pr
 
 	return store.initialize(
 		configuration.GetStringSlice(prefix+"addresses"),
+		configuration.GetString(prefix+"username"),
 		configuration.GetString(prefix+"password"),
+		configuration.GetString(prefix+"keyPrefix"),
 		configuration.GetBool(prefix+"useReadOnly"),
 		configuration.GetBool(prefix+"routeByLatency"),
 		tlsConfig,
@@ -44,9 +46,13 @@ func (store *RedisCluster3Store) Initialize(configuration util.Configuration, pr
 	)
 }
 
-func (store *RedisCluster3Store) initialize(addresses []string, password string, readOnly, routeByLatency bool, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
+func (store *RedisCluster3Store) initialize(addresses []string, username string, password string, keyPrefix string, readOnly, routeByLatency bool, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
+	if err := checkUserPassword(username, password); err != nil {
+		return err
+	}
 	options := &redis.ClusterOptions{
 		Addrs:          addresses,
+		Username:       username,
 		Password:       password,
 		ReadOnly:       readOnly,
 		RouteByLatency: routeByLatency,
@@ -55,5 +61,6 @@ func (store *RedisCluster3Store) initialize(addresses []string, password string,
 	settings.ApplyToCluster(options)
 	store.Client = redis.NewClusterClient(options)
 	store.redsync = redsync.New(goredis.NewPool(store.Client))
+	store.keyPrefix = keyPrefix
 	return
 }

weed/filer/redis3/redis_sentinel_store.go

@@ -37,12 +37,19 @@ func (store *Redis3SentinelStore) Initialize(configuration util.Configuration, p
 		configuration.GetString(prefix+"sentinel_username"),
 		configuration.GetString(prefix+"sentinel_password"),
 		configuration.GetInt(prefix+"database"),
+		configuration.GetString(prefix+"keyPrefix"),
 		tlsConfig,
 		redis_conf.Read(configuration, prefix),
 	)
 }
 
-func (store *Redis3SentinelStore) initialize(addresses []string, masterName string, username string, password string, sentinelUsername string, sentinelPassword string, database int, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
+func (store *Redis3SentinelStore) initialize(addresses []string, masterName string, username string, password string, sentinelUsername string, sentinelPassword string, database int, keyPrefix string, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
+	if err := checkUserPassword(username, password); err != nil {
+		return err
+	}
+	if err := checkUserPassword(sentinelUsername, sentinelPassword); err != nil {
+		return err
+	}
 	options := &redis.FailoverOptions{
 		MasterName:       masterName,
 		SentinelAddrs:    addresses,
@@ -56,5 +63,6 @@ func (store *Redis3SentinelStore) initialize(addresses []string, masterName stri
 	settings.ApplyToFailover(options)
 	store.Client = redis.NewFailoverClient(options)
 	store.redsync = redsync.New(goredis.NewPool(store.Client))
+	store.keyPrefix = keyPrefix
 	return
 }

weed/filer/redis3/redis_store.go

@@ -31,16 +31,22 @@ func (store *Redis3Store) Initialize(configuration util.Configuration, prefix st
 	}
 	return store.initialize(
 		configuration.GetString(prefix+"address"),
+		configuration.GetString(prefix+"username"),
 		configuration.GetString(prefix+"password"),
 		configuration.GetInt(prefix+"database"),
+		configuration.GetString(prefix+"keyPrefix"),
 		tlsConfig,
 		redis_conf.Read(configuration, prefix),
 	)
 }
 
-func (store *Redis3Store) initialize(hostPort string, password string, database int, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
+func (store *Redis3Store) initialize(hostPort string, username string, password string, database int, keyPrefix string, tlsConfig *tls.Config, settings redis_conf.Settings) (err error) {
+	if err := checkUserPassword(username, password); err != nil {
+		return err
+	}
 	options := &redis.Options{
 		Addr:      hostPort,
+		Username:  username,
 		Password:  password,
 		DB:        database,
 		TLSConfig: tlsConfig,
@@ -48,5 +54,6 @@ func (store *Redis3Store) initialize(hostPort string, password string, database
 	settings.ApplyTo(options)
 	store.Client = redis.NewClient(options)
 	store.redsync = redsync.New(goredis.NewPool(store.Client))
+	store.keyPrefix = keyPrefix
 	return
 }

weed/filer/redis3/universal_redis_store.go

@@ -19,8 +19,26 @@ const (
 )
 
 type UniversalRedis3Store struct {
-	Client  redis.UniversalClient
-	redsync *redsync.Redsync
+	Client    redis.UniversalClient
+	redsync   *redsync.Redsync
+	keyPrefix string
+}
+
+func (store *UniversalRedis3Store) getKey(key string) string {
+	if store.keyPrefix == "" {
+		return key
+	}
+	return store.keyPrefix + key
+}
+
+// checkUserPassword rejects a username with an empty password: go-redis only
+// sends AUTH when a password is set, so the connection would silently run as
+// the default user instead of the configured ACL user.
+func checkUserPassword(username, password string) error {
+	if username != "" && password == "" {
+		return fmt.Errorf("username %q requires a password: with an empty password no AUTH is sent and the connection uses the default user", username)
+	}
+	return nil
 }
 
 func (store *UniversalRedis3Store) BeginTransaction(ctx context.Context) (context.Context, error) {
@@ -42,7 +60,7 @@ func (store *UniversalRedis3Store) InsertEntry(ctx context.Context, entry *filer
 	dir, name := entry.FullPath.DirAndName()
 
 	if name != "" {
-		if err = insertChild(ctx, store, genDirectoryListKey(dir), name); err != nil {
+		if err = insertChild(ctx, store, store.getKey(genDirectoryListKey(dir)), name); err != nil {
 			return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
 		}
 	}
@@ -60,7 +78,7 @@ func (store *UniversalRedis3Store) doInsertEntry(ctx context.Context, entry *fil
 		value = util.MaybeGzipData(value)
 	}
 
-	if err = store.Client.Set(ctx, string(entry.FullPath), value, time.Duration(entry.TtlSec)*time.Second).Err(); err != nil {
+	if err = store.Client.Set(ctx, store.getKey(string(entry.FullPath)), value, time.Duration(entry.TtlSec)*time.Second).Err(); err != nil {
 		return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
 	}
 	return nil
@@ -73,7 +91,7 @@ func (store *UniversalRedis3Store) UpdateEntry(ctx context.Context, entry *filer
 
 func (store *UniversalRedis3Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
 
-	data, err := store.Client.Get(ctx, string(fullpath)).Result()
+	data, err := store.Client.Get(ctx, store.getKey(string(fullpath))).Result()
 	if err == redis.Nil {
 		return nil, filer_pb.ErrNotFound
 	}
@@ -99,7 +117,7 @@ func (store *UniversalRedis3Store) DeleteEntry(ctx context.Context, fullpath uti
 	// children it describes. Dropping it here would also discard an entry that
 	// arrived after the caller judged this directory empty, and nothing else
 	// records that the entry is there.
-	_, err = store.Client.Del(ctx, string(fullpath)).Result()
+	_, err = store.Client.Del(ctx, store.getKey(string(fullpath))).Result()
 	if err != nil {
 		return fmt.Errorf("delete %s : %v", fullpath, err)
 	}
@@ -107,7 +125,7 @@ func (store *UniversalRedis3Store) DeleteEntry(ctx context.Context, fullpath uti
 	dir, name := fullpath.DirAndName()
 
 	if name != "" {
-		if err = removeChild(ctx, store, genDirectoryListKey(dir), name); err != nil {
+		if err = removeChild(ctx, store, store.getKey(genDirectoryListKey(dir)), name); err != nil {
 			return fmt.Errorf("DeleteEntry %s in parent dir: %v", fullpath, err)
 		}
 	}
@@ -117,15 +135,15 @@ func (store *UniversalRedis3Store) DeleteEntry(ctx context.Context, fullpath uti
 
 func (store *UniversalRedis3Store) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
 
-	dirListKey := genDirectoryListKey(string(fullpath))
+	dirListKey := store.getKey(genDirectoryListKey(string(fullpath)))
 	if err = removeChildren(ctx, store, dirListKey, func(name string) error {
 		path := util.NewFullPath(string(fullpath), name)
-		_, err = store.Client.Del(ctx, string(path)).Result()
+		_, err = store.Client.Del(ctx, store.getKey(string(path))).Result()
 		if err != nil {
 			return fmt.Errorf("DeleteFolderChildren %s in parent dir: %v", fullpath, err)
 		}
 		// not efficient, but need to remove if it is a directory
-		store.Client.Del(ctx, genDirectoryListKey(string(path)))
+		store.Client.Del(ctx, store.getKey(genDirectoryListKey(string(path))))
 		return nil
 	}); err != nil {
 		return err
@@ -145,7 +163,7 @@ func (store *UniversalRedis3Store) ListDirectoryPrefixedEntries(ctx context.Cont
 
 func (store *UniversalRedis3Store) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 
-	dirListKey := genDirectoryListKey(string(dirPath))
+	dirListKey := store.getKey(genDirectoryListKey(string(dirPath)))
 	counter := int64(0)
 
 	var callbackErr error
@@ -167,7 +185,7 @@ func (store *UniversalRedis3Store) ListDirectoryEntries(ctx context.Context, dir
 		} else {
 			if entry.TtlSec > 0 {
 				if entry.Attr.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
-					store.Client.Del(ctx, string(path)).Result()
+					store.Client.Del(ctx, store.getKey(string(path))).Result()
 					store.Client.ZRem(ctx, dirListKey, fileName).Result()
 					return true
 				}

weed/filer/redis3/universal_redis_store_kv.go

@@ -10,7 +10,7 @@ import (
 
 func (store *UniversalRedis3Store) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
 
-	_, err = store.Client.Set(ctx, string(key), value, 0).Result()
+	_, err = store.Client.Set(ctx, store.getKey(string(key)), value, 0).Result()
 
 	if err != nil {
 		return fmt.Errorf("kv put: %w", err)
@@ -21,7 +21,7 @@ func (store *UniversalRedis3Store) KvPut(ctx context.Context, key []byte, value
 
 func (store *UniversalRedis3Store) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
 
-	data, err := store.Client.Get(ctx, string(key)).Result()
+	data, err := store.Client.Get(ctx, store.getKey(string(key))).Result()
 
 	if err == redis.Nil {
 		return nil, filer.ErrKvNotFound
@@ -32,7 +32,7 @@ func (store *UniversalRedis3Store) KvGet(ctx context.Context, key []byte) (value
 
 func (store *UniversalRedis3Store) KvDelete(ctx context.Context, key []byte) (err error) {
 
-	_, err = store.Client.Del(ctx, string(key)).Result()
+	_, err = store.Client.Del(ctx, store.getKey(string(key))).Result()
 
 	if err != nil {
 		return fmt.Errorf("kv delete: %w", err)

weed/filer/redis3/universal_redis_store_test.go

@@ -127,3 +127,70 @@ func TestRemovingTheLastChildDropsTheListing(t *testing.T) {
 		t.Errorf("listing key should be gone once the last child is removed, still present")
 	}
 }
+
+// A configured keyPrefix must scope every key the store writes: entry values,
+// directory listings (and their skiplist nodes and locks), and Kv keys. The
+// raw-key assertions below would still pass a round-trip check if reads and
+// writes both dropped the prefix, so they check the actual key space.
+func TestKeyPrefixScopesAllKeys(t *testing.T) {
+	store, dir := newTestStore(t)
+	store.keyPrefix = string(dir) + "_"
+	kvKey := string(dir) + "_k"
+	ctx := context.Background()
+
+	now := time.Now()
+	child := dir.Child("obj")
+	if err := store.InsertEntry(ctx, &filer.Entry{
+		FullPath: child,
+		Attr:     filer.Attr{Crtime: now, Mtime: now, Mode: 0644},
+	}); err != nil {
+		t.Fatalf("InsertEntry %s: %v", child, err)
+	}
+	if _, err := store.FindEntry(ctx, child); err != nil {
+		t.Fatalf("FindEntry %s: %v", child, err)
+	}
+	if names := listNames(t, store, dir); len(names) != 1 || names[0] != "obj" {
+		t.Fatalf("list %s: got %v", dir, names)
+	}
+
+	for _, raw := range []string{string(child), genDirectoryListKey(string(dir))} {
+		if n, err := store.Client.Exists(ctx, raw).Result(); err != nil || n != 0 {
+			t.Fatalf("key %q should not exist unprefixed: n=%d err=%v", raw, n, err)
+		}
+		if n, err := store.Client.Exists(ctx, store.keyPrefix+raw).Result(); err != nil || n != 1 {
+			t.Fatalf("key %q should exist under the prefix: n=%d err=%v", raw, n, err)
+		}
+	}
+
+	if err := store.KvPut(ctx, []byte(kvKey), []byte("v")); err != nil {
+		t.Fatalf("KvPut: %v", err)
+	}
+	if v, err := store.KvGet(ctx, []byte(kvKey)); err != nil || string(v) != "v" {
+		t.Fatalf("KvGet: v=%q err=%v", v, err)
+	}
+	if _, err := store.Client.Get(ctx, kvKey).Result(); err != redis.Nil {
+		t.Fatalf("kv key should not exist unprefixed: err=%v", err)
+	}
+	if v, err := store.Client.Get(ctx, store.keyPrefix+kvKey).Result(); err != nil || v != "v" {
+		t.Fatalf("kv key should exist under the prefix: v=%q err=%v", v, err)
+	}
+	if err := store.KvDelete(ctx, []byte(kvKey)); err != nil {
+		t.Fatalf("KvDelete: %v", err)
+	}
+	if _, err := store.KvGet(ctx, []byte(kvKey)); err != filer.ErrKvNotFound {
+		t.Fatalf("KvGet after delete: err=%v", err)
+	}
+
+	if err := store.DeleteFolderChildren(ctx, dir); err != nil {
+		t.Fatalf("DeleteFolderChildren %s: %v", dir, err)
+	}
+	if names := listNames(t, store, dir); len(names) != 0 {
+		t.Fatalf("listing should be empty once the children are deleted, got %v", names)
+	}
+	if err := store.DeleteEntry(ctx, dir); err != nil {
+		t.Fatalf("DeleteEntry %s: %v", dir, err)
+	}
+	if keys, err := store.Client.Keys(ctx, store.keyPrefix+"*").Result(); err != nil || len(keys) != 0 {
+		t.Fatalf("no prefixed keys should remain after cleanup: keys=%v err=%v", keys, err)
+	}
+}