[Fix] Oauth2 Client credentials flow use scopes from the keyfile as w…

…ell (#1244)

### Motivation

As the issue shows when using Pulsarctl which works with a context as configuration style, Oauth2 is used under the hood.

However the library does not expose any way to inject the context configuration `scope` value, and then relies on reading that from the `keyFile` for `client_credentials flow`.

However that is not being utilized in the current code as the scope value is not read from from the file.

*Explain here the context, and why you're making that change. What is the problem you're trying to solve.*
To allow the usage of oauth2 with a keyfile in Pulsarctl 3+ which it doesn't right now

### Modifications
Alters so reading the keyFile for client credentials also returns the `scope` value.

The value is then split on spaces, and added to a temporary slice.
After that the value of any additionalScopes that was already in the options is then added to the temp slice as well.
In the end the additionalScopes property on the options is set to the temp slice containing values from previous additionalScopes and the ones from the keyFile.

(cherry picked from commit c3b0633)

oauth2/client_credentials_flow.go

@@ -20,6 +20,8 @@ package oauth2
 import (
 	"net/http"
 
+	"strings"
+
 	"github.com/apache/pulsar-client-go/oauth2/clock"
 
 	"github.com/pkg/errors"
@@ -68,7 +70,6 @@ func newClientCredentialsFlow(
 // NewDefaultClientCredentialsFlow provides an easy way to build up a default
 // client credentials flow with all the correct configuration.
 func NewDefaultClientCredentialsFlow(options ClientCredentialsFlowOptions) (*ClientCredentialsFlow, error) {
-
 	credsProvider := NewClientCredentialsProviderFromKeyFile(options.KeyFile)
 	keyFile, err := credsProvider.GetClientCredentials()
 	if err != nil {
@@ -81,7 +82,6 @@ func NewDefaultClientCredentialsFlow(options ClientCredentialsFlowOptions) (*Cli
 	}
 
 	tokenRetriever := NewTokenRetriever(&http.Client{})
-
 	return newClientCredentialsFlow(
 		options,
 		keyFile,
@@ -94,13 +94,25 @@ var _ Flow = &ClientCredentialsFlow{}
 
 func (c *ClientCredentialsFlow) Authorize(audience string) (*AuthorizationGrant, error) {
 	var err error
+
+	// Merge the scopes of the options AdditionalScopes with the scopes read from the keyFile config
+	var scopesToAdd []string
+	if len(c.options.AdditionalScopes) > 0 {
+		scopesToAdd = append(scopesToAdd, c.options.AdditionalScopes...)
+	}
+
+	if c.keyfile.Scope != "" {
+		scopesSplit := strings.Split(c.keyfile.Scope, " ")
+		scopesToAdd = append(scopesToAdd, scopesSplit...)
+	}
+
 	grant := &AuthorizationGrant{
 		Type:              GrantTypeClientCredentials,
 		Audience:          audience,
 		ClientID:          c.keyfile.ClientID,
 		ClientCredentials: c.keyfile,
 		TokenEndpoint:     c.oidcWellKnownEndpoints.TokenEndpoint,
-		Scopes:            c.options.AdditionalScopes,
+		Scopes:            scopesToAdd,
 	}
 
 	// test the credentials and obtain an initial access token

oauth2/client_credentials_flow_test.go

@@ -47,6 +47,7 @@ var clientCredentials = KeyFile{
 	ClientSecret: "test_clientSecret",
 	ClientEmail:  "test_clientEmail",
 	IssuerURL:    "http://issuer",
+	Scope:        "test_scope",
 }
 
 var _ = Describe("ClientCredentialsFlow", func() {
@@ -64,9 +65,11 @@ var _ = Describe("ClientCredentialsFlow", func() {
 		})
 
 		It("invokes TokenExchanger with credentials", func() {
+			additionalScope := "additional_scope"
 			provider := newClientCredentialsFlow(
 				ClientCredentialsFlowOptions{
-					KeyFile: "test_keyfile",
+					KeyFile:          "test_keyfile",
+					AdditionalScopes: []string{additionalScope},
 				},
 				&clientCredentials,
 				oidcEndpoints,
@@ -81,6 +84,7 @@ var _ = Describe("ClientCredentialsFlow", func() {
 				ClientID:      clientCredentials.ClientID,
 				ClientSecret:  clientCredentials.ClientSecret,
 				Audience:      "test_audience",
+				Scopes:        []string{additionalScope, clientCredentials.Scope},
 			}))
 		})
 

oauth2/client_credentials_provider.go

@@ -38,6 +38,7 @@ type KeyFile struct {
 	ClientSecret string `json:"client_secret"`
 	ClientEmail  string `json:"client_email"`
 	IssuerURL    string `json:"issuer_url"`
+	Scope        string `json:"scope"`
 }
 
 func NewClientCredentialsProviderFromKeyFile(keyFile string) *KeyFileProvider {

oauth2/client_credentials_provider_test.go

@@ -34,12 +34,14 @@ func TestNewClientCredentialsProviderFromKeyFile(t *testing.T) {
 	ClientSecret := "CLIENT_SECRET"
 	ClientEmail := "CLIENT_EMAIL"
 	IssuerURL := "ISSUER_URL"
+	Scope := "SCOPE"
 	keyFile := &KeyFile{
 		Type:         oauthType,
 		ClientID:     clientID,
 		ClientSecret: ClientSecret,
 		ClientEmail:  ClientEmail,
 		IssuerURL:    IssuerURL,
+		Scope:        Scope,
 	}
 
 	b, err := json.Marshal(keyFile)