feat: TLS authentication

api/v1alpha1/common.go

@@ -45,6 +45,9 @@ type PulsarAuthentication struct {
 
 	// +optional
 	OAuth2 *PulsarAuthenticationOAuth2 `json:"oauth2,omitempty"`
+
+	// +optional
+	TLS *PulsarAuthenticationTLS `json:"tls,omitempty"`
 }
 
 // PulsarResourceLifeCyclePolicy indicates whether it will keep or delete the resource
@@ -67,6 +70,12 @@ type PulsarAuthenticationOAuth2 struct {
 	Key            ValueOrSecretRef `json:"key"`
 }
 
+// PulsarAuthenticationTLS indicates the parameters which are need by pulsar TLS Authentication
+type PulsarAuthenticationTLS struct {
+	ClientCertificatePath    string `json:"clientCertificatePath"`
+	ClientCertificateKeyPath string `json:"clientCertificateKeyPath"`
+}
+
 // IsPulsarResourceReady returns true if resource satisfies with these condition
 // 1. The instance is not deleted
 // 2. Status ObservedGeneration is equal with meta.ObservedGeneration

api/v1alpha1/pulsarconnection_types.go

@@ -29,7 +29,7 @@ type PulsarConnectionSpec struct {
 	// AdminServiceURL is the admin service url of the pulsar cluster
 	// +optional
 	// +kubebuilder:validation:Pattern="^https?://.+$"
-	AdminServiceURL string `json:"adminServiceURL"`
+	AdminServiceURL string `json:"adminServiceURL,omitempty"`
 
 	// Authentication defines authentication configurations
 	// +optional
@@ -58,6 +58,19 @@ type PulsarConnectionSpec struct {
 	// set when enabling the Geo Replication
 	// +optional
 	ClusterName string `json:"clusterName,omitempty"`
+
+	// TLSEnableHostnameVerification indicates whether to verify the hostname of the broker.
+	// Only used when using secure urls.
+	// +optional
+	TLSEnableHostnameVerification bool `json:"tlsEnableHostnameVerification,omitempty"`
+
+	// TLSAllowInsecureConnection indicates whether to allow insecure connection to the broker.
+	// +optional
+	TLSAllowInsecureConnection bool `json:"tlsAllowInsecureConnection,omitempty"`
+
+	// TLSTrustCertsFilePath Path for the TLS certificate used to validate the broker endpoint when using TLS.
+	// +optional
+	TLSTrustCertsFilePath string `json:"tlsTrustCertsFilePath,omitempty"`
 }
 
 // PulsarConnectionStatus defines the observed state of PulsarConnection

charts/pulsar-resources-operator/crds/resource.streamnative.io_pulsarconnections.yaml

@@ -128,6 +128,18 @@ spec:
                     - issuerEndpoint
                     - key
                     type: object
+                  tls:
+                    description: PulsarAuthenticationTLS indicates the parameters
+                      which are need by pulsar TLS Authentication
+                    properties:
+                      clientCertificateKeyPath:
+                        type: string
+                      clientCertificatePath:
+                        type: string
+                    required:
+                    - clientCertificateKeyPath
+                    - clientCertificatePath
+                    type: object
                   token:
                     description: ValueOrSecretRef is a string or a secret reference
                       of the authentication
@@ -165,6 +177,18 @@ spec:
                 description: ClusterName indicates the local cluster name of the pulsar
                   cluster. It should set when enabling the Geo Replication
                 type: string
+              tlsAllowInsecureConnection:
+                description: TLSAllowInsecureConnection indicates whether to allow
+                  insecure connection to the broker.
+                type: boolean
+              tlsEnableHostnameVerification:
+                description: TLSEnableHostnameVerification indicates whether to verify
+                  the hostname of the broker. Only used when using secure urls.
+                type: boolean
+              tlsTrustCertsFilePath:
+                description: TLSTrustCertsFilePath Path for the TLS certificate used
+                  to validate the broker endpoint when using TLS.
+                type: string
             type: object
           status:
             description: PulsarConnectionStatus defines the observed state of PulsarConnection

charts/pulsar-resources-operator/templates/deployment.yaml

@@ -75,6 +75,14 @@ spec:
           periodSeconds: 20
         resources:
           {{- toYaml .Values.resources | nindent 10 }}
+        {{- if .Values.volumeMounts }}
+        volumeMounts:
+          {{- toYaml .Values.volumeMounts | nindent 10 }}
+        {{- end }}
+      {{- if .Values.volumes }}
+      volumes:
+        {{- toYaml .Values.volumes | nindent 8 }}
+      {{- end }}
       nodeSelector:
         {{- toYaml .Values.nodeSelector | nindent 8 }}
       affinity:

charts/pulsar-resources-operator/values.schema.json

@@ -333,6 +333,32 @@
             },
             "additionalProperties": true
         },
+        "volumes": {
+            "$id": "#/properties/volumes",
+            "type": "array",
+            "title": "The volumes schema",
+            "description": "An explanation about the purpose of this instance.",
+            "examples": [
+                []
+            ],
+            "additionalItems": true,
+            "items": {
+                "$id": "#/properties/volumes/items"
+            }
+        },
+        "volumeMounts": {
+            "$id": "#/properties/volumeMounts",
+            "type": "array",
+            "title": "The volumeMounts schema",
+            "description": "An explanation about the purpose of this instance.",
+            "examples": [
+                []
+            ],
+            "additionalItems": true,
+            "items": {
+                "$id": "#/properties/volumeMounts/items"
+            }
+        },
         "nodeSelector": {
             "$id": "#/properties/nodeSelector",
             "type": "object",

config/crd/bases/resource.streamnative.io_pulsarconnections.yaml

@@ -128,6 +128,18 @@ spec:
                     - issuerEndpoint
                     - key
                     type: object
+                  tls:
+                    description: PulsarAuthenticationTLS indicates the parameters
+                      which are need by pulsar TLS Authentication
+                    properties:
+                      clientCertificateKeyPath:
+                        type: string
+                      clientCertificatePath:
+                        type: string
+                    required:
+                    - clientCertificateKeyPath
+                    - clientCertificatePath
+                    type: object
                   token:
                     description: ValueOrSecretRef is a string or a secret reference
                       of the authentication
@@ -165,6 +177,18 @@ spec:
                 description: ClusterName indicates the local cluster name of the pulsar
                   cluster. It should set when enabling the Geo Replication
                 type: string
+              tlsAllowInsecureConnection:
+                description: TLSAllowInsecureConnection indicates whether to allow
+                  insecure connection to the broker.
+                type: boolean
+              tlsEnableHostnameVerification:
+                description: TLSEnableHostnameVerification indicates whether to verify
+                  the hostname of the broker. Only used when using secure urls.
+                type: boolean
+              tlsTrustCertsFilePath:
+                description: TLSTrustCertsFilePath Path for the TLS certificate used
+                  to validate the broker endpoint when using TLS.
+                type: string
             type: object
           status:
             description: PulsarConnectionStatus defines the observed state of PulsarConnection

docs/pulsar_connection.md

@@ -117,6 +117,24 @@ Other `PulsarConnection` configuration examples:
           # Use the keyFile contents as the oauth2 key value
           value: {"type":"sn_service_account","client_id":"zvex72oGvFQMBQGZ2ozMxOus2s4tQASJ","client_secret":"60J6fo81j-h69_vVvYvqFOHs2NfOyy6pqGqwIhTgnxpQ7O3UH8PdCbVtdm_SJjIf","client_email":"[email protected]","issuer_url":"https://auth.streamnative.cloud"}
 
+* TLS authentication
+
+  ```yaml
+  apiVersion: resource.streamnative.io/v1alpha1
+  kind: PulsarConnection
+  metadata:
+    name: test-tls-auth-pulsar-connection
+    namespace: test
+  spec:
+    adminServiceURL: http://test-pulsar-sn-platform-broker.test.svc.cluster.local:8080
+  brokerServiceURL: pulsar://test-pulsar-sn-platform-broker.test.svc.cluster.local:6650
+  clusterName: pulsar-cluster
+  authentication:
+    tls:
+      clientCertificateKeyPath: /certs/tls.key
+      clientCertificatePath: /certs/tls.crt
+  ```
+
 This table lists specifications available for the `PulsarConnection` resource.
 
 | Option | Description | Required or not |
@@ -127,7 +145,9 @@ This table lists specifications available for the `PulsarConnection` resource.
 | `brokerServiceSecureURL` | The broker service URL for secure connection to the Pulsar cluster, such as `pulsar+ssl://cluster-broker.test.svc.cluster.local:6651`. This option is required for configuring Geo-replication when TLS is enabled. This option is available for version `0.3.0` or above. | No |
 | `adminServiceSecureURL` | The admin service URL for secure connection to the Pulsar cluster, such as `https://cluster-broker.test.svc.cluster.local:443`. This option is available for version `0.3.0` or above. | No |
 | `clusterName` | The Pulsar cluster name. You can use the `pulsar-admin clusters list` command to get the Pulsar cluster name. This option is required for configuring Geo-replication. Provided from `0.3.0` | No |
-
+| `tlsAllowInsecureConnection` | A flag that indicates whether to allow insecure connection to the broker. Provided from `0.5.0` | No |
+| `tlsEnableHostnameVerification` | A flag that indicates wether hostname verification is enabled. Provided from `0.5.0` | No |
+| `tlsTrustCertsFilePath` | The path to the certificate used during hostname verfification. Provided from `0.5.0` | No |
 
 1. Apply the YAML file to create the Pulsar Connection. 
 

pkg/admin/interface.go

@@ -15,6 +15,7 @@
 package admin
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 
@@ -177,6 +178,10 @@ type PulsarAdminConfig struct {
 	ClientID       string
 	Audience       string
 	Key            string
+
+	// TLS Authentication related configuration
+	ClientCertificatePath    string
+	ClientCertificateKeyPath string
 }
 
 // NewPulsarAdmin initialize a pulsar admin client with configuration
@@ -187,8 +192,10 @@ func NewPulsarAdmin(conf PulsarAdminConfig) (PulsarAdmin, error) {
 	var adminClient admin.Client
 
 	config := &config.Config{
-		WebServiceURL:              conf.WebServiceURL,
-		TLSAllowInsecureConnection: true,
+		WebServiceURL:                 conf.WebServiceURL,
+		TLSAllowInsecureConnection:    conf.TLSAllowInsecureConnection,
+		TLSEnableHostnameVerification: conf.TLSEnableHostnameVerification,
+		TLSTrustCertsFilePath:         conf.TLSTrustCertsFilePath,
 		// V2 admin endpoint contains operations for tenant, namespace and topic.
 		PulsarAPIVersion: config.V2,
 	}
@@ -218,9 +225,22 @@ func NewPulsarAdmin(conf PulsarAdminConfig) (PulsarAdmin, error) {
 			return nil, err
 		}
 		adminClient = admin.NewWithAuthProvider(config, oauthProvider)
-	} else {
+	} else if conf.Token != "" {
 		config.Token = conf.Token
 
+		adminClient, err = admin.New(config)
+		if err != nil {
+			return nil, err
+		}
+	} else if conf.ClientCertificatePath != "" {
+		config.AuthPlugin = auth.TLSPluginName
+		config.AuthParams = fmt.Sprintf("{\"tlsCertFile\": %q, \"tlsKeyFile\": %q}", conf.ClientCertificatePath, conf.ClientCertificateKeyPath)
+
+		adminClient, err = admin.New(config)
+		if err != nil {
+			return nil, err
+		}
+	} else {
 		adminClient, err = admin.New(config)
 		if err != nil {
 			return nil, err

pkg/connection/reconciler.go

@@ -192,8 +192,21 @@ func (r *PulsarConnectionReconciler) MakePulsarAdminConfig(ctx context.Context)
 	if r.connection.Spec.AdminServiceURL == "" && r.connection.Spec.AdminServiceSecureURL == "" {
 		return nil, fmt.Errorf("adminServiceURL or adminServiceSecureURL must not be empty")
 	}
+
+	tlsEnableHostnameVerification := r.connection.Spec.TLSEnableHostnameVerification
+	tlsAllowInsecureConnection := r.connection.Spec.TLSAllowInsecureConnection
+	tlsTrustCertsFilePath := r.connection.Spec.TLSTrustCertsFilePath
+
+	if r.connection.Spec.AdminServiceSecureURL == "" {
+		tlsEnableHostnameVerification = false
+		tlsAllowInsecureConnection = true
+		tlsTrustCertsFilePath = ""
+	}
 	cfg := admin.PulsarAdminConfig{
-		WebServiceURL: r.connection.Spec.AdminServiceURL,
+		WebServiceURL:                 r.connection.Spec.AdminServiceURL,
+		TLSAllowInsecureConnection:    tlsAllowInsecureConnection,
+		TLSEnableHostnameVerification: tlsEnableHostnameVerification,
+		TLSTrustCertsFilePath:         tlsTrustCertsFilePath,
 	}
 	hasAuth := false
 	if authn := r.connection.Spec.Authentication; authn != nil {
@@ -219,6 +232,10 @@ func (r *PulsarConnectionReconciler) MakePulsarAdminConfig(ctx context.Context)
 				cfg.Key = *value
 			}
 		}
+		if tls := authn.TLS; tls != nil {
+			cfg.ClientCertificatePath = tls.ClientCertificatePath
+			cfg.ClientCertificateKeyPath = tls.ClientCertificateKeyPath
+		}
 	}
 	return &cfg, nil
 }