Skip to content

kafka: decouple security SASL types from Sarama #5952

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ti-chi-bot merged 4 commits into from
Aug 12, 2026
Merged

kafka: decouple security SASL types from Sarama #5952

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b4b8cd5
first commit
3AceShowHand Aug 11, 2026
3fb0837
move sasl to kafka
3AceShowHand Aug 12, 2026
0c7fc18
move scram to kafka package
3AceShowHand Aug 12, 2026
12be7dd
add more tests
3AceShowHand Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
139 changes: 0 additions & 139 deletions pkg/security/sasl.go
View file
Open in desktop

This file was deleted.

16 changes: 8 additions & 8 deletions pkg/sink/kafka/oauth2_token_provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,27 +56,27 @@ func newTokenProvider(ctx context.Context, o *options) (sarama.AccessTokenProvid
// client credentials library as defined by the spec, however non-compliant
// auth server implementations may want a custom type
endpointParams := url.Values{}
if o.SASL.OAuth2.GrantType != "" {
endpointParams.Set("grant_type", o.SASL.OAuth2.GrantType)
if o.sasl.oauth2.grantType != "" {
endpointParams.Set("grant_type", o.sasl.oauth2.grantType)
}

// audience is an optional parameter that can be used to specify the
// intended audience of the token.
if o.SASL.OAuth2.Audience != "" {
endpointParams.Set("audience", o.SASL.OAuth2.Audience)
if o.sasl.oauth2.audience != "" {
endpointParams.Set("audience", o.sasl.oauth2.audience)
}

tokenURL, err := url.Parse(o.SASL.OAuth2.TokenURL)
tokenURL, err := url.Parse(o.sasl.oauth2.tokenURL)
if err != nil {
return nil, errors.WrapError(errors.ErrKafkaInvalidConfig, err)
}

cfg := clientcredentials.Config{
ClientID: o.SASL.OAuth2.ClientID,
ClientSecret: o.SASL.OAuth2.ClientSecret,
ClientID: o.sasl.oauth2.clientID,
ClientSecret: o.sasl.oauth2.clientSecret,
TokenURL: tokenURL.String(),
EndpointParams: endpointParams,
Scopes: o.SASL.OAuth2.Scopes,
Scopes: o.sasl.oauth2.scopes,
}
return &tokenProvider{
tokenSource: cfg.TokenSource(ctx),
Expand Down
41 changes: 20 additions & 21 deletions pkg/sink/kafka/oauth2_token_provider_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"testing"

"github.com/pingcap/ticdc/pkg/errors"
"github.com/pingcap/ticdc/pkg/security"
"github.com/stretchr/testify/require"
"golang.org/x/oauth2"
)
Expand All @@ -30,13 +29,13 @@ func TestNewTokenProviderRejectsInvalidTokenURL(t *testing.T) {
t.Parallel()

options := &options{
SASL: &security.SASL{
OAuth2: security.OAuth2{
ClientID: "client-id",
ClientSecret: "client-secret",
TokenURL: "http://test.com/Segment%%2815197306101420000%29",
Scopes: []string{"scope1", "scope2"},
GrantType: "client_credentials",
sasl: &saslConfig{
oauth2: oauth2Config{
clientID: "client-id",
clientSecret: "client-secret",
tokenURL: "http://test.com/Segment%%2815197306101420000%29",
scopes: []string{"scope1", "scope2"},
grantType: "client_credentials",
},
},
}
Expand Down Expand Up @@ -75,14 +74,14 @@ func TestTokenProviderRequestsToken(t *testing.T) {
t.Cleanup(server.Close)

options := &options{
SASL: &security.SASL{
OAuth2: security.OAuth2{
ClientID: "client-id",
ClientSecret: "client-secret",
TokenURL: server.URL + "/oauth2/token",
Scopes: []string{"scope1", "scope2"},
GrantType: "custom_grant",
Audience: "test-audience",
sasl: &saslConfig{
oauth2: oauth2Config{
clientID: "client-id",
clientSecret: "client-secret",
tokenURL: server.URL + "/oauth2/token",
scopes: []string{"scope1", "scope2"},
grantType: "custom_grant",
audience: "test-audience",
},
},
}
Expand Down Expand Up @@ -115,11 +114,11 @@ func TestTokenProviderPropagatesEndpointError(t *testing.T) {
t.Cleanup(server.Close)

options := &options{
SASL: &security.SASL{
OAuth2: security.OAuth2{
ClientID: "client-id",
ClientSecret: "client-secret",
TokenURL: server.URL,
sasl: &saslConfig{
oauth2: oauth2Config{
clientID: "client-id",
clientSecret: "client-secret",
tokenURL: server.URL,
},
},
}
Expand Down
60 changes: 31 additions & 29 deletions pkg/sink/kafka/options.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ type options struct {
EnableTLS bool
Credential *security.Credential
InsecureSkipVerify bool
SASL *security.SASL
sasl *saslConfig

// Timeout for network configurations, default to `10s`
DialTimeout time.Duration
Expand All @@ -186,7 +186,7 @@ func NewOptions() *options {
RequiredAcks: WaitForAll,
Credential: &security.Credential{},
InsecureSkipVerify: false,
SASL: &security.SASL{},
sasl: &saslConfig{},
AutoCreate: true,
DialTimeout: defaultTimeout,
WriteTimeout: defaultTimeout,
Expand Down Expand Up @@ -430,56 +430,56 @@ func (o *options) applyTLS(params *urlConfig) error {

func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConfig) error {
if urlParameter.SASLUser != nil && *urlParameter.SASLUser != "" {
o.SASL.SASLUser = *urlParameter.SASLUser
o.sasl.user = *urlParameter.SASLUser
}

if urlParameter.SASLPassword != nil && *urlParameter.SASLPassword != "" {
o.SASL.SASLPassword = *urlParameter.SASLPassword
o.sasl.password = *urlParameter.SASLPassword
}

if urlParameter.SASLMechanism != nil && *urlParameter.SASLMechanism != "" {
mechanism, err := security.SASLMechanismFromString(*urlParameter.SASLMechanism)
mechanism, err := saslMechanismFromString(*urlParameter.SASLMechanism)
if err != nil {
return errors.WrapError(errors.ErrKafkaInvalidConfig, err)
return err
}
o.SASL.SASLMechanism = mechanism
o.sasl.mechanism = mechanism
}

if urlParameter.SASLGssAPIAuthType != nil && *urlParameter.SASLGssAPIAuthType != "" {
authType, err := security.AuthTypeFromString(*urlParameter.SASLGssAPIAuthType)
authType, err := gssapiAuthTypeFromString(*urlParameter.SASLGssAPIAuthType)
if err != nil {
return errors.WrapError(errors.ErrKafkaInvalidConfig, err)
return err
}
o.SASL.GSSAPI.AuthType = authType
o.sasl.gssapi.authType = authType
}

if urlParameter.SASLGssAPIKeytabPath != nil && *urlParameter.SASLGssAPIKeytabPath != "" {
o.SASL.GSSAPI.KeyTabPath = *urlParameter.SASLGssAPIKeytabPath
o.sasl.gssapi.keyTabPath = *urlParameter.SASLGssAPIKeytabPath
}

if urlParameter.SASLGssAPIKerberosConfigPath != nil &&
*urlParameter.SASLGssAPIKerberosConfigPath != "" {
o.SASL.GSSAPI.KerberosConfigPath = *urlParameter.SASLGssAPIKerberosConfigPath
o.sasl.gssapi.kerberosConfigPath = *urlParameter.SASLGssAPIKerberosConfigPath
}

if urlParameter.SASLGssAPIServiceName != nil && *urlParameter.SASLGssAPIServiceName != "" {
o.SASL.GSSAPI.ServiceName = *urlParameter.SASLGssAPIServiceName
o.sasl.gssapi.serviceName = *urlParameter.SASLGssAPIServiceName
}

if urlParameter.SASLGssAPIUser != nil && *urlParameter.SASLGssAPIUser != "" {
o.SASL.GSSAPI.Username = *urlParameter.SASLGssAPIUser
o.sasl.gssapi.username = *urlParameter.SASLGssAPIUser
}

if urlParameter.SASLGssAPIPassword != nil && *urlParameter.SASLGssAPIPassword != "" {
o.SASL.GSSAPI.Password = *urlParameter.SASLGssAPIPassword
o.sasl.gssapi.password = *urlParameter.SASLGssAPIPassword
}

if urlParameter.SASLGssAPIRealm != nil && *urlParameter.SASLGssAPIRealm != "" {
o.SASL.GSSAPI.Realm = *urlParameter.SASLGssAPIRealm
o.sasl.gssapi.realm = *urlParameter.SASLGssAPIRealm
}

if urlParameter.SASLGssAPIDisablePafxfast != nil {
o.SASL.GSSAPI.DisablePAFXFAST = *urlParameter.SASLGssAPIDisablePafxfast
o.sasl.gssapi.disablePAFXFAST = *urlParameter.SASLGssAPIDisablePafxfast
}

if sinkConfig != nil && sinkConfig.KafkaConfig != nil {
Expand All @@ -488,7 +488,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
if clientID == "" {
return errors.ErrKafkaInvalidConfig.GenWithStack("OAuth2 client ID cannot be empty")
}
o.SASL.OAuth2.ClientID = clientID
o.sasl.oauth2.clientID = clientID
}

if sinkConfig.KafkaConfig.SASLOAuthClientSecret != nil {
Expand All @@ -503,7 +503,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
if err != nil {
return errors.ErrKafkaInvalidConfig.GenWithStack("OAuth2 client secret is not base64 encoded")
}
o.SASL.OAuth2.ClientSecret = string(decodedClientSecret)
o.sasl.oauth2.clientSecret = string(decodedClientSecret)
}

if sinkConfig.KafkaConfig.SASLOAuthTokenURL != nil {
Expand All @@ -512,32 +512,34 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
return errors.ErrKafkaInvalidConfig.GenWithStack(
"OAuth2 token URL cannot be empty")
}
o.SASL.OAuth2.TokenURL = tokenURL
o.sasl.oauth2.tokenURL = tokenURL
}

if o.SASL.OAuth2.IsEnable() {
if o.SASL.SASLMechanism != security.OAuthMechanism {
if o.sasl.oauth2.clientID != "" ||
o.sasl.oauth2.clientSecret != "" ||
o.sasl.oauth2.tokenURL != "" {
if o.sasl.mechanism != oauthMechanism {
return errors.ErrKafkaInvalidConfig.GenWithStack(
"OAuth2 is only supported with SASL mechanism type OAUTHBEARER, but got %s",
o.SASL.SASLMechanism)
o.sasl.mechanism)
}

if err := o.SASL.OAuth2.Validate(); err != nil {
return errors.WrapError(errors.ErrKafkaInvalidConfig, err)
if err := o.sasl.oauth2.validate(); err != nil {
return err
}
o.SASL.OAuth2.SetDefault()
o.sasl.oauth2.grantType = "client_credentials"
}

if sinkConfig.KafkaConfig.SASLOAuthScopes != nil {
o.SASL.OAuth2.Scopes = sinkConfig.KafkaConfig.SASLOAuthScopes
o.sasl.oauth2.scopes = sinkConfig.KafkaConfig.SASLOAuthScopes
}

if sinkConfig.KafkaConfig.SASLOAuthGrantType != nil {
o.SASL.OAuth2.GrantType = *sinkConfig.KafkaConfig.SASLOAuthGrantType
o.sasl.oauth2.grantType = *sinkConfig.KafkaConfig.SASLOAuthGrantType
}

if sinkConfig.KafkaConfig.SASLOAuthAudience != nil {
o.SASL.OAuth2.Audience = *sinkConfig.KafkaConfig.SASLOAuthAudience
o.sasl.oauth2.audience = *sinkConfig.KafkaConfig.SASLOAuthAudience
}
}

Expand Down
Loading
Loading