Configurable Pulsar TLS parameters (#51)

* configuration Pulsar TLS parameters, consolidate pulsar.Client creation in a function

* upversion apache-go-client to pick brokerTlsUrl redirect fix

Author: zzzming

go.mod

@@ -4,7 +4,7 @@ go 1.13
 
 require (
 	github.com/DataDog/zstd v1.4.4 // indirect
-	github.com/apache/pulsar-client-go v0.0.0-20200214184451-fc390a6a37f3
+	github.com/apache/pulsar-client-go v0.1.1-0.20200425133951-6edc8f4ef954
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/ghodss/yaml v1.0.0
 	github.com/golang/snappy v0.0.1 // indirect
@@ -18,5 +18,4 @@ require (
 	github.com/xdg/stringprep v1.0.0 // indirect
 	go.mongodb.org/mongo-driver v1.2.0
 	golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413 // indirect
-	golang.org/x/text v0.3.2 // indirect
 )

go.sum

@@ -12,7 +12,7 @@ ALL_PKGS=""
 
 cd $DIR/../src
 for d in */ ; do
-    if [[ ${d} != "unit-test/" && ${d} != "e2e/" ]] # exclude unit-test for test coverage
+    if [[ ${d} != "unit-test/" && ${d} != "e2e/" && ${d} != "docs/" ]] # exclude unit-test for test coverage
     then
         pkg=${d%/}
         ALL_PKGS=${ALL_PKGS}","${BASE_PKG_DIR}${pkg}

scripts/test_coverage.sh

@@ -4,13 +4,13 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar"
 	"github.com/kafkaesque-io/pulsar-beam/src/model"
+	"github.com/kafkaesque-io/pulsar-beam/src/pulsardriver"
 	"github.com/kafkaesque-io/pulsar-beam/src/util"
 
 	log "github.com/sirupsen/logrus"
@@ -52,27 +52,13 @@ func (s *PulsarHandler) Init() error {
 	s.topicName = util.GetConfig().DbName
 	tokenStr := util.GetConfig().DbPassword
 
-	s.logger.Infof("pulsar URL: %s token: %s", pulsarURL, tokenStr)
-	clientOpt := pulsar.ClientOptions{
-		URL:               pulsarURL,
-		OperationTimeout:  30 * time.Second,
-		ConnectionTimeout: 30 * time.Second,
-	}
-
-	if tokenStr != "" {
-		clientOpt.Authentication = pulsar.NewAuthenticationToken(tokenStr)
-	}
-
-	if strings.HasPrefix(pulsarURL, "pulsar+ssl://") {
-		trustStore := util.GetConfig().TrustStore //"/etc/ssl/certs/ca-bundle.crt"
-		if trustStore == "" {
-			return fmt.Errorf("this is fatal that we are missing trustStore while pulsar+ssl is required")
-		}
-		clientOpt.TLSTrustCertsFilePath = trustStore
+	s.logger.Infof("database pulsar URL: %s", pulsarURL)
+	if log.GetLevel() == log.DebugLevel {
+		s.logger.Debugf("database pulsar token string is %s", tokenStr)
 	}
 
 	var err error
-	s.client, err = pulsar.NewClient(clientOpt)
+	s.client, err = pulsardriver.NewPulsarClient(pulsarURL, tokenStr)
 	if err != nil {
 		// this would be a serious problem so that we return with error
 		return err

src/db/pulsardb.go

@@ -61,23 +61,7 @@ func (c *PulsarClient) GetClient(url, tokenStr string) (pulsar.Client, error) {
 		return c.client, nil
 	}
 
-	clientOpt := pulsar.ClientOptions{
-		URL:               url,
-		OperationTimeout:  time.Duration(clientOpsTimeout) * time.Second,
-		ConnectionTimeout: time.Duration(clientConnectTimeout) * time.Second,
-	}
-
-	if tokenStr != "" {
-		clientOpt.Authentication = pulsar.NewAuthenticationToken(tokenStr)
-	}
-
-	if strings.HasPrefix(url, "pulsar+ssl://") {
-		trustStore := util.AssignString(util.GetConfig().TrustStore, "/etc/ssl/certs/ca-bundle.crt")
-		clientOpt.TLSTrustCertsFilePath = trustStore
-	}
-
-	driver, err := pulsar.NewClient(clientOpt)
-
+	driver, err := NewPulsarClient(url, tokenStr)
 	if err != nil {
 		log.Errorf("failed instantiate pulsar client %v", err)
 		return nil, fmt.Errorf("Could not instantiate Pulsar client: %v", err)
@@ -110,3 +94,40 @@ func (c *PulsarClient) Reconnect() (pulsar.Client, error) {
 	c.Close()
 	return c.GetClient(c.pulsarURL, c.token)
 }
+
+// NewPulsarClient always creates a new pulsar.Client connection
+func NewPulsarClient(url, tokenStr string) (pulsar.Client, error) {
+	clientOpt := pulsar.ClientOptions{
+		URL:               url,
+		OperationTimeout:  time.Duration(clientOpsTimeout) * time.Second,
+		ConnectionTimeout: time.Duration(clientConnectTimeout) * time.Second,
+	}
+
+	if tokenStr != "" {
+		clientOpt.Authentication = pulsar.NewAuthenticationToken(tokenStr)
+	}
+
+	if strings.HasPrefix(url, "pulsar+ssl://") {
+		trustStore := util.GetConfig().TrustStore //"/etc/ssl/certs/ca-bundle.crt"
+		if trustStore == "" {
+			return nil, fmt.Errorf("this is fatal that we are missing trustStore while pulsar+ssl is required")
+		}
+		clientOpt.TLSTrustCertsFilePath = trustStore
+	}
+
+	// default is false for these two configuration parameters
+	clientOpt.TLSAllowInsecureConnection = util.StringToBool(util.GetConfig().PulsarTLSAllowInsecureConnection)
+	clientOpt.TLSValidateHostname = util.StringToBool(util.GetConfig().PulsarTLSValidateHostname)
+
+	driver, err := pulsar.NewClient(clientOpt)
+
+	if err != nil {
+		log.Errorf("failed instantiate pulsar client %v", err)
+		return nil, fmt.Errorf("Could not instantiate Pulsar client: %v", err)
+	}
+	if log.GetLevel() == log.DebugLevel {
+		log.Debugf("pulsar client url %s\n token %s", url, tokenStr)
+	}
+
+	return driver, nil
+}

src/pulsardriver/pulsar-client.go

@@ -369,3 +369,28 @@ func TestConcurrencyTTLCache(t *testing.T) {
 	assert(t, 14 == cache.Count(), "some objects have expired")
 	assert(t, !object1.isClosed, "object1 has not expired yet")
 }
+
+func TestStringToBoo(t *testing.T) {
+	// true cases
+	assert(t, StringToBool("true"), "string true yields boolean true")
+	assert(t, StringToBool("True"), "string True yields boolean true")
+	assert(t, StringToBool(" tRue"), "string tRue with space yields boolean true")
+	assert(t, StringToBool("yes"), "string true yields boolean true")
+	assert(t, StringToBool("1"), "string true yields boolean true")
+	assert(t, StringToBool("enable"), "string true yields boolean true")
+	assert(t, StringToBool(" Enabled "), "string Enabled with space yields boolean true")
+	assert(t, StringToBool("ok"), "string ok yields boolean true")
+	assert(t, StringToBool("Ok"), "string Ok yields boolean true")
+
+	// false cases
+	assert(t, !StringToBool(" "), "string space yields boolean false")
+	assert(t, !StringToBool(""), "string empty string yields boolean false")
+	assert(t, !StringToBool(" t rue"), "string t rue with space yields boolean false")
+	assert(t, !StringToBool("no"), "string no yields boolean false")
+	assert(t, !StringToBool("10"), "string 10 yields boolean false")
+	assert(t, !StringToBool("0"), "string 0 yields boolean false")
+	assert(t, !StringToBool("notok"), "string notok yields boolean false")
+	assert(t, !StringToBool("disable"), "string disable yields boolean false")
+	assert(t, !StringToBool("adsfasdf"), "string any string yields boolean false")
+
+}

src/unit-test/util_test.go

@@ -44,6 +44,14 @@ type Configuration struct {
 	SuperRoles       string `json:"SuperRoles"`
 	PulsarBrokerURL  string `json:"PulsarBrokerURL"`
 
+	// Configure whether the Pulsar client accept untrusted TLS certificate from broker (default: false)
+	// Set to `true` to enable
+	PulsarTLSAllowInsecureConnection string `json:"PulsarTLSAllowInsecureConnection"`
+
+	// Configure whether the Pulsar client verify the validity of the host name from broker (default: false)
+	// Set to `true` to enable
+	PulsarTLSValidateHostname string `json:"PulsarTLSValidateHostname"`
+
 	// Webhook consumers pool checked interval to stop deleted consumers and start new ones
 	// default value 180s
 	PbDbInterval string `json:"PbDbInterval"`

src/util/config.go

@@ -110,3 +110,16 @@ func GetEnvInt(env string, defaultNum int) int {
 	}
 	return defaultNum
 }
+
+// StringToBool format various strings to boolean
+// strconv.ParseBool only covers `true` and `false` cases
+func StringToBool(str string) bool {
+	s := strings.ToLower(strings.TrimSpace(str))
+
+	// `1` is true because the default Golang boolean is initialized as false
+	if s == "true" || s == "yes" || s == "enable" || s == "enabled" || s == "1" || s == "ok" {
+		return true
+	}
+
+	return false
+}