Ensure the bootstrap rotation code is tested by forcing rotation

Expose both a Stop() method (for cleanup) and a method to force
cert rotation, but only expose Stop() on the interface.

Verify that we choose the correct client.

cmd/kubelet/app/BUILD

@@ -8,8 +8,22 @@ load(
 
 go_test(
     name = "go_default_test",
-    srcs = ["server_test.go"],
+    srcs = [
+        "server_bootstrap_test.go",
+        "server_test.go",
+    ],
     embed = [":go_default_library"],
+    deps = [
+        "//staging/src/k8s.io/api/certificates/v1beta1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/client-go/rest:go_default_library",
+        "//staging/src/k8s.io/client-go/util/cert:go_default_library",
+        "//vendor/github.com/cloudflare/cfssl/config:go_default_library",
+        "//vendor/github.com/cloudflare/cfssl/signer:go_default_library",
+        "//vendor/github.com/cloudflare/cfssl/signer/local:go_default_library",
+    ],
 )
 
 go_library(
@@ -119,6 +133,7 @@ go_library(
         "//staging/src/k8s.io/client-go/tools/clientcmd:go_default_library",
         "//staging/src/k8s.io/client-go/tools/record:go_default_library",
         "//staging/src/k8s.io/client-go/util/cert:go_default_library",
+        "//staging/src/k8s.io/client-go/util/certificate:go_default_library",
         "//staging/src/k8s.io/cloud-provider:go_default_library",
         "//staging/src/k8s.io/csi-api/pkg/client/clientset/versioned:go_default_library",
         "//staging/src/k8s.io/kubelet/config/v1beta1:go_default_library",

cmd/kubelet/app/server.go

@@ -56,6 +56,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/record"
 	certutil "k8s.io/client-go/util/cert"
+	"k8s.io/client-go/util/certificate"
 	cloudprovider "k8s.io/cloud-provider"
 	csiclientset "k8s.io/csi-api/pkg/client/clientset/versioned"
 	kubeletconfigv1beta1 "k8s.io/kubelet/config/v1beta1"
@@ -743,42 +744,20 @@ func buildKubeletClientConfig(s *options.KubeletServer, nodeName types.NodeName)
 			return nil, nil, err
 		}
 
-		newClientFn := func(current *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
-			// If we have a valid certificate, use that to fetch CSRs. Otherwise use the bootstrap
-			// credentials.
-			// XXX: When an external bootstrap source is available, it should be possible to always use that source
-			//      to retrieve new credentials.
-			config := certConfig
-			if current != nil {
-				config = clientConfig
-			}
-			client, err := clientset.NewForConfig(config)
-			if err != nil {
-				return nil, err
-			}
-			return client.CertificatesV1beta1().CertificateSigningRequests(), nil
-		}
-
-		clientCertificateManager, err := kubeletcertificate.NewKubeletClientCertificateManager(
-			s.CertDirectory,
-			nodeName,
-			clientConfig.CertFile,
-			clientConfig.KeyFile,
-			newClientFn,
-		)
+		clientCertificateManager, err := buildClientCertificateManager(certConfig, clientConfig, s.CertDirectory, nodeName)
 		if err != nil {
 			return nil, nil, err
 		}
 
 		// the rotating transport will use the cert from the cert manager instead of these files
-		transportConfig := *clientConfig
+		transportConfig := restclient.CopyConfig(clientConfig)
 		transportConfig.CertFile = ""
 		transportConfig.KeyFile = ""
 
 		// we set exitAfter to five minutes because we use this client configuration to request new certs - if we are unable
 		// to request new certs, we will be unable to continue normal operation. Exiting the process allows a wrapper
 		// or the bootstrapping credentials to potentially lay down new initial config.
-		closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, &transportConfig, clientCertificateManager, 5*time.Minute)
+		closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, transportConfig, clientCertificateManager, 5*time.Minute)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -786,7 +765,7 @@ func buildKubeletClientConfig(s *options.KubeletServer, nodeName types.NodeName)
 		klog.V(2).Info("Starting client certificate rotation.")
 		clientCertificateManager.Start()
 
-		return &transportConfig, closeAllConns, nil
+		return transportConfig, closeAllConns, nil
 	}
 
 	if len(s.BootstrapKubeconfig) > 0 {
@@ -802,6 +781,35 @@ func buildKubeletClientConfig(s *options.KubeletServer, nodeName types.NodeName)
 	return clientConfig, nil, nil
 }
 
+// buildClientCertificateManager creates a certificate manager that will use certConfig to request a client certificate
+// if no certificate is available, or the most recent clientConfig (which is assumed to point to the cert that the manager will
+// write out).
+func buildClientCertificateManager(certConfig, clientConfig *restclient.Config, certDir string, nodeName types.NodeName) (certificate.Manager, error) {
+	newClientFn := func(current *tls.Certificate) (certificatesclient.CertificateSigningRequestInterface, error) {
+		// If we have a valid certificate, use that to fetch CSRs. Otherwise use the bootstrap
+		// credentials. In the future it would be desirable to change the behavior of bootstrap
+		// to always fall back to the external bootstrap credentials when such credentials are
+		// provided by a fundamental trust system like cloud VM identity or an HSM module.
+		config := certConfig
+		if current != nil {
+			config = clientConfig
+		}
+		client, err := clientset.NewForConfig(config)
+		if err != nil {
+			return nil, err
+		}
+		return client.CertificatesV1beta1().CertificateSigningRequests(), nil
+	}
+
+	return kubeletcertificate.NewKubeletClientCertificateManager(
+		certDir,
+		nodeName,
+		clientConfig.CertFile,
+		clientConfig.KeyFile,
+		newClientFn,
+	)
+}
+
 // getNodeName returns the node name according to the cloud provider
 // if cloud provider is specified. Otherwise, returns the hostname of the node.
 func getNodeName(cloud cloudprovider.Interface, hostname string) (types.NodeName, error) {

cmd/kubelet/app/server_bootstrap_test.go

@@ -0,0 +1,281 @@
+/*
+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 app
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	cryptorand "crypto/rand"
+	"crypto/x509"
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	cfsslconfig "github.com/cloudflare/cfssl/config"
+	cfsslsigner "github.com/cloudflare/cfssl/signer"
+	cfssllocal "github.com/cloudflare/cfssl/signer/local"
+
+	certapi "k8s.io/api/certificates/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	restclient "k8s.io/client-go/rest"
+	certutil "k8s.io/client-go/util/cert"
+)
+
+// Test_buildClientCertificateManager validates that we can build a local client cert
+// manager that will use the bootstrap client until we get a valid cert, then use our
+// provided identity on subsequent requests.
+func Test_buildClientCertificateManager(t *testing.T) {
+	testDir, err := ioutil.TempDir("", "kubeletcert")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { os.RemoveAll(testDir) }()
+
+	serverPrivateKey, err := ecdsa.GenerateKey(elliptic.P256(), cryptorand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	serverCA, err := certutil.NewSelfSignedCACert(certutil.Config{
+		CommonName: "the-test-framework",
+	}, serverPrivateKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	server := &csrSimulator{
+		t:                t,
+		serverPrivateKey: serverPrivateKey,
+		serverCA:         serverCA,
+	}
+	s := httptest.NewServer(server)
+	defer s.Close()
+
+	config1 := &restclient.Config{
+		UserAgent: "FirstClient",
+		Host:      s.URL,
+	}
+	config2 := &restclient.Config{
+		UserAgent: "SecondClient",
+		Host:      s.URL,
+	}
+
+	nodeName := types.NodeName("test")
+	m, err := buildClientCertificateManager(config1, config2, testDir, nodeName)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m.Stop()
+	r := m.(rotater)
+
+	// get an expired CSR (simulating historical output)
+	server.backdate = 2 * time.Hour
+	server.expectUserAgent = "FirstClient"
+	ok, err := r.RotateCerts()
+	if !ok || err != nil {
+		t.Fatalf("unexpected rotation err: %t %v", ok, err)
+	}
+	if cert := m.Current(); cert != nil {
+		t.Fatalf("Unexpected cert, should be expired: %#v", cert)
+	}
+	fi := getFileInfo(testDir)
+	if len(fi) != 2 {
+		t.Fatalf("Unexpected directory contents: %#v", fi)
+	}
+
+	// if m.Current() == nil, then we try again and get a valid
+	// client
+	server.backdate = 0
+	server.expectUserAgent = "FirstClient"
+	if ok, err := r.RotateCerts(); !ok || err != nil {
+		t.Fatalf("unexpected rotation err: %t %v", ok, err)
+	}
+	if cert := m.Current(); cert == nil {
+		t.Fatalf("Unexpected cert, should be valid: %#v", cert)
+	}
+	fi = getFileInfo(testDir)
+	if len(fi) != 2 {
+		t.Fatalf("Unexpected directory contents: %#v", fi)
+	}
+
+	// if m.Current() != nil, then we should use the second client
+	server.expectUserAgent = "SecondClient"
+	if ok, err := r.RotateCerts(); !ok || err != nil {
+		t.Fatalf("unexpected rotation err: %t %v", ok, err)
+	}
+	if cert := m.Current(); cert == nil {
+		t.Fatalf("Unexpected cert, should be valid: %#v", cert)
+	}
+	fi = getFileInfo(testDir)
+	if len(fi) != 2 {
+		t.Fatalf("Unexpected directory contents: %#v", fi)
+	}
+}
+
+func getFileInfo(dir string) map[string]os.FileInfo {
+	fi := make(map[string]os.FileInfo)
+	filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+		if path == dir {
+			return nil
+		}
+		fi[path] = info
+		if !info.IsDir() {
+			os.Remove(path)
+		}
+		return nil
+	})
+	return fi
+}
+
+type rotater interface {
+	RotateCerts() (bool, error)
+}
+
+func getCSR(req *http.Request) (*certapi.CertificateSigningRequest, error) {
+	if req.Body == nil {
+		return nil, nil
+	}
+	body, err := ioutil.ReadAll(req.Body)
+	if err != nil {
+		return nil, err
+	}
+	csr := &certapi.CertificateSigningRequest{}
+	if err := json.Unmarshal(body, csr); err != nil {
+		return nil, err
+	}
+	return csr, nil
+}
+
+func mustMarshal(obj interface{}) []byte {
+	data, err := json.Marshal(obj)
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
+type csrSimulator struct {
+	t *testing.T
+
+	serverPrivateKey *ecdsa.PrivateKey
+	serverCA         *x509.Certificate
+	backdate         time.Duration
+
+	expectUserAgent string
+
+	lock sync.Mutex
+	csr  *certapi.CertificateSigningRequest
+}
+
+func (s *csrSimulator) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	t := s.t
+
+	t.Logf("Request %s %s %s", req.Method, req.URL, req.UserAgent())
+
+	if len(s.expectUserAgent) > 0 && req.UserAgent() != s.expectUserAgent {
+		t.Errorf("Unexpected user agent: %s", req.UserAgent())
+	}
+
+	switch {
+	case req.Method == "POST" && req.URL.Path == "/apis/certificates.k8s.io/v1beta1/certificatesigningrequests":
+		csr, err := getCSR(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if csr.Name == "" {
+			csr.Name = "test-csr"
+		}
+
+		csr.UID = types.UID("1")
+		csr.ResourceVersion = "1"
+		data := mustMarshal(csr)
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(data)
+
+		csr = csr.DeepCopy()
+		csr.ResourceVersion = "2"
+		var usages []string
+		for _, usage := range csr.Spec.Usages {
+			usages = append(usages, string(usage))
+		}
+		policy := &cfsslconfig.Signing{
+			Default: &cfsslconfig.SigningProfile{
+				Usage:        usages,
+				Expiry:       time.Hour,
+				ExpiryString: time.Hour.String(),
+				Backdate:     s.backdate,
+			},
+		}
+		cfs, err := cfssllocal.NewSigner(s.serverPrivateKey, s.serverCA, cfsslsigner.DefaultSigAlgo(s.serverPrivateKey), policy)
+		if err != nil {
+			t.Fatal(err)
+		}
+		csr.Status.Certificate, err = cfs.Sign(cfsslsigner.SignRequest{
+			Request: string(csr.Spec.Request),
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		csr.Status.Conditions = []certapi.CertificateSigningRequestCondition{
+			{Type: certapi.CertificateApproved},
+		}
+		s.csr = csr
+
+	case req.Method == "GET" && req.URL.Path == "/apis/certificates.k8s.io/v1beta1/certificatesigningrequests" && req.URL.RawQuery == "fieldSelector=metadata.name%3Dtest-csr&limit=500":
+		if s.csr == nil {
+			t.Fatalf("no csr")
+		}
+		csr := s.csr.DeepCopy()
+
+		data := mustMarshal(&certapi.CertificateSigningRequestList{
+			ListMeta: metav1.ListMeta{
+				ResourceVersion: "2",
+			},
+			Items: []certapi.CertificateSigningRequest{
+				*csr,
+			},
+		})
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(data)
+
+	case req.Method == "GET" && req.URL.Path == "/apis/certificates.k8s.io/v1beta1/certificatesigningrequests" && req.URL.RawQuery == "fieldSelector=metadata.name%3Dtest-csr&resourceVersion=2&watch=true":
+		if s.csr == nil {
+			t.Fatalf("no csr")
+		}
+		csr := s.csr.DeepCopy()
+
+		data := mustMarshal(&metav1.WatchEvent{
+			Type: "ADDED",
+			Object: runtime.RawExtension{
+				Raw: mustMarshal(csr),
+			},
+		})
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(data)
+
+	default:
+		t.Fatalf("unexpected request: %s %s", req.Method, req.URL)
+	}
+}

pkg/kubelet/certificate/transport_test.go

@@ -124,7 +124,9 @@ func (f *fakeManager) SetCertificateSigningRequestClient(certificatesclient.Cert
 
 func (f *fakeManager) ServerHealthy() bool { return f.healthy }
 
-func (f *fakeManager) Start() {}
+func (f *fakeManager) Start()                     {}
+func (f *fakeManager) Stop()                      {}
+func (f *fakeManager) RotateCerts() (bool, error) { return false, nil }
 
 func (f *fakeManager) Current() *tls.Certificate {
 	if val := f.cert.Load(); val != nil {

staging/src/k8s.io/client-go/util/certificate/certificate_manager.go

@@ -50,6 +50,8 @@ var certificateWaitBackoff = wait.Backoff{Duration: 30 * time.Second, Steps: 4,
 type Manager interface {
 	// Start the API server status sync loop.
 	Start()
+	// Stop the cert manager loop.
+	Stop()
 	// Current returns the currently selected certificate from the
 	// certificate manager, as well as the associated certificate and key data
 	// in PEM format.
@@ -164,6 +166,8 @@ type manager struct {
 	// the clientFn must only be accessed under the clientAccessLock
 	clientAccessLock sync.Mutex
 	clientFn         CSRClientFunc
+	stopCh           chan struct{}
+	stopped          bool
 }
 
 // NewManager returns a new certificate manager. A certificate manager is
@@ -184,6 +188,7 @@ func NewManager(config *Config) (Manager, error) {
 	}
 
 	m := manager{
+		stopCh:                make(chan struct{}),
 		clientFn:              config.ClientFn,
 		getTemplate:           getTemplate,
 		dynamicTemplate:       config.GetTemplate != nil,
@@ -219,6 +224,17 @@ func (m *manager) ServerHealthy() bool {
 	return m.serverHealth
 }
 
+// Stop terminates the manager.
+func (m *manager) Stop() {
+	m.clientAccessLock.Lock()
+	defer m.clientAccessLock.Unlock()
+	if m.stopped {
+		return
+	}
+	close(m.stopCh)
+	m.stopped = true
+}
+
 // Start will start the background work of rotating the certificates.
 func (m *manager) Start() {
 	// Certificate rotation depends on access to the API server certificate
@@ -232,7 +248,7 @@ func (m *manager) Start() {
 	klog.V(2).Infof("Certificate rotation is enabled.")
 
 	templateChanged := make(chan struct{})
-	go wait.Forever(func() {
+	go wait.Until(func() {
 		deadline := m.nextRotationDeadline()
 		if sleepInterval := deadline.Sub(time.Now()); sleepInterval > 0 {
 			klog.V(2).Infof("Waiting %v for next certificate rotation", sleepInterval)
@@ -267,17 +283,17 @@ func (m *manager) Start() {
 			utilruntime.HandleError(fmt.Errorf("Reached backoff limit, still unable to rotate certs: %v", err))
 			wait.PollInfinite(32*time.Second, m.rotateCerts)
 		}
-	}, time.Second)
+	}, time.Second, m.stopCh)
 
 	if m.dynamicTemplate {
-		go wait.Forever(func() {
+		go wait.Until(func() {
 			// check if the current template matches what we last requested
 			if !m.certSatisfiesTemplate() && !reflect.DeepEqual(m.getLastRequest(), m.getTemplate()) {
 				// if the template is different, queue up an interrupt of the rotation deadline loop.
 				// if we've requested a CSR that matches the new template by the time the interrupt is handled, the interrupt is disregarded.
 				templateChanged <- struct{}{}
 			}
-		}, time.Second)
+		}, time.Second, m.stopCh)
 	}
 }
 
@@ -332,11 +348,19 @@ func (m *manager) getClient() (certificatesclient.CertificateSigningRequestInter
 	return m.clientFn(current)
 }
 
+// RotateCerts is exposed for testing only and is not a part of the public interface.
+// Returns true if it changed the cert, false otherwise. Error is only returned in
+// exceptional cases.
+func (m *manager) RotateCerts() (bool, error) {
+	return m.rotateCerts()
+}
+
 // rotateCerts attempts to request a client cert from the server, wait a reasonable
 // period of time for it to be signed, and then update the cert on disk. If it cannot
 // retrieve a cert, it will return false. It will only return error in exceptional cases.
 // This method also keeps track of "server health" by interpreting the responses it gets
 // from the server on the various calls it makes.
+// TODO: return errors, have callers handle and log them correctly
 func (m *manager) rotateCerts() (bool, error) {
 	klog.V(2).Infof("Rotating certificates")