diff --git a/pulsar/client.go b/pulsar/client.go index ac0d14d21f..433870dd3b 100644 --- a/pulsar/client.go +++ b/pulsar/client.go @@ -18,7 +18,9 @@ package pulsar import ( + "context" "crypto/tls" + "net" "time" "github.com/apache/pulsar-client-go/pulsar/auth" @@ -103,6 +105,11 @@ type ClientOptions struct { // If your application is sensitive to service disruption, set this explicitly (e.g., 10s or 15s). ConnectionTimeout time.Duration + // Dialer is an optional custom dialer used to establish TCP connections to + // brokers. When nil, a net.Dialer honouring ConnectionTimeout is used. + // The address passed to the dialer is the broker's host:port + Dialer func(ctx context.Context, network, addr string) (net.Conn, error) + // Set the operation timeout (default: 30 seconds) // Producer-create, subscribe and unsubscribe operations will be retried until this interval, after which the // operation will be marked as failed diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go index f1cfeb043c..85694ca86e 100644 --- a/pulsar/client_impl.go +++ b/pulsar/client_impl.go @@ -166,7 +166,7 @@ func newClient(options ClientOptions) (Client, error) { c := &client{ cnxPool: internal.NewConnectionPool(tlsConfig, authProvider, connectionTimeout, keepAliveInterval, - maxConnectionsPerHost, logger, metrics, options.Description, connectionMaxIdleTime), + maxConnectionsPerHost, logger, metrics, options.Description, connectionMaxIdleTime, options.Dialer), log: logger, metrics: metrics, memLimit: internal.NewMemoryLimitController(memLimitBytes, defaultMemoryLimitTriggerThreshold), diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go index 9fd8cef36f..67d5b1741b 100644 --- a/pulsar/internal/connection.go +++ b/pulsar/internal/connection.go @@ -138,6 +138,7 @@ type dataRequest struct { type connection struct { started int32 connectionTimeout time.Duration + dialer func(ctx context.Context, network, addr string) (net.Conn, error) closeOnce sync.Once // mu protects the fields below against concurrency accesses. @@ -182,6 +183,7 @@ type connection struct { // connectionOptions defines configurations for creating connection. type connectionOptions struct { + dialer func(ctx context.Context, network, addr string) (net.Conn, error) logicalAddr *url.URL physicalAddr *url.URL tls *TLSOptions @@ -195,6 +197,7 @@ type connectionOptions struct { func newConnection(opts connectionOptions) *connection { cnx := &connection{ + dialer: opts.dialer, connectionTimeout: opts.connectionTimeout, keepAliveInterval: opts.keepAliveInterval, logicalAddr: opts.logicalAddr, @@ -258,13 +261,31 @@ func (c *connection) connect() bool { tlsConfig *tls.Config ) + // time.Duration is initialized to 0 by default, net.Dialer's default timeout is no timeout + // therefore if c.connectionTimeout is 0, it means no timeout. + // As in tls.DialWithDialer, this deadline spans the dial and the TLS + // handshake together, not just the dial. + ctx := context.Background() + if c.connectionTimeout.Nanoseconds() > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, c.connectionTimeout) + defer cancel() + } + + // kept separate from the TLS handshake below so that a custom dialer returns + // a plain net.Conn and the handshake is performed here, not inside + // tls.DialWithDialer. + dial := func() (net.Conn, error) { + if c.dialer != nil { + return c.dialer(ctx, "tcp", c.physicalAddr.Host) + } + var d net.Dialer + return d.DialContext(ctx, "tcp", c.physicalAddr.Host) + } + if c.tlsOptions == nil { // Clear text connection - if c.connectionTimeout.Nanoseconds() > 0 { - cnx, err = net.DialTimeout("tcp", c.physicalAddr.Host, c.connectionTimeout) - } else { - cnx, err = net.Dial("tcp", c.physicalAddr.Host) - } + cnx, err = dial() } else { // TLS connection tlsConfig, err = c.getTLSConfig() @@ -273,10 +294,28 @@ func (c *connection) connect() bool { return false } - // time.Duration is initialized to 0 by default, net.Dialer's default timeout is no timeout - // therefore if c.connectionTimeout is 0, it means no timeout - d := &net.Dialer{Timeout: c.connectionTimeout} - cnx, err = tls.DialWithDialer(d, "tcp", c.physicalAddr.Host, tlsConfig) + // tls.DialWithDialer infers ServerName from the address being dialed when + // the config does not set one; tls.Client does not, and would verify + // against an empty name. Preserve that so behaviour is unchanged + // if ServerName is left empty. + if tlsConfig.ServerName == "" { + host := c.physicalAddr.Hostname() + if host != "" { + tlsConfig = tlsConfig.Clone() + tlsConfig.ServerName = host + } + } + + var rawConn net.Conn + rawConn, err = dial() + if err == nil { + tlsConn := tls.Client(rawConn, tlsConfig) + if err = tlsConn.HandshakeContext(ctx); err != nil { + _ = rawConn.Close() + } else { + cnx = tlsConn + } + } } if err != nil { diff --git a/pulsar/internal/connection_pool.go b/pulsar/internal/connection_pool.go index cd082188b6..8870cc4875 100644 --- a/pulsar/internal/connection_pool.go +++ b/pulsar/internal/connection_pool.go @@ -18,7 +18,9 @@ package internal import ( + "context" "fmt" + "net" "net/url" "sync" "sync/atomic" @@ -58,6 +60,7 @@ type connectionPool struct { metrics *Metrics log log.Logger description string + dialer func(ctx context.Context, network, addr string) (net.Conn, error) } // NewConnectionPool init connection pool. @@ -70,7 +73,8 @@ func NewConnectionPool( logger log.Logger, metrics *Metrics, description string, - connectionMaxIdleTime time.Duration) ConnectionPool { + connectionMaxIdleTime time.Duration, + dialer func(ctx context.Context, network, addr string) (net.Conn, error)) ConnectionPool { p := &connectionPool{ connections: make(map[string]*connection), tlsOptions: tlsOptions, @@ -82,6 +86,7 @@ func NewConnectionPool( metrics: metrics, closeCh: make(chan struct{}), description: description, + dialer: dialer, } go p.checkAndCleanIdleConnections(connectionMaxIdleTime) return p @@ -114,6 +119,7 @@ func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.U if conn == nil { conn = newConnection(connectionOptions{ + dialer: p.dialer, logicalAddr: logicalAddr, physicalAddr: physicalAddr, tls: p.tlsOptions, diff --git a/pulsar/internal/connection_test.go b/pulsar/internal/connection_test.go index 92831cab93..6b089c27ee 100644 --- a/pulsar/internal/connection_test.go +++ b/pulsar/internal/connection_test.go @@ -19,12 +19,16 @@ package internal import ( "context" + "crypto/tls" + "errors" + "net" "net/url" "sync" "sync/atomic" "testing" "time" + "github.com/apache/pulsar-client-go/pulsar/auth" pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto" "github.com/apache/pulsar-client-go/pulsar/log" "github.com/prometheus/client_golang/prometheus" @@ -224,3 +228,257 @@ func newMockMetrics() *Metrics { }), } } + +func testConnectionOptions(t *testing.T, physicalAddr string) connectionOptions { + t.Helper() + + addr, err := url.Parse(physicalAddr) + assert.NoError(t, err) + + return connectionOptions{ + logicalAddr: addr, + physicalAddr: addr, + auth: auth.NewAuthDisabled(), + logger: log.DefaultNopLogger(), + metrics: newMockMetrics(), + } +} + +// listen starts a TCP listener that accepts and immediately closes connections, +// so connect() can complete a plaintext dial without a broker. +func listen(t *testing.T) net.Listener { + t.Helper() + + l, err := net.Listen("tcp", "127.0.0.1:0") + assert.NoError(t, err) + t.Cleanup(func() { _ = l.Close() }) + + go func() { + for { + c, err := l.Accept() + if err != nil { + return + } + _ = c.Close() + } + }() + return l +} + +func TestConnectionDialerIsUsed(t *testing.T) { + l := listen(t) + + var ( + gotNetwork string + gotAddr string + gotSet bool + ) + + opts := testConnectionOptions(t, "pulsar://"+l.Addr().String()) + opts.connectionTimeout = 10 * time.Second + opts.dialer = func(ctx context.Context, network, addr string) (net.Conn, error) { + gotNetwork, gotAddr, gotSet = network, addr, true + + deadline, ok := ctx.Deadline() + assert.True(t, ok, "dialer context should carry the connection timeout") + assert.WithinDuration(t, time.Now().Add(10*time.Second), deadline, time.Second) + + return net.Dial(network, addr) + } + + cnx := newConnection(opts) + assert.True(t, cnx.connect()) + cnx.Close() + + assert.True(t, gotSet, "dialer should have been called") + assert.Equal(t, "tcp", gotNetwork) + assert.Equal(t, l.Addr().String(), gotAddr) +} + +func TestConnectionDialerNoTimeoutHasNoDeadline(t *testing.T) { + l := listen(t) + + opts := testConnectionOptions(t, "pulsar://"+l.Addr().String()) + opts.dialer = func(ctx context.Context, network, addr string) (net.Conn, error) { + _, ok := ctx.Deadline() + assert.False(t, ok, "no ConnectionTimeout should mean no deadline") + return net.Dial(network, addr) + } + + cnx := newConnection(opts) + assert.True(t, cnx.connect()) + cnx.Close() +} + +func TestConnectionDialerError(t *testing.T) { + l := listen(t) + + opts := testConnectionOptions(t, "pulsar://"+l.Addr().String()) + opts.dialer = func(_ context.Context, _, _ string) (net.Conn, error) { + return nil, errors.New("dial rejected") + } + + cnx := newConnection(opts) + assert.False(t, cnx.connect(), "a dialer error should fail the connection") +} + +// The dialer returns a plain net.Conn and the library performs the TLS +// handshake itself, so a custom dialer must not bypass certificate +// verification. +func TestConnectionDialerWithTLS(t *testing.T) { + cert, err := tls.LoadX509KeyPair("../../integration-tests/certs/broker-cert.pem", + "../../integration-tests/certs/broker-key.pem") + assert.NoError(t, err) + + l, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: []tls.Certificate{cert}}) + assert.NoError(t, err) + defer l.Close() + + go func() { + for { + c, err := l.Accept() + if err != nil { + return + } + _ = c.(*tls.Conn).Handshake() + _ = c.Close() + } + }() + + dialed := false + opts := testConnectionOptions(t, "pulsar+ssl://"+l.Addr().String()) + opts.tls = &TLSOptions{TrustCertsFilePath: "../../integration-tests/certs/cacert.pem"} + opts.dialer = func(ctx context.Context, network, addr string) (net.Conn, error) { + dialed = true + return net.Dial(network, addr) + } + + cnx := newConnection(opts) + // The broker certificate is not valid for 127.0.0.1, so the handshake the + // library performs on the dialer's conn must reject it. + assert.False(t, cnx.connect()) + assert.True(t, dialed, "dialer should be used for TLS connections too") + + // ...and it succeeds once the name matches. + opts.tls.ServerName = "localhost" + opts.tls.ValidateHostname = true + cnx = newConnection(opts) + assert.True(t, cnx.connect()) + cnx.Close() +} + +// With ValidateHostname off, getTLSConfig() leaves ServerName empty. +// tls.DialWithDialer used to infer it from the dialed address; tls.Client does +// not, so connect() fills it in and verification still works. +func TestConnectionTLSServerNameInferredFromAddress(t *testing.T) { + cert, err := tls.LoadX509KeyPair("../../integration-tests/certs/broker-cert.pem", + "../../integration-tests/certs/broker-key.pem") + assert.NoError(t, err) + + l, err := tls.Listen("tcp", "localhost:0", &tls.Config{Certificates: []tls.Certificate{cert}}) + assert.NoError(t, err) + defer l.Close() + + go func() { + for { + c, err := l.Accept() + if err != nil { + return + } + _ = c.(*tls.Conn).Handshake() + _ = c.Close() + } + }() + + _, port, err := net.SplitHostPort(l.Addr().String()) + assert.NoError(t, err) + + opts := testConnectionOptions(t, "pulsar+ssl://localhost:"+port) + opts.tls = &TLSOptions{TrustCertsFilePath: "../../integration-tests/certs/cacert.pem"} + + cnx := newConnection(opts) + assert.True(t, cnx.connect()) + cnx.Close() +} + +// A peer that completes the TCP accept then never speaks TLS. +func TestConnectionTLSHandshakeBlackhole(t *testing.T) { + l, err := net.Listen("tcp", "127.0.0.1:0") + assert.NoError(t, err) + defer l.Close() + + accepted := make(chan net.Conn, 4) + go func() { + for { + c, err := l.Accept() + if err != nil { + return + } + accepted <- c // hold it open, send nothing + } + }() + + _, port, _ := net.SplitHostPort(l.Addr().String()) + opts := testConnectionOptions(t, "pulsar+ssl://localhost:"+port) + opts.connectionTimeout = 2 * time.Second + opts.tls = &TLSOptions{TrustCertsFilePath: "../../integration-tests/certs/cacert.pem"} + + cnx := newConnection(opts) + + done := make(chan bool, 1) + start := time.Now() + go func() { done <- cnx.connect() }() + + select { + case ok := <-done: + assert.False(t, ok) + t.Logf("connect() returned after %v", time.Since(start)) + case <-time.After(10 * time.Second): + t.Fatalf("connect() HUNG: still blocked after 10s with ConnectionTimeout=2s") + } +} + +// After connect() returns, the connection must remain usable indefinitely: +// the connect deadline must not linger on the socket. +func TestConnectionNoLingeringDeadlineAfterHandshake(t *testing.T) { + cert, err := tls.LoadX509KeyPair("../../integration-tests/certs/broker-cert.pem", + "../../integration-tests/certs/broker-key.pem") + assert.NoError(t, err) + + l, err := tls.Listen("tcp", "localhost:0", &tls.Config{Certificates: []tls.Certificate{cert}}) + assert.NoError(t, err) + defer l.Close() + + srvDone := make(chan struct{}) + go func() { + defer close(srvDone) + c, err := l.Accept() + if err != nil { + return + } + _ = c.(*tls.Conn).Handshake() + // Stay silent past the connect timeout, then send a byte. + time.Sleep(1500 * time.Millisecond) + _, _ = c.Write([]byte{0x42}) + time.Sleep(2 * time.Second) + _ = c.Close() + }() + + _, port, _ := net.SplitHostPort(l.Addr().String()) + opts := testConnectionOptions(t, "pulsar+ssl://localhost:"+port) + opts.connectionTimeout = 500 * time.Millisecond + opts.tls = &TLSOptions{TrustCertsFilePath: "../../integration-tests/certs/cacert.pem"} + + cnx := newConnection(opts) + assert.True(t, cnx.connect()) + + // Read well after the 500ms connect timeout would have elapsed. + buf := make([]byte, 1) + n, err := cnx.cnx.Read(buf) + assert.NoError(t, err, "read after connect timeout elapsed must not fail with i/o timeout") + assert.Equal(t, 1, n) + assert.Equal(t, byte(0x42), buf[0]) + + cnx.Close() + <-srvDone +}