diff --git a/changelog/unreleased/nats-authentication.md b/changelog/unreleased/nats-authentication.md
new file mode 100644
index 0000000000..36daa6dbb0
--- /dev/null
+++ b/changelog/unreleased/nats-authentication.md
@@ -0,0 +1,5 @@
+Enhancement: Allow authentication for nats connections
+
+Allows configuring username/password for nats connections
+
+https://github.com/cs3org/reva/pull/4412
diff --git a/pkg/storage/cache/cache.go b/pkg/storage/cache/cache.go
index c02d246242..0ff7229c42 100644
--- a/pkg/storage/cache/cache.go
+++ b/pkg/storage/cache/cache.go
@@ -51,6 +51,8 @@ type Config struct {
 	TTL                time.Duration `mapstructure:"cache_ttl"`
 	Size               int           `mapstructure:"cache_size"`
 	DisablePersistence bool          `mapstructure:"cache_disable_persistence"`
+	AuthUsername       string        `mapstructure:"cache_auth_username"`
+	AuthPassword       string        `mapstructure:"cache_auth_password"`
 }
 
 // Cache handles key value operations on caches
@@ -240,5 +242,6 @@ func getStore(cfg Config) microstore.Store {
 		store.TTL(cfg.TTL),
 		store.Size(cfg.Size),
 		store.DisablePersistence(cfg.DisablePersistence),
+		store.Authentication(cfg.AuthUsername, cfg.AuthPassword),
 	)
 }
diff --git a/pkg/storage/utils/decomposedfs/decomposedfs.go b/pkg/storage/utils/decomposedfs/decomposedfs.go
index 23384d5536..b6ee40d295 100644
--- a/pkg/storage/utils/decomposedfs/decomposedfs.go
+++ b/pkg/storage/utils/decomposedfs/decomposedfs.go
@@ -145,6 +145,7 @@ func NewDefault(m map[string]interface{}, bs tree.Blobstore, es events.Stream) (
 		microstore.Database(o.IDCache.Database),
 		microstore.Table(o.IDCache.Table),
 		store.DisablePersistence(o.IDCache.DisablePersistence),
+		store.Authentication(o.IDCache.AuthUsername, o.IDCache.AuthPassword),
 	))
 
 	permissionsSelector, err := pool.PermissionsSelector(o.PermissionsSVC, pool.WithTLSMode(o.PermTLSMode))
diff --git a/pkg/store/options.go b/pkg/store/options.go
index c4e177e546..75ba99c6bb 100644
--- a/pkg/store/options.go
+++ b/pkg/store/options.go
@@ -89,3 +89,17 @@ func DisablePersistence(val bool) store.Option {
 		o.Context = context.WithValue(o.Context, disablePersistanceContextKey{}, val)
 	}
 }
+
+type authenticationContextKey struct{}
+
+// Authentication configures the username and password to use for authentication.
+// Only supported by the `natsjskv` implementation.
+func Authentication(username, password string) store.Option {
+	return func(o *store.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+
+		o.Context = context.WithValue(o.Context, authenticationContextKey{}, []string{username, password})
+	}
+}
diff --git a/pkg/store/store.go b/pkg/store/store.go
index 67ef00dd2b..f26edff2e2 100644
--- a/pkg/store/store.go
+++ b/pkg/store/store.go
@@ -127,6 +127,10 @@ func Create(opts ...microstore.Option) microstore.Store {
 		// host, port, clusterid
 		natsOptions := nats.GetDefaultOptions()
 		natsOptions.Name = "TODO" // we can pass in the service name to allow identifying the client, but that requires adding a custom context option
+		if auth, ok := options.Context.Value(authenticationContextKey{}).([]string); ok && len(auth) == 2 {
+			natsOptions.User = auth[0]
+			natsOptions.Password = auth[1]
+		}
 		return natsjs.NewStore(
 			append(opts,
 				natsjs.NatsOptions(natsOptions), // always pass in properly initialized default nats options
@@ -141,6 +145,10 @@ func Create(opts ...microstore.Option) microstore.Store {
 
 		natsOptions := nats.GetDefaultOptions()
 		natsOptions.Name = "TODO" // we can pass in the service name to allow identifying the client, but that requires adding a custom context option
+		if auth, ok := options.Context.Value(authenticationContextKey{}).([]string); ok && len(auth) == 2 {
+			natsOptions.User = auth[0]
+			natsOptions.Password = auth[1]
+		}
 		return natsjskv.NewStore(
 			append(opts,
 				natsjs.NatsOptions(natsOptions), // always pass in properly initialized default nats options