(feat) Add auto-discovery in k8s | Adarsh
This commit is contained in:
Adarsh K Kumar
360
vendor/k8s.io/client-go/plugin/pkg/client/auth/exec/exec.go
generated
vendored
Normal file
360
vendor/k8s.io/client-go/plugin/pkg/client/auth/exec/exec.go
generated
vendored
Normal file
@ -0,0 +1,360 @@
|
||||
/*
|
||||
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 exec
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"reflect"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/davecgh/go-spew/spew"
|
||||
"golang.org/x/crypto/ssh/terminal"
|
||||
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/client-go/pkg/apis/clientauthentication"
|
||||
"k8s.io/client-go/pkg/apis/clientauthentication/v1alpha1"
|
||||
"k8s.io/client-go/pkg/apis/clientauthentication/v1beta1"
|
||||
"k8s.io/client-go/tools/clientcmd/api"
|
||||
"k8s.io/client-go/transport"
|
||||
"k8s.io/client-go/util/connrotation"
|
||||
"k8s.io/klog"
|
||||
)
|
||||
|
||||
const execInfoEnv = "KUBERNETES_EXEC_INFO"
|
||||
|
||||
var scheme = runtime.NewScheme()
|
||||
var codecs = serializer.NewCodecFactory(scheme)
|
||||
|
||||
func init() {
|
||||
v1.AddToGroupVersion(scheme, schema.GroupVersion{Version: "v1"})
|
||||
utilruntime.Must(v1alpha1.AddToScheme(scheme))
|
||||
utilruntime.Must(v1beta1.AddToScheme(scheme))
|
||||
utilruntime.Must(clientauthentication.AddToScheme(scheme))
|
||||
}
|
||||
|
||||
var (
|
||||
// Since transports can be constantly re-initialized by programs like kubectl,
|
||||
// keep a cache of initialized authenticators keyed by a hash of their config.
|
||||
globalCache = newCache()
|
||||
// The list of API versions we accept.
|
||||
apiVersions = map[string]schema.GroupVersion{
|
||||
v1alpha1.SchemeGroupVersion.String(): v1alpha1.SchemeGroupVersion,
|
||||
v1beta1.SchemeGroupVersion.String(): v1beta1.SchemeGroupVersion,
|
||||
}
|
||||
)
|
||||
|
||||
func newCache() *cache {
|
||||
return &cache{m: make(map[string]*Authenticator)}
|
||||
}
|
||||
|
||||
var spewConfig = &spew.ConfigState{DisableMethods: true, Indent: " "}
|
||||
|
||||
func cacheKey(c *api.ExecConfig) string {
|
||||
return spewConfig.Sprint(c)
|
||||
}
|
||||
|
||||
type cache struct {
|
||||
mu sync.Mutex
|
||||
m map[string]*Authenticator
|
||||
}
|
||||
|
||||
func (c *cache) get(s string) (*Authenticator, bool) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
a, ok := c.m[s]
|
||||
return a, ok
|
||||
}
|
||||
|
||||
// put inserts an authenticator into the cache. If an authenticator is already
|
||||
// associated with the key, the first one is returned instead.
|
||||
func (c *cache) put(s string, a *Authenticator) *Authenticator {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
existing, ok := c.m[s]
|
||||
if ok {
|
||||
return existing
|
||||
}
|
||||
c.m[s] = a
|
||||
return a
|
||||
}
|
||||
|
||||
// GetAuthenticator returns an exec-based plugin for providing client credentials.
|
||||
func GetAuthenticator(config *api.ExecConfig) (*Authenticator, error) {
|
||||
return newAuthenticator(globalCache, config)
|
||||
}
|
||||
|
||||
func newAuthenticator(c *cache, config *api.ExecConfig) (*Authenticator, error) {
|
||||
key := cacheKey(config)
|
||||
if a, ok := c.get(key); ok {
|
||||
return a, nil
|
||||
}
|
||||
|
||||
gv, ok := apiVersions[config.APIVersion]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("exec plugin: invalid apiVersion %q", config.APIVersion)
|
||||
}
|
||||
|
||||
a := &Authenticator{
|
||||
cmd: config.Command,
|
||||
args: config.Args,
|
||||
group: gv,
|
||||
|
||||
stdin: os.Stdin,
|
||||
stderr: os.Stderr,
|
||||
interactive: terminal.IsTerminal(int(os.Stdout.Fd())),
|
||||
now: time.Now,
|
||||
environ: os.Environ,
|
||||
}
|
||||
|
||||
for _, env := range config.Env {
|
||||
a.env = append(a.env, env.Name+"="+env.Value)
|
||||
}
|
||||
|
||||
return c.put(key, a), nil
|
||||
}
|
||||
|
||||
// Authenticator is a client credential provider that rotates credentials by executing a plugin.
|
||||
// The plugin input and output are defined by the API group client.authentication.k8s.io.
|
||||
type Authenticator struct {
|
||||
// Set by the config
|
||||
cmd string
|
||||
args []string
|
||||
group schema.GroupVersion
|
||||
env []string
|
||||
|
||||
// Stubbable for testing
|
||||
stdin io.Reader
|
||||
stderr io.Writer
|
||||
interactive bool
|
||||
now func() time.Time
|
||||
environ func() []string
|
||||
|
||||
// Cached results.
|
||||
//
|
||||
// The mutex also guards calling the plugin. Since the plugin could be
|
||||
// interactive we want to make sure it's only called once.
|
||||
mu sync.Mutex
|
||||
cachedCreds *credentials
|
||||
exp time.Time
|
||||
|
||||
onRotate func()
|
||||
}
|
||||
|
||||
type credentials struct {
|
||||
token string
|
||||
cert *tls.Certificate
|
||||
}
|
||||
|
||||
// UpdateTransportConfig updates the transport.Config to use credentials
|
||||
// returned by the plugin.
|
||||
func (a *Authenticator) UpdateTransportConfig(c *transport.Config) error {
|
||||
c.Wrap(func(rt http.RoundTripper) http.RoundTripper {
|
||||
return &roundTripper{a, rt}
|
||||
})
|
||||
|
||||
if c.TLS.GetCert != nil {
|
||||
return errors.New("can't add TLS certificate callback: transport.Config.TLS.GetCert already set")
|
||||
}
|
||||
c.TLS.GetCert = a.cert
|
||||
|
||||
var dial func(ctx context.Context, network, addr string) (net.Conn, error)
|
||||
if c.Dial != nil {
|
||||
dial = c.Dial
|
||||
} else {
|
||||
dial = (&net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}).DialContext
|
||||
}
|
||||
d := connrotation.NewDialer(dial)
|
||||
a.onRotate = d.CloseAll
|
||||
c.Dial = d.DialContext
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type roundTripper struct {
|
||||
a *Authenticator
|
||||
base http.RoundTripper
|
||||
}
|
||||
|
||||
func (r *roundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
// If a user has already set credentials, use that. This makes commands like
|
||||
// "kubectl get --token (token) pods" work.
|
||||
if req.Header.Get("Authorization") != "" {
|
||||
return r.base.RoundTrip(req)
|
||||
}
|
||||
|
||||
creds, err := r.a.getCreds()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("getting credentials: %v", err)
|
||||
}
|
||||
if creds.token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+creds.token)
|
||||
}
|
||||
|
||||
res, err := r.base.RoundTrip(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if res.StatusCode == http.StatusUnauthorized {
|
||||
resp := &clientauthentication.Response{
|
||||
Header: res.Header,
|
||||
Code: int32(res.StatusCode),
|
||||
}
|
||||
if err := r.a.maybeRefreshCreds(creds, resp); err != nil {
|
||||
klog.Errorf("refreshing credentials: %v", err)
|
||||
}
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (a *Authenticator) credsExpired() bool {
|
||||
if a.exp.IsZero() {
|
||||
return false
|
||||
}
|
||||
return a.now().After(a.exp)
|
||||
}
|
||||
|
||||
func (a *Authenticator) cert() (*tls.Certificate, error) {
|
||||
creds, err := a.getCreds()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return creds.cert, nil
|
||||
}
|
||||
|
||||
func (a *Authenticator) getCreds() (*credentials, error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
if a.cachedCreds != nil && !a.credsExpired() {
|
||||
return a.cachedCreds, nil
|
||||
}
|
||||
|
||||
if err := a.refreshCredsLocked(nil); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.cachedCreds, nil
|
||||
}
|
||||
|
||||
// maybeRefreshCreds executes the plugin to force a rotation of the
|
||||
// credentials, unless they were rotated already.
|
||||
func (a *Authenticator) maybeRefreshCreds(creds *credentials, r *clientauthentication.Response) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
// Since we're not making a new pointer to a.cachedCreds in getCreds, no
|
||||
// need to do deep comparison.
|
||||
if creds != a.cachedCreds {
|
||||
// Credentials already rotated.
|
||||
return nil
|
||||
}
|
||||
|
||||
return a.refreshCredsLocked(r)
|
||||
}
|
||||
|
||||
// refreshCredsLocked executes the plugin and reads the credentials from
|
||||
// stdout. It must be called while holding the Authenticator's mutex.
|
||||
func (a *Authenticator) refreshCredsLocked(r *clientauthentication.Response) error {
|
||||
cred := &clientauthentication.ExecCredential{
|
||||
Spec: clientauthentication.ExecCredentialSpec{
|
||||
Response: r,
|
||||
Interactive: a.interactive,
|
||||
},
|
||||
}
|
||||
|
||||
env := append(a.environ(), a.env...)
|
||||
if a.group == v1alpha1.SchemeGroupVersion {
|
||||
// Input spec disabled for beta due to lack of use. Possibly re-enable this later if
|
||||
// someone wants it back.
|
||||
//
|
||||
// See: https://github.com/kubernetes/kubernetes/issues/61796
|
||||
data, err := runtime.Encode(codecs.LegacyCodec(a.group), cred)
|
||||
if err != nil {
|
||||
return fmt.Errorf("encode ExecCredentials: %v", err)
|
||||
}
|
||||
env = append(env, fmt.Sprintf("%s=%s", execInfoEnv, data))
|
||||
}
|
||||
|
||||
stdout := &bytes.Buffer{}
|
||||
cmd := exec.Command(a.cmd, a.args...)
|
||||
cmd.Env = env
|
||||
cmd.Stderr = a.stderr
|
||||
cmd.Stdout = stdout
|
||||
if a.interactive {
|
||||
cmd.Stdin = a.stdin
|
||||
}
|
||||
|
||||
if err := cmd.Run(); err != nil {
|
||||
return fmt.Errorf("exec: %v", err)
|
||||
}
|
||||
|
||||
_, gvk, err := codecs.UniversalDecoder(a.group).Decode(stdout.Bytes(), nil, cred)
|
||||
if err != nil {
|
||||
return fmt.Errorf("decoding stdout: %v", err)
|
||||
}
|
||||
if gvk.Group != a.group.Group || gvk.Version != a.group.Version {
|
||||
return fmt.Errorf("exec plugin is configured to use API version %s, plugin returned version %s",
|
||||
a.group, schema.GroupVersion{Group: gvk.Group, Version: gvk.Version})
|
||||
}
|
||||
|
||||
if cred.Status == nil {
|
||||
return fmt.Errorf("exec plugin didn't return a status field")
|
||||
}
|
||||
if cred.Status.Token == "" && cred.Status.ClientCertificateData == "" && cred.Status.ClientKeyData == "" {
|
||||
return fmt.Errorf("exec plugin didn't return a token or cert/key pair")
|
||||
}
|
||||
if (cred.Status.ClientCertificateData == "") != (cred.Status.ClientKeyData == "") {
|
||||
return fmt.Errorf("exec plugin returned only certificate or key, not both")
|
||||
}
|
||||
|
||||
if cred.Status.ExpirationTimestamp != nil {
|
||||
a.exp = cred.Status.ExpirationTimestamp.Time
|
||||
} else {
|
||||
a.exp = time.Time{}
|
||||
}
|
||||
|
||||
newCreds := &credentials{
|
||||
token: cred.Status.Token,
|
||||
}
|
||||
if cred.Status.ClientKeyData != "" && cred.Status.ClientCertificateData != "" {
|
||||
cert, err := tls.X509KeyPair([]byte(cred.Status.ClientCertificateData), []byte(cred.Status.ClientKeyData))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed parsing client key/certificate: %v", err)
|
||||
}
|
||||
newCreds.cert = &cert
|
||||
}
|
||||
|
||||
oldCreds := a.cachedCreds
|
||||
a.cachedCreds = newCreds
|
||||
// Only close all connections when TLS cert rotates. Token rotation doesn't
|
||||
// need the extra noise.
|
||||
if a.onRotate != nil && oldCreds != nil && !reflect.DeepEqual(oldCreds.cert, a.cachedCreds.cert) {
|
||||
a.onRotate()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
8
vendor/k8s.io/client-go/plugin/pkg/client/auth/gcp/OWNERS
generated
vendored
Normal file
8
vendor/k8s.io/client-go/plugin/pkg/client/auth/gcp/OWNERS
generated
vendored
Normal file
@ -0,0 +1,8 @@
|
||||
# See the OWNERS docs at https://go.k8s.io/owners
|
||||
|
||||
approvers:
|
||||
- cjcullen
|
||||
- jlowdermilk
|
||||
reviewers:
|
||||
- cjcullen
|
||||
- jlowdermilk
|
||||
383
vendor/k8s.io/client-go/plugin/pkg/client/auth/gcp/gcp.go
generated
vendored
Normal file
383
vendor/k8s.io/client-go/plugin/pkg/client/auth/gcp/gcp.go
generated
vendored
Normal file
@ -0,0 +1,383 @@
|
||||
/*
|
||||
Copyright 2016 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 gcp
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/oauth2"
|
||||
"golang.org/x/oauth2/google"
|
||||
"k8s.io/apimachinery/pkg/util/net"
|
||||
"k8s.io/apimachinery/pkg/util/yaml"
|
||||
restclient "k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/util/jsonpath"
|
||||
"k8s.io/klog"
|
||||
)
|
||||
|
||||
func init() {
|
||||
if err := restclient.RegisterAuthProviderPlugin("gcp", newGCPAuthProvider); err != nil {
|
||||
klog.Fatalf("Failed to register gcp auth plugin: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
var (
|
||||
// Stubbable for testing
|
||||
execCommand = exec.Command
|
||||
|
||||
// defaultScopes:
|
||||
// - cloud-platform is the base scope to authenticate to GCP.
|
||||
// - userinfo.email is used to authenticate to GKE APIs with gserviceaccount
|
||||
// email instead of numeric uniqueID.
|
||||
defaultScopes = []string{
|
||||
"https://www.googleapis.com/auth/cloud-platform",
|
||||
"https://www.googleapis.com/auth/userinfo.email"}
|
||||
)
|
||||
|
||||
// gcpAuthProvider is an auth provider plugin that uses GCP credentials to provide
|
||||
// tokens for kubectl to authenticate itself to the apiserver. A sample json config
|
||||
// is provided below with all recognized options described.
|
||||
//
|
||||
// {
|
||||
// 'auth-provider': {
|
||||
// # Required
|
||||
// "name": "gcp",
|
||||
//
|
||||
// 'config': {
|
||||
// # Authentication options
|
||||
// # These options are used while getting a token.
|
||||
//
|
||||
// # comma-separated list of GCP API scopes. default value of this field
|
||||
// # is "https://www.googleapis.com/auth/cloud-platform,https://www.googleapis.com/auth/userinfo.email".
|
||||
// # to override the API scopes, specify this field explicitly.
|
||||
// "scopes": "https://www.googleapis.com/auth/cloud-platform"
|
||||
//
|
||||
// # Caching options
|
||||
//
|
||||
// # Raw string data representing cached access token.
|
||||
// "access-token": "ya29.CjWdA4GiBPTt",
|
||||
// # RFC3339Nano expiration timestamp for cached access token.
|
||||
// "expiry": "2016-10-31 22:31:9.123",
|
||||
//
|
||||
// # Command execution options
|
||||
// # These options direct the plugin to execute a specified command and parse
|
||||
// # token and expiry time from the output of the command.
|
||||
//
|
||||
// # Command to execute for access token. Command output will be parsed as JSON.
|
||||
// # If "cmd-args" is not present, this value will be split on whitespace, with
|
||||
// # the first element interpreted as the command, remaining elements as args.
|
||||
// "cmd-path": "/usr/bin/gcloud",
|
||||
//
|
||||
// # Arguments to pass to command to execute for access token.
|
||||
// "cmd-args": "config config-helper --output=json"
|
||||
//
|
||||
// # JSONPath to the string field that represents the access token in
|
||||
// # command output. If omitted, defaults to "{.access_token}".
|
||||
// "token-key": "{.credential.access_token}",
|
||||
//
|
||||
// # JSONPath to the string field that represents expiration timestamp
|
||||
// # of the access token in the command output. If omitted, defaults to
|
||||
// # "{.token_expiry}"
|
||||
// "expiry-key": ""{.credential.token_expiry}",
|
||||
//
|
||||
// # golang reference time in the format that the expiration timestamp uses.
|
||||
// # If omitted, defaults to time.RFC3339Nano
|
||||
// "time-fmt": "2006-01-02 15:04:05.999999999"
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
//
|
||||
type gcpAuthProvider struct {
|
||||
tokenSource oauth2.TokenSource
|
||||
persister restclient.AuthProviderConfigPersister
|
||||
}
|
||||
|
||||
func newGCPAuthProvider(_ string, gcpConfig map[string]string, persister restclient.AuthProviderConfigPersister) (restclient.AuthProvider, error) {
|
||||
ts, err := tokenSource(isCmdTokenSource(gcpConfig), gcpConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cts, err := newCachedTokenSource(gcpConfig["access-token"], gcpConfig["expiry"], persister, ts, gcpConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &gcpAuthProvider{cts, persister}, nil
|
||||
}
|
||||
|
||||
func isCmdTokenSource(gcpConfig map[string]string) bool {
|
||||
_, ok := gcpConfig["cmd-path"]
|
||||
return ok
|
||||
}
|
||||
|
||||
func tokenSource(isCmd bool, gcpConfig map[string]string) (oauth2.TokenSource, error) {
|
||||
// Command-based token source
|
||||
if isCmd {
|
||||
cmd := gcpConfig["cmd-path"]
|
||||
if len(cmd) == 0 {
|
||||
return nil, fmt.Errorf("missing access token cmd")
|
||||
}
|
||||
if gcpConfig["scopes"] != "" {
|
||||
return nil, fmt.Errorf("scopes can only be used when kubectl is using a gcp service account key")
|
||||
}
|
||||
var args []string
|
||||
if cmdArgs, ok := gcpConfig["cmd-args"]; ok {
|
||||
args = strings.Fields(cmdArgs)
|
||||
} else {
|
||||
fields := strings.Fields(cmd)
|
||||
cmd = fields[0]
|
||||
args = fields[1:]
|
||||
}
|
||||
return newCmdTokenSource(cmd, args, gcpConfig["token-key"], gcpConfig["expiry-key"], gcpConfig["time-fmt"]), nil
|
||||
}
|
||||
|
||||
// Google Application Credentials-based token source
|
||||
scopes := parseScopes(gcpConfig)
|
||||
ts, err := google.DefaultTokenSource(context.Background(), scopes...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cannot construct google default token source: %v", err)
|
||||
}
|
||||
return ts, nil
|
||||
}
|
||||
|
||||
// parseScopes constructs a list of scopes that should be included in token source
|
||||
// from the config map.
|
||||
func parseScopes(gcpConfig map[string]string) []string {
|
||||
scopes, ok := gcpConfig["scopes"]
|
||||
if !ok {
|
||||
return defaultScopes
|
||||
}
|
||||
if scopes == "" {
|
||||
return []string{}
|
||||
}
|
||||
return strings.Split(gcpConfig["scopes"], ",")
|
||||
}
|
||||
|
||||
func (g *gcpAuthProvider) WrapTransport(rt http.RoundTripper) http.RoundTripper {
|
||||
var resetCache map[string]string
|
||||
if cts, ok := g.tokenSource.(*cachedTokenSource); ok {
|
||||
resetCache = cts.baseCache()
|
||||
} else {
|
||||
resetCache = make(map[string]string)
|
||||
}
|
||||
return &conditionalTransport{&oauth2.Transport{Source: g.tokenSource, Base: rt}, g.persister, resetCache}
|
||||
}
|
||||
|
||||
func (g *gcpAuthProvider) Login() error { return nil }
|
||||
|
||||
type cachedTokenSource struct {
|
||||
lk sync.Mutex
|
||||
source oauth2.TokenSource
|
||||
accessToken string
|
||||
expiry time.Time
|
||||
persister restclient.AuthProviderConfigPersister
|
||||
cache map[string]string
|
||||
}
|
||||
|
||||
func newCachedTokenSource(accessToken, expiry string, persister restclient.AuthProviderConfigPersister, ts oauth2.TokenSource, cache map[string]string) (*cachedTokenSource, error) {
|
||||
var expiryTime time.Time
|
||||
if parsedTime, err := time.Parse(time.RFC3339Nano, expiry); err == nil {
|
||||
expiryTime = parsedTime
|
||||
}
|
||||
if cache == nil {
|
||||
cache = make(map[string]string)
|
||||
}
|
||||
return &cachedTokenSource{
|
||||
source: ts,
|
||||
accessToken: accessToken,
|
||||
expiry: expiryTime,
|
||||
persister: persister,
|
||||
cache: cache,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (t *cachedTokenSource) Token() (*oauth2.Token, error) {
|
||||
tok := t.cachedToken()
|
||||
if tok.Valid() && !tok.Expiry.IsZero() {
|
||||
return tok, nil
|
||||
}
|
||||
tok, err := t.source.Token()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cache := t.update(tok)
|
||||
if t.persister != nil {
|
||||
if err := t.persister.Persist(cache); err != nil {
|
||||
klog.V(4).Infof("Failed to persist token: %v", err)
|
||||
}
|
||||
}
|
||||
return tok, nil
|
||||
}
|
||||
|
||||
func (t *cachedTokenSource) cachedToken() *oauth2.Token {
|
||||
t.lk.Lock()
|
||||
defer t.lk.Unlock()
|
||||
return &oauth2.Token{
|
||||
AccessToken: t.accessToken,
|
||||
TokenType: "Bearer",
|
||||
Expiry: t.expiry,
|
||||
}
|
||||
}
|
||||
|
||||
func (t *cachedTokenSource) update(tok *oauth2.Token) map[string]string {
|
||||
t.lk.Lock()
|
||||
defer t.lk.Unlock()
|
||||
t.accessToken = tok.AccessToken
|
||||
t.expiry = tok.Expiry
|
||||
ret := map[string]string{}
|
||||
for k, v := range t.cache {
|
||||
ret[k] = v
|
||||
}
|
||||
ret["access-token"] = t.accessToken
|
||||
ret["expiry"] = t.expiry.Format(time.RFC3339Nano)
|
||||
return ret
|
||||
}
|
||||
|
||||
// baseCache is the base configuration value for this TokenSource, without any cached ephemeral tokens.
|
||||
func (t *cachedTokenSource) baseCache() map[string]string {
|
||||
t.lk.Lock()
|
||||
defer t.lk.Unlock()
|
||||
ret := map[string]string{}
|
||||
for k, v := range t.cache {
|
||||
ret[k] = v
|
||||
}
|
||||
delete(ret, "access-token")
|
||||
delete(ret, "expiry")
|
||||
return ret
|
||||
}
|
||||
|
||||
type commandTokenSource struct {
|
||||
cmd string
|
||||
args []string
|
||||
tokenKey string
|
||||
expiryKey string
|
||||
timeFmt string
|
||||
}
|
||||
|
||||
func newCmdTokenSource(cmd string, args []string, tokenKey, expiryKey, timeFmt string) *commandTokenSource {
|
||||
if len(timeFmt) == 0 {
|
||||
timeFmt = time.RFC3339Nano
|
||||
}
|
||||
if len(tokenKey) == 0 {
|
||||
tokenKey = "{.access_token}"
|
||||
}
|
||||
if len(expiryKey) == 0 {
|
||||
expiryKey = "{.token_expiry}"
|
||||
}
|
||||
return &commandTokenSource{
|
||||
cmd: cmd,
|
||||
args: args,
|
||||
tokenKey: tokenKey,
|
||||
expiryKey: expiryKey,
|
||||
timeFmt: timeFmt,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *commandTokenSource) Token() (*oauth2.Token, error) {
|
||||
fullCmd := strings.Join(append([]string{c.cmd}, c.args...), " ")
|
||||
cmd := execCommand(c.cmd, c.args...)
|
||||
var stderr bytes.Buffer
|
||||
cmd.Stderr = &stderr
|
||||
output, err := cmd.Output()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error executing access token command %q: err=%v output=%s stderr=%s", fullCmd, err, output, string(stderr.Bytes()))
|
||||
}
|
||||
token, err := c.parseTokenCmdOutput(output)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error parsing output for access token command %q: %v", fullCmd, err)
|
||||
}
|
||||
return token, nil
|
||||
}
|
||||
|
||||
func (c *commandTokenSource) parseTokenCmdOutput(output []byte) (*oauth2.Token, error) {
|
||||
output, err := yaml.ToJSON(output)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var data interface{}
|
||||
if err := json.Unmarshal(output, &data); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
accessToken, err := parseJSONPath(data, "token-key", c.tokenKey)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error parsing token-key %q from %q: %v", c.tokenKey, string(output), err)
|
||||
}
|
||||
expiryStr, err := parseJSONPath(data, "expiry-key", c.expiryKey)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error parsing expiry-key %q from %q: %v", c.expiryKey, string(output), err)
|
||||
}
|
||||
var expiry time.Time
|
||||
if t, err := time.Parse(c.timeFmt, expiryStr); err != nil {
|
||||
klog.V(4).Infof("Failed to parse token expiry from %s (fmt=%s): %v", expiryStr, c.timeFmt, err)
|
||||
} else {
|
||||
expiry = t
|
||||
}
|
||||
|
||||
return &oauth2.Token{
|
||||
AccessToken: accessToken,
|
||||
TokenType: "Bearer",
|
||||
Expiry: expiry,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func parseJSONPath(input interface{}, name, template string) (string, error) {
|
||||
j := jsonpath.New(name)
|
||||
buf := new(bytes.Buffer)
|
||||
if err := j.Parse(template); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := j.Execute(buf, input); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return buf.String(), nil
|
||||
}
|
||||
|
||||
type conditionalTransport struct {
|
||||
oauthTransport *oauth2.Transport
|
||||
persister restclient.AuthProviderConfigPersister
|
||||
resetCache map[string]string
|
||||
}
|
||||
|
||||
var _ net.RoundTripperWrapper = &conditionalTransport{}
|
||||
|
||||
func (t *conditionalTransport) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
if len(req.Header.Get("Authorization")) != 0 {
|
||||
return t.oauthTransport.Base.RoundTrip(req)
|
||||
}
|
||||
|
||||
res, err := t.oauthTransport.RoundTrip(req)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if res.StatusCode == 401 {
|
||||
klog.V(4).Infof("The credentials that were supplied are invalid for the target cluster")
|
||||
t.persister.Persist(t.resetCache)
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (t *conditionalTransport) WrappedRoundTripper() http.RoundTripper { return t.oauthTransport.Base }
|
||||
Reference in New Issue
Block a user