feat: adds support for TLS configuration options in exotic jetstream …

…EventBus settings (#3381)

Signed-off-by: fullykubed <[email protected]>

USERS.md

@@ -36,6 +36,7 @@ Organizations below are **officially** using Argo Events. Please send a PR with
 1. [MariaDB](https://mariadb.com/)
 1. [Mobimeo GmbH](https://mobimeo.com/en/home/)
 1. [OneCause](https://www.onecause.com/)
+1. [Panfactum](https://panfactum.com/)
 1. [Pinnacle Reliability](https://pinnaclereliability.com/)
 1. [PDOK](https://pdok.nl)
 1. [Phrase](https://www.phrase.com/)

api/jsonschema/schema.json

@@ -2355,6 +2355,10 @@
         "streamConfig": {
           "type": "string"
         },
+        "tls": {
+          "$ref": "#/definitions/io.argoproj.events.v1alpha1.TLSConfig",
+          "description": "SSL/TLS settings for the NATS client"
+        },
         "url": {
           "description": "JetStream (Nats) URL",
           "type": "string"

docs/APIs.md

@@ -11332,6 +11332,26 @@ Secret for auth
 
 </tr>
 
+<tr>
+
+<td>
+
+<code>tls</code></br> <em> <a href="#argoproj.io/v1alpha1.TLSConfig">
+TLSConfig </a> </em>
+</td>
+
+<td>
+
+<em>(Optional)</em>
+<p>
+
+SSL/TLS settings for the NATS client
+</p>
+
+</td>
+
+</tr>
+
 </tbody>
 
 </table>
@@ -19815,6 +19835,7 @@ TLSConfig
 <a href="#argoproj.io/v1alpha1.BitbucketServerEventSource">BitbucketServerEventSource</a>,
 <a href="#argoproj.io/v1alpha1.EmitterEventSource">EmitterEventSource</a>,
 <a href="#argoproj.io/v1alpha1.HTTPTrigger">HTTPTrigger</a>,
+<a href="#argoproj.io/v1alpha1.JetStreamConfig">JetStreamConfig</a>,
 <a href="#argoproj.io/v1alpha1.KafkaBus">KafkaBus</a>,
 <a href="#argoproj.io/v1alpha1.KafkaEventSource">KafkaEventSource</a>,
 <a href="#argoproj.io/v1alpha1.KafkaTrigger">KafkaTrigger</a>,

pkg/apis/events/v1alpha1/jetstream_eventbus.go

@@ -102,4 +102,7 @@ type JetStreamConfig struct {
 	AccessSecret *corev1.SecretKeySelector `json:"accessSecret,omitempty" protobuf:"bytes,2,opt,name=accessSecret"`
 	// +optional
 	StreamConfig string `json:"streamConfig,omitempty" protobuf:"bytes,3,opt,name=streamConfig"`
+	// SSL/TLS settings for the NATS client
+	// +optional
+	TLS *TLSConfig `json:"tls,omitempty" protobuf:"bytes,4,opt,name=tls"`
 }

pkg/eventbus/driver.go

@@ -52,7 +52,7 @@ func GetEventSourceDriver(ctx context.Context, eventBusConfig v1alpha1.BusConfig
 		}
 		dvr = stansource.NewSourceSTAN(eventBusConfig.NATS.URL, *eventBusConfig.NATS.ClusterID, eventSourceName, defaultSubject, auth, logger)
 	case v1alpha1.EventBusJetStream:
-		dvr, err = jetstreamsource.NewSourceJetstream(eventBusConfig.JetStream.URL, eventSourceName, eventBusConfig.JetStream.StreamConfig, auth, logger) // don't need to pass in subject because subjects will be derived from dependencies
+		dvr, err = jetstreamsource.NewSourceJetstream(eventBusConfig.JetStream.URL, eventSourceName, eventBusConfig.JetStream.StreamConfig, auth, logger, eventBusConfig.JetStream.TLS) // don't need to pass in subject because subjects will be derived from dependencies
 		if err != nil {
 			return nil, err
 		}
@@ -96,7 +96,7 @@ func GetSensorDriver(ctx context.Context, eventBusConfig v1alpha1.BusConfig, sen
 		dvr = stansensor.NewSensorSTAN(eventBusConfig.NATS.URL, *eventBusConfig.NATS.ClusterID, sensorSpec.Name, auth, logger)
 		return dvr, nil
 	case v1alpha1.EventBusJetStream:
-		dvr, err = jetstreamsensor.NewSensorJetstream(eventBusConfig.JetStream.URL, sensorSpec, eventBusConfig.JetStream.StreamConfig, auth, logger) // don't need to pass in subject because subjects will be derived from dependencies
+		dvr, err = jetstreamsensor.NewSensorJetstream(eventBusConfig.JetStream.URL, sensorSpec, eventBusConfig.JetStream.StreamConfig, auth, logger, eventBusConfig.JetStream.TLS) // don't need to pass in subject because subjects will be derived from dependencies
 		return dvr, err
 	case v1alpha1.EventBusKafka:
 		dvr = kafkasensor.NewKafkaSensor(eventBusConfig.Kafka, sensorSpec, hostname, logger)

pkg/eventbus/jetstream/base/jetstream.go

@@ -16,6 +16,7 @@ import (
 type Jetstream struct {
 	url  string
 	auth *eventbuscommon.Auth
+	tls  *v1alpha1.TLSConfig
 
 	MgmtConnection JetstreamConnection
 
@@ -24,10 +25,11 @@ type Jetstream struct {
 	Logger *zap.SugaredLogger
 }
 
-func NewJetstream(url string, streamSettings string, auth *eventbuscommon.Auth, logger *zap.SugaredLogger) (*Jetstream, error) {
+func NewJetstream(url string, streamSettings string, auth *eventbuscommon.Auth, logger *zap.SugaredLogger, tls *v1alpha1.TLSConfig) (*Jetstream, error) {
 	js := &Jetstream{
 		url:            url,
 		auth:           auth,
+		tls:            tls,
 		Logger:         logger,
 		streamSettings: streamSettings,
 	}
@@ -67,9 +69,19 @@ func (stream *Jetstream) MakeConnection() (*JetstreamConnection, error) {
 			conn.NATSConnected = true
 			log.Info("Reconnected to NATS server")
 		}),
-		nats.Secure(&tls.Config{
+	}
+
+	if stream.tls != nil {
+		tlsConfig, err := sharedutil.GetTLSConfig(stream.tls)
+		if err != nil {
+			return nil, err
+		}
+		opts = append(opts, nats.Secure(tlsConfig))
+		log.Info("Client-side TLS configuration enabled on the NATS connection")
+	} else {
+		opts = append(opts, nats.Secure(&tls.Config{
 			InsecureSkipVerify: true,
-		}),
+		}))
 	}
 
 	switch stream.auth.Strategy {

pkg/eventbus/jetstream/eventsource/source_jetstream.go

@@ -1,6 +1,7 @@
 package eventsource
 
 import (
+	"github.com/argoproj/argo-events/pkg/apis/events/v1alpha1"
 	eventbuscommon "github.com/argoproj/argo-events/pkg/eventbus/common"
 	jetstreambase "github.com/argoproj/argo-events/pkg/eventbus/jetstream/base"
 	"go.uber.org/zap"
@@ -11,8 +12,8 @@ type SourceJetstream struct {
 	eventSourceName string
 }
 
-func NewSourceJetstream(url, eventSourceName string, streamConfig string, auth *eventbuscommon.Auth, logger *zap.SugaredLogger) (*SourceJetstream, error) {
-	baseJetstream, err := jetstreambase.NewJetstream(url, streamConfig, auth, logger)
+func NewSourceJetstream(url, eventSourceName string, streamConfig string, auth *eventbuscommon.Auth, logger *zap.SugaredLogger, tls *v1alpha1.TLSConfig) (*SourceJetstream, error) {
+	baseJetstream, err := jetstreambase.NewJetstream(url, streamConfig, auth, logger, tls)
 	if err != nil {
 		return nil, err
 	}

pkg/eventbus/jetstream/eventsource/source_jetstream_test.go

@@ -20,7 +20,7 @@ func TestNewSourceJetstream(t *testing.T) {
 	logger := zap.NewExample().Sugar()
 
 	auth := &common.Auth{}
-	sourceJetstream, err := NewSourceJetstream(testURL, testEventSource, testStreamConfig, auth, logger)
+	sourceJetstream, err := NewSourceJetstream(testURL, testEventSource, testStreamConfig, auth, logger, nil)
 	assert.NotNil(t, sourceJetstream)
 	assert.Nil(t, err)
 }
@@ -29,7 +29,7 @@ func TestSourceJetstream_Connect(t *testing.T) {
 	logger := zap.NewExample().Sugar()
 
 	auth := &common.Auth{}
-	sourceJetstream, err := NewSourceJetstream(testURL, testEventSource, testStreamConfig, auth, logger)
+	sourceJetstream, err := NewSourceJetstream(testURL, testEventSource, testStreamConfig, auth, logger, nil)
 	assert.NotNil(t, sourceJetstream)
 	assert.Nil(t, err)
 
@@ -44,7 +44,7 @@ func TestSourceJetstream_Initialize_Failure(t *testing.T) {
 	auth := &common.Auth{
 		Strategy: v1alpha1.AuthStrategyNone,
 	}
-	sourceJetstream, err := NewSourceJetstream(testURL, testEventSource, testStreamConfig, auth, logger)
+	sourceJetstream, err := NewSourceJetstream(testURL, testEventSource, testStreamConfig, auth, logger, nil)
 	assert.NotNil(t, sourceJetstream)
 	assert.Nil(t, err)
 

pkg/eventbus/jetstream/sensor/sensor_jetstream.go

@@ -30,14 +30,14 @@ type SensorJetstream struct {
 	keyValueStore nats.KeyValue
 }
 
-func NewSensorJetstream(url string, sensorSpec *v1alpha1.Sensor, streamConfig string, auth *eventbuscommon.Auth, logger *zap.SugaredLogger) (*SensorJetstream, error) {
+func NewSensorJetstream(url string, sensorSpec *v1alpha1.Sensor, streamConfig string, auth *eventbuscommon.Auth, logger *zap.SugaredLogger, tls *v1alpha1.TLSConfig) (*SensorJetstream, error) {
 	if sensorSpec == nil {
 		errStr := SensorNilError
 		logger.Errorf(errStr)
 		return nil, fmt.Errorf(errStr)
 	}
 
-	baseJetstream, err := eventbusjetstreambase.NewJetstream(url, streamConfig, auth, logger)
+	baseJetstream, err := eventbusjetstreambase.NewJetstream(url, streamConfig, auth, logger, tls)
 	if err != nil {
 		return nil, err
 	}

pkg/reconciler/eventsource/resource.go

@@ -7,6 +7,10 @@ import (
 	"fmt"
 	"sort"
 
+	"github.com/argoproj/argo-events/pkg/apis/events/v1alpha1"
+	aev1 "github.com/argoproj/argo-events/pkg/apis/events/v1alpha1"
+	controllerscommon "github.com/argoproj/argo-events/pkg/reconciler/common"
+	sharedutil "github.com/argoproj/argo-events/pkg/shared/util"
 	"github.com/imdario/mergo"
 	"go.uber.org/zap"
 	appv1 "k8s.io/api/apps/v1"
@@ -17,11 +21,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/argoproj/argo-events/pkg/apis/events/v1alpha1"
-	aev1 "github.com/argoproj/argo-events/pkg/apis/events/v1alpha1"
-	controllerscommon "github.com/argoproj/argo-events/pkg/reconciler/common"
-	sharedutil "github.com/argoproj/argo-events/pkg/shared/util"
 )
 
 // AdaptorArgs are the args needed to create a sensor deployment
@@ -222,14 +221,29 @@ func buildDeployment(args *AdaptorArgs, eventBus *aev1.EventBus) (*appv1.Deploym
 
 	var secretObjs []interface{}
 	var accessSecret *corev1.SecretKeySelector
+	var caCertSecret *corev1.SecretKeySelector
+	var clientCertSecret *corev1.SecretKeySelector
+	var clientKeySecret *corev1.SecretKeySelector
 	switch {
 	case eventBus.Status.Config.NATS != nil:
+		caCertSecret = nil
+		clientCertSecret = nil
+		clientKeySecret = nil
 		accessSecret = eventBus.Status.Config.NATS.AccessSecret
 		secretObjs = []interface{}{eventSourceCopy}
 	case eventBus.Status.Config.JetStream != nil:
+		tlsOptions := eventBus.Status.Config.JetStream.TLS
+		if tlsOptions != nil {
+			caCertSecret = tlsOptions.CACertSecret
+			clientCertSecret = tlsOptions.ClientCertSecret
+			clientKeySecret = tlsOptions.ClientKeySecret
+		}
 		accessSecret = eventBus.Status.Config.JetStream.AccessSecret
 		secretObjs = []interface{}{eventSourceCopy}
 	case eventBus.Status.Config.Kafka != nil:
+		caCertSecret = nil
+		clientCertSecret = nil
+		clientKeySecret = nil
 		accessSecret = nil
 		secretObjs = []interface{}{eventSourceCopy, eventBus} // kafka requires secrets for sasl and tls
 	default:
@@ -259,6 +273,43 @@ func buildDeployment(args *AdaptorArgs, eventBus *aev1.EventBus) (*appv1.Deploym
 		})
 	}
 
+	uniqueCertVolumeMap := make(map[string][]corev1.KeyToPath)
+	for _, secret := range []*corev1.SecretKeySelector{caCertSecret, clientCertSecret, clientKeySecret} {
+		if secret != nil {
+			uniqueCertVolumeMap[secret.Name] = append(uniqueCertVolumeMap[secret.Name], corev1.KeyToPath{
+				Key:  secret.Key,
+				Path: secret.Key,
+			})
+		}
+	}
+
+	// We deduplicate the certificate secret mounts to ensure every secret under the TLS config is only mounted once
+	// because the secrets MUST be mounted at /argo-events/secrets/<secret-name>
+	// in order for util.GetTLSConfig to work without modification
+	for secretName, items := range uniqueCertVolumeMap {
+		// The names of volumes MUST be valid DNS_LABELs; as the secret names are user-supplied,
+		// we perform some input cleansing to ensure they conform
+		volumeName := sharedutil.ConvertToDNSLabel(secretName)
+
+		optional := false
+
+		volumes = append(volumes, corev1.Volume{
+			Name: volumeName,
+			VolumeSource: corev1.VolumeSource{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: secretName,
+					Items:      items,
+					Optional:   &optional,
+				},
+			},
+		})
+
+		volumeMounts = append(volumeMounts, corev1.VolumeMount{
+			Name:      volumeName,
+			MountPath: fmt.Sprintf("/argo-events/secrets/%s", secretName),
+		})
+	}
+
 	// secrets
 	volSecrets, volSecretMounts := sharedutil.VolumesFromSecretsOrConfigMaps(v1alpha1.SecretKeySelectorType, secretObjs...)
 	volumes = append(volumes, volSecrets...)