diff --git a/README.md b/README.md index d24e463..60bd4e9 100644 --- a/README.md +++ b/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 diff --git a/docs/reference-config.yaml b/docs/reference-config.yaml index 7be865a..c4c0f6c 100644 --- a/docs/reference-config.yaml +++ b/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: diff --git a/go.mod b/go.mod index 1f0a95a..e8a3f35 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index d25914b..17d3050 100644 --- a/go.sum +++ b/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= diff --git a/kafka/client_config_helper.go b/kafka/client_config_helper.go index ae671dd..76c9afd 100644 --- a/kafka/client_config_helper.go +++ b/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 diff --git a/kafka/config_sasl.go b/kafka/config_sasl.go index 73c82de..e37a60a 100644 --- a/kafka/config_sasl.go +++ b/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() diff --git a/kafka/config_sasl_aws.go b/kafka/config_sasl_aws.go new file mode 100644 index 0000000..bc94969 --- /dev/null +++ b/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"` +}