feat: add TLS support for Keeper connections

Author: loomts

pkg/backup/create.go

@@ -794,7 +794,7 @@ func (b *Backuper) createBackupNamedCollections(ctx context.Context, backupPath
 			}
 
 			k := keeper.Keeper{}
-			if err = k.Connect(ctx, b.ch); err != nil {
+			if err = k.Connect(ctx, b.ch, b.cfg); err != nil {
 				return 0, err
 			}
 			defer k.Close()
@@ -844,7 +844,7 @@ func (b *Backuper) createBackupRBACReplicated(ctx context.Context, rbacBackup st
 	rbacDataSize := uint64(0)
 	if err = b.ch.SelectContext(ctx, &replicatedRBAC, "SELECT name FROM system.user_directories WHERE type='replicated'"); err == nil && len(replicatedRBAC) > 0 {
 		k := keeper.Keeper{}
-		if err = k.Connect(ctx, b.ch); err != nil {
+		if err = k.Connect(ctx, b.ch, b.cfg); err != nil {
 			return 0, err
 		}
 		defer k.Close()

pkg/backup/restore.go

@@ -500,7 +500,7 @@ func (b *Backuper) restoreRBAC(ctx context.Context, backupName string, disks []c
 	replicatedUserDirectories := make([]clickhouse.UserDirectory, 0)
 	if err = b.ch.SelectContext(ctx, &replicatedUserDirectories, "SELECT name FROM system.user_directories WHERE type='replicated'"); err == nil && len(replicatedUserDirectories) > 0 {
 		k = &keeper.Keeper{}
-		if connErr := k.Connect(ctx, b.ch); connErr != nil {
+		if connErr := k.Connect(ctx, b.ch, b.cfg); connErr != nil {
 			return errors.Wrap(connErr, "but can't connect to keeper")
 		}
 		defer k.Close()

pkg/clickhouse/clickhouse.go

@@ -2,8 +2,6 @@ package clickhouse
 
 import (
 	"context"
-	"crypto/tls"
-	"crypto/x509"
 	"database/sql"
 	"fmt"
 	"os"
@@ -18,6 +16,7 @@ import (
 	"github.com/Altinity/clickhouse-backup/v2/pkg/common"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/Altinity/clickhouse-backup/v2/pkg/metadata"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 	"github.com/antchfx/xmlquery"
@@ -87,31 +86,9 @@ func (ch *ClickHouse) Connect() error {
 	}
 
 	if ch.Config.Secure {
-		tlsConfig := &tls.Config{
-			InsecureSkipVerify: ch.Config.SkipVerify,
-		}
-		if ch.Config.TLSKey != "" || ch.Config.TLSCert != "" || ch.Config.TLSCa != "" {
-			if ch.Config.TLSCert != "" || ch.Config.TLSKey != "" {
-				cert, err := tls.LoadX509KeyPair(ch.Config.TLSCert, ch.Config.TLSKey)
-				if err != nil {
-					log.Error().Msgf("tls.LoadX509KeyPair error: %v", err)
-					return err
-				}
-				tlsConfig.Certificates = []tls.Certificate{cert}
-			}
-			if ch.Config.TLSCa != "" {
-				caCert, err := os.ReadFile(ch.Config.TLSCa)
-				if err != nil {
-					log.Error().Msgf("read `tls_ca` file %s return error: %v ", ch.Config.TLSCa, err)
-					return err
-				}
-				caCertPool := x509.NewCertPool()
-				if caCertPool.AppendCertsFromPEM(caCert) != true {
-					log.Error().Msgf("AppendCertsFromPEM %s return false", ch.Config.TLSCa)
-					return fmt.Errorf("AppendCertsFromPEM %s return false", ch.Config.TLSCa)
-				}
-				tlsConfig.RootCAs = caCertPool
-			}
+		tlsConfig, err := utils.NewTLSConfig(ch.Config.TLSCa, ch.Config.TLSCert, ch.Config.TLSKey, ch.Config.SkipVerify)
+		if err != nil {
+			return err
 		}
 		opt.TLS = tlsConfig
 	}

pkg/keeper/keeper.go

@@ -3,9 +3,11 @@ package keeper
 import (
 	"bufio"
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"os"
 	"path"
 	"strconv"
@@ -18,6 +20,8 @@ import (
 	"github.com/rs/zerolog/log"
 
 	"github.com/Altinity/clickhouse-backup/v2/pkg/clickhouse"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/utils"
 	"github.com/go-zookeeper/zk"
 )
 
@@ -59,7 +63,7 @@ type Keeper struct {
 }
 
 // Connect - connect to any zookeeper server from /var/lib/clickhouse/preprocessed_configs/config.xml
-func (k *Keeper) Connect(ctx context.Context, ch *clickhouse.ClickHouse) error {
+func (k *Keeper) Connect(ctx context.Context, ch *clickhouse.ClickHouse, cfg *config.Config) error {
 	configFile, doc, err := ch.ParseXML(ctx, "config.xml")
 	if err != nil {
 		return errors.Wrapf(err, "can't parse config.xml from %s, error", configFile)
@@ -83,6 +87,7 @@ func (k *Keeper) Connect(ctx context.Context, ch *clickhouse.ClickHouse) error {
 		return errors.WithStack(fmt.Errorf("/zookeeper/node not exists in %s", configFile))
 	}
 	keeperHosts := make([]string, len(nodeList))
+	isSecure := false
 	for i, node := range nodeList {
 		hostNode := node.SelectElement("host")
 		if hostNode == nil {
@@ -93,11 +98,37 @@ func (k *Keeper) Connect(ctx context.Context, ch *clickhouse.ClickHouse) error {
 		if portNode != nil {
 			port = portNode.InnerText()
 		}
+		secureNode := node.SelectElement("secure")
+		if secureNode != nil && (secureNode.InnerText() == "1" || secureNode.InnerText() == "true") {
+			isSecure = true
+		}
 		keeperHosts[i] = fmt.Sprintf("%s:%s", hostNode.InnerText(), port)
 	}
-	conn, _, err := zk.Connect(keeperHosts, sessionTimeout, zk.WithLogger(newKeeperLogger()))
-	if err != nil {
-		return err
+	var conn *zk.Conn
+	if isSecure {
+		log.Info().Msgf("isSecure=%v, keeperHosts=%v, caPath=%v, certPath=%v, keyPath=%v, skipVerify=%v, use TLS for keeper connection", isSecure, keeperHosts, cfg.ClickHouse.TLSCa, cfg.ClickHouse.TLSCert, cfg.ClickHouse.TLSKey, cfg.ClickHouse.SkipVerify)
+		tlsConfig, err := utils.NewTLSConfig(cfg.ClickHouse.TLSCa, cfg.ClickHouse.TLSCert, cfg.ClickHouse.TLSKey, cfg.ClickHouse.SkipVerify)
+		if err != nil {
+			return errors.Wrap(err, "can't create TLS config")
+		}
+		conn, _, err = zk.Connect(keeperHosts, sessionTimeout, zk.WithLogger(newKeeperLogger()), zk.WithDialer(func(network, address string, timeout time.Duration) (net.Conn, error) {
+			tlsConn, dialErr := tls.DialWithDialer(&net.Dialer{Timeout: timeout}, network, address, tlsConfig)
+			if dialErr != nil {
+				log.Error().Msgf("TLS dial to %s failed: %v", address, dialErr)
+				return nil, dialErr
+			}
+			return tlsConn, nil
+		}))
+		if err != nil {
+			log.Error().Msgf("zk.Connect with TLS failed: %v", err)
+			return err
+		}
+	} else {
+		log.Info().Msgf("isSecure=%v, keeperHosts=%v", isSecure, keeperHosts)
+		conn, _, err = zk.Connect(keeperHosts, sessionTimeout, zk.WithLogger(newKeeperLogger()))
+		if err != nil {
+			return err
+		}
 	}
 	if digestNode := zookeeperNode.SelectElement("digest"); digestNode != nil {
 		if err = conn.AddAuth("digest", []byte(digestNode.InnerText())); err != nil {

pkg/utils/utils.go

@@ -2,12 +2,16 @@ package utils
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
-	"github.com/rs/zerolog/log"
+	"os"
 	"os/exec"
 	"regexp"
 	"strings"
 	"time"
+
+	"github.com/rs/zerolog/log"
 )
 
 const (
@@ -72,3 +76,28 @@ func ExecCmdOut(ctx context.Context, timeout time.Duration, cmd string, args ...
 	cancel()
 	return string(out), err
 }
+
+func NewTLSConfig(caPath, certPath, keyPath string, skipVerify bool) (*tls.Config, error) {
+	tlsConfig := &tls.Config{
+		InsecureSkipVerify: skipVerify,
+	}
+	if certPath != "" || keyPath != "" {
+		cert, err := tls.LoadX509KeyPair(certPath, keyPath)
+		if err != nil {
+			return nil, fmt.Errorf("tls.LoadX509KeyPair error: %v", err)
+		}
+		tlsConfig.Certificates = []tls.Certificate{cert}
+	}
+	if caPath != "" {
+		caCert, err := os.ReadFile(caPath)
+		if err != nil {
+			return nil, fmt.Errorf("read ca file %s return error: %v", caPath, err)
+		}
+		caCertPool := x509.NewCertPool()
+		if !caCertPool.AppendCertsFromPEM(caCert) {
+			return nil, fmt.Errorf("AppendCertsFromPEM %s return false", caPath)
+		}
+		tlsConfig.RootCAs = caCertPool
+	}
+	return tlsConfig, nil
+}