Add lazy Namespace storage client

go.mod

@@ -3,6 +3,7 @@ module github.com/buildkite/agent/v4
 go 1.26.5
 
 require (
+	buf.build/gen/go/namespace/cloud/protocolbuffers/go v1.36.12-20260820164744-a1973bcf4d87.1
 	cloud.google.com/go/compute/metadata v0.9.0
 	cloud.google.com/go/kms v1.33.0
 	connectrpc.com/connect v1.20.0
@@ -70,11 +71,14 @@ require (
 	golang.org/x/sys v0.47.0
 	golang.org/x/term v0.45.0
 	google.golang.org/api v0.292.0
+	google.golang.org/grpc v1.83.0
 	google.golang.org/protobuf v1.36.12
 	gopkg.in/yaml.v3 v3.0.1
+	namespacelabs.dev/integrations v0.0.11-0.20260508113815-6a8135624a35 // Pinned for CreateNewVersionIfExists support.
 )
 
 require (
+	buf.build/gen/go/namespace/cloud/grpc/go v1.6.2-20260820164744-a1973bcf4d87.1 // indirect
 	cloud.google.com/go v0.123.0 // indirect
 	cloud.google.com/go/auth v0.22.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
@@ -114,6 +118,7 @@ require (
 	github.com/go-logr/logr v1.4.4 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/goccy/go-json v0.10.6 // indirect
+	github.com/golang-jwt/jwt/v4 v4.5.2 // indirect
 	github.com/golang-jwt/jwt/v5 v5.3.1 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
@@ -157,7 +162,6 @@ require (
 	google.golang.org/genproto v0.0.0-20260406210006-6f92a3bedf2d // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect
-	google.golang.org/grpc v1.83.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gotest.tools/gotestsum v1.13.0 // indirect
 	mvdan.cc/gofumpt v0.9.2 // indirect

go.sum

@@ -1,3 +1,7 @@
+buf.build/gen/go/namespace/cloud/grpc/go v1.6.2-20260820164744-a1973bcf4d87.1 h1:Szd8+tmae26rv8IG14gItvykV9vhFzvrBcUg47ZAVHM=
+buf.build/gen/go/namespace/cloud/grpc/go v1.6.2-20260820164744-a1973bcf4d87.1/go.mod h1:v2Fn3cyk8eT581ReENee/l/lFl1JNYMU/WspzgSjXiU=
+buf.build/gen/go/namespace/cloud/protocolbuffers/go v1.36.12-20260820164744-a1973bcf4d87.1 h1:rha4+KA9AGYPQktS1LYeHjpXDafnpoPLziO0FGhd1iw=
+buf.build/gen/go/namespace/cloud/protocolbuffers/go v1.36.12-20260820164744-a1973bcf4d87.1/go.mod h1:qZ8DAMLNpt2zvGyC5UsmblfkkKyUWbVHEkLodBFoMKg=
 cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE=
 cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU=
 cloud.google.com/go/auth v0.22.0 h1:Xp9wAKkLoeaYb5pYZZoQGz4E9sdPxIbzS3gywZE3ciQ=
@@ -168,6 +172,8 @@ github.com/goccy/go-json v0.10.6 h1:p8HrPJzOakx/mn/bQtjgNjdTcN+/S6FcG2CTtQOrHVU=
 github.com/goccy/go-json v0.10.6/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
 github.com/gofrs/flock v0.13.0 h1:95JolYOvGMqeH31+FC7D2+uULf6mG61mEZ/A8dRYMzw=
 github.com/gofrs/flock v0.13.0/go.mod h1:jxeyy9R1auM5S6JYDBhDt+E2TCo7DkratH4Pgi8P+Z0=
+github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI=
+github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0=
 github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY=
 github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
@@ -188,8 +194,8 @@ github.com/googleapis/enterprise-certificate-proxy v0.3.19 h1:mMOE7DN2+p76/EdIrm
 github.com/googleapis/enterprise-certificate-proxy v0.3.19/go.mod h1:rSEsBUemEBZEexP2y6jPp16LUmUbjmSbcPMQizR0o4k=
 github.com/googleapis/gax-go/v2 v2.23.0 h1:Tchl7qkvE7Ip3y+ztvNufYFvkfqTe7NfLTYGIdJRLuE=
 github.com/googleapis/gax-go/v2 v2.23.0/go.mod h1:rBQKOVJCdb8IFEzg+FCwlt1LP/xMDGuqUXhUG+XMXEg=
-github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
-github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
+github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
 github.com/gowebpki/jcs v1.0.1 h1:Qjzg8EOkrOTuWP7DqQ1FbYtcpEbeTzUoTN9bptp8FOU=
 github.com/gowebpki/jcs v1.0.1/go.mod h1:CID1cNZ+sHp1CCpAR8mPf6QRtagFBgPJE0FCUQ6+BrI=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk=
@@ -389,3 +395,5 @@ gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q=
 gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA=
 mvdan.cc/gofumpt v0.9.2 h1:zsEMWL8SVKGHNztrx6uZrXdp7AX8r421Vvp23sz7ik4=
 mvdan.cc/gofumpt v0.9.2/go.mod h1:iB7Hn+ai8lPvofHd9ZFGVg2GOr8sBUw1QUWjNbmIL/s=
+namespacelabs.dev/integrations v0.0.11-0.20260508113815-6a8135624a35 h1:khQ002qSHfL0QocNYvjbEJHCHZCNrlNbfZJA/hPCqPA=
+namespacelabs.dev/integrations v0.0.11-0.20260508113815-6a8135624a35/go.mod h1:E8wND7kuqalgSfiaGuxkfYd6xRq918FGX8vpnUmwOu8=

internal/cache/cache.go

@@ -34,6 +34,19 @@ type cacheOps interface {
 	ListCaches() []configuration.Cache
 }
 
+type cacheClientCloser interface {
+	close() error
+}
+
+func withClientCleanup(l logger.Logger, c cacheClientCloser, run func() error) error {
+	defer func() {
+		if closeErr := c.close(); closeErr != nil {
+			l.Warnf("Failed to close Namespace storage client: %v", closeErr)
+		}
+	}()
+	return run()
+}
+
 // RunSave saves caches based on the provided configuration and logs results as
 // each cache is processed.
 func RunSave(ctx context.Context, l logger.Logger, apiClient *api.Client, cfg Config) error {
@@ -45,7 +58,9 @@ func RunSave(ctx context.Context, l logger.Logger, apiClient *api.Client, cfg Co
 		l.Infof("No caches defined in the cache configuration file, nothing to save")
 		return nil
 	}
-	return saveWithClient(ctx, l, c, cacheIDs, cfg.Concurrency)
+	return withClientCleanup(l, c, func() error {
+		return saveWithClient(ctx, l, c, cacheIDs, cfg.Concurrency)
+	})
 }
 
 // RunRestore restores caches based on the provided configuration and logs results
@@ -59,7 +74,9 @@ func RunRestore(ctx context.Context, l logger.Logger, apiClient *api.Client, cfg
 		l.Infof("No caches defined in the cache configuration file, nothing to restore")
 		return nil
 	}
-	return restoreWithClient(ctx, l, c, cacheIDs, cfg.Concurrency)
+	return withClientCleanup(l, c, func() error {
+		return restoreWithClient(ctx, l, c, cacheIDs, cfg.Concurrency)
+	})
 }
 
 // ListCaches returns all cache definitions configured on the client.

internal/cache/cache_test.go

@@ -43,6 +43,63 @@ func (m *mockCacheClient) ListCaches() []configuration.Cache {
 	return nil
 }
 
+type fakeCacheClientCloser struct {
+	closeFunc func() error
+}
+
+func (c fakeCacheClientCloser) close() error {
+	return c.closeFunc()
+}
+
+func TestWithClientCleanup(t *testing.T) {
+	t.Run("closes after work completes", func(t *testing.T) {
+		workComplete := false
+		closed := false
+		closer := fakeCacheClientCloser{closeFunc: func() error {
+			if !workComplete {
+				t.Error("client closed before work completed")
+			}
+			closed = true
+			return nil
+		}}
+
+		err := withClientCleanup(logger.Discard, closer, func() error {
+			workComplete = true
+			return nil
+		})
+		if err != nil {
+			t.Fatalf("withClientCleanup error = %v, want nil", err)
+		}
+		if !closed {
+			t.Error("client was not closed")
+		}
+	})
+
+	t.Run("close failure is logged without masking work error", func(t *testing.T) {
+		workErr := errors.New("worker failed")
+		closeErr := errors.New("close failed")
+		log := logger.NewBuffer()
+		closer := fakeCacheClientCloser{closeFunc: func() error { return closeErr }}
+
+		err := withClientCleanup(log, closer, func() error { return workErr })
+		if !errors.Is(err, workErr) {
+			t.Fatalf("withClientCleanup error = %v, want %v", err, workErr)
+		}
+		if got := strings.Join(log.Messages, "\n"); !strings.Contains(got, closeErr.Error()) {
+			t.Errorf("log messages = %q, want close error", got)
+		}
+	})
+
+	t.Run("close failure does not turn success into failure", func(t *testing.T) {
+		closeErr := errors.New("close failed")
+		closer := fakeCacheClientCloser{closeFunc: func() error { return closeErr }}
+
+		if err := withClientCleanup(logger.Discard, closer, func() error { return nil }); err != nil {
+			t.Fatalf("withClientCleanup error = %v, want nil", err)
+		}
+	})
+}
+
 // Test helpers
 
 func createTempCacheConfig(t *testing.T, content string) string {
@@ -396,6 +453,12 @@ func TestNewClient_ValidCacheIDs(t *testing.T) {
 	if got := client; got == nil {
 		t.Fatalf("newClient(logger.Discard, nil, cfg) = %v, want non-nil value", got)
 	}
+	if client.nscClient == nil {
+		t.Fatal("newClient did not create a Namespace client holder")
+	}
+	if err := client.close(); err != nil {
+		t.Fatalf("close unused client: %v", err)
+	}
 	if diff := cmp.Diff(cacheIDs, []string{"cache1", "cache2"}); diff != "" {
 		t.Fatalf("newClient(logger.Discard, nil, cfg) diff (-got +want):\n%s", diff)
 	}

internal/cache/client.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/buildkite/agent/v4/api"
 	"github.com/buildkite/agent/v4/internal/cache/configuration"
+	"github.com/buildkite/agent/v4/internal/cache/store"
 	"github.com/buildkite/agent/v4/logger"
 )
 
@@ -47,16 +48,17 @@ var (
 
 // client is a configured handle for cache Save and Restore operations.
 //
-// It is not a network connection; it just bundles the API client, storage
-// bucket and the expanded, validated cache definitions used by every call.
-// Safe for concurrent use; honours context cancellation.
+// It bundles the API client, storage bucket, expanded cache definitions, and a
+// lazily initialized Namespace connection. Safe for concurrent use; honours
+// context cancellation.
 type client struct {
 	api        cacheAPI
 	bucketURL  string
 	format     string
 	platform   string
 	registry   string
 	caches     []configuration.Cache
+	nscClient  *store.NscClient
 	onProgress ProgressCallback
 }
 
@@ -100,6 +102,7 @@ func newClient(l logger.Logger, apiClient cacheAPI, cfg Config) (*client, []stri
 		platform:  fmt.Sprintf("%s/%s", runtime.GOOS, runtime.GOARCH),
 		registry:  registry,
 		caches:    expanded,
+		nscClient: store.NewNscClient(),
 		onProgress: func(cacheID, stage, message string, _, _ int) {
 			l.WithFields(
 				logger.StringField("cache_id", cacheID),
@@ -116,6 +119,10 @@ func newClient(l logger.Logger, apiClient cacheAPI, cfg Config) (*client, []stri
 	return c, names, nil
 }
 
+func (c *client) close() error {
+	return c.nscClient.Close()
+}
+
 // resolveCacheNames returns requested if non-empty (after validating every name
 // exists), otherwise returns every cache name configured on the client.
 func (c *client) resolveCacheNames(requested []string) ([]string, error) {

internal/cache/restore.go

@@ -407,7 +407,7 @@ func (c *client) downloadCache(ctx context.Context, retrieveResp api.CacheEntryR
 	)
 
 	// Create blob store
-	blobStore, err := store.NewBlobStore(ctx, retrieveResp.Store, bucketURL)
+	blobStore, err := store.NewBlobStore(ctx, retrieveResp.Store, bucketURL, c.nscClient)
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, "failed to create blob store")
@@ -422,7 +422,9 @@ func (c *client) downloadCache(ctx context.Context, retrieveResp api.CacheEntryR
 		return "", "", nil, fmt.Errorf("failed to create temp directory: %w", err)
 	}
 
-	archiveFile = filepath.Join(tmpDir, storeObjectName)
+	// The object name comes from the cache API. Keep it as the remote lookup key,
+	// but never use it as a local path component.
+	archiveFile = filepath.Join(tmpDir, "archive")
 
 	// Download archive
 	transferInfo, err = blobStore.Download(ctx, storeObjectName, archiveFile)

internal/cache/save.go

@@ -277,7 +277,7 @@ func (c *client) Save(ctx context.Context, cacheID string) (SaveResult, error) {
 	c.callProgress(cacheID, "uploading", "Uploading cache archive", 0, int(archiveInfo.Size))
 
 	// Upload archive
-	blobStore, err := store.NewBlobStore(ctx, registryResp.Store, c.bucketURL)
+	blobStore, err := store.NewBlobStore(ctx, registryResp.Store, c.bucketURL, c.nscClient)
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, "failed to create blob store")

internal/cache/store/blob.go

@@ -20,13 +20,13 @@ type Blob interface {
 	Download(ctx context.Context, key, destPath string) (*TransferInfo, error)
 }
 
-func NewBlobStore(ctx context.Context, store, bucketURL string) (Blob, error) {
+func NewBlobStore(ctx context.Context, store, bucketURL string, nscClient *NscClient) (Blob, error) {
 	switch store {
 	case AgentManaged:
 		scheme, _, _ := strings.Cut(bucketURL, "://")
 		switch scheme {
 		case nscScheme:
-			return NewNscStore(bucketURL)
+			return NewNscStore(bucketURL, nscClient)
 		case "file":
 			// Supported only for local testing, kept consistent with validateCacheStore.
 			return NewLocalFileBlob(ctx, bucketURL)

internal/cache/store/file.go

@@ -53,6 +53,27 @@ type FileMetadata struct {
 	Version   int    `json:"version"`          // Metadata schema version
 }
 
+func validateFilePath(filePath string) error {
+	if filePath == "" {
+		return fmt.Errorf("file path cannot be empty")
+	}
+
+	cleanPath := filepath.Clean(filePath)
+	dangerousChars := []string{";", "&", "|", "`", "$", "(", ")", "{", "}", "[", "]", "<", ">", "\"", "'"}
+	if runtime.GOOS != "windows" {
+		dangerousChars = append(dangerousChars, "\\")
+	}
+	for _, char := range dangerousChars {
+		if strings.Contains(cleanPath, char) {
+			return fmt.Errorf("file path contains potentially dangerous character: %s", char)
+		}
+	}
+	if strings.Contains(cleanPath, "..") {
+		return fmt.Errorf("file path contains path traversal sequence")
+	}
+	return nil
+}
+
 // NewLocalFileBlob creates a new local file storage backend from a file:// URL.
 //
 // Supported URL formats:

internal/cache/store/file_test.go

@@ -10,6 +10,41 @@ import (
 	"testing"
 )
 
+func TestValidateFilePath(t *testing.T) {
+	tests := []struct {
+		name      string
+		filePath  string
+		wantError string
+	}{
+		{name: "valid simple path", filePath: "test.txt"},
+		{name: "valid relative path", filePath: "dir/subdir/file.txt"},
+		{name: "valid absolute path", filePath: "/tmp/test.txt"},
+		{name: "empty path", wantError: "file path cannot be empty"},
+		{name: "path with semicolon", filePath: "file;rm -rf /", wantError: "file path contains potentially dangerous character: ;"},
+		{name: "path with ampersand", filePath: "file&malicious", wantError: "file path contains potentially dangerous character: &"},
+		{name: "path with pipe", filePath: "file|cat /etc/passwd", wantError: "file path contains potentially dangerous character: |"},
+		{name: "path with backtick", filePath: "file`whoami`", wantError: "file path contains potentially dangerous character: `"},
+		{name: "path with dollar sign", filePath: "file$(whoami)", wantError: "file path contains potentially dangerous character: $"},
+		{name: "path traversal attempt", filePath: "../../../etc/passwd", wantError: "file path contains path traversal sequence"},
+		{name: "path with quotes", filePath: `file"test"`, wantError: `file path contains potentially dangerous character: "`},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateFilePath(tt.filePath)
+			if tt.wantError == "" {
+				if err != nil {
+					t.Fatalf("validateFilePath: %v", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tt.wantError) {
+				t.Fatalf("validateFilePath error = %v, want error containing %q", err, tt.wantError)
+			}
+		})
+	}
+}
+
 func TestNewLocalFileBlob(t *testing.T) {
 	ctx := t.Context()
 
@@ -560,7 +595,7 @@ func TestNewBlobStoreLocalFile(t *testing.T) {
 	ctx := t.Context()
 	tmpDir := t.TempDir()
 
-	blob, err := NewBlobStore(ctx, LocalFileStore, fileURL(tmpDir))
+	blob, err := NewBlobStore(ctx, LocalFileStore, fileURL(tmpDir), nil)
 	if err != nil {
 		t.Fatalf("NewBlobStore: %v", err)
 	}
@@ -576,7 +611,7 @@ func TestNewBlobStoreLocalFile(t *testing.T) {
 // file:// is accepted for agent_managed (local testing) and must reach the local
 // file store rather than the S3 store, matching validateCacheStore.
 func TestNewBlobStoreAgentManagedFile(t *testing.T) {
-	blob, err := NewBlobStore(t.Context(), AgentManaged, fileURL(t.TempDir()))
+	blob, err := NewBlobStore(t.Context(), AgentManaged, fileURL(t.TempDir()), nil)
 	if err != nil {
 		t.Fatalf("NewBlobStore: %v", err)
 	}