[stacked on #803] Add clusterTrustBundle as a SystemInfo volume data source

.github/workflows/pr-workflow.yaml

@@ -114,6 +114,14 @@ jobs:
         E2E_TEMPLATE_NAME: counter-microvm
         E2E_TEMPLATE_READY_TIMEOUT: 600s
       run: hack/run-e2e-kind.sh ./internal/e2e/suites/demo -v -args --no-color
+    - name: Run E2E tests (micro-VM identity)
+      # Same identity suite (actorMetadata + clusterTrustBundle SystemInfo
+      # projections, restore-identity, bundle rotation), re-run with the probe
+      # on the micro-VM runtime so both sandboxes' volume delivery is covered.
+      env:
+        E2E_PROBE_SANDBOX_CLASS: microvm
+        E2E_PROBE_READY_TIMEOUT: 600s
+      run: hack/run-e2e-kind.sh ./internal/e2e/suites/identity -v -args --no-color
     - name: Dump diagnostics on failure
       if: failure()
       run: |

cmd/ateapi/internal/controlapi/cluster_trust_bundle.go

@@ -0,0 +1,100 @@
+//  Copyright 2026 Google LLC
+//
+//  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 controlapi
+
+import (
+	"fmt"
+
+	"github.com/agent-substrate/substrate/internal/pemutil"
+	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
+	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	certlisters "k8s.io/client-go/listers/certificates/v1beta1"
+)
+
+// resolveClusterTrustBundles fills the PemBundle bytes of every
+// clusterTrustBundle data source in workloadSpec, resolving each referenced
+// ClusterTrustBundle through the lister and sanitizing its PEM the way
+// kubelet does for projections (CERTIFICATE blocks only, deduplicated).
+//
+// The wire spec carries only {path, pem_bundle}; the bundle NAME lives in the
+// ActorTemplate, so resolution walks the template and locates the matching
+// wire entry by volume name + path. It must run on the resume path, before
+// the spec is sent to atelet: a missing, empty, or unparseable bundle fails
+// the actor start with an error naming the bundle, per the SystemInfo volume
+// contract. atelet never talks to the Kubernetes API.
+func resolveClusterTrustBundles(lister certlisters.ClusterTrustBundleLister, template *atev1alpha1.ActorTemplate, workloadSpec *ateletpb.WorkloadSpec) error {
+	if template == nil {
+		return nil
+	}
+
+	// Wire entries indexed by (volume name, path) for filling in place.
+	type key struct{ volume, path string }
+	wire := map[key]*ateletpb.ClusterTrustBundleDataSource{}
+	for _, vol := range workloadSpec.GetVolumes() {
+		for _, ds := range vol.GetSystemInfo().GetDataSources() {
+			if ctb := ds.GetClusterTrustBundle(); ctb != nil {
+				wire[key{vol.GetName(), ctb.GetPath()}] = ctb
+			}
+		}
+	}
+	if len(wire) == 0 {
+		return nil
+	}
+	// A nil lister means ateapi found the ClusterTrustBundle API unavailable
+	// at startup (certificates.k8s.io/v1beta1 is feature-gated; see
+	// cmd/ateapi/main.go). Fail the start rather than panic on the lister.
+	if lister == nil {
+		return fmt.Errorf("this cluster does not serve the ClusterTrustBundle API (certificates.k8s.io/v1beta1), required by this actor's SystemInfo volumes")
+	}
+
+	// One resolution per distinct bundle name, shared across references.
+	resolved := map[string][]byte{}
+	for _, vol := range template.Spec.Volumes {
+		if vol.VolumeSource.SystemInfo == nil {
+			continue
+		}
+		for _, ds := range vol.VolumeSource.SystemInfo.DataSources {
+			if ds.ClusterTrustBundle == nil {
+				continue
+			}
+			name := ds.ClusterTrustBundle.Name
+
+			pemBundle, ok := resolved[name]
+			if !ok {
+				bundle, err := lister.Get(name)
+				if apierrors.IsNotFound(err) {
+					return fmt.Errorf("ClusterTrustBundle %q not found (referenced by volume %q)", name, vol.Name)
+				} else if err != nil {
+					return fmt.Errorf("while reading ClusterTrustBundle %q: %w", name, err)
+				}
+				pemBundle, err = pemutil.SanitizeCertificateBundle([]byte(bundle.Spec.TrustBundle))
+				if err != nil {
+					return fmt.Errorf("ClusterTrustBundle %q has an unusable trust bundle: %w", name, err)
+				}
+				resolved[name] = pemBundle
+			}
+
+			entry, ok := wire[key{vol.Name, ds.ClusterTrustBundle.Path}]
+			if !ok {
+				// The wire spec is built from this same template, so a missing
+				// entry means the two views diverged — a bug, not user error.
+				return fmt.Errorf("internal error: no wire entry for ClusterTrustBundle %q at volume %q path %q", name, vol.Name, ds.ClusterTrustBundle.Path)
+			}
+			entry.PemBundle = pemBundle
+		}
+	}
+	return nil
+}

cmd/ateapi/internal/controlapi/cluster_trust_bundle_test.go

@@ -0,0 +1,149 @@
+//  Copyright 2026 Google LLC
+//
+//  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 controlapi
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"strings"
+	"testing"
+	"time"
+
+	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
+	certsv1beta1 "k8s.io/api/certificates/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	certlisters "k8s.io/client-go/listers/certificates/v1beta1"
+	"k8s.io/client-go/tools/cache"
+)
+
+// testCertPEM mints a throwaway self-signed certificate, PEM-encoded.
+func testCertPEM(t *testing.T) []byte {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	der, err := x509.CreateCertificate(rand.Reader, &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "test"},
+		NotBefore:    time.Now(),
+		NotAfter:     time.Now().Add(time.Hour),
+	}, &x509.Certificate{SerialNumber: big.NewInt(1)}, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
+}
+
+func ctbLister(t *testing.T, bundles ...*certsv1beta1.ClusterTrustBundle) certlisters.ClusterTrustBundleLister {
+	t.Helper()
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+	for _, b := range bundles {
+		if err := indexer.Add(b); err != nil {
+			t.Fatal(err)
+		}
+	}
+	return certlisters.NewClusterTrustBundleLister(indexer)
+}
+
+func ctbTemplate(volumeName, bundleName, path string) *atev1alpha1.ActorTemplate {
+	return &atev1alpha1.ActorTemplate{
+		Spec: atev1alpha1.ActorTemplateSpec{
+			Volumes: []atev1alpha1.Volume{{
+				Name: volumeName,
+				VolumeSource: atev1alpha1.VolumeSource{
+					SystemInfo: &atev1alpha1.SystemInfoVolumeSource{
+						DataSources: []atev1alpha1.SystemInfoDataSource{
+							{ClusterTrustBundle: &atev1alpha1.ClusterTrustBundleDataSource{Name: bundleName, Path: path}},
+						},
+					},
+				},
+			}},
+		},
+	}
+}
+
+func TestResolveClusterTrustBundles(t *testing.T) {
+	certPEM := testCertPEM(t)
+	junk := "garbage\n" + string(pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: []byte("x")}))
+
+	template := ctbTemplate("system-info", "egress-trust", "trust/ca.pem")
+	spec, err := workloadSpecFromActorTemplate(template, nil)
+	if err != nil {
+		t.Fatalf("workloadSpecFromActorTemplate: %v", err)
+	}
+
+	t.Run("resolves and sanitizes into the wire spec", func(t *testing.T) {
+		lister := ctbLister(t, &certsv1beta1.ClusterTrustBundle{
+			ObjectMeta: metav1.ObjectMeta{Name: "egress-trust"},
+			// Junk around the certificate proves kubelet-style sanitization:
+			// only the CERTIFICATE block survives, and the duplicate is dropped.
+			Spec: certsv1beta1.ClusterTrustBundleSpec{TrustBundle: junk + string(certPEM) + string(certPEM)},
+		})
+		if err := resolveClusterTrustBundles(lister, template, spec); err != nil {
+			t.Fatalf("resolveClusterTrustBundles: %v", err)
+		}
+		got := spec.GetVolumes()[0].GetSystemInfo().GetDataSources()[0].GetClusterTrustBundle()
+		if got.GetPath() != "trust/ca.pem" {
+			t.Errorf("path = %q, want %q", got.GetPath(), "trust/ca.pem")
+		}
+		if string(got.GetPemBundle()) != string(certPEM) {
+			t.Errorf("pem bundle = %q, want the sanitized certificate", got.GetPemBundle())
+		}
+	})
+
+	t.Run("missing bundle fails naming it", func(t *testing.T) {
+		spec, _ := workloadSpecFromActorTemplate(template, nil)
+		err := resolveClusterTrustBundles(ctbLister(t), template, spec)
+		if err == nil || !strings.Contains(err.Error(), `"egress-trust"`) || !strings.Contains(err.Error(), "not found") {
+			t.Errorf("error = %v, want not-found naming the bundle", err)
+		}
+	})
+
+	t.Run("unusable bundle fails naming it", func(t *testing.T) {
+		spec, _ := workloadSpecFromActorTemplate(template, nil)
+		lister := ctbLister(t, &certsv1beta1.ClusterTrustBundle{
+			ObjectMeta: metav1.ObjectMeta{Name: "egress-trust"},
+			Spec:       certsv1beta1.ClusterTrustBundleSpec{TrustBundle: junk},
+		})
+		err := resolveClusterTrustBundles(lister, template, spec)
+		if err == nil || !strings.Contains(err.Error(), `"egress-trust"`) || !strings.Contains(err.Error(), "unusable") {
+			t.Errorf("error = %v, want unusable-bundle naming the bundle", err)
+		}
+	})
+
+	t.Run("no CTB sources is a no-op even with a nil lister", func(t *testing.T) {
+		plain := &atev1alpha1.ActorTemplate{}
+		spec, _ := workloadSpecFromActorTemplate(plain, nil)
+		if err := resolveClusterTrustBundles(nil, plain, spec); err != nil {
+			t.Fatalf("resolveClusterTrustBundles: %v", err)
+		}
+	})
+
+	t.Run("CTB sources with a nil lister fail with a clear error, not a panic", func(t *testing.T) {
+		// nil lister = ateapi booted on a cluster without the feature-gated
+		// ClusterTrustBundle API (see cmd/ateapi/main.go).
+		spec, _ := workloadSpecFromActorTemplate(template, nil)
+		err := resolveClusterTrustBundles(nil, template, spec)
+		if err == nil || !strings.Contains(err.Error(), "does not serve the ClusterTrustBundle API") {
+			t.Errorf("error = %v, want API-unavailable error", err)
+		}
+	})
+}

cmd/ateapi/internal/controlapi/functional_test.go

@@ -315,6 +315,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 	ateletFactory, ateletInformer := AteletInformer(k8sClient)
 	scFactory := informers.NewSharedInformerFactory(k8sClient, 0)
 	scLister := scFactory.Storage().V1().StorageClasses().Lister()
+	ctbLister := scFactory.Certificates().V1beta1().ClusterTrustBundles().Lister()
 
 	substrateInformerFactory := externalversions.NewSharedInformerFactory(substrateClient, 0)
 	actorTemplateLister := substrateInformerFactory.Api().V1alpha1().ActorTemplates().Lister()
@@ -366,7 +367,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 	volPlugins := map[string]volume.VolumePluginControlPlane{
 		mockDriverName: mockPlugin,
 	}
-	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins)
+	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, ctbLister, dialer, instruments, "", volPlugins)
 
 	// 5. Start REAL gRPC Server for ATE API
 	grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ateinterceptors.ServerUnaryInterceptor))

cmd/ateapi/internal/controlapi/service.go

@@ -24,6 +24,7 @@ import (
 	"github.com/agent-substrate/substrate/internal/volume/csi"
 	listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	certlisters "k8s.io/client-go/listers/certificates/v1beta1"
 	storagev1listers "k8s.io/client-go/listers/storage/v1"
 )
 
@@ -59,6 +60,7 @@ func NewService(
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	csiDriverConfigLister listersv1alpha1.CSIDriverConfigLister,
 	storageClassLister storagev1listers.StorageClassLister,
+	clusterTrustBundleLister certlisters.ClusterTrustBundleLister,
 	dialer *AteletDialer,
 	instruments *Instruments,
 	egressGatewayAddress string,
@@ -75,7 +77,7 @@ func NewService(
 		instruments:           instruments,
 		volumePlugins:         volumePlugins,
 	}
-	s.actorWorkflow = NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s)
+	s.actorWorkflow = NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, clusterTrustBundleLister, instruments, egressGatewayAddress, s)
 	return s
 }
 

cmd/ateapi/internal/controlapi/workflow.go

@@ -30,6 +30,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	grpcCodes "google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	certlisters "k8s.io/client-go/listers/certificates/v1beta1"
 	storagev1listers "k8s.io/client-go/listers/storage/v1"
 )
 
@@ -67,17 +68,18 @@ func markSkipped(ctx context.Context, reason string) {
 
 // ActorWorkflow handles the workflows for actor's resume / suspend operations.
 type ActorWorkflow struct {
-	store                store.Interface
-	workerCache          *workercache.Cache
-	scheduler            scheduling.Scheduler
-	dialer               *AteletDialer
-	actorTemplateLister  listersv1alpha1.ActorTemplateLister
-	workerPoolLister     listersv1alpha1.WorkerPoolLister
-	sandboxConfigLister  listersv1alpha1.SandboxConfigLister
-	storageClassLister   storagev1listers.StorageClassLister
-	instruments          *Instruments
-	egressGatewayAddress string
-	pluginRegistry       VolumePluginRegistry
+	store                    store.Interface
+	workerCache              *workercache.Cache
+	scheduler                scheduling.Scheduler
+	dialer                   *AteletDialer
+	actorTemplateLister      listersv1alpha1.ActorTemplateLister
+	workerPoolLister         listersv1alpha1.WorkerPoolLister
+	sandboxConfigLister      listersv1alpha1.SandboxConfigLister
+	storageClassLister       storagev1listers.StorageClassLister
+	clusterTrustBundleLister certlisters.ClusterTrustBundleLister
+	instruments              *Instruments
+	egressGatewayAddress     string
+	pluginRegistry           VolumePluginRegistry
 }
 
 // NewActorWorkflow creates a new ActorWorkflow. instruments may be nil.
@@ -89,22 +91,24 @@ func NewActorWorkflow(
 	workerPoolLister listersv1alpha1.WorkerPoolLister,
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	storageClassLister storagev1listers.StorageClassLister,
+	clusterTrustBundleLister certlisters.ClusterTrustBundleLister,
 	instruments *Instruments,
 	egressGatewayAddress string,
 	pluginRegistry VolumePluginRegistry,
 ) *ActorWorkflow {
 	return &ActorWorkflow{
-		store:                store,
-		workerCache:          workerCache,
-		scheduler:            scheduling.New(workerCache, scheduling.WithMeter(otel.Meter("ateapi"))),
-		dialer:               dialer,
-		actorTemplateLister:  actorTemplateLister,
-		workerPoolLister:     workerPoolLister,
-		sandboxConfigLister:  sandboxConfigLister,
-		storageClassLister:   storageClassLister,
-		instruments:          instruments,
-		egressGatewayAddress: egressGatewayAddress,
-		pluginRegistry:       pluginRegistry,
+		store:                    store,
+		workerCache:              workerCache,
+		scheduler:                scheduling.New(workerCache, scheduling.WithMeter(otel.Meter("ateapi"))),
+		dialer:                   dialer,
+		actorTemplateLister:      actorTemplateLister,
+		workerPoolLister:         workerPoolLister,
+		sandboxConfigLister:      sandboxConfigLister,
+		storageClassLister:       storageClassLister,
+		clusterTrustBundleLister: clusterTrustBundleLister,
+		instruments:              instruments,
+		egressGatewayAddress:     egressGatewayAddress,
+		pluginRegistry:           pluginRegistry,
 	}
 }
 

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -611,6 +611,12 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou
 	if err != nil {
 		return tele, err
 	}
+	// The spec is about to be sent to atelet, so resolve referenced
+	// ClusterTrustBundles into it now; a missing or unusable bundle fails the
+	// actor start here, with an error naming the bundle.
+	if err := resolveClusterTrustBundles(w.clusterTrustBundleLister, actorTemplate, workloadSpec); err != nil {
+		return tele, err
+	}
 	egressGateway := w.egressGateway()
 
 	if local := actor.GetLocalSnapshotInfo(); local != nil {