feat: pluggable snapshot storage broker with per-snapshot signed URLs

cmd/ateapi/internal/controlapi/broker_wiring.go

@@ -0,0 +1,87 @@
+// 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"
+	"time"
+
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/storagebroker"
+	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
+)
+
+// snapshotCapabilityTTL bounds how long a minted snapshot read/write capability
+// is valid. It must comfortably exceed a restore or checkpoint transfer, but
+// stay short so a leaked URL exposes one snapshot only briefly.
+const snapshotCapabilityTTL = 15 * time.Minute
+
+// readAccessFor mints a read capability for each snapshot URI the node will
+// read, returning them keyed by URI. A restore that combines the actor's own
+// snapshot with a golden snapshot passes both; the caller may pass an unset
+// (empty) URI unconditionally and it is skipped. With no broker configured it
+// returns nil and the node reads with its built-in storage client.
+func (w *ActorWorkflow) readAccessFor(ctx context.Context, snapshotURIs ...string) (map[string]*ateletpb.SignedObjectAccess, error) {
+	if w.broker == nil {
+		return nil, nil
+	}
+	return mintAccess(ctx, w.broker.MintRead, snapshotURIs...)
+}
+
+// writeAccessFor mints a write capability for each snapshot URI the node will
+// write, returning them keyed by URI. With no broker configured it returns nil.
+func (w *ActorWorkflow) writeAccessFor(ctx context.Context, snapshotURIs ...string) (map[string]*ateletpb.SignedObjectAccess, error) {
+	if w.broker == nil {
+		return nil, nil
+	}
+	return mintAccess(ctx, w.broker.MintWrite, snapshotURIs...)
+}
+
+// mintAccess mints one capability per distinct, non-empty URI with mint and maps
+// each onto the wire message keyed by URI. It returns nil (not an empty map)
+// when nothing was minted, so a caller can assign the result directly.
+func mintAccess(ctx context.Context, mint func(context.Context, string, time.Duration) (storagebroker.Capability, error), snapshotURIs ...string) (map[string]*ateletpb.SignedObjectAccess, error) {
+	out := make(map[string]*ateletpb.SignedObjectAccess, len(snapshotURIs))
+	for _, uri := range snapshotURIs {
+		if uri == "" || out[uri] != nil {
+			continue
+		}
+		cap, err := mint(ctx, uri, snapshotCapabilityTTL)
+		if err != nil {
+			return nil, fmt.Errorf("minting snapshot capability for %q: %w", uri, err)
+		}
+		out[uri] = signedAccessFromCapability(cap)
+	}
+	if len(out) == 0 {
+		return nil, nil
+	}
+	return out, nil
+}
+
+// signedAccessFromCapability maps a broker capability onto the wire message
+// atelet consumes. A read capability populates the read fields, a write
+// capability the write fields; the rest stay empty.
+func signedAccessFromCapability(cap storagebroker.Capability) *ateletpb.SignedObjectAccess {
+	return &ateletpb.SignedObjectAccess{
+		PrefixUrl:      cap.PrefixURL,
+		ReadToken:      cap.ReadToken,
+		ReadObjectUrls: cap.ReadObjectURLs,
+		WriteMethod:    cap.WriteMethod,
+		WriteToken:     cap.WriteToken,
+		WriteHeaders:   cap.WriteHeaders,
+		PostUrl:        cap.PostURL,
+		PostFields:     cap.PostFields,
+	}
+}

cmd/ateapi/internal/controlapi/broker_wiring_test.go

@@ -0,0 +1,139 @@
+// 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"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/storagebroker"
+)
+
+// fakeBroker records the URIs it is asked to sign and returns a capability whose
+// tokens embed the URI, so a test can tell entries apart.
+type fakeBroker struct {
+	err         error
+	mintedRead  []string
+	mintedWrite []string
+}
+
+func (f *fakeBroker) MintRead(_ context.Context, uri string, _ time.Duration) (storagebroker.Capability, error) {
+	if f.err != nil {
+		return storagebroker.Capability{}, f.err
+	}
+	f.mintedRead = append(f.mintedRead, uri)
+	return storagebroker.Capability{PrefixURL: "https://read/" + uri, ReadToken: "rt-" + uri}, nil
+}
+
+func (f *fakeBroker) MintWrite(_ context.Context, uri string, _ time.Duration) (storagebroker.Capability, error) {
+	if f.err != nil {
+		return storagebroker.Capability{}, f.err
+	}
+	f.mintedWrite = append(f.mintedWrite, uri)
+	return storagebroker.Capability{
+		PrefixURL:   "https://write/" + uri,
+		WriteMethod: storagebroker.WriteMethodPUT,
+		WriteToken:  "wt-" + uri,
+	}, nil
+}
+
+func TestReadAccessFor_MintsOnePerURI(t *testing.T) {
+	fb := &fakeBroker{}
+	w := &ActorWorkflow{broker: fb}
+
+	// The empty URI (an unset golden) must be skipped, not signed.
+	m, err := w.readAccessFor(context.Background(), "s3://bucket/actor", "s3://bucket/golden", "")
+	if err != nil {
+		t.Fatalf("readAccessFor: %v", err)
+	}
+	if len(m) != 2 {
+		t.Fatalf("want 2 capabilities, got %d: %v", len(m), m)
+	}
+	if got := m["s3://bucket/actor"].GetReadToken(); got != "rt-s3://bucket/actor" {
+		t.Errorf("actor read token = %q, want rt-s3://bucket/actor", got)
+	}
+	if got := m["s3://bucket/golden"].GetReadToken(); got != "rt-s3://bucket/golden" {
+		t.Errorf("golden read token = %q, want rt-s3://bucket/golden", got)
+	}
+	if _, ok := m[""]; ok {
+		t.Errorf("empty URI must not be signed")
+	}
+}
+
+func TestReadAccessFor_DedupsRepeatedURI(t *testing.T) {
+	fb := &fakeBroker{}
+	w := &ActorWorkflow{broker: fb}
+
+	m, err := w.readAccessFor(context.Background(), "s3://bucket/a", "s3://bucket/a")
+	if err != nil {
+		t.Fatalf("readAccessFor: %v", err)
+	}
+	if len(m) != 1 {
+		t.Fatalf("want 1 capability, got %d", len(m))
+	}
+	if len(fb.mintedRead) != 1 {
+		t.Errorf("broker minted %d times, want 1 (deduped)", len(fb.mintedRead))
+	}
+}
+
+func TestReadWriteAccessFor_NilBrokerReturnsNil(t *testing.T) {
+	w := &ActorWorkflow{} // no broker configured
+
+	if m, err := w.readAccessFor(context.Background(), "s3://bucket/a"); err != nil || m != nil {
+		t.Errorf("readAccessFor with no broker = (%v, %v), want (nil, nil)", m, err)
+	}
+	if m, err := w.writeAccessFor(context.Background(), "s3://bucket/a"); err != nil || m != nil {
+		t.Errorf("writeAccessFor with no broker = (%v, %v), want (nil, nil)", m, err)
+	}
+}
+
+func TestAccessFor_AllEmptyReturnsNil(t *testing.T) {
+	w := &ActorWorkflow{broker: &fakeBroker{}}
+
+	m, err := w.readAccessFor(context.Background(), "", "")
+	if err != nil {
+		t.Fatalf("readAccessFor: %v", err)
+	}
+	if m != nil {
+		t.Errorf("want nil map when nothing to sign, got %v", m)
+	}
+}
+
+func TestWriteAccessFor_MintsWriteCapability(t *testing.T) {
+	fb := &fakeBroker{}
+	w := &ActorWorkflow{broker: fb}
+
+	m, err := w.writeAccessFor(context.Background(), "s3://bucket/dest")
+	if err != nil {
+		t.Fatalf("writeAccessFor: %v", err)
+	}
+	sa := m["s3://bucket/dest"]
+	if sa.GetWriteMethod() != storagebroker.WriteMethodPUT || sa.GetWriteToken() != "wt-s3://bucket/dest" {
+		t.Errorf("write capability = %+v, want PUT with wt-s3://bucket/dest", sa)
+	}
+}
+
+func TestAccessFor_PropagatesMintError(t *testing.T) {
+	w := &ActorWorkflow{broker: &fakeBroker{err: errors.New("boom")}}
+
+	if _, err := w.readAccessFor(context.Background(), "s3://bucket/a"); err == nil {
+		t.Errorf("readAccessFor: want error, got nil")
+	}
+	if _, err := w.writeAccessFor(context.Background(), "s3://bucket/a"); err == nil {
+		t.Errorf("writeAccessFor: want error, got nil")
+	}
+}

cmd/ateapi/internal/controlapi/functionaltest/common_test.go

@@ -178,7 +178,7 @@ func setupTestWithVolumePlugins(t *testing.T, ns string, plugins map[string]volu
 			mockDriverName: mockPlugin,
 		}
 	}
-	service := controlapi.NewRPCService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins)
+	service := controlapi.NewRPCService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, csiDriverConfigLister, scLister, dialer, instruments, "", volPlugins, nil)
 
 	// 5. Start REAL gRPC Server for ATE API
 	grpcServer := grpc.NewServer(grpc.ChainUnaryInterceptor(

cmd/ateapi/internal/controlapi/service.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"sync"
 
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/storagebroker"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
 	"github.com/agent-substrate/substrate/internal/resources"
@@ -70,6 +71,7 @@ func NewRPCService(
 	instruments *Instruments,
 	egressGatewayAddress string,
 	volumePlugins map[string]volume.VolumePluginControlPlane,
+	broker storagebroker.Broker,
 ) *RPCService {
 	impl := newServiceImpl(persistence, actorTemplateLister, storageClassLister)
 	s := &RPCService{
@@ -81,7 +83,7 @@ func NewRPCService(
 		instruments:           instruments,
 		volumePlugins:         volumePlugins,
 	}
-	s.actorWorkflow = NewActorWorkflow(impl, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s)
+	s.actorWorkflow = NewActorWorkflow(impl, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, storageClassLister, instruments, egressGatewayAddress, s, broker)
 	s.workerWorkflow = NewWorkerWorkflow(impl)
 	return s
 }

cmd/ateapi/internal/controlapi/workflow.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/scheduling"
+	"github.com/agent-substrate/substrate/cmd/ateapi/internal/storagebroker"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
 	"github.com/agent-substrate/substrate/internal/resources"
@@ -79,6 +80,9 @@ type ActorWorkflow struct {
 	instruments          *Instruments
 	egressGatewayAddress string
 	pluginRegistry       VolumePluginRegistry
+	// broker, when set, mints short-lived per-snapshot signed access so atelet
+	// reads and writes snapshots over plain HTTP with no cloud credential.
+	broker storagebroker.Broker
 }
 
 // NewActorWorkflow creates a new ActorWorkflow. instruments may be nil.
@@ -93,6 +97,7 @@ func NewActorWorkflow(
 	instruments *Instruments,
 	egressGatewayAddress string,
 	pluginRegistry VolumePluginRegistry,
+	broker storagebroker.Broker,
 ) *ActorWorkflow {
 	return &ActorWorkflow{
 		store:                store,
@@ -106,6 +111,7 @@ func NewActorWorkflow(
 		instruments:          instruments,
 		egressGatewayAddress: egressGatewayAddress,
 		pluginRegistry:       pluginRegistry,
+		broker:               broker,
 	}
 }
 

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -708,6 +708,11 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou
 		if !src.GoldenSnapshotURI.IsZero() {
 			req.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN
 			req.GoldenSnapshotUri = src.GoldenSnapshotURI.String()
+			// atelet combines the local pause snapshot with the golden snapshot
+			// it reads from object storage; mint a read capability for it.
+			if req.SignedAccess, err = w.readAccessFor(ctx, src.GoldenSnapshotURI.String()); err != nil {
+				return tele, maybeCrashActor(ctx, w.store, actorRef, err, "while minting golden snapshot read access", ateattr.OperationResume)
+			}
 		}
 		tele.WireSnapshotScope = ateattr.SnapshotScopeValue(req.Scope)
 
@@ -751,6 +756,11 @@ func (w *ActorWorkflow) ensureAteletRestored(ctx context.Context, actorRef resou
 			CpuMilli:          cpuMilli,
 			MemoryBytes:       memBytes,
 		}
+		// The node reads the actor's snapshot and, on a golden-data restore, the
+		// golden snapshot too; mint a read capability scoped to each.
+		if req.SignedAccess, err = w.readAccessFor(ctx, src.SnapshotURI.String(), src.GoldenSnapshotURI.String()); err != nil {
+			return tele, maybeCrashActor(ctx, w.store, actorRef, err, "while minting snapshot read access", ateattr.OperationResume)
+		}
 		_, err = client.Restore(ctx, req)
 		return tele, maybeCrashActor(ctx, w.store, actorRef, err, "while restoring durable snapshot", ateattr.OperationResume)
 	} else {

cmd/ateapi/internal/controlapi/workflow_suspend.go

@@ -259,6 +259,10 @@ func (w *ActorWorkflow) ensureAteletSuspended(ctx context.Context, actorRef reso
 		Scope:    actorSnapshotContentScopeToAtelet(commitSnapshotScope(actor.GetMetadata().GetAtespace(), actorTemplate)),
 		ActorUid: actor.GetMetadata().Uid,
 	}
+	// The node writes the actor's snapshot; mint a write capability for it.
+	if req.SignedAccess, err = w.writeAccessFor(ctx, snapshotURI.String()); err != nil {
+		return "", err
+	}
 	wireSnapshotScope = ateattr.SnapshotScopeValue(req.Scope)
 
 	_, err = client.Checkpoint(ctx, req)
@@ -311,6 +315,11 @@ func (w *ActorWorkflow) ensurePausedSnapshotUploaded(ctx context.Context, actorR
 		// from the captured scope in the snapshot's manifest where possible.
 		DesiredScope: actorSnapshotContentScopeToAtelet(commitSnapshotScope(actor.GetMetadata().GetAtespace(), actorTemplate)),
 	}
+	// The node writes the paused snapshot to its destination; mint a write
+	// capability for it.
+	if req.SignedAccess, err = w.writeAccessFor(ctx, snapshotURI.String()); err != nil {
+		return "", err
+	}
 	wireSnapshotScope = ateattr.SnapshotScopeValue(req.DesiredScope)
 
 	_, err = client.UploadPausedCheckpoint(ctx, req)

cmd/ateapi/internal/controlapi/workflow_suspend_test.go

@@ -703,7 +703,7 @@ func TestSuspendActor_PausedWithoutLocalSnapshotCrashes(t *testing.T) {
 	}); err != nil {
 		t.Fatalf("add template to indexer: %v", err)
 	}
-	w := NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil)
+	w := NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil, nil)
 
 	seedWorkflowActor(t, ctx, st, resources.ActorRef{Atespace: "team-a", Name: "id1"}, "ns", "tmpl1", ateapipb.ActorState_ACTOR_STATE_PAUSED)
 

cmd/ateapi/internal/controlapi/workflow_testutil_test.go

@@ -50,7 +50,7 @@ func newTestActorWorkflow(t *testing.T, st store.Interface, tmplNamespace, tmplN
 	}); err != nil {
 		t.Fatalf("add template to indexer: %v", err)
 	}
-	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil)
+	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil, "", nil, nil)
 }
 
 // seedWorkflowActor stores an actor with the given state, bound to the given