Add flags for enabling TLS for store

cmd/keeper/keeper.go

@@ -60,6 +60,9 @@ type config struct {
 	id                      string
 	storeBackend            string
 	storeEndpoints          string
+	storeCertFile           string
+	storeKeyFile            string
+	storeCACertFile         string
 	dataDir                 string
 	clusterName             string
 	listenAddress           string
@@ -90,6 +93,9 @@ func init() {
 	cmdKeeper.PersistentFlags().StringVar(&cfg.id, "id", "", "keeper id (must be unique in the cluster and can contain only lower-case letters, numbers and the underscore character). If not provided a random id will be generated.")
 	cmdKeeper.PersistentFlags().StringVar(&cfg.storeBackend, "store-backend", "", "store backend type (etcd or consul)")
 	cmdKeeper.PersistentFlags().StringVar(&cfg.storeEndpoints, "store-endpoints", "", "a comma-delimited list of store endpoints (defaults: 127.0.0.1:2379 for etcd, 127.0.0.1:8500 for consul)")
+	cmdKeeper.PersistentFlags().StringVar(&cfg.storeCertFile, "store-cert", "", "path to the client server TLS cert file")
+	cmdKeeper.PersistentFlags().StringVar(&cfg.storeKeyFile, "store-key", "", "path to the client server TLS key file")
+	cmdKeeper.PersistentFlags().StringVar(&cfg.storeCACertFile, "store-cacert", "", "path to the client server TLS trusted CA key file")
 	cmdKeeper.PersistentFlags().StringVar(&cfg.dataDir, "data-dir", "", "data directory")
 	cmdKeeper.PersistentFlags().StringVar(&cfg.clusterName, "cluster-name", "", "cluster name")
 	cmdKeeper.PersistentFlags().StringVar(&cfg.listenAddress, "listen-address", "localhost", "keeper listening address")
@@ -245,7 +251,13 @@ type PostgresKeeper struct {
 func NewPostgresKeeper(id string, cfg *config, stop chan bool, end chan error) (*PostgresKeeper, error) {
 	storePath := filepath.Join(common.StoreBasePath, cfg.clusterName)
 
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create store: %v", err)
 	}

cmd/proxy/proxy.go

@@ -46,20 +46,26 @@ var cmdProxy = &cobra.Command{
 }
 
 type config struct {
-	storeBackend   string
-	storeEndpoints string
-	clusterName    string
-	listenAddress  string
-	port           string
-	stopListening  bool
-	debug          bool
+	storeBackend    string
+	storeEndpoints  string
+	storeCertFile   string
+	storeKeyFile    string
+	storeCACertFile string
+	clusterName     string
+	listenAddress   string
+	port            string
+	stopListening   bool
+	debug           bool
 }
 
 var cfg config
 
 func init() {
 	cmdProxy.PersistentFlags().StringVar(&cfg.storeBackend, "store-backend", "", "store backend type (etcd or consul)")
 	cmdProxy.PersistentFlags().StringVar(&cfg.storeEndpoints, "store-endpoints", "", "a comma-delimited list of store endpoints (defaults: 127.0.0.1:2379 for etcd, 127.0.0.1:8500 for consul)")
+	cmdProxy.PersistentFlags().StringVar(&cfg.storeCertFile, "store-cert", "", "path to the client server TLS cert file")
+	cmdProxy.PersistentFlags().StringVar(&cfg.storeKeyFile, "store-key", "", "path to the client server TLS key file")
+	cmdProxy.PersistentFlags().StringVar(&cfg.storeCACertFile, "store-cacert", "", "path to the client server TLS trusted CA key file")
 	cmdProxy.PersistentFlags().StringVar(&cfg.clusterName, "cluster-name", "", "cluster name")
 	cmdProxy.PersistentFlags().StringVar(&cfg.listenAddress, "listen-address", "127.0.0.1", "proxy listening address")
 	cmdProxy.PersistentFlags().StringVar(&cfg.port, "port", "5432", "proxy listening port")
@@ -83,7 +89,13 @@ type ClusterChecker struct {
 func NewClusterChecker(id string, cfg config) (*ClusterChecker, error) {
 	storePath := filepath.Join(common.StoreBasePath, cfg.clusterName)
 
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create store: %v", err)
 	}

cmd/sentinel/sentinel.go

@@ -58,6 +58,9 @@ const (
 type config struct {
 	storeBackend            string
 	storeEndpoints          string
+	storeCertFile           string
+	storeKeyFile            string
+	storeCACertFile         string
 	clusterName             string
 	listenAddress           string
 	port                    string
@@ -74,6 +77,9 @@ var cfg config
 func init() {
 	cmdSentinel.PersistentFlags().StringVar(&cfg.storeBackend, "store-backend", "", "store backend type (etcd or consul)")
 	cmdSentinel.PersistentFlags().StringVar(&cfg.storeEndpoints, "store-endpoints", "", "a comma-delimited list of store endpoints (defaults: 127.0.0.1:2379 for etcd, 127.0.0.1:8500 for consul)")
+	cmdSentinel.PersistentFlags().StringVar(&cfg.storeCertFile, "store-cert", "", "path to the client server TLS cert file")
+	cmdSentinel.PersistentFlags().StringVar(&cfg.storeKeyFile, "store-key", "", "path to the client server TLS key file")
+	cmdSentinel.PersistentFlags().StringVar(&cfg.storeCACertFile, "store-cacert", "", "path to the client server TLS trusted CA key file")
 	cmdSentinel.PersistentFlags().StringVar(&cfg.clusterName, "cluster-name", "", "cluster name")
 	cmdSentinel.PersistentFlags().StringVar(&cfg.listenAddress, "listen-address", "localhost", "sentinel listening address")
 	cmdSentinel.PersistentFlags().StringVar(&cfg.port, "port", "6431", "sentinel listening port")
@@ -663,7 +669,13 @@ func NewSentinel(id string, cfg *config, stop chan bool, end chan bool) (*Sentin
 	}
 
 	storePath := filepath.Join(common.StoreBasePath, cfg.clusterName)
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create store: %v", err)
 	}

cmd/stolonctl/config_get.go

@@ -55,7 +55,13 @@ func getConfig(e *store.StoreManager) (*cluster.NilConfig, error) {
 func configGet(cmd *cobra.Command, args []string) {
 	storePath := filepath.Join(common.StoreBasePath, cfg.clusterName)
 
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
 	if err != nil {
 		die("cannot create store: %v", err)
 	}

cmd/stolonctl/config_patch.go

@@ -98,7 +98,13 @@ func configPatch(cmd *cobra.Command, args []string) {
 	}
 
 	storePath := filepath.Join(common.StoreBasePath, cfg.clusterName)
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
 	if err != nil {
 		die("cannot create store: %v", err)
 	}

cmd/stolonctl/config_replace.go

@@ -95,7 +95,13 @@ func configReplace(cmd *cobra.Command, args []string) {
 
 	storePath := filepath.Join(common.StoreBasePath, cfg.clusterName)
 
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
 	if err != nil {
 		die("cannot create store: %v", err)
 	}

cmd/stolonctl/listclusters.go

@@ -36,7 +36,14 @@ func init() {
 }
 
 func getClusters(storeBasePath string) ([]string, error) {
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
+
 	if err != nil {
 		return nil, fmt.Errorf("cannot create store: %v", err)
 	}

cmd/stolonctl/status.go

@@ -84,7 +84,13 @@ func status(cmd *cobra.Command, args []string) {
 	}
 	storePath := filepath.Join(common.StoreBasePath, cfg.clusterName)
 
-	kvstore, err := store.NewStore(store.Backend(cfg.storeBackend), cfg.storeEndpoints)
+	kvstore, err := store.NewStore(
+		store.Backend(cfg.storeBackend),
+		cfg.storeEndpoints,
+		cfg.storeCertFile,
+		cfg.storeKeyFile,
+		cfg.storeCACertFile,
+	)
 	if err != nil {
 		die("cannot create store: %v", err)
 	}

cmd/stolonctl/stolonctl.go

@@ -38,16 +38,22 @@ var cmdStolonCtl = &cobra.Command{
 }
 
 type config struct {
-	storeBackend   string
-	storeEndpoints string
-	clusterName    string
+	storeBackend    string
+	storeEndpoints  string
+	storeCertFile   string
+	storeKeyFile    string
+	storeCACertFile string
+	clusterName     string
 }
 
 var cfg config
 
 func init() {
 	cmdStolonCtl.PersistentFlags().StringVar(&cfg.storeBackend, "store-backend", "", "store backend type (etcd or consul)")
 	cmdStolonCtl.PersistentFlags().StringVar(&cfg.storeEndpoints, "store-endpoints", "", "a comma-delimited list of store endpoints (defaults: 127.0.0.1:2379 for etcd, 127.0.0.1:8500 for consul)")
+	cmdStolonCtl.PersistentFlags().StringVar(&cfg.storeCertFile, "store-cert", "", "path to the client server TLS cert file")
+	cmdStolonCtl.PersistentFlags().StringVar(&cfg.storeKeyFile, "store-key", "", "path to the client server TLS key file")
+	cmdStolonCtl.PersistentFlags().StringVar(&cfg.storeCACertFile, "store-cacert", "", "path to the client server TLS trusted CA key file")
 	cmdStolonCtl.PersistentFlags().StringVar(&cfg.clusterName, "cluster-name", "", "cluster name")
 }
 

pkg/store/store.go

@@ -15,8 +15,11 @@
 package store
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"path/filepath"
 	"strings"
 	"time"
@@ -70,7 +73,7 @@ type StoreManager struct {
 	store       kvstore.Store
 }
 
-func NewStore(backend Backend, addrsStr string) (kvstore.Store, error) {
+func NewStore(backend Backend, addrsStr, certFile, keyFile, caCertFile string) (kvstore.Store, error) {
 
 	var kvbackend kvstore.Backend
 	switch backend {
@@ -92,7 +95,42 @@ func NewStore(backend Backend, addrsStr string) (kvstore.Store, error) {
 	}
 	addrs := strings.Split(addrsStr, ",")
 
-	store, err := libkv.NewStore(kvbackend, addrs, &kvstore.Config{ConnectionTimeout: 10 * time.Second})
+	var tlsC *tls.Config = nil
+	if certFile != "" && keyFile != "" {
+		cc := &kvstore.ClientTLSConfig{
+			CertFile:   certFile,
+			KeyFile:    keyFile,
+			CACertFile: caCertFile,
+		}
+
+		var certPool *x509.CertPool = nil
+		if caCertFile != "" {
+			if pemBytes, err := ioutil.ReadFile(cc.CACertFile); err == nil {
+				certPool = x509.NewCertPool()
+				certPool.AppendCertsFromPEM(pemBytes)
+			} else {
+				return nil, err
+			}
+		}
+
+		if tlsCert, err := tls.LoadX509KeyPair(cc.CertFile, cc.KeyFile); err == nil {
+			tlsC = &tls.Config{
+				RootCAs:      certPool,
+				Certificates: []tls.Certificate{tlsCert},
+			}
+		} else {
+			return nil, err
+		}
+	}
+
+	store, err := libkv.NewStore(
+		kvbackend,
+		addrs,
+		&kvstore.Config{
+			TLS:               tlsC,
+			ConnectionTimeout: 10 * time.Second,
+		},
+	)
 	if err != nil {
 		return nil, err
 	}

tests/integration/config_test.go

@@ -53,7 +53,13 @@ func TestServerParameters(t *testing.T) {
 
 	storePath := filepath.Join(common.StoreBasePath, clusterName)
 
-	kvstore, err := store.NewStore(tstore.storeBackend, storeEndpoints)
+	kvstore, err := store.NewStore(
+		tstore.storeBackend,
+		storeEndpoints,
+		"",
+		"",
+		"",
+	)
 	if err != nil {
 		t.Fatalf("cannot create store: %v", err)
 	}

tests/integration/ha_test.go

@@ -67,7 +67,13 @@ func TestInitWithMultipleKeepers(t *testing.T) {
 
 	storePath := filepath.Join(common.StoreBasePath, clusterName)
 
-	kvstore, err := store.NewStore(tstore.storeBackend, storeEndpoints)
+	kvstore, err := store.NewStore(
+		tstore.storeBackend,
+		storeEndpoints,
+		"",
+		"",
+		"",
+	)
 	if err != nil {
 		t.Fatalf("cannot create store: %v", err)
 	}
@@ -131,7 +137,13 @@ func setupServers(t *testing.T, clusterName, dir string, numKeepers, numSentinel
 
 	storePath := filepath.Join(common.StoreBasePath, clusterName)
 
-	kvstore, err := store.NewStore(tstore.storeBackend, storeEndpoints)
+	kvstore, err := store.NewStore(
+		tstore.storeBackend,
+		storeEndpoints,
+		"",
+		"",
+		"",
+	)
 	if err != nil {
 		t.Fatalf("cannot create store: %v", err)
 	}

tests/integration/init_test.go

@@ -118,7 +118,13 @@ func TestInitUsers(t *testing.T) {
 	clusterName = uuid.NewV4().String()
 	storePath := filepath.Join(common.StoreBasePath, clusterName)
 
-	kvstore, err := store.NewStore(tstore.storeBackend, storeEndpoints)
+	kvstore, err := store.NewStore(
+		tstore.storeBackend,
+		storeEndpoints,
+		"",
+		"",
+		"",
+	)
 	if err != nil {
 		t.Fatalf("cannot create store: %v", err)
 	}
@@ -154,7 +160,13 @@ func TestInitUsers(t *testing.T) {
 	clusterName = uuid.NewV4().String()
 	storePath = filepath.Join(common.StoreBasePath, clusterName)
 
-	kvstore, err = store.NewStore(tstore.storeBackend, storeEndpoints)
+	kvstore, err = store.NewStore(
+		tstore.storeBackend,
+		storeEndpoints,
+		"",
+		"",
+		"",
+	)
 	if err != nil {
 		t.Fatalf("cannot create store: %v", err)
 	}
@@ -216,7 +228,13 @@ func TestInitialClusterConfig(t *testing.T) {
 	storeEndpoints := fmt.Sprintf("%s:%s", tstore.listenAddress, tstore.port)
 	storePath := filepath.Join(common.StoreBasePath, clusterName)
 
-	kvstore, err := store.NewStore(tstore.storeBackend, storeEndpoints)
+	kvstore, err := store.NewStore(
+		tstore.storeBackend,
+		storeEndpoints,
+		"",
+		"",
+		"",
+	)
 	if err != nil {
 		t.Fatalf("cannot create store: %v", err)
 	}

tests/integration/proxy_test.go

@@ -77,7 +77,13 @@ func TestProxyListening(t *testing.T) {
 
 	storePath := filepath.Join(common.StoreBasePath, clusterName)
 
-	kvstore, err := store.NewStore(tstore.storeBackend, storeEndpoints)
+	kvstore, err := store.NewStore(
+		tstore.storeBackend,
+		storeEndpoints,
+		"",
+		"",
+		"",
+	)
 	if err != nil {
 		t.Fatalf("cannot create store: %v", err)
 	}