fix: fix tls insecure connection (#731)

Signed-off-by: Zixuan Liu <[email protected]>

pkg/auth/auth_provider.go

@@ -37,7 +37,7 @@ type Transport struct {
 	T http.RoundTripper
 }
 
-func GetAuthProvider(config *common.Config) (*Provider, error) {
+func GetAuthProvider(config *common.Config) (Provider, error) {
 	var provider Provider
 	defaultTransport, err := NewDefaultTransport(config)
 	if err != nil {
@@ -65,7 +65,7 @@ func GetAuthProvider(config *common.Config) (*Provider, error) {
 				config.IssuerEndpoint, config.ClientID, config.Audience, config.Scope, defaultTransport)
 		}
 	}
-	return &provider, err
+	return provider, err
 }
 
 // GetDefaultTransport gets a default transport.

pkg/pulsar/admin.go

@@ -18,7 +18,6 @@
 package pulsar
 
 import (
-	"fmt"
 	"net/http"
 	"net/url"
 	"path"
@@ -61,28 +60,11 @@ type pulsarClient struct {
 
 // New returns a new client
 func New(config *common.Config) (Client, error) {
-	if len(config.WebServiceURL) == 0 {
-		config.WebServiceURL = DefaultWebServiceURL
-	}
-
-	c := &pulsarClient{
-		APIVersion: config.PulsarAPIVersion,
-		Client: &cli.Client{
-			ServiceURL:  config.WebServiceURL,
-			VersionInfo: ReleaseVersion,
-			HTTPClient: &http.Client{
-				Timeout: DefaultHTTPTimeOutDuration,
-			},
-		},
-	}
-
 	authProvider, err := auth.GetAuthProvider(config)
-	if !utils.IsNilFixed(authProvider) {
-		c.Client.HTTPClient.Transport = *authProvider
-	} else {
-		fmt.Printf("No Auth Provider found\n")
+	if err != nil {
+		return nil, err
 	}
-	return c, err
+	return NewPulsarClientWithAuthProvider(config, authProvider)
 }
 
 // NewWithAuthProvider creates a client with auth provider.
@@ -98,12 +80,31 @@ func NewWithAuthProvider(config *common.Config, authProvider auth.Provider) Clie
 // NewPulsarClientWithAuthProvider create a client with auth provider.
 func NewPulsarClientWithAuthProvider(config *common.Config,
 	authProvider auth.Provider) (Client, error) {
-	defaultTransport, err := auth.NewDefaultTransport(config)
-	if err != nil {
-		return nil, err
+	var transport http.RoundTripper
+
+	if authProvider != nil {
+		transport = authProvider.Transport()
+		if transport != nil {
+			transport = authProvider
+		}
+	}
+
+	if transport == nil {
+		defaultTransport, err := auth.NewDefaultTransport(config)
+		if err != nil {
+			return nil, err
+		}
+		if authProvider != nil {
+			authProvider.WithTransport(authProvider)
+		} else {
+			transport = defaultTransport
+		}
 	}
 
-	authProvider.WithTransport(defaultTransport)
+	webServiceURL := config.WebServiceURL
+	if len(webServiceURL) == 0 {
+		config.WebServiceURL = DefaultWebServiceURL
+	}
 
 	c := &pulsarClient{
 		APIVersion: config.PulsarAPIVersion,
@@ -112,7 +113,7 @@ func NewPulsarClientWithAuthProvider(config *common.Config,
 			VersionInfo: ReleaseVersion,
 			HTTPClient: &http.Client{
 				Timeout:   DefaultHTTPTimeOutDuration,
-				Transport: authProvider,
+				Transport: transport,
 			},
 		},
 	}

pkg/pulsar/admin_test.go

@@ -18,11 +18,13 @@
 package pulsar
 
 import (
+	"net/http"
 	"testing"
 
+	"github.com/streamnative/pulsar-admin-go/pkg/auth"
 	"github.com/streamnative/pulsar-admin-go/pkg/pulsar/common"
-
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestPulsarClientEndpointEscapes(t *testing.T) {
@@ -31,3 +33,74 @@ func TestPulsarClientEndpointEscapes(t *testing.T) {
 	expected := "/admin/v2/myendpoint/abc%25%3F%20%2Fdef/ghi"
 	assert.Equal(t, expected, actual)
 }
+
+func TestNew(t *testing.T) {
+	config := &common.Config{}
+	admin, err := New(config)
+	require.NoError(t, err)
+	require.NotNil(t, admin)
+}
+
+func TestNewWithAuthProvider(t *testing.T) {
+	config := &common.Config{}
+
+	tokenAuth, err := auth.NewAuthenticationToken("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."+
+		"eyJzdWIiOiJhZG1pbiIsImlhdCI6MTUxNjIzOTAyMn0.sVt6cyu3HKd89LcQvZVMNbqT0DTl3FvG9oYbj8hBDqU", nil)
+	require.NoError(t, err)
+	require.NotNil(t, tokenAuth)
+
+	admin, err := NewPulsarClientWithAuthProvider(config, tokenAuth)
+	require.NoError(t, err)
+	require.NotNil(t, admin)
+}
+
+type customAuthProvider struct {
+	transport http.RoundTripper
+}
+
+var _ auth.Provider = &customAuthProvider{}
+
+func (c *customAuthProvider) RoundTrip(req *http.Request) (*http.Response, error) {
+	panic("implement me")
+}
+
+func (c *customAuthProvider) Transport() http.RoundTripper {
+	return c.transport
+}
+
+func (c *customAuthProvider) WithTransport(transport http.RoundTripper) {
+	c.transport = transport
+}
+
+func TestNewWithCustomAuthProviderWithTransport(t *testing.T) {
+	config := &common.Config{}
+	defaultTransport, err := auth.NewDefaultTransport(config)
+	require.NoError(t, err)
+
+	customAuthProvider := &customAuthProvider{
+		transport: defaultTransport,
+	}
+
+	admin, err := NewPulsarClientWithAuthProvider(config, customAuthProvider)
+	require.NoError(t, err)
+	require.NotNil(t, admin)
+
+	// Expected the transport of customAuthProvider will not be overwritten.
+	require.Equal(t, defaultTransport, admin.(*pulsarClient).Client.HTTPClient.Transport)
+}
+
+func TestNewWithTlsAllowInsecure(t *testing.T) {
+	config := &common.Config{
+		TLSAllowInsecureConnection: true,
+	}
+	admin, err := New(config)
+	require.NoError(t, err)
+	require.NotNil(t, admin)
+
+	pulsarClientS := admin.(*pulsarClient)
+	require.NotNil(t, pulsarClientS.Client.HTTPClient.Transport)
+	tr := pulsarClientS.Client.HTTPClient.Transport.(*http.Transport)
+	require.NotNil(t, tr)
+	require.NotNil(t, tr.TLSClientConfig)
+	require.True(t, tr.TLSClientConfig.InsecureSkipVerify)
+}

pkg/pulsar/utils/utils_test.go

@@ -58,4 +58,10 @@ func TestIsNilFixed(t *testing.T) {
 
 	var ch chan string
 	assert.True(t, IsNilFixed(ch))
+
+	var nilInterface People
+	assert.True(t, IsNilFixed(nilInterface))
+
+	// pointer to an interface, the IsNilFixed method cannot check this.
+	assert.False(t, IsNilFixed(&nilInterface))
 }