Merge pull request #67359 from mikedanese/reloadtoken

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions here: https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md.

client: periodically reload InClusterConfig token

/sig auth
/sig api-machinery

```release-note
NONE
```

Kubernetes-commit: 7b6647a418c660f2c87f183f706b297f1cb573ca

vendor/k8s.io/client-go/rest/config.go

@@ -18,6 +18,7 @@ package rest
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net"
@@ -29,7 +30,6 @@ import (
 	"time"
 
 	"github.com/golang/glog"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -44,6 +44,8 @@ const (
 	DefaultBurst int     = 10
 )
 
+var ErrNotInCluster = errors.New("unable to load in-cluster configuration, KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT must be defined")
+
 // Config holds the common attributes that can be passed to a Kubernetes client on
 // initialization.
 type Config struct {
@@ -308,20 +310,26 @@ func DefaultKubernetesUserAgent() string {
 
 // InClusterConfig returns a config object which uses the service account
 // kubernetes gives to pods. It's intended for clients that expect to be
-// running inside a pod running on kubernetes. It will return an error if
-// called from a process not running in a kubernetes environment.
+// running inside a pod running on kubernetes. It will return ErrNotInCluster
+// if called from a process not running in a kubernetes environment.
 func InClusterConfig() (*Config, error) {
+	const (
+		tokenFile  = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+		rootCAFile = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+	)
 	host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
 	if len(host) == 0 || len(port) == 0 {
-		return nil, fmt.Errorf("unable to load in-cluster configuration, KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT must be defined")
+		return nil, ErrNotInCluster
 	}
 
-	token, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token")
-	if err != nil {
+	ts := newCachedPathTokenSource(tokenFile)
+
+	if _, err := ts.Token(); err != nil {
 		return nil, err
 	}
+
 	tlsClientConfig := TLSClientConfig{}
-	rootCAFile := "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+
 	if _, err := certutil.NewPool(rootCAFile); err != nil {
 		glog.Errorf("Expected to load root CA config from %s, but got err: %v", rootCAFile, err)
 	} else {
@@ -331,8 +339,8 @@ func InClusterConfig() (*Config, error) {
 	return &Config{
 		// TODO: switch to using cluster DNS.
 		Host:            "https://" + net.JoinHostPort(host, port),
-		BearerToken:     string(token),
 		TLSClientConfig: tlsClientConfig,
+		WrapTransport:   TokenSourceWrapTransport(ts),
 	}, nil
 }
 

vendor/k8s.io/client-go/rest/token_source.go

@@ -0,0 +1,138 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed 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 rest
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/golang/glog"
+	"golang.org/x/oauth2"
+)
+
+// TokenSourceWrapTransport returns a WrapTransport that injects bearer tokens
+// authentication from an oauth2.TokenSource.
+func TokenSourceWrapTransport(ts oauth2.TokenSource) func(http.RoundTripper) http.RoundTripper {
+	return func(rt http.RoundTripper) http.RoundTripper {
+		return &tokenSourceTransport{
+			base: rt,
+			ort: &oauth2.Transport{
+				Source: ts,
+				Base:   rt,
+			},
+		}
+	}
+}
+
+func newCachedPathTokenSource(path string) oauth2.TokenSource {
+	return &cachingTokenSource{
+		now:    time.Now,
+		leeway: 1 * time.Minute,
+		base: &fileTokenSource{
+			path: path,
+			// This period was picked because it is half of the minimum validity
+			// duration for a token provisioned by they TokenRequest API. This is
+			// unsophisticated and should induce rotation at a frequency that should
+			// work with the token volume source.
+			period: 5 * time.Minute,
+		},
+	}
+}
+
+type tokenSourceTransport struct {
+	base http.RoundTripper
+	ort  http.RoundTripper
+}
+
+func (tst *tokenSourceTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	// This is to allow --token to override other bearer token providers.
+	if req.Header.Get("Authorization") != "" {
+		return tst.base.RoundTrip(req)
+	}
+	return tst.ort.RoundTrip(req)
+}
+
+type fileTokenSource struct {
+	path   string
+	period time.Duration
+}
+
+var _ = oauth2.TokenSource(&fileTokenSource{})
+
+func (ts *fileTokenSource) Token() (*oauth2.Token, error) {
+	tokb, err := ioutil.ReadFile(ts.path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read token file %q: %v", ts.path, err)
+	}
+	tok := strings.TrimSpace(string(tokb))
+	if len(tok) == 0 {
+		return nil, fmt.Errorf("read empty token from file %q", ts.path)
+	}
+
+	return &oauth2.Token{
+		AccessToken: tok,
+		Expiry:      time.Now().Add(ts.period),
+	}, nil
+}
+
+type cachingTokenSource struct {
+	base   oauth2.TokenSource
+	leeway time.Duration
+
+	sync.RWMutex
+	tok *oauth2.Token
+
+	// for testing
+	now func() time.Time
+}
+
+var _ = oauth2.TokenSource(&cachingTokenSource{})
+
+func (ts *cachingTokenSource) Token() (*oauth2.Token, error) {
+	now := ts.now()
+	// fast path
+	ts.RLock()
+	tok := ts.tok
+	ts.RUnlock()
+
+	if tok != nil && tok.Expiry.Add(-1*ts.leeway).After(now) {
+		return tok, nil
+	}
+
+	// slow path
+	ts.Lock()
+	defer ts.Unlock()
+	if tok := ts.tok; tok != nil && tok.Expiry.Add(-1*ts.leeway).After(now) {
+		return tok, nil
+	}
+
+	tok, err := ts.base.Token()
+	if err != nil {
+		if ts.tok == nil {
+			return nil, err
+		}
+		glog.Errorf("Unable to rotate token: %v", err)
+		return ts.tok, nil
+	}
+
+	ts.tok = tok
+	return tok, nil
+}

vendor/k8s.io/client-go/tools/cache/listwatch.go

@@ -18,7 +18,6 @@ package cache
 
 import (
 	"context"
-	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -91,13 +90,6 @@ func NewFilteredListWatchFromClient(c Getter, resource string, namespace string,
 	return &ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
 }
 
-func timeoutFromListOptions(options metav1.ListOptions) time.Duration {
-	if options.TimeoutSeconds != nil {
-		return time.Duration(*options.TimeoutSeconds) * time.Second
-	}
-	return 0
-}
-
 // List a set of apiserver resources
 func (lw *ListWatch) List(options metav1.ListOptions) (runtime.Object, error) {
 	if !lw.DisableChunking {

vendor/k8s.io/client-go/util/cert/cert.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"crypto/ecdsa"
 	"crypto/elliptic"
+	"crypto/rand"
 	cryptorand "crypto/rand"
 	"crypto/rsa"
 	"crypto/x509"
@@ -87,7 +88,7 @@ func NewSelfSignedCACert(cfg Config, key *rsa.PrivateKey) (*x509.Certificate, er
 
 // NewSignedCert creates a signed certificate using the given CA certificate and key
 func NewSignedCert(cfg Config, key *rsa.PrivateKey, caCert *x509.Certificate, caKey *rsa.PrivateKey) (*x509.Certificate, error) {
-	serial, err := cryptorand.Int(cryptorand.Reader, new(big.Int).SetInt64(math.MaxInt64))
+	serial, err := rand.Int(rand.Reader, new(big.Int).SetInt64(math.MaxInt64))
 	if err != nil {
 		return nil, err
 	}