diff --git a/pkg/security/sasl.go b/pkg/security/sasl.go deleted file mode 100644 index 6b503b5bea..0000000000 --- a/pkg/security/sasl.go +++ /dev/null @@ -1,139 +0,0 @@ -// Copyright 2020 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 security - -import ( - "strings" - - "github.com/IBM/sarama" - "github.com/pingcap/errors" -) - -// SASLMechanism defines SASL mechanism. -type SASLMechanism string - -// The mechanisms we currently support. -const ( - // UnknownMechanism means the SASL mechanism is unknown. - UnknownMechanism SASLMechanism = "" - // PlainMechanism means the SASL mechanism is plain. - PlainMechanism SASLMechanism = sarama.SASLTypePlaintext - // SCRAM256Mechanism means the SASL mechanism is SCRAM-SHA-256. - SCRAM256Mechanism SASLMechanism = sarama.SASLTypeSCRAMSHA256 - // SCRAM512Mechanism means the SASL mechanism is SCRAM-SHA-512. - SCRAM512Mechanism SASLMechanism = sarama.SASLTypeSCRAMSHA512 - // GSSAPIMechanism means the SASL mechanism is GSSAPI. - GSSAPIMechanism SASLMechanism = sarama.SASLTypeGSSAPI - // OAuthMechanism means the SASL mechanism is OAuth2. - OAuthMechanism SASLMechanism = sarama.SASLTypeOAuth -) - -// SASLMechanismFromString converts the string to SASL mechanism. -func SASLMechanismFromString(s string) (SASLMechanism, error) { - switch strings.ToLower(s) { - case "plain": - return PlainMechanism, nil - case "scram-sha-256": - return SCRAM256Mechanism, nil - case "scram-sha-512": - return SCRAM512Mechanism, nil - case "gssapi": - return GSSAPIMechanism, nil - case "oauthbearer": - return OAuthMechanism, nil - default: - return UnknownMechanism, errors.Errorf("unknown %s SASL mechanism", s) - } -} - -// SASL holds necessary path parameter to support sasl-scram -type SASL struct { - SASLUser string - SASLPassword string - SASLMechanism SASLMechanism - GSSAPI GSSAPI - OAuth2 OAuth2 -} - -// OAuth2 holds necessary parameters to support sasl-oauth2. -type OAuth2 struct { - ClientID string - ClientSecret string - TokenURL string - Scopes []string - GrantType string - Audience string -} - -// Validate validates the parameters of OAuth2. -// Some parameters are required, some are optional. -func (o *OAuth2) Validate() error { - if len(o.ClientID) == 0 { - return errors.New("OAuth2 client id is empty") - } - if len(o.ClientSecret) == 0 { - return errors.New("OAuth2 client secret is empty") - } - if len(o.TokenURL) == 0 { - return errors.New("OAuth2 token url is empty") - } - return nil -} - -// SetDefault sets the default value of OAuth2. -func (o *OAuth2) SetDefault() { - o.GrantType = "client_credentials" -} - -// IsEnable checks whether the OAuth2 is enabled. -// One of values of ClientID, ClientSecret and TokenURL is not empty means enabled. -func (o *OAuth2) IsEnable() bool { - return len(o.ClientID) > 0 || len(o.ClientSecret) > 0 || len(o.TokenURL) > 0 -} - -// GSSAPIAuthType defines the type of GSSAPI authentication. -type GSSAPIAuthType int - -const ( - // UnknownAuth means the auth type is unknown. - UnknownAuth GSSAPIAuthType = 0 - // UserAuth means the auth type is user. - UserAuth GSSAPIAuthType = sarama.KRB5_USER_AUTH - // KeyTabAuth means the auth type is keytab. - KeyTabAuth GSSAPIAuthType = sarama.KRB5_KEYTAB_AUTH -) - -// AuthTypeFromString convent the string to GSSAPIAuthType. -func AuthTypeFromString(s string) (GSSAPIAuthType, error) { - switch strings.ToLower(s) { - case "user": - return UserAuth, nil - case "keytab": - return KeyTabAuth, nil - default: - return UnknownAuth, errors.Errorf("unknown %s auth type", s) - } -} - -// GSSAPI holds necessary path parameter to support sasl-gssapi. -type GSSAPI struct { - AuthType GSSAPIAuthType `toml:"sasl-gssapi-auth-type" json:"sasl-gssapi-auth-type"` - KeyTabPath string `toml:"sasl-gssapi-keytab-path" json:"sasl-gssapi-keytab-path"` - KerberosConfigPath string `toml:"sasl-gssapi-kerberos-config-path" json:"sasl-gssapi-kerberos-config-path"` - ServiceName string `toml:"sasl-gssapi-service-name" json:"sasl-gssapi-service-name"` - Username string `toml:"sasl-gssapi-user" json:"sasl-gssapi-user"` - Password string `toml:"sasl-gssapi-password" json:"sasl-gssapi-password"` - Realm string `toml:"sasl-gssapi-realm" json:"sasl-gssapi-realm"` - DisablePAFXFAST bool `toml:"sasl-gssapi-disable-pafxfast" json:"sasl-gssapi-disable-pafxfast"` -} diff --git a/pkg/sink/kafka/oauth2_token_provider.go b/pkg/sink/kafka/oauth2_token_provider.go index b38e150d93..6a474b6725 100644 --- a/pkg/sink/kafka/oauth2_token_provider.go +++ b/pkg/sink/kafka/oauth2_token_provider.go @@ -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), diff --git a/pkg/sink/kafka/oauth2_token_provider_test.go b/pkg/sink/kafka/oauth2_token_provider_test.go index d796d80d4e..d7d0473c32 100644 --- a/pkg/sink/kafka/oauth2_token_provider_test.go +++ b/pkg/sink/kafka/oauth2_token_provider_test.go @@ -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" ) @@ -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", }, }, } @@ -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", }, }, } @@ -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, }, }, } diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 35ead5fcec..81cc997d6f 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -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 @@ -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, @@ -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 { @@ -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 { @@ -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 { @@ -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 } } diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index cd5f52c73e..1fe52bab86 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -173,26 +173,35 @@ func TestApplySASL(t *testing.T) { name string uri string kafkaConfig *config.KafkaConfig - expected security.SASL + expected saslConfig expectErr string }{ {name: "no params", uri: baseURI}, { name: "valid PLAIN SASL", uri: baseURI + "?sasl-user=user&sasl-password=password&sasl-mechanism=plain", - expected: security.SASL{ - SASLUser: "user", - SASLPassword: "password", - SASLMechanism: security.PlainMechanism, + expected: saslConfig{ + user: "user", + password: "password", + mechanism: plainMechanism, }, }, { - name: "valid SCRAM SASL", + name: "valid SCRAM-SHA-256 SASL", + uri: baseURI + "?sasl-user=user&sasl-password=password&sasl-mechanism=scram-sha-256", + expected: saslConfig{ + user: "user", + password: "password", + mechanism: scram256Mechanism, + }, + }, + { + name: "valid SCRAM-SHA-512 SASL", uri: baseURI + "?sasl-user=user&sasl-password=password&sasl-mechanism=SCRAM-SHA-512", - expected: security.SASL{ - SASLUser: "user", - SASLPassword: "password", - SASLMechanism: security.SCRAM512Mechanism, + expected: saslConfig{ + user: "user", + password: "password", + mechanism: scram512Mechanism, }, }, { @@ -202,15 +211,15 @@ func TestApplySASL(t *testing.T) { "&sasl-gssapi-service-name=a&sasl-gssapi-user=user" + "&sasl-gssapi-password=pwd&sasl-gssapi-realm=realm" + "&sasl-gssapi-disable-pafxfast=false", - expected: security.SASL{ - SASLMechanism: security.GSSAPIMechanism, - GSSAPI: security.GSSAPI{ - AuthType: security.UserAuth, - KerberosConfigPath: "/root/config", - ServiceName: "a", - Username: "user", - Password: "pwd", - Realm: "realm", + expected: saslConfig{ + mechanism: gssapiMechanism, + gssapi: gssapiConfig{ + authType: userAuth, + kerberosConfigPath: "/root/config", + serviceName: "a", + username: "user", + password: "pwd", + realm: "realm", }, }, }, @@ -221,27 +230,27 @@ func TestApplySASL(t *testing.T) { "&sasl-gssapi-service-name=a&sasl-gssapi-user=user" + "&sasl-gssapi-keytab-path=/root/keytab&sasl-gssapi-realm=realm" + "&sasl-gssapi-disable-pafxfast=false", - expected: security.SASL{ - SASLMechanism: security.GSSAPIMechanism, - GSSAPI: security.GSSAPI{ - AuthType: security.KeyTabAuth, - KeyTabPath: "/root/keytab", - KerberosConfigPath: "/root/config", - ServiceName: "a", - Username: "user", - Realm: "realm", + expected: saslConfig{ + mechanism: gssapiMechanism, + gssapi: gssapiConfig{ + authType: keyTabAuth, + keyTabPath: "/root/keytab", + kerberosConfigPath: "/root/config", + serviceName: "a", + username: "user", + realm: "realm", }, }, }, { name: "invalid mechanism", uri: baseURI + "?sasl-mechanism=a", - expectErr: "unknown a SASL mechanism", + expectErr: "unknown SASL mechanism: a", }, { name: "invalid GSSAPI auth type", uri: baseURI + "?sasl-mechanism=gssapi&sasl-gssapi-auth-type=keyta1b", - expectErr: "unknown keyta1b auth type", + expectErr: "unknown auth type: keyta1b", }, { name: "valid OAUTHBEARER SASL", @@ -251,13 +260,13 @@ func TestApplySASL(t *testing.T) { SASLOAuthClientSecret: aws.String("Y2xpZW50X3NlY3JldA=="), SASLOAuthTokenURL: aws.String("127.0.0.1:9093/token"), }, - expected: security.SASL{ - SASLMechanism: security.OAuthMechanism, - OAuth2: security.OAuth2{ - ClientID: "client_id", - ClientSecret: "client_secret", - TokenURL: "127.0.0.1:9093/token", - GrantType: "client_credentials", + expected: saslConfig{ + mechanism: oauthMechanism, + oauth2: oauth2Config{ + clientID: "client_id", + clientSecret: "client_secret", + tokenURL: "127.0.0.1:9093/token", + grantType: "client_credentials", }, }, }, @@ -331,7 +340,7 @@ func TestApplySASL(t *testing.T) { } require.NoError(t, err) - require.Equal(t, test.expected, *options.SASL) + require.Equal(t, test.expected, *options.sasl) }) } } @@ -966,16 +975,16 @@ func TestMerge(t *testing.T) { require.Equal(t, time.Minute+time.Second, c.DialTimeout) require.Equal(t, 2*time.Minute+time.Second, c.WriteTimeout) require.Equal(t, 1, int(c.RequiredAcks)) - require.Equal(t, "abc", c.SASL.SASLUser) - require.Equal(t, "123", c.SASL.SASLPassword) - require.Equal(t, "plain", strings.ToLower(string(c.SASL.SASLMechanism))) - require.Equal(t, 2, int(c.SASL.GSSAPI.AuthType)) - require.Equal(t, "SASLGssAPIKeytabPath", c.SASL.GSSAPI.KeyTabPath) - require.Equal(t, "service", c.SASL.GSSAPI.ServiceName) - require.Equal(t, "user", c.SASL.GSSAPI.Username) - require.Equal(t, "pass", c.SASL.GSSAPI.Password) - require.Equal(t, "realm", c.SASL.GSSAPI.Realm) - require.Equal(t, true, c.SASL.GSSAPI.DisablePAFXFAST) + require.Equal(t, "abc", c.sasl.user) + require.Equal(t, "123", c.sasl.password) + require.Equal(t, "plain", strings.ToLower(string(c.sasl.mechanism))) + require.Equal(t, 2, int(c.sasl.gssapi.authType)) + require.Equal(t, "SASLGssAPIKeytabPath", c.sasl.gssapi.keyTabPath) + require.Equal(t, "service", c.sasl.gssapi.serviceName) + require.Equal(t, "user", c.sasl.gssapi.username) + require.Equal(t, "pass", c.sasl.gssapi.password) + require.Equal(t, "realm", c.sasl.gssapi.realm) + require.Equal(t, true, c.sasl.gssapi.disablePAFXFAST) require.Equal(t, true, c.EnableTLS) require.Equal(t, "ca.pem", c.Credential.CAPath) require.Equal(t, "cert.pem", c.Credential.CertPath) @@ -1048,16 +1057,16 @@ func TestMerge(t *testing.T) { require.Equal(t, time.Minute+time.Second, c.DialTimeout) require.Equal(t, 2*time.Minute+time.Second, c.WriteTimeout) require.Equal(t, 1, int(c.RequiredAcks)) - require.Equal(t, "abc", c.SASL.SASLUser) - require.Equal(t, "123", c.SASL.SASLPassword) - require.Equal(t, "plain", strings.ToLower(string(c.SASL.SASLMechanism))) - require.Equal(t, 2, int(c.SASL.GSSAPI.AuthType)) - require.Equal(t, "SASLGssAPIKeytabPath", c.SASL.GSSAPI.KeyTabPath) - require.Equal(t, "service", c.SASL.GSSAPI.ServiceName) - require.Equal(t, "user", c.SASL.GSSAPI.Username) - require.Equal(t, "pass", c.SASL.GSSAPI.Password) - require.Equal(t, "realm", c.SASL.GSSAPI.Realm) - require.Equal(t, true, c.SASL.GSSAPI.DisablePAFXFAST) + require.Equal(t, "abc", c.sasl.user) + require.Equal(t, "123", c.sasl.password) + require.Equal(t, "plain", strings.ToLower(string(c.sasl.mechanism))) + require.Equal(t, 2, int(c.sasl.gssapi.authType)) + require.Equal(t, "SASLGssAPIKeytabPath", c.sasl.gssapi.keyTabPath) + require.Equal(t, "service", c.sasl.gssapi.serviceName) + require.Equal(t, "user", c.sasl.gssapi.username) + require.Equal(t, "pass", c.sasl.gssapi.password) + require.Equal(t, "realm", c.sasl.gssapi.realm) + require.Equal(t, true, c.sasl.gssapi.disablePAFXFAST) require.Equal(t, true, c.EnableTLS) require.Equal(t, "ca.pem", c.Credential.CAPath) require.Equal(t, "cert.pem", c.Credential.CertPath) diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go index 51dbd2384e..932de956b6 100644 --- a/pkg/sink/kafka/sarama_config.go +++ b/pkg/sink/kafka/sarama_config.go @@ -23,7 +23,6 @@ import ( "github.com/IBM/sarama" "github.com/pingcap/log" "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/security" "go.uber.org/zap" ) @@ -141,37 +140,37 @@ func completeSaramaKafkaVersion(config *sarama.Config, o *options) error { } func completeSaramaSASLConfig(ctx context.Context, config *sarama.Config, o *options) error { - if o.SASL != nil && o.SASL.SASLMechanism != "" { + if o.sasl != nil && o.sasl.mechanism != "" { config.Net.SASL.Enable = true - config.Net.SASL.Mechanism = sarama.SASLMechanism(o.SASL.SASLMechanism) - switch o.SASL.SASLMechanism { - case SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512, SASLTypePlaintext: - config.Net.SASL.User = o.SASL.SASLUser - config.Net.SASL.Password = o.SASL.SASLPassword - if strings.EqualFold(string(o.SASL.SASLMechanism), SASLTypeSCRAMSHA256) { + config.Net.SASL.Mechanism = sarama.SASLMechanism(o.sasl.mechanism) + switch o.sasl.mechanism { + case scram256Mechanism, scram512Mechanism, plainMechanism: + config.Net.SASL.User = o.sasl.user + config.Net.SASL.Password = o.sasl.password + if strings.EqualFold(string(o.sasl.mechanism), string(scram256Mechanism)) { config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { - return &security.XDGSCRAMClient{HashGeneratorFcn: security.SHA256} + return &xdgSCRAMClient{HashGeneratorFcn: sha256HashGenerator} } - } else if strings.EqualFold(string(o.SASL.SASLMechanism), SASLTypeSCRAMSHA512) { + } else if strings.EqualFold(string(o.sasl.mechanism), string(scram512Mechanism)) { config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { - return &security.XDGSCRAMClient{HashGeneratorFcn: security.SHA512} + return &xdgSCRAMClient{HashGeneratorFcn: sha512HashGenerator} } } - case SASLTypeGSSAPI: - config.Net.SASL.GSSAPI.AuthType = int(o.SASL.GSSAPI.AuthType) - config.Net.SASL.GSSAPI.Username = o.SASL.GSSAPI.Username - config.Net.SASL.GSSAPI.ServiceName = o.SASL.GSSAPI.ServiceName - config.Net.SASL.GSSAPI.KerberosConfigPath = o.SASL.GSSAPI.KerberosConfigPath - config.Net.SASL.GSSAPI.Realm = o.SASL.GSSAPI.Realm - config.Net.SASL.GSSAPI.DisablePAFXFAST = o.SASL.GSSAPI.DisablePAFXFAST - switch o.SASL.GSSAPI.AuthType { - case security.UserAuth: - config.Net.SASL.GSSAPI.Password = o.SASL.GSSAPI.Password - case security.KeyTabAuth: - config.Net.SASL.GSSAPI.KeyTabPath = o.SASL.GSSAPI.KeyTabPath + case gssapiMechanism: + config.Net.SASL.GSSAPI.AuthType = int(o.sasl.gssapi.authType) + config.Net.SASL.GSSAPI.Username = o.sasl.gssapi.username + config.Net.SASL.GSSAPI.ServiceName = o.sasl.gssapi.serviceName + config.Net.SASL.GSSAPI.KerberosConfigPath = o.sasl.gssapi.kerberosConfigPath + config.Net.SASL.GSSAPI.Realm = o.sasl.gssapi.realm + config.Net.SASL.GSSAPI.DisablePAFXFAST = o.sasl.gssapi.disablePAFXFAST + switch o.sasl.gssapi.authType { + case userAuth: + config.Net.SASL.GSSAPI.Password = o.sasl.gssapi.password + case keyTabAuth: + config.Net.SASL.GSSAPI.KeyTabPath = o.sasl.gssapi.keyTabPath } - case SASLTypeOAuth: + case oauthMechanism: p, err := newTokenProvider(ctx, o) if err != nil { return err diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go index 051c0f59f2..d8070361ba 100644 --- a/pkg/sink/kafka/sarama_config_test.go +++ b/pkg/sink/kafka/sarama_config_test.go @@ -70,10 +70,10 @@ func TestNewSaramaConfig(t *testing.T) { saslOptions := NewOptions() saslOptions.Version = "2.6.0" saslOptions.ClientID = "test-sasl-scram" - saslOptions.SASL = &security.SASL{ - SASLUser: "user", - SASLPassword: "password", - SASLMechanism: sarama.SASLTypeSCRAMSHA256, + saslOptions.sasl = &saslConfig{ + user: "user", + password: "password", + mechanism: scram256Mechanism, } cfg, err = newSaramaConfig(ctx, saslOptions) @@ -198,35 +198,112 @@ func TestNewSaramaConfigMaxRetryFromSinkURI(t *testing.T) { func TestCompleteSaramaSASLConfig(t *testing.T) { t.Parallel() - // Test that SASL is turned on correctly. - options := NewOptions() - options.SASL = &security.SASL{ - SASLUser: "user", - SASLPassword: "password", - SASLMechanism: "", - GSSAPI: security.GSSAPI{}, + tests := []struct { + name string + sasl *saslConfig + verify func(*testing.T, *sarama.Config) + }{ + { + name: "disabled", + sasl: &saslConfig{}, + verify: func(t *testing.T, config *sarama.Config) { + require.False(t, config.Net.SASL.Enable) + }, + }, + { + name: "PLAIN", + sasl: &saslConfig{user: "user", password: "password", mechanism: plainMechanism}, + verify: func(t *testing.T, config *sarama.Config) { + require.Equal(t, "user", config.Net.SASL.User) + require.Equal(t, "password", config.Net.SASL.Password) + require.Nil(t, config.Net.SASL.SCRAMClientGeneratorFunc) + }, + }, + { + name: "SCRAM-SHA-256", + sasl: &saslConfig{user: "user", password: "password", mechanism: scram256Mechanism}, + verify: func(t *testing.T, config *sarama.Config) { + require.Equal(t, "user", config.Net.SASL.User) + require.Equal(t, "password", config.Net.SASL.Password) + require.NotNil(t, config.Net.SASL.SCRAMClientGeneratorFunc) + }, + }, + { + name: "SCRAM-SHA-512", + sasl: &saslConfig{user: "user", password: "password", mechanism: scram512Mechanism}, + verify: func(t *testing.T, config *sarama.Config) { + require.Equal(t, "user", config.Net.SASL.User) + require.Equal(t, "password", config.Net.SASL.Password) + require.NotNil(t, config.Net.SASL.SCRAMClientGeneratorFunc) + }, + }, + { + name: "GSSAPI user auth", + sasl: &saslConfig{mechanism: gssapiMechanism, gssapi: gssapiConfig{ + authType: userAuth, + kerberosConfigPath: "/etc/krb5.conf", + serviceName: "kafka", + username: "user", + password: "password", + realm: "EXAMPLE.COM", + disablePAFXFAST: true, + }}, + verify: func(t *testing.T, config *sarama.Config) { + require.Equal(t, int(userAuth), config.Net.SASL.GSSAPI.AuthType) + require.Equal(t, "/etc/krb5.conf", config.Net.SASL.GSSAPI.KerberosConfigPath) + require.Equal(t, "kafka", config.Net.SASL.GSSAPI.ServiceName) + require.Equal(t, "user", config.Net.SASL.GSSAPI.Username) + require.Equal(t, "password", config.Net.SASL.GSSAPI.Password) + require.Empty(t, config.Net.SASL.GSSAPI.KeyTabPath) + require.Equal(t, "EXAMPLE.COM", config.Net.SASL.GSSAPI.Realm) + require.True(t, config.Net.SASL.GSSAPI.DisablePAFXFAST) + }, + }, + { + name: "GSSAPI keytab auth", + sasl: &saslConfig{mechanism: gssapiMechanism, gssapi: gssapiConfig{ + authType: keyTabAuth, + keyTabPath: "/tmp/user.keytab", + kerberosConfigPath: "/etc/krb5.conf", + serviceName: "kafka", + username: "user", + password: "unused", + realm: "EXAMPLE.COM", + }}, + verify: func(t *testing.T, config *sarama.Config) { + require.Equal(t, int(keyTabAuth), config.Net.SASL.GSSAPI.AuthType) + require.Equal(t, "/tmp/user.keytab", config.Net.SASL.GSSAPI.KeyTabPath) + require.Empty(t, config.Net.SASL.GSSAPI.Password) + }, + }, + { + name: "OAUTHBEARER", + sasl: &saslConfig{mechanism: oauthMechanism, oauth2: oauth2Config{ + clientID: "client-id", + clientSecret: "client-secret", + tokenURL: "http://127.0.0.1/token", + }}, + verify: func(t *testing.T, config *sarama.Config) { + require.NotNil(t, config.Net.SASL.TokenProvider) + }, + }, } - ctx := context.Background() - saramaConfig := sarama.NewConfig() - completeSaramaSASLConfig(ctx, saramaConfig, options) - require.False(t, saramaConfig.Net.SASL.Enable) - options.SASL.SASLMechanism = "plain" - completeSaramaSASLConfig(ctx, saramaConfig, options) - require.True(t, saramaConfig.Net.SASL.Enable) - // Test that the SCRAMClientGeneratorFunc is set up correctly. - options = NewOptions() - options.SASL = &security.SASL{ - SASLUser: "user", - SASLPassword: "password", - SASLMechanism: "plain", - GSSAPI: security.GSSAPI{}, + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + config := sarama.NewConfig() + options := NewOptions() + options.sasl = test.sasl + err := completeSaramaSASLConfig(t.Context(), config, options) + require.NoError(t, err) + if test.sasl.mechanism != "" { + require.True(t, config.Net.SASL.Enable) + require.Equal(t, sarama.SASLMechanism(test.sasl.mechanism), config.Net.SASL.Mechanism) + } + test.verify(t, config) + }) } - saramaConfig = sarama.NewConfig() - completeSaramaSASLConfig(ctx, saramaConfig, options) - require.Nil(t, saramaConfig.Net.SASL.SCRAMClientGeneratorFunc) - options.SASL.SASLMechanism = "SCRAM-SHA-512" - completeSaramaSASLConfig(ctx, saramaConfig, options) - require.NotNil(t, saramaConfig.Net.SASL.SCRAMClientGeneratorFunc) } func TestSaramaTimeout(t *testing.T) { diff --git a/pkg/sink/kafka/sasl_config.go b/pkg/sink/kafka/sasl_config.go new file mode 100644 index 0000000000..95389a1774 --- /dev/null +++ b/pkg/sink/kafka/sasl_config.go @@ -0,0 +1,124 @@ +// Copyright 2020 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 ( + "strings" + + "github.com/pingcap/ticdc/pkg/errors" +) + +// saslMechanism defines a SASL mechanism. +type saslMechanism string + +// The mechanisms we currently support. +const ( + // plainMechanism means the SASL mechanism is plain. + plainMechanism saslMechanism = "PLAIN" + // scram256Mechanism means the SASL mechanism is SCRAM-SHA-256. + scram256Mechanism saslMechanism = "SCRAM-SHA-256" + // scram512Mechanism means the SASL mechanism is SCRAM-SHA-512. + scram512Mechanism saslMechanism = "SCRAM-SHA-512" + // gssapiMechanism means the SASL mechanism is GSSAPI. + gssapiMechanism saslMechanism = "GSSAPI" + // oauthMechanism means the SASL mechanism is OAUTHBEARER. + oauthMechanism saslMechanism = "OAUTHBEARER" +) + +// saslMechanismFromString converts the string to a SASL mechanism. +func saslMechanismFromString(s string) (saslMechanism, error) { + switch strings.ToLower(s) { + case "plain": + return plainMechanism, nil + case "scram-sha-256": + return scram256Mechanism, nil + case "scram-sha-512": + return scram512Mechanism, nil + case "gssapi": + return gssapiMechanism, nil + case "oauthbearer": + return oauthMechanism, nil + default: + return "", errors.ErrKafkaInvalidConfig.FastGen("unknown SASL mechanism: %s", s) + } +} + +// saslConfig holds the configuration for SASL authentication. +type saslConfig struct { + user string + password string + mechanism saslMechanism + gssapi gssapiConfig + oauth2 oauth2Config +} + +// oauth2Config holds necessary parameters to support sasl-oauth2. +type oauth2Config struct { + clientID string + clientSecret string + tokenURL string + scopes []string + grantType string + audience string +} + +// validate validates the OAuth2 parameters. +func (o *oauth2Config) validate() error { + if len(o.clientID) == 0 { + return errors.ErrKafkaInvalidConfig.FastGen("OAuth2 client id is empty") + } + if len(o.clientSecret) == 0 { + return errors.ErrKafkaInvalidConfig.FastGen("OAuth2 client secret is empty") + } + if len(o.tokenURL) == 0 { + return errors.ErrKafkaInvalidConfig.FastGen("OAuth2 token url is empty") + } + return nil +} + +// gssapiAuthType defines the type of GSSAPI authentication. +type gssapiAuthType int + +const ( + // unknownAuth means the auth type is unknown. + unknownAuth gssapiAuthType = 0 + // userAuth means the auth type is user. + userAuth gssapiAuthType = 1 + // keyTabAuth means the auth type is keytab. + keyTabAuth gssapiAuthType = 2 +) + +// gssapiAuthTypeFromString converts the string to a GSSAPI authentication type. +func gssapiAuthTypeFromString(s string) (gssapiAuthType, error) { + switch strings.ToLower(s) { + case "user": + return userAuth, nil + case "keytab": + return keyTabAuth, nil + default: + return unknownAuth, errors.ErrKafkaInvalidConfig.FastGen("unknown auth type: %s", s) + } +} + +// gssapiConfig holds necessary parameters to support sasl-gssapi. +type gssapiConfig struct { + authType gssapiAuthType + keyTabPath string + kerberosConfigPath string + serviceName string + username string + password string + realm string + disablePAFXFAST bool +} diff --git a/pkg/security/sasl_test.go b/pkg/sink/kafka/sasl_config_test.go similarity index 59% rename from pkg/security/sasl_test.go rename to pkg/sink/kafka/sasl_config_test.go index 71f3c90754..d335750ea5 100644 --- a/pkg/security/sasl_test.go +++ b/pkg/sink/kafka/sasl_config_test.go @@ -11,11 +11,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -package security +package kafka import ( "testing" + "github.com/pingcap/ticdc/pkg/errors" "github.com/stretchr/testify/require" ) @@ -32,18 +33,10 @@ func TestSASLMechanismFromString(t *testing.T) { name: "random mechanism", s: "random", expectedMechanism: "", - expectErr: "unknown random SASL mechanism", - }, - { - name: "lower case plain mechanism", - s: "plain", - expectedMechanism: "PLAIN", - }, - { - name: "upper case plain mechanism", - s: "PLAIN", - expectedMechanism: "PLAIN", + expectErr: "unknown SASL mechanism: random", }, + {name: "lower case plain mechanism", s: "plain", expectedMechanism: "PLAIN"}, + {name: "upper case plain mechanism", s: "PLAIN", expectedMechanism: "PLAIN"}, { name: "lower case scram-sha-256 mechanism", s: "scram-sha-256", @@ -64,33 +57,35 @@ func TestSASLMechanismFromString(t *testing.T) { s: "SCRAM-SHA-512", expectedMechanism: "SCRAM-SHA-512", }, + {name: "lower case gssapi mechanism", s: "gssapi", expectedMechanism: "GSSAPI"}, + {name: "upper case GSSAPI mechanism", s: "GSSAPI", expectedMechanism: "GSSAPI"}, { - name: "lower case gssapi mechanism", - s: "gssapi", - expectedMechanism: "GSSAPI", + name: "lower case oauthbearer mechanism", + s: "oauthbearer", + expectedMechanism: "OAUTHBEARER", }, { - name: "upper case GSSAPI mechanism", - s: "GSSAPI", - expectedMechanism: "GSSAPI", + name: "upper case OAUTHBEARER mechanism", + s: "OAUTHBEARER", + expectedMechanism: "OAUTHBEARER", }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { t.Parallel() - mechanism, err := SASLMechanismFromString(test.s) + mechanism, err := saslMechanismFromString(test.s) + require.Equal(t, test.expectedMechanism, string(mechanism)) if test.expectErr != "" { - require.Error(t, err) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) require.Regexp(t, test.expectErr, err.Error()) } else { require.NoError(t, err) - require.Equal(t, test.expectedMechanism, string(mechanism)) } }) } } -func TestAuthTypeFromString(t *testing.T) { +func TestGSSAPIAuthTypeFromString(t *testing.T) { t.Parallel() tests := []struct { @@ -99,45 +94,24 @@ func TestAuthTypeFromString(t *testing.T) { expectedType int expectErr string }{ - { - name: "unknown", - s: "a", - expectedType: 0, - expectErr: "unknown a auth type", - }, - { - name: "lower case user", - s: "user", - expectedType: 1, - }, - { - name: "upper case user", - s: "USER", - expectedType: 1, - }, - { - name: "lower case keytab", - s: "keytab", - expectedType: 2, - }, - { - name: "upper case keytab", - s: "KEYTAB", - expectedType: 2, - }, + {name: "unknown", s: "a", expectedType: 0, expectErr: "unknown auth type: a"}, + {name: "lower case user", s: "user", expectedType: 1}, + {name: "upper case user", s: "USER", expectedType: 1}, + {name: "lower case keytab", s: "keytab", expectedType: 2}, + {name: "upper case keytab", s: "KEYTAB", expectedType: 2}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { t.Parallel() - authType, err := AuthTypeFromString(test.s) + authType, err := gssapiAuthTypeFromString(test.s) + require.Equal(t, test.expectedType, int(authType)) if test.expectErr != "" { - require.Error(t, err) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) require.Regexp(t, test.expectErr, err.Error()) } else { require.NoError(t, err) - require.Equal(t, test.expectedType, int(authType)) } }) } diff --git a/pkg/security/scram_client.go b/pkg/sink/kafka/scram_client.go similarity index 64% rename from pkg/security/scram_client.go rename to pkg/sink/kafka/scram_client.go index 481ee545e4..85d2fff890 100644 --- a/pkg/security/scram_client.go +++ b/pkg/sink/kafka/scram_client.go @@ -11,7 +11,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package security +package kafka import ( "crypto/sha256" @@ -22,21 +22,17 @@ import ( ) var ( - // SHA256 func - SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() } - // SHA512 func - SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() } + sha256HashGenerator scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() } + sha512HashGenerator scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() } ) -// XDGSCRAMClient xdg scram client -type XDGSCRAMClient struct { +type xdgSCRAMClient struct { *scram.Client *scram.ClientConversation scram.HashGeneratorFcn } -// Begin xdg scram client Begin -func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) { +func (x *xdgSCRAMClient) Begin(userName, password, authzID string) (err error) { x.Client, err = x.NewClient(userName, password, authzID) if err != nil { return err @@ -45,13 +41,11 @@ func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) { return nil } -// Step xdg scram client Step -func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) { +func (x *xdgSCRAMClient) Step(challenge string) (response string, err error) { response, err = x.ClientConversation.Step(challenge) return } -// Done xdg scram client Done -func (x *XDGSCRAMClient) Done() bool { +func (x *xdgSCRAMClient) Done() bool { return x.ClientConversation.Done() } diff --git a/pkg/sink/kafka/scram_client_test.go b/pkg/sink/kafka/scram_client_test.go new file mode 100644 index 0000000000..ae0829f2a7 --- /dev/null +++ b/pkg/sink/kafka/scram_client_test.go @@ -0,0 +1,85 @@ +// 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 ( + "testing" + + "github.com/IBM/sarama" + "github.com/stretchr/testify/require" + "github.com/xdg/scram" +) + +func TestSCRAMClientGeneratorHandshake(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + mechanism saslMechanism + hashGenerator scram.HashGeneratorFcn + }{ + {name: "SHA-256", mechanism: scram256Mechanism, hashGenerator: sha256HashGenerator}, + {name: "SHA-512", mechanism: scram512Mechanism, hashGenerator: sha512HashGenerator}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + const ( + username = "user" + password = "password" + ) + options := NewOptions() + options.sasl = &saslConfig{ + user: username, + password: password, + mechanism: test.mechanism, + } + config := sarama.NewConfig() + require.NoError(t, completeSaramaSASLConfig(t.Context(), config, options)) + require.NotNil(t, config.Net.SASL.SCRAMClientGeneratorFunc) + + client := config.Net.SASL.SCRAMClientGeneratorFunc() + require.NoError(t, client.Begin(username, password, "")) + + credentialClient, err := test.hashGenerator.NewClient(username, password, "") + require.NoError(t, err) + credentials := credentialClient.GetStoredCredentials(scram.KeyFactors{ + Salt: "salt", + Iters: 4096, + }) + server, err := test.hashGenerator.NewServer( + func(string) (scram.StoredCredentials, error) { return credentials, nil }) + require.NoError(t, err) + serverConversation := server.NewConversation() + + clientMessage, err := client.Step("") + require.NoError(t, err) + serverMessage, err := serverConversation.Step(clientMessage) + require.NoError(t, err) + clientMessage, err = client.Step(serverMessage) + require.NoError(t, err) + serverMessage, err = serverConversation.Step(clientMessage) + require.NoError(t, err) + clientMessage, err = client.Step(serverMessage) + require.NoError(t, err) + + require.Empty(t, clientMessage) + require.True(t, client.Done()) + require.True(t, serverConversation.Done()) + require.True(t, serverConversation.Valid()) + }) + } +}