Clickhouse SPIFFE MTLs and SQL over HTTPS support

clickhouse_https_bug_test.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/go-kit/log"
+	"github.com/stretchr/testify/require"
+)
+
+// TestClickHouseHTTPSURLBug tests the bug where clickhouse+https URLs get incorrectly prefixed with tcp://
+func TestClickHouseHTTPSURLBug(t *testing.T) {
+	// Create a job with a clickhouse+https connection
+	job := &Job{
+		log:         log.NewNopLogger(),
+		mtlsSetup:   &MockMTLSSetup{ShouldFail: false},
+		Connections: []string{"clickhouse+https://127.0.0.1:19999/default?secure=true&tls_config=spiffe&username=clickhouse"},
+		Timeout:     1 * time.Second,
+	}
+
+	// Update connections to create the connection object
+	job.updateConnections()
+
+	// Verify we have exactly one connection
+	require.Len(t, job.conns, 1, "Should have exactly one connection")
+	conn := job.conns[0]
+
+	// Test the connection with a fast timeout to avoid hanging
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+
+	// Helper function to check for the bug
+	checkConnectionError := func(attemptNum int, err error) bool {
+		if err == nil {
+			t.Logf("Attempt %d: Connection unexpectedly succeeded", attemptNum)
+			return false
+		}
+
+		// Check for different manifestations of the URL corruption bug
+		if strings.Contains(err.Error(), "tcp://clickhouse+https") {
+			t.Errorf("Attempt %d: Bug reproduced: URL incorrectly prefixed with tcp://. Error: %v", attemptNum, err)
+			return true // Bug found
+		} else if strings.Contains(err.Error(), "lookup clickhouse+https") {
+			t.Errorf("Attempt %d: Bug reproduced: URL corruption causing DNS lookup of 'clickhouse+https'. Error: %v", attemptNum, err)
+			return true // Bug found
+		} else if strings.Contains(err.Error(), "clickhouse+https") && !strings.Contains(err.Error(), "127.0.0.1:19999") {
+			t.Errorf("Attempt %d: Bug reproduced: URL contains malformed 'clickhouse+https' without proper hostname. Error: %v", attemptNum, err)
+			return true // Bug found
+		} else {
+			// If we don't see any URL corruption bug, the connection attempt might fail for other reasons
+			// which is fine - we just want to ensure the URL is processed correctly
+			t.Logf("Attempt %d: Connection failed as expected (no URL corruption bug detected): %v", attemptNum, err)
+			return false
+		}
+	}
+
+	// Try to ping the database twice - the bug seems to occur on the retry mechanism
+	// First attempt
+	err1 := conn.connect(ctx, job)
+	bugFound := checkConnectionError(1, err1)
+	if bugFound {
+		return
+	}
+
+	// Second attempt - this is where the bug typically manifests
+	err2 := conn.connect(ctx, job)
+	bugFound = checkConnectionError(2, err2)
+	if bugFound {
+		return
+	}
+
+	// If neither attempt reproduced the bug, log that both attempts completed
+	t.Logf("Both connection attempts completed without reproducing the tcp:// prefix bug")
+}

clickhouse_tls_test.go

@@ -0,0 +1,181 @@
+package main
+
+import (
+	"crypto/tls"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/go-kit/log"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestClickHouseTLSConfigurationIntegration(t *testing.T) {
+	// Test the ClickHouse TLS configuration through the actual updateConnections method
+	tests := []struct {
+		name           string
+		driver         string
+		connectionURL  string
+		shouldSetupTLS bool
+		expectedURL    string
+	}{
+		{
+			name:           "clickhouse with tls_config=spiffe should setup TLS",
+			driver:         "clickhouse",
+			connectionURL:  "clickhouse+https://localhost:9000/default?tls_config=spiffe&other_param=value",
+			shouldSetupTLS: true,
+			expectedURL:    "clickhouse+https://localhost:9000/default?other_param=value",
+		},
+		{
+			name:           "clickhouse without tls_config should not setup TLS",
+			driver:         "clickhouse",
+			connectionURL:  "clickhouse://localhost:9000/default?other_param=value",
+			shouldSetupTLS: false,
+			expectedURL:    "clickhouse://localhost:9000/default?other_param=value",
+		},
+		{
+			name:           "non-clickhouse driver should not setup TLS",
+			driver:         "postgres",
+			connectionURL:  "postgres://localhost:5432/db?tls_config=spiffe",
+			shouldSetupTLS: false,
+			expectedURL:    "postgres://localhost:5432/db?tls_config=spiffe",
+		},
+		{
+			name:           "clickhouse with different tls_config should not setup TLS",
+			driver:         "clickhouse",
+			connectionURL:  "clickhouse://localhost:9000/default?tls_config=custom",
+			shouldSetupTLS: false,
+			expectedURL:    "clickhouse://localhost:9000/default?tls_config=custom",
+		},
+		{
+			name:           "remove multiple tls_config parameters",
+			driver:         "clickhouse",
+			connectionURL:  "clickhouse://localhost:9000/default?tls_config=spiffe&param1=value1&tls_config=other&param2=value2",
+			shouldSetupTLS: true,
+			expectedURL:    "clickhouse://localhost:9000/default?param1=value1&param2=value2",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Use a mock that doesn't fail for this test
+			mockSetup := &MockMTLSSetup{ShouldFail: false}
+			job := &Job{
+				log:         log.NewNopLogger(),
+				mtlsSetup:   mockSetup,
+				Connections: []string{tt.connectionURL}, // Set the connection URL
+			}
+
+			// Test the actual updateConnections method (this is what production code uses)
+			job.updateConnections()
+
+			// Verify the results by checking the created connections
+			if tt.shouldSetupTLS {
+				// Should have exactly one connection
+				require.Len(t, job.conns, 1, "Should have exactly one connection")
+				conn := job.conns[0]
+
+				assert.NotNil(t, conn.tlsConfig, "TLS config should be set")
+				assert.Equal(t, uint16(tls.VersionTLS12), conn.tlsConfig.MinVersion, "TLS minimum version should be TLS 1.2")
+				assert.True(t, mockSetup.SetupCalled, "SetupMTLS should have been called")
+				assert.NotNil(t, conn.tlsConfig.GetClientCertificate, "GetClientCertificate should be set")
+				assert.Equal(t, tt.expectedURL, conn.url, "URL should match expected after processing")
+			} else {
+				require.Len(t, job.conns, 1, "Should have exactly one connection")
+				conn := job.conns[0]
+				assert.Nil(t, conn.tlsConfig, "TLS config should not be set for non-ClickHouse drivers")
+				assert.False(t, mockSetup.SetupCalled, "SetupMTLS should not have been called for non-ClickHouse drivers")
+				assert.Equal(t, tt.expectedURL, conn.url, "URL should remain unchanged for non-ClickHouse drivers")
+			}
+		})
+	}
+}
+
+type MockMTLSSetup struct {
+	ShouldFail  bool
+	SetupCalled bool
+}
+
+func (m *MockMTLSSetup) SetupMTLS(job *Job, tlsConfig *tls.Config) error {
+	m.SetupCalled = true
+	if m.ShouldFail {
+		return errors.New("mock setup failure")
+	}
+	// Mock the GetClientCertificate function
+	tlsConfig.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
+		return &tls.Certificate{}, nil
+	}
+	return nil
+}
+
+func TestClickHouseTLSConfigurationWithMocking(t *testing.T) {
+	tests := []struct {
+		name            string
+		connectionURL   string
+		setupShouldFail bool
+		shouldSetupTLS  bool
+		expectedURL     string
+		expectError     bool
+	}{
+		{
+			name:            "successful TLS setup",
+			connectionURL:   "clickhouse://localhost:9000/default?tls_config=spiffe&other=value",
+			setupShouldFail: false,
+			shouldSetupTLS:  true,
+			expectedURL:     "clickhouse://localhost:9000/default?other=value",
+			expectError:     false,
+		},
+		{
+			name:            "mTLS setup failure",
+			connectionURL:   "clickhouse://localhost:9000/default?tls_config=spiffe",
+			setupShouldFail: true,
+			shouldSetupTLS:  false,
+			expectError:     true,
+		},
+		{
+			name:            "no TLS setup needed",
+			connectionURL:   "clickhouse://localhost:9000/default?other=value",
+			setupShouldFail: false,
+			shouldSetupTLS:  false,
+			expectedURL:     "clickhouse://localhost:9000/default?other=value",
+			expectError:     false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockSetup := &MockMTLSSetup{ShouldFail: tt.setupShouldFail}
+			job := &Job{
+				log:       log.NewNopLogger(),
+				mtlsSetup: mockSetup,
+			}
+
+			result := job.configureClickHouseTLS(tt.connectionURL, tt.connectionURL)
+
+			if tt.expectError {
+				assert.Error(t, result.Error)
+				return
+			}
+
+			assert.NoError(t, result.Error)
+			assert.Equal(t, tt.expectedURL, result.ModifiedURL)
+
+			if tt.shouldSetupTLS {
+				assert.NotNil(t, result.TLSConfig)
+				assert.Equal(t, uint16(tls.VersionTLS12), result.TLSConfig.MinVersion)
+				assert.True(t, mockSetup.SetupCalled)
+				assert.NotNil(t, result.TLSConfig.GetClientCertificate)
+			} else {
+				assert.Nil(t, result.TLSConfig)
+				if strings.Contains(tt.connectionURL, "tls_config=spiffe") {
+					// Should have been called but failed
+					assert.True(t, mockSetup.SetupCalled)
+				} else {
+					// Should not have been called at all
+					assert.False(t, mockSetup.SetupCalled)
+				}
+			}
+		})
+	}
+}

config.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
@@ -142,10 +143,38 @@ func (c *cronConfig) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	return nil
 }
 
+// MTLSIdentity allows to configure which TLS certificate will be presented by
+// the client when connecting to the database server
+type MTLSIdentity struct {
+	CertPath string `yaml:"cert_path"`
+	KeyPath  string `yaml:"key_path"`
+}
+
+// MTLSSetup interface for dependency injection of mTLS setup logic
+type MTLSSetup interface {
+	SetupMTLS(job *Job, tlsConfig *tls.Config) error
+}
+
+// FilesystemMTLSSetup implements MTLSSetup using filesystem-based certificates
+type FilesystemMTLSSetup struct{}
+
+// NewFilesystemMTLSSetup creates a new filesystem-based mTLS setup
+func NewFilesystemMTLSSetup() MTLSSetup {
+	return &FilesystemMTLSSetup{}
+}
+
+// TLSConfigResult holds the result of TLS configuration
+type TLSConfigResult struct {
+	TLSConfig   *tls.Config
+	ModifiedURL string
+	Error       error
+}
+
 // Job is a collection of connections and queries
 type Job struct {
 	log          log.Logger
 	conns        []*connection
+	mtlsSetup    MTLSSetup     // Injectable dependency for mTLS setup
 	Name         string        `yaml:"name"`          // name of this job
 	KeepAlive    bool          `yaml:"keepalive"`     // keep connection between runs?
 	Interval     time.Duration `yaml:"interval"`      // interval at which this job is run
@@ -154,6 +183,8 @@ type Job struct {
 	Queries      []*Query      `yaml:"queries"`
 	StartupSQL   []string      `yaml:"startup_sql"` // SQL executed on startup
 	Iterator     Iterator      `yaml:"iterator"`    // Iterator configuration
+	MTLSIdentity MTLSIdentity  `yaml:"mtls_identity"`
+	Timeout      time.Duration `yaml:"timeout"` // timeout for database operations (default: 30s)
 }
 
 type connection struct {
@@ -165,8 +196,9 @@ type connection struct {
 	user                string
 	tokenExpirationTime time.Time
 	iteratorValues      []string
-    	snowflakeConfig *gosnowflake.Config
-    	snowflakeDSN    string
+	snowflakeConfig     *gosnowflake.Config
+	snowflakeDSN        string
+	tlsConfig           *tls.Config
 }
 
 // Query is an SQL query that is executed on a connection

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.34.0
 	github.com/aws/aws-sdk-go v1.50.6
 	github.com/cenkalti/backoff v2.2.1+incompatible
+	github.com/cloudflare/certinel v0.4.1
 	github.com/go-kit/log v0.2.1
 	github.com/go-sql-driver/mysql v1.8.1
 	github.com/gobwas/glob v0.2.3
@@ -18,6 +19,7 @@ require (
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/segmentio/go-athena v0.0.0-20230626212750-5fac08ed8dab
 	github.com/snowflakedb/gosnowflake v1.13.3
+	github.com/stretchr/testify v1.10.0
 	github.com/vertica/vertica-sql-go v1.3.3
 	go.uber.org/automaxprocs v1.5.3
 	google.golang.org/api v0.197.0
@@ -55,10 +57,12 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/danieljoos/wincred v1.2.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dvsekhvalnov/jose2go v1.6.0 // indirect
 	github.com/elastic/go-sysinfo v1.11.2 // indirect
 	github.com/elastic/go-windows v1.0.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.7 // indirect
 	github.com/go-faster/city v1.0.1 // indirect
 	github.com/go-faster/errors v0.7.1 // indirect
@@ -102,6 +106,7 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect

go.sum

@@ -96,6 +96,8 @@ github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/cloudflare/certinel v0.4.1 h1:b0nGqKxEjCe6aS3SoZf0HwjkzfCCAqGzZj8iB9ZJGW0=
+github.com/cloudflare/certinel v0.4.1/go.mod h1:hcx0SA3fmeMzo6egeOzN/29/xfA4+bhZttHvR20a4YA=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I=
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
@@ -129,6 +131,8 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
+github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
+github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
 github.com/gabriel-vasile/mimetype v1.4.7 h1:SKFKl7kD0RiPdbht0s7hFtjl489WcQ1VyPW8ZzUMYCA=
 github.com/gabriel-vasile/mimetype v1.4.7/go.mod h1:GDlAgAyIRT27BhFl53XNAFtfjzOkLaF35JdEG0P7LtU=
 github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw=
@@ -621,6 +625,8 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU=
+gotest.tools/v3 v3.5.1/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=