Support oauth2 authentication for pulsar-client-go

go.mod

@@ -1,17 +1,18 @@
 module github.com/apache/pulsar-client-go
 
-go 1.12
+go 1.13
 
 require (
 	github.com/DataDog/zstd v1.4.6-0.20200617134701-89f69fb7df32
+	github.com/apache/pulsar-client-go/oauth2 v0.0.0-00010101000000-000000000000
 	github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6
 	github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b
 	github.com/gogo/protobuf v1.3.1
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/klauspost/compress v1.10.8
 	github.com/kr/pretty v0.2.0 // indirect
 	github.com/pierrec/lz4 v2.0.5+incompatible
-	github.com/pkg/errors v0.8.1
+	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.7.1
 	github.com/sirupsen/logrus v1.4.2
 	github.com/spaolacci/murmur3 v1.1.0
@@ -20,3 +21,5 @@ require (
 	github.com/stretchr/testify v1.4.0
 	github.com/yahoo/athenz v1.8.55
 )
+
+replace github.com/apache/pulsar-client-go/oauth2 => ./oauth2

oauth2/auth.go

@@ -0,0 +1,120 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package oauth2
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/apache/pulsar-client-go/oauth2/clock"
+	"github.com/dgrijalva/jwt-go"
+	"golang.org/x/oauth2"
+)
+
+const (
+	ClaimNameUserName = "https://pulsar.apache.org/username"
+)
+
+// Flow abstracts an OAuth 2.0 authentication and authorization flow
+type Flow interface {
+	// Authorize obtains an authorization grant based on an OAuth 2.0 authorization flow.
+	// The method returns a grant which may contain an initial access token.
+	Authorize(audience string) (*AuthorizationGrant, error)
+}
+
+// AuthorizationGrantRefresher refreshes OAuth 2.0 authorization grant
+type AuthorizationGrantRefresher interface {
+	// Refresh refreshes an authorization grant to contain a fresh access token
+	Refresh(grant *AuthorizationGrant) (*AuthorizationGrant, error)
+}
+
+type AuthorizationGrantType string
+
+const (
+	// GrantTypeClientCredentials represents a client credentials grant
+	GrantTypeClientCredentials AuthorizationGrantType = "client_credentials"
+
+	// GrantTypeDeviceCode represents a device code grant
+	GrantTypeDeviceCode AuthorizationGrantType = "device_code"
+)
+
+// AuthorizationGrant is a credential representing the resource owner's authorization
+// to access its protected resources, and is used by the client to obtain an access token
+type AuthorizationGrant struct {
+	// Type describes the type of authorization grant represented by this structure
+	Type AuthorizationGrantType `json:"type"`
+
+	// Audience is the intended audience of the access tokens
+	Audience string `json:"audience,omitempty"`
+
+	// ClientID is an OAuth2 client identifier used by some flows
+	ClientID string `json:"client_id,omitempty"`
+
+	// ClientCredentials is credentials data for the client credentials grant type
+	ClientCredentials *KeyFile `json:"client_credentials,omitempty"`
+
+	// the token endpoint
+	TokenEndpoint string `json:"token_endpoint"`
+
+	// Token contains an access token in the client credentials grant type,
+	// and a refresh token in the device authorization grant type
+	Token *oauth2.Token `json:"token,omitempty"`
+}
+
+// TokenResult holds token information
+type TokenResult struct {
+	AccessToken  string `json:"access_token"`
+	IDToken      string `json:"id_token"`
+	RefreshToken string `json:"refresh_token"`
+	ExpiresIn    int    `json:"expires_in"`
+}
+
+// Issuer holds information about the issuer of tokens
+type Issuer struct {
+	IssuerEndpoint string
+	ClientID       string
+	Audience       string
+}
+
+func convertToOAuth2Token(token *TokenResult, clock clock.Clock) oauth2.Token {
+	return oauth2.Token{
+		AccessToken:  token.AccessToken,
+		TokenType:    "bearer",
+		RefreshToken: token.RefreshToken,
+		Expiry:       clock.Now().Add(time.Duration(token.ExpiresIn) * time.Second),
+	}
+}
+
+// ExtractUserName extracts the username claim from an authorization grant
+func ExtractUserName(token oauth2.Token) (string, error) {
+	p := jwt.Parser{}
+	claims := jwt.MapClaims{}
+	if _, _, err := p.ParseUnverified(token.AccessToken, claims); err != nil {
+		return "", fmt.Errorf("unable to decode the access token: %v", err)
+	}
+	username, ok := claims[ClaimNameUserName]
+	if !ok {
+		return "", fmt.Errorf("access token doesn't contain a username claim")
+	}
+	switch v := username.(type) {
+	case string:
+		return v, nil
+	default:
+		return "", fmt.Errorf("access token contains an unsupported username claim")
+	}
+}

oauth2/auth_suite_test.go

@@ -0,0 +1,65 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package oauth2
+
+import (
+	"context"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestAuth(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "cloud-cli Auth Suite")
+}
+
+type MockTokenExchanger struct {
+	CalledWithRequest        interface{}
+	ReturnsTokens            *TokenResult
+	ReturnsError             error
+	RefreshCalledWithRequest *RefreshTokenExchangeRequest
+}
+
+func (te *MockTokenExchanger) ExchangeCode(req AuthorizationCodeExchangeRequest) (*TokenResult, error) {
+	te.CalledWithRequest = &req
+	return te.ReturnsTokens, te.ReturnsError
+}
+
+func (te *MockTokenExchanger) ExchangeRefreshToken(req RefreshTokenExchangeRequest) (*TokenResult, error) {
+	te.RefreshCalledWithRequest = &req
+	return te.ReturnsTokens, te.ReturnsError
+}
+
+func (te *MockTokenExchanger) ExchangeClientCredentials(req ClientCredentialsExchangeRequest) (*TokenResult, error) {
+	te.CalledWithRequest = &req
+	return te.ReturnsTokens, te.ReturnsError
+}
+
+func (te *MockTokenExchanger) ExchangeDeviceCode(ctx context.Context,
+	req DeviceCodeExchangeRequest) (*TokenResult, error) {
+	te.CalledWithRequest = &req
+	return te.ReturnsTokens, te.ReturnsError
+}
+
+var oidcEndpoints = OIDCWellKnownEndpoints{
+	AuthorizationEndpoint:       "http://issuer/auth/authorize",
+	TokenEndpoint:               "http://issuer/auth/token",
+	DeviceAuthorizationEndpoint: "http://issuer/auth/authorize/device",
+}