Add EXTERNAL SASL configuration (#209)

* Add EXTERNAL SASL configuration

Fixes #207

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

README.md

@@ -21,6 +21,7 @@ Go client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rabbitmq-serv
         * [Multi hosts](#multi-hosts)
         * [Load Balancer](#load-balancer)
         * [TLS](#tls)
+		* [Sasl Mechanisms](#sasl-mechanisms)
       * [Streams](#streams)
 		* [Statistics](#streams-statistics)
     * [Publish messages](#publish-messages)
@@ -175,6 +176,29 @@ env, err := stream.NewEnvironment(
 )
 ```
 
+### Sasl Mechanisms
+
+To configure SASL you need to set the `SaslMechanism` parameter `Environment.SetSaslConfiguration`:
+```golang
+cfg := new(tls.Config)
+cfg.ServerName = "my_server_name"
+cfg.RootCAs = x509.NewCertPool()
+
+if ca, err := os.ReadFile("certs/ca_certificate.pem"); err == nil {
+	cfg.RootCAs.AppendCertsFromPEM(ca)
+}
+
+if cert, err := tls.LoadX509KeyPair("certs/client/cert.pem", "certs/client/key.pem"); err == nil {
+cfg.Certificates = append(cfg.Certificates, cert)
+}
+
+env, err := stream.NewEnvironment(stream.NewEnvironmentOptions().
+	SetUri("rabbitmq-stream+tls://my_server_name:5551/").
+	IsTLS(true).
+	SetSaslConfiguration(stream.SaslConfigurationExternal). // SASL EXTERNAL
+	SetTLSConfig(cfg))
+```
+
 ### Streams
 
 To define streams you need to use the the `environment` interfaces `DeclareStream` and `DeleteStream`.

go.sum

@@ -11,6 +11,7 @@ github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMo
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/go-logr/logr v1.2.3 h1:2DntVwHkVopvECVRSlL5PSo9eG+cAkDCuckLubN+rq0=
 github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 h1:p104kn46Q8WdvHunIJ9dAyjPVtrBPhSr3KT2yUst43I=
 github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
@@ -32,6 +33,7 @@ github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8
 github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 h1:yAJXTCF9TqKcTiHJAE8dj7HMvPfh66eeA2JYW7eFpSE=
 github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -151,6 +153,7 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.10/go.mod h1:Uh6Zz+xoGYZom868N8YTex3t7RhtHDBrE8Gzo9bV56E=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.2.0 h1:G6AHpWxTMGY1KyEYoAQ5WTtIekUUvDNjan3ugu60JvE=
 golang.org/x/tools v0.2.0/go.mod h1:y4OqIKeOV/fWJetJ8bXPU1sEVniLMIyDAZWeHdV+NTA=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

pkg/amqp/decode.go

@@ -42,9 +42,9 @@ type unmarshaler interface {
 //
 // If i implements unmarshaler, i.unmarshal() will be called.
 //
-// Pointers to primitive types will be decoded via the appropriate read[Type] function.
+// Pointers to primitive types will be decoded via the appropriate read[type] function.
 //
-// If i is a pointer to a pointer (**Type), it will be dereferenced and a new instance
+// If it is a pointer to a pointer (**Type), it will be dereferenced and a new instance
 // of (*Type) is allocated via reflection.
 //
 // Common map types (map[string]string, map[Symbol]interface{}, and

pkg/stream/client.go

@@ -14,6 +14,22 @@ import (
 	"time"
 )
 
+// SaslConfiguration see
+//
+//	SaslConfigurationPlain    = "PLAIN"
+//	SaslConfigurationExternal = "EXTERNAL"
+//
+
+type SaslConfiguration struct {
+	Mechanism string
+}
+
+func newSaslConfigurationDefault() *SaslConfiguration {
+	return &SaslConfiguration{
+		Mechanism: SaslConfigurationPlain,
+	}
+}
+
 type TuneState struct {
 	requestedMaxFrameSize int
 	requestedHeartbeat    int
@@ -42,13 +58,14 @@ type Client struct {
 	coordinator          *Coordinator
 	broker               *Broker
 	tcpParameters        *TCPParameters
+	saslConfiguration    *SaslConfiguration
 
 	mutex            *sync.Mutex
 	metadataListener metadataListener
 	lastHeartBeat    HeartBeat
 }
 
-func newClient(connectionName string, broker *Broker, tcpParameters *TCPParameters) *Client {
+func newClient(connectionName string, broker *Broker, tcpParameters *TCPParameters, saslConfiguration *SaslConfiguration) *Client {
 	var clientBroker = broker
 	if broker == nil {
 		clientBroker = newBrokerDefault()
@@ -57,10 +74,15 @@ func newClient(connectionName string, broker *Broker, tcpParameters *TCPParamete
 		tcpParameters = newTCPParameterDefault()
 	}
 
+	if saslConfiguration == nil {
+		saslConfiguration = newSaslConfigurationDefault()
+	}
+
 	c := &Client{
 		coordinator:          NewCoordinator(),
 		broker:               clientBroker,
 		tcpParameters:        tcpParameters,
+		saslConfiguration:    saslConfiguration,
 		destructor:           &sync.Once{},
 		mutex:                &sync.Mutex{},
 		clientProperties:     ClientProperties{items: make(map[string]string)},
@@ -224,10 +246,15 @@ func (c *Client) authenticate(user string, password string) error {
 	}
 	saslMechanism := ""
 	for i := 0; i < len(saslMechanisms); i++ {
-		if saslMechanisms[i] == "PLAIN" {
-			saslMechanism = "PLAIN"
+		if saslMechanisms[i] == c.saslConfiguration.Mechanism {
+			saslMechanism = c.saslConfiguration.Mechanism
+			break
 		}
 	}
+	if saslMechanism == "" {
+		return fmt.Errorf("no matching mechanism found")
+	}
+
 	response := unicodeNull + user + unicodeNull + password
 	saslResponse := []byte(response)
 	return c.sendSaslAuthenticate(saslMechanism, saslResponse)

pkg/stream/constants.go

@@ -194,3 +194,8 @@ func lookUpCommand(command uint16) string {
 
 	return constLookup[command]
 }
+
+const (
+	SaslConfigurationPlain    = "PLAIN"
+	SaslConfigurationExternal = "EXTERNAL"
+)

pkg/stream/coordinator_test.go

@@ -14,7 +14,7 @@ var _ = Describe("Coordinator", func() {
 			client *Client
 		)
 		BeforeEach(func() {
-			client = newClient("test-client", nil, nil)
+			client = newClient("test-client", nil, nil, nil)
 
 		})
 		AfterEach(func() {
@@ -131,7 +131,7 @@ var _ = Describe("Coordinator", func() {
 			client *Client
 		)
 		BeforeEach(func() {
-			client = newClient("test-client", nil, nil)
+			client = newClient("test-client", nil, nil, nil)
 
 		})
 		AfterEach(func() {
@@ -185,7 +185,7 @@ var _ = Describe("Coordinator", func() {
 			client *Client
 		)
 		BeforeEach(func() {
-			client = newClient("test-client", nil, nil)
+			client = newClient("test-client", nil, nil, nil)
 
 		})
 		AfterEach(func() {

pkg/stream/environment.go

@@ -25,7 +25,7 @@ func NewEnvironment(options *EnvironmentOptions) (*Environment, error) {
 	if options == nil {
 		options = NewEnvironmentOptions()
 	}
-	client := newClient("go-stream-locator", nil, options.TCPParameters)
+	client := newClient("go-stream-locator", nil, options.TCPParameters, options.SaslConfiguration)
 	defer func(client *Client) {
 		err := client.Close()
 		if err != nil {
@@ -38,6 +38,12 @@ func NewEnvironment(options *EnvironmentOptions) (*Environment, error) {
 		return nil, fmt.Errorf(" MaxConsumersPerClient and MaxProducersPerClient must be between 1 and 254")
 	}
 
+	if options.SaslConfiguration != nil {
+		if options.SaslConfiguration.Mechanism != SaslConfigurationPlain && options.SaslConfiguration.Mechanism != SaslConfigurationExternal {
+			return nil, fmt.Errorf("SaslConfiguration mechanism must be PLAIN or EXTERNAL")
+		}
+	}
+
 	if len(options.ConnectionParameters) == 0 {
 		options.ConnectionParameters = []*Broker{newBrokerDefault()}
 	}
@@ -76,7 +82,7 @@ func NewEnvironment(options *EnvironmentOptions) (*Environment, error) {
 }
 func (env *Environment) newReconnectClient() (*Client, error) {
 	broker := env.options.ConnectionParameters[0]
-	client := newClient("go-stream-locator", broker, env.options.TCPParameters)
+	client := newClient("go-stream-locator", broker, env.options.TCPParameters, env.options.SaslConfiguration)
 
 	err := client.connect()
 	tentatives := 1
@@ -86,7 +92,8 @@ func (env *Environment) newReconnectClient() (*Client, error) {
 		time.Sleep(time.Duration(tentatives) * time.Second)
 		rand.Seed(time.Now().UnixNano())
 		n := rand.Intn(len(env.options.ConnectionParameters))
-		client = newClient("stream-locator", env.options.ConnectionParameters[n], env.options.TCPParameters)
+		client = newClient("stream-locator", env.options.ConnectionParameters[n], env.options.TCPParameters,
+			env.options.SaslConfiguration)
 		tentatives = tentatives + 1
 		err = client.connect()
 
@@ -261,6 +268,7 @@ func (env *Environment) IsClosed() bool {
 type EnvironmentOptions struct {
 	ConnectionParameters  []*Broker
 	TCPParameters         *TCPParameters
+	SaslConfiguration     *SaslConfiguration
 	MaxProducersPerClient int
 	MaxConsumersPerClient int
 	AddressResolver       *AddressResolver
@@ -272,6 +280,7 @@ func NewEnvironmentOptions() *EnvironmentOptions {
 		MaxConsumersPerClient: 1,
 		ConnectionParameters:  []*Broker{},
 		TCPParameters:         newTCPParameterDefault(),
+		SaslConfiguration:     newSaslConfigurationDefault(),
 	}
 }
 
@@ -328,6 +337,14 @@ func (envOptions *EnvironmentOptions) SetVHost(vhost string) *EnvironmentOptions
 	return envOptions
 }
 
+func (envOptions *EnvironmentOptions) SetSaslConfiguration(value string) *EnvironmentOptions {
+	if envOptions.SaslConfiguration == nil {
+		envOptions.SaslConfiguration = newSaslConfigurationDefault()
+	}
+	envOptions.SaslConfiguration.Mechanism = value
+	return envOptions
+}
+
 func (envOptions *EnvironmentOptions) SetTLSConfig(config *tls.Config) *EnvironmentOptions {
 	if envOptions.TCPParameters == nil {
 		envOptions.TCPParameters = newTCPParameterDefault()
@@ -506,7 +523,7 @@ func (c *Client) maybeCleanConsumers(streamName string) {
 	}
 }
 
-func (cc *environmentCoordinator) newProducer(leader *Broker, tcpParameters *TCPParameters, streamName string,
+func (cc *environmentCoordinator) newProducer(leader *Broker, tcpParameters *TCPParameters, saslConfiguration *SaslConfiguration, streamName string,
 	options *ProducerOptions) (*Producer, error) {
 	cc.mutex.Lock()
 	defer cc.mutex.Unlock()
@@ -521,7 +538,7 @@ func (cc *environmentCoordinator) newProducer(leader *Broker, tcpParameters *TCP
 	}
 
 	if clientResult == nil {
-		clientResult = cc.newClientForProducer(leader, tcpParameters)
+		clientResult = cc.newClientForProducer(leader, tcpParameters, saslConfiguration)
 	}
 
 	err := clientResult.connect()
@@ -538,7 +555,7 @@ func (cc *environmentCoordinator) newProducer(leader *Broker, tcpParameters *TCP
 		if err != nil {
 			return nil, err
 		}
-		clientResult = cc.newClientForProducer(leader, tcpParameters)
+		clientResult = cc.newClientForProducer(leader, tcpParameters, saslConfiguration)
 		err = clientResult.connect()
 		if err != nil {
 			return nil, err
@@ -555,8 +572,8 @@ func (cc *environmentCoordinator) newProducer(leader *Broker, tcpParameters *TCP
 	return producer, nil
 }
 
-func (cc *environmentCoordinator) newClientForProducer(leader *Broker, tcpParameters *TCPParameters) *Client {
-	clientResult := newClient("go-stream-producer", leader, tcpParameters)
+func (cc *environmentCoordinator) newClientForProducer(leader *Broker, tcpParameters *TCPParameters, saslConfiguration *SaslConfiguration) *Client {
+	clientResult := newClient("go-stream-producer", leader, tcpParameters, saslConfiguration)
 	chMeta := make(chan metaDataUpdateEvent, 1)
 	clientResult.metadataListener = chMeta
 	go func(ch <-chan metaDataUpdateEvent, cl *Client) {
@@ -575,7 +592,7 @@ func (cc *environmentCoordinator) newClientForProducer(leader *Broker, tcpParame
 	return clientResult
 }
 
-func (cc *environmentCoordinator) newConsumer(leader *Broker, tcpParameters *TCPParameters,
+func (cc *environmentCoordinator) newConsumer(leader *Broker, tcpParameters *TCPParameters, saslConfiguration *SaslConfiguration,
 	streamName string, messagesHandler MessagesHandler,
 	options *ConsumerOptions) (*Consumer, error) {
 	cc.mutex.Lock()
@@ -591,7 +608,7 @@ func (cc *environmentCoordinator) newConsumer(leader *Broker, tcpParameters *TCP
 	}
 
 	if clientResult == nil {
-		clientResult = newClient("go-stream-consumer", leader, tcpParameters)
+		clientResult = newClient("go-stream-consumer", leader, tcpParameters, saslConfiguration)
 		chMeta := make(chan metaDataUpdateEvent)
 		clientResult.metadataListener = chMeta
 		go func(ch <-chan metaDataUpdateEvent, cl *Client) {
@@ -675,8 +692,8 @@ func (ps *producersEnvironment) newProducer(clientLocator *Client, streamName st
 	}
 	leader.cloneFrom(clientLocator.broker, resolver)
 
-	producer, err := ps.producersCoordinator[coordinatorKey].newProducer(leader, clientLocator.tcpParameters, streamName,
-		options)
+	producer, err := ps.producersCoordinator[coordinatorKey].newProducer(leader, clientLocator.tcpParameters,
+		clientLocator.saslConfiguration, streamName, options)
 	if err != nil {
 		return nil, err
 	}
@@ -740,7 +757,7 @@ func (ps *consumersEnvironment) NewSubscriber(clientLocator *Client, streamName
 	}
 	consumerBroker.cloneFrom(clientLocator.broker, resolver)
 	consumer, err := ps.consumersCoordinator[coordinatorKey].
-		newConsumer(consumerBroker, clientLocator.tcpParameters, streamName, messagesHandler, consumerOptions)
+		newConsumer(consumerBroker, clientLocator.tcpParameters, clientLocator.saslConfiguration, streamName, messagesHandler, consumerOptions)
 	if err != nil {
 		return nil, err
 	}

pkg/stream/environment_test.go

@@ -343,6 +343,13 @@ var _ = Describe("Environment test", func() {
 		Expect(err).To(HaveOccurred())
 	})
 
+	It("Fail to set SetSaslConfiguration", func() {
+		_, err := NewEnvironment(NewEnvironmentOptions().
+			SetSaslConfiguration("IS_NOT_VALID").
+			IsTLS(true))
+		Expect(err).To(HaveOccurred())
+	})
+
 	It("Set TCP parameters", func() {
 		// weak test, atm I don't have other ways to test these values
 		// just validate that the connection still works