diff --git a/agents/drivers/zookeeper/agent_test.go b/agents/drivers/zookeeper/agent_test.go index d998925f23..007996489e 100644 --- a/agents/drivers/zookeeper/agent_test.go +++ b/agents/drivers/zookeeper/agent_test.go @@ -97,7 +97,7 @@ func TestConnectionConfiguration(t *testing.T) { } }) } - if !hasTLSOptions(connectionConfig{CACertPath: "/tmp/ca.pem"}) { + if !tlsOptionsPresent(connectionConfig{CACertPath: "/tmp/ca.pem"}) { t.Fatal("TLS path was not detected") } for input, expected := range map[string]int{"": 16, "bad": 16, "0": 1, "32": 32, "100": 64} { @@ -113,7 +113,9 @@ func TestConnectionValidationHappensBeforeNetwork(t *testing.T) { config connectionConfig message string }{ - {"tls", connectionConfig{SSL: true}, "ZooKeeper TLS is not supported"}, + {"tls client cert without key", connectionConfig{ClientCertPath: "/tmp/client.crt"}, "client key (client_key_path) is required"}, + {"tls client key without cert", connectionConfig{ClientKeyPath: "/tmp/client.key"}, "client certificate (client_cert_path) is required"}, + {"tls missing CA file", connectionConfig{CACertPath: "/tmp/missing-ca.pem"}, "load ZooKeeper CA certificate"}, {"auth", connectionConfig{AuthScheme: "sasl"}, `Unsupported auth_scheme "sasl"`}, {"sasl username", connectionConfig{AuthScheme: saslDigestAuthScheme, Password: "secret"}, `username is required when auth_scheme = "sasl_digest"`}, {"sasl password", connectionConfig{AuthScheme: saslDigestAuthScheme, Username: "user"}, `password is required when auth_scheme = "sasl_digest"`}, diff --git a/agents/drivers/zookeeper/connection.go b/agents/drivers/zookeeper/connection.go index 8b93241d40..2eef6b30b2 100644 --- a/agents/drivers/zookeeper/connection.go +++ b/agents/drivers/zookeeper/connection.go @@ -2,6 +2,7 @@ package main import ( "context" + "crypto/tls" "encoding/json" "errors" "fmt" @@ -147,8 +148,9 @@ func (service *server) connectionInfo() (map[string]any, error) { } func openClient(config connectionConfig) (*clientSession, error) { - if hasTLSOptions(config) { - return nil, errors.New("ZooKeeper TLS is not supported") + tlsConfig, err := buildZooKeeperTLSConfig(config) + if err != nil { + return nil, err } authScheme := resolveAuthScheme(config) if authScheme != defaultAuthScheme && authScheme != saslDigestAuthScheme { @@ -179,16 +181,16 @@ func openClient(config connectionConfig) (*clientSession, error) { } connectionTimeout := millisecondsOrDefault(config.ConnectionTimeoutMS, defaultConnectionTimeout) probeTimeout := minDuration(defaultProbeTimeout, connectionTimeout) - if err := requireReachableServer(target.Servers, probeTimeout); err != nil { + if err := requireReachableServer(target.Servers, probeTimeout, tlsConfig); err != nil { return nil, err } - dialer := newZooKeeperDialer(connectionTimeout, nil) + dialer := newZooKeeperDialer(connectionTimeout, nil, tlsConfig) if authScheme == saslDigestAuthScheme { dialer = newZooKeeperDialer(connectionTimeout, &saslDigestCredentials{ Username: strings.TrimSpace(config.Username), Password: config.Password, - }) + }, tlsConfig) } sessionTimeout := millisecondsOrDefault(config.SessionTimeoutMS, defaultSessionTimeout) @@ -254,13 +256,13 @@ func openClient(config connectionConfig) (*clientSession, error) { return session, nil } -func newZooKeeperDialer(connectionTimeout time.Duration, credentials *saslDigestCredentials) zk.Dialer { +func newZooKeeperDialer(connectionTimeout time.Duration, credentials *saslDigestCredentials, tlsConfig *tls.Config) zk.Dialer { return func(network, address string, libraryTimeout time.Duration) (net.Conn, error) { timeout := libraryTimeout if timeout <= 0 || connectionTimeout < timeout { timeout = connectionTimeout } - connection, err := net.DialTimeout(network, address, timeout) + connection, err := dialZooKeeperConnection(address, timeout, tlsConfig) if err != nil { return nil, err } @@ -294,13 +296,17 @@ func databaseInfo(config connectionConfig) map[string]any { if err != nil { return info } - if version := detectServerVersion(target.Servers, millisecondsOrDefault(config.ConnectionTimeoutMS, defaultConnectionTimeout)); version != "" { + tlsConfig, err := buildZooKeeperTLSConfig(config) + if err != nil { + return info + } + if version := detectServerVersion(target.Servers, millisecondsOrDefault(config.ConnectionTimeoutMS, defaultConnectionTimeout), tlsConfig); version != "" { info["productVersion"] = version } return info } -func detectServerVersion(servers []string, timeout time.Duration) string { +func detectServerVersion(servers []string, timeout time.Duration, tlsConfig *tls.Config) string { deadline := timeout if deadline <= 0 || deadline > 2*time.Second { deadline = 2 * time.Second @@ -313,7 +319,7 @@ func detectServerVersion(servers []string, timeout time.Duration) string { // ZooKeeper 3.5+ whitelists only "srvr" by default; "envi"/"stat" // are opt-in, so probe srvr first for default-config clusters. for _, command := range []string{"srvr", "envi", "stat"} { - connection, err := net.DialTimeout("tcp", address, deadline) + connection, err := dialZooKeeperConnection(address, deadline, tlsConfig) if err != nil { continue } @@ -378,7 +384,7 @@ func parseConnectTarget(value string) (connectTarget, error) { return connectTarget{Servers: servers, Chroot: chroot}, nil } -func requireReachableServer(servers []string, timeout time.Duration) error { +func requireReachableServer(servers []string, timeout time.Duration, tlsConfig *tls.Config) error { ctx, cancel := context.WithTimeout(context.Background(), timeout+500*time.Millisecond) defer cancel() workers := minInt(len(servers), maximumReachabilityWorkers) @@ -394,7 +400,7 @@ func requireReachableServer(servers []string, timeout time.Duration) error { if err != nil { continue } - connection, err := net.DialTimeout("tcp", address, timeout) + connection, err := dialZooKeeperConnection(address, timeout, tlsConfig) if err == nil { connection.Close() select { @@ -485,16 +491,6 @@ func connectionURLParams(config connectionConfig) url.Values { return parsed } -func hasTLSOptions(config connectionConfig) bool { - return config.SSL || firstNonBlank( - config.CACertPath, - config.ClientCertPath, - config.ClientKeyPath, - config.CertPath, - config.KeyPath, - ) != "" -} - func joinPrefix(chroot, namespace string) string { parts := make([]string, 0, 2) if normalized := normalizePrefix(chroot); normalized != "" { diff --git a/agents/drivers/zookeeper/integration_test.go b/agents/drivers/zookeeper/integration_test.go index c2697d623e..b9981ca796 100644 --- a/agents/drivers/zookeeper/integration_test.go +++ b/agents/drivers/zookeeper/integration_test.go @@ -171,3 +171,32 @@ func BenchmarkZooKeeperOperations(b *testing.B) { } }) } + +func TestZooKeeperTLSIntegration(t *testing.T) { + connectString := os.Getenv("DBX_ZOOKEEPER_TLS_TEST_CONNECT_STRING") + if connectString == "" { + t.Skip("DBX_ZOOKEEPER_TLS_TEST_CONNECT_STRING is not set") + } + caCertPath := os.Getenv("DBX_ZOOKEEPER_TLS_TEST_CA_CERT_PATH") + if caCertPath == "" { + t.Skip("DBX_ZOOKEEPER_TLS_TEST_CA_CERT_PATH is not set") + } + service := &server{statLookupConcurrency: 4} + connection := map[string]any{ + "zookeeper_connect_string": connectString, + "ssl": true, + "ca_cert_path": caCertPath, + } + if _, err := service.connect(mustJSON(map[string]any{"connection": connection})); err != nil { + t.Fatalf("connect over TLS: %v", err) + } + t.Cleanup(service.closeClient) + root := fmt.Sprintf("/dbx-go-tls-integration-%d", time.Now().UnixNano()) + if _, err := service.put(mustJSON(map[string]any{"key": root + "/value", "value": map[string]string{"encoding": "utf8", "data": "secret"}})); err != nil { + t.Fatal(err) + } + result, err := service.get(mustJSON(map[string]any{"key": root + "/value"})) + if err != nil || result["found"] != true { + t.Fatalf("get=%#v err=%v", result, err) + } +} diff --git a/agents/drivers/zookeeper/tls.go b/agents/drivers/zookeeper/tls.go new file mode 100644 index 0000000000..e17c2f9ef0 --- /dev/null +++ b/agents/drivers/zookeeper/tls.go @@ -0,0 +1,119 @@ +package main + +import ( + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "net/url" + "os" + "strconv" + "strings" + "time" +) + +// dialZooKeeperConnection opens a raw ZooKeeper transport connection. When +// tlsConfig is non-nil the connection is wrapped in a TLS handshake so the +// caller can talk to a secure (clientCnxn with ssl) ZooKeeper port. +func dialZooKeeperConnection(address string, timeout time.Duration, tlsConfig *tls.Config) (net.Conn, error) { + dialer := &net.Dialer{Timeout: timeout} + if tlsConfig == nil { + return dialer.Dial("tcp", address) + } + config := tlsConfig.Clone() + if config.ServerName == "" { + host, _, err := net.SplitHostPort(address) + if err != nil { + return nil, fmt.Errorf("parse ZooKeeper TLS address %q: %w", address, err) + } + config.ServerName = host + } + return tls.DialWithDialer(dialer, "tcp", address, config) +} + +// buildZooKeeperTLSConfig turns the TLS-related fields of a connectionConfig +// into a *tls.Config. It returns (nil, nil) when TLS is not requested so the +// caller can keep using the plain TCP dialer for non-secure clusters. +func buildZooKeeperTLSConfig(config connectionConfig) (*tls.Config, error) { + if !tlsOptionsPresent(config) { + return nil, nil + } + tlsConfig := &tls.Config{ + MinVersion: tls.VersionTLS12, + } + caPath := firstNonBlank(config.CACertPath) + if caPath != "" { + pool, err := loadCACertPool(caPath) + if err != nil { + return nil, fmt.Errorf("load ZooKeeper CA certificate: %w", err) + } + tlsConfig.RootCAs = pool + } + certPath := firstNonBlank(config.ClientCertPath, config.CertPath) + keyPath := firstNonBlank(config.ClientKeyPath, config.KeyPath) + switch { + case certPath != "" && keyPath != "": + certificate, err := tls.LoadX509KeyPair(certPath, keyPath) + if err != nil { + return nil, fmt.Errorf("load ZooKeeper client certificate: %w", err) + } + tlsConfig.Certificates = []tls.Certificate{certificate} + case certPath != "": + return nil, errors.New("client key (client_key_path) is required when a client certificate is provided") + case keyPath != "": + return nil, errors.New("client certificate (client_cert_path) is required when a client key is provided") + } + // insecure_skip_verify is exposed only as a connection URL parameter + // (e.g. ?insecure_skip_verify=true). It defaults to off, so callers that + // don't set it get full certificate-chain verification. Only enable it for + // trusted or test-only clusters, since it disables that verification. + if paramBool(connectionURLParams(config), "insecure_skip_verify") { + tlsConfig.InsecureSkipVerify = true + } + return tlsConfig, nil +} + +// tlsOptionsPresent reports whether the connection configuration requests a +// TLS-secured connection, either via the explicit ssl flag or by supplying any +// certificate/material paths. +func tlsOptionsPresent(config connectionConfig) bool { + return config.SSL || firstNonBlank( + config.CACertPath, + config.ClientCertPath, + config.ClientKeyPath, + config.CertPath, + config.KeyPath, + ) != "" +} + +// loadCACertPool reads a PEM-encoded trust store (one or more CERTIFICATE +// blocks) and returns an x509.CertPool built from it. It uses +// x509.CertPool.AppendCertsFromPEM (the same helper etcd-go relies on) so a +// single call parses every CERTIFICATE block in the file, including CA-chain +// bundles, without a manual decode loop. +func loadCACertPool(path string) (*x509.CertPool, error) { + contents, err := os.ReadFile(path) + if err != nil { + return nil, err + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(contents) { + return nil, errors.New("PEM CA file contains no CERTIFICATE blocks") + } + return pool, nil +} + +// paramBool parses a boolean connection URL parameter, treating missing or +// unparseable values as false. +func paramBool(params url.Values, name string) bool { + value := strings.TrimSpace(params.Get(name)) + if value == "" { + return false + } + parsed, err := strconv.ParseBool(value) + if err != nil { + return false + } + return parsed +} diff --git a/agents/drivers/zookeeper/tls_test.go b/agents/drivers/zookeeper/tls_test.go new file mode 100644 index 0000000000..37b0ee5e04 --- /dev/null +++ b/agents/drivers/zookeeper/tls_test.go @@ -0,0 +1,131 @@ +package main + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "os" + "path/filepath" + "testing" + "time" +) + +func writePEM(t *testing.T, dir, name, blockType string, der []byte) string { + t.Helper() + path := filepath.Join(dir, name) + if err := os.WriteFile(path, pem.EncodeToMemory(&pem.Block{Type: blockType, Bytes: der}), 0o600); err != nil { + t.Fatalf("write %s: %v", name, err) + } + return path +} + +func generateCertificate(t *testing.T, dir string, ca *x509.Certificate, caKey *ecdsa.PrivateKey) (certPath, keyPath string) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate key: %v", err) + } + template := &x509.Certificate{ + SerialNumber: big.NewInt(time.Now().UnixNano()), + Subject: pkix.Name{CommonName: "dbx-test"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth, x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + IsCA: ca == nil, + } + signer := ca + signerKey := caKey + if ca == nil { + signer = template + signerKey = key + } + der, err := x509.CreateCertificate(rand.Reader, template, signer, &key.PublicKey, signerKey) + if err != nil { + t.Fatalf("create certificate: %v", err) + } + certPath = writePEM(t, dir, "cert.pem", "CERTIFICATE", der) + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + t.Fatalf("marshal key: %v", err) + } + keyPath = writePEM(t, dir, "key.pem", "EC PRIVATE KEY", keyDER) + return certPath, keyPath +} + +func TestBuildZooKeeperTLSConfig(t *testing.T) { + dir := t.TempDir() + caCertPath, caKeyPath := generateCertificate(t, dir, nil, nil) + caDER, _ := os.ReadFile(caCertPath) + caBlock, _ := pem.Decode(caDER) + ca, err := x509.ParseCertificate(caBlock.Bytes) + if err != nil { + t.Fatalf("parse ca: %v", err) + } + keyDER, _ := os.ReadFile(caKeyPath) + caKeyBlock, _ := pem.Decode(keyDER) + caKey, err := x509.ParseECPrivateKey(caKeyBlock.Bytes) + if err != nil { + t.Fatalf("parse ca key: %v", err) + } + + t.Run("disabled without options", func(t *testing.T) { + config, err := buildZooKeeperTLSConfig(connectionConfig{}) + if err != nil || config != nil { + t.Fatalf("expected (nil, nil), got (%v, %v)", config, err) + } + }) + + t.Run("ca only sets RootCAs", func(t *testing.T) { + caPath, _ := generateCertificate(t, dir, ca, caKey) + config, err := buildZooKeeperTLSConfig(connectionConfig{CACertPath: caPath}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if config == nil || config.RootCAs == nil { + t.Fatal("expected tls.Config with RootCAs") + } + if config.MinVersion != 0x0303 { // TLS 1.2 + t.Fatalf("expected MinVersion TLS1.2, got %x", config.MinVersion) + } + }) + + t.Run("client cert and key load", func(t *testing.T) { + clientCert, clientKey := generateCertificate(t, dir, ca, caKey) + config, err := buildZooKeeperTLSConfig(connectionConfig{ + ClientCertPath: clientCert, + ClientKeyPath: clientKey, + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(config.Certificates) != 1 { + t.Fatalf("expected 1 client certificate, got %d", len(config.Certificates)) + } + }) + + t.Run("client cert without key errors", func(t *testing.T) { + clientCert, _ := generateCertificate(t, dir, ca, caKey) + if _, err := buildZooKeeperTLSConfig(connectionConfig{ClientCertPath: clientCert}); err == nil { + t.Fatal("expected error when client cert has no key") + } + }) + + t.Run("insecure skip verify", func(t *testing.T) { + config, err := buildZooKeeperTLSConfig(connectionConfig{ + SSL: true, + URLParams: "insecure_skip_verify=true", + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !config.InsecureSkipVerify { + t.Fatal("expected InsecureSkipVerify true") + } + }) +} diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index 59ce4b2fca..badd87f8c0 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -2546,10 +2546,6 @@ function applyProfile(val: string, preserveConnectionFields = false) { if (profile.type === "zookeeper") { form.value.database = undefined; form.value.connection_string = ""; - form.value.ssl = false; - form.value.ca_cert_path = ""; - form.value.client_cert_path = ""; - form.value.client_key_path = ""; } if (profile.type === "nacos") { resetNacosFields(); @@ -3173,6 +3169,7 @@ const tlsCapableDatabaseTypes = new Set([ "influxdb", "victoriametrics", "cassandra", + "zookeeper", ]); const supportsTlsToggle = computed(() => tlsCapableDatabaseTypes.has(form.value.db_type)); const supportsCaCertificatePath = computed(() => form.value.db_type === "clickhouse" || form.value.db_type === "victoriametrics"); @@ -4385,7 +4382,12 @@ function connectionConfigForSubmit(id: string, generatedName = "", validatePlugi config.port = firstEndpoint.port; } config.database = undefined; - config.ssl = false; + config.ca_cert_path = config.ca_cert_path?.trim() || ""; + config.client_cert_path = config.client_cert_path?.trim() || ""; + config.client_key_path = config.client_key_path?.trim() || ""; + if ((config.client_cert_path && !config.client_key_path) || (!config.client_cert_path && config.client_key_path)) { + throw new Error(t("connection.etcdClientCertPairRequired")); + } } if (config.db_type === "etcd") { config.etcd_endpoints = normalizeEndpointLines(config.etcd_endpoints || ""); @@ -4400,12 +4402,12 @@ function connectionConfigForSubmit(id: string, generatedName = "", validatePlugi if ((config.client_cert_path && !config.client_key_path) || (!config.client_cert_path && config.client_key_path)) { throw new Error(t("connection.etcdClientCertPairRequired")); } - } else if (form.value.db_type !== "consul") { + } else if (form.value.db_type !== "consul" && config.db_type !== "zookeeper") { config.etcd_endpoints = undefined; config.client_cert_path = undefined; config.client_key_path = undefined; } - if (config.db_type !== "mysql" && config.db_type !== "clickhouse" && config.db_type !== "etcd" && config.db_type !== "consul" && config.db_type !== "starrocks" && config.db_type !== "mongodb" && config.db_type !== "victoriametrics") { + if (config.db_type !== "mysql" && config.db_type !== "clickhouse" && config.db_type !== "etcd" && config.db_type !== "consul" && config.db_type !== "starrocks" && config.db_type !== "mongodb" && config.db_type !== "victoriametrics" && config.db_type !== "zookeeper") { config.ca_cert_path = undefined; } else { config.ca_cert_path = config.ca_cert_path?.trim() || ""; @@ -8342,7 +8344,7 @@ function openExternalUrl(url: string) { -