feat(redis): add mTLS support for Redis connection initialization (#6738)

* feat(redis): add mTLS support for Redis connection initialization

- Enhanced the Redis2Store initialization to support mutual TLS (mTLS) by adding configuration options for CA certificate, client certificate, and client key paths.
- Updated the Redis client setup to use TLS configuration when mTLS is enabled, ensuring secure connections to the Redis server.

* feat(redis): extend Redis3Store initialization to support mTLS

- Added configuration options for enabling mutual TLS (mTLS) in Redis3Store.
- Implemented logic to load client certificates and CA certificates for secure Redis connections.
- Updated the Redis client setup to utilize TLS configuration when mTLS is enabled.

---------

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

2 files changed, 102 insertions, 12 deletions

diff --git a/weed/filer/redis2/redis_store.go b/weed/filer/redis2/redis_store.go
index a271dbcd1..5e7bc019e 100644
--- a/weed/filer/redis2/redis_store.go
+++ b/weed/filer/redis2/redis_store.go
@@ -1,8 +1,14 @@
 package redis2
 
 import (
+	"crypto/tls"
+	"crypto/x509"
+	"net"
+	"os"
+
 	"github.com/redis/go-redis/v9"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
@@ -24,15 +30,54 @@ func (store *Redis2Store) Initialize(configuration util.Configuration, prefix st
 		configuration.GetString(prefix+"password"),
 		configuration.GetInt(prefix+"database"),
 		configuration.GetStringSlice(prefix+"superLargeDirectories"),
+		configuration.GetBool(prefix+"enable_mtls"),
+		configuration.GetString(prefix+"ca_cert_path"),
+		configuration.GetString(prefix+"client_cert_path"),
+		configuration.GetString(prefix+"client_key_path"),
 	)
 }
 
-func (store *Redis2Store) initialize(hostPort string, password string, database int, superLargeDirectories []string) (err error) {
-	store.Client = redis.NewClient(&redis.Options{
-		Addr:     hostPort,
-		Password: password,
-		DB:       database,
-	})
+func (store *Redis2Store) initialize(hostPort string, password string, database int, superLargeDirectories []string, enableMtls bool, caCertPath string, clientCertPath string, clientKeyPath string) (err error) {
+	if enableMtls {
+		clientCert, err := tls.LoadX509KeyPair(clientCertPath, clientKeyPath)
+		if err != nil {
+			glog.Fatalf("Error loading client certificate and key pair: %v", err)
+		}
+
+		caCertBytes, err := os.ReadFile(caCertPath)
+		if err != nil {
+			glog.Fatalf("Error reading CA certificate file: %v", err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		if ok := caCertPool.AppendCertsFromPEM(caCertBytes); !ok {
+			glog.Fatalf("Error appending CA certificate to pool")
+		}
+
+		redisHost, _, err := net.SplitHostPort(hostPort)
+		if err != nil {
+			glog.Fatalf("Error parsing redis host and port from %s: %v", hostPort, err)
+		}
+
+		tlsConfig := &tls.Config{
+			Certificates: []tls.Certificate{clientCert},
+			RootCAs: caCertPool,
+			ServerName: redisHost,
+			MinVersion: tls.VersionTLS12,
+		}
+		store.Client = redis.NewClient(&redis.Options{
+			Addr:     hostPort,
+			Password: password,
+			DB:       database,
+			TLSConfig: tlsConfig,
+		})
+	} else {
+		store.Client = redis.NewClient(&redis.Options{
+			Addr:     hostPort,
+			Password: password,
+			DB:       database,
+		})
+	}
 	store.loadSuperLargeDirectories(superLargeDirectories)
 	return
 }
diff --git a/weed/filer/redis3/redis_store.go b/weed/filer/redis3/redis_store.go
index fbcb6d10d..3bb0ce46f 100644
--- a/weed/filer/redis3/redis_store.go
+++ b/weed/filer/redis3/redis_store.go
@@ -1,10 +1,16 @@
 package redis3
 
 import (
+	"crypto/tls"
+	"crypto/x509"
+	"net"
+	"os"
+
 	"github.com/go-redsync/redsync/v4"
 	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
 	"github.com/redis/go-redis/v9"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
@@ -25,15 +31,54 @@ func (store *Redis3Store) Initialize(configuration util.Configuration, prefix st
 		configuration.GetString(prefix+"address"),
 		configuration.GetString(prefix+"password"),
 		configuration.GetInt(prefix+"database"),
+		configuration.GetBool(prefix+"enable_mtls"),
+		configuration.GetString(prefix+"ca_cert_path"),
+		configuration.GetString(prefix+"client_cert_path"),
+		configuration.GetString(prefix+"client_key_path"),
 	)
 }
 
-func (store *Redis3Store) initialize(hostPort string, password string, database int) (err error) {
-	store.Client = redis.NewClient(&redis.Options{
-		Addr:     hostPort,
-		Password: password,
-		DB:       database,
-	})
+func (store *Redis3Store) initialize(hostPort string, password string, database int, enableMtls bool, caCertPath string, clientCertPath string, clientKeyPath string) (err error) {
+	if enableMtls {
+		clientCert, err := tls.LoadX509KeyPair(clientCertPath, clientKeyPath)
+		if err != nil {
+			glog.Fatalf("Error loading client certificate and key pair: %v", err)
+		}
+
+		caCertBytes, err := os.ReadFile(caCertPath)
+		if err != nil {
+			glog.Fatalf("Error reading CA certificate file: %v", err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		if ok := caCertPool.AppendCertsFromPEM(caCertBytes); !ok {
+			glog.Fatalf("Error appending CA certificate to pool")
+		}
+
+		redisHost, _, err := net.SplitHostPort(hostPort)
+		if err != nil {
+			glog.Fatalf("Error parsing redis host and port from %s: %v", hostPort, err)
+		}
+
+		tlsConfig := &tls.Config{
+			Certificates: []tls.Certificate{clientCert},
+			RootCAs:      caCertPool,
+			ServerName:   redisHost,
+			MinVersion:   tls.VersionTLS12,
+		}
+		store.Client = redis.NewClient(&redis.Options{
+			Addr:      hostPort,
+			Password:  password,
+			DB:        database,
+			TLSConfig: tlsConfig,
+		})
+	} else {
+		store.Client = redis.NewClient(&redis.Options{
+			Addr:     hostPort,
+			Password: password,
+			DB:       database,
+		})
+	}
 	store.redsync = redsync.New(goredis.NewPool(store.Client))
 	return
 }