feat(pam): clickhouse

packages/gateway-v2/capabilities.go

@@ -0,0 +1,3 @@
+package gatewayv2
+
+const CapabilitySupportedAccountTypes = "supported_account_types"

packages/gateway-v2/gateway.go

@@ -460,6 +460,7 @@ func (g *Gateway) registerHeartBeat(ctx context.Context, errCh chan error) {
 		if g.pkcs11Module != nil {
 			capabilities[CapabilityPkcs11] = true
 		}
+		capabilities[CapabilitySupportedAccountTypes] = pam.GetSupportedResourceTypes()
 		req := api.GatewayHeartbeatRequest{Capabilities: capabilities}
 		if err := api.CallGatewayHeartBeatV2(g.httpClient, req); err != nil {
 			log.Warn().Msgf("Heartbeat failed: %v", err)

packages/gateway-v2/test_connection_handler.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 	"time"
 
+	clickhousehandler "github.com/Infisical/infisical-merge/packages/pam/handlers/clickhouse"
 	mssqlhandler "github.com/Infisical/infisical-merge/packages/pam/handlers/mssql"
 	oraclehandler "github.com/Infisical/infisical-merge/packages/pam/handlers/oracle"
 	snowflakehandler "github.com/Infisical/infisical-merge/packages/pam/handlers/snowflake"
@@ -62,6 +63,7 @@ const (
 	testConnModeKubernetes = "kubernetes"
 	testConnModeSSH        = "ssh"
 	testConnModeSnowflake  = "snowflake"
+	testConnModeClickhouse = "clickhouse"
 	testConnModeTCP        = "tcp"
 )
 
@@ -118,6 +120,15 @@ type snowflakeTestParams struct {
 	Role                 string `json:"role"`
 }
 
+type clickhouseTestParams struct {
+	Username              string `json:"username"`
+	Password              string `json:"password"`
+	Database              string `json:"database"`
+	SslEnabled            bool   `json:"sslEnabled"`
+	SslRejectUnauthorized *bool  `json:"sslRejectUnauthorized"`
+	SslCertificate        string `json:"sslCertificate"`
+}
+
 type ldapTestParams struct {
 	Username               string `json:"username"`
 	Password               string `json:"password"`
@@ -689,6 +700,32 @@ func handleTestConnection(w http.ResponseWriter, r *http.Request) {
 			defer proxy.Close()
 			return authFailure(proxy.Probe(ctx))
 		}
+	case testConnModeClickhouse:
+		var params clickhouseTestParams
+		if !decode(&params) {
+			return
+		}
+		redactSecrets = append(redactSecrets, params.Password)
+		op = func() error {
+			var tlsConfig *tls.Config
+			if params.SslEnabled {
+				var err error
+				if tlsConfig, err = buildTestTLSConfig(target.host, params.SslCertificate, params.SslRejectUnauthorized); err != nil {
+					return connectFailure(err)
+				}
+			}
+			if err := dialTarget(ctx, target.host, target.port); err != nil {
+				return connectFailure(err)
+			}
+			return authFailure(clickhousehandler.TestConnection(ctx, clickhousehandler.ClickHouseProxyConfig{
+				TargetAddr: net.JoinHostPort(target.host, strconv.Itoa(target.port)),
+				Username:   params.Username,
+				Password:   params.Password,
+				Database:   params.Database,
+				EnableTLS:  params.SslEnabled,
+				TLSConfig:  tlsConfig,
+			}))
+		}
 	case testConnModeSSH:
 		var params sshTestParams
 		if !decode(&params) {

packages/pam/handlers/clickhouse/listener.go

@@ -0,0 +1,54 @@
+package clickhouse
+
+import (
+	"errors"
+	"net"
+	"net/http"
+	"sync"
+)
+
+type singleConnListener struct {
+	conns  chan net.Conn
+	closed chan struct{}
+	once   sync.Once
+}
+
+func newSingleConnListener(conn net.Conn) *singleConnListener {
+	listener := &singleConnListener{conns: make(chan net.Conn, 1), closed: make(chan struct{})}
+	listener.conns <- &closeNotifyConn{Conn: conn, onClose: listener.Close}
+	return listener
+}
+
+type closeNotifyConn struct {
+	net.Conn
+	onClose func() error
+	once    sync.Once
+}
+
+func (c *closeNotifyConn) Close() error {
+	err := c.Conn.Close()
+	c.once.Do(func() { _ = c.onClose() })
+	return err
+}
+
+func (l *singleConnListener) Accept() (net.Conn, error) {
+	select {
+	case conn := <-l.conns:
+		return conn, nil
+	case <-l.closed:
+		return nil, net.ErrClosed
+	}
+}
+
+func (l *singleConnListener) Close() error {
+	l.once.Do(func() { close(l.closed) })
+	return nil
+}
+
+func (l *singleConnListener) Addr() net.Addr {
+	return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}
+}
+
+func isListenerDone(err error) bool {
+	return errors.Is(err, net.ErrClosed) || errors.Is(err, http.ErrServerClosed)
+}