|
9 | 9 | "fmt" |
10 | 10 | "log/slog" |
11 | 11 | "net/url" |
| 12 | + "os" |
| 13 | + "path/filepath" |
12 | 14 | "slices" |
13 | 15 | "strings" |
14 | 16 | "time" |
@@ -232,32 +234,114 @@ func (c *ClickHouseConnector) ValidateCheck(ctx context.Context) error { |
232 | 234 | return nil |
233 | 235 | } |
234 | 236 |
|
235 | | -func Connect(ctx context.Context, env map[string]string, config *protos.ClickhouseConfig) (clickhouse.Conn, error) { |
236 | | - var tlsSetting *tls.Config |
237 | | - if !config.DisableTls { |
238 | | - tlsSetting = &tls.Config{MinVersion: tls.VersionTLS13} |
239 | | - if config.Certificate != nil || config.PrivateKey != nil { |
240 | | - if config.Certificate == nil || config.PrivateKey == nil { |
241 | | - return nil, fmt.Errorf("both certificate and private key must be provided if using certificate-based authentication") |
242 | | - } |
243 | | - cert, err := tls.X509KeyPair([]byte(*config.Certificate), []byte(*config.PrivateKey)) |
244 | | - if err != nil { |
245 | | - return nil, fmt.Errorf("failed to parse provided certificate: %w", err) |
246 | | - } |
247 | | - tlsSetting.Certificates = []tls.Certificate{cert} |
| 237 | +// configureDirectoryTLS configures the tls.Config by loading certificate files |
| 238 | +// from a directory. It expects tls.crt and tls.key files, and optionally ca.crt. |
| 239 | +// This is typically used when a Kubernetes Secret is mounted as a volume. |
| 240 | +// |
| 241 | +// Client certificates are loaded via GetClientCertificate so that every TLS |
| 242 | +// handshake re-reads the files from disk, automatically picking up rotated |
| 243 | +// certificates (e.g. renewed by cert-manager) without requiring a reconnect. |
| 244 | +func configureDirectoryTLS(tlsConfig *tls.Config, dir string) error { |
| 245 | + certPath := filepath.Join(dir, "tls.crt") |
| 246 | + keyPath := filepath.Join(dir, "tls.key") |
| 247 | + caPath := filepath.Join(dir, "ca.crt") |
| 248 | + |
| 249 | + // Verify that the required files are readable at configuration time |
| 250 | + if _, err := os.ReadFile(certPath); err != nil { |
| 251 | + return fmt.Errorf("failed to read TLS certificate from %q: %w", certPath, err) |
| 252 | + } |
| 253 | + if _, err := os.ReadFile(keyPath); err != nil { |
| 254 | + return fmt.Errorf("failed to read TLS private key from %q: %w", keyPath, err) |
| 255 | + } |
| 256 | + |
| 257 | + // Load CA certificate upfront — RootCAs must be set before the TLS |
| 258 | + // handshake so the server certificate can be verified. |
| 259 | + caCertPEM, err := os.ReadFile(caPath) |
| 260 | + if err == nil && len(caCertPEM) > 0 { |
| 261 | + caPool := x509.NewCertPool() |
| 262 | + if !caPool.AppendCertsFromPEM(caCertPEM) { |
| 263 | + return fmt.Errorf("failed to parse CA certificate from %q", caPath) |
248 | 264 | } |
249 | | - if config.RootCa != nil { |
250 | | - caPool := x509.NewCertPool() |
251 | | - if !caPool.AppendCertsFromPEM([]byte(*config.RootCa)) { |
252 | | - return nil, fmt.Errorf("failed to parse provided root CA") |
253 | | - } |
254 | | - tlsSetting.RootCAs = caPool |
| 265 | + tlsConfig.RootCAs = caPool |
| 266 | + } |
| 267 | + |
| 268 | + // Use GetClientCertificate so the cert/key are re-read on every TLS |
| 269 | + // handshake, ensuring rotated certificates are picked up immediately. |
| 270 | + tlsConfig.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) { |
| 271 | + certPEM, err := os.ReadFile(certPath) |
| 272 | + if err != nil { |
| 273 | + return nil, fmt.Errorf("failed to read TLS certificate from %q: %w", certPath, err) |
| 274 | + } |
| 275 | + keyPEM, err := os.ReadFile(keyPath) |
| 276 | + if err != nil { |
| 277 | + return nil, fmt.Errorf("failed to read TLS private key from %q: %w", keyPath, err) |
| 278 | + } |
| 279 | + cert, err := tls.X509KeyPair(certPEM, keyPEM) |
| 280 | + if err != nil { |
| 281 | + return nil, fmt.Errorf("failed to parse TLS certificate from directory %q: %w", dir, err) |
| 282 | + } |
| 283 | + return &cert, nil |
| 284 | + } |
| 285 | + |
| 286 | + return nil |
| 287 | +} |
| 288 | + |
| 289 | +// configureInlineTLS configures the tls.Config using inline certificate/key |
| 290 | +// PEM values from the ClickHouse peer config. |
| 291 | +func configureInlineTLS(tlsConfig *tls.Config, config *protos.ClickhouseConfig) error { |
| 292 | + if config.Certificate != nil || config.PrivateKey != nil { |
| 293 | + if config.Certificate == nil || config.PrivateKey == nil { |
| 294 | + return errors.New("both certificate and private key must be provided if using certificate-based authentication") |
| 295 | + } |
| 296 | + cert, err := tls.X509KeyPair([]byte(*config.Certificate), []byte(*config.PrivateKey)) |
| 297 | + if err != nil { |
| 298 | + return fmt.Errorf("failed to parse provided certificate: %w", err) |
| 299 | + } |
| 300 | + tlsConfig.Certificates = []tls.Certificate{cert} |
| 301 | + } |
| 302 | + if config.RootCa != nil { |
| 303 | + caPool := x509.NewCertPool() |
| 304 | + if !caPool.AppendCertsFromPEM([]byte(*config.RootCa)) { |
| 305 | + return errors.New("failed to parse provided root CA") |
| 306 | + } |
| 307 | + tlsConfig.RootCAs = caPool |
| 308 | + } |
| 309 | + return nil |
| 310 | +} |
| 311 | + |
| 312 | +// buildTLSConfig builds the TLS configuration for a ClickHouse connection. |
| 313 | +// When a TLS certificate directory is configured, it loads certificates from the directory. |
| 314 | +// Otherwise, inline certificates are used. |
| 315 | +func buildTLSConfig(config *protos.ClickhouseConfig) (*tls.Config, error) { |
| 316 | + if config.DisableTls { |
| 317 | + return nil, nil |
| 318 | + } |
| 319 | + |
| 320 | + tlsConfig := &tls.Config{ |
| 321 | + MinVersion: tls.VersionTLS13, |
| 322 | + ServerName: config.TlsHost, |
| 323 | + } |
| 324 | + certDir := config.GetTlsCertificateDirectory() |
| 325 | + |
| 326 | + if certDir != "" { |
| 327 | + if err := configureDirectoryTLS(tlsConfig, certDir); err != nil { |
| 328 | + return nil, err |
255 | 329 | } |
256 | | - if config.TlsHost != "" { |
257 | | - tlsSetting.ServerName = config.TlsHost |
| 330 | + } else { |
| 331 | + if err := configureInlineTLS(tlsConfig, config); err != nil { |
| 332 | + return nil, err |
258 | 333 | } |
259 | 334 | } |
260 | 335 |
|
| 336 | + return tlsConfig, nil |
| 337 | +} |
| 338 | + |
| 339 | +func Connect(ctx context.Context, env map[string]string, config *protos.ClickhouseConfig) (clickhouse.Conn, error) { |
| 340 | + tlsSetting, err := buildTLSConfig(config) |
| 341 | + if err != nil { |
| 342 | + return nil, err |
| 343 | + } |
| 344 | + |
261 | 345 | settings := clickhouse.Settings{ |
262 | 346 | // See: https://clickhouse.com/docs/en/cloud/reference/shared-merge-tree#consistency |
263 | 347 | "select_sequential_consistency": uint64(1), |
|
0 commit comments