Merge pull request #46694 from matt-tyler/bug-gcpauthdefaultcred

Automatic merge from submit-queue (batch tested with PRs 34515, 47236, 46694, 47819, 47792)

Clear auth config when gcp app default credentials fail

**What this PR does / why we need it**:

Specific use case is when utilizing multiple gcp accounts, the user may provide credentials for the wrong account.

This change ensures the incorrect credentials are not cached in auth config, and logs an appropriate message.

**Which issue this PR fixes** : fixes #38075

**Special notes for your reviewer**:

**Release note**:

```release-note
Tokens retrieved from Google Cloud with application default credentials will not be cached if the client fails authorization
```

staging/src/k8s.io/client-go/plugin/pkg/client/auth/gcp/gcp.go

@@ -124,7 +124,7 @@ func newGCPAuthProvider(_ string, gcpConfig map[string]string, persister restcli
 }
 
 func (g *gcpAuthProvider) WrapTransport(rt http.RoundTripper) http.RoundTripper {
-	return &conditionalTransport{&oauth2.Transport{Source: g.tokenSource, Base: rt}}
+	return &conditionalTransport{&oauth2.Transport{Source: g.tokenSource, Base: rt}, g.persister}
 }
 
 func (g *gcpAuthProvider) Login() error { return nil }
@@ -284,11 +284,25 @@ func parseJSONPath(input interface{}, name, template string) (string, error) {
 
 type conditionalTransport struct {
 	oauthTransport *oauth2.Transport
+	persister      restclient.AuthProviderConfigPersister
 }
 
 func (t *conditionalTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 	if len(req.Header.Get("Authorization")) != 0 {
 		return t.oauthTransport.Base.RoundTrip(req)
 	}
-	return t.oauthTransport.RoundTrip(req)
+
+	res, err := t.oauthTransport.RoundTrip(req)
+
+	if err != nil {
+		return nil, err
+	}
+
+	if res.StatusCode == 401 {
+		glog.V(4).Infof("The credentials that were supplied are invalid for the target cluster")
+		emptyCache := make(map[string]string)
+		t.persister.Persist(emptyCache)
+	}
+
+	return res, nil
 }

staging/src/k8s.io/client-go/plugin/pkg/client/auth/gcp/gcp_test.go

@@ -18,6 +18,7 @@ package gcp
 
 import (
 	"fmt"
+	"net/http"
 	"os"
 	"os/exec"
 	"reflect"
@@ -323,3 +324,65 @@ func TestCachedTokenSource(t *testing.T) {
 		t.Errorf("got cache %v, want %v", got, cache)
 	}
 }
+
+type MockTransport struct {
+	res *http.Response
+}
+
+func (t *MockTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	return t.res, nil
+}
+
+func TestClearingCredentials(t *testing.T) {
+
+	fakeExpiry := time.Now().Add(time.Hour)
+
+	cache := map[string]string{
+		"access-token": "fakeToken",
+		"expiry":       fakeExpiry.String(),
+	}
+
+	cts := cachedTokenSource{
+		source:      nil,
+		accessToken: cache["access-token"],
+		expiry:      fakeExpiry,
+		persister:   nil,
+		cache:       nil,
+	}
+
+	tests := []struct {
+		name  string
+		res   http.Response
+		cache map[string]string
+	}{
+		{
+			"Unauthorized",
+			http.Response{StatusCode: 401},
+			make(map[string]string),
+		},
+		{
+			"Authorized",
+			http.Response{StatusCode: 200},
+			cache,
+		},
+	}
+
+	persister := &fakePersister{}
+	req := http.Request{Header: http.Header{}}
+
+	for _, tc := range tests {
+		authProvider := gcpAuthProvider{&cts, persister}
+
+		fakeTransport := MockTransport{&tc.res}
+
+		transport := (authProvider.WrapTransport(&fakeTransport))
+		persister.Persist(cache)
+
+		transport.RoundTrip(&req)
+
+		if got := persister.read(); !reflect.DeepEqual(got, tc.cache) {
+			t.Errorf("got cache %v, want %v", got, tc.cache)
+		}
+	}
+
+}