Authentication provider for OAuth 2.0

- based on cloud-cli @ bc645b16ca7b7474b132ee1da8b56da35025a616

go.mod

@@ -3,18 +3,23 @@ module github.com/apache/pulsar-client-go
 go 1.12
 
 require (
-	github.com/DataDog/zstd v1.4.6-0.20200617134701-89f69fb7df32
+	github.com/DataDog/zstd v1.4.5
+	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/klauspost/compress v1.10.5
+	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/sirupsen/logrus v1.4.1
 	github.com/spaolacci/murmur3 v1.1.0
 	github.com/spf13/cobra v0.0.3
 	github.com/spf13/pflag v1.0.3 // indirect
 	github.com/stretchr/testify v1.4.0
 	github.com/yahoo/athenz v1.8.55
+	k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19
 )
+
+replace github.com/apache/pulsar-client-go/oauth2 => ./oauth2

oauth2/auth.go

@@ -0,0 +1,96 @@
+// Copyright (c) 2020 StreamNative, Inc.. All Rights Reserved.
+
+package auth
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/dgrijalva/jwt-go"
+	"golang.org/x/oauth2"
+	"k8s.io/utils/clock"
+)
+
+const (
+	ClaimNameUserName = "https://streamnative.io/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() (*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"`
+
+	// ClientCredentials is credentials data for the client credentials grant type
+	ClientCredentials *KeyFile `json:"client_credentials,omitempty"`
+
+	// 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,44 @@
+// Copyright (c) 2020 StreamNative, Inc.. All Rights Reserved.
+
+package auth
+
+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
+}