Queue multi-level trust

pkg/queue/sharedmain/cache.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2018 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 sharedmain
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"fmt"
+	"os"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+
+	"knative.dev/control-protocol/pkg/certificates"
+	netcfg "knative.dev/networking/pkg/config"
+	"knative.dev/pkg/logging"
+)
+
+type CertCache struct {
+	logger      *zap.SugaredLogger
+	trustConfig netcfg.Trust
+
+	certificate   *tls.Certificate
+	ServerTLSConf tls.Config
+
+	certificatesMux sync.RWMutex
+	caCerts         []*x509.Certificate
+	caLastUpdate    time.Time
+}
+
+func (cr *CertCache) init() {
+	cr.ServerTLSConf.MinVersion = tls.VersionTLS12
+	cr.ServerTLSConf.ClientCAs = x509.NewCertPool()
+	cr.ServerTLSConf.GetCertificate = cr.GetCertificate
+	switch cr.trustConfig {
+	case netcfg.TrustIdentity, netcfg.TrustMutual:
+		cr.ServerTLSConf.ClientAuth = tls.RequireAndVerifyClientCert
+		cr.ServerTLSConf.VerifyConnection = func(cs tls.ConnectionState) error {
+			if len(cs.PeerCertificates) == 0 {
+				// Should never happen on a server side
+				cr.logger.Info("mTLS: Failed to verify client connection. Certificate is missing\n")
+				return fmt.Errorf("mTLS: Failed to verify client connection. Certificate is missing")
+			}
+			for _, match := range cs.PeerCertificates[0].DNSNames {
+				// Activator currently supports a single routingId which is the default "0"
+				// Working with other routingId is not yet implemented
+				if match == certificates.DataPlaneRoutingName("0") {
+					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 to verify client connection for DNSNames: %v\n", cs.PeerCertificates[0].DNSNames)
+			return fmt.Errorf("mTLS: Failed to verify client connection for DNSNames: %v", cs.PeerCertificates[0].DNSNames)
+		}
+	}
+}
+
+// NewCertCache starts secretInformer.
+func NewCertCache(ctx context.Context, trust netcfg.Trust) *CertCache {
+	cr := &CertCache{
+		logger:      logging.FromContext(ctx),
+		trustConfig: trust,
+	}
+	cr.init()
+
+	cr.updateCache()
+
+	return cr
+}
+
+func (cr *CertCache) updateCache() {
+	info, err := os.Stat(caPath)
+	if err != nil {
+		cr.logger.Warnw("failed to stat secret ca", zap.Error(err))
+		return
+	}
+	if cr.caLastUpdate == info.ModTime() {
+		return
+	}
+	cr.logger.Info("Secret reloaded")
+
+	cr.certificatesMux.Lock()
+	defer cr.certificatesMux.Unlock()
+
+	cr.caLastUpdate = info.ModTime()
+
+	//caPath
+	caBlock, err := os.ReadFile(caPath)
+	if err != nil {
+		cr.logger.Warnw("failed to parse secret ca", zap.Error(err))
+		return
+	}
+	block, _ := pem.Decode(caBlock)
+	if block == nil {
+		cr.logger.Warnw("failed to parse CA")
+		return
+	}
+	ca, err := x509.ParseCertificate(block.Bytes)
+	if err != nil {
+		cr.logger.Warnw("failed to parse CA", zap.Error(err))
+		return
+	}
+
+	cert, err := tls.LoadX509KeyPair(certPath, keyPath)
+	if err != nil {
+		cr.logger.Warnw("failed to parse secret", zap.Error(err))
+		return
+	}
+
+	cr.certificate = &cert
+	for _, usedCa := range cr.caCerts {
+		if usedCa.Equal(ca) {
+			return
+		}
+	}
+	cr.caCerts = append(cr.caCerts, ca)
+	cr.ServerTLSConf.ClientCAs.AddCert(ca)
+}
+
+// GetCertificate returns the cached certificates.
+func (cr *CertCache) GetCertificate(_ *tls.ClientHelloInfo) (*tls.Certificate, error) {
+	cr.updateCache()
+	return cr.certificate, nil
+}

pkg/queue/sharedmain/main.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	"knative.dev/control-protocol/pkg/certificates"
+	netcfg "knative.dev/networking/pkg/config"
 	netstats "knative.dev/networking/pkg/http/stats"
 	pkglogging "knative.dev/pkg/logging"
 	"knative.dev/pkg/logging/logkey"
@@ -65,6 +66,9 @@ const (
 	// keyPath is the path for the server certificate key mounted by queue-proxy.
 	keyPath = queue.CertDirectory + "/" + certificates.PrivateKeyName
 
+	// caPath is the path for the server certificate ca mounted by queue-proxy.
+	caPath = queue.CertDirectory + "/" + certificates.CaCertName
+
 	// PodInfoAnnotationsPath is an exported path for the annotations file
 	// This path is used by QP Options (Extensions).
 	PodInfoAnnotationsPath = queue.PodInfoDirectory + "/" + queue.PodInfoAnnotationsFilename
@@ -126,6 +130,9 @@ type Env struct {
 
 	// ServingPodIP is the pod ip address
 	ServingPodIP string `split_words:"true" required:"true"`
+
+	// TrustConfig is the level of Trust used
+	TrustConfig string `split_words:"true" required:"true"`
 }
 
 // Defaults provides Options (QP Extensions) with the default bahaviour of QP
@@ -250,9 +257,11 @@ func Main(opts ...Option) error {
 		"admin": adminServer(":"+strconv.Itoa(networking.QueueAdminPort), adminHandler),
 	}
 
+	var certCache *CertCache
 	if tlsEnabled {
 		// Drop admin http server since the admin TLS server is listening on the same port
 		delete(httpServers, "admin")
+		certCache = NewCertCache(d.Ctx, netcfg.Trust(d.Env.TrustConfig))
 	} else {
 		tlsServers = map[string]*http.Server{}
 	}
@@ -271,9 +280,10 @@ func Main(opts ...Option) error {
 	}
 	for name, server := range tlsServers {
 		go func(name string, s *http.Server) {
+			s.TLSConfig = &certCache.ServerTLSConf
 			// Don't forward ErrServerClosed as that indicates we're already shutting down.
 			logger.Info("Starting tls server ", name, s.Addr)
-			if err := s.ListenAndServeTLS(certPath, keyPath); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			if err := s.ListenAndServeTLS("", ""); err != nil && !errors.Is(err, http.ErrServerClosed) {
 				errCh <- fmt.Errorf("%s server failed to serve: %w", name, err)
 			}
 		}(name, server)

pkg/reconciler/revision/resources/deploy.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	netcfg "knative.dev/networking/pkg/config"
 	netheader "knative.dev/networking/pkg/http/header"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/ptr"
@@ -193,7 +194,7 @@ func makePodSpec(rev *v1.Revision, cfg *config.Config) (*corev1.PodSpec, error)
 		extraVolumes = append(extraVolumes, *tokenVolume)
 	}
 
-	if cfg.Network.InternalEncryption {
+	if cfg.Network.DataplaneTrust != netcfg.TrustDisabled {
 		queueContainer.VolumeMounts = append(queueContainer.VolumeMounts, varCertVolumeMount)
 		extraVolumes = append(extraVolumes, certVolume(networking.ServingCertName))
 	}

pkg/reconciler/revision/resources/deploy_test.go

@@ -129,6 +129,10 @@ var (
 			ValueFrom: &corev1.EnvVarSource{
 				FieldRef: &corev1.ObjectFieldSelector{FieldPath: "status.podIP"},
 			},
+		}, {
+			Name:  "TRUST_CONFIG",
+			Value: "disabled",
+			// No logging configuration
 		}, {
 			Name: "SERVING_LOGGING_CONFIG",
 			// No logging configuration

pkg/reconciler/revision/resources/queue.go

@@ -310,6 +310,9 @@ func makeQueueContainer(rev *v1.Revision, cfg *config.Config) (*corev1.Container
 					FieldPath: "status.podIP",
 				},
 			},
+		}, {
+			Name:  "TRUST_CONFIG",
+			Value: string(cfg.Network.DataplaneTrust),
 		}, {
 			Name:  "SERVING_LOGGING_CONFIG",
 			Value: cfg.Logging.LoggingConfig,

pkg/reconciler/revision/resources/queue_test.go

@@ -410,7 +410,10 @@ func TestMakeQueueContainer(t *testing.T) {
 				}
 			}
 			cfg := &config.Config{
-				Tracing:       &traceConfig,
+				Tracing: &traceConfig,
+				Network: &netcfg.Config{
+					DataplaneTrust: "disabled",
+				},
 				Logging:       &test.lc,
 				Observability: &test.oc,
 				Deployment:    &test.dc,
@@ -929,6 +932,7 @@ var defaultEnv = map[string]string{
 	"TRACING_CONFIG_DEBUG":                             "false",
 	"TRACING_CONFIG_SAMPLE_RATE":                       "0",
 	"TRACING_CONFIG_ZIPKIN_ENDPOINT":                   "",
+	"TRUST_CONFIG":                                     "disabled",
 	"USER_PORT":                                        strconv.Itoa(v1.DefaultUserPort),
 	"ROOT_CA":                                          "",
 }
@@ -998,9 +1002,11 @@ func revConfig() *config.Config {
 			Defaults:   defaults,
 			Features:   &apicfg.Features{},
 		},
-		Deployment:    &deploymentConfig,
-		Logging:       &logConfig,
-		Network:       &netcfg.Config{},
+		Deployment: &deploymentConfig,
+		Logging:    &logConfig,
+		Network: &netcfg.Config{
+			DataplaneTrust: "disabled",
+		},
 		Observability: &obsConfig,
 		Tracing:       &traceConfig,
 	}

pkg/reconciler/revision/revision.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	appsv1listers "k8s.io/client-go/listers/apps/v1"
 	cachingclientset "knative.dev/caching/pkg/client/clientset/versioned"
+	netcfg "knative.dev/networking/pkg/config"
 	clientset "knative.dev/serving/pkg/client/clientset/versioned"
 	revisionreconciler "knative.dev/serving/pkg/client/injection/reconciler/serving/v1/revision"
 
@@ -139,7 +140,7 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, rev *v1.Revision) pkgrec
 	}
 
 	// Deploy certificate when internal-encryption is enabled.
-	if config.FromContext(ctx).Network.InternalEncryption {
+	if config.FromContext(ctx).Network.DataplaneTrust != netcfg.TrustDisabled {
 		if err := c.reconcileSecret(ctx, rev); err != nil {
 			return err
 		}

pkg/reconciler/revision/table_test.go

@@ -892,6 +892,8 @@ func reconcilerTestConfig() *config.Config {
 		},
 		Logging: &logging.Config{},
 		Tracing: &tracingconfig.Config{},
-		Network: &netcfg.Config{},
+		Network: &netcfg.Config{
+			DataplaneTrust: netcfg.TrustDisabled,
+		},
 	}
 }