diff --git a/api/v2/model.go b/api/v2/model.go index 0575a73e23..50f916ab79 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -458,6 +458,15 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( Token: c.Sink.KafkaConfig.GlueSchemaRegistryConfig.Token, } } + var awsMSKIAMConfig *config.AWSMSKIAMConfig + if c.Sink.KafkaConfig.AWSMSKIAM != nil { + awsMSKIAMConfig = &config.AWSMSKIAMConfig{ + Region: c.Sink.KafkaConfig.AWSMSKIAM.Region, + RoleARN: c.Sink.KafkaConfig.AWSMSKIAM.RoleARN, + RoleSessionName: c.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName, + ExternalID: c.Sink.KafkaConfig.AWSMSKIAM.ExternalID, + } + } kafkaConfig = &config.KafkaConfig{ PartitionNum: c.Sink.KafkaConfig.PartitionNum, @@ -488,6 +497,8 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( SASLOAuthScopes: c.Sink.KafkaConfig.SASLOAuthScopes, SASLOAuthGrantType: c.Sink.KafkaConfig.SASLOAuthGrantType, SASLOAuthAudience: c.Sink.KafkaConfig.SASLOAuthAudience, + SASLOAuthProvider: c.Sink.KafkaConfig.SASLOAuthProvider, + AWSMSKIAM: awsMSKIAMConfig, EnableTLS: c.Sink.KafkaConfig.EnableTLS, CA: c.Sink.KafkaConfig.CA, Cert: c.Sink.KafkaConfig.Cert, @@ -789,6 +800,15 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { Token: cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig.Token, } } + var awsMSKIAMConfig *AWSMSKIAMConfig + if cloned.Sink.KafkaConfig.AWSMSKIAM != nil { + awsMSKIAMConfig = &AWSMSKIAMConfig{ + Region: cloned.Sink.KafkaConfig.AWSMSKIAM.Region, + RoleARN: cloned.Sink.KafkaConfig.AWSMSKIAM.RoleARN, + RoleSessionName: cloned.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName, + ExternalID: cloned.Sink.KafkaConfig.AWSMSKIAM.ExternalID, + } + } kafkaConfig = &KafkaConfig{ PartitionNum: cloned.Sink.KafkaConfig.PartitionNum, @@ -819,6 +839,8 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { SASLOAuthScopes: cloned.Sink.KafkaConfig.SASLOAuthScopes, SASLOAuthGrantType: cloned.Sink.KafkaConfig.SASLOAuthGrantType, SASLOAuthAudience: cloned.Sink.KafkaConfig.SASLOAuthAudience, + SASLOAuthProvider: cloned.Sink.KafkaConfig.SASLOAuthProvider, + AWSMSKIAM: awsMSKIAMConfig, EnableTLS: cloned.Sink.KafkaConfig.EnableTLS, CA: cloned.Sink.KafkaConfig.CA, Cert: cloned.Sink.KafkaConfig.Cert, @@ -1506,6 +1528,8 @@ type KafkaConfig struct { SASLOAuthScopes []string `json:"sasl_oauth_scopes,omitempty" toml:"sasl-oauth-scopes,omitempty"` SASLOAuthGrantType *string `json:"sasl_oauth_grant_type,omitempty" toml:"sasl-oauth-grant-type,omitempty"` SASLOAuthAudience *string `json:"sasl_oauth_audience,omitempty" toml:"sasl-oauth-audience,omitempty"` + SASLOAuthProvider *string `json:"sasl_oauth_provider,omitempty" toml:"sasl-oauth-provider,omitempty"` + AWSMSKIAM *AWSMSKIAMConfig `json:"aws_msk_iam,omitempty" toml:"aws-msk-iam,omitempty"` EnableTLS *bool `json:"enable_tls,omitempty" toml:"enable-tls,omitempty"` CA *string `json:"ca,omitempty" toml:"ca,omitempty"` Cert *string `json:"cert,omitempty" toml:"cert,omitempty"` @@ -1573,6 +1597,14 @@ type GlueSchemaRegistryConfig struct { Token string `json:"token,omitempty" toml:"token,omitempty"` } +// AWSMSKIAMConfig represents an Amazon MSK IAM authentication configuration. +type AWSMSKIAMConfig struct { + Region string `json:"region" toml:"region"` + RoleARN string `json:"role_arn,omitempty" toml:"role-arn,omitempty"` + RoleSessionName string `json:"role_session_name,omitempty" toml:"role-session-name,omitempty"` + ExternalID string `json:"external_id,omitempty" toml:"external-id,omitempty"` +} + // OpenProtocolConfig represents the configurations for open protocol encoding type OpenProtocolConfig struct { OutputOldValue bool `json:"output_old_value" toml:"output-old-value"` diff --git a/api/v2/model_test.go b/api/v2/model_test.go index 88fb821cdb..ffd5faeef0 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -41,6 +41,16 @@ func TestReplicaConfigConversion(t *testing.T) { SpoolDiskQuota: util.AddressOf(int64(1024)), SpoolBaseDir: util.AddressOf("/tmp/ticdc-spool"), }, + KafkaConfig: &KafkaConfig{ + SASLMechanism: util.AddressOf("OAUTHBEARER"), + SASLOAuthProvider: util.AddressOf("AWS_MSK_IAM"), + AWSMSKIAM: &AWSMSKIAMConfig{ + Region: "us-west-2", + RoleARN: "arn:aws:iam::123456789012:role/ticdc-msk", + RoleSessionName: "ticdc", + ExternalID: "external-id", + }, + }, }, Mounter: &MounterConfig{ WorkerNum: util.AddressOf(16), @@ -73,6 +83,11 @@ func TestReplicaConfigConversion(t *testing.T) { require.True(t, util.GetOrZero(internalCfg.Sink.CloudStorageConfig.UseTableIDAsPath)) require.Equal(t, int64(1024), util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolDiskQuota)) require.Equal(t, "/tmp/ticdc-spool", util.GetOrZero(internalCfg.Sink.CloudStorageConfig.SpoolBaseDir)) + require.Equal(t, "AWS_MSK_IAM", util.GetOrZero(internalCfg.Sink.KafkaConfig.SASLOAuthProvider)) + require.Equal(t, "us-west-2", internalCfg.Sink.KafkaConfig.AWSMSKIAM.Region) + require.Equal(t, "arn:aws:iam::123456789012:role/ticdc-msk", internalCfg.Sink.KafkaConfig.AWSMSKIAM.RoleARN) + require.Equal(t, "ticdc", internalCfg.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName) + require.Equal(t, "external-id", internalCfg.Sink.KafkaConfig.AWSMSKIAM.ExternalID) require.Equal(t, internalCfg.Mounter.WorkerNum, *apiCfg.Mounter.WorkerNum) require.True(t, util.GetOrZero(internalCfg.Scheduler.EnableTableAcrossNodes)) require.Equal(t, 1000, util.GetOrZero(internalCfg.Scheduler.RegionThreshold)) @@ -100,6 +115,11 @@ func TestReplicaConfigConversion(t *testing.T) { require.True(t, *apiCfgBack.Sink.CloudStorageConfig.UseTableIDAsPath) require.Equal(t, int64(1024), *apiCfgBack.Sink.CloudStorageConfig.SpoolDiskQuota) require.Equal(t, "/tmp/ticdc-spool", *apiCfgBack.Sink.CloudStorageConfig.SpoolBaseDir) + require.Equal(t, "AWS_MSK_IAM", *apiCfgBack.Sink.KafkaConfig.SASLOAuthProvider) + require.Equal(t, "us-west-2", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.Region) + require.Equal(t, "arn:aws:iam::123456789012:role/ticdc-msk", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.RoleARN) + require.Equal(t, "ticdc", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.RoleSessionName) + require.Equal(t, "external-id", apiCfgBack.Sink.KafkaConfig.AWSMSKIAM.ExternalID) require.Equal(t, 16, *apiCfgBack.Mounter.WorkerNum) require.True(t, *apiCfgBack.Scheduler.EnableTableAcrossNodes) require.Equal(t, "correctness", *apiCfgBack.Integrity.IntegrityCheckLevel) diff --git a/go.mod b/go.mod index 564652b685..59ab6cf676 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/KimMachineGun/automemlimit v0.2.4 github.com/agiledragon/gomonkey/v2 v2.11.0 github.com/apache/pulsar-client-go v0.13.0 + github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4 github.com/aws/aws-sdk-go v1.55.7 github.com/aws/aws-sdk-go-v2 v1.41.5 github.com/aws/aws-sdk-go-v2/config v1.32.2 diff --git a/go.sum b/go.sum index 2a9ecdef0e..d5a403efb6 100644 --- a/go.sum +++ b/go.sum @@ -138,6 +138,8 @@ github.com/ardielle/ardielle-tools v1.5.4/go.mod h1:oZN+JRMnqGiIhrzkRN9l26Cej9dE github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so= github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw= +github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4 h1:2jAwFwA0Xgcx94dUId+K24yFabsKYDtAhCgyMit6OqE= +github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4/go.mod h1:MVYeeOhILFFemC/XlYTClvBjYZrg/EPd3ts885KrNTI= github.com/aws/aws-sdk-go v1.32.6/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0= github.com/aws/aws-sdk-go v1.44.204/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI= github.com/aws/aws-sdk-go v1.55.7 h1:UJrkFq7es5CShfBwlWAC8DA077vp8PyVbQd3lqLiztE= diff --git a/pkg/config/sink.go b/pkg/config/sink.go index dca4d50572..1bc4fd54ba 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -457,6 +457,14 @@ type CodecConfig struct { OutputRowKey *bool `toml:"output-row-key" json:"output-row-key,omitempty"` } +// AWSMSKIAMConfig represents an Amazon MSK IAM authentication configuration. +type AWSMSKIAMConfig struct { + Region string `toml:"region" json:"region"` + RoleARN string `toml:"role-arn" json:"role-arn,omitempty"` + RoleSessionName string `toml:"role-session-name" json:"role-session-name,omitempty"` + ExternalID string `toml:"external-id" json:"external-id,omitempty"` +} + // KafkaConfig represents a kafka sink configuration type KafkaConfig struct { PartitionNum *int32 `toml:"partition-num" json:"partition-num,omitempty"` @@ -487,6 +495,8 @@ type KafkaConfig struct { SASLOAuthScopes []string `toml:"sasl-oauth-scopes" json:"sasl-oauth-scopes,omitempty"` SASLOAuthGrantType *string `toml:"sasl-oauth-grant-type" json:"sasl-oauth-grant-type,omitempty"` SASLOAuthAudience *string `toml:"sasl-oauth-audience" json:"sasl-oauth-audience,omitempty"` + SASLOAuthProvider *string `toml:"sasl-oauth-provider" json:"sasl-oauth-provider,omitempty"` + AWSMSKIAM *AWSMSKIAMConfig `toml:"aws-msk-iam" json:"aws-msk-iam,omitempty"` EnableTLS *bool `toml:"enable-tls" json:"enable-tls,omitempty"` CA *string `toml:"ca" json:"ca,omitempty"` Cert *string `toml:"cert" json:"cert,omitempty"` diff --git a/pkg/config/sink_test.go b/pkg/config/sink_test.go index 507bb250ef..d02a5aced2 100644 --- a/pkg/config/sink_test.go +++ b/pkg/config/sink_test.go @@ -17,11 +17,40 @@ import ( "net/url" "testing" + "github.com/BurntSushi/toml" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/util" "github.com/stretchr/testify/require" ) +func TestDecodeAWSMSKIAMConfig(t *testing.T) { + t.Parallel() + + var replicaConfig ReplicaConfig + metadata, err := toml.Decode(` +[sink.kafka-config] +sasl-mechanism = "OAUTHBEARER" +sasl-oauth-provider = "AWS_MSK_IAM" +enable-tls = true + +[sink.kafka-config.aws-msk-iam] +region = "ap-northeast-1" +role-arn = "arn:aws:iam::123456789012:role/TiCDCMSKProducer" +role-session-name = "ticdc" +external-id = "external-id" +`, &replicaConfig) + require.NoError(t, err) + require.Empty(t, metadata.Undecoded()) + require.NotNil(t, replicaConfig.Sink) + require.NotNil(t, replicaConfig.Sink.KafkaConfig) + require.Equal(t, &AWSMSKIAMConfig{ + Region: "ap-northeast-1", + RoleARN: "arn:aws:iam::123456789012:role/TiCDCMSKProducer", + RoleSessionName: "ticdc", + ExternalID: "external-id", + }, replicaConfig.Sink.KafkaConfig.AWSMSKIAM) +} + func TestValidateTxnAtomicity(t *testing.T) { t.Parallel() testCases := []struct { diff --git a/pkg/security/sasl.go b/pkg/security/sasl.go index 6b503b5bea..aa80b9cabe 100644 --- a/pkg/security/sasl.go +++ b/pkg/security/sasl.go @@ -64,6 +64,16 @@ type SASL struct { SASLMechanism SASLMechanism GSSAPI GSSAPI OAuth2 OAuth2 + OAuthProvider string + AWSMSKIAM AWSMSKIAM +} + +// AWSMSKIAM holds parameters for Amazon MSK IAM authentication. +type AWSMSKIAM struct { + Region string + RoleARN string + RoleSessionName string + ExternalID string } // OAuth2 holds necessary parameters to support sasl-oauth2. diff --git a/pkg/sink/kafka/aws_msk_iam_token_provider.go b/pkg/sink/kafka/aws_msk_iam_token_provider.go new file mode 100644 index 0000000000..594a52987f --- /dev/null +++ b/pkg/sink/kafka/aws_msk_iam_token_provider.go @@ -0,0 +1,54 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "context" + + "github.com/IBM/sarama" + "github.com/aws/aws-msk-iam-sasl-signer-go/signer" + "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/security" +) + +type awsMSKIAMTokenProvider struct { + ctx context.Context + config security.AWSMSKIAM +} + +var _ sarama.AccessTokenProvider = (*awsMSKIAMTokenProvider)(nil) + +func newAWSMSKIAMTokenProvider( + ctx context.Context, config security.AWSMSKIAM, +) sarama.AccessTokenProvider { + return &awsMSKIAMTokenProvider{ctx: ctx, config: config} +} + +func (p *awsMSKIAMTokenProvider) Token() (*sarama.AccessToken, error) { + var ( + token string + err error + ) + if p.config.RoleARN == "" { + token, _, err = signer.GenerateAuthToken(p.ctx, p.config.Region) + } else { + token, _, err = signer.GenerateAuthTokenFromRoleWithExternalId( + p.ctx, p.config.Region, p.config.RoleARN, + p.config.RoleSessionName, p.config.ExternalID) + } + if err != nil { + return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err) + } + return &sarama.AccessToken{Token: token}, nil +} diff --git a/pkg/sink/kafka/aws_msk_iam_token_provider_test.go b/pkg/sink/kafka/aws_msk_iam_token_provider_test.go new file mode 100644 index 0000000000..aa0c7f43d8 --- /dev/null +++ b/pkg/sink/kafka/aws_msk_iam_token_provider_test.go @@ -0,0 +1,55 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "context" + "testing" + + "github.com/IBM/sarama" + "github.com/pingcap/ticdc/pkg/security" + "github.com/stretchr/testify/require" +) + +func TestAWSMSKIAMTokenProvider(t *testing.T) { + t.Setenv("AWS_ACCESS_KEY_ID", "test-access-key") + t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret-key") + t.Setenv("AWS_EC2_METADATA_DISABLED", "true") + + provider := newAWSMSKIAMTokenProvider(t.Context(), security.AWSMSKIAM{ + Region: "us-east-1", + }) + token, err := provider.Token() + require.NoError(t, err) + require.NotEmpty(t, token.Token) +} + +func TestCompleteSaramaSASLConfigUsesAWSMSKIAMProvider(t *testing.T) { + saramaConfig := sarama.NewConfig() + options := NewOptions() + options.SASL.SASLMechanism = security.OAuthMechanism + options.SASL.OAuthProvider = SASLOAuthProviderAWSMSKIAM + options.SASL.AWSMSKIAM = security.AWSMSKIAM{ + Region: "ap-northeast-1", + RoleARN: "arn:aws:iam::123456789012:role/TiCDCMSKProducer", + RoleSessionName: "ticdc", + ExternalID: "external-id", + } + + err := completeSaramaSASLConfig(context.Background(), saramaConfig, options) + require.NoError(t, err) + provider, ok := saramaConfig.Net.SASL.TokenProvider.(*awsMSKIAMTokenProvider) + require.True(t, ok) + require.Equal(t, options.SASL.AWSMSKIAM, provider.config) +} diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index da1dfb3640..65016318f5 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -70,6 +70,8 @@ const ( SASLTypeGSSAPI = "GSSAPI" // SASLTypeOAuth represents the SASL/OAUTHBEARER mechanism (Kafka 2.0.0+) SASLTypeOAuth = "OAUTHBEARER" + // SASLOAuthProviderAWSMSKIAM uses AWS IAM credentials to authenticate with Amazon MSK. + SASLOAuthProviderAWSMSKIAM = "AWS_MSK_IAM" ) // RequiredAcks is used in Produce Requests to tell the broker how many replica acknowledgements @@ -338,6 +340,10 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if err != nil { return err } + if o.SASL.OAuthProvider == SASLOAuthProviderAWSMSKIAM && !o.EnableTLS { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS MSK IAM authentication requires TLS to be enabled") + } return nil } @@ -483,6 +489,29 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf } if sinkConfig != nil && sinkConfig.KafkaConfig != nil { + kafkaConfig := sinkConfig.KafkaConfig + if kafkaConfig.SASLOAuthProvider != nil { + o.SASL.OAuthProvider = strings.TrimSpace(*kafkaConfig.SASLOAuthProvider) + if o.SASL.OAuthProvider == "" { + return errors.ErrKafkaInvalidConfig.GenWithStack("SASL OAuth provider cannot be empty") + } + } + if kafkaConfig.AWSMSKIAM != nil { + o.SASL.AWSMSKIAM.Region = strings.TrimSpace(kafkaConfig.AWSMSKIAM.Region) + o.SASL.AWSMSKIAM.RoleARN = strings.TrimSpace(kafkaConfig.AWSMSKIAM.RoleARN) + o.SASL.AWSMSKIAM.RoleSessionName = strings.TrimSpace(kafkaConfig.AWSMSKIAM.RoleSessionName) + o.SASL.AWSMSKIAM.ExternalID = strings.TrimSpace(kafkaConfig.AWSMSKIAM.ExternalID) + } + standardOAuthConfigPresent := kafkaConfig.SASLOAuthClientID != nil || + kafkaConfig.SASLOAuthClientSecret != nil || kafkaConfig.SASLOAuthTokenURL != nil || + kafkaConfig.SASLOAuthScopes != nil || kafkaConfig.SASLOAuthGrantType != nil || + kafkaConfig.SASLOAuthAudience != nil + if o.SASL.OAuthProvider != "" && standardOAuthConfigPresent { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "SASL OAuth provider %s cannot be used with standard OAuth2 configuration", + o.SASL.OAuthProvider) + } + if sinkConfig.KafkaConfig.SASLOAuthClientID != nil { clientID := *sinkConfig.KafkaConfig.SASLOAuthClientID if clientID == "" { @@ -539,6 +568,29 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf if sinkConfig.KafkaConfig.SASLOAuthAudience != nil { o.SASL.OAuth2.Audience = *sinkConfig.KafkaConfig.SASLOAuthAudience } + + awsMSKIAM := o.SASL.AWSMSKIAM + awsConfigPresent := o.SASL.OAuthProvider != "" || kafkaConfig.AWSMSKIAM != nil + if awsConfigPresent { + if o.SASL.OAuthProvider != SASLOAuthProviderAWSMSKIAM { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "unsupported SASL OAuth provider %q", o.SASL.OAuthProvider) + } + if o.SASL.SASLMechanism != security.OAuthMechanism { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS MSK IAM is only supported with SASL mechanism type OAUTHBEARER, but got %s", + o.SASL.SASLMechanism) + } + if awsMSKIAM.Region == "" { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS region cannot be empty for AWS MSK IAM authentication") + } + if awsMSKIAM.RoleARN == "" && + (awsMSKIAM.RoleSessionName != "" || awsMSKIAM.ExternalID != "") { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "AWS role ARN is required when role session name or external ID is configured") + } + } } return nil diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 025df9086b..ce79fba05b 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -345,6 +345,148 @@ func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) { } } +func TestApplyAWSMSKIAM(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + kafkaConfig *config.KafkaConfig + expectedErr string + }{ + { + name: "default credential chain", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String("AWS_MSK_IAM"), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), + }, + }, + { + name: "assume role", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + RoleARN: "arn:aws:iam::123456789012:role/TiCDCMSKProducer", + RoleSessionName: "ticdc", + ExternalID: "external-id", + }, + EnableTLS: aws.Bool(true), + }, + }, + { + name: "TLS disabled", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + }, + expectedErr: "AWS MSK IAM authentication requires TLS", + }, + { + name: "missing region", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + AWSMSKIAM: &config.AWSMSKIAMConfig{}, + EnableTLS: aws.Bool(true), + }, + expectedErr: "AWS region cannot be empty", + }, + { + name: "unsupported provider", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String("unknown"), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), + }, + expectedErr: "unsupported SASL OAuth provider", + }, + { + name: "wrong SASL mechanism", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypePlaintext), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), + }, + expectedErr: "AWS MSK IAM is only supported with SASL mechanism type OAUTHBEARER", + }, + { + name: "external ID without role ARN", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + ExternalID: "external-id", + }, + EnableTLS: aws.Bool(true), + }, + expectedErr: "AWS role ARN is required", + }, + { + name: "mixed OAuth providers", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + SASLOAuthClientID: aws.String("client-id"), + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), + }, + expectedErr: "cannot be used with standard OAuth2 configuration", + }, + { + name: "AWS provider with standard OAuth scopes", + kafkaConfig: &config.KafkaConfig{ + SASLMechanism: aws.String(SASLTypeOAuth), + SASLOAuthProvider: aws.String(SASLOAuthProviderAWSMSKIAM), + SASLOAuthScopes: []string{"scope"}, + AWSMSKIAM: &config.AWSMSKIAMConfig{ + Region: "ap-northeast-1", + }, + EnableTLS: aws.Bool(true), + }, + expectedErr: "cannot be used with standard OAuth2 configuration", + }, + } + + changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic") + require.NoError(t, err) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + options := NewOptions() + err := options.Apply(changefeedID, sinkURI, &config.SinkConfig{ + KafkaConfig: test.kafkaConfig, + }) + if test.expectedErr != "" { + require.ErrorContains(t, err, test.expectedErr) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) + return + } + require.NoError(t, err) + require.Equal(t, SASLOAuthProviderAWSMSKIAM, options.SASL.OAuthProvider) + require.Equal(t, test.kafkaConfig.AWSMSKIAM.Region, options.SASL.AWSMSKIAM.Region) + require.True(t, options.EnableTLS) + }) + } +} + func TestSetPartitionNum(t *testing.T) { options := NewOptions() changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index 51dbd2384e..7a2693c4c7 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -172,7 +172,15 @@ func completeSaramaSASLConfig(ctx context.Context, config *sarama.Config, o *opt } case SASLTypeOAuth: - p, err := newTokenProvider(ctx, o) + var ( + p sarama.AccessTokenProvider + err error + ) + if o.SASL.OAuthProvider == SASLOAuthProviderAWSMSKIAM { + p = newAWSMSKIAMTokenProvider(ctx, o.SASL.AWSMSKIAM) + } else { + p, err = newTokenProvider(ctx, o) + } if err != nil { return err }