valkey: enable tls

CONTEXT.md

@@ -223,6 +223,10 @@ Proxy-level config (`TLSClientAuth`) requesting or requiring client certificates
 **Outbound mTLS to Backends**:
 Proxy-level config (`EnableMTLS`) presenting a client certificate when connecting to upstream backends, with hot cert rotation.
 
+**Valkey Swarm mTLS**:
+Mutual-TLS transport between skipper (the client) and a valkey swarm shard: skipper presents a client certificate and verifies the valkey server certificate against a configured CA bundle. Configured via `-swarm-valkey-enable-tls` with a static keypair (no rotation).
+_Avoid_: valkey TLS, redis TLS
+
 ## Kubernetes Integration
 
 **Ingress**:

config/config.go

@@ -361,6 +361,13 @@ type Config struct {
 	SwarmValkeyDialTimeout        time.Duration `yaml:"swarm-valkey-dial-timeout"`
 	SwarmValkeyKeepAlive          time.Duration `yaml:"swarm-valkey-keepalive"`
 	SwarmValkeyUpdateInterval     time.Duration `yaml:"swarm-valkey-update-interval"`
+	SwarmEnableTLS                bool          `yaml:"swarm-enable-tls"`
+	// swarm TLS
+	SwarmCaFile         string         `yaml:"swarm-ca"`
+	SwarmClientCertFile string         `yaml:"swarm-client-cert"`
+	SwarmClientKeyFile  string         `yaml:"swarm-client-key"`
+	SwarmCA             *x509.CertPool `yaml:"-"`
+
 	// swim based
 	SwarmKubernetesNamespace          string        `yaml:"swarm-namespace"`
 	SwarmKubernetesLabelSelectorKey   string        `yaml:"swarm-label-selector-key"`
@@ -776,6 +783,10 @@ func NewConfig() *Config {
 	flag.DurationVar(&cfg.SwarmValkeyDialTimeout, "swarm-valkey-dial-timeout", net.DefaultDialTimeout, "set valkey client dial timeout")
 	flag.DurationVar(&cfg.SwarmValkeyKeepAlive, "swarm-valkey-keepalive", net.DefaultKeepAlive, "set valkey keepalive probes interval")
 	flag.DurationVar(&cfg.SwarmValkeyUpdateInterval, "swarm-valkey-update-interval", net.DefaultUpdateInterval, "set update interval to update valkey addresses")
+	flag.StringVar(&cfg.SwarmClientCertFile, "swarm-client-cert", "", "valkey client certificate")
+	flag.StringVar(&cfg.SwarmClientKeyFile, "swarm-client-key", "", "valkey client key")
+	flag.StringVar(&cfg.SwarmCaFile, "swarm-ca", "", "Comma-separated CA bundle file paths used to verify the valkey server certificate")
+	flag.BoolVar(&cfg.SwarmEnableTLS, "swarm-enable-tls", false, "Enables mutual TLS for the valkey swarm connection. It uses -swarm-valkey-client-cert and -swarm-valkey-client-key as the client keypair and -swarm-valkey-ca to verify the valkey server certificate. It only supports one cert and one key file.")
 	// swim
 	flag.StringVar(&cfg.SwarmKubernetesNamespace, "swarm-namespace", swarm.DefaultNamespace, "Kubernetes namespace to find swarm peer instances")
 	flag.StringVar(&cfg.SwarmKubernetesLabelSelectorKey, "swarm-label-selector-key", swarm.DefaultLabelSelectorKey, "Kubernetes labelselector key to find swarm peer instances")
@@ -941,6 +952,21 @@ func (c *Config) ParseArgs(progname string, args []string) error {
 		c.Certificates = certificates
 	}
 
+	if c.SwarmCaFile != "" {
+		if c.SwarmCA == nil {
+			c.SwarmCA = x509.NewCertPool()
+		}
+		for f := range strings.SplitSeq(c.SwarmCaFile, ",") {
+			pem, err := os.ReadFile(f)
+			if err != nil {
+				return fmt.Errorf("valkey failed to read %q: %v", f, err)
+			}
+			if !c.SwarmCA.AppendCertsFromPEM(pem) {
+				return fmt.Errorf("valkey failed to append CA cert %q", f)
+			}
+		}
+	}
+
 	if c.TLSKeyLogFile != "" {
 		f, err := os.OpenFile(c.TLSKeyLogFile, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
 		if err != nil {
@@ -1253,6 +1279,11 @@ func (c *Config) ToOptions() skipper.Options {
 		SwarmValkeyDialTimeout:        c.SwarmValkeyDialTimeout,
 		SwarmValkeyKeepAlive:          c.SwarmValkeyKeepAlive,
 		SwarmValkeyUpdateInterval:     c.SwarmValkeyUpdateInterval,
+		SwarmEnableTLS:                c.SwarmEnableTLS,
+		SwarmClientCertFile:           c.SwarmClientCertFile,
+		SwarmClientKeyFile:            c.SwarmClientKeyFile,
+		SwarmCA:                       c.SwarmCA,
+		SwarmClientRefreshInterval:    c.ClientCertRefreshInterval,
 		// swim based
 		SwarmKubernetesNamespace:          c.SwarmKubernetesNamespace,
 		SwarmKubernetesLabelSelectorKey:   c.SwarmKubernetesLabelSelectorKey,

docs/tutorials/ratelimit.md

@@ -197,6 +197,24 @@ If you have [routesrv proxy](https://opensource.zalando.com/skipper/kubernetes/i
 you need to configure Skipper with the flag `-swarm-valkey-remote=http://<routesrv-service-name>.<routesrv-namespace>.svc.cluster.local/swarm/valkey/shards`.
 `Routesrv` will be responsible for collecting Valkey endpoints and Skipper will poll them from it.
 
+#### TLS / mutual TLS
+
+To connect to Valkey over mutual TLS, enable TLS and provide Skipper's client
+keypair plus the CA bundle that signs the Valkey server certificate:
+
+```
+-swarm-valkey-enable-tls
+-swarm-valkey-client-cert=/path/client.crt
+-swarm-valkey-client-key=/path/client.key
+-swarm-valkey-ca=/path/ca.crt
+```
+
+Skipper presents the client certificate to Valkey and verifies the Valkey
+server certificate against the CA bundle. The server certificate must carry a
+SAN matching the address Skipper dials via `-swarm-valkey-urls`. Multiple CA
+files may be given comma-separated. Only one client cert and one key file are
+supported, and the keypair is loaded once at startup (no rotation).
+
 #### Implementation
 
 The implementation use [Valkey-Go

net/httpclient.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"net/http/httptrace"
 	"net/url"
+	"os"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -63,6 +64,9 @@ type CertReloader struct {
 // GetClientCertificate returns the new rotated *tls.Certificate.
 // You have to use Close() in order to not leak a goroutine.
 func NewCertReloader(certFile, keyFile string, interval time.Duration, log logging.Logger) (*CertReloader, error) {
+	if interval <= 0 {
+		interval = defaultRefreshInterval
+	}
 	sp := secrets.NewSecretPaths(interval)
 	if err := sp.Add(certFile); err != nil {
 		sp.Close()
@@ -100,6 +104,14 @@ func NewCertReloader(certFile, keyFile string, interval time.Duration, log loggi
 	go cr.refreshLoop(interval)
 	return cr, nil
 }
+func MustNewCertReloader(certFile, keyFile string, interval time.Duration, log logging.Logger) *CertReloader {
+	cr, err := NewCertReloader(certFile, keyFile, interval, log)
+	if err != nil {
+		log.Errorf("Failed to initialize cert reloader: %v", err)
+		os.Exit(2)
+	}
+	return cr
+}
 
 func (cr *CertReloader) refreshLoop(interval time.Duration) {
 	ticker := time.NewTicker(interval)

net/redisclient.go

@@ -2,6 +2,8 @@ package net
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"log"
 	"sync"
@@ -78,6 +80,22 @@ type RedisOptions struct {
 
 	// HashAlgorithm is one of rendezvous, rendezvousVnodes, jump, mpchash, defaults to github.com/go-redis/redis default
 	HashAlgorithm string
+
+	EnableTLS bool
+
+	CA *x509.CertPool
+
+	// ClientCertFile is the path to a PEM-encoded client certificate for mTLS to backends.
+	// Must be set together with ClientKeyFile. When set, GetClientCertificate is used for cert rotation.
+	ClientCertFile string
+
+	// ClientKeyFile is the path to a PEM-encoded private key for mTLS to backends.
+	// Must be set together with ClientCertFile.
+	ClientKeyFile string
+
+	// ClientCertRefreshInterval is how often ClientCertFile/ClientKeyFile are re-read.
+	// Defaults to 5 minutes if zero.
+	ClientCertRefreshInterval time.Duration
 }
 
 // RedisRingClient is a redis client that does access redis by
@@ -226,6 +244,13 @@ func NewRedisRingClient(ro *RedisOptions) *RedisRingClient {
 			opt.MaintNotificationsConfig = &maintnotifications.Config{
 				Mode: maintnotifications.ModeDisabled,
 			}
+			if ro.EnableTLS && ro.ClientCertFile != "" && ro.ClientKeyFile != "" {
+				cr := MustNewCertReloader(ro.ClientCertFile, ro.ClientKeyFile, ro.ClientCertRefreshInterval, ro.Log)
+				opt.TLSConfig = &tls.Config{
+					GetClientCertificate: cr.GetClientCertificate,
+					RootCAs:              ro.CA,
+				}
+			}
 
 			return redis.NewClient(opt)
 		},

net/valkey.go

@@ -2,6 +2,8 @@ package net
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
 	"fmt"
 	"math"
@@ -74,6 +76,22 @@ type ValkeyOptions struct {
 	// Hook see https://pkg.go.dev/github.com/valkey-io/valkey-go/valkeyhook
 	Hook valkeyhook.Hook
 
+	EnableTLS bool
+
+	CA *x509.CertPool
+
+	// ClientCertFile is the path to a PEM-encoded client certificate for mTLS to backends.
+	// Must be set together with ClientKeyFile. When set, GetClientCertificate is used for cert rotation.
+	ClientCertFile string
+
+	// ClientKeyFile is the path to a PEM-encoded private key for mTLS to backends.
+	// Must be set together with ClientCertFile.
+	ClientKeyFile string
+
+	// ClientCertRefreshInterval is how often ClientCertFile/ClientKeyFile are re-read.
+	// Defaults to 5 minutes if zero.
+	ClientCertRefreshInterval time.Duration
+
 	// EnableOTel enables OpenTelemetry adapter, see https://pkg.go.dev/github.com/valkey-io/valkey-go/valkeyotel
 	EnableOTel bool
 	// OTelOptions
@@ -117,6 +135,14 @@ func createValkeyClient(addr string, opt *ValkeyOptions) (valkey.Client, error)
 		err error
 	)
 
+	if opt.EnableTLS && opt.ClientCertFile != "" && opt.ClientKeyFile != "" {
+		cr := MustNewCertReloader(opt.ClientCertFile, opt.ClientKeyFile, opt.ClientCertRefreshInterval, opt.Log)
+		clientOptions.TLSConfig = &tls.Config{
+			GetClientCertificate: cr.GetClientCertificate,
+			RootCAs:              opt.CA,
+		}
+	}
+
 	if opt.EnableOTel {
 		cli, err = valkeyotel.NewClient(clientOptions, opt.OTelOptions...)
 	} else {

proxy/proxy.go

@@ -842,21 +842,11 @@ func WithParams(p Params) *Proxy {
 	log := &logging.DefaultLog{}
 	var cr *snet.CertReloader
 	if p.EnableMTLS && p.ClientCertFile != "" && p.ClientKeyFile != "" {
-		interval := p.ClientCertRefreshInterval
-		if interval == 0 {
-			interval = 5 * time.Minute
-		}
-		var err error
-		cr, err = snet.NewCertReloader(p.ClientCertFile, p.ClientKeyFile, interval, log)
-		if err != nil {
-			log.Errorf("Failed to initialize cert reloader in proxy: %v", err)
-			os.Exit(2)
-		} else {
-			if tr.TLSClientConfig == nil {
-				tr.TLSClientConfig = &tls.Config{}
-			}
-			tr.TLSClientConfig.GetClientCertificate = cr.GetClientCertificate
+		cr = snet.MustNewCertReloader(p.ClientCertFile, p.ClientKeyFile, p.ClientCertRefreshInterval, log)
+		if tr.TLSClientConfig == nil {
+			tr.TLSClientConfig = &tls.Config{}
 		}
+		tr.TLSClientConfig.GetClientCertificate = cr.GetClientCertificate
 	}
 
 	h2cTr := newTransport(p)

skipper.go

@@ -1153,6 +1153,13 @@ type Options struct {
 	SwarmValkeyKeepAlive          time.Duration
 	SwarmValkeyConnLifetime       time.Duration
 	SwarmValkeyUpdateInterval     time.Duration
+	// swarm TLS
+	SwarmEnableTLS             bool
+	SwarmCA                    *x509.CertPool
+	SwarmClientCertFile        string
+	SwarmClientKeyFile         string
+	SwarmClientRefreshInterval time.Duration
+
 	// swim based swarm
 	SwarmKubernetesNamespace          string
 	SwarmKubernetesLabelSelectorKey   string
@@ -2194,37 +2201,47 @@ func run(o Options, sig chan os.Signal, idleConnsCH chan struct{}) error {
 			log.Infof("Valkey based swarm with %d shards", len(o.SwarmValkeyURLs))
 
 			valkeyOptions = &skpnet.ValkeyOptions{
-				Addrs:            o.SwarmValkeyURLs,
-				UpdateInterval:   o.SwarmValkeyUpdateInterval,
-				Username:         o.SwarmValkeyUsername,
-				Password:         o.SwarmValkeyPassword,
-				ConnWriteTimeout: o.SwarmValkeyConnWriteTimeout,
-				DialTimeout:      o.SwarmValkeyDialTimeout,
-				KeepAlive:        o.SwarmValkeyKeepAlive,
-				ConnLifetime:     o.SwarmValkeyConnLifetime,
-				Tracer:           tracer,
-				Log:              log.New(),
+				Addrs:                     o.SwarmValkeyURLs,
+				UpdateInterval:            o.SwarmValkeyUpdateInterval,
+				Username:                  o.SwarmValkeyUsername,
+				Password:                  o.SwarmValkeyPassword,
+				ConnWriteTimeout:          o.SwarmValkeyConnWriteTimeout,
+				DialTimeout:               o.SwarmValkeyDialTimeout,
+				KeepAlive:                 o.SwarmValkeyKeepAlive,
+				ConnLifetime:              o.SwarmValkeyConnLifetime,
+				EnableTLS:                 o.SwarmEnableTLS,
+				ClientCertFile:            o.SwarmClientCertFile,
+				ClientKeyFile:             o.SwarmClientKeyFile,
+				ClientCertRefreshInterval: o.SwarmClientRefreshInterval,
+				CA:                        o.SwarmCA,
+				Tracer:                    tracer,
+				Log:                       log.New(),
 			}
 
 		} else if len(o.SwarmRedisURLs) > 0 || o.KubernetesRedisServiceName != "" || o.SwarmRedisEndpointsRemoteURL != "" {
 			log.Infof("Redis based swarm with %d shards", len(o.SwarmRedisURLs))
 
 			redisOptions = &skpnet.RedisOptions{
-				Addrs:               o.SwarmRedisURLs,
-				Username:            o.SwarmRedisUsername,
-				Password:            o.SwarmRedisPassword,
-				HashAlgorithm:       o.SwarmRedisHashAlgorithm,
-				DialTimeout:         o.SwarmRedisDialTimeout,
-				ReadTimeout:         o.SwarmRedisReadTimeout,
-				WriteTimeout:        o.SwarmRedisWriteTimeout,
-				PoolTimeout:         o.SwarmRedisPoolTimeout,
-				MinIdleConns:        o.SwarmRedisMinIdleConns,
-				MaxIdleConns:        o.SwarmRedisMaxIdleConns,
-				ConnMetricsInterval: o.SwarmRedisConnMetricsInterval,
-				UpdateInterval:      o.SwarmRedisUpdateInterval,
-				HeartbeatFrequency:  o.SwarmRedisHeartbeatFrequency,
-				Tracer:              tracer,
-				Log:                 log.New(),
+				Addrs:                     o.SwarmRedisURLs,
+				Username:                  o.SwarmRedisUsername,
+				Password:                  o.SwarmRedisPassword,
+				HashAlgorithm:             o.SwarmRedisHashAlgorithm,
+				DialTimeout:               o.SwarmRedisDialTimeout,
+				ReadTimeout:               o.SwarmRedisReadTimeout,
+				WriteTimeout:              o.SwarmRedisWriteTimeout,
+				PoolTimeout:               o.SwarmRedisPoolTimeout,
+				MinIdleConns:              o.SwarmRedisMinIdleConns,
+				MaxIdleConns:              o.SwarmRedisMaxIdleConns,
+				ConnMetricsInterval:       o.SwarmRedisConnMetricsInterval,
+				UpdateInterval:            o.SwarmRedisUpdateInterval,
+				HeartbeatFrequency:        o.SwarmRedisHeartbeatFrequency,
+				EnableTLS:                 o.SwarmEnableTLS,
+				ClientCertFile:            o.SwarmClientCertFile,
+				ClientKeyFile:             o.SwarmClientKeyFile,
+				ClientCertRefreshInterval: o.SwarmClientRefreshInterval,
+				CA:                        o.SwarmCA,
+				Tracer:                    tracer,
+				Log:                       log.New(),
 			}
 
 		} else {