Pulsar PubSub: allow TLS configuration for hostname and insecure connections

pubsub/pulsar/metadata.go

@@ -20,21 +20,23 @@ import (
 )
 
 type pulsarMetadata struct {
-	Host                    string                    `mapstructure:"host"`
-	ConsumerID              string                    `mapstructure:"consumerID"`
-	EnableTLS               bool                      `mapstructure:"enableTLS"`
-	DisableBatching         bool                      `mapstructure:"disableBatching"`
-	BatchingMaxPublishDelay time.Duration             `mapstructure:"batchingMaxPublishDelay"`
-	BatchingMaxSize         uint                      `mapstructure:"batchingMaxSize"`
-	BatchingMaxMessages     uint                      `mapstructure:"batchingMaxMessages"`
-	Tenant                  string                    `mapstructure:"tenant"`
-	Namespace               string                    `mapstructure:"namespace"`
-	Persistent              bool                      `mapstructure:"persistent"`
-	RedeliveryDelay         time.Duration             `mapstructure:"redeliveryDelay"`
-	internalTopicSchemas    map[string]schemaMetadata `mapstructure:"-"`
-	PublicKey               string                    `mapstructure:"publicKey"`
-	PrivateKey              string                    `mapstructure:"privateKey"`
-	Keys                    string                    `mapstructure:"keys"`
+	Host                       string                    `mapstructure:"host"`
+	ConsumerID                 string                    `mapstructure:"consumerID"`
+	EnableTLS                  bool                      `mapstructure:"enableTLS"`
+	DisableBatching            bool                      `mapstructure:"disableBatching"`
+	BatchingMaxPublishDelay    time.Duration             `mapstructure:"batchingMaxPublishDelay"`
+	BatchingMaxSize            uint                      `mapstructure:"batchingMaxSize"`
+	BatchingMaxMessages        uint                      `mapstructure:"batchingMaxMessages"`
+	Tenant                     string                    `mapstructure:"tenant"`
+	Namespace                  string                    `mapstructure:"namespace"`
+	Persistent                 bool                      `mapstructure:"persistent"`
+	RedeliveryDelay            time.Duration             `mapstructure:"redeliveryDelay"`
+	internalTopicSchemas       map[string]schemaMetadata `mapstructure:"-"`
+	PublicKey                  string                    `mapstructure:"publicKey"`
+	PrivateKey                 string                    `mapstructure:"privateKey"`
+	Keys                       string                    `mapstructure:"keys"`
+	TLSAllowInsecureConnection bool                      `mapstructure:"tlsAllowInsecureConnection"`
+	TLSValidateHostname        bool                      `mapstructure:"tlsValidateHostname"`
 
 	Token                            string `mapstructure:"token"`
 	oauth2.ClientCredentialsMetadata `mapstructure:",squash"`

pubsub/pulsar/metadata.yaml

@@ -57,7 +57,7 @@ metadata:
     sensitive: true
     description: "Address of the Pulsar broker."
     example: |
-      "localhost:6650", "http://pulsar-pj54qwwdpz4b-pulsar.ap-sg.public.pulsar.com:8080"
+      "localhost:6650", "http://pulsar-pj54qwwdpz4b-pulsar.ap-sg.public.pulsar.com:8080", "pulsar+ssl://localhost:6651"
   - name: consumerID
     type: string
     description: "Used to set the subscription name or consumer ID."
@@ -74,6 +74,18 @@ metadata:
       Enable TLS.
     default: 'false'
     example: '"true", "false"'
+  - name: tlsAllowInsecureConnection
+    type: bool
+    description: |
+      Allow insecure TLS connections
+    default: 'false'
+    example: '"true", "false"'
+  - name: tlsValidateHostname
+    type: bool
+    description: |
+      Validate hostname when using TLS
+    default: 'false'
+    example: '"true", "false"'
   - name: tenant
     description: |
       The topic tenant within the instance. Tenants are essential to multi-tenancy in Pulsar, and spread

pubsub/pulsar/pulsar.go

@@ -38,29 +38,34 @@ import (
 )
 
 const (
-	host                    = "host"
-	consumerID              = "consumerID"
-	enableTLS               = "enableTLS"
-	deliverAt               = "deliverAt"
-	deliverAfter            = "deliverAfter"
-	disableBatching         = "disableBatching"
-	batchingMaxPublishDelay = "batchingMaxPublishDelay"
-	batchingMaxSize         = "batchingMaxSize"
-	batchingMaxMessages     = "batchingMaxMessages"
-	tenant                  = "tenant"
-	namespace               = "namespace"
-	persistent              = "persistent"
-	redeliveryDelay         = "redeliveryDelay"
-	avroProtocol            = "avro"
-	jsonProtocol            = "json"
-	protoProtocol           = "proto"
-	partitionKey            = "partitionKey"
-
-	defaultTenant     = "public"
-	defaultNamespace  = "default"
-	cachedNumProducer = 10
-	pulsarPrefix      = "pulsar://"
-	pulsarToken       = "token"
+	host                       = "host"
+	consumerID                 = "consumerID"
+	enableTLS                  = "enableTLS"
+	deliverAt                  = "deliverAt"
+	deliverAfter               = "deliverAfter"
+	disableBatching            = "disableBatching"
+	batchingMaxPublishDelay    = "batchingMaxPublishDelay"
+	batchingMaxSize            = "batchingMaxSize"
+	batchingMaxMessages        = "batchingMaxMessages"
+	tenant                     = "tenant"
+	namespace                  = "namespace"
+	persistent                 = "persistent"
+	redeliveryDelay            = "redeliveryDelay"
+	avroProtocol               = "avro"
+	jsonProtocol               = "json"
+	protoProtocol              = "proto"
+	partitionKey               = "partitionKey"
+	tlsAllowInsecureConnection = "tlsAllowInsecureConnection"
+	tlsValidateHostname        = "tlsValidateHostname"
+
+	defaultTenant                     = "public"
+	defaultNamespace                  = "default"
+	defaultTLSAllowInsecureConnection = false
+	defaultTLSValidateHostname        = false
+	cachedNumProducer                 = 10
+	pulsarPrefix                      = "pulsar://"
+	pulsarSSLPrefix                   = "pulsar+ssl://"
+	pulsarToken                       = "token"
 	// topicFormat is the format for pulsar, which have a well-defined structure: {persistent|non-persistent}://tenant/namespace/topic,
 	// see https://pulsar.apache.org/docs/en/concepts-messaging/#topics for details.
 	topicFormat                = "%s://%s/%s/%s"
@@ -113,15 +118,17 @@ func NewPulsar(l logger.Logger) pubsub.PubSub {
 
 func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
 	m := pulsarMetadata{
-		Persistent:              true,
-		Tenant:                  defaultTenant,
-		Namespace:               defaultNamespace,
-		internalTopicSchemas:    map[string]schemaMetadata{},
-		DisableBatching:         false,
-		BatchingMaxPublishDelay: defaultBatchingMaxPublishDelay,
-		BatchingMaxMessages:     defaultMaxMessages,
-		BatchingMaxSize:         defaultMaxBatchSize,
-		RedeliveryDelay:         defaultRedeliveryDelay,
+		Persistent:                 true,
+		Tenant:                     defaultTenant,
+		Namespace:                  defaultNamespace,
+		internalTopicSchemas:       map[string]schemaMetadata{},
+		DisableBatching:            false,
+		BatchingMaxPublishDelay:    defaultBatchingMaxPublishDelay,
+		BatchingMaxMessages:        defaultMaxMessages,
+		BatchingMaxSize:            defaultMaxBatchSize,
+		RedeliveryDelay:            defaultRedeliveryDelay,
+		TLSAllowInsecureConnection: defaultTLSAllowInsecureConnection,
+		TLSValidateHostname:        defaultTLSValidateHostname,
 	}
 
 	if err := kitmd.DecodeMetadata(meta.Properties, &m); err != nil {
@@ -132,6 +139,11 @@ func parsePulsarMetadata(meta pubsub.Metadata) (*pulsarMetadata, error) {
 		return nil, errors.New("pulsar error: missing pulsar host")
 	}
 
+	// If tls is disabled allow insecure connections
+	if !m.EnableTLS {
+		m.TLSAllowInsecureConnection = true
+	}
+
 	for k, v := range meta.Properties {
 		switch {
 		case strings.HasSuffix(k, topicJSONSchemaIdentifier):
@@ -163,16 +175,13 @@ func (p *Pulsar) Init(ctx context.Context, metadata pubsub.Metadata) error {
 	if err != nil {
 		return err
 	}
-	pulsarURL := m.Host
-	if !strings.HasPrefix(m.Host, "http://") &&
-		!strings.HasPrefix(m.Host, "https://") {
-		pulsarURL = fmt.Sprintf("%s%s", pulsarPrefix, m.Host)
-	}
+	pulsarURL := getPulsarURL(m)
 	options := pulsar.ClientOptions{
 		URL:                        pulsarURL,
 		OperationTimeout:           30 * time.Second,
 		ConnectionTimeout:          30 * time.Second,
-		TLSAllowInsecureConnection: !m.EnableTLS,
+		TLSAllowInsecureConnection: m.TLSAllowInsecureConnection,
+		TLSValidateHostname:        m.TLSValidateHostname,
 	}
 
 	switch {
@@ -538,3 +547,18 @@ func isValidPEM(val string) bool {
 	block, _ := pem.Decode([]byte(val))
 	return block != nil
 }
+
+func getPulsarURL(m *pulsarMetadata) string {
+	pulsarURL := m.Host
+	if !strings.HasPrefix(m.Host, "http://") &&
+		!strings.HasPrefix(m.Host, "https://") &&
+		!strings.HasPrefix(m.Host, "pulsar+ssl://") &&
+		!strings.HasPrefix(m.Host, "pulsar://") {
+		if m.EnableTLS {
+			pulsarURL = fmt.Sprintf("%s%s", pulsarSSLPrefix, m.Host)
+		} else {
+			pulsarURL = fmt.Sprintf("%s%s", pulsarPrefix, m.Host)
+		}
+	}
+	return pulsarURL
+}

pubsub/pulsar/pulsar_test.go

@@ -46,6 +46,8 @@ func TestParsePulsarMetadata(t *testing.T) {
 	assert.Equal(t, uint(100), meta.BatchingMaxSize)
 	assert.Equal(t, uint(200), meta.BatchingMaxMessages)
 	assert.Empty(t, meta.internalTopicSchemas)
+	assert.True(t, meta.TLSAllowInsecureConnection)
+	assert.False(t, meta.TLSValidateHostname)
 }
 
 func TestParsePulsarSchemaMetadata(t *testing.T) {