transport.go

Documentation: k8s.io/client-go/transport

     1  /*
     2  Copyright 2015 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package transport
    18  
    19  import (
    20  	"context"
    21  	"crypto/tls"
    22  	"crypto/x509"
    23  	"encoding/pem"
    24  	"fmt"
    25  	"net/http"
    26  	"os"
    27  	"sync"
    28  	"time"
    29  
    30  	utilnet "k8s.io/apimachinery/pkg/util/net"
    31  	"k8s.io/klog/v2"
    32  )
    33  
    34  // New returns an http.RoundTripper that will provide the authentication
    35  // or transport level security defined by the provided Config.
    36  func New(config *Config) (http.RoundTripper, error) {
    37  	// Set transport level security
    38  	if config.Transport != nil && (config.HasCA() || config.HasCertAuth() || config.HasCertCallback() || config.TLS.Insecure) {
    39  		return nil, fmt.Errorf("using a custom transport with TLS certificate options or the insecure flag is not allowed")
    40  	}
    41  
    42  	if !isValidHolders(config) {
    43  		return nil, fmt.Errorf("misconfigured holder for dialer or cert callback")
    44  	}
    45  
    46  	var (
    47  		rt  http.RoundTripper
    48  		err error
    49  	)
    50  
    51  	if config.Transport != nil {
    52  		rt = config.Transport
    53  	} else {
    54  		rt, err = tlsCache.get(config)
    55  		if err != nil {
    56  			return nil, err
    57  		}
    58  	}
    59  
    60  	return HTTPWrappersForConfig(config, rt)
    61  }
    62  
    63  func isValidHolders(config *Config) bool {
    64  	if config.TLS.GetCertHolder != nil && config.TLS.GetCertHolder.GetCert == nil {
    65  		return false
    66  	}
    67  
    68  	if config.DialHolder != nil && config.DialHolder.Dial == nil {
    69  		return false
    70  	}
    71  
    72  	return true
    73  }
    74  
    75  // TLSConfigFor returns a tls.Config that will provide the transport level security defined
    76  // by the provided Config. Will return nil if no transport level security is requested.
    77  func TLSConfigFor(c *Config) (*tls.Config, error) {
    78  	if !(c.HasCA() || c.HasCertAuth() || c.HasCertCallback() || c.TLS.Insecure || len(c.TLS.ServerName) > 0 || len(c.TLS.NextProtos) > 0) {
    79  		return nil, nil
    80  	}
    81  	if c.HasCA() && c.TLS.Insecure {
    82  		return nil, fmt.Errorf("specifying a root certificates file with the insecure flag is not allowed")
    83  	}
    84  	if err := loadTLSFiles(c); err != nil {
    85  		return nil, err
    86  	}
    87  
    88  	tlsConfig := &tls.Config{
    89  		// Can't use SSLv3 because of POODLE and BEAST
    90  		// Can't use TLSv1.0 because of POODLE and BEAST using CBC cipher
    91  		// Can't use TLSv1.1 because of RC4 cipher usage
    92  		MinVersion:         tls.VersionTLS12,
    93  		InsecureSkipVerify: c.TLS.Insecure,
    94  		ServerName:         c.TLS.ServerName,
    95  		NextProtos:         c.TLS.NextProtos,
    96  	}
    97  
    98  	if c.HasCA() {
    99  		/*
   100  			kubernetes mutual (2-way) x509 between client and apiserver:
   101  
   102  				1. apiserver sending its apiserver certificate along with its publickey to client
   103  				>2. client verifies the apiserver certificate sent against its cluster certificate authority data
   104  				3. client sending its client certificate along with its public key to the apiserver
   105  				4. apiserver verifies the client certificate sent against its cluster certificate authority data
   106  
   107  				description:
   108  					here, with this block,
   109  					cluster certificate authority data gets loaded into TLS before the handshake process
   110  					for client to later during the handshake verify the apiserver certificate
   111  
   112  				normal args related to this stage:
   113  					--certificate-authority='':
   114  						Path to a cert file for the certificate authority
   115  
   116  					(retrievable from "kubectl options" command)
   117  					(suggested by @deads2k)
   118  
   119  				see also:
   120  					- for the step 1, see: staging/src/k8s.io/apiserver/pkg/server/options/serving.go
   121  					- for the step 3, see: a few lines below in this file
   122  					- for the step 4, see: staging/src/k8s.io/apiserver/pkg/authentication/request/x509/x509.go
   123  		*/
   124  
   125  		rootCAs, err := rootCertPool(c.TLS.CAData)
   126  		if err != nil {
   127  			return nil, fmt.Errorf("unable to load root certificates: %w", err)
   128  		}
   129  		tlsConfig.RootCAs = rootCAs
   130  	}
   131  
   132  	var staticCert *tls.Certificate
   133  	// Treat cert as static if either key or cert was data, not a file
   134  	if c.HasCertAuth() && !c.TLS.ReloadTLSFiles {
   135  		// If key/cert were provided, verify them before setting up
   136  		// tlsConfig.GetClientCertificate.
   137  		cert, err := tls.X509KeyPair(c.TLS.CertData, c.TLS.KeyData)
   138  		if err != nil {
   139  			return nil, err
   140  		}
   141  		staticCert = &cert
   142  	}
   143  
   144  	var dynamicCertLoader func() (*tls.Certificate, error)
   145  	if c.TLS.ReloadTLSFiles {
   146  		dynamicCertLoader = cachingCertificateLoader(c.TLS.CertFile, c.TLS.KeyFile)
   147  	}
   148  
   149  	if c.HasCertAuth() || c.HasCertCallback() {
   150  
   151  		/*
   152  			    kubernetes mutual (2-way) x509 between client and apiserver:
   153  
   154  					1. apiserver sending its apiserver certificate along with its publickey to client
   155  					2. client verifies the apiserver certificate sent against its cluster certificate authority data
   156  					>3. client sending its client certificate along with its public key to the apiserver
   157  					4. apiserver verifies the client certificate sent against its cluster certificate authority data
   158  
   159  					description:
   160  						here, with this callback function,
   161  						client certificate and pub key get loaded into TLS during the handshake process
   162  						for apiserver to later in the step 4 verify the client certificate
   163  
   164  					normal args related to this stage:
   165  						--client-certificate='':
   166  							Path to a client certificate file for TLS
   167  						--client-key='':
   168  							Path to a client key file for TLS
   169  
   170  						(retrievable from "kubectl options" command)
   171  						(suggested by @deads2k)
   172  
   173  					see also:
   174  						- for the step 1, see: staging/src/k8s.io/apiserver/pkg/server/options/serving.go
   175  						- for the step 2, see: a few lines above in this file
   176  						- for the step 4, see: staging/src/k8s.io/apiserver/pkg/authentication/request/x509/x509.go
   177  		*/
   178  
   179  		tlsConfig.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
   180  			// Note: static key/cert data always take precedence over cert
   181  			// callback.
   182  			if staticCert != nil {
   183  				return staticCert, nil
   184  			}
   185  			// key/cert files lead to ReloadTLSFiles being set - takes precedence over cert callback
   186  			if dynamicCertLoader != nil {
   187  				return dynamicCertLoader()
   188  			}
   189  			if c.HasCertCallback() {
   190  				cert, err := c.TLS.GetCertHolder.GetCert()
   191  				if err != nil {
   192  					return nil, err
   193  				}
   194  				// GetCert may return empty value, meaning no cert.
   195  				if cert != nil {
   196  					return cert, nil
   197  				}
   198  			}
   199  
   200  			// Both c.TLS.CertData/KeyData were unset and GetCert didn't return
   201  			// anything. Return an empty tls.Certificate, no client cert will
   202  			// be sent to the server.
   203  			return &tls.Certificate{}, nil
   204  		}
   205  	}
   206  
   207  	return tlsConfig, nil
   208  }
   209  
   210  // loadTLSFiles copies the data from the CertFile, KeyFile, and CAFile fields into the CertData,
   211  // KeyData, and CAFile fields, or returns an error. If no error is returned, all three fields are
   212  // either populated or were empty to start.
   213  func loadTLSFiles(c *Config) error {
   214  	var err error
   215  	c.TLS.CAData, err = dataFromSliceOrFile(c.TLS.CAData, c.TLS.CAFile)
   216  	if err != nil {
   217  		return err
   218  	}
   219  
   220  	// Check that we are purely loading from files
   221  	if len(c.TLS.CertFile) > 0 && len(c.TLS.CertData) == 0 && len(c.TLS.KeyFile) > 0 && len(c.TLS.KeyData) == 0 {
   222  		c.TLS.ReloadTLSFiles = true
   223  	}
   224  
   225  	c.TLS.CertData, err = dataFromSliceOrFile(c.TLS.CertData, c.TLS.CertFile)
   226  	if err != nil {
   227  		return err
   228  	}
   229  
   230  	c.TLS.KeyData, err = dataFromSliceOrFile(c.TLS.KeyData, c.TLS.KeyFile)
   231  	return err
   232  }
   233  
   234  // dataFromSliceOrFile returns data from the slice (if non-empty), or from the file,
   235  // or an error if an error occurred reading the file
   236  func dataFromSliceOrFile(data []byte, file string) ([]byte, error) {
   237  	if len(data) > 0 {
   238  		return data, nil
   239  	}
   240  	if len(file) > 0 {
   241  		fileData, err := os.ReadFile(file)
   242  		if err != nil {
   243  			return []byte{}, err
   244  		}
   245  		return fileData, nil
   246  	}
   247  	return nil, nil
   248  }
   249  
   250  // rootCertPool returns nil if caData is empty.  When passed along, this will mean "use system CAs".
   251  // When caData is not empty, it will be the ONLY information used in the CertPool.
   252  func rootCertPool(caData []byte) (*x509.CertPool, error) {
   253  	// What we really want is a copy of x509.systemRootsPool, but that isn't exposed.  It's difficult to build (see the go
   254  	// code for a look at the platform specific insanity), so we'll use the fact that RootCAs == nil gives us the system values
   255  	// It doesn't allow trusting either/or, but hopefully that won't be an issue
   256  	if len(caData) == 0 {
   257  		return nil, nil
   258  	}
   259  
   260  	// if we have caData, use it
   261  	certPool := x509.NewCertPool()
   262  	if ok := certPool.AppendCertsFromPEM(caData); !ok {
   263  		return nil, createErrorParsingCAData(caData)
   264  	}
   265  	return certPool, nil
   266  }
   267  
   268  // createErrorParsingCAData ALWAYS returns an error.  We call it because know we failed to AppendCertsFromPEM
   269  // but we don't know the specific error because that API is just true/false
   270  func createErrorParsingCAData(pemCerts []byte) error {
   271  	for len(pemCerts) > 0 {
   272  		var block *pem.Block
   273  		block, pemCerts = pem.Decode(pemCerts)
   274  		if block == nil {
   275  			return fmt.Errorf("unable to parse bytes as PEM block")
   276  		}
   277  
   278  		if block.Type != "CERTIFICATE" || len(block.Headers) != 0 {
   279  			continue
   280  		}
   281  
   282  		if _, err := x509.ParseCertificate(block.Bytes); err != nil {
   283  			return fmt.Errorf("failed to parse certificate: %w", err)
   284  		}
   285  	}
   286  	return fmt.Errorf("no valid certificate authority data seen")
   287  }
   288  
   289  // WrapperFunc wraps an http.RoundTripper when a new transport
   290  // is created for a client, allowing per connection behavior
   291  // to be injected.
   292  type WrapperFunc func(rt http.RoundTripper) http.RoundTripper
   293  
   294  // Wrappers accepts any number of wrappers and returns a wrapper
   295  // function that is the equivalent of calling each of them in order. Nil
   296  // values are ignored, which makes this function convenient for incrementally
   297  // wrapping a function.
   298  func Wrappers(fns ...WrapperFunc) WrapperFunc {
   299  	if len(fns) == 0 {
   300  		return nil
   301  	}
   302  	// optimize the common case of wrapping a possibly nil transport wrapper
   303  	// with an additional wrapper
   304  	if len(fns) == 2 && fns[0] == nil {
   305  		return fns[1]
   306  	}
   307  	return func(rt http.RoundTripper) http.RoundTripper {
   308  		base := rt
   309  		for _, fn := range fns {
   310  			if fn != nil {
   311  				base = fn(base)
   312  			}
   313  		}
   314  		return base
   315  	}
   316  }
   317  
   318  // ContextCanceller prevents new requests after the provided context is finished.
   319  // err is returned when the context is closed, allowing the caller to provide a context
   320  // appropriate error.
   321  func ContextCanceller(ctx context.Context, err error) WrapperFunc {
   322  	return func(rt http.RoundTripper) http.RoundTripper {
   323  		return &contextCanceller{
   324  			ctx: ctx,
   325  			rt:  rt,
   326  			err: err,
   327  		}
   328  	}
   329  }
   330  
   331  type contextCanceller struct {
   332  	ctx context.Context
   333  	rt  http.RoundTripper
   334  	err error
   335  }
   336  
   337  func (b *contextCanceller) RoundTrip(req *http.Request) (*http.Response, error) {
   338  	select {
   339  	case <-b.ctx.Done():
   340  		return nil, b.err
   341  	default:
   342  		return b.rt.RoundTrip(req)
   343  	}
   344  }
   345  
   346  func tryCancelRequest(rt http.RoundTripper, req *http.Request) {
   347  	type canceler interface {
   348  		CancelRequest(*http.Request)
   349  	}
   350  	switch rt := rt.(type) {
   351  	case canceler:
   352  		rt.CancelRequest(req)
   353  	case utilnet.RoundTripperWrapper:
   354  		tryCancelRequest(rt.WrappedRoundTripper(), req)
   355  	default:
   356  		klog.Warningf("Unable to cancel request for %T", rt)
   357  	}
   358  }
   359  
   360  type certificateCacheEntry struct {
   361  	cert  *tls.Certificate
   362  	err   error
   363  	birth time.Time
   364  }
   365  
   366  // isStale returns true when this cache entry is too old to be usable
   367  func (c *certificateCacheEntry) isStale() bool {
   368  	return time.Since(c.birth) > time.Second
   369  }
   370  
   371  func newCertificateCacheEntry(certFile, keyFile string) certificateCacheEntry {
   372  	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
   373  	return certificateCacheEntry{cert: &cert, err: err, birth: time.Now()}
   374  }
   375  
   376  // cachingCertificateLoader ensures that we don't hammer the filesystem when opening many connections
   377  // the underlying cert files are read at most once every second
   378  func cachingCertificateLoader(certFile, keyFile string) func() (*tls.Certificate, error) {
   379  	current := newCertificateCacheEntry(certFile, keyFile)
   380  	var currentMtx sync.RWMutex
   381  
   382  	return func() (*tls.Certificate, error) {
   383  		currentMtx.RLock()
   384  		if current.isStale() {
   385  			currentMtx.RUnlock()
   386  
   387  			currentMtx.Lock()
   388  			defer currentMtx.Unlock()
   389  
   390  			if current.isStale() {
   391  				current = newCertificateCacheEntry(certFile, keyFile)
   392  			}
   393  		} else {
   394  			defer currentMtx.RUnlock()
   395  		}
   396  
   397  		return current.cert, current.err
   398  	}
   399  }
   400
View as plain text