SASL Authentication Fails with AWS MSK using SASL_AWS_IAM Algorithm

[rsi2m]

Hi, team! First off, thanks for building and maintaining xk6-kafka! 🙏

I'm encountering an issue while using xk6-kafka inside Docker container to connect to AWS MSK and consume messages. The problem arises during the SASL authentication process when the SASL_AWS_IAM algorithm is used. The error I encounter is:

time="2025-01-17T12:28:03Z" level=error msg="Failed to create dialer., OriginalError: could not successfully authenticate to <redacted>:9098 with SASL: [58] SASL Authentication Failed: SASL Authentication failed" error="Failed to create dialer., OriginalError: could not successfully authenticate to <redacted>:9098 with SASL: [58] SASL Authentication Failed: SASL Authentication failed"
time="2025-01-17T12:28:03Z" level=error msg="GoError: Failed to create dialer., OriginalError: could not successfully authenticate to <redacted>:9098 with SASL: [58] SASL Authentication Failed: SASL Authentication failed\n\tat file:///k6/tests/msk.js:16:20(43)\n" hint="script exception"

Here's how my script looks:

import {
  SASL_AWS_IAM, TLS_1_2,
  Reader,
} from "k6/x/kafka";

const brokers = [""]

const saslConfig = {
  algorithm: SASL_AWS_IAM,
}

const tlsConfig = {
  enableTls: true, 
  insecureSkipTlsVerify: true, 
  minVersion: TLS_1_2,
}

const reader = new Reader({
  brokers: brokers,
  topic: "events",
  partition: 0,
  offset: 0,
  sasl: saslConfig,
  tls: tlsConfig,
})

export default function () {
  let messages = reader.consume({ limit: 10 })
  check(messages, {
    "10 messages returned": (msgs) => msgs.length === 10
  })
}

Additional Information:

• The error originates from the saslAuthenticate method in kafka-go:

for completed := false; !completed; {
		challenge, err := conn.saslAuthenticate(state)
		switch {
		case err == nil:
		case errors.Is(err, io.EOF):
			// the broker may communicate a failed exchange by closing the
			// connection (esp. in the case where we're passing opaque sasl
			// data over the wire since there's no protocol info).
			return SASLAuthenticationFailed
		default:
			return err
		}

		completed, state, err = sess.Next(ctx, challenge)
		if err != nil {
			return fmt.Errorf("SASL authentication process has failed: %w", err)
		}
	} 

• TLS appears to function correctly, as there are no handshake errors.
• The issue I'm facing looks simillar to the one reported in kafka-go. Author managed to fix it eventually, but solution is based on (old) aws-sdk-go and not applicable anymore, because aws-sdk-go-v2 is used everywhere now.

Any help or guidance on how to resolve this would be greatly appreciated. Let me know if you need more details about my setup or environment. Thanks for the great work on xk6-kafka!

[mostafa]

Hey @rsi2m,

Have you read my article about load testing AWS MSK? Have you set the env-vars for authentication with AWS on your machine?

[rsi2m]

Thanks for a quick response, @mostafa!
Yeah, I've read your guide about, in fact the config is taken from there as well. Sadly, I don't have the following properties (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN) exposed neither via env-var nor in credentials file.
I've tried using github.com/aws/aws-sdk-go-v2/credentials/ec2rolecreds to pull credentials in runtime and expose them as a env-vars ( in auth.go) :

tempCreds, err2 := ec2rolecreds.Retrieve(saslContext)
_ = os.Setenv("AWS_ACCESS_KEY_ID", tempCreds.AccessKeyID)
_ = os.Setenv("AWS_SECRET_ACCESS_KEY", tempCreds.SecretAccessKey)
_ = os.Setenv("AWS_SESSION_TOKEN", tempCreds.SessionToken)

While actual credentials were pulled correctly, SASL auth still failed.

[mostafa]

@rsi2m

I recommend configuring the AWS CLI with the given credentials, and then running the test to see if it works. Just export the env-vars and run aws configure. Have you tried connecting the logger?