Load certificate when they are updated when internal-encryption is enabled

cmd/activator/main.go

@@ -19,7 +19,6 @@ package main
 import (
 	"context"
 	"crypto/tls"
-	"crypto/x509"
 	"errors"
 	"fmt"
 	"log"
@@ -41,7 +40,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 
-	"knative.dev/control-protocol/pkg/certificates"
 	network "knative.dev/networking/pkg"
 	netcfg "knative.dev/networking/pkg/config"
 	netprobe "knative.dev/networking/pkg/http/probe"
@@ -60,6 +58,7 @@ import (
 	tracingconfig "knative.dev/pkg/tracing/config"
 	"knative.dev/pkg/version"
 	"knative.dev/pkg/websocket"
+	"knative.dev/serving/pkg/activator/certificate"
 	activatorconfig "knative.dev/serving/pkg/activator/config"
 	activatorhandler "knative.dev/serving/pkg/activator/handler"
 	activatornet "knative.dev/serving/pkg/activator/net"
@@ -162,32 +161,15 @@ func main() {
 	// Enable TLS against queue-proxy when internal-encryption is enabled.
 	tlsEnabled := networkConfig.InternalEncryption
 
+	var certCache *certificate.CertCache
+
 	// Enable TLS client when queue-proxy-ca is specified.
 	// At this moment activator with TLS does not disable HTTP.
 	// See also https://github.com/knative/serving/issues/12808.
 	if tlsEnabled {
 		logger.Info("Internal Encryption is enabled")
-		caSecret, err := kubeClient.CoreV1().Secrets(system.Namespace()).Get(ctx, netcfg.ServingInternalCertName, metav1.GetOptions{})
-		if err != nil {
-			logger.Fatalw("Failed to get secret", zap.Error(err))
-		}
-
-		pool, err := x509.SystemCertPool()
-		if err != nil {
-			pool = x509.NewCertPool()
-		}
-
-		if ok := pool.AppendCertsFromPEM(caSecret.Data[certificates.CaCertName]); !ok {
-			logger.Fatalw("Failed to append ca cert to the RootCAs")
-		}
-
-		tlsConf := &tls.Config{
-			RootCAs:            pool,
-			InsecureSkipVerify: false,
-			ServerName:         certificates.FakeDnsName,
-			MinVersion:         tls.VersionTLS12,
-		}
-		transport = pkgnet.NewProxyAutoTLSTransport(env.MaxIdleProxyConns, env.MaxIdleProxyConnsPerHost, tlsConf)
+		certCache = certificate.NewCertCache(ctx)
+		transport = pkgnet.NewProxyAutoTLSTransport(env.MaxIdleProxyConns, env.MaxIdleProxyConnsPerHost, &certCache.TLSConf)
 	}
 
 	// Start throttler.
@@ -300,20 +282,12 @@ func main() {
 	// At this moment activator with TLS does not disable HTTP.
 	// See also https://github.com/knative/serving/issues/12808.
 	if tlsEnabled {
-		secret, err := kubeClient.CoreV1().Secrets(system.Namespace()).Get(ctx, netcfg.ServingInternalCertName, metav1.GetOptions{})
-		if err != nil {
-			logger.Fatalw("failed to get secret", zap.Error(err))
-		}
-		cert, err := tls.X509KeyPair(secret.Data[certificates.CertName], secret.Data[certificates.PrivateKeyName])
-		if err != nil {
-			logger.Fatalw("failed to load certs", zap.Error(err))
-		}
-
-		// TODO: Implement the secret (certificate) rotation like knative.dev/pkg/webhook/certificates/.
-		// Also, the current activator must be restarted when updating the secret.
 		name, server := "https", pkgnet.NewServer(":"+strconv.Itoa(networking.BackendHTTPSPort), ah)
 		go func(name string, s *http.Server) {
-			s.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12}
+			s.TLSConfig = &tls.Config{
+				MinVersion:     tls.VersionTLS12,
+				GetCertificate: certCache.GetCertificate,
+			}
 			// Don't forward ErrServerClosed as that indicates we're already shutting down.
 			if err := s.ListenAndServeTLS("", ""); err != nil && !errors.Is(err, http.ErrServerClosed) {
 				errCh <- fmt.Errorf("%s server failed: %w", name, err)

pkg/activator/certificate/cache.go

@@ -0,0 +1,116 @@
+/*
+Copyright 2023 The Knative 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 certificate
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"sync"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/tools/cache"
+
+	"knative.dev/control-protocol/pkg/certificates"
+	netcfg "knative.dev/networking/pkg/config"
+	"knative.dev/pkg/controller"
+	secretinformer "knative.dev/pkg/injection/clients/namespacedkube/informers/core/v1/secret"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/system"
+)
+
+// CertCache caches certificates and CA pool.
+type CertCache struct {
+	secretInformer v1.SecretInformer
+	logger         *zap.SugaredLogger
+
+	certificate *tls.Certificate
+	TLSConf     tls.Config
+
+	certificatesMux sync.RWMutex
+}
+
+// NewCertCache starts secretInformer.
+func NewCertCache(ctx context.Context) *CertCache {
+	secretInformer := secretinformer.Get(ctx)
+
+	cr := &CertCache{
+		secretInformer: secretInformer,
+		logger:         logging.FromContext(ctx),
+	}
+
+	secret, err := cr.secretInformer.Lister().Secrets(system.Namespace()).Get(netcfg.ServingInternalCertName)
+	if err != nil {
+		cr.logger.Warnw("failed to get secret", zap.Error(err))
+		return nil
+	}
+
+	cr.updateCache(secret)
+
+	secretInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterWithNameAndNamespace(system.Namespace(), netcfg.ServingInternalCertName),
+		Handler: cache.ResourceEventHandlerFuncs{
+			UpdateFunc: cr.handleCertificateUpdate,
+			AddFunc:    cr.handleCertificateAdd,
+		},
+	})
+
+	return cr
+}
+
+func (cr *CertCache) handleCertificateAdd(added interface{}) {
+	if secret, ok := added.(*corev1.Secret); ok {
+		cr.updateCache(secret)
+	}
+}
+
+func (cr *CertCache) updateCache(secret *corev1.Secret) {
+	cr.certificatesMux.Lock()
+	defer cr.certificatesMux.Unlock()
+
+	cert, err := tls.X509KeyPair(secret.Data[certificates.CertName], secret.Data[certificates.PrivateKeyName])
+	if err != nil {
+		cr.logger.Warnw("failed to parse secret", zap.Error(err))
+		return
+	}
+	cr.certificate = &cert
+
+	pool := x509.NewCertPool()
+	block, _ := pem.Decode(secret.Data[certificates.CaCertName])
+	ca, err := x509.ParseCertificate(block.Bytes)
+	if err != nil {
+		cr.logger.Warnw("failed to parse CA", zap.Error(err))
+		return
+	}
+	pool.AddCert(ca)
+
+	cr.TLSConf.RootCAs = pool
+	cr.TLSConf.ServerName = certificates.LegacyFakeDnsName
+	cr.TLSConf.MinVersion = tls.VersionTLS12
+}
+
+func (cr *CertCache) handleCertificateUpdate(_, new interface{}) {
+	cr.handleCertificateAdd(new)
+}
+
+// GetCertificate returns the cached certificates.
+func (cr *CertCache) GetCertificate(_ *tls.ClientHelloInfo) (*tls.Certificate, error) {
+	return cr.certificate, nil
+}