Merge pull request #124574 from zhangweikop/master

enable kubelet server to dynamically load tls certificate files

pkg/features/kube_features.go

@@ -640,6 +640,14 @@ const (
 	// Allow almost all printable ASCII characters in environment variables
 	RelaxedEnvironmentVariableValidation featuregate.Feature = "RelaxedEnvironmentVariableValidation"
 
+	// owner: @zhangweikop
+	// beta: v1.31
+	//
+	// Enable kubelet tls server to update certificate if the specified certificate files are changed.
+	// This feature is useful when specifying tlsCertFile & tlsPrivateKeyFile in kubelet Configuration.
+	// No effect for other cases such as using serverTLSbootstap.
+	ReloadKubeletServerCertificateFile featuregate.Feature = "ReloadKubeletServerCertificateFile"
+
 	// owner: @mikedanese
 	// alpha: v1.7
 	// beta: v1.12
@@ -1117,6 +1125,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	RelaxedEnvironmentVariableValidation: {Default: false, PreRelease: featuregate.Alpha},
 
+	ReloadKubeletServerCertificateFile: {Default: true, PreRelease: featuregate.Beta},
+
 	RotateKubeletServerCertificate: {Default: true, PreRelease: featuregate.Beta},
 
 	RuntimeClassInImageCriAPI: {Default: false, PreRelease: featuregate.Alpha},

pkg/kubelet/certificate/kubelet.go

@@ -17,6 +17,7 @@ limitations under the License.
 package certificate
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"crypto/x509/pkix"
@@ -24,15 +25,18 @@ import (
 	"math"
 	"net"
 	"sort"
+	"sync/atomic"
 	"time"
 
 	certificates "k8s.io/api/certificates/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/util/certificate"
 	compbasemetrics "k8s.io/component-base/metrics"
 	"k8s.io/component-base/metrics/legacyregistry"
+	"k8s.io/klog/v2"
 	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
 	"k8s.io/kubernetes/pkg/kubelet/metrics"
 	netutils "k8s.io/utils/net"
@@ -234,3 +238,66 @@ func NewKubeletClientCertificateManager(
 
 	return m, nil
 }
+
+// NewKubeletServerCertificateDynamicFileManager creates a certificate manager based on reading and watching certificate and key files.
+// The returned struct implements certificate.Manager interface, enabling using it like other CertificateManager in this package.
+// But the struct doesn't communicate with API server to perform certificate request at all.
+func NewKubeletServerCertificateDynamicFileManager(certFile, keyFile string) (certificate.Manager, error) {
+	c, err := dynamiccertificates.NewDynamicServingContentFromFiles("kubelet-server-cert-files", certFile, keyFile)
+	if err != nil {
+		return nil, fmt.Errorf("unable to set up dynamic certificate manager for kubelet server cert files: %w", err)
+	}
+	m := &kubeletServerCertificateDynamicFileManager{
+		dynamicCertificateContent: c,
+		certFile:                  certFile,
+		keyFile:                   keyFile,
+	}
+	m.Enqueue()
+	c.AddListener(m)
+	return m, nil
+}
+
+// kubeletServerCertificateDynamicFileManager uses a dynamic CertKeyContentProvider based on cert and key files.
+type kubeletServerCertificateDynamicFileManager struct {
+	cancelFn                  context.CancelFunc
+	certFile                  string
+	keyFile                   string
+	dynamicCertificateContent *dynamiccertificates.DynamicCertKeyPairContent
+	currentTLSCertificate     atomic.Pointer[tls.Certificate]
+}
+
+// Enqueue implements the functions to be notified when the serving cert content changes.
+func (m *kubeletServerCertificateDynamicFileManager) Enqueue() {
+	certContent, keyContent := m.dynamicCertificateContent.CurrentCertKeyContent()
+	cert, err := tls.X509KeyPair(certContent, keyContent)
+	if err != nil {
+		klog.ErrorS(err, "invalid certificate and key pair from file", "certFile", m.certFile, "keyFile", m.keyFile)
+		return
+	}
+	m.currentTLSCertificate.Store(&cert)
+	klog.V(4).InfoS("loaded certificate and key pair in kubelet server certificate manager", "certFile", m.certFile, "keyFile", m.keyFile)
+}
+
+// Current returns the last valid certificate key pair loaded from files.
+func (m *kubeletServerCertificateDynamicFileManager) Current() *tls.Certificate {
+	return m.currentTLSCertificate.Load()
+}
+
+// Start starts watching the certificate and key files
+func (m *kubeletServerCertificateDynamicFileManager) Start() {
+	var ctx context.Context
+	ctx, m.cancelFn = context.WithCancel(context.Background())
+	go m.dynamicCertificateContent.Run(ctx, 1)
+}
+
+// Stop stops watching the certificate and key files
+func (m *kubeletServerCertificateDynamicFileManager) Stop() {
+	if m.cancelFn != nil {
+		m.cancelFn()
+	}
+}
+
+// ServerHealthy always returns true since the file manager doesn't communicate with any server
+func (m *kubeletServerCertificateDynamicFileManager) ServerHealthy() bool {
+	return true
+}

pkg/kubelet/certificate/kubelet_test.go

@@ -17,11 +17,19 @@ limitations under the License.
 package certificate
 
 import (
+	"bytes"
+	"context"
+	"fmt"
 	"net"
+	"os"
+	"path/filepath"
 	"reflect"
 	"testing"
+	"time"
 
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/cert"
 	netutils "k8s.io/utils/net"
 )
 
@@ -100,3 +108,156 @@ func TestAddressesToHostnamesAndIPs(t *testing.T) {
 		})
 	}
 }
+
+func removeThenCreate(name string, data []byte, perm os.FileMode) error {
+	if err := os.Remove(name); err != nil {
+		if !os.IsNotExist(err) {
+			return err
+		}
+	}
+	return os.WriteFile(name, data, perm)
+}
+
+func createCertAndKeyFiles(certDir string) (string, string, error) {
+	cert, key, err := cert.GenerateSelfSignedCertKey("k8s.io", nil, nil)
+	if err != nil {
+		return "", "", nil
+	}
+
+	certPath := filepath.Join(certDir, "kubelet.cert")
+	keyPath := filepath.Join(certDir, "kubelet.key")
+	if err := removeThenCreate(certPath, cert, os.FileMode(0644)); err != nil {
+		return "", "", err
+	}
+
+	if err := removeThenCreate(keyPath, key, os.FileMode(0600)); err != nil {
+		return "", "", err
+	}
+
+	return certPath, keyPath, nil
+}
+
+// createCertAndKeyFilesUsingRename creates cert and key files under a parent dir `identity` as
+// <certDir>/identity/kubelet.cert, <certDir>/identity/kubelet.key
+func createCertAndKeyFilesUsingRename(certDir string) (string, string, error) {
+	cert, key, err := cert.GenerateSelfSignedCertKey("k8s.io", nil, nil)
+	if err != nil {
+		return "", "", nil
+	}
+
+	var certKeyPathFn = func(dataDir string) (string, string, string) {
+		outputDir := filepath.Join(certDir, dataDir)
+		return outputDir, filepath.Join(outputDir, "kubelet.cert"), filepath.Join(outputDir, "kubelet.key")
+	}
+
+	writeDir, writeCertPath, writeKeyPath := certKeyPathFn("identity.tmp")
+	if err := os.Mkdir(writeDir, 0777); err != nil {
+		return "", "", err
+	}
+
+	if err := removeThenCreate(writeCertPath, cert, os.FileMode(0644)); err != nil {
+		return "", "", err
+	}
+
+	if err := removeThenCreate(writeKeyPath, key, os.FileMode(0600)); err != nil {
+		return "", "", err
+	}
+
+	targetDir, certPath, keyPath := certKeyPathFn("identity")
+	if err := os.RemoveAll(targetDir); err != nil {
+		if !os.IsNotExist(err) {
+			return "", "", err
+		}
+	}
+	if err := os.Rename(writeDir, targetDir); err != nil {
+		return "", "", err
+	}
+
+	return certPath, keyPath, nil
+}
+
+func TestKubeletServerCertificateFromFiles(t *testing.T) {
+	// test two common ways of certificate file updates:
+	// 1. delete and write the cert and key files directly
+	// 2. create the cert and key files under a child dir and perform dir rename during update
+	tests := []struct {
+		name      string
+		useRename bool
+	}{
+		{
+			name:      "remove and create",
+			useRename: false,
+		},
+		{
+			name:      "rename cert dir",
+			useRename: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			createFn := createCertAndKeyFiles
+			if tt.useRename {
+				createFn = createCertAndKeyFilesUsingRename
+			}
+
+			certDir := t.TempDir()
+			certPath, keyPath, err := createFn(certDir)
+			if err != nil {
+				t.Fatalf("Unable to setup cert files: %v", err)
+			}
+
+			m, err := NewKubeletServerCertificateDynamicFileManager(certPath, keyPath)
+			if err != nil {
+				t.Fatalf("Unable to create certificte provider: %v", err)
+			}
+
+			m.Start()
+			defer m.Stop()
+
+			c := m.Current()
+			if c == nil {
+				t.Fatal("failed to provide valid certificate")
+			}
+			time.Sleep(100 * time.Millisecond)
+			c2 := m.Current()
+			if c2 == nil {
+				t.Fatal("failed to provide valid certificate")
+			}
+			if c2 != c {
+				t.Errorf("expected the same loaded certificate object when there is no cert file change, got different")
+			}
+
+			// simulate certificate files updated in the background
+			if _, _, err := createFn(certDir); err != nil {
+				t.Fatalf("got errors when rotating certificate files in the test: %v", err)
+			}
+
+			err = wait.PollUntilContextTimeout(context.Background(),
+				100*time.Millisecond, 10*time.Second, true,
+				func(_ context.Context) (bool, error) {
+					c3 := m.Current()
+					if c3 == nil {
+						return false, fmt.Errorf("expected valid certificate regardless of file changes, but got nil")
+					}
+					if bytes.Equal(c.Certificate[0], c3.Certificate[0]) {
+						t.Logf("loaded certificate is not updated")
+						return false, nil
+					}
+					return true, nil
+				})
+			if err != nil {
+				t.Errorf("failed to provide the updated certificate after file changes: %v", err)
+			}
+
+			if err = os.Remove(certPath); err != nil {
+				t.Errorf("could not delete file in order to perform test")
+			}
+
+			time.Sleep(1 * time.Second)
+			if m.Current() == nil {
+				t.Errorf("expected the manager still provides cached content when certificate file was not available")
+			}
+		})
+	}
+}

pkg/kubelet/kubelet.go

@@ -781,17 +781,28 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	}
 	klet.imageManager = imageManager
 
-	if kubeCfg.ServerTLSBootstrap && kubeDeps.TLSOptions != nil && utilfeature.DefaultFeatureGate.Enabled(features.RotateKubeletServerCertificate) {
-		klet.serverCertificateManager, err = kubeletcertificate.NewKubeletServerCertificateManager(klet.kubeClient, kubeCfg, klet.nodeName, klet.getLastObservedNodeAddresses, certDirectory)
-		if err != nil {
-			return nil, fmt.Errorf("failed to initialize certificate manager: %v", err)
-		}
-		kubeDeps.TLSOptions.Config.GetCertificate = func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
-			cert := klet.serverCertificateManager.Current()
-			if cert == nil {
-				return nil, fmt.Errorf("no serving certificate available for the kubelet")
+	if kubeDeps.TLSOptions != nil {
+		if kubeCfg.ServerTLSBootstrap && utilfeature.DefaultFeatureGate.Enabled(features.RotateKubeletServerCertificate) {
+			klet.serverCertificateManager, err = kubeletcertificate.NewKubeletServerCertificateManager(klet.kubeClient, kubeCfg, klet.nodeName, klet.getLastObservedNodeAddresses, certDirectory)
+			if err != nil {
+				return nil, fmt.Errorf("failed to initialize certificate manager: %w", err)
+			}
+
+		} else if kubeDeps.TLSOptions.CertFile != "" && kubeDeps.TLSOptions.KeyFile != "" && utilfeature.DefaultFeatureGate.Enabled(features.ReloadKubeletServerCertificateFile) {
+			klet.serverCertificateManager, err = kubeletcertificate.NewKubeletServerCertificateDynamicFileManager(kubeDeps.TLSOptions.CertFile, kubeDeps.TLSOptions.KeyFile)
+			if err != nil {
+				return nil, fmt.Errorf("failed to initialize file based certificate manager: %w", err)
+			}
+		}
+
+		if klet.serverCertificateManager != nil {
+			kubeDeps.TLSOptions.Config.GetCertificate = func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
+				cert := klet.serverCertificateManager.Current()
+				if cert == nil {
+					return nil, fmt.Errorf("no serving certificate available for the kubelet")
+				}
+				return cert, nil
 			}
-			return cert, nil
 		}
 	}