Dataplane-trust Adding mTLS and TLS To Activator

cmd/activator/main.go

@@ -18,7 +18,6 @@ package main
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"log"
@@ -32,6 +31,7 @@ import (
 	"go.uber.org/zap"
 
 	// Injection related imports.
+
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/injection"
 	"knative.dev/serving/pkg/activator"
@@ -145,7 +145,6 @@ func main() {
 	// (via keep-alive) to send real requests, avoiding needing an extra
 	// reconnect for the first request after the probe succeeds.
 	logger.Debugf("MaxIdleProxyConns: %d, MaxIdleProxyConnsPerHost: %d", env.MaxIdleProxyConns, env.MaxIdleProxyConnsPerHost)
-	transport := pkgnet.NewProxyAutoTransport(env.MaxIdleProxyConns, env.MaxIdleProxyConnsPerHost)
 
 	// Fetch networking configuration to determine whether EnableMeshPodAddressability
 	// is enabled or not.
@@ -158,23 +157,24 @@ func main() {
 		logger.Fatalw("Failed to construct network config", zap.Error(err))
 	}
 
-	// Enable TLS against queue-proxy when internal-encryption is enabled.
-	tlsEnabled := networkConfig.InternalEncryption
-
+	// Enable TLS both as client and as server if DataplaneTrust != TrustDisabled
+	tlsEnabled := networkConfig.DataplaneTrust != netcfg.TrustDisabled
+	var roundtrip http.RoundTripper
 	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")
-		certCache = certificate.NewCertCache(ctx)
-		transport = pkgnet.NewProxyAutoTLSTransport(env.MaxIdleProxyConns, env.MaxIdleProxyConnsPerHost, &certCache.TLSConf)
+		logger.Info("Dataplane trust %q is used", networkConfig.DataplaneTrust)
+		certCache = certificate.NewCertCache(ctx, networkConfig.DataplaneTrust)
+		roundtrip = activatorhandler.NewProxyAutoTLSTransport(env.MaxIdleProxyConns, env.MaxIdleProxyConnsPerHost, &certCache.ClientTLSConf)
+	} else {
+		roundtrip = activatorhandler.NewProxyAutoTransport(env.MaxIdleProxyConns, env.MaxIdleProxyConnsPerHost)
 	}
 
 	// Start throttler.
 	throttler := activatornet.NewThrottler(ctx, env.PodIP)
-	go throttler.Run(ctx, transport, networkConfig.EnableMeshPodAddressability, networkConfig.MeshCompatibilityMode)
+	go throttler.Run(ctx, roundtrip, networkConfig.EnableMeshPodAddressability, networkConfig.MeshCompatibilityMode)
 
 	oct := tracing.NewOpenCensusTracer(tracing.WithExporterFull(networking.ActivatorServiceName, env.PodIP, logger))
 	defer oct.Shutdown(context.Background())
@@ -189,7 +189,7 @@ func main() {
 
 	// Set up our config store
 	configMapWatcher := configmapinformer.NewInformedWatcher(kubeClient, system.Namespace())
-	configStore := activatorconfig.NewStore(logger, tracerUpdater)
+	configStore := activatorconfig.NewStore(logger, networkConfig.DataplaneTrust, tracerUpdater)
 	configStore.WatchConfigs(configMapWatcher)
 
 	statCh := make(chan []asmetrics.StatMessage)
@@ -208,7 +208,7 @@ func main() {
 
 	// Create activation handler chain
 	// Note: innermost handlers are specified first, ie. the last handler in the chain will be executed first
-	ah := activatorhandler.New(ctx, throttler, transport, networkConfig.EnableMeshPodAddressability, logger, tlsEnabled)
+	ah := activatorhandler.New(ctx, throttler, roundtrip, networkConfig.EnableMeshPodAddressability, logger, tlsEnabled)
 	ah = handler.NewTimeoutHandler(ah, "activator request timeout", func(r *http.Request) (time.Duration, time.Duration, time.Duration) {
 		if rev := activatorhandler.RevisionFrom(r.Context()); rev != nil {
 			var responseStartTimeout = 0 * time.Second
@@ -278,16 +278,13 @@ func main() {
 		}(name, server)
 	}
 
-	// Enable TLS server when internal-encryption is specified.
+	// Enable TLS server when DataPlaneTrust is not netcfg.TrustDisabled.
 	// At this moment activator with TLS does not disable HTTP.
 	// See also https://github.com/knative/serving/issues/12808.
 	if tlsEnabled {
 		name, server := "https", pkgnet.NewServer(":"+strconv.Itoa(networking.BackendHTTPSPort), ah)
 		go func(name string, s *http.Server) {
-			s.TLSConfig = &tls.Config{
-				MinVersion:     tls.VersionTLS12,
-				GetCertificate: certCache.GetCertificate,
-			}
+			s.TLSConfig = &certCache.ServerTLSConf
 			// 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

@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/pem"
+	"fmt"
 	"sync"
 
 	"go.uber.org/zap"
@@ -40,23 +41,26 @@ import (
 type CertCache struct {
 	secretInformer v1.SecretInformer
 	logger         *zap.SugaredLogger
+	trust          netcfg.Trust
 
-	certificate *tls.Certificate
-	TLSConf     tls.Config
+	certificate   *tls.Certificate
+	ClientTLSConf tls.Config
+	ServerTLSConf tls.Config
 
 	certificatesMux sync.RWMutex
 }
 
 // NewCertCache starts secretInformer.
-func NewCertCache(ctx context.Context) *CertCache {
+func NewCertCache(ctx context.Context, trust netcfg.Trust) *CertCache {
 	secretInformer := secretinformer.Get(ctx)
 
 	cr := &CertCache{
 		secretInformer: secretInformer,
 		logger:         logging.FromContext(ctx),
+		trust:          trust,
 	}
 
-	secret, err := cr.secretInformer.Lister().Secrets(system.Namespace()).Get(netcfg.ServingInternalCertName)
+	secret, err := cr.secretInformer.Lister().Secrets(system.Namespace()).Get(netcfg.ServingRoutingCertName)
 	if err != nil {
 		cr.logger.Warnw("failed to get secret", zap.Error(err))
 		return nil
@@ -65,7 +69,7 @@ func NewCertCache(ctx context.Context) *CertCache {
 	cr.updateCache(secret)
 
 	secretInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterWithNameAndNamespace(system.Namespace(), netcfg.ServingInternalCertName),
+		FilterFunc: controller.FilterWithNameAndNamespace(system.Namespace(), netcfg.ServingRoutingCertName),
 		Handler: cache.ResourceEventHandlerFuncs{
 			UpdateFunc: cr.handleCertificateUpdate,
 			AddFunc:    cr.handleCertificateAdd,
@@ -101,9 +105,32 @@ func (cr *CertCache) updateCache(secret *corev1.Secret) {
 	}
 	pool.AddCert(ca)
 
-	cr.TLSConf.RootCAs = pool
-	cr.TLSConf.ServerName = certificates.LegacyFakeDnsName
-	cr.TLSConf.MinVersion = tls.VersionTLS12
+	cr.ClientTLSConf.RootCAs = pool
+	cr.ClientTLSConf.ServerName = certificates.LegacyFakeDnsName
+	cr.ClientTLSConf.MinVersion = tls.VersionTLS12
+	cr.ClientTLSConf.Certificates = []tls.Certificate{cert}
+
+	cr.ServerTLSConf.MinVersion = tls.VersionTLS12
+	cr.ServerTLSConf.GetCertificate = cr.GetCertificate
+
+	switch cr.trust {
+	case netcfg.TrustIdentity, netcfg.TrustMutual:
+		cr.ServerTLSConf.ClientAuth = tls.RequireAndVerifyClientCert
+		cr.ServerTLSConf.ClientCAs = pool
+		cr.ServerTLSConf.VerifyConnection = func(cs tls.ConnectionState) error {
+			for _, match := range cs.PeerCertificates[0].DNSNames {
+				if match == "kn-routing-0" { // routingId not yet supported
+					return nil
+				}
+				//Until all ingresses work with updated dataplane certificates  - allow also any legacy certificate
+				if match == certificates.LegacyFakeDnsName {
+					return nil
+				}
+			}
+			cr.logger.Info("mTLS: Failed Client with DNSNames: %v\n", cs.PeerCertificates[0].DNSNames)
+			return fmt.Errorf("mTLS Failed to approve %v", cs.PeerCertificates[0].DNSNames)
+		}
+	}
 }
 
 func (cr *CertCache) handleCertificateUpdate(_, new interface{}) {

pkg/activator/certificate/cache_test.go

@@ -47,12 +47,13 @@ func fakeCertCache(ctx context.Context) *CertCache {
 	cr := &CertCache{
 		secretInformer: secretInformer,
 		certificate:    nil,
-		TLSConf:        tls.Config{},
+		ClientTLSConf:  tls.Config{},
 		logger:         logging.FromContext(ctx),
+		trust:          netcfg.TrustMutual,
 	}
 
 	secretInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterWithNameAndNamespace(system.Namespace(), netcfg.ServingInternalCertName),
+		FilterFunc: controller.FilterWithNameAndNamespace(system.Namespace(), netcfg.ServingRoutingCertName),
 		Handler: cache.ResourceEventHandlerFuncs{
 			UpdateFunc: cr.handleCertificateUpdate,
 			AddFunc:    cr.handleCertificateAdd,
@@ -62,7 +63,28 @@ func fakeCertCache(ctx context.Context) *CertCache {
 	return cr
 }
 
-func TestReconcile(t *testing.T) {
+func routingCertCache(ctx context.Context) *CertCache {
+	secretInformer := fakesecretinformer.Get(ctx)
+
+	cr := &CertCache{
+		secretInformer: secretInformer,
+		certificate:    nil,
+		ClientTLSConf:  tls.Config{},
+		logger:         logging.FromContext(ctx),
+	}
+
+	secretInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterWithNameAndNamespace(system.Namespace(), netcfg.ServingRoutingCertName),
+		Handler: cache.ResourceEventHandlerFuncs{
+			UpdateFunc: cr.handleCertificateUpdate,
+			AddFunc:    cr.handleCertificateAdd,
+		},
+	})
+
+	return cr
+}
+
+func TestFakeReconcile(t *testing.T) {
 	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
 	cr := fakeCertCache(ctx)
 
@@ -78,7 +100,105 @@ func TestReconcile(t *testing.T) {
 
 	secret := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      netcfg.ServingInternalCertName,
+			Name:      netcfg.ServingRoutingCertName,
+			Namespace: system.Namespace(),
+		},
+		Data: map[string][]byte{
+			certificates.CaCertName:     ca,
+			certificates.PrivateKeyName: tlsKey,
+			certificates.CertName:       tlsCrt,
+		},
+	}
+
+	fakekubeclient.Get(ctx).CoreV1().Secrets(system.Namespace()).Create(ctx, secret, metav1.CreateOptions{})
+	fakesecretinformer.Get(ctx).Informer().GetIndexer().Add(secret)
+
+	// Wait for the resources to be created and the handler is called.
+	if err := wait.PollImmediate(10*time.Millisecond, 2*time.Second, func() (bool, error) {
+		// To access cert.Certificate, take a lock.
+		cr.certificatesMux.RLock()
+		defer cr.certificatesMux.RUnlock()
+		cert, _ := cr.GetCertificate(nil)
+		return cert != nil, nil
+	}); err != nil {
+		t.Fatal("timeout to get the secret:", err)
+	}
+
+	cs := tls.ConnectionState{PeerCertificates: []*x509.Certificate{{DNSNames: []string{"ddd", "xxx", "kn-routing-0", "ddd", "xxx"}}}}
+	if verifyErr := cr.ServerTLSConf.VerifyConnection(cs); verifyErr != nil {
+		t.Error("cr.ServerTLSConf.VerifyConnection: expected to find kn-routing-0")
+	}
+
+	cs = tls.ConnectionState{PeerCertificates: []*x509.Certificate{{DNSNames: []string{"ddd", "xxx", certificates.LegacyFakeDnsName, "ddd", "xxx"}}}}
+	if verifyErr := cr.ServerTLSConf.VerifyConnection(cs); verifyErr != nil {
+		t.Error("cr.ServerTLSConf.VerifyConnection: expected to find kn-routing-0")
+	}
+	cs = tls.ConnectionState{PeerCertificates: []*x509.Certificate{{DNSNames: []string{"ddd", "xxx"}}}}
+	if verifyErr := cr.ServerTLSConf.VerifyConnection(cs); verifyErr == nil {
+		t.Error("cr.ServerTLSConf.VerifyConnection: expected error")
+	}
+	cs = tls.ConnectionState{PeerCertificates: []*x509.Certificate{{DNSNames: []string{}}}}
+	if verifyErr := cr.ServerTLSConf.VerifyConnection(cs); verifyErr == nil {
+		t.Error("cr.ServerTLSConf.VerifyConnection: expected error")
+	}
+
+	// Update cert and key but keep using old CA, then the error is expected.
+	secret.Data[certificates.CertName] = newTLSCrt
+	secret.Data[certificates.PrivateKeyName] = newTLSKey
+	newCert, _ := tls.X509KeyPair(newTLSCrt, newTLSKey)
+
+	fakekubeclient.Get(ctx).CoreV1().Secrets(system.Namespace()).Update(ctx, secret, metav1.UpdateOptions{})
+	if err := wait.PollImmediate(10*time.Millisecond, 5*time.Second, func() (bool, error) {
+		// To access cert.Certificate, take a lock.
+		cr.certificatesMux.RLock()
+		defer cr.certificatesMux.RUnlock()
+		cert, err := cr.GetCertificate(nil)
+		return err == nil && reflect.DeepEqual(newCert.Certificate, cert.Certificate), nil
+	}); err != nil {
+		t.Fatalf("timeout to update the cert: %v", err)
+	}
+
+	// Update CA, now the error is gone.
+	secret.Data[certificates.CaCertName] = newCA
+
+	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: %v", zap.Error(err))
+		return
+	}
+
+	pool.AddCert(ca)
+
+	fakekubeclient.Get(ctx).CoreV1().Secrets(system.Namespace()).Update(ctx, secret, metav1.UpdateOptions{})
+	if err := wait.PollImmediate(10*time.Millisecond, 10*time.Second, func() (bool, error) {
+		// To access cr.TLSConf.RootCAs, take a lock.
+		cr.certificatesMux.RLock()
+		defer cr.certificatesMux.RUnlock()
+		return err == nil && pool.Equal(cr.ClientTLSConf.RootCAs), nil
+	}); err != nil {
+		t.Fatalf("timeout to update the cert: %v", err)
+	}
+}
+
+func TestRoutingReconcile(t *testing.T) {
+	ctx, cancel, informers := rtesting.SetupFakeContextWithCancel(t)
+	cr := routingCertCache(ctx)
+
+	waitInformers, err := rtesting.RunAndSyncInformers(ctx, informers...)
+	if err != nil {
+		cancel()
+		t.Fatal("failed to start informers:", err)
+	}
+	t.Cleanup(func() {
+		cancel()
+		waitInformers()
+	})
+
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      netcfg.ServingRoutingCertName,
 			Namespace: system.Namespace(),
 		},
 		Data: map[string][]byte{
@@ -136,7 +256,7 @@ func TestReconcile(t *testing.T) {
 		// To access cr.TLSConf.RootCAs, take a lock.
 		cr.certificatesMux.RLock()
 		defer cr.certificatesMux.RUnlock()
-		return err == nil && pool.Equal(cr.TLSConf.RootCAs), nil
+		return err == nil && pool.Equal(cr.ClientTLSConf.RootCAs), nil
 	}); err != nil {
 		t.Fatalf("timeout to update the cert: %v", err)
 	}

pkg/activator/config/store.go

@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"go.uber.org/atomic"
+	netcfg "knative.dev/networking/pkg/config"
 	"knative.dev/pkg/configmap"
 	tracingconfig "knative.dev/pkg/tracing/config"
 )
@@ -29,6 +30,7 @@ type cfgKey struct{}
 // Config is the configuration for the activator.
 type Config struct {
 	Tracing *tracingconfig.Config
+	Trust   netcfg.Trust
 }
 
 // FromContext obtains a Config injected into the passed context.
@@ -45,14 +47,15 @@ type Store struct {
 }
 
 // NewStore creates a new configuration Store.
-func NewStore(logger configmap.Logger, onAfterStore ...func(name string, value interface{})) *Store {
+func NewStore(logger configmap.Logger, trust netcfg.Trust, onAfterStore ...func(name string, value interface{})) *Store {
 	s := &Store{}
 
 	// Append an update function to run after a ConfigMap has updated to update the
 	// current state of the Config.
 	onAfterStore = append(onAfterStore, func(_ string, _ interface{}) {
 		s.current.Store(&Config{
 			Tracing: s.UntypedLoad(tracingconfig.ConfigName).(*tracingconfig.Config).DeepCopy(),
+			Trust:   trust,
 		})
 	})
 	s.UntypedStore = configmap.NewUntypedStore(

pkg/activator/config/store_test.go

@@ -37,7 +37,7 @@ var tracingConfig = &corev1.ConfigMap{
 
 func TestStore(t *testing.T) {
 	logger := ltesting.TestLogger(t)
-	store := NewStore(logger)
+	store := NewStore(logger, "Disabled")
 	store.OnConfigChanged(tracingConfig)
 
 	ctx := store.ToContext(context.Background())
@@ -68,7 +68,7 @@ func TestStore(t *testing.T) {
 
 func BenchmarkStoreToContext(b *testing.B) {
 	logger := ltesting.TestLogger(b)
-	store := NewStore(logger)
+	store := NewStore(logger, "Disabled")
 	store.OnConfigChanged(tracingConfig)
 
 	b.Run("sequential", func(b *testing.B) {