feat: support AWS MSK IAM authentication (AWS_MSK_IAM SASL mechanism)

README.md

@@ -7,7 +7,7 @@ KMinion.
 ## 🚀 Features
 
 - **Kafka versions:** Supports all Kafka versions v0.11+
-- **Supported SASL mechanisms:** plain, scram-sha-256/512, gssapi/kerberos
+- **Supported SASL mechanisms:** plain, scram-sha-256/512, gssapi/kerberos, oauthbearer, aws-msk-iam
 - **TLS support:** TLS is supported, regardless whether you need mTLS, a custom CA, encrypted keys or just the trusted
   root certs
 - **Consumer Group Lags:** Number of messages a consumer group is lagging behind the latest offset

docs/reference-config.yaml

@@ -46,7 +46,7 @@ kafka:
     username: ""
     # Password to use for PLAIN or SCRAM mechanism
     password: ""
-    # Mechanism to use for SASL Authentication. Valid values are PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI, OAUTHBEARER
+    # Mechanism to use for SASL Authentication. Valid values are PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI, OAUTHBEARER, AWS_MSK_IAM
     mechanism: "PLAIN"
     # GSSAPI / Kerberos config properties
     gssapi:
@@ -65,6 +65,12 @@ kafka:
       clientId: ""
       clientSecret: ""
       scope: ""
+    # AWS_MSK_IAM config properties
+    aws:
+      region: ""
+      roleArn: ""
+      externalId: ""
+      roleSessionName: ""
 
 minion:
   consumerGroups:

go.mod

@@ -3,6 +3,10 @@ module github.com/cloudhut/kminion/v2
 go 1.26.5
 
 require (
+	github.com/aws/aws-sdk-go-v2 v1.37.1
+	github.com/aws/aws-sdk-go-v2/config v1.30.2
+	github.com/aws/aws-sdk-go-v2/credentials v1.18.2
+	github.com/aws/aws-sdk-go-v2/service/sts v1.35.1
 	github.com/google/uuid v1.6.0
 	github.com/jcmturner/gokrb5/v8 v8.4.4
 	github.com/jellydator/ttlcache/v2 v2.11.1
@@ -22,6 +26,15 @@ require (
 )
 
 require (
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.14 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.26.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.31.1 // indirect
+	github.com/aws/smithy-go v1.27.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect

go.sum

@@ -12,15 +12,41 @@ github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmV
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/aws/aws-sdk-go-v2 v1.9.2/go.mod h1:cK/D0BBs0b/oWPIcX/Z/obahJK1TT7IPVjy53i/mX/4=
+github.com/aws/aws-sdk-go-v2 v1.37.1 h1:SMUxeNz3Z6nqGsXv0JuJXc8w5YMtrQMuIBmDx//bBDY=
+github.com/aws/aws-sdk-go-v2 v1.37.1/go.mod h1:9Q0OoGQoboYIAJyslFyF1f5K1Ryddop8gqMhWx/n4Wg=
 github.com/aws/aws-sdk-go-v2/config v1.8.3/go.mod h1:4AEiLtAb8kLs7vgw2ZV3p2VZ1+hBavOc84hqxVNpCyw=
+github.com/aws/aws-sdk-go-v2/config v1.30.2 h1:YE1BmSc4fFYqFgN1mN8uzrtc7R9x+7oSWeX8ckoltAw=
+github.com/aws/aws-sdk-go-v2/config v1.30.2/go.mod h1:UNrLGZ6jfAVjgVJpkIxjLufRJqTXCVYOpkeVf83kwBo=
 github.com/aws/aws-sdk-go-v2/credentials v1.4.3/go.mod h1:FNNC6nQZQUuyhq5aE5c7ata8o9e4ECGmS4lAXC7o1mQ=
+github.com/aws/aws-sdk-go-v2/credentials v1.18.2 h1:mfm0GKY/PHLhs7KO0sUaOtFnIQ15Qqxt+wXbO/5fIfs=
+github.com/aws/aws-sdk-go-v2/credentials v1.18.2/go.mod h1:v0SdJX6ayPeZFQxgXUKw5RhLpAoZUuynxWDfh8+Eknc=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.6.0/go.mod h1:gqlclDEZp4aqJOancXK6TN24aKhT0W0Ae9MHk3wzTMM=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.1 h1:owmNBboeA0kHKDcdF8KiSXmrIuXZustfMGGytv6OMkM=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.1/go.mod h1:Bg1miN59SGxrZqlP8vJZSmXW+1N8Y1MjQDq1OfuNod8=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1 h1:ksZXBYv80EFTcgc8OJO48aQ8XDWXIQL7gGasPeCoTzI=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.1/go.mod h1:HSksQyyJETVZS7uM54cir0IgxttTD+8aEoJMPGepHBI=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1 h1:+dn/xF/05utS7tUhjIcndbuaPjfll2LhbH1cCDGLYUQ=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.1/go.mod h1:hyAGz30LHdm5KBZDI58MXx5lDVZ5CUfvfTZvMu4HCZo=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.2.4/go.mod h1:ZcBrrI3zBKlhGFNYWvju0I3TR93I7YIgAfy82Fh4lcQ=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
 github.com/aws/aws-sdk-go-v2/service/appconfig v1.4.2/go.mod h1:FZ3HkCe+b10uFZZkFdvf98LHW21k49W8o8J366lqVKY=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.14 h1:SA43nfaY7+1jjMNIc2ywu99JLJLButtIdLP6j+bT870=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.14/go.mod h1:Du3llKcwbQvHsTXSLzTOGQz0DTDBMEzdg7DAGu7inrY=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.3.2/go.mod h1:72HRZDLMtmVQiLG2tLfQcaWLCssELvGl+Zf2WVxMmR8=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.1 h1:ky79ysLMxhwk5rxJtS+ILd3Mc8kC5fhsLBrP27r6h4I=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.1/go.mod h1:+2MmkvFvPYM1vsozBWduoLJUi5maxFk5B7KJFECujhY=
 github.com/aws/aws-sdk-go-v2/service/sso v1.4.2/go.mod h1:NBvT9R1MEF+Ud6ApJKM0G+IkPchKS7p7c2YPKwHmBOk=
+github.com/aws/aws-sdk-go-v2/service/sso v1.26.1 h1:uWaz3DoNK9MNhm7i6UGxqufwu3BEuJZm72WlpGwyVtY=
+github.com/aws/aws-sdk-go-v2/service/sso v1.26.1/go.mod h1:ILpVNjL0BO+Z3Mm0SbEeUoYS9e0eJWV1BxNppp0fcb8=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.31.1 h1:XdG6/o1/ZDmn3wJU5SRAejHaWgKS4zHv0jBamuKuS2k=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.31.1/go.mod h1:oiotGTKadCOCl3vg/tYh4k45JlDF81Ka8rdumNhEnIQ=
 github.com/aws/aws-sdk-go-v2/service/sts v1.7.2/go.mod h1:8EzeIqfWt2wWT4rJVu3f21TfrhJ8AEMzVybRNSb/b4g=
+github.com/aws/aws-sdk-go-v2/service/sts v1.35.1 h1:iF4Xxkc0H9c/K2dS0zZw3SCkj0Z7n6AMnUiiyoJND+I=
+github.com/aws/aws-sdk-go-v2/service/sts v1.35.1/go.mod h1:0bxIatfN0aLq4mjoLDeBpOjOke68OsFlXPDFJ7V0MYw=
 github.com/aws/smithy-go v1.8.0/go.mod h1:SObp3lf9smib00L/v3U2eAKG8FyQ7iLrJnQiAmR5n+E=
+github.com/aws/smithy-go v1.27.5 h1:d1ro7KpYOYwP6m73YFa+Kc/A130VsAdX68SpsJwARMM=
+github.com/aws/smithy-go v1.27.5/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=

kafka/client_config_helper.go

@@ -10,10 +10,15 @@ import (
 	"net"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
 	"github.com/jcmturner/gokrb5/v8/client"
 	"github.com/jcmturner/gokrb5/v8/keytab"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/sasl"
+	saslaws "github.com/twmb/franz-go/pkg/sasl/aws"
 	"github.com/twmb/franz-go/pkg/sasl/kerberos"
 	"github.com/twmb/franz-go/pkg/sasl/oauth"
 	"github.com/twmb/franz-go/pkg/sasl/plain"
@@ -127,6 +132,44 @@ func NewKgoConfig(cfg Config, logger *zap.Logger) ([]kgo.Opt, error) {
 			})
 			opts = append(opts, kgo.SASL(mechanism))
 		}
+
+		// AWS MSK IAM
+		if cfg.SASL.Mechanism == SASLMechanismAWSMSKIAM {
+			var loadOptions []func(*awsconfig.LoadOptions) error
+			if cfg.SASL.AWS.Region != "" {
+				loadOptions = append(loadOptions, awsconfig.WithRegion(cfg.SASL.AWS.Region))
+			}
+			awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), loadOptions...)
+			if err != nil {
+				return nil, fmt.Errorf("failed to load aws config: %w", err)
+			}
+
+			credentialsProvider := awsCfg.Credentials
+			if cfg.SASL.AWS.RoleARN != "" {
+				stsClient := sts.NewFromConfig(awsCfg)
+				credentialsProvider = aws.NewCredentialsCache(stscreds.NewAssumeRoleProvider(stsClient, cfg.SASL.AWS.RoleARN, func(options *stscreds.AssumeRoleOptions) {
+					if cfg.SASL.AWS.RoleSessionName != "" {
+						options.RoleSessionName = cfg.SASL.AWS.RoleSessionName
+					}
+					if cfg.SASL.AWS.ExternalID != "" {
+						options.ExternalID = aws.String(cfg.SASL.AWS.ExternalID)
+					}
+				}))
+			}
+
+			mechanism := saslaws.ManagedStreamingIAM(func(ctx context.Context) (saslaws.Auth, error) {
+				credentials, err := credentialsProvider.Retrieve(ctx)
+				if err != nil {
+					return saslaws.Auth{}, fmt.Errorf("failed to retrieve aws credentials: %w", err)
+				}
+				return saslaws.Auth{
+					AccessKey:    credentials.AccessKeyID,
+					SecretKey:    credentials.SecretAccessKey,
+					SessionToken: credentials.SessionToken,
+				}, nil
+			})
+			opts = append(opts, kgo.SASL(mechanism))
+		}
 	}
 
 	// Configure TLS

kafka/config_sasl.go

@@ -8,6 +8,7 @@ const (
 	SASLMechanismScramSHA512 = "SCRAM-SHA-512"
 	SASLMechanismGSSAPI      = "GSSAPI"
 	SASLMechanismOAuthBearer = "OAUTHBEARER"
+	SASLMechanismAWSMSKIAM   = "AWS_MSK_IAM"
 )
 
 // SASLConfig for Kafka Client
@@ -20,6 +21,7 @@ type SASLConfig struct {
 	// SASL Mechanisms that require more configuration than username & password
 	GSSAPI      SASLGSSAPIConfig  `koanf:"gssapi"`
 	OAuthBearer OAuthBearerConfig `koanf:"oauth"`
+	AWS         AWSConfig         `koanf:"aws"`
 }
 
 // SetDefaults for SASL Config
@@ -36,7 +38,7 @@ func (c *SASLConfig) Validate() error {
 	}
 
 	switch c.Mechanism {
-	case SASLMechanismPlain, SASLMechanismScramSHA256, SASLMechanismScramSHA512, SASLMechanismGSSAPI:
+	case SASLMechanismPlain, SASLMechanismScramSHA256, SASLMechanismScramSHA512, SASLMechanismGSSAPI, SASLMechanismAWSMSKIAM:
 		// Valid and supported
 	case SASLMechanismOAuthBearer:
 		return c.OAuthBearer.Validate()

kafka/config_sasl_aws.go

@@ -0,0 +1,8 @@
+package kafka
+
+type AWSConfig struct {
+	Region          string `koanf:"region"`
+	RoleARN         string `koanf:"roleArn"`
+	ExternalID      string `koanf:"externalId"`
+	RoleSessionName string `koanf:"roleSessionName"`
+}