SASL: Support AWS_MSK_IAM mechanism

[jgiles]

Describe the solution you'd like

• Add a package with a sasl.Mechanism implementation supporting SASL mechanism AWS_MSK_IAM for IAM-based authentication to AWS MSK managed Kafka.
• Alternatively (if there are concerns about including AWS-specific code in this repo), establish and document a mechanism for registering + using externally-implemented SASL mechanisms (probably this is just a documentation update pointing at sasl.Mechanism and then listing known OSS impls)

Supporting documentation

• MSK is a popular managed Kafka solution on AWS
• AWS announces IAM auth support: https://aws.amazon.com/about-aws/whats-new/2021/05/introducing-iam-access-control-amazon-msk/
• Reference implementation in Java: https://github.com/aws/aws-msk-iam-auth
• Go library for underlying sigv4 signing: https://pkg.go.dev/github.com/aws/aws-sdk-go@v1.38.37/aws/signer/v4
• Related issues
  • More auth mechanisms support #539
  • SASL: add support for OAUTHBEARER mechanism #238
  • SASL: add support for GSSAPI mechanism #237

[twmb]

For potential implementation reference, I've added support for AWS_MSK_IAM in my franz-go client.

[achille-roussel]

Hello @jgiles, thanks for opening the discussion.

We've historically refrained from bringing in more SASL mechanisms in kafka-go due to the external dependencies that they tend to require, and the often unfortunate consequences of forcing external dependencies on programs that depend on kafka-go (too many times we have had builds broken by external dependencies disappearing or not respecting versioning rules, etc...).

We do recognize that there is a need to integrate kafka-go with these SASL mechanisms, so we are thinking of introducing them as a separate modules within the repository, probably with a structure like this:

sasl
├── msk
│   ├── go.mod
│   ├── go.sum
│   └── msk.go
├── plain
│   └── plain.go
├── scram
│   └── scram.go
├── sasl.go
└── sasl_test.go

The sasl/msk module would depend on kafka-go and other dependencies (like the AWS SDK), but it won't force the external dependencies on clients that don't import it.

Let me know what you think about this.

@twmb thanks for sharing the work you did, this will very likely be a useful reference to support adding AWS_MSK_IAM to kafka-go 👍

[jgiles]

@achille-roussel I like the plan to use Go modules within this repo for supported SASL "plugins" with dependencies that should be isolated from the rest of the repo. I have seen this strategy used effectively in other open-source Go projects with similar needs.

[achille-roussel]

Great!

@jgiles would you be able to submit a pull request to add the AWS_MSK_IAM SASL mechanism then?

[jgiles]

I or someone on my team might be able to, but the timing would be uncertain based on some ongoing internal design+prioritization discussions.

@achille-roussel we will post on this issue if we pick this up for development (or establish a clear timeline for doing so). For now, a couple questions:

• Your designation as "Assignee" here does not indicate an intent to implement, just triage, correct?
• If we picked this up, a module structured as outlined above and implemented mirroring the referenced Java + Go implementations would match your expectations for review?

Just to check that we wouldn't be duplicating work or heading off on the wrong direction :-)

[achille-roussel]

Your designation as "Assignee" here does not indicate an intent to implement, just triage, correct?

This is correct, we use assignees to state who is the point of contact in the maintainers team.

If we picked this up, a module structured as outlined above and implemented mirroring the referenced Java + Go implementations would match your expectations for review?

I believe the could would have to work with the existing SASL interfaces we have in kafka-go https://github.com/segmentio/kafka-go/blob/master/sasl/sasl.go#L1-L44

No other concerns otherwise. I'll be happy to discuss further if there are other design decisions you'd want feedback on.

[cmaher]

I've started on an implementation of this, and, in testing, I've run into another issue that does not seem to be tied to the SASL mechanism. It's deep in the internals of the sasl handshake, before the actual sasl mechanism is called, and I was wondering if the maintainers would have some insights.

Using this example (with real values, of course), and this sasl mechanism, I'm hitting and error in the first part of the sasl handshake. When ApiVersions() is called, only 7 bytes ([]uint8{21, 3, 3, 0, 2, 2, 80}) are received in response, and these fail to parse in peekResponseSizeAndID. I'm running the java version in comparison, and even when I use the same correlation ID and client id (resulting, in the same byte-for-byte request, the response is still the incorrect 7 bytes. I was wondering if you would know where to start digging into what could be causing this result, or if I have overlooked something simple.

Edit: The above issue was due to my failure to configure TLS. The error was coming back as a simple EOF, which wasn't very helpful, but that's a separate issue

[dominicbarnes]

This should be supported as of v0.4.25!

[andrew-yangy]

I ran into a connection issue when I was trying to implement this new msk iam feature, took me hours to figure it out so put my solution in case helpful:

	dialer := &kafka.Dialer{
		SASLMechanism: &aws_msk_iam.Mechanism{
			Signer: sigv4.NewSigner(creds),
			Region: "AWS_REGION",
		},
		TLS: &tls.Config{}, 
	}

TLS: &tls.Config{} is required

[thadamski]

@ddvkid Are you by chance using the EC2RoleProvider for auth w/ MSK? I've been able to get your code example working (thanks for the TLS tip!) only when using EnvProvider w/ env. variables, but not with EC2RoleProvider.

I've tested the role itself w/ ec2Metadata using a test kafka-cli inside a pod/container in our EKS cluster, so I believe the role and eks serviceAccount are working.

Error:

{"level":"error","error":"[58] SASL Authentication Failed: SASL Authentication failed","time":"2022-01-28T14:01:38Z","caller":"/go/src/github.com/my-org/my-project/kafka_test.go:51","message":"error while reading batch"}

Example Code I'm using:

package main

import (
	"context"
	"crypto/tls"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/credentials/ec2rolecreds"
	"github.com/aws/aws-sdk-go/aws/ec2metadata"
	"github.com/aws/aws-sdk-go/aws/session"
	sigv4 "github.com/aws/aws-sdk-go/aws/signer/v4"
	"github.com/rs/zerolog/log"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/aws_msk_iam"
	"os"
	"strings"
	"time"
)

func main() {
	sess, err := session.NewSession()
	if err != nil {
		log.Logger.Fatal().Err(err)
	}
	awsCredentials := credentials.NewChainCredentials(
		[]credentials.Provider{
			&credentials.EnvProvider{},
			&ec2rolecreds.EC2RoleProvider{
				Client: ec2metadata.New(sess),
			},
		})

	kafkaDialer := &kafka.Dialer{
		Timeout:   10 * time.Second,
		DualStack: true,
		SASLMechanism: &aws_msk_iam.Mechanism{
			Signer: sigv4.NewSigner(awsCredentials),
			Region: os.Getenv("AWS_DEFAULT_REGION"),
		},
		TLS: &tls.Config{},
	}

	brokers := strings.Split(os.Getenv("KAFKA_SERVERS"), ",")
	topic := "quickstart-events"
	batchSize := int(10e6) // 10MB
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   brokers,
		Topic:     topic,
		Dialer:    kafkaDialer,
		Partition: 0,
		MinBytes:  batchSize,
		MaxBytes:  batchSize,
	})

	for {
		m, err := r.ReadMessage(context.Background())
		if err != nil {
			log.Logger.Err(err).Msg("error while reading batch") // <-- Error log from above
			break
		}
		log.Logger.Info().Msgf("Got Kafka Event %q", string(m.Value))
	}
}

[thadamski]

I created a separate issue in the event this is a bug and wrong usage of the sdk: #832