feat: amqp event source authentication (#1252)

* feat: amqp event source authentication Signed-off-by: Derek Wang <[email protected]> * example Signed-off-by: Derek Wang <[email protected]>
argoproj · Jun 22, 2021 · e92d105 · e92d105
1 parent d403c44
commit e92d105
Show file tree

Hide file tree

Showing 12 changed files with 508 additions and 368 deletions.
diff --git a/api/event-source.html b/api/event-source.html
diff --git a/api/event-source.md b/api/event-source.md
diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
diff --git a/eventsources/sources/amqp/start.go b/eventsources/sources/amqp/start.go
@@ -68,21 +68,35 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 	amqpEventSource := &el.AMQPEventSource
 	var conn *amqplib.Connection
 	if err := common.Connect(amqpEventSource.ConnectionBackoff, func() error {
+		c := amqplib.Config{
+			Heartbeat: 10 * time.Second,
+			Locale:    "en_US",
+		}
 		if amqpEventSource.TLS != nil {
 			tlsConfig, err := common.GetTLSConfig(amqpEventSource.TLS)
 			if err != nil {
 				return errors.Wrap(err, "failed to get the tls configuration")
 			}
-			conn, err = amqplib.DialTLS(amqpEventSource.URL, tlsConfig)
+			c.TLSClientConfig = tlsConfig
+		}
+		if amqpEventSource.Auth != nil {
+			username, err := common.GetSecretFromVolume(amqpEventSource.Auth.Username)
 			if err != nil {
-				return err
+				return errors.Wrap(err, "username not founnd")
 			}
-		} else {
-			var err error
-			conn, err = amqplib.Dial(amqpEventSource.URL)
+			password, err := common.GetSecretFromVolume(amqpEventSource.Auth.Password)
 			if err != nil {
-				return err
+				return errors.Wrap(err, "password not founnd")
 			}
+			c.SASL = []amqplib.Authentication{&amqplib.AMQPlainAuth{
+				Username: username,
+				Password: password,
+			}}
+		}
+		var err error
+		conn, err = amqplib.DialConfig(amqpEventSource.URL, c)
+		if err != nil {
+			return err
 		}
 		return nil
 	}); err != nil {

diff --git a/eventsources/sources/amqp/validate.go b/eventsources/sources/amqp/validate.go
@@ -49,5 +49,8 @@ func validate(eventSource *v1alpha1.AMQPEventSource) error {
 	if eventSource.TLS != nil {
 		return apicommon.ValidateTLSConfig(eventSource.TLS)
 	}
+	if eventSource.Auth != nil {
+		return apicommon.ValidateBasicAuth(eventSource.Auth)
+	}
 	return nil
 }
diff --git a/examples/event-sources/amqp.yaml b/examples/event-sources/amqp.yaml
@@ -54,6 +54,15 @@ spec:
         exclusive: false
         noLocal: false
         noWait: false
+      # username and password for authentication
+      # use secret selectors
+      auth:
+        username:
+          name: my-secret
+          key: username
+        password:
+          name: my-secret
+          key: password
 
 
 #    example-tls:

diff --git a/pkg/apis/common/validate.go b/pkg/apis/common/validate.go
@@ -1,6 +1,8 @@
 package common
 
-import "errors"
+import (
+	fmt "fmt"
+)
 
 // ValidateTLSConfig validates a TLS configuration.
 func ValidateTLSConfig(tlsConfig *TLSConfig) error {
@@ -22,11 +24,24 @@ func ValidateTLSConfig(tlsConfig *TLSConfig) error {
 	}
 
 	if !caCertSet && !clientCertSet && !clientKeySet {
-		return errors.New("invalid tls config, please configure either caCertSecret, or clientCertSecret and clientKeySecret, or both")
+		return fmt.Errorf("invalid tls config, please configure either caCertSecret, or clientCertSecret and clientKeySecret, or both")
 	}
 
 	if (clientCertSet || clientKeySet) && (!clientCertSet || !clientKeySet) {
-		return errors.New("invalid tls config, both clientCertSecret and clientKeySecret need to be configured")
+		return fmt.Errorf("invalid tls config, both clientCertSecret and clientKeySecret need to be configured")
+	}
+	return nil
+}
+
+func ValidateBasicAuth(auth *BasicAuth) error {
+	if auth == nil {
+		return nil
+	}
+	if auth.Username == nil {
+		return fmt.Errorf("username missing")
+	}
+	if auth.Password == nil {
+		return fmt.Errorf("password missing")
 	}
 	return nil
 }
@@ -39,12 +54,12 @@ func ValidateSASLConfig(saslConfig *SASLConfig) error {
 	switch saslConfig.Mechanism {
 	case "", "PLAIN", "OAUTHBEARER", "SCRAM-SHA-256", "SCRAM-SHA-512", "GSSAPI":
 	default:
-		return errors.New("invalid sasl config. Possible values for SASL Mechanism are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`")
+		return fmt.Errorf("invalid sasl config. Possible values for SASL Mechanism are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`")
 	}
 
 	// user and password must both be set
 	if saslConfig.UserSecret == nil || saslConfig.PasswordSecret == nil {
-		return errors.New("invalid sasl config, both userSecret and passwordSecret must be defined")
+		return fmt.Errorf("invalid sasl config, both userSecret and passwordSecret must be defined")
 	}
 
 	return nil