Skip to content

Commit 8d530db

Browse files
yangpei1214yangpei1214
authored
[DSG-8949] support Schema Registry bearer auth with static token (#1103)
1 parent ba5aa40 commit 8d530db

File tree

6 files changed

+189
-16
lines changed

6 files changed

+189
-16
lines changed

‎examples/confluent_cloud_example/confluent_cloud_example.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ func main() {
8989
panic(fmt.Sprintf("Failed to create producer: %s", err))
9090
}
9191

92-
client, err := schemaregistry.NewClient(schemaregistry.NewConfigWithAuthentication(
92+
client, err := schemaregistry.NewClient(schemaregistry.NewConfigWithBasicAuthentication(
9393
schemaRegistryAPIEndpoint,
9494
schemaRegistryAPIKey,
9595
schemaRegistryAPISecret))

‎schemaregistry/config.go‎

Lines changed: 36 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,15 @@ type Config struct {
3838
// SaslUsername specifies the password for SASL.
3939
SaslPassword string
4040

41+
// BearerAuthToken specifies the token for authentication.
42+
BearerAuthToken string
43+
// BearerAuthCredentialsSource specifies how to determine the credentials.
44+
BearerAuthCredentialsSource string
45+
// BearerAuthLogicalCluster specifies the target SR logical cluster id. It is required for Confluent Cloud Schema Registry
46+
BearerAuthLogicalCluster string
47+
// BearerAuthIdentityPoolID specifies the identity pool ID. It is required for Confluent Cloud Schema Registry
48+
BearerAuthIdentityPoolID string
49+
4150
// SslCertificateLocation specifies the location of SSL certificates.
4251
SslCertificateLocation string
4352
// SslKeyLocation specifies the location of SSL keys.
@@ -64,9 +73,6 @@ func NewConfig(url string) *Config {
6473

6574
c.SchemaRegistryURL = url
6675

67-
c.BasicAuthUserInfo = ""
68-
c.BasicAuthCredentialsSource = "URL"
69-
7076
c.SaslMechanism = "GSSAPI"
7177
c.SaslUsername = ""
7278
c.SaslPassword = ""
@@ -84,6 +90,7 @@ func NewConfig(url string) *Config {
8490

8591
// NewConfigWithAuthentication returns a new configuration instance using basic authentication.
8692
// For Confluent Cloud, use the API key for the username and the API secret for the password.
93+
// This method is deprecated.
8794
func NewConfigWithAuthentication(url string, username string, password string) *Config {
8895
c := NewConfig(url)
8996

@@ -92,3 +99,29 @@ func NewConfigWithAuthentication(url string, username string, password string) *
9299

93100
return c
94101
}
102+
103+
// NewConfigWithBasicAuthentication returns a new configuration instance using basic authentication.
104+
// For Confluent Cloud, use the API key for the username and the API secret for the password.
105+
func NewConfigWithBasicAuthentication(url string, username string, password string) *Config {
106+
c := NewConfig(url)
107+
108+
c.BasicAuthUserInfo = fmt.Sprintf("%s:%s", username, password)
109+
c.BasicAuthCredentialsSource = "USER_INFO"
110+
111+
return c
112+
}
113+
114+
// NewConfigWithBearerAuthentication returns a new configuration instance using bearer authentication.
115+
// For Confluent Cloud, targetSr(`bearer.auth.logical.cluster` and
116+
// identityPoolID(`bearer.auth.identity.pool.id`) is required
117+
func NewConfigWithBearerAuthentication(url, token, targetSr, identityPoolID string) *Config {
118+
119+
c := NewConfig(url)
120+
121+
c.BearerAuthToken = token
122+
c.BearerAuthCredentialsSource = "STATIC_TOKEN"
123+
c.BearerAuthLogicalCluster = targetSr
124+
c.BearerAuthIdentityPoolID = identityPoolID
125+
126+
return c
127+
}

‎schemaregistry/config_test.go‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,16 @@ import (
2323
func TestConfigWithAuthentication(t *testing.T) {
2424
maybeFail = initFailFunc(t)
2525

26-
c := NewConfigWithAuthentication("mock://", "username", "password")
26+
c := NewConfigWithBasicAuthentication("mock://", "username", "password")
2727

2828
maybeFail("BasicAuthCredentialsSource", expect(c.BasicAuthCredentialsSource, "USER_INFO"))
2929
maybeFail("BasicAuthUserInfo", expect(c.BasicAuthUserInfo, "username:password"))
3030
}
31+
32+
func TestConfigWithBearerAuth(t *testing.T) {
33+
maybeFail = initFailFunc(t)
34+
c := NewConfigWithBearerAuthentication("mock://", "token", "lsrc-123", "poolID")
35+
maybeFail("BearerAuthCredentialsSource", expect(c.BearerAuthCredentialsSource, "STATIC_TOKEN"))
36+
maybeFail("BearerAuthLogicalCluster", expect(c.BearerAuthLogicalCluster, "lsrc-123"))
37+
maybeFail("BearerAuthIdentityPoolID", expect(c.BearerAuthIdentityPoolID, "poolID"))
38+
}

‎schemaregistry/rest_service.go‎

Lines changed: 57 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,9 @@ const (
5454
subjectConfig = config + "/%s"
5555
mode = "/mode"
5656
modeConfig = mode + "/%s"
57+
58+
targetSRClusterKey = "Target-Sr-Cluster"
59+
targetIdentityPoolIDKey = "Confluent-Identity-Pool-Id"
5760
)
5861

5962
// REST API request
@@ -251,28 +254,71 @@ func configureUSERINFOAuth(conf *Config, header http.Header) error {
251254

252255
}
253256

257+
func configureStaticTokenAuth(conf *Config, header http.Header) error {
258+
bearerToken := conf.BearerAuthToken
259+
if len(bearerToken) == 0 {
260+
return fmt.Errorf("config bearer.auth.token must be specified when bearer.auth.credentials.source is" +
261+
" specified with STATIC_TOKEN")
262+
}
263+
header.Add("Authorization", fmt.Sprintf("Bearer %s", bearerToken))
264+
setBearerAuthExtraHeaders(conf, header)
265+
return nil
266+
}
267+
268+
func setBearerAuthExtraHeaders(conf *Config, header http.Header) {
269+
targetIdentityPoolID := conf.BearerAuthIdentityPoolID
270+
if len(targetIdentityPoolID) > 0 {
271+
header.Add(targetIdentityPoolIDKey, targetIdentityPoolID)
272+
}
273+
274+
targetSr := conf.BearerAuthLogicalCluster
275+
if len(targetSr) > 0 {
276+
header.Add(targetSRClusterKey, targetSr)
277+
}
278+
}
279+
254280
// newAuthHeader returns a base64 encoded userinfo string identified on the configured credentials source
255281
func newAuthHeader(service *url.URL, conf *Config) (http.Header, error) {
256282
// Remove userinfo from url regardless of source to avoid confusion/conflicts
257283
defer func() {
258284
service.User = nil
259285
}()
260286

261-
source := conf.BasicAuthCredentialsSource
262-
263287
header := http.Header{}
264288

289+
basicSource := conf.BasicAuthCredentialsSource
290+
bearerSource := conf.BearerAuthCredentialsSource
291+
265292
var err error
266-
switch strings.ToUpper(source) {
267-
case "URL":
268-
err = configureURLAuth(service, header)
269-
case "SASL_INHERIT":
270-
err = configureSASLAuth(conf, header)
271-
case "USER_INFO":
272-
err = configureUSERINFOAuth(conf, header)
273-
default:
274-
err = fmt.Errorf("unrecognized value for basic.auth.credentials.source %s", source)
293+
if len(basicSource) != 0 && len(bearerSource) != 0 {
294+
return header, fmt.Errorf("only one of basic.auth.credentials.source or bearer.auth.credentials.source" +
295+
" may be specified")
296+
} else if len(basicSource) != 0 {
297+
switch strings.ToUpper(basicSource) {
298+
case "URL":
299+
err = configureURLAuth(service, header)
300+
case "SASL_INHERIT":
301+
err = configureSASLAuth(conf, header)
302+
case "USER_INFO":
303+
err = configureUSERINFOAuth(conf, header)
304+
default:
305+
err = fmt.Errorf("unrecognized value for basic.auth.credentials.source %s", basicSource)
306+
}
307+
} else if len(bearerSource) != 0 {
308+
switch strings.ToUpper(bearerSource) {
309+
case "STATIC_TOKEN":
310+
err = configureStaticTokenAuth(conf, header)
311+
//case "OAUTHBEARER":
312+
// err = configureOauthBearerAuth(conf, header)
313+
//case "SASL_OAUTHBEARER_INHERIT":
314+
// err = configureSASLOauth()
315+
//case "CUSTOM":
316+
// err = configureCustomOauth(conf, header)
317+
default:
318+
err = fmt.Errorf("unrecognized value for bearer.auth.credentials.source %s", bearerSource)
319+
}
275320
}
321+
276322
return header, err
277323
}
278324

‎schemaregistry/rest_service_test.go‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ package schemaregistry
1818

1919
import (
2020
"crypto/tls"
21+
"net/url"
2122
"strings"
2223
"testing"
2324
)
@@ -83,3 +84,87 @@ func TestConfigureTLS(t *testing.T) {
8384
t.Errorf("Should work with valid CA, certificate and key, got %s", err)
8485
}
8586
}
87+
88+
func TestNewAuthHeader(t *testing.T) {
89+
url, err := url.Parse("mock://")
90+
if err != nil {
91+
t.Errorf("Should work with empty config, got %s", err)
92+
}
93+
94+
config := &Config{}
95+
96+
config.BearerAuthCredentialsSource = "STATIC_TOKEN"
97+
config.BasicAuthCredentialsSource = "URL"
98+
99+
_, err = newAuthHeader(url, config)
100+
if err == nil {
101+
t.Errorf("Should not work with both basic auth source and bearer auth source")
102+
}
103+
104+
// testing bearer auth
105+
config.BasicAuthCredentialsSource = ""
106+
_, err = newAuthHeader(url, config)
107+
if err == nil {
108+
t.Errorf("Should not work if bearer auth token is empty")
109+
}
110+
111+
config.BearerAuthToken = "token"
112+
config.BearerAuthLogicalCluster = "lsrc-123"
113+
config.BearerAuthIdentityPoolID = "poolID"
114+
headers, err := newAuthHeader(url, config)
115+
if err != nil {
116+
t.Errorf("Should work with bearer auth token, got %s", err)
117+
} else {
118+
if val, exists := headers["Authorization"]; !exists || len(val) == 0 ||
119+
!strings.EqualFold(val[0], "Bearer token") {
120+
t.Errorf("Should have header with key Authorization")
121+
}
122+
if val, exists := headers[targetIdentityPoolIDKey]; !exists || len(val) == 0 ||
123+
!strings.EqualFold(val[0], "poolID") {
124+
t.Errorf("Should have header with key Confluent-Identity-Pool-Id")
125+
}
126+
if val, exists := headers[targetSRClusterKey]; !exists || len(val) == 0 ||
127+
!strings.EqualFold(val[0], "lsrc-123") {
128+
t.Errorf("Should have header with key Target-Sr-Cluster")
129+
}
130+
}
131+
132+
config.BearerAuthCredentialsSource = "other"
133+
_, err = newAuthHeader(url, config)
134+
if err == nil {
135+
t.Errorf("Should not work if bearer auth source is invalid")
136+
}
137+
138+
// testing basic auth
139+
config.BearerAuthCredentialsSource = ""
140+
config.BasicAuthCredentialsSource = "USER_INFO"
141+
config.BasicAuthUserInfo = "username:password"
142+
_, err = newAuthHeader(url, config)
143+
if err != nil {
144+
t.Errorf("Should work with basic auth token, got %s", err)
145+
}
146+
147+
config.BasicAuthCredentialsSource = "URL"
148+
_, err = newAuthHeader(url, config)
149+
if err != nil {
150+
t.Errorf("Should work with basic auth token, got %s", err)
151+
} else if val, exists := headers["Authorization"]; !exists || len(val) == 0 {
152+
t.Errorf("Should have header with key Authorization")
153+
}
154+
155+
config.BasicAuthCredentialsSource = "SASL_INHERIT"
156+
config.SaslUsername = "username"
157+
config.SaslPassword = "password"
158+
_, err = newAuthHeader(url, config)
159+
if err != nil {
160+
t.Errorf("Should work with basic auth token, got %s", err)
161+
} else if val, exists := headers["Authorization"]; !exists || len(val) == 0 {
162+
t.Errorf("Should have header with key Authorization")
163+
}
164+
165+
config.BasicAuthCredentialsSource = "other"
166+
_, err = newAuthHeader(url, config)
167+
if err == nil {
168+
t.Errorf("Should not work if basic auth source is invalid")
169+
}
170+
}

‎schemaregistry/schemaregistry_client.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -218,6 +218,7 @@ type Client interface {
218218
func NewClient(conf *Config) (Client, error) {
219219

220220
urlConf := conf.SchemaRegistryURL
221+
// for testing
221222
if strings.HasPrefix(urlConf, "mock://") {
222223
url, err := url.Parse(urlConf)
223224
if err != nil {

0 commit comments

Comments
 (0)