Merge pull request #86816 from deads2k/auto-reload-csr

add dynamic reloading for CSR signing controllers

pkg/controller/.import-restrictions

@@ -74,6 +74,7 @@
         "k8s.io/apimachinery/pkg/version",
         "k8s.io/api/imagepolicy/v1alpha1",
         "k8s.io/apiserver/pkg/admission",
+        "k8s.io/apiserver/pkg/server/dynamiccertificates",
         "k8s.io/apiserver/pkg/storage",
         "k8s.io/api/batch/v2alpha1",
         "k8s.io/apiserver/pkg/registry/rest",

pkg/controller/certificates/authority/authority.go

@@ -30,6 +30,11 @@ var serialNumberLimit = new(big.Int).Lsh(big.NewInt(1), 128)
 // CertificateAuthority implements a certificate authority that supports policy
 // based signing. It's used by the signing controller.
 type CertificateAuthority struct {
+	// RawCert is an optional field to determine if signing cert/key pairs have changed
+	RawCert []byte
+	// RawKey is an optional field to determine if signing cert/key pairs have changed
+	RawKey []byte
+
 	Certificate *x509.Certificate
 	PrivateKey  crypto.Signer
 	Backdate    time.Duration

pkg/controller/certificates/signer/BUILD

@@ -26,13 +26,17 @@ go_test(
 
 go_library(
     name = "go_default_library",
-    srcs = ["signer.go"],
+    srcs = [
+        "ca_provider.go",
+        "signer.go",
+    ],
     importpath = "k8s.io/kubernetes/pkg/controller/certificates/signer",
     deps = [
         "//pkg/apis/certificates/v1beta1:go_default_library",
         "//pkg/controller/certificates:go_default_library",
         "//pkg/controller/certificates/authority:go_default_library",
         "//staging/src/k8s.io/api/certificates/v1beta1:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/server/dynamiccertificates:go_default_library",
         "//staging/src/k8s.io/client-go/informers/certificates/v1beta1:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/util/cert:go_default_library",

pkg/controller/certificates/signer/ca_provider.go

@@ -0,0 +1,101 @@
+/*
+Copyright 2020 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 signer
+
+import (
+	"bytes"
+	"crypto"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
+	"k8s.io/client-go/util/cert"
+	"k8s.io/client-go/util/keyutil"
+	"k8s.io/kubernetes/pkg/controller/certificates/authority"
+)
+
+func newCAProvider(caFile, caKeyFile string) (*caProvider, error) {
+	caLoader, err := dynamiccertificates.NewDynamicServingContentFromFiles("csr-controller", caFile, caKeyFile)
+	if err != nil {
+		return nil, fmt.Errorf("error reading CA cert file %q: %v", caFile, err)
+	}
+
+	ret := &caProvider{
+		caLoader: caLoader,
+	}
+	if err := ret.setCA(); err != nil {
+		return nil, err
+	}
+
+	return ret, nil
+}
+
+type caProvider struct {
+	caValue  atomic.Value
+	caLoader *dynamiccertificates.DynamicCertKeyPairContent
+}
+
+// setCA unconditionally stores the current cert/key content
+func (p *caProvider) setCA() error {
+	certPEM, keyPEM := p.caLoader.CurrentCertKeyContent()
+
+	certs, err := cert.ParseCertsPEM(certPEM)
+	if err != nil {
+		return fmt.Errorf("error reading CA cert file %q: %v", p.caLoader.Name(), err)
+	}
+	if len(certs) != 1 {
+		return fmt.Errorf("error reading CA cert file %q: expected 1 certificate, found %d", p.caLoader.Name(), len(certs))
+	}
+
+	key, err := keyutil.ParsePrivateKeyPEM(keyPEM)
+	if err != nil {
+		return fmt.Errorf("error reading CA key file %q: %v", p.caLoader.Name(), err)
+	}
+	priv, ok := key.(crypto.Signer)
+	if !ok {
+		return fmt.Errorf("error reading CA key file %q: key did not implement crypto.Signer", p.caLoader.Name())
+	}
+
+	ca := &authority.CertificateAuthority{
+		RawCert: certPEM,
+		RawKey:  keyPEM,
+
+		Certificate: certs[0],
+		PrivateKey:  priv,
+		Backdate:    5 * time.Minute,
+	}
+	p.caValue.Store(ca)
+
+	return nil
+}
+
+// currentCA provides the curent value of the CA.
+// It always check for a stale value.  This is cheap because it's all an in memory cache of small slices.
+func (p *caProvider) currentCA() (*authority.CertificateAuthority, error) {
+	certPEM, keyPEM := p.caLoader.CurrentCertKeyContent()
+	currCA := p.caValue.Load().(*authority.CertificateAuthority)
+	if bytes.Equal(currCA.RawCert, certPEM) && bytes.Equal(currCA.RawKey, keyPEM) {
+		return currCA, nil
+	}
+
+	// the bytes weren't equal, so we have to set and then load
+	if err := p.setCA(); err != nil {
+		return currCA, err
+	}
+	return p.caValue.Load().(*authority.CertificateAuthority), nil
+}

pkg/controller/certificates/signer/signer.go

@@ -18,82 +18,72 @@ limitations under the License.
 package signer
 
 import (
-	"crypto"
 	"encoding/pem"
 	"fmt"
-	"io/ioutil"
 	"time"
 
 	capi "k8s.io/api/certificates/v1beta1"
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
 	certificatesinformers "k8s.io/client-go/informers/certificates/v1beta1"
 	clientset "k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/util/cert"
-	"k8s.io/client-go/util/keyutil"
 	capihelper "k8s.io/kubernetes/pkg/apis/certificates/v1beta1"
 	"k8s.io/kubernetes/pkg/controller/certificates"
 	"k8s.io/kubernetes/pkg/controller/certificates/authority"
 )
 
+type CSRSigningController struct {
+	certificateController *certificates.CertificateController
+	dynamicCertReloader   dynamiccertificates.ControllerRunner
+}
+
 func NewCSRSigningController(
 	client clientset.Interface,
 	csrInformer certificatesinformers.CertificateSigningRequestInformer,
 	caFile, caKeyFile string,
 	certTTL time.Duration,
-) (*certificates.CertificateController, error) {
+) (*CSRSigningController, error) {
 	signer, err := newSigner(caFile, caKeyFile, client, certTTL)
 	if err != nil {
 		return nil, err
 	}
-	return certificates.NewCertificateController(
-		"csrsigning",
-		client,
-		csrInformer,
-		signer.handle,
-	), nil
+
+	return &CSRSigningController{
+		certificateController: certificates.NewCertificateController(
+			"csrsigning",
+			client,
+			csrInformer,
+			signer.handle,
+		),
+		dynamicCertReloader: signer.caProvider.caLoader,
+	}, nil
+}
+
+// Run the main goroutine responsible for watching and syncing jobs.
+func (c *CSRSigningController) Run(workers int, stopCh <-chan struct{}) {
+	go c.dynamicCertReloader.Run(workers, stopCh)
+
+	c.certificateController.Run(workers, stopCh)
 }
 
 type signer struct {
-	ca      *authority.CertificateAuthority
+	caProvider *caProvider
+
 	client  clientset.Interface
 	certTTL time.Duration
 }
 
 func newSigner(caFile, caKeyFile string, client clientset.Interface, certificateDuration time.Duration) (*signer, error) {
-	certPEM, err := ioutil.ReadFile(caFile)
+	caProvider, err := newCAProvider(caFile, caKeyFile)
 	if err != nil {
-		return nil, fmt.Errorf("error reading CA cert file %q: %v", caFile, err)
+		return nil, err
 	}
 
-	certs, err := cert.ParseCertsPEM(certPEM)
-	if err != nil {
-		return nil, fmt.Errorf("error reading CA cert file %q: %v", caFile, err)
+	ret := &signer{
+		caProvider: caProvider,
+		client:     client,
+		certTTL:    certificateDuration,
 	}
-	if len(certs) != 1 {
-		return nil, fmt.Errorf("error reading CA cert file %q: expected 1 certificate, found %d", caFile, len(certs))
-	}
-
-	keyPEM, err := ioutil.ReadFile(caKeyFile)
-	if err != nil {
-		return nil, fmt.Errorf("error reading CA key file %q: %v", caKeyFile, err)
-	}
-	key, err := keyutil.ParsePrivateKeyPEM(keyPEM)
-	if err != nil {
-		return nil, fmt.Errorf("error reading CA key file %q: %v", caKeyFile, err)
-	}
-	priv, ok := key.(crypto.Signer)
-	if !ok {
-		return nil, fmt.Errorf("error reading CA key file %q: key did not implement crypto.Signer", caKeyFile)
-	}
-
-	return &signer{
-		ca: &authority.CertificateAuthority{
-			Certificate: certs[0],
-			PrivateKey:  priv,
-			Backdate:    5 * time.Minute,
-		},
-		client:  client,
-		certTTL: certificateDuration,
-	}, nil
+	return ret, nil
 }
 
 func (s *signer) handle(csr *capi.CertificateSigningRequest) error {
@@ -117,7 +107,11 @@ func (s *signer) sign(csr *capi.CertificateSigningRequest) (*capi.CertificateSig
 		return nil, fmt.Errorf("unable to parse csr %q: %v", csr.Name, err)
 	}
 
-	der, err := s.ca.Sign(x509cr.Raw, authority.PermissiveSigningPolicy{
+	currCA, err := s.caProvider.currentCA()
+	if err != nil {
+		return nil, err
+	}
+	der, err := currCA.Sign(x509cr.Raw, authority.PermissiveSigningPolicy{
 		TTL:    s.certTTL,
 		Usages: csr.Spec.Usages,
 	})

pkg/controller/certificates/signer/signer_test.go

@@ -38,8 +38,13 @@ func TestSigner(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to create signer: %v", err)
 	}
-	s.ca.Now = clock.Now
-	s.ca.Backdate = 0
+	currCA, err := s.caProvider.currentCA()
+	if err != nil {
+		t.Fatal(err)
+	}
+	currCA.Now = clock.Now
+	currCA.Backdate = 0
+	s.caProvider.caValue.Store(currCA)
 
 	csrb, err := ioutil.ReadFile("./testdata/kubelet.csr")
 	if err != nil {

staging/src/k8s.io/apiserver/pkg/server/dynamiccertificates/dynamic_serving_content.go

@@ -29,8 +29,8 @@ import (
 	"k8s.io/klog"
 )
 
-// DynamicFileServingContent provides a CertKeyContentProvider that can dynamically react to new file content
-type DynamicFileServingContent struct {
+// DynamicCertKeyPairContent provides a CertKeyContentProvider that can dynamically react to new file content
+type DynamicCertKeyPairContent struct {
 	name string
 
 	// certFile is the name of the certificate file to read.
@@ -39,7 +39,7 @@ type DynamicFileServingContent struct {
 	keyFile string
 
 	// servingCert is a certKeyContent that contains the last read, non-zero length content of the key and cert
-	servingCert atomic.Value
+	certKeyPair atomic.Value
 
 	listeners []Listener
 
@@ -47,24 +47,24 @@ type DynamicFileServingContent struct {
 	queue workqueue.RateLimitingInterface
 }
 
-var _ Notifier = &DynamicFileServingContent{}
-var _ CertKeyContentProvider = &DynamicFileServingContent{}
-var _ ControllerRunner = &DynamicFileServingContent{}
+var _ Notifier = &DynamicCertKeyPairContent{}
+var _ CertKeyContentProvider = &DynamicCertKeyPairContent{}
+var _ ControllerRunner = &DynamicCertKeyPairContent{}
 
 // NewDynamicServingContentFromFiles returns a dynamic CertKeyContentProvider based on a cert and key filename
-func NewDynamicServingContentFromFiles(purpose, certFile, keyFile string) (*DynamicFileServingContent, error) {
+func NewDynamicServingContentFromFiles(purpose, certFile, keyFile string) (*DynamicCertKeyPairContent, error) {
 	if len(certFile) == 0 || len(keyFile) == 0 {
 		return nil, fmt.Errorf("missing filename for serving cert")
 	}
 	name := fmt.Sprintf("%s::%s::%s", purpose, certFile, keyFile)
 
-	ret := &DynamicFileServingContent{
+	ret := &DynamicCertKeyPairContent{
 		name:     name,
 		certFile: certFile,
 		keyFile:  keyFile,
 		queue:    workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), fmt.Sprintf("DynamicCABundle-%s", purpose)),
 	}
-	if err := ret.loadServingCert(); err != nil {
+	if err := ret.loadCertKeyPair(); err != nil {
 		return nil, err
 	}
 
@@ -72,12 +72,12 @@ func NewDynamicServingContentFromFiles(purpose, certFile, keyFile string) (*Dyna
 }
 
 // AddListener adds a listener to be notified when the serving cert content changes.
-func (c *DynamicFileServingContent) AddListener(listener Listener) {
+func (c *DynamicCertKeyPairContent) AddListener(listener Listener) {
 	c.listeners = append(c.listeners, listener)
 }
 
 // loadServingCert determines the next set of content for the file.
-func (c *DynamicFileServingContent) loadServingCert() error {
+func (c *DynamicCertKeyPairContent) loadCertKeyPair() error {
 	cert, err := ioutil.ReadFile(c.certFile)
 	if err != nil {
 		return err
@@ -102,12 +102,12 @@ func (c *DynamicFileServingContent) loadServingCert() error {
 	}
 
 	// check to see if we have a change. If the values are the same, do nothing.
-	existing, ok := c.servingCert.Load().(*certKeyContent)
+	existing, ok := c.certKeyPair.Load().(*certKeyContent)
 	if ok && existing != nil && existing.Equal(newCertKey) {
 		return nil
 	}
 
-	c.servingCert.Store(newCertKey)
+	c.certKeyPair.Store(newCertKey)
 
 	for _, listener := range c.listeners {
 		listener.Enqueue()
@@ -117,12 +117,12 @@ func (c *DynamicFileServingContent) loadServingCert() error {
 }
 
 // RunOnce runs a single sync loop
-func (c *DynamicFileServingContent) RunOnce() error {
-	return c.loadServingCert()
+func (c *DynamicCertKeyPairContent) RunOnce() error {
+	return c.loadCertKeyPair()
 }
 
 // Run starts the controller and blocks until stopCh is closed.
-func (c *DynamicFileServingContent) Run(workers int, stopCh <-chan struct{}) {
+func (c *DynamicCertKeyPairContent) Run(workers int, stopCh <-chan struct{}) {
 	defer utilruntime.HandleCrash()
 	defer c.queue.ShutDown()
 
@@ -143,19 +143,19 @@ func (c *DynamicFileServingContent) Run(workers int, stopCh <-chan struct{}) {
 	<-stopCh
 }
 
-func (c *DynamicFileServingContent) runWorker() {
+func (c *DynamicCertKeyPairContent) runWorker() {
 	for c.processNextWorkItem() {
 	}
 }
 
-func (c *DynamicFileServingContent) processNextWorkItem() bool {
+func (c *DynamicCertKeyPairContent) processNextWorkItem() bool {
 	dsKey, quit := c.queue.Get()
 	if quit {
 		return false
 	}
 	defer c.queue.Done(dsKey)
 
-	err := c.loadServingCert()
+	err := c.loadCertKeyPair()
 	if err == nil {
 		c.queue.Forget(dsKey)
 		return true
@@ -168,12 +168,12 @@ func (c *DynamicFileServingContent) processNextWorkItem() bool {
 }
 
 // Name is just an identifier
-func (c *DynamicFileServingContent) Name() string {
+func (c *DynamicCertKeyPairContent) Name() string {
 	return c.name
 }
 
-// CurrentCertKeyContent provides serving cert byte content
-func (c *DynamicFileServingContent) CurrentCertKeyContent() ([]byte, []byte) {
-	certKeyContent := c.servingCert.Load().(*certKeyContent)
+// CurrentCertKeyContent provides cert and key byte content
+func (c *DynamicCertKeyPairContent) CurrentCertKeyContent() ([]byte, []byte) {
+	certKeyContent := c.certKeyPair.Load().(*certKeyContent)
 	return certKeyContent.cert, certKeyContent.key
 }

staging/src/k8s.io/apiserver/pkg/server/dynamiccertificates/dynamic_sni_content.go

@@ -18,7 +18,7 @@ package dynamiccertificates
 
 // DynamicFileSNIContent provides a SNICertKeyContentProvider that can dynamically react to new file content
 type DynamicFileSNIContent struct {
-	*DynamicFileServingContent
+	*DynamicCertKeyPairContent
 	sniNames []string
 }
 
@@ -34,10 +34,10 @@ func NewDynamicSNIContentFromFiles(purpose, certFile, keyFile string, sniNames .
 	}
 
 	ret := &DynamicFileSNIContent{
-		DynamicFileServingContent: servingContent,
+		DynamicCertKeyPairContent: servingContent,
 		sniNames:                  sniNames,
 	}
-	if err := ret.loadServingCert(); err != nil {
+	if err := ret.loadCertKeyPair(); err != nil {
 		return nil, err
 	}