diff --git a/.github/workflows/pr-workflow.yaml b/.github/workflows/pr-workflow.yaml index 68c9cc30d0..d0419089e9 100644 --- a/.github/workflows/pr-workflow.yaml +++ b/.github/workflows/pr-workflow.yaml @@ -120,14 +120,24 @@ jobs: E2E_SANDBOX_CLASS: microvm run: hack/run-e2e-kind.sh -v -args --no-color - name: Deploy MITM egress (sdsmint) - # Swap the passthrough egress gateway for the sdsmint variant, which - # mints per-SNI leaves from the egress-mitm-ca-pool (created here if - # missing). --deploy-atenet redeploys only the atenet components (the - # rest of the control plane is unchanged), keeping this step cheap. - # Cluster-wide, so it must come AFTER the standard lanes: once egress - # TLS is intercepted, their passthrough assumptions - # (TestActorEgressHTTPS's end-to-end TLS with the origin) no longer hold. - run: hack/install-ate-kind.sh --deploy-atenet --experimental-use-sdsmint + # Swap the passthrough egress gateway for the sdsmint variant (minting + # per-SNI leaves from the egress-mitm-ca-pool, created here if missing) + # and redeploy ate-api-server with --inject-egress-trust-bundle. The + # system deploy rather than --deploy-ate-apiserver: only it renders + # through the kind overlay that points ate-api-server at rustfs. + # Cluster-wide, so it must come AFTER the standard lanes: intercepted + # egress breaks their passthrough assumptions, and every actor started + # from here on needs the bundle. + run: hack/install-ate-kind.sh --deploy-ate-system --experimental-use-sdsmint + - name: Recreate counter demos under injection + # The pre-flip counter goldens no longer match the injected spec, and + # gVisor refuses mismatched restores (see docs/api-guide.md). The deploy + # keeps an existing template, so tear both demos down first. Demo-only + # steps on purpose: redeploying the control plane here would drop the + # sdsmint switch and the flag with it. + run: | + hack/install-ate-kind.sh --delete-demo-counter --delete-demo-counter-microvm + hack/install-ate-kind.sh --deploy-demo-counter --deploy-demo-counter-microvm - name: Run E2E tests (egress MITM trust) # The consumption half of the trust-bundle chain: an actor does TLS with # the MITM gateway's minted leaf using ONLY the projected bundle, plus a diff --git a/.gitignore b/.gitignore index b650ba710a..85c8a74a15 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Binaries /ateapi +/probe /substratectl /kubectl-ate /atelet diff --git a/cmd/ateapi/internal/controlapi/egresstrust.go b/cmd/ateapi/internal/controlapi/egresstrust.go new file mode 100644 index 0000000000..5b9b9bca3e --- /dev/null +++ b/cmd/ateapi/internal/controlapi/egresstrust.go @@ -0,0 +1,112 @@ +// 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 ( + "context" + "fmt" + "log/slog" + "strings" + + "github.com/agent-substrate/substrate/internal/proto/ateletpb" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +// The auto-injected egress trust volume: under --inject-egress-trust-bundle +// every actor trusts the egress gateway's MITM CA with nothing declared in +// its template. +const ( + // The '.' makes template collisions impossible: template volume names + // must be DNS labels, and ate.dev names belong to the platform. + egressTrustVolumeName = "trust.ate.dev" + + // Must stay on atelet's allowlist (cmd/atelet/trustbundle.go), which + // resolves it at every Run/Restore and fails actor start without it. + egressTrustBundleName = "egress-mitm.ate.dev" + + egressTrustMountPath = "/run/substrate/certs" + egressTrustBundleFile = "egress-mitm.ate.dev.pem" +) + +// injectEgressTrustVolume mounts the egress trust volume into every container +// except those that own the path (ownsEgressTrustPath); if all opt out, no +// volume is injected, so the actor is not gated on a bundle nothing reads. +func injectEgressTrustVolume(ctx context.Context, actor *ateapipb.Actor, spec *ateletpb.WorkloadSpec) error { + for _, vol := range spec.GetVolumes() { + if vol.GetName() == egressTrustVolumeName { + return fmt.Errorf("volume name %q is reserved for the injected egress trust bundle", egressTrustVolumeName) + } + } + mounted := false + for _, ctr := range spec.GetContainers() { + if path, ok := ownsEgressTrustPath(spec, ctr); ok { + slog.InfoContext(ctx, "Not mounting the egress trust volume: the template owns the path", + slog.String("atespace", actor.GetMetadata().GetAtespace()), + slog.String("actor", actor.GetMetadata().GetName()), + slog.String("container", ctr.GetName()), + slog.String("mountPath", path)) + continue + } + ctr.VolumeMounts = append(ctr.VolumeMounts, &ateletpb.VolumeMount{ + Name: egressTrustVolumeName, + MountPath: egressTrustMountPath, + }) + mounted = true + } + if !mounted { + return nil + } + spec.Volumes = append(spec.Volumes, &ateletpb.Volume{ + Name: egressTrustVolumeName, + Source: &ateletpb.Volume_SystemInfo{ + SystemInfo: &ateletpb.SystemInfoVolume{ + DataSources: []*ateletpb.SystemInfoDataSource{{ + DataSource: &ateletpb.SystemInfoDataSource_TrustBundle{ + TrustBundle: &ateletpb.TrustBundleDataSource{ + Name: egressTrustBundleName, + Path: egressTrustBundleFile, + }, + }, + }}, + }, + }, + }) + return nil +} + +// ownsEgressTrustPath returns the template mount that opts ctr out: at or +// below the reserved path, or an image volume above it — micro-VM bind +// ordering would shadow one mount with the other either way. +func ownsEgressTrustPath(spec *ateletpb.WorkloadSpec, ctr *ateletpb.Container) (string, bool) { + for _, m := range ctr.GetVolumeMounts() { + mp := m.GetMountPath() + if mp == egressTrustMountPath || strings.HasPrefix(mp, egressTrustMountPath+"/") { + return mp, true + } + if strings.HasPrefix(egressTrustMountPath, mp+"/") && isImageVolume(spec, m.GetName()) { + return mp, true + } + } + return "", false +} + +func isImageVolume(spec *ateletpb.WorkloadSpec, name string) bool { + for _, vol := range spec.GetVolumes() { + if vol.GetName() == name { + return vol.GetImage() != nil + } + } + return false +} diff --git a/cmd/ateapi/internal/controlapi/egresstrust_test.go b/cmd/ateapi/internal/controlapi/egresstrust_test.go new file mode 100644 index 0000000000..07021a8bdc --- /dev/null +++ b/cmd/ateapi/internal/controlapi/egresstrust_test.go @@ -0,0 +1,282 @@ +// 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 ( + "context" + "strings" + "testing" + + "github.com/agent-substrate/substrate/internal/proto/ateletpb" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/google/go-cmp/cmp" + "google.golang.org/protobuf/testing/protocmp" +) + +// egressTrustVolume is the volume injectEgressTrustVolume appends. +func egressTrustVolume() *ateletpb.Volume { + return &ateletpb.Volume{ + Name: egressTrustVolumeName, + Source: &ateletpb.Volume_SystemInfo{ + SystemInfo: &ateletpb.SystemInfoVolume{ + DataSources: []*ateletpb.SystemInfoDataSource{ + {DataSource: &ateletpb.SystemInfoDataSource_TrustBundle{ + TrustBundle: &ateletpb.TrustBundleDataSource{ + Name: egressTrustBundleName, + Path: egressTrustBundleFile, + }, + }}, + }, + }, + }, + } +} + +func egressTrustMount() *ateletpb.VolumeMount { + return &ateletpb.VolumeMount{Name: egressTrustVolumeName, MountPath: egressTrustMountPath} +} + +func TestInjectEgressTrustVolume(t *testing.T) { + tests := []struct { + name string + spec *ateletpb.WorkloadSpec + want *ateletpb.WorkloadSpec + }{ + { + name: "mounts the bundle into every container", + spec: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main"}, + {Name: "sidecar", VolumeMounts: []*ateletpb.VolumeMount{{Name: "home", MountPath: "/home/user"}}}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "home", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + }, + want: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{egressTrustMount()}}, + {Name: "sidecar", VolumeMounts: []*ateletpb.VolumeMount{{Name: "home", MountPath: "/home/user"}, egressTrustMount()}}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "home", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + egressTrustVolume(), + }, + }, + }, + { + name: "a container mounting the reserved path keeps its own mount", + spec: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{{Name: "trust", MountPath: egressTrustMountPath}}}, + {Name: "sidecar"}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "trust", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + }, + want: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{{Name: "trust", MountPath: egressTrustMountPath}}}, + {Name: "sidecar", VolumeMounts: []*ateletpb.VolumeMount{egressTrustMount()}}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "trust", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + egressTrustVolume(), + }, + }, + }, + { + name: "no volume at all when every container overrides", + spec: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{{Name: "trust", MountPath: egressTrustMountPath}}}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "trust", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + }, + want: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{{Name: "trust", MountPath: egressTrustMountPath}}}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "trust", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + }, + }, + { + name: "a mount nested below the reserved path opts the container out", + spec: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "extra", MountPath: egressTrustMountPath + "/extra"}, + }}, + {Name: "sidecar"}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "extra", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + }, + want: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "extra", MountPath: egressTrustMountPath + "/extra"}, + }}, + {Name: "sidecar", VolumeMounts: []*ateletpb.VolumeMount{egressTrustMount()}}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "extra", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + egressTrustVolume(), + }, + }, + }, + { + name: "a mount above the reserved path does not override", + spec: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "run", MountPath: "/run/substrate"}, + }}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "run", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + }, + want: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "run", MountPath: "/run/substrate"}, + egressTrustMount(), + }}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "run", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + egressTrustVolume(), + }, + }, + }, + { + // Guards the prefix boundary: a sibling sharing the reserved + // path as a string prefix is not below it. + name: "a sibling path sharing the prefix does not opt out", + spec: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "certs-extra", MountPath: egressTrustMountPath + "-extra"}, + }}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "certs-extra", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + }, + want: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "certs-extra", MountPath: egressTrustMountPath + "-extra"}, + egressTrustMount(), + }}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "certs-extra", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + egressTrustVolume(), + }, + }, + }, + { + // An image volume above the path opts out where a durableDir + // does not: the micro-VM runtime binds image volumes after + // systemInfo, which would shadow the injected mount. + name: "an image volume mounted above the reserved path opts the container out", + spec: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "rootfs", MountPath: "/run/substrate"}, + }}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "rootfs", Source: &ateletpb.Volume_Image{Image: &ateletpb.ImageVolumeSource{Reference: "img@sha256:abc"}}}, + }, + }, + want: &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{ + {Name: "main", VolumeMounts: []*ateletpb.VolumeMount{ + {Name: "rootfs", MountPath: "/run/substrate"}, + }}, + }, + Volumes: []*ateletpb.Volume{ + {Name: "rootfs", Source: &ateletpb.Volume_Image{Image: &ateletpb.ImageVolumeSource{Reference: "img@sha256:abc"}}}, + }, + }, + }, + { + name: "no containers, no volume", + spec: &ateletpb.WorkloadSpec{}, + want: &ateletpb.WorkloadSpec{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := injectEgressTrustVolume(context.Background(), nil, tt.spec); err != nil { + t.Fatalf("injectEgressTrustVolume() error: %v", err) + } + if diff := cmp.Diff(tt.want, tt.spec, protocmp.Transform()); diff != "" { + t.Errorf("injectEgressTrustVolume() mismatch (-want +got):\n%s", diff) + } + }) + } +} + +func TestInjectEgressTrustVolumeRejectsReservedName(t *testing.T) { + spec := &ateletpb.WorkloadSpec{ + Containers: []*ateletpb.Container{{Name: "main"}}, + Volumes: []*ateletpb.Volume{ + {Name: egressTrustVolumeName, Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}}, + }, + } + err := injectEgressTrustVolume(context.Background(), nil, spec) + if err == nil { + t.Fatalf("injectEgressTrustVolume() = nil, want reserved-name error") + } + if !strings.Contains(err.Error(), egressTrustVolumeName) || !strings.Contains(err.Error(), "reserved") { + t.Errorf("injectEgressTrustVolume() error %q, want it to name %q as reserved", err, egressTrustVolumeName) + } +} + +// TestWorkloadSpecInjectsEgressTrustVolume pins that the ActorWorkflow +// wrapper applies injection if and only if the deployment enables it. +func TestWorkloadSpecInjectsEgressTrustVolume(t *testing.T) { + template := &ateapipb.ActorTemplate{ + Metadata: &ateapipb.ResourceMetadata{Atespace: "agent-ns", Name: "tmpl1"}, + Containers: []*ateapipb.Container{{Name: "main", Image: "main"}}, + } + + for _, inject := range []bool{false, true} { + w := NewActorWorkflow(nil, nil, nil, nil, nil, nil, "", inject, nil, nil) + got, err := w.workloadSpec(context.Background(), template, nil) + if err != nil { + t.Fatalf("workloadSpec(inject=%v) error: %v", inject, err) + } + want := &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "main", Image: "main"}}} + if inject { + want.Containers[0].VolumeMounts = []*ateletpb.VolumeMount{egressTrustMount()} + want.Volumes = []*ateletpb.Volume{egressTrustVolume()} + } + if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" { + t.Errorf("workloadSpec(inject=%v) mismatch (-want +got):\n%s", inject, diff) + } + } +} diff --git a/cmd/ateapi/internal/controlapi/functionaltest/common_test.go b/cmd/ateapi/internal/controlapi/functionaltest/common_test.go index 05d7a3a615..a9f723e9c6 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/common_test.go +++ b/cmd/ateapi/internal/controlapi/functionaltest/common_test.go @@ -103,7 +103,15 @@ type testContext struct { // setupTest sets up a fully isolated test environment. func setupTest(t *testing.T, ns string) *testContext { t.Helper() - return setupTestWithVolumePlugins(t, ns, nil) + return setupTestFull(t, ns, nil, false) +} + +// setupTestWithEgressTrustInjection is setupTest with the service running +// --inject-egress-trust-bundle, for pinning the injected volume in the wire +// spec of each atelet RPC. +func setupTestWithEgressTrustInjection(t *testing.T, ns string) *testContext { + t.Helper() + return setupTestFull(t, ns, nil, true) } // setupTestWithVolumePlugins is setupTest with the default mock volume plugin @@ -111,6 +119,11 @@ func setupTest(t *testing.T, ns string) *testContext { // plugin pass it here rather than swapping it into the running RPCService, so each // test owns its own plugin set. func setupTestWithVolumePlugins(t *testing.T, ns string, plugins map[string]volume.VolumePluginControlPlane) *testContext { + t.Helper() + return setupTestFull(t, ns, plugins, false) +} + +func setupTestFull(t *testing.T, ns string, plugins map[string]volume.VolumePluginControlPlane, injectEgressTrustBundle bool) *testContext { t.Helper() // 1. Start an isolated PostgreSQL-backed store. persistence, cleanupStore := storetest.SetupTestStore(t) @@ -185,7 +198,7 @@ func setupTestWithVolumePlugins(t *testing.T, ns string, plugins map[string]volu } } objectStore := objectstoretest.New() - service := controlapi.NewRPCService(persistence, wc, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins, objectStore) + service := controlapi.NewRPCService(persistence, wc, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", injectEgressTrustBundle, volPlugins, objectStore) // 5. Start REAL gRPC Server for ATE API grpcServer := grpc.NewServer(grpc.ChainUnaryInterceptor( diff --git a/cmd/ateapi/internal/controlapi/functionaltest/egresstrust_test.go b/cmd/ateapi/internal/controlapi/functionaltest/egresstrust_test.go new file mode 100644 index 0000000000..879b1cb0be --- /dev/null +++ b/cmd/ateapi/internal/controlapi/functionaltest/egresstrust_test.go @@ -0,0 +1,166 @@ +// 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 functionaltest + +import ( + "context" + "testing" + + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +// Spelled as literals rather than imported: these names are the contract +// with atelet's allowlist and workload readers, so constant drift must fail +// here. +const ( + injectedTrustVolume = "trust.ate.dev" + injectedTrustBundle = "egress-mitm.ate.dev" + injectedTrustMountPath = "/run/substrate/certs" +) + +// TestActorLifecycle_InjectsEgressTrustVolume asserts the injected volume on +// every wire spec across resume, pause, resume-from-paused, and suspend, +// pinning that each RPC flow builds its spec through the injection wrapper. +func TestActorLifecycle_InjectsEgressTrustVolume(t *testing.T) { + ns := namespaceForTest("ns-egress-trust") + tc := setupTestWithEgressTrustInjection(t, ns) + defer tc.cleanup() + + createTemplate(t, tc, ns) + worker := createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1") + + const actorName = "actor-egress-trust" + ref := &ateapipb.ObjectRef{Atespace: testAtespace, Name: actorName} + if _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{ + Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: actorName}, + ActorTemplate: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "tmpl1"}, + }, + }); err != nil { + t.Fatalf("CreateActor failed: %v", err) + } + + // Boot skips the golden snapshot createTemplate seeds; without it the + // first resume is a Restore and no Run spec is ever recorded. + if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Actor: ref, Boot: true}); err != nil { + t.Fatalf("ResumeActor failed: %v", err) + } + assertEgressTrustInjected(t, "Run", lockedSpec(tc, func() *ateletpb.WorkloadSpec { return tc.fakeAtelet.RunRequest.GetSpec() })) + + if _, err := tc.client.PauseActor(context.Background(), &ateapipb.PauseActorRequest{Actor: ref}); err != nil { + t.Fatalf("PauseActor failed: %v", err) + } + pause := lockedCheckpoint(tc) + if got := pause.GetType(); got != ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL { + t.Errorf("pause checkpoint type = %v, want CHECKPOINT_TYPE_LOCAL", got) + } + assertEgressTrustInjected(t, "Checkpoint (pause)", pause.GetSpec()) + + // Pause and suspend release the worker after the RPC returns; the next + // resume needs it free again. + waitForWorkerAvailable(t, tc, worker) + if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Actor: ref}); err != nil { + t.Fatalf("ResumeActor from paused failed: %v", err) + } + assertEgressTrustInjected(t, "Restore", lockedSpec(tc, func() *ateletpb.WorkloadSpec { return tc.fakeAtelet.RestoreRequest.GetSpec() })) + + if _, err := tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{Actor: ref}); err != nil { + t.Fatalf("SuspendActor failed: %v", err) + } + // The type assertion is what proves this is the suspend checkpoint and + // not a stale read of the pause one. + suspend := lockedCheckpoint(tc) + if got := suspend.GetType(); got != ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL { + t.Errorf("suspend checkpoint type = %v, want CHECKPOINT_TYPE_EXTERNAL", got) + } + assertEgressTrustInjected(t, "Checkpoint (suspend)", suspend.GetSpec()) + + // Terminate: delete only runs from SUSPENDED or CRASHED, and only a + // still-assigned worker gets the RPC, so resume again and crash the + // actor in the store with its assignment intact. + waitForWorkerAvailable(t, tc, worker) + if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Actor: ref}); err != nil { + t.Fatalf("ResumeActor from suspended failed: %v", err) + } + actorRef := resources.ActorRef{Atespace: testAtespace, Name: actorName} + current, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Actor: ref}) + if err != nil { + t.Fatalf("GetActor failed: %v", err) + } + if _, err := tc.persistence.UpdateActor(context.Background(), actorRef, store.PreconditionFrom(current), func(toUpdate *ateapipb.Actor) error { + toUpdate.Status.State = ateapipb.ActorState_ACTOR_STATE_CRASHED + return nil + }); err != nil { + t.Fatalf("UpdateActor failed: %v", err) + } + if _, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{Actor: ref}); err != nil { + t.Fatalf("DeleteActor failed: %v", err) + } + assertEgressTrustInjected(t, "Terminate", lockedSpec(tc, func() *ateletpb.WorkloadSpec { return tc.fakeAtelet.TerminateRequest.GetSpec() })) +} + +func lockedSpec(tc *testContext, get func() *ateletpb.WorkloadSpec) *ateletpb.WorkloadSpec { + tc.fakeAtelet.Lock.Lock() + defer tc.fakeAtelet.Lock.Unlock() + return get() +} + +func lockedCheckpoint(tc *testContext) *ateletpb.CheckpointRequest { + tc.fakeAtelet.Lock.Lock() + defer tc.fakeAtelet.Lock.Unlock() + return tc.fakeAtelet.CheckpointRequest +} + +func assertEgressTrustInjected(t *testing.T, op string, spec *ateletpb.WorkloadSpec) { + t.Helper() + if spec == nil { + t.Fatalf("%s: no workload spec recorded", op) + } + + found := false + for _, vol := range spec.GetVolumes() { + if vol.GetName() != injectedTrustVolume { + continue + } + found = true + sources := vol.GetSystemInfo().GetDataSources() + if len(sources) != 1 { + t.Errorf("%s: volume %q has %d data sources, want 1", op, injectedTrustVolume, len(sources)) + break + } + tb := sources[0].GetTrustBundle() + if tb.GetName() != injectedTrustBundle || tb.GetPath() != injectedTrustBundle+".pem" { + t.Errorf("%s: volume %q projects {name %q, path %q}, want {name %q, path %q}", op, injectedTrustVolume, tb.GetName(), tb.GetPath(), injectedTrustBundle, injectedTrustBundle+".pem") + } + } + if !found { + t.Errorf("%s: spec is missing the injected volume %q", op, injectedTrustVolume) + } + + for _, ctr := range spec.GetContainers() { + mounted := false + for _, m := range ctr.GetVolumeMounts() { + if m.GetName() == injectedTrustVolume && m.GetMountPath() == injectedTrustMountPath { + mounted = true + } + } + if !mounted { + t.Errorf("%s: container %q is missing the injected mount %q at %q", op, ctr.GetName(), injectedTrustVolume, injectedTrustMountPath) + } + } +} diff --git a/cmd/ateapi/internal/controlapi/functionaltest/main_test.go b/cmd/ateapi/internal/controlapi/functionaltest/main_test.go index 6b74af0f3b..78cdcb7faa 100644 --- a/cmd/ateapi/internal/controlapi/functionaltest/main_test.go +++ b/cmd/ateapi/internal/controlapi/functionaltest/main_test.go @@ -127,6 +127,9 @@ type FakeAteletServer struct { UploadRequest *ateletpb.UploadPausedCheckpointRequest FailUpload error + TerminateCalled bool + TerminateRequest *ateletpb.TerminateRequest + // objectStore, when set, receives the objects a checkpoint or an upload // writes, so the control plane's copy and release steps have real external // snapshots to act on. setupTest points it at the test's own store. @@ -175,6 +178,8 @@ func (f *FakeAteletServer) Reset() { f.FailUpload = nil f.objectStore = nil + f.TerminateCalled = false + f.TerminateRequest = nil } func (f *FakeAteletServer) UploadPausedCheckpoint(ctx context.Context, req *ateletpb.UploadPausedCheckpointRequest) (*ateletpb.UploadPausedCheckpointResponse, error) { @@ -233,6 +238,15 @@ func (f *FakeAteletServer) Restore(ctx context.Context, req *ateletpb.RestoreReq return &ateletpb.RestoreResponse{}, nil } +func (f *FakeAteletServer) Terminate(ctx context.Context, req *ateletpb.TerminateRequest) (*ateletpb.TerminateResponse, error) { + f.Lock.Lock() + defer f.Lock.Unlock() + + f.TerminateCalled = true + f.TerminateRequest = proto.Clone(req).(*ateletpb.TerminateRequest) + return &ateletpb.TerminateResponse{}, nil +} + func (f *FakeAteletServer) lastRestoreRequest() *ateletpb.RestoreRequest { f.Lock.Lock() defer f.Lock.Unlock() diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go index 50cca164f7..f07d6a8a65 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -74,6 +74,7 @@ func NewRPCService( dialer *AteletDialer, instruments *Instruments, egressGatewayAddress string, + injectEgressTrustBundle bool, volumePlugins map[string]volume.VolumePluginControlPlane, objectStore objectstore.Store, ) *RPCService { @@ -89,7 +90,7 @@ func NewRPCService( volumePlugins: volumePlugins, objectStore: objectStore, } - s.actorWorkflow = NewActorWorkflow(impl, workerCache, dialer, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s, objectStore) + s.actorWorkflow = NewActorWorkflow(impl, workerCache, dialer, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, injectEgressTrustBundle, s, objectStore) s.workerWorkflow = NewWorkerWorkflow(impl) return s } diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index fc3873beb7..fb169455eb 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -77,8 +77,10 @@ type ActorWorkflow struct { storageClassLister storagev1listers.StorageClassLister instruments *Instruments egressGatewayAddress string - pluginRegistry VolumePluginRegistry - objectStore objectstore.Store + // injectEgressTrustBundle enables the policy in egresstrust.go. + injectEgressTrustBundle bool + pluginRegistry VolumePluginRegistry + objectStore objectstore.Store } // NewActorWorkflow creates a new ActorWorkflow. instruments may be nil. @@ -94,20 +96,22 @@ func NewActorWorkflow( storageClassLister storagev1listers.StorageClassLister, instruments *Instruments, egressGatewayAddress string, + injectEgressTrustBundle bool, pluginRegistry VolumePluginRegistry, objectStore objectstore.Store, ) *ActorWorkflow { return &ActorWorkflow{ - store: store, - workerCache: workerCache, - scheduler: scheduling.New(workerCache, scheduling.WithMeter(otel.Meter("ateapi"))), - dialer: dialer, - sandboxConfigLister: sandboxConfigLister, - storageClassLister: storageClassLister, - instruments: instruments, - egressGatewayAddress: egressGatewayAddress, - pluginRegistry: pluginRegistry, - objectStore: objectStore, + store: store, + workerCache: workerCache, + scheduler: scheduling.New(workerCache, scheduling.WithMeter(otel.Meter("ateapi"))), + dialer: dialer, + sandboxConfigLister: sandboxConfigLister, + storageClassLister: storageClassLister, + instruments: instruments, + egressGatewayAddress: egressGatewayAddress, + injectEgressTrustBundle: injectEgressTrustBundle, + pluginRegistry: pluginRegistry, + objectStore: objectStore, } } diff --git a/cmd/ateapi/internal/controlapi/workflow_delete.go b/cmd/ateapi/internal/controlapi/workflow_delete.go index 6c744dbb24..4ad54b8f07 100644 --- a/cmd/ateapi/internal/controlapi/workflow_delete.go +++ b/cmd/ateapi/internal/controlapi/workflow_delete.go @@ -163,7 +163,7 @@ func (w *ActorWorkflow) ensureAteletTerminated(ctx context.Context, actorRef res var workloadSpec *ateletpb.WorkloadSpec if actorTemplate != nil { - spec, err := workloadSpecFromActorTemplate(actorTemplate, actor) + spec, err := w.workloadSpec(ctx, actorTemplate, actor) if err != nil { return err } diff --git a/cmd/ateapi/internal/controlapi/workflow_pause.go b/cmd/ateapi/internal/controlapi/workflow_pause.go index 129011b255..8080d06857 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause.go +++ b/cmd/ateapi/internal/controlapi/workflow_pause.go @@ -178,7 +178,7 @@ func (w *ActorWorkflow) ensureAteletPaused(ctx context.Context, actorRef resourc } client := ateletpb.NewAteomHerderClient(ateletConn) - workloadSpec, err := workloadSpecFromActorTemplate(actorTemplate, actor) + workloadSpec, err := w.workloadSpec(ctx, actorTemplate, actor) if err != nil { return "", err } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index c3b6fca08d..7cbe9e9d10 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -657,7 +657,7 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou } client := ateletpb.NewAteomHerderClient(ateletConn) - workloadSpec, err := workloadSpecFromActorTemplate(actorTemplate, actor) + workloadSpec, err := w.workloadSpec(ctx, actorTemplate, actor) if err != nil { return tele, err } diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go index d76c9591ae..0a93e06906 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend.go @@ -231,7 +231,7 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso } client := ateletpb.NewAteomHerderClient(ateletConn) - workloadSpec, err := workloadSpecFromActorTemplate(actorTemplate, actor) + workloadSpec, err := w.workloadSpec(ctx, actorTemplate, actor) if err != nil { return "", err } diff --git a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go index af78631005..002ff33fe1 100644 --- a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go @@ -57,7 +57,7 @@ func newTestActorWorkflow(t *testing.T, st store.Interface, tmplAtespace, tmplNa }); err != nil && !errors.Is(err, store.ErrAlreadyExists) { t.Fatalf("create test ActorTemplate: %v", err) } - return NewActorWorkflow(st, nil, nil, nil, nil, nil, "", nil, objectstoretest.New()) + return NewActorWorkflow(st, nil, nil, nil, nil, nil, "", false, nil, objectstoretest.New()) } // newFinalizeWorkflow builds an ActorWorkflow over persistence with an diff --git a/cmd/ateapi/internal/controlapi/workload_spec.go b/cmd/ateapi/internal/controlapi/workload_spec.go index f5048ece26..765e231e0b 100644 --- a/cmd/ateapi/internal/controlapi/workload_spec.go +++ b/cmd/ateapi/internal/controlapi/workload_spec.go @@ -15,6 +15,7 @@ package controlapi import ( + "context" "fmt" "github.com/agent-substrate/substrate/internal/proto/ateletpb" @@ -46,6 +47,22 @@ func toAteletResources(r *ateapipb.Resources) (*ateletpb.ResourceLimits, error) return out, nil } +// workloadSpec lowers the template and applies deployment policy (the +// injected egress trust volume). Every atelet RPC must build its spec here +// so Run, Restore, Checkpoint, and Terminate agree on the volume set. +func (w *ActorWorkflow) workloadSpec(ctx context.Context, actorTemplate *ateapipb.ActorTemplate, actor *ateapipb.Actor) (*ateletpb.WorkloadSpec, error) { + spec, err := workloadSpecFromActorTemplate(actorTemplate, actor) + if err != nil { + return nil, err + } + if w.injectEgressTrustBundle { + if err := injectEgressTrustVolume(ctx, actor, spec); err != nil { + return nil, err + } + } + return spec, nil +} + // workloadSpecFromActorTemplate builds a WorkloadSpec from the template; // container env is copied verbatim. func workloadSpecFromActorTemplate(actorTemplate *ateapipb.ActorTemplate, actor *ateapipb.Actor) (*ateletpb.WorkloadSpec, error) { diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go index e35af167cf..fd9811be30 100644 --- a/cmd/ateapi/main.go +++ b/cmd/ateapi/main.go @@ -74,8 +74,9 @@ var ( postgresConnectionString = pflag.String("postgres-connection-string", "", "PostgreSQL connection string (libpq DSN or URI).") postgresSchema = pflag.String("postgres-schema", "public", "PostgreSQL schema for Substrate tables. This overrides a search_path connection parameter.") - actorIDJWTPoolFile = pflag.String("actor-id-jwt-pool", "", "The file that contains the serialized JWT authority pool for signing actor JWTs") - egressGatewayAddress = pflag.String("egress-gateway-address", "", "Address of the egress PEP. Empty disables tunneled egress.") + actorIDJWTPoolFile = pflag.String("actor-id-jwt-pool", "", "The file that contains the serialized JWT authority pool for signing actor JWTs") + egressGatewayAddress = pflag.String("egress-gateway-address", "", "Address of the egress PEP. Empty disables tunneled egress.") + injectEgressTrustBundle = pflag.Bool("inject-egress-trust-bundle", false, "Mount the egress-mitm.ate.dev trust bundle into every actor at /run/substrate/certs/egress-mitm.ate.dev.pem. Requires the sdsmint (MITM) egress gateway, which publishes that bundle; without it, actors fail to start. Set at install time: changing it later invalidates existing golden and paused snapshots.") actorIDCAPoolFile = pflag.String("actor-id-ca-pool", "", "The file that contains the CA pool for signing actor JWTs") podIdentityCACerts = pflag.String("pod-identity-ca-certs", "", "The file that contains the pod-identity CA bundle, used both for verifying client certificates presented to the gRPC server and for verifying atelet serving certificates when dialing atelet. If empty, client-cert verification is disabled and atelet dials will fail.") @@ -198,7 +199,7 @@ func main() { volPlugins := make(map[string]volume.VolumePluginControlPlane) ateletDialer := controlapi.NewAteletDialer(workerPodInformer.GetIndexer(), ateletPodInformer.GetIndexer(), *ateletClientCredBundle, *podIdentityCACerts) - controlSrv := controlapi.NewRPCService(persistence, workerCache, sandboxConfigLister, csiDriverConfigLister, storageClassLister, ateletDialer, instruments, *egressGatewayAddress, volPlugins, objectStore) + controlSrv := controlapi.NewRPCService(persistence, workerCache, sandboxConfigLister, csiDriverConfigLister, storageClassLister, ateletDialer, instruments, *egressGatewayAddress, *injectEgressTrustBundle, volPlugins, objectStore) // Drive stored ActorTemplates through the golden actor flow. templateReconciler := controlapi.NewActorTemplateReconciler(persistence, controlSrv) diff --git a/docs/api-guide.md b/docs/api-guide.md index 58f9532feb..bdc9a3d08d 100644 --- a/docs/api-guide.md +++ b/docs/api-guide.md @@ -203,6 +203,17 @@ spec: atelet resolves the bundle on the node when the actor starts, reading the backing object through a cluster-wide watch (the same informer dynamic refresh will later hang off) and sanitizing it the way kubelet does for projections: only `CERTIFICATE` PEM blocks are kept, deduplicated, with block headers stripped and the anchors deliberately shuffled — order carries no meaning, so consumers must not depend on it. The actor itself never talks to any bundle backend. Starting the actor fails, with an error naming the bundle, if the name is not on the allowlist, the bundle's backend is unavailable in this deployment, or the resolved bundle is missing, empty, or contains no certificates. Bundle contents are re-resolved on every Run/Restore. +##### The auto-injected egress trust volume + +A deployment that intercepts actor egress TLS (the MITM egress gateway) can run ateapi with `--inject-egress-trust-bundle`. Every actor then receives a read-only volume projecting the `egress-mitm.ate.dev` bundle to `/run/substrate/certs/egress-mitm.ate.dev.pem`, with no template involvement — the example above becomes unnecessary for the egress bundle. + +- The injected volume is named `trust.ate.dev`. Template volume names must be DNS labels, so no template volume can collide with it — like worker metadata keys under `ate.dev/`, names in the `ate.dev` domain belong to the platform. +- A container that mounts one of its own volumes at `/run/substrate/certs` — or anywhere below it — keeps its mounts and receives no injected one: declaring the path is how a template takes ownership of it, the override rule kubelet applies to serviceaccount token mounts. (Mounts below the path opt out too because the injected parent bind would shadow them on the micro-VM runtime, which mounts binds grouped by volume kind rather than parent-first. An `image` volume mounted above the path opts the container out for the mirror-image reason: the micro-VM runtime binds image volumes after systemInfo, so that parent would shadow the injected mount.) Every opt-out is logged by ateapi with the actor, container, and colliding mount path. If every container opts out, no volume is injected at all. +- Injection is fail-closed like any trustBundle projection: while the bundle is absent, actors that would receive the injection do not start. Enable the flag only in deployments that publish the bundle; `hack/install-ate.sh` passes it whenever an invocation that deploys ate-api-server runs under `--experimental-use-sdsmint`. +- The volume is per-activation policy, not part of the template: it never appears in the ActorTemplate resource. Enable the flag at install time, before templates and workloads exist — it takes effect for sandboxes created afterwards (cold boots, and the golden snapshots captured from them). A snapshot taken under the other setting no longer matches the spec it would be restored with: gVisor refuses such a restore outright, and a micro-VM guest resumes its snapshotted mount table without the file until its next cold boot. Flipping the flag on a live deployment therefore means recreating templates (regenerating their goldens) and discarding paused or suspended actors. + +Workloads pick the anchors up by pointing their TLS stack at the file (for example `SSL_CERT_FILE=/run/substrate/certs/egress-mitm.ate.dev.pem` for OpenSSL-based stacks); substrate does not set environment variables on the actor's behalf. + ### Container Fields Each entry in `containers` describes one process to run in the actor's sandbox. diff --git a/hack/install-ate.sh b/hack/install-ate.sh index f4c2f16e86..814220b138 100755 --- a/hack/install-ate.sh +++ b/hack/install-ate.sh @@ -79,7 +79,9 @@ function usage() { echo "" echo "Experiments:" echo "" - echo " --experimental-use-sdsmint Deploy the egress gateway with per-SNI certificate minting (experimental)" + echo " --experimental-use-sdsmint Deploy the egress gateway with per-SNI certificate minting, and run" + echo " ate-api-server (when this invocation deploys it) with" + echo " --inject-egress-trust-bundle (experimental)" echo " --experimental-additional-egress-extproc-service NS/SVC:PORT" echo " Run an additional ext_proc authorization filter, served by that Service." echo " Requires --experimental-use-sdsmint. (experimental)" @@ -350,6 +352,60 @@ atenet_egress_manifest() { fi } +# patch_ate_api_server_manifest adds --inject-egress-trust-bundle under +# --experimental-use-sdsmint, guarded like ensure_egress_mitm_ca_pool_secret. +# Spliced into the args because kustomize strips comment markers. +patch_ate_api_server_manifest() { + if [[ "$(atenet_router)" == "agentgateway" || "${ATE_EXPERIMENTAL_USE_SDSMINT:-false}" != "true" ]]; then + cat + return + fi + # index() rather than a regex class: the runner's awk is mawk, whose + # POSIX-class support is unreliable, and a non-matching pattern degrades + # to a silent passthrough. + awk '{ + print + i = index($0, "- --egress-gateway-address=") + if (i > 0 && substr($0, 1, i - 1) ~ /^ *$/) { + print substr($0, 1, i - 1) "- --inject-egress-trust-bundle" + } + }' +} + +# wait_for_ate_api_server_drain waits for replaced ate-api-server pods to +# exit. rollout status returns once they are marked deleted, but +# --drain-delay keeps them serving new RPCs after SIGTERM — work sequenced +# right after could still get specs built under the previous flags. +wait_for_ate_api_server_drain() { + for _ in $(seq 1 30); do + run_kubectl -n ate-system get pods -l app=ate-api-server \ + -o jsonpath='{.items[*].metadata.deletionTimestamp}' | grep -q . || return 0 + sleep 2 + done + echo "Error: timed out waiting for replaced ate-api-server pods to exit" >&2 + return 1 +} + +# verify_ate_api_server_injection fails the install when the live egress +# gateway is the sdsmint variant but ate-api-server lacks injection. Checks +# cluster state, not invocation flags: a plain redeploy must not strip it. +verify_ate_api_server_injection() { + local egress_containers="" + egress_containers="$(run_kubectl -n ate-system get deployment atenet-egress \ + -o jsonpath='{.spec.template.spec.initContainers[*].name} {.spec.template.spec.containers[*].name}' 2>/dev/null || true)" + local api_args="" + api_args="$(run_kubectl -n ate-system get deployment ate-api-server \ + -o jsonpath='{.spec.template.spec.containers[0].args}')" + if [[ "${egress_containers}" == *sdsmint* && "${api_args}" != *inject-egress-trust-bundle* ]]; then + echo "Error: the sdsmint (MITM) egress gateway is deployed but ate-api-server lacks --inject-egress-trust-bundle;" >&2 + echo "redeploy ate-api-server with --experimental-use-sdsmint so actors keep receiving the egress trust bundle." >&2 + return 1 + fi + if [[ "${egress_containers}" != *sdsmint* && "${api_args}" == *inject-egress-trust-bundle* ]]; then + echo "Warning: ate-api-server runs --inject-egress-trust-bundle but the deployed egress gateway is not the sdsmint variant." >&2 + fi +} + render_atenet_egress_manifest() { if [[ "$(atenet_router)" == "agentgateway" ]]; then # The markers live inside Envoy's bootstrap, so there is nowhere here to @@ -949,7 +1005,7 @@ deploy_ate_system() { fi local manifests="" - manifests="$(render_ate_system_manifests)" + manifests="$(render_ate_system_manifests | patch_ate_api_server_manifest)" echo "${manifests}" | run_kubectl apply -f - reconcile_cloudsql_proxy_sidecar @@ -965,6 +1021,8 @@ deploy_ate_system() { run_kubectl rollout status statefulset/postgres -n ate-system --timeout="$(rollout_timeout)" fi run_kubectl rollout status deployment/ate-api-server -n ate-system --timeout="$(rollout_timeout)" + wait_for_ate_api_server_drain + verify_ate_api_server_injection run_kubectl rollout status deployment/ate-controller -n ate-system --timeout="$(rollout_timeout)" run_kubectl rollout status deployment/atenet-router -n ate-system --timeout="$(rollout_timeout)" run_kubectl rollout status deployment/atenet-egress -n ate-system --timeout="$(rollout_timeout)" @@ -1005,9 +1063,11 @@ deploy_ate_apiserver() { apply_otel_config apply_otel_endpoint_override - run_ko apply -f manifests/ate-install/ate-api-server.yaml + run_ko resolve -f manifests/ate-install/ate-api-server.yaml | patch_ate_api_server_manifest | run_kubectl apply -f - reconcile_cloudsql_proxy_sidecar run_kubectl rollout status deployment/ate-api-server -n ate-system --timeout="$(rollout_timeout)" + wait_for_ate_api_server_drain + verify_ate_api_server_injection } # Reconciles the Cloud SQL Auth Proxy sidecar and Workload Identity diff --git a/internal/e2e/fixtures/probe/main.go b/internal/e2e/fixtures/probe/main.go index f2a5fe1606..719541df89 100644 --- a/internal/e2e/fixtures/probe/main.go +++ b/internal/e2e/fixtures/probe/main.go @@ -46,6 +46,11 @@ const ( trustFile = "/run/ate/trust-bundle.pem" ) +// injectedTrustFile is where ateapi's auto-injected egress trust volume +// lands. Nothing in probe.yaml.tmpl declares it; that absence is what the +// injection e2e asserts on. +const injectedTrustFile = "/run/substrate/certs/egress-mitm.ate.dev.pem" + // procStatus is where the kernel reports this process's capability sets. Asking // the kernel — rather than reading back the OCI spec atelet wrote — is the whole // point: it is what proves the sandbox actually applied the requested set. @@ -295,8 +300,9 @@ func memTotalBytes() (int64, error) { // fetch GETs ?url= over the actor's normal egress path and reports the // outcome, doing TLS with the trust anchors selected by ?roots=: "bundle" -// (the default) loads the projected trust bundle at trustFile, "system" uses -// the image's system roots. TestActorEgressMITMTrust documents why each mode +// (the default) loads the projected trust bundle at trustFile, "injected" +// loads the auto-injected bundle at injectedTrustFile, "system" uses the +// image's system roots. TestActorEgressMITMTrust documents why each mode // passes or fails. TLS failures land in the "error" field rather than the // HTTP status: a verification failure is a result for the suite to assert // on, not a broken probe. @@ -308,20 +314,25 @@ func fetch(w http.ResponseWriter, r *http.Request) { writeJSON(w, resp) return } + rootsFile := "" roots := r.URL.Query().Get("roots") switch roots { - case "", "bundle", "system": + case "", "bundle": + rootsFile = trustFile + case "injected": + rootsFile = injectedTrustFile + case "system": default: // Fail closed on typos: silently treating an unknown value as // "bundle" would flip a suite's negative control into a positive // fetch with a misleading failure message. - resp["error"] = "unknown roots value " + strconv.Quote(roots) + " (want bundle or system)" + resp["error"] = "unknown roots value " + strconv.Quote(roots) + " (want bundle, injected, or system)" writeJSON(w, resp) return } tlsCfg := &tls.Config{} - if roots != "system" { - b, err := os.ReadFile(trustFile) + if rootsFile != "" { + b, err := os.ReadFile(rootsFile) if err != nil { resp["error"] = "reading trust bundle: " + err.Error() writeJSON(w, resp) @@ -329,7 +340,7 @@ func fetch(w http.ResponseWriter, r *http.Request) { } pool := x509.NewCertPool() if !pool.AppendCertsFromPEM(b) { - resp["error"] = "no certificates parsed from " + trustFile + resp["error"] = "no certificates parsed from " + rootsFile writeJSON(w, resp) return } diff --git a/internal/e2e/suites/egressmitm/egressmitm_test.go b/internal/e2e/suites/egressmitm/egressmitm_test.go index 806d9dd114..6af454b133 100644 --- a/internal/e2e/suites/egressmitm/egressmitm_test.go +++ b/internal/e2e/suites/egressmitm/egressmitm_test.go @@ -22,7 +22,9 @@ package egressmitm import ( "context" "encoding/json" + "encoding/pem" "io" + "maps" "net/http" "net/url" "os" @@ -33,6 +35,7 @@ import ( "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) const probeTemplate = "probe" @@ -56,9 +59,10 @@ var probeNamespace string // the passthrough gateway cluster-wide, so CI runs it as separate steps // after the standard lanes (see pr-workflow.yaml) — once per sandbox class, // since trust delivery differs per class (gVisor RO bind vs the micro-VM -// unified virtio-fs share). Locally: +// unified virtio-fs share). TestActorEgressTrustAutoInjection additionally +// needs ate-api-server running --inject-egress-trust-bundle, so: // -// hack/install-ate-kind.sh --deploy-atenet --experimental-use-sdsmint +// hack/install-ate-kind.sh --deploy-ate-system --experimental-use-sdsmint // E2E_EGRESS_MITM=1 hack/run-e2e-kind.sh ./internal/e2e/suites/egressmitm -v -args --no-color // E2E_EGRESS_MITM=1 E2E_SANDBOX_CLASS=microvm hack/run-e2e-kind.sh ./internal/e2e/suites/egressmitm -v -args --no-color // @@ -66,7 +70,7 @@ var probeNamespace string // (hack/run-microvm-demo-kind.sh, or hack/install-microvm-deps.sh --install). func TestActorEgressMITMTrust(t *testing.T) { if os.Getenv("E2E_EGRESS_MITM") == "" { - t.Skip("needs the sdsmint (MITM) egress gateway: deploy with hack/install-ate-kind.sh --deploy-atenet --experimental-use-sdsmint, then set E2E_EGRESS_MITM=1") + t.Skip("needs the sdsmint (MITM) egress gateway: deploy with hack/install-ate-kind.sh --deploy-ate-system --experimental-use-sdsmint, then set E2E_EGRESS_MITM=1") } env, err := e2e.CheckEnv("BUCKET_NAME", "KO_DOCKER_REPO") if err != nil { @@ -88,8 +92,8 @@ func TestActorEgressMITMTrust(t *testing.T) { probeNamespace, _ = e2e.DeployProbe(t, env["BUCKET_NAME"], "egressmitm", e2e.WithTrustBundle()) const id = "probe-mitm" - createAndResumeActor(t, ctx, clients, id) - waitForActorState(t, ctx, clients, id, ateapipb.ActorState_ACTOR_STATE_RUNNING) + createAndResumeActor(t, ctx, clients, probeNamespace, id) + waitForActorState(t, ctx, clients, probeNamespace, id, ateapipb.ActorState_ACTOR_STATE_RUNNING) rc, err := e2e.NewRouterClient(ctx) if err != nil { @@ -108,7 +112,7 @@ func TestActorEgressMITMTrust(t *testing.T) { deadline := time.Now().Add(2 * time.Minute) var pos fetchResponse for { - pos = probeFetch(t, ctx, rc, id, origin, "bundle") + pos = probeFetch(t, ctx, rc, probeNamespace, id, origin, "bundle") isCertErr := strings.Contains(pos.Error, "certificate") || strings.Contains(pos.Error, "x509") if pos.Error == "" || !isCertErr || time.Now().After(deadline) { break @@ -122,7 +126,7 @@ func TestActorEgressMITMTrust(t *testing.T) { t.Fatalf("fetch %s via projected bundle: status %s, want 200", origin, pos.Status) } - neg := probeFetch(t, ctx, rc, id, origin, "system") + neg := probeFetch(t, ctx, rc, probeNamespace, id, origin, "system") if neg.Error == "" { t.Errorf("fetch with system roots unexpectedly succeeded (status %s): the minted leaf should chain to no public CA — is the sdsmint (MITM) gateway actually deployed, or is egress running in passthrough mode?", neg.Status) } else if !strings.Contains(neg.Error, "certificate") && !strings.Contains(neg.Error, "x509") { @@ -130,19 +134,151 @@ func TestActorEgressMITMTrust(t *testing.T) { } } +// TestActorEgressTrustAutoInjection deploys the probe WITHOUT WithTrustBundle, +// so anything it reads arrived through ateapi's injection: the injected file +// must match the published bundle (as cert sets — sanitization shuffles), and +// TLS through the MITM gateway must complete with only those anchors. +// +// Needs --inject-egress-trust-bundle on ateapi as well as the sdsmint +// gateway. Locally: +// +// hack/install-ate-kind.sh --deploy-ate-system --experimental-use-sdsmint +// E2E_EGRESS_MITM=1 hack/run-e2e-kind.sh ./internal/e2e/suites/egressmitm -v -args --no-color +func TestActorEgressTrustAutoInjection(t *testing.T) { + if os.Getenv("E2E_EGRESS_MITM") == "" { + t.Skip("needs the sdsmint (MITM) egress gateway and injection: deploy with hack/install-ate-kind.sh --deploy-ate-system --experimental-use-sdsmint, then set E2E_EGRESS_MITM=1") + } + env, err := e2e.CheckEnv("BUCKET_NAME", "KO_DOCKER_REPO") + if err != nil { + t.Fatalf("CheckEnv failed: %v", err) + } + ctx := context.Background() + clients := e2e.GetClients() + + // Without WithTrustBundle, DeployProbe does not wait for the bundle, but + // under injection even this fixture's golden boot fails closed without it. + e2e.EnsureEgressTrustBundle(t, ctx, clients) + + ns, _ := e2e.DeployProbe(t, env["BUCKET_NAME"], "inject") + + const id = "probe-inject" + createAndResumeActor(t, ctx, clients, ns, id) + waitForActorState(t, ctx, clients, ns, id, ateapipb.ActorState_ACTOR_STATE_RUNNING) + + rc, err := e2e.NewRouterClient(ctx) + if err != nil { + t.Fatalf("NewRouterClient: %v", err) + } + defer rc.Close() + + const injectedPath = "/run/substrate/certs/egress-mitm.ate.dev.pem" + read := probeReadFile(t, ctx, rc, ns, id, injectedPath) + if read.Error != "" { + t.Fatalf("reading %s in the actor: %s — is ateapi running with --inject-egress-trust-bundle? (redeploy with hack/install-ate-kind.sh --deploy-ate-system --experimental-use-sdsmint)", injectedPath, read.Error) + } + ctb, err := clients.K8s.CertificatesV1beta1().ClusterTrustBundles().Get(ctx, e2e.EgressTrustBundleObjectName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("getting ClusterTrustBundle %q: %v", e2e.EgressTrustBundleObjectName, err) + } + gotCerts := certSet(t, read.Content, injectedPath) + wantCerts := certSet(t, ctb.Spec.TrustBundle, "ClusterTrustBundle "+e2e.EgressTrustBundleObjectName) + if !maps.Equal(gotCerts, wantCerts) { + t.Errorf("injected bundle at %s carries %d certificate(s) that do not match the %d in ClusterTrustBundle %q", injectedPath, len(gotCerts), len(wantCerts), e2e.EgressTrustBundleObjectName) + } + + // Same propagation-retry rationale as TestActorEgressMITMTrust. + const origin = "https://example.com/" + deadline := time.Now().Add(2 * time.Minute) + var pos fetchResponse + for { + pos = probeFetch(t, ctx, rc, ns, id, origin, "injected") + isCertErr := strings.Contains(pos.Error, "certificate") || strings.Contains(pos.Error, "x509") + if pos.Error == "" || !isCertErr || time.Now().After(deadline) { + break + } + time.Sleep(5 * time.Second) + } + if pos.Error != "" { + t.Fatalf("TLS through the MITM egress gateway with the injected trust bundle failed: %s", pos.Error) + } + if pos.Status != "200" { + t.Fatalf("fetch %s via injected bundle: status %s, want 200", origin, pos.Status) + } +} + +// certSet parses every CERTIFICATE block into a set of DER bytes, the +// comparable form of a sanitized (deduplicated, shuffled) projection. +func certSet(t *testing.T, pemBundle, source string) map[string]bool { + t.Helper() + set := map[string]bool{} + rest := []byte(pemBundle) + for { + var block *pem.Block + block, rest = pem.Decode(rest) + if block == nil { + break + } + if block.Type == "CERTIFICATE" { + set[string(block.Bytes)] = true + } + } + if len(set) == 0 { + t.Fatalf("no CERTIFICATE PEM blocks in %s", source) + } + return set +} + type fetchResponse struct { Status string `json:"status"` Error string `json:"error"` } +type readFileResponse struct { + Content string `json:"content"` + Error string `json:"error"` +} + +// probeReadFile reads a file inside the actor via the probe, with the same +// router-warmup retry as probeFetch. Probe-level read failures are results, +// returned for the caller to assert on. +func probeReadFile(t *testing.T, ctx context.Context, rc *e2e.RouterClient, ns, id, filePath string) readFileResponse { + t.Helper() + path := "/readfile?path=" + url.QueryEscape(filePath) + ref := resources.ActorRef{Atespace: ns, Name: id} + + deadline := time.Now().Add(30 * time.Second) + for { + resp, err := rc.Get(ctx, ref, path) + if err != nil { + t.Fatalf("GET %s for %q: %v", path, id, err) + } + body, readErr := io.ReadAll(resp.Body) + resp.Body.Close() + if readErr != nil { + t.Fatalf("reading %s response for %q: %v", path, id, readErr) + } + if resp.StatusCode == http.StatusOK { + var out readFileResponse + if err := json.Unmarshal(body, &out); err != nil { + t.Fatalf("decoding %s response for %q: %v (body %q)", path, id, err, body) + } + return out + } + if time.Now().After(deadline) { + t.Fatalf("GET %s for %q: status %d, body %q", path, id, resp.StatusCode, body) + } + time.Sleep(2 * time.Second) + } +} + // probeFetch asks the probe to fetch origin with the given roots mode. // Router-level failures are retried for up to 30s (a resume can return // before the route reaches the router's xDS snapshot); probe-level TLS // failures are results, returned for the caller to assert on. -func probeFetch(t *testing.T, ctx context.Context, rc *e2e.RouterClient, id, origin, roots string) fetchResponse { +func probeFetch(t *testing.T, ctx context.Context, rc *e2e.RouterClient, ns, id, origin, roots string) fetchResponse { t.Helper() path := "/fetch?roots=" + roots + "&url=" + url.QueryEscape(origin) - ref := resources.ActorRef{Atespace: probeNamespace, Name: id} + ref := resources.ActorRef{Atespace: ns, Name: id} deadline := time.Now().Add(30 * time.Second) for { @@ -173,21 +309,21 @@ func probeFetch(t *testing.T, ctx context.Context, rc *e2e.RouterClient, id, ori // lifecycle (actor records outlive the fixture namespace); DeployProbe has // already waited for the template's golden snapshot. -func createAndResumeActor(t *testing.T, ctx context.Context, clients *e2e.Clients, id string) { +func createAndResumeActor(t *testing.T, ctx context.Context, clients *e2e.Clients, ns, id string) { t.Helper() - ref := &ateapipb.ObjectRef{Atespace: probeNamespace, Name: id} + ref := &ateapipb.ObjectRef{Atespace: ns, Name: id} _, _ = clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: ref}) _, _ = clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{Actor: ref}) if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ - Metadata: &ateapipb.ResourceMetadata{Atespace: probeNamespace, Name: id}, - ActorTemplate: &ateapipb.ObjectRef{Atespace: probeNamespace, Name: probeTemplate}, + Metadata: &ateapipb.ResourceMetadata{Atespace: ns, Name: id}, + ActorTemplate: &ateapipb.ObjectRef{Atespace: ns, Name: probeTemplate}, }}); err != nil { t.Fatalf("CreateActor %q: %v", id, err) } t.Cleanup(func() { _, _ = clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: ref}) if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{Actor: ref}); err != nil { - t.Logf("cleanup: DeleteActor %q failed, actor leaked (remove with: kubectl ate delete actor %s -a %s): %v", id, id, probeNamespace, err) + t.Logf("cleanup: DeleteActor %q failed, actor leaked (remove with: kubectl ate delete actor %s -a %s): %v", id, id, ns, err) } }) if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{Actor: ref}); err != nil { @@ -195,12 +331,12 @@ func createAndResumeActor(t *testing.T, ctx context.Context, clients *e2e.Client } } -func waitForActorState(t *testing.T, ctx context.Context, clients *e2e.Clients, actorName string, want ateapipb.ActorState) { +func waitForActorState(t *testing.T, ctx context.Context, clients *e2e.Clients, ns, actorName string, want ateapipb.ActorState) { t.Helper() deadline := time.Now().Add(60 * time.Second) for time.Now().Before(deadline) { resp, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{ - Actor: &ateapipb.ObjectRef{Atespace: probeNamespace, Name: actorName}, + Actor: &ateapipb.ObjectRef{Atespace: ns, Name: actorName}, }) if err == nil && resp.GetStatus().GetState() == want { return