kubelet: Support ClusterTrustBundlePEM projections

2022-10-21 23:13:42 -07:00
parent e83baddbb1
commit 1ebe5774d0
17 changed files with 1322 additions and 34 deletions
--- a/pkg/api/pod/util.go
+++ b/pkg/api/pod/util.go
@@ -1002,18 +1002,14 @@ func dropDisabledClusterTrustBundleProjection(podSpec, oldPodSpec *api.PodSpec)
 		return
 	}

-	for _, v := range podSpec.Volumes {
-		if v.Projected == nil {
+	for i := range podSpec.Volumes {
+		if podSpec.Volumes[i].Projected == nil {
 			continue
 		}

-		filteredSources := []api.VolumeProjection{}
-		for _, s := range v.Projected.Sources {
-			if s.ClusterTrustBundle == nil {
-				filteredSources = append(filteredSources, s)
-			}
+		for j := range podSpec.Volumes[i].Projected.Sources {
+			podSpec.Volumes[i].Projected.Sources[j].ClusterTrustBundle = nil
 		}
-		v.Projected.Sources = filteredSources
 	}
 }

--- a/pkg/api/pod/util_test.go
+++ b/pkg/api/pod/util_test.go
@@ -3237,3 +3237,156 @@ func TestMarkPodProposedForResize(t *testing.T) {
 		})
 	}
 }
+
+func TestDropClusterTrustBundleProjectedVolumes(t *testing.T) {
+	testCases := []struct {
+		description                         string
+		clusterTrustBundleProjectionEnabled bool
+		oldPod                              *api.PodSpec
+		newPod                              *api.PodSpec
+		wantPod                             *api.PodSpec
+	}{
+		{
+			description: "feature gate disabled, cannot add CTB volume to pod",
+			oldPod: &api.PodSpec{
+				Volumes: []api.Volume{},
+			},
+			newPod: &api.PodSpec{
+				Volumes: []api.Volume{
+					{
+						Name: "foo",
+						VolumeSource: api.VolumeSource{
+							Projected: &api.ProjectedVolumeSource{
+								Sources: []api.VolumeProjection{
+									{
+										ClusterTrustBundle: &api.ClusterTrustBundleProjection{
+											Name: pointer.String("foo"),
+										},
+									},
+								},
+							}},
+					},
+				},
+			},
+			wantPod: &api.PodSpec{
+				Volumes: []api.Volume{
+					{
+						Name: "foo",
+						VolumeSource: api.VolumeSource{
+							Projected: &api.ProjectedVolumeSource{
+								Sources: []api.VolumeProjection{
+									{},
+								},
+							}},
+					},
+				},
+			},
+		},
+		{
+			description: "feature gate disabled, can keep CTB volume on pod",
+			oldPod: &api.PodSpec{
+				Volumes: []api.Volume{
+					{
+						Name: "foo",
+						VolumeSource: api.VolumeSource{
+							Projected: &api.ProjectedVolumeSource{
+								Sources: []api.VolumeProjection{
+									{
+										ClusterTrustBundle: &api.ClusterTrustBundleProjection{
+											Name: pointer.String("foo"),
+										},
+									},
+								},
+							}},
+					},
+				},
+			},
+			newPod: &api.PodSpec{
+				Volumes: []api.Volume{
+					{
+						Name: "foo",
+						VolumeSource: api.VolumeSource{
+							Projected: &api.ProjectedVolumeSource{
+								Sources: []api.VolumeProjection{
+									{
+										ClusterTrustBundle: &api.ClusterTrustBundleProjection{
+											Name: pointer.String("foo"),
+										},
+									},
+								},
+							}},
+					},
+				},
+			},
+			wantPod: &api.PodSpec{
+				Volumes: []api.Volume{
+					{
+						Name: "foo",
+						VolumeSource: api.VolumeSource{
+							Projected: &api.ProjectedVolumeSource{
+								Sources: []api.VolumeProjection{
+									{
+										ClusterTrustBundle: &api.ClusterTrustBundleProjection{
+											Name: pointer.String("foo"),
+										},
+									},
+								},
+							}},
+					},
+				},
+			},
+		},
+		{
+			description:                         "feature gate enabled, can add CTB volume to pod",
+			clusterTrustBundleProjectionEnabled: true,
+			oldPod: &api.PodSpec{
+				Volumes: []api.Volume{},
+			},
+			newPod: &api.PodSpec{
+				Volumes: []api.Volume{
+					{
+						Name: "foo",
+						VolumeSource: api.VolumeSource{
+							Projected: &api.ProjectedVolumeSource{
+								Sources: []api.VolumeProjection{
+									{
+										ClusterTrustBundle: &api.ClusterTrustBundleProjection{
+											Name: pointer.String("foo"),
+										},
+									},
+								},
+							}},
+					},
+				},
+			},
+			wantPod: &api.PodSpec{
+				Volumes: []api.Volume{
+					{
+						Name: "foo",
+						VolumeSource: api.VolumeSource{
+							Projected: &api.ProjectedVolumeSource{
+								Sources: []api.VolumeProjection{
+									{
+										ClusterTrustBundle: &api.ClusterTrustBundleProjection{
+											Name: pointer.String("foo"),
+										},
+									},
+								},
+							}},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.description, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ClusterTrustBundleProjection, tc.clusterTrustBundleProjectionEnabled)()
+
+			dropDisabledClusterTrustBundleProjection(tc.newPod, tc.oldPod)
+			if diff := cmp.Diff(tc.newPod, tc.wantPod); diff != "" {
+				t.Fatalf("Unexpected modification to new pod; diff (-got +want)\n%s", diff)
+			}
+		})
+	}
+}
--- a/pkg/kubelet/clustertrustbundle/clustertrustbundle_manager.go
+++ b/pkg/kubelet/clustertrustbundle/clustertrustbundle_manager.go
@@ -0,0 +1,261 @@
+/*
+Copyright 2023 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 clustertrustbundle abstracts access to ClusterTrustBundles so that
+// projected volumes can use them.
+package clustertrustbundle
+
+import (
+	"encoding/pem"
+	"fmt"
+	"math/rand"
+	"time"
+
+	certificatesv1alpha1 "k8s.io/api/certificates/v1alpha1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	lrucache "k8s.io/apimachinery/pkg/util/cache"
+	"k8s.io/apimachinery/pkg/util/sets"
+	certinformersv1alpha1 "k8s.io/client-go/informers/certificates/v1alpha1"
+	certlistersv1alpha1 "k8s.io/client-go/listers/certificates/v1alpha1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+)
+
+const (
+	maxLabelSelectorLength = 100 * 1024
+)
+
+// Manager abstracts over the ability to get trust anchors.
+type Manager interface {
+	GetTrustAnchorsByName(name string, allowMissing bool) ([]byte, error)
+	GetTrustAnchorsBySigner(signerName string, labelSelector *metav1.LabelSelector, allowMissing bool) ([]byte, error)
+}
+
+// InformerManager is the "real" manager.  It uses informers to track
+// ClusterTrustBundle objects.
+type InformerManager struct {
+	ctbInformer cache.SharedIndexInformer
+	ctbLister   certlistersv1alpha1.ClusterTrustBundleLister
+
+	normalizationCache *lrucache.LRUExpireCache
+	cacheTTL           time.Duration
+}
+
+var _ Manager = (*InformerManager)(nil)
+
+// NewInformerManager returns an initialized InformerManager.
+func NewInformerManager(bundles certinformersv1alpha1.ClusterTrustBundleInformer, cacheSize int, cacheTTL time.Duration) (*InformerManager, error) {
+	// We need to call Informer() before calling start on the shared informer
+	// factory, or the informer won't be registered to be started.
+	m := &InformerManager{
+		ctbInformer:        bundles.Informer(),
+		ctbLister:          bundles.Lister(),
+		normalizationCache: lrucache.NewLRUExpireCache(cacheSize),
+		cacheTTL:           cacheTTL,
+	}
+
+	// Have the informer bust cache entries when it sees updates that could
+	// apply to them.
+	_, err := m.ctbInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj any) {
+			ctb, ok := obj.(*certificatesv1alpha1.ClusterTrustBundle)
+			if !ok {
+				return
+			}
+			klog.InfoS("Dropping all cache entries for signer", "signerName", ctb.Spec.SignerName)
+			m.dropCacheFor(ctb)
+		},
+		UpdateFunc: func(old, new any) {
+			ctb, ok := new.(*certificatesv1alpha1.ClusterTrustBundle)
+			if !ok {
+				return
+			}
+			klog.InfoS("Dropping cache for ClusterTrustBundle", "signerName", ctb.Spec.SignerName)
+			m.dropCacheFor(new.(*certificatesv1alpha1.ClusterTrustBundle))
+		},
+		DeleteFunc: func(obj any) {
+			ctb, ok := obj.(*certificatesv1alpha1.ClusterTrustBundle)
+			if !ok {
+				tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					return
+				}
+				ctb, ok = tombstone.Obj.(*certificatesv1alpha1.ClusterTrustBundle)
+				if !ok {
+					return
+				}
+			}
+			klog.InfoS("Dropping cache for ClusterTrustBundle", "signerName", ctb.Spec.SignerName)
+			m.dropCacheFor(ctb)
+		},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("while registering event handler on informer: %w", err)
+	}
+
+	return m, nil
+}
+
+func (m *InformerManager) dropCacheFor(ctb *certificatesv1alpha1.ClusterTrustBundle) {
+	if ctb.Spec.SignerName != "" {
+		m.normalizationCache.RemoveAll(func(key any) bool {
+			return key.(cacheKeyType).signerName == ctb.Spec.SignerName
+		})
+	} else {
+		m.normalizationCache.RemoveAll(func(key any) bool {
+			return key.(cacheKeyType).ctbName == ctb.ObjectMeta.Name
+		})
+	}
+}
+
+// GetTrustAnchorsByName returns normalized and deduplicated trust anchors from
+// a single named ClusterTrustBundle.
+func (m *InformerManager) GetTrustAnchorsByName(name string, allowMissing bool) ([]byte, error) {
+	if !m.ctbInformer.HasSynced() {
+		return nil, fmt.Errorf("ClusterTrustBundle informer has not yet synced")
+	}
+
+	cacheKey := cacheKeyType{ctbName: name}
+
+	if cachedAnchors, ok := m.normalizationCache.Get(cacheKey); ok {
+		return cachedAnchors.([]byte), nil
+	}
+
+	ctb, err := m.ctbLister.Get(name)
+	if k8serrors.IsNotFound(err) && allowMissing {
+		return []byte{}, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("while getting ClusterTrustBundle: %w", err)
+	}
+
+	pemTrustAnchors, err := m.normalizeTrustAnchors([]*certificatesv1alpha1.ClusterTrustBundle{ctb})
+	if err != nil {
+		return nil, fmt.Errorf("while normalizing trust anchors: %w", err)
+	}
+
+	m.normalizationCache.Add(cacheKey, pemTrustAnchors, m.cacheTTL)
+
+	return pemTrustAnchors, nil
+}
+
+// GetTrustAnchorsBySigner returns normalized and deduplicated trust anchors
+// from a set of selected ClusterTrustBundles.
+func (m *InformerManager) GetTrustAnchorsBySigner(signerName string, labelSelector *metav1.LabelSelector, allowMissing bool) ([]byte, error) {
+	if !m.ctbInformer.HasSynced() {
+		return nil, fmt.Errorf("ClusterTrustBundle informer has not yet synced")
+	}
+
+	// Note that this function treats nil as "match nothing", and non-nil but
+	// empty as "match everything".
+	selector, err := metav1.LabelSelectorAsSelector(labelSelector)
+	if err != nil {
+		return nil, fmt.Errorf("while parsing label selector: %w", err)
+	}
+
+	cacheKey := cacheKeyType{signerName: signerName, labelSelector: selector.String()}
+
+	if lsLen := len(cacheKey.labelSelector); lsLen > maxLabelSelectorLength {
+		return nil, fmt.Errorf("label selector length (%d) is larger than %d", lsLen, maxLabelSelectorLength)
+	}
+
+	if cachedAnchors, ok := m.normalizationCache.Get(cacheKey); ok {
+		return cachedAnchors.([]byte), nil
+	}
+
+	rawCTBList, err := m.ctbLister.List(selector)
+	if err != nil {
+		return nil, fmt.Errorf("while listing ClusterTrustBundles matching label selector %v: %w", labelSelector, err)
+	}
+
+	ctbList := []*certificatesv1alpha1.ClusterTrustBundle{}
+	for _, ctb := range rawCTBList {
+		if ctb.Spec.SignerName == signerName {
+			ctbList = append(ctbList, ctb)
+		}
+	}
+
+	if len(ctbList) == 0 {
+		if allowMissing {
+			return []byte{}, nil
+		}
+		return nil, fmt.Errorf("combination of signerName and labelSelector matched zero ClusterTrustBundles")
+	}
+
+	pemTrustAnchors, err := m.normalizeTrustAnchors(ctbList)
+	if err != nil {
+		return nil, fmt.Errorf("while normalizing trust anchors: %w", err)
+	}
+
+	m.normalizationCache.Add(cacheKey, pemTrustAnchors, m.cacheTTL)
+
+	return pemTrustAnchors, nil
+}
+
+func (m *InformerManager) normalizeTrustAnchors(ctbList []*certificatesv1alpha1.ClusterTrustBundle) ([]byte, error) {
+	// Deduplicate trust anchors from all ClusterTrustBundles.
+	trustAnchorSet := sets.Set[string]{}
+	for _, ctb := range ctbList {
+		rest := []byte(ctb.Spec.TrustBundle)
+		var b *pem.Block
+		for {
+			b, rest = pem.Decode(rest)
+			if b == nil {
+				break
+			}
+			trustAnchorSet = trustAnchorSet.Insert(string(b.Bytes))
+		}
+	}
+
+	// Give the list a stable ordering that changes each time Kubelet restarts.
+	trustAnchorList := sets.List(trustAnchorSet)
+	rand.Shuffle(len(trustAnchorList), func(i, j int) {
+		trustAnchorList[i], trustAnchorList[j] = trustAnchorList[j], trustAnchorList[i]
+	})
+
+	pemTrustAnchors := []byte{}
+	for _, ta := range trustAnchorList {
+		b := &pem.Block{
+			Type:  "CERTIFICATE",
+			Bytes: []byte(ta),
+		}
+		pemTrustAnchors = append(pemTrustAnchors, pem.EncodeToMemory(b)...)
+	}
+
+	return pemTrustAnchors, nil
+}
+
+type cacheKeyType struct {
+	ctbName       string
+	signerName    string
+	labelSelector string
+}
+
+// NoopManager always returns an error, for use in static kubelet mode.
+type NoopManager struct{}
+
+var _ Manager = (*NoopManager)(nil)
+
+// GetTrustAnchorsByName implements Manager.
+func (m *NoopManager) GetTrustAnchorsByName(name string, allowMissing bool) ([]byte, error) {
+	return nil, fmt.Errorf("ClusterTrustBundle projection is not supported in static kubelet mode")
+}
+
+// GetTrustAnchorsBySigner implements Manager.
+func (m *NoopManager) GetTrustAnchorsBySigner(signerName string, labelSelector *metav1.LabelSelector, allowMissing bool) ([]byte, error) {
+	return nil, fmt.Errorf("ClusterTrustBundle projection is not supported in static kubelet mode")
+}
--- a/pkg/kubelet/clustertrustbundle/clustertrustbundle_manager_test.go
+++ b/pkg/kubelet/clustertrustbundle/clustertrustbundle_manager_test.go
@@ -0,0 +1,474 @@
+/*
+Copyright 2023 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 clustertrustbundle
+
+import (
+	"bytes"
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"fmt"
+	"math/big"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	certificatesv1alpha1 "k8s.io/api/certificates/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+)
+
+func TestBeforeSynced(t *testing.T) {
+	kc := fake.NewSimpleClientset()
+
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(kc, 0)
+
+	ctbInformer := informerFactory.Certificates().V1alpha1().ClusterTrustBundles()
+	ctbManager, _ := NewInformerManager(ctbInformer, 256, 5*time.Minute)
+
+	_, err := ctbManager.GetTrustAnchorsByName("foo", false)
+	if err == nil {
+		t.Fatalf("Got nil error, wanted non-nil")
+	}
+}
+
+func TestGetTrustAnchorsByName(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	ctb1 := &certificatesv1alpha1.ClusterTrustBundle{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "ctb1",
+		},
+		Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+			TrustBundle: mustMakeRoot(t, "root1"),
+		},
+	}
+
+	ctb2 := &certificatesv1alpha1.ClusterTrustBundle{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "ctb2",
+		},
+		Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+			TrustBundle: mustMakeRoot(t, "root2"),
+		},
+	}
+
+	kc := fake.NewSimpleClientset(ctb1, ctb2)
+
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(kc, 0)
+
+	ctbInformer := informerFactory.Certificates().V1alpha1().ClusterTrustBundles()
+	ctbManager, _ := NewInformerManager(ctbInformer, 256, 5*time.Minute)
+
+	informerFactory.Start(ctx.Done())
+	if !cache.WaitForCacheSync(ctx.Done(), ctbInformer.Informer().HasSynced) {
+		t.Fatalf("Timed out waiting for informer to sync")
+	}
+
+	gotBundle, err := ctbManager.GetTrustAnchorsByName("ctb1", false)
+	if err != nil {
+		t.Fatalf("Error while calling GetTrustAnchorsByName: %v", err)
+	}
+
+	if diff := diffBundles(gotBundle, []byte(ctb1.Spec.TrustBundle)); diff != "" {
+		t.Fatalf("Got bad bundle; diff (-got +want)\n%s", diff)
+	}
+
+	gotBundle, err = ctbManager.GetTrustAnchorsByName("ctb2", false)
+	if err != nil {
+		t.Fatalf("Error while calling GetTrustAnchorsByName: %v", err)
+	}
+
+	if diff := diffBundles(gotBundle, []byte(ctb2.Spec.TrustBundle)); diff != "" {
+		t.Fatalf("Got bad bundle; diff (-got +want)\n%s", diff)
+	}
+
+	_, err = ctbManager.GetTrustAnchorsByName("not-found", false)
+	if err == nil { // EQUALS nil
+		t.Fatalf("While looking up nonexisting ClusterTrustBundle, got nil error, wanted non-nil")
+	}
+
+	_, err = ctbManager.GetTrustAnchorsByName("not-found", true)
+	if err != nil {
+		t.Fatalf("Unexpected error while calling GetTrustAnchorsByName for nonexistent CTB with allowMissing: %v", err)
+	}
+}
+
+func TestGetTrustAnchorsByNameCaching(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	ctb1 := &certificatesv1alpha1.ClusterTrustBundle{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foo",
+		},
+		Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+			TrustBundle: mustMakeRoot(t, "root1"),
+		},
+	}
+
+	ctb2 := &certificatesv1alpha1.ClusterTrustBundle{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foo",
+		},
+		Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+			TrustBundle: mustMakeRoot(t, "root2"),
+		},
+	}
+
+	kc := fake.NewSimpleClientset(ctb1)
+
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(kc, 0)
+
+	ctbInformer := informerFactory.Certificates().V1alpha1().ClusterTrustBundles()
+	ctbManager, _ := NewInformerManager(ctbInformer, 256, 5*time.Minute)
+
+	informerFactory.Start(ctx.Done())
+	if !cache.WaitForCacheSync(ctx.Done(), ctbInformer.Informer().HasSynced) {
+		t.Fatalf("Timed out waiting for informer to sync")
+	}
+
+	t.Run("foo should yield the first certificate", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsByName("foo", false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ctb1.Spec.TrustBundle
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("foo should still yield the first certificate", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsByName("foo", false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ctb1.Spec.TrustBundle
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	if err := kc.CertificatesV1alpha1().ClusterTrustBundles().Delete(ctx, ctb1.ObjectMeta.Name, metav1.DeleteOptions{}); err != nil {
+		t.Fatalf("Error while deleting the old CTB: %v", err)
+	}
+	if _, err := kc.CertificatesV1alpha1().ClusterTrustBundles().Create(ctx, ctb2, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("Error while adding new CTB: %v", err)
+	}
+
+	// We need to sleep long enough for the informer to notice the new
+	// ClusterTrustBundle, but much less than the 5 minutes of the cache TTL.
+	// This shows us that the informer is properly clearing the cache.
+	time.Sleep(5 * time.Second)
+
+	t.Run("foo should yield the new certificate", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsByName("foo", false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ctb2.Spec.TrustBundle
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+}
+
+func TestGetTrustAnchorsBySignerName(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	ctb1 := mustMakeCTB("signer-a-label-a-1", "foo.bar/a", map[string]string{"label": "a"}, mustMakeRoot(t, "0"))
+	ctb2 := mustMakeCTB("signer-a-label-a-2", "foo.bar/a", map[string]string{"label": "a"}, mustMakeRoot(t, "1"))
+	ctb2dup := mustMakeCTB("signer-a-label-2-dup", "foo.bar/a", map[string]string{"label": "a"}, ctb2.Spec.TrustBundle)
+	ctb3 := mustMakeCTB("signer-a-label-b-1", "foo.bar/a", map[string]string{"label": "b"}, mustMakeRoot(t, "2"))
+	ctb4 := mustMakeCTB("signer-b-label-a-1", "foo.bar/b", map[string]string{"label": "a"}, mustMakeRoot(t, "3"))
+
+	kc := fake.NewSimpleClientset(ctb1, ctb2, ctb2dup, ctb3, ctb4)
+
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(kc, 0)
+
+	ctbInformer := informerFactory.Certificates().V1alpha1().ClusterTrustBundles()
+	ctbManager, _ := NewInformerManager(ctbInformer, 256, 5*time.Minute)
+
+	informerFactory.Start(ctx.Done())
+	if !cache.WaitForCacheSync(ctx.Done(), ctbInformer.Informer().HasSynced) {
+		t.Fatalf("Timed out waiting for informer to sync")
+	}
+
+	t.Run("big labelselector should cause error", func(t *testing.T) {
+		longString := strings.Builder{}
+		for i := 0; i < 63; i++ {
+			longString.WriteString("v")
+		}
+		matchLabels := map[string]string{}
+		for i := 0; i < 100*1024/63+1; i++ {
+			matchLabels[fmt.Sprintf("key-%d", i)] = longString.String()
+		}
+
+		_, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/a", &metav1.LabelSelector{MatchLabels: matchLabels}, false)
+		if err == nil || !strings.Contains(err.Error(), "label selector length") {
+			t.Fatalf("Bad error, got %v, wanted it to contain \"label selector length\"", err)
+		}
+	})
+
+	t.Run("signer-a label-a should yield two sorted certificates", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/a", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "a"}}, false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ctb1.Spec.TrustBundle + ctb2.Spec.TrustBundle
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("signer-a with nil selector should yield zero certificates", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/a", nil, true)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ""
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("signer-b with empty selector should yield one certificates", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/b", &metav1.LabelSelector{}, false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		if diff := diffBundles(gotBundle, []byte(ctb4.Spec.TrustBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("signer-a label-b should yield one certificate", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/a", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "b"}}, false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		if diff := diffBundles(gotBundle, []byte(ctb3.Spec.TrustBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("signer-b label-a should yield one certificate", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/b", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "a"}}, false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		if diff := diffBundles(gotBundle, []byte(ctb4.Spec.TrustBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("signer-b label-b allowMissing=true should yield zero certificates", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/b", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "b"}}, true)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		if diff := diffBundles(gotBundle, []byte{}); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("signer-b label-b allowMissing=false should yield zero certificates (error)", func(t *testing.T) {
+		_, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/b", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "b"}}, false)
+		if err == nil { // EQUALS nil
+			t.Fatalf("Got nil error while calling GetTrustAnchorsBySigner, wanted non-nil")
+		}
+	})
+}
+
+func TestGetTrustAnchorsBySignerNameCaching(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	ctb1 := mustMakeCTB("signer-a-label-a-1", "foo.bar/a", map[string]string{"label": "a"}, mustMakeRoot(t, "0"))
+	ctb2 := mustMakeCTB("signer-a-label-a-2", "foo.bar/a", map[string]string{"label": "a"}, mustMakeRoot(t, "1"))
+
+	kc := fake.NewSimpleClientset(ctb1)
+
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(kc, 0)
+
+	ctbInformer := informerFactory.Certificates().V1alpha1().ClusterTrustBundles()
+	ctbManager, _ := NewInformerManager(ctbInformer, 256, 5*time.Minute)
+
+	informerFactory.Start(ctx.Done())
+	if !cache.WaitForCacheSync(ctx.Done(), ctbInformer.Informer().HasSynced) {
+		t.Fatalf("Timed out waiting for informer to sync")
+	}
+
+	t.Run("signer-a label-a should yield one certificate", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/a", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "a"}}, false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ctb1.Spec.TrustBundle
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	t.Run("signer-a label-a should yield the same result when called again", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/a", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "a"}}, false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ctb1.Spec.TrustBundle
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+
+	if err := kc.CertificatesV1alpha1().ClusterTrustBundles().Delete(ctx, ctb1.ObjectMeta.Name, metav1.DeleteOptions{}); err != nil {
+		t.Fatalf("Error while deleting the old CTB: %v", err)
+	}
+	if _, err := kc.CertificatesV1alpha1().ClusterTrustBundles().Create(ctx, ctb2, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("Error while adding new CTB: %v", err)
+	}
+
+	// We need to sleep long enough for the informer to notice the new
+	// ClusterTrustBundle, but much less than the 5 minutes of the cache TTL.
+	// This shows us that the informer is properly clearing the cache.
+	time.Sleep(5 * time.Second)
+
+	t.Run("signer-a label-a should return the new certificate", func(t *testing.T) {
+		gotBundle, err := ctbManager.GetTrustAnchorsBySigner("foo.bar/a", &metav1.LabelSelector{MatchLabels: map[string]string{"label": "a"}}, false)
+		if err != nil {
+			t.Fatalf("Got error while calling GetTrustAnchorsBySigner: %v", err)
+		}
+
+		wantBundle := ctb2.Spec.TrustBundle
+
+		if diff := diffBundles(gotBundle, []byte(wantBundle)); diff != "" {
+			t.Fatalf("Bad bundle; diff (-got +want)\n%s", diff)
+		}
+	})
+}
+
+func mustMakeRoot(t *testing.T, cn string) string {
+	pub, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("Error while generating key: %v", err)
+	}
+
+	template := &x509.Certificate{
+		SerialNumber: big.NewInt(0),
+		Subject: pkix.Name{
+			CommonName: cn,
+		},
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+	}
+
+	cert, err := x509.CreateCertificate(rand.Reader, template, template, pub, priv)
+	if err != nil {
+		t.Fatalf("Error while making certificate: %v", err)
+	}
+
+	return string(pem.EncodeToMemory(&pem.Block{
+		Type:    "CERTIFICATE",
+		Headers: nil,
+		Bytes:   cert,
+	}))
+}
+
+func mustMakeCTB(name, signerName string, labels map[string]string, bundle string) *certificatesv1alpha1.ClusterTrustBundle {
+	return &certificatesv1alpha1.ClusterTrustBundle{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: labels,
+		},
+		Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+			SignerName:  signerName,
+			TrustBundle: bundle,
+		},
+	}
+}
+
+func diffBundles(a, b []byte) string {
+	var block *pem.Block
+
+	aBlocks := []*pem.Block{}
+	for {
+		block, a = pem.Decode(a)
+		if block == nil {
+			break
+		}
+		aBlocks = append(aBlocks, block)
+	}
+	sort.Slice(aBlocks, func(i, j int) bool {
+		if aBlocks[i].Type < aBlocks[j].Type {
+			return true
+		} else if aBlocks[i].Type == aBlocks[j].Type {
+			comp := bytes.Compare(aBlocks[i].Bytes, aBlocks[j].Bytes)
+			return comp <= 0
+		} else {
+			return false
+		}
+	})
+
+	bBlocks := []*pem.Block{}
+	for {
+		block, b = pem.Decode(b)
+		if block == nil {
+			break
+		}
+		bBlocks = append(bBlocks, block)
+	}
+	sort.Slice(bBlocks, func(i, j int) bool {
+		if bBlocks[i].Type < bBlocks[j].Type {
+			return true
+		} else if bBlocks[i].Type == bBlocks[j].Type {
+			comp := bytes.Compare(bBlocks[i].Bytes, bBlocks[j].Bytes)
+			return comp <= 0
+		} else {
+			return false
+		}
+	})
+
+	return cmp.Diff(aBlocks, bBlocks)
+}
--- a/pkg/kubelet/config/common.go
+++ b/pkg/kubelet/config/common.go
@@ -19,6 +19,7 @@ package config
 import (
 	"crypto/md5"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"strings"

@@ -102,6 +103,9 @@ func applyDefaults(pod *api.Pod, source string, isFile bool, nodeName types.Node

 type defaultFunc func(pod *api.Pod) error

+// A static pod tried to use a ClusterTrustBundle projected volume source.
+var ErrStaticPodTriedToUseClusterTrustBundle = errors.New("static pods may not use ClusterTrustBundle projected volume sources")
+
 // tryDecodeSinglePod takes data and tries to extract valid Pod config information from it.
 func tryDecodeSinglePod(data []byte, defaultFn defaultFunc) (parsed bool, pod *v1.Pod, err error) {
 	// JSON is valid YAML, so this should work for everything.
@@ -136,6 +140,19 @@ func tryDecodeSinglePod(data []byte, defaultFn defaultFunc) (parsed bool, pod *v
 		klog.ErrorS(err, "Pod failed to convert to v1", "pod", klog.KObj(newPod))
 		return true, nil, err
 	}
+
+	for _, v := range v1Pod.Spec.Volumes {
+		if v.Projected == nil {
+			continue
+		}
+
+		for _, s := range v.Projected.Sources {
+			if s.ClusterTrustBundle != nil {
+				return true, nil, ErrStaticPodTriedToUseClusterTrustBundle
+			}
+		}
+	}
+
 	return true, v1Pod, nil
 }

--- a/pkg/kubelet/config/common_test.go
+++ b/pkg/kubelet/config/common_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package config

 import (
+	"errors"
 	"reflect"
 	"testing"

@@ -31,6 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/validation"
 	"k8s.io/kubernetes/pkg/securitycontext"
+	"k8s.io/utils/ptr"
 )

 func noDefault(*core.Pod) error { return nil }
@@ -107,6 +109,76 @@ func TestDecodeSinglePod(t *testing.T) {
 	}
 }

+func TestDecodeSinglePodRejectsClusterTrustBundleVolumes(t *testing.T) {
+	grace := int64(30)
+	enableServiceLinks := v1.DefaultEnableServiceLinks
+	pod := &v1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test",
+			UID:       "12345",
+			Namespace: "mynamespace",
+		},
+		Spec: v1.PodSpec{
+			RestartPolicy:                 v1.RestartPolicyAlways,
+			DNSPolicy:                     v1.DNSClusterFirst,
+			TerminationGracePeriodSeconds: &grace,
+			Containers: []v1.Container{{
+				Name:                     "image",
+				Image:                    "test/image",
+				ImagePullPolicy:          "IfNotPresent",
+				TerminationMessagePath:   "/dev/termination-log",
+				TerminationMessagePolicy: v1.TerminationMessageReadFile,
+				SecurityContext:          securitycontext.ValidSecurityContextWithContainerDefaults(),
+				VolumeMounts: []v1.VolumeMount{
+					{
+						Name:      "ctb-volume",
+						MountPath: "/var/run/ctb-volume",
+					},
+				},
+			}},
+			Volumes: []v1.Volume{
+				{
+					Name: "ctb-volume",
+					VolumeSource: v1.VolumeSource{
+						Projected: &v1.ProjectedVolumeSource{
+							Sources: []v1.VolumeProjection{
+								{
+									ClusterTrustBundle: &v1.ClusterTrustBundleProjection{
+										Name: ptr.To("my-ctb"),
+										Path: "ctb-file",
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			SecurityContext:    &v1.PodSecurityContext{},
+			SchedulerName:      v1.DefaultSchedulerName,
+			EnableServiceLinks: &enableServiceLinks,
+		},
+		Status: v1.PodStatus{
+			PodIP: "1.2.3.4",
+			PodIPs: []v1.PodIP{
+				{
+					IP: "1.2.3.4",
+				},
+			},
+		},
+	}
+	json, err := runtime.Encode(clientscheme.Codecs.LegacyCodec(v1.SchemeGroupVersion), pod)
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	_, _, err = tryDecodeSinglePod(json, noDefault)
+	if !errors.Is(err, ErrStaticPodTriedToUseClusterTrustBundle) {
+		t.Errorf("Got error %q, want %q", err, ErrStaticPodTriedToUseClusterTrustBundle)
+	}
+}
+
 func TestDecodePodList(t *testing.T) {
 	grace := int64(30)
 	enableServiceLinks := v1.DefaultEnableServiceLinks
--- a/pkg/kubelet/kubelet.go
+++ b/pkg/kubelet/kubelet.go
@@ -75,6 +75,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	kubeletcertificate "k8s.io/kubernetes/pkg/kubelet/certificate"
 	"k8s.io/kubernetes/pkg/kubelet/cloudresource"
+	"k8s.io/kubernetes/pkg/kubelet/clustertrustbundle"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
 	draplugin "k8s.io/kubernetes/pkg/kubelet/cm/dra/plugin"
 	"k8s.io/kubernetes/pkg/kubelet/config"
@@ -451,7 +452,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	var serviceLister corelisters.ServiceLister
 	var serviceHasSynced cache.InformerSynced
 	if kubeDeps.KubeClient != nil {
-		kubeInformers := informers.NewSharedInformerFactory(kubeDeps.KubeClient, 0)
+		kubeInformers := informers.NewSharedInformerFactoryWithOptions(kubeDeps.KubeClient, 0)
 		serviceLister = kubeInformers.Core().V1().Services().Lister()
 		serviceHasSynced = kubeInformers.Core().V1().Services().Informer().HasSynced
 		kubeInformers.Start(wait.NeverStop)
@@ -793,11 +794,26 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,

 	tokenManager := token.NewManager(kubeDeps.KubeClient)

+	var clusterTrustBundleManager clustertrustbundle.Manager
+	if kubeDeps.KubeClient != nil && utilfeature.DefaultFeatureGate.Enabled(features.ClusterTrustBundleProjection) {
+		kubeInformers := informers.NewSharedInformerFactoryWithOptions(kubeDeps.KubeClient, 0)
+		clusterTrustBundleManager, err = clustertrustbundle.NewInformerManager(kubeInformers.Certificates().V1alpha1().ClusterTrustBundles(), 2*int(kubeCfg.MaxPods), 5*time.Minute)
+		if err != nil {
+			return nil, fmt.Errorf("while starting informer-based ClusterTrustBundle manager: %w", err)
+		}
+		kubeInformers.Start(wait.NeverStop)
+		klog.InfoS("Started ClusterTrustBundle informer")
+	} else {
+		// In static kubelet mode, use a no-op manager.
+		clusterTrustBundleManager = &clustertrustbundle.NoopManager{}
+		klog.InfoS("Not starting ClusterTrustBundle informer because we are in static kubelet mode")
+	}
+
 	// NewInitializedVolumePluginMgr initializes some storageErrors on the Kubelet runtimeState (in csi_plugin.go init)
 	// which affects node ready status. This function must be called before Kubelet is initialized so that the Node
 	// ReadyState is accurate with the storage state.
 	klet.volumePluginMgr, err =
-		NewInitializedVolumePluginMgr(klet, secretManager, configMapManager, tokenManager, kubeDeps.VolumePlugins, kubeDeps.DynamicPluginProber)
+		NewInitializedVolumePluginMgr(klet, secretManager, configMapManager, tokenManager, clusterTrustBundleManager, kubeDeps.VolumePlugins, kubeDeps.DynamicPluginProber)
 	if err != nil {
 		return nil, err
 	}
--- a/pkg/kubelet/kubelet_test.go
+++ b/pkg/kubelet/kubelet_test.go
@@ -59,6 +59,7 @@ import (
 	"k8s.io/kubernetes/pkg/features"
 	kubeletconfiginternal "k8s.io/kubernetes/pkg/kubelet/apis/config"
 	cadvisortest "k8s.io/kubernetes/pkg/kubelet/cadvisor/testing"
+	"k8s.io/kubernetes/pkg/kubelet/clustertrustbundle"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
 	"k8s.io/kubernetes/pkg/kubelet/config"
 	"k8s.io/kubernetes/pkg/kubelet/configmap"
@@ -379,7 +380,7 @@ func newTestKubeletWithImageList(

 	var prober volume.DynamicPluginProber // TODO (#51147) inject mock
 	kubelet.volumePluginMgr, err =
-		NewInitializedVolumePluginMgr(kubelet, kubelet.secretManager, kubelet.configMapManager, token.NewManager(kubelet.kubeClient), allPlugins, prober)
+		NewInitializedVolumePluginMgr(kubelet, kubelet.secretManager, kubelet.configMapManager, token.NewManager(kubelet.kubeClient), &clustertrustbundle.NoopManager{}, allPlugins, prober)
 	require.NoError(t, err, "Failed to initialize VolumePluginMgr")

 	kubelet.volumeManager = kubeletvolume.NewVolumeManager(
--- a/pkg/kubelet/runonce_test.go
+++ b/pkg/kubelet/runonce_test.go
@@ -35,6 +35,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	utiltesting "k8s.io/client-go/util/testing"
 	cadvisortest "k8s.io/kubernetes/pkg/kubelet/cadvisor/testing"
+	"k8s.io/kubernetes/pkg/kubelet/clustertrustbundle"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
 	"k8s.io/kubernetes/pkg/kubelet/configmap"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
@@ -72,6 +73,7 @@ func TestRunOnce(t *testing.T) {
 	}, nil).AnyTimes()
 	fakeSecretManager := secret.NewFakeManager()
 	fakeConfigMapManager := configmap.NewFakeManager()
+	clusterTrustBundleManager := &clustertrustbundle.NoopManager{}
 	podManager := kubepod.NewBasicPodManager()
 	fakeRuntime := &containertest.FakeRuntime{}
 	podStartupLatencyTracker := kubeletutil.NewPodStartupLatencyTracker()
@@ -103,7 +105,7 @@ func TestRunOnce(t *testing.T) {

 	plug := &volumetest.FakeVolumePlugin{PluginName: "fake", Host: nil}
 	kb.volumePluginMgr, err =
-		NewInitializedVolumePluginMgr(kb, fakeSecretManager, fakeConfigMapManager, nil, []volume.VolumePlugin{plug}, nil /* prober */)
+		NewInitializedVolumePluginMgr(kb, fakeSecretManager, fakeConfigMapManager, nil, clusterTrustBundleManager, []volume.VolumePlugin{plug}, nil /* prober */)
 	if err != nil {
 		t.Fatalf("failed to initialize VolumePluginMgr: %v", err)
 	}
--- a/pkg/kubelet/volume_host.go
+++ b/pkg/kubelet/volume_host.go
@@ -27,6 +27,7 @@ import (

 	authenticationv1 "k8s.io/api/authentication/v1"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
@@ -35,6 +36,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	cloudprovider "k8s.io/cloud-provider"
+	"k8s.io/kubernetes/pkg/kubelet/clustertrustbundle"
 	"k8s.io/kubernetes/pkg/kubelet/configmap"
 	"k8s.io/kubernetes/pkg/kubelet/secret"
 	"k8s.io/kubernetes/pkg/kubelet/token"
@@ -55,6 +57,7 @@ func NewInitializedVolumePluginMgr(
 	secretManager secret.Manager,
 	configMapManager configmap.Manager,
 	tokenManager *token.Manager,
+	clusterTrustBundleManager clustertrustbundle.Manager,
 	plugins []volume.VolumePlugin,
 	prober volume.DynamicPluginProber) (*volume.VolumePluginMgr, error) {

@@ -75,15 +78,16 @@ func NewInitializedVolumePluginMgr(
 	}

 	kvh := &kubeletVolumeHost{
-		kubelet:          kubelet,
-		volumePluginMgr:  volume.VolumePluginMgr{},
-		secretManager:    secretManager,
-		configMapManager: configMapManager,
-		tokenManager:     tokenManager,
-		informerFactory:  informerFactory,
-		csiDriverLister:  csiDriverLister,
-		csiDriversSynced: csiDriversSynced,
-		exec:             utilexec.New(),
+		kubelet:                   kubelet,
+		volumePluginMgr:           volume.VolumePluginMgr{},
+		secretManager:             secretManager,
+		configMapManager:          configMapManager,
+		tokenManager:              tokenManager,
+		clusterTrustBundleManager: clusterTrustBundleManager,
+		informerFactory:           informerFactory,
+		csiDriverLister:           csiDriverLister,
+		csiDriversSynced:          csiDriversSynced,
+		exec:                      utilexec.New(),
 	}

 	if err := kvh.volumePluginMgr.InitPlugins(plugins, prober, kvh); err != nil {
@@ -104,15 +108,16 @@ func (kvh *kubeletVolumeHost) GetPluginDir(pluginName string) string {
 }

 type kubeletVolumeHost struct {
-	kubelet          *Kubelet
-	volumePluginMgr  volume.VolumePluginMgr
-	secretManager    secret.Manager
-	tokenManager     *token.Manager
-	configMapManager configmap.Manager
-	informerFactory  informers.SharedInformerFactory
-	csiDriverLister  storagelisters.CSIDriverLister
-	csiDriversSynced cache.InformerSynced
-	exec             utilexec.Interface
+	kubelet                   *Kubelet
+	volumePluginMgr           volume.VolumePluginMgr
+	secretManager             secret.Manager
+	tokenManager              *token.Manager
+	configMapManager          configmap.Manager
+	clusterTrustBundleManager clustertrustbundle.Manager
+	informerFactory           informers.SharedInformerFactory
+	csiDriverLister           storagelisters.CSIDriverLister
+	csiDriversSynced          cache.InformerSynced
+	exec                      utilexec.Interface
 }

 func (kvh *kubeletVolumeHost) SetKubeletError(err error) {
@@ -266,6 +271,14 @@ func (kvh *kubeletVolumeHost) DeleteServiceAccountTokenFunc() func(podUID types.
 	return kvh.tokenManager.DeleteServiceAccountToken
 }

+func (kvh *kubeletVolumeHost) GetTrustAnchorsByName(name string, allowMissing bool) ([]byte, error) {
+	return kvh.clusterTrustBundleManager.GetTrustAnchorsByName(name, allowMissing)
+}
+
+func (kvh *kubeletVolumeHost) GetTrustAnchorsBySigner(signerName string, labelSelector *metav1.LabelSelector, allowMissing bool) ([]byte, error) {
+	return kvh.clusterTrustBundleManager.GetTrustAnchorsBySigner(signerName, labelSelector, allowMissing)
+}
+
 func (kvh *kubeletVolumeHost) GetNodeLabels() (map[string]string, error) {
 	node, err := kvh.kubelet.GetNode()
 	if err != nil {
--- a/pkg/volume/plugins.go
+++ b/pkg/volume/plugins.go
@@ -333,6 +333,13 @@ type KubeletVolumeHost interface {
 	WaitForCacheSync() error
 	// Returns hostutil.HostUtils
 	GetHostUtil() hostutil.HostUtils
+
+	// Returns trust anchors from the named ClusterTrustBundle.
+	GetTrustAnchorsByName(name string, allowMissing bool) ([]byte, error)
+
+	// Returns trust anchors from the ClusterTrustBundles selected by signer
+	// name and label selector.
+	GetTrustAnchorsBySigner(signerName string, labelSelector *metav1.LabelSelector, allowMissing bool) ([]byte, error)
 }

 // AttachDetachVolumeHost is a AttachDetach Controller specific interface that plugins can use
--- a/pkg/volume/projected/projected.go
+++ b/pkg/volume/projected/projected.go
@@ -45,6 +45,7 @@ const (

 type projectedPlugin struct {
 	host                      volume.VolumeHost
+	kvHost                    volume.KubeletVolumeHost
 	getSecret                 func(namespace, name string) (*v1.Secret, error)
 	getConfigMap              func(namespace, name string) (*v1.ConfigMap, error)
 	getServiceAccountToken    func(namespace, name string, tr *authenticationv1.TokenRequest) (*authenticationv1.TokenRequest, error)
@@ -69,6 +70,7 @@ func getPath(uid types.UID, volName string, host volume.VolumeHost) string {

 func (plugin *projectedPlugin) Init(host volume.VolumeHost) error {
 	plugin.host = host
+	plugin.kvHost = host.(volume.KubeletVolumeHost)
 	plugin.getSecret = host.GetSecretFunc()
 	plugin.getConfigMap = host.GetConfigMapFunc()
 	plugin.getServiceAccountToken = host.GetServiceAccountTokenFunc()
@@ -353,6 +355,42 @@ func (s *projectedVolumeMounter) collectData(mounterArgs volume.MounterArgs) (ma
 				Mode:   mode,
 				FsUser: mounterArgs.FsUser,
 			}
+		case source.ClusterTrustBundle != nil:
+			allowEmpty := false
+			if source.ClusterTrustBundle.Optional != nil && *source.ClusterTrustBundle.Optional {
+				allowEmpty = true
+			}
+
+			var trustAnchors []byte
+			if source.ClusterTrustBundle.Name != nil {
+				var err error
+				trustAnchors, err = s.plugin.kvHost.GetTrustAnchorsByName(*source.ClusterTrustBundle.Name, allowEmpty)
+				if err != nil {
+					errlist = append(errlist, err)
+					continue
+				}
+			} else if source.ClusterTrustBundle.SignerName != nil {
+				var err error
+				trustAnchors, err = s.plugin.kvHost.GetTrustAnchorsBySigner(*source.ClusterTrustBundle.SignerName, source.ClusterTrustBundle.LabelSelector, allowEmpty)
+				if err != nil {
+					errlist = append(errlist, err)
+					continue
+				}
+			} else {
+				errlist = append(errlist, fmt.Errorf("ClusterTrustBundle projection requires either name or signerName to be set"))
+				continue
+			}
+
+			mode := *s.source.DefaultMode
+			if mounterArgs.FsUser != nil || mounterArgs.FsGroup != nil {
+				mode = 0600
+			}
+
+			payload[source.ClusterTrustBundle.Path] = volumeutil.FileProjection{
+				Data:   trustAnchors,
+				Mode:   mode,
+				FsUser: mounterArgs.FsUser,
+			}
 		}
 	}
 	return payload, utilerrors.NewAggregate(errlist)
--- a/pkg/volume/projected/projected_test.go
+++ b/pkg/volume/projected/projected_test.go
@@ -17,7 +17,13 @@ limitations under the License.
 package projected

 import (
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
 	"fmt"
+	"math/big"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -26,6 +32,7 @@ import (

 	"github.com/google/go-cmp/cmp"
 	authenticationv1 "k8s.io/api/authentication/v1"
+	certificatesv1alpha1 "k8s.io/api/certificates/v1alpha1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -872,13 +879,172 @@ func TestCollectDataWithServiceAccountToken(t *testing.T) {
 	}
 }

+func TestCollectDataWithClusterTrustBundle(t *testing.T) {
+	// This test is limited by the use of a fake clientset and volume host.  We
+	// can't meaningfully test that label selectors end up doing the correct
+	// thing for example.
+
+	goodCert1 := mustMakeRoot(t, "root1")
+
+	testCases := []struct {
+		name string
+
+		source  v1.ProjectedVolumeSource
+		bundles []runtime.Object
+
+		fsUser  *int64
+		fsGroup *int64
+
+		wantPayload map[string]util.FileProjection
+		wantErr     error
+	}{
+		{
+			name: "single ClusterTrustBundle by name",
+			source: v1.ProjectedVolumeSource{
+				Sources: []v1.VolumeProjection{
+					{
+						ClusterTrustBundle: &v1.ClusterTrustBundleProjection{
+							Name: utilptr.String("foo"),
+							Path: "bundle.pem",
+						},
+					},
+				},
+				DefaultMode: utilptr.Int32(0644),
+			},
+			bundles: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "foo",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						TrustBundle: string(goodCert1),
+					},
+				},
+			},
+			wantPayload: map[string]util.FileProjection{
+				"bundle.pem": {
+					Data: []byte(goodCert1),
+					Mode: 0644,
+				},
+			},
+		},
+		{
+			name: "single ClusterTrustBundle by signer name",
+			source: v1.ProjectedVolumeSource{
+				Sources: []v1.VolumeProjection{
+					{
+						ClusterTrustBundle: &v1.ClusterTrustBundleProjection{
+							SignerName: utilptr.String("foo.example/bar"), // Note: fake client doesn't understand selection by signer name.
+							LabelSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{
+									"key": "non-value", // Note: fake client doesn't actually act on label selectors.
+								},
+							},
+							Path: "bundle.pem",
+						},
+					},
+				},
+				DefaultMode: utilptr.Int32(0644),
+			},
+			bundles: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "foo:example:bar",
+						Labels: map[string]string{
+							"key": "value",
+						},
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  "foo.example/bar",
+						TrustBundle: string(goodCert1),
+					},
+				},
+			},
+			wantPayload: map[string]util.FileProjection{
+				"bundle.pem": {
+					Data: []byte(goodCert1),
+					Mode: 0644,
+				},
+			},
+		},
+		{
+			name: "single ClusterTrustBundle by name, non-default mode",
+			source: v1.ProjectedVolumeSource{
+				Sources: []v1.VolumeProjection{
+					{
+						ClusterTrustBundle: &v1.ClusterTrustBundleProjection{
+							Name: utilptr.String("foo"),
+							Path: "bundle.pem",
+						},
+					},
+				},
+				DefaultMode: utilptr.Int32(0600),
+			},
+			bundles: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "foo",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						TrustBundle: string(goodCert1),
+					},
+				},
+			},
+			wantPayload: map[string]util.FileProjection{
+				"bundle.pem": {
+					Data: []byte(goodCert1),
+					Mode: 0600,
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			pod := &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "default",
+					UID:       types.UID("test_pod_uid"),
+				},
+				Spec: v1.PodSpec{ServiceAccountName: "foo"},
+			}
+
+			client := fake.NewSimpleClientset(tc.bundles...)
+
+			tempDir, host := newTestHost(t, client)
+			defer os.RemoveAll(tempDir)
+
+			var myVolumeMounter = projectedVolumeMounter{
+				projectedVolume: &projectedVolume{
+					sources: tc.source.Sources,
+					podUID:  pod.UID,
+					plugin: &projectedPlugin{
+						host:   host,
+						kvHost: host.(volume.KubeletVolumeHost),
+					},
+				},
+				source: tc.source,
+				pod:    pod,
+			}
+
+			gotPayload, err := myVolumeMounter.collectData(volume.MounterArgs{FsUser: tc.fsUser, FsGroup: tc.fsGroup})
+			if err != nil {
+				t.Fatalf("Unexpected failure making payload: %v", err)
+			}
+			if diff := cmp.Diff(tc.wantPayload, gotPayload); diff != "" {
+				t.Fatalf("Bad payload; diff (-want +got)\n%s", diff)
+			}
+		})
+	}
+}
+
 func newTestHost(t *testing.T, clientset clientset.Interface) (string, volume.VolumeHost) {
 	tempDir, err := os.MkdirTemp("", "projected_volume_test.")
 	if err != nil {
 		t.Fatalf("can't make a temp rootdir: %v", err)
 	}

-	return tempDir, volumetest.NewFakeVolumeHost(t, tempDir, clientset, emptydir.ProbeVolumePlugins())
+	return tempDir, volumetest.NewFakeKubeletVolumeHost(t, tempDir, clientset, emptydir.ProbeVolumePlugins())
 }

 func TestCanSupport(t *testing.T) {
@@ -1322,3 +1488,30 @@ func doTestCleanAndTeardown(plugin volume.VolumePlugin, podUID types.UID, testVo
 		t.Errorf("TearDown() failed: %v", err)
 	}
 }
+
+func mustMakeRoot(t *testing.T, cn string) string {
+	pub, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("Error while generating key: %v", err)
+	}
+
+	template := &x509.Certificate{
+		SerialNumber: big.NewInt(0),
+		Subject: pkix.Name{
+			CommonName: cn,
+		},
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+	}
+
+	cert, err := x509.CreateCertificate(rand.Reader, template, template, pub, priv)
+	if err != nil {
+		t.Fatalf("Error while making certificate: %v", err)
+	}
+
+	return string(pem.EncodeToMemory(&pem.Block{
+		Type:    "CERTIFICATE",
+		Headers: nil,
+		Bytes:   cert,
+	}))
+}
--- a/pkg/volume/testing/volume_host.go
+++ b/pkg/volume/testing/volume_host.go
@@ -17,6 +17,7 @@ limitations under the License.
 package testing

 import (
+	"bytes"
 	"context"
 	"fmt"
 	"net"
@@ -437,3 +438,30 @@ func (f *fakeKubeletVolumeHost) WaitForCacheSync() error {
 func (f *fakeKubeletVolumeHost) GetHostUtil() hostutil.HostUtils {
 	return f.hostUtil
 }
+
+func (f *fakeKubeletVolumeHost) GetTrustAnchorsByName(name string, allowMissing bool) ([]byte, error) {
+	ctb, err := f.kubeClient.CertificatesV1alpha1().ClusterTrustBundles().Get(context.Background(), name, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("while getting ClusterTrustBundle %s: %w", name, err)
+	}
+
+	return []byte(ctb.Spec.TrustBundle), nil
+}
+
+// Note: we do none of the deduplication and sorting that the real deal should do.
+func (f *fakeKubeletVolumeHost) GetTrustAnchorsBySigner(signerName string, labelSelector *metav1.LabelSelector, allowMissing bool) ([]byte, error) {
+	ctbList, err := f.kubeClient.CertificatesV1alpha1().ClusterTrustBundles().List(context.Background(), metav1.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("while listing all ClusterTrustBundles: %w", err)
+	}
+
+	fullSet := bytes.Buffer{}
+	for i, ctb := range ctbList.Items {
+		fullSet.WriteString(ctb.Spec.TrustBundle)
+		if i != len(ctbList.Items)-1 {
+			fullSet.WriteString("\n")
+		}
+	}
+
+	return fullSet.Bytes(), nil
+}
--- a/plugin/pkg/admission/noderestriction/admission.go
+++ b/plugin/pkg/admission/noderestriction/admission.go
@@ -258,6 +258,17 @@ func (p *Plugin) admitPodCreate(nodeName string, a admission.Attributes) error {
 	if hasConfigMaps {
 		return admission.NewForbidden(a, fmt.Errorf("node %q can not create pods that reference configmaps", nodeName))
 	}
+
+	for _, vol := range pod.Spec.Volumes {
+		if vol.VolumeSource.Projected != nil {
+			for _, src := range vol.VolumeSource.Projected.Sources {
+				if src.ClusterTrustBundle != nil {
+					return admission.NewForbidden(a, fmt.Errorf("node %q can not create pods that reference clustertrustbundles", nodeName))
+				}
+			}
+		}
+	}
+
 	for _, v := range pod.Spec.Volumes {
 		if v.PersistentVolumeClaim != nil {
 			return admission.NewForbidden(a, fmt.Errorf("node %q can not create pods that reference persistentvolumeclaims", nodeName))
--- a/plugin/pkg/admission/noderestriction/admission_test.go
+++ b/plugin/pkg/admission/noderestriction/admission_test.go
@@ -394,6 +394,9 @@ func Test_nodePlugin_Admit(t *testing.T) {
 	configmappod, _ := makeTestPod("ns", "myconfigmappod", "mynode", true)
 	configmappod.Spec.Volumes = []api.Volume{{VolumeSource: api.VolumeSource{ConfigMap: &api.ConfigMapVolumeSource{LocalObjectReference: api.LocalObjectReference{Name: "foo"}}}}}

+	ctbpod, _ := makeTestPod("ns", "myctbpod", "mynode", true)
+	ctbpod.Spec.Volumes = []api.Volume{{VolumeSource: api.VolumeSource{Projected: &api.ProjectedVolumeSource{Sources: []api.VolumeProjection{{ClusterTrustBundle: &api.ClusterTrustBundleProjection{Name: pointer.String("foo")}}}}}}}
+
 	pvcpod, _ := makeTestPod("ns", "mypvcpod", "mynode", true)
 	pvcpod.Spec.Volumes = []api.Volume{{VolumeSource: api.VolumeSource{PersistentVolumeClaim: &api.PersistentVolumeClaimVolumeSource{ClaimName: "foo"}}}}

@@ -866,6 +869,12 @@ func Test_nodePlugin_Admit(t *testing.T) {
 			attributes: admission.NewAttributesRecord(configmappod, nil, podKind, configmappod.Namespace, configmappod.Name, podResource, "", admission.Create, &metav1.CreateOptions{}, false, mynode),
 			err:        "reference configmaps",
 		},
+		{
+			name:       "forbid create of pod referencing clustertrustbundle",
+			podsGetter: noExistingPods,
+			attributes: admission.NewAttributesRecord(ctbpod, nil, podKind, ctbpod.Namespace, ctbpod.Name, podResource, "", admission.Create, &metav1.CreateOptions{}, false, mynode),
+			err:        "reference clustertrustbundles",
+		},
 		{
 			name:       "forbid create of pod referencing persistentvolumeclaim",
 			podsGetter: noExistingPods,
--- a/plugin/pkg/admission/serviceaccount/admission.go
+++ b/plugin/pkg/admission/serviceaccount/admission.go
@@ -210,9 +210,6 @@ func (s *Plugin) Validate(ctx context.Context, a admission.Attributes, o admissi
 					if projSource.ServiceAccountToken != nil {
 						return admission.NewForbidden(a, fmt.Errorf("a mirror pod may not use ServiceAccountToken volume projections"))
 					}
-					if projSource.ClusterTrustBundle != nil {
-						return admission.NewForbidden(a, fmt.Errorf("a mirror pod may not use ClusterTrustBundle volume projections"))
-					}
 				}
 			}
 		}