From c641ca61ca7a4d81219e739e92dd2263fe4d4491 Mon Sep 17 00:00:00 2001 From: Eric Curtin Date: Fri, 23 Oct 2026 21:40:26 +0100 Subject: [PATCH 1/5] csi: reject a volume ID that escapes the staging directory NodeStage and NodeUnstage joined the raw ID onto the staging prefix, so an ID like ../x made atelet create and remove directories outside it. Signed-off-by: Eric Curtin --- internal/volume/csi/plugin.go | 24 ++++++++-- internal/volume/csi/plugin_test.go | 71 ++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 4 deletions(-) diff --git a/internal/volume/csi/plugin.go b/internal/volume/csi/plugin.go index c75f119bc4..088101a13b 100644 --- a/internal/volume/csi/plugin.go +++ b/internal/volume/csi/plugin.go @@ -286,7 +286,10 @@ func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest) return "", nil } - stagingPath := filepath.Join(p.stagingDirPrefix, req.VolumeID) + stagingPath, err := p.stagingDir(req.VolumeID) + if err != nil { + return "", err + } if err := os.MkdirAll(stagingPath, 0750); err != nil { return "", fmt.Errorf("failed to create staging directory %q: %w", stagingPath, err) } @@ -299,7 +302,7 @@ func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest) PublishContext: req.PublishContext, } - _, err := p.client.NodeStageVolume(ctx, csiReq) + _, err = p.client.NodeStageVolume(ctx, csiReq) if status.Code(err) == codes.Unimplemented { slog.WarnContext(ctx, "CSI NodeStageVolume is unimplemented by driver; skipping staging", slog.String("volume_id", req.VolumeID)) p.disableNodeCap(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) @@ -334,13 +337,16 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error { return nil } - stagingPath := filepath.Join(p.stagingDirPrefix, volumeID) + stagingPath, err := p.stagingDir(volumeID) + if err != nil { + return err + } req := &csi.NodeUnstageVolumeRequest{ VolumeId: volumeID, StagingTargetPath: stagingPath, } - _, err := p.client.NodeUnstageVolume(ctx, req) + _, err = p.client.NodeUnstageVolume(ctx, req) if status.Code(err) == codes.Unimplemented { slog.WarnContext(ctx, "CSI NodeUnstageVolume is unimplemented by driver; skipping unstaging", slog.String("volume_id", volumeID)) p.disableNodeCap(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) @@ -352,6 +358,16 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error { return nil } +// stagingDir returns the staging directory for volumeID. The ID is the +// driver's own string and may contain slashes, but it must name a directory +// below stagingDirPrefix. +func (p *Plugin) stagingDir(volumeID string) (string, error) { + if !filepath.IsLocal(volumeID) || filepath.Clean(volumeID) == "." { + return "", fmt.Errorf("volume ID %q does not name a directory below the staging directory", volumeID) + } + return filepath.Join(p.stagingDirPrefix, volumeID), nil +} + // removeStagingDir removes an empty staging directory, logging any failure. func removeStagingDir(ctx context.Context, stagingPath string) { if err := os.Remove(stagingPath); err != nil && !os.IsNotExist(err) { diff --git a/internal/volume/csi/plugin_test.go b/internal/volume/csi/plugin_test.go index c47bb4dabd..c881c2780a 100644 --- a/internal/volume/csi/plugin_test.go +++ b/internal/volume/csi/plugin_test.go @@ -418,6 +418,77 @@ func TestPlugin_MountVolume(t *testing.T) { } } +func TestPlugin_StagingDirStaysBelowPrefix(t *testing.T) { + tests := []struct { + name string + volumeID string + wantErr bool + }{ + {name: "plain", volumeID: "vol-1"}, + {name: "slashes", volumeID: "projects/p/disks/d-1"}, + {name: "parent", volumeID: "../x", wantErr: true}, + {name: "parent after a name", volumeID: "a/../../x", wantErr: true}, + {name: "absolute", volumeID: "/x", wantErr: true}, + {name: "prefix itself", volumeID: ".", wantErr: true}, + {name: "empty", volumeID: "", wantErr: true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var stageCalled, unstageCalled int + driver := &mockCSIDriver{ + nodeStageVolumeFunc: func(context.Context, *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) { + stageCalled++ + return &csi.NodeStageVolumeResponse{}, nil + }, + nodeUnstageVolumeFunc: func(context.Context, *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) { + unstageCalled++ + return &csi.NodeUnstageVolumeResponse{}, nil + }, + } + endpoint, cleanup := startMockCSIDriver(t, driver) + defer cleanup() + client, err := NewCSIClient(endpoint, nil) + if err != nil { + t.Fatalf("failed to create CSI client: %v", err) + } + defer client.Close() + + root := t.TempDir() + plugin := NewPlugin(client) + plugin.stagingDirPrefix = filepath.Join(root, "staging") + ctx := context.Background() + plugin.InitNodeCapabilities(ctx) + + err = plugin.MountVolume(ctx, volume.MountVolumeRequest{VolumeID: tc.volumeID, TargetPath: filepath.Join(root, "target")}) + if gotErr := err != nil; gotErr != tc.wantErr { + t.Fatalf("MountVolume() error = %v, wantErr %v", err, tc.wantErr) + } + err = plugin.UnmountVolume(ctx, tc.volumeID, filepath.Join(root, "target")) + if gotErr := err != nil; gotErr != tc.wantErr { + t.Fatalf("UnmountVolume() error = %v, wantErr %v", err, tc.wantErr) + } + + wantCalls := 1 + if tc.wantErr { + wantCalls = 0 + } + if stageCalled != wantCalls || unstageCalled != wantCalls { + t.Errorf("NodeStageVolume calls = %d, NodeUnstageVolume calls = %d, want %d each", stageCalled, unstageCalled, wantCalls) + } + // Everything the plugin creates must be under the staging prefix. + entries, err := os.ReadDir(root) + if err != nil { + t.Fatal(err) + } + for _, e := range entries { + if e.Name() != "staging" { + t.Errorf("%q was created outside the staging directory", filepath.Join(root, e.Name())) + } + } + }) + } +} + func TestPlugin_UnmountVolume(t *testing.T) { driver := &mockCSIDriver{} endpoint, cleanup := startMockCSIDriver(t, driver) From 7322fad9099cb9faca823beb547351655a668fee Mon Sep 17 00:00:00 2001 From: Eric Curtin Date: Fri, 23 Oct 2026 23:21:19 +0100 Subject: [PATCH 2/5] csi: unstaging a rejected volume ID is a no-op, so Terminate still works Signed-off-by: Eric Curtin --- internal/volume/csi/plugin.go | 5 ++++- internal/volume/csi/plugin_test.go | 7 ++++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/internal/volume/csi/plugin.go b/internal/volume/csi/plugin.go index 088101a13b..8a2577230d 100644 --- a/internal/volume/csi/plugin.go +++ b/internal/volume/csi/plugin.go @@ -339,7 +339,10 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error { stagingPath, err := p.stagingDir(volumeID) if err != nil { - return err + // stageVolume rejects this ID too, so nothing was staged. Failing here + // would leave the actor that carries it unable to terminate. + slog.WarnContext(ctx, "Skipping unstage of a volume ID that cannot have been staged", slog.String("volume_id", volumeID), slog.Any("error", err)) + return nil } req := &csi.NodeUnstageVolumeRequest{ VolumeId: volumeID, diff --git a/internal/volume/csi/plugin_test.go b/internal/volume/csi/plugin_test.go index c881c2780a..1986321c46 100644 --- a/internal/volume/csi/plugin_test.go +++ b/internal/volume/csi/plugin_test.go @@ -463,9 +463,10 @@ func TestPlugin_StagingDirStaysBelowPrefix(t *testing.T) { if gotErr := err != nil; gotErr != tc.wantErr { t.Fatalf("MountVolume() error = %v, wantErr %v", err, tc.wantErr) } - err = plugin.UnmountVolume(ctx, tc.volumeID, filepath.Join(root, "target")) - if gotErr := err != nil; gotErr != tc.wantErr { - t.Fatalf("UnmountVolume() error = %v, wantErr %v", err, tc.wantErr) + // Nothing was staged for a rejected ID, so unmounting it must still + // succeed or the actor could never be terminated. + if err := plugin.UnmountVolume(ctx, tc.volumeID, filepath.Join(root, "target")); err != nil { + t.Fatalf("UnmountVolume() error = %v", err) } wantCalls := 1 From 0c81411e09f82eced055832cc8f8aeb873535810 Mon Sep 17 00:00:00 2001 From: Eric Curtin Date: Sat, 24 Oct 2026 09:47:26 +0100 Subject: [PATCH 3/5] csi: note an older build may have staged a rejected ID Signed-off-by: Eric Curtin --- internal/volume/csi/plugin.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/internal/volume/csi/plugin.go b/internal/volume/csi/plugin.go index 8a2577230d..cfcd83526d 100644 --- a/internal/volume/csi/plugin.go +++ b/internal/volume/csi/plugin.go @@ -339,8 +339,9 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error { stagingPath, err := p.stagingDir(volumeID) if err != nil { - // stageVolume rejects this ID too, so nothing was staged. Failing here - // would leave the actor that carries it unable to terminate. + // stageVolume rejects this ID too, so this build staged nothing. An + // older build may have, and that directory is left behind. Failing + // here would leave the actor that carries it unable to terminate. slog.WarnContext(ctx, "Skipping unstage of a volume ID that cannot have been staged", slog.String("volume_id", volumeID), slog.Any("error", err)) return nil } From 1c6737e82f694f546809e876da42b5e5ee2857e8 Mon Sep 17 00:00:00 2001 From: Eric Curtin Date: Mon, 9 Nov 2026 21:48:36 +0000 Subject: [PATCH 4/5] csi: stage at a per-driver hash of the volume ID Signed-off-by: Eric Curtin --- internal/volume/csi/plugin.go | 49 +++++++--------- internal/volume/csi/plugin_test.go | 93 ++++++++++++++++-------------- 2 files changed, 71 insertions(+), 71 deletions(-) diff --git a/internal/volume/csi/plugin.go b/internal/volume/csi/plugin.go index cfcd83526d..c136ab5dfc 100644 --- a/internal/volume/csi/plugin.go +++ b/internal/volume/csi/plugin.go @@ -16,8 +16,10 @@ package csi import ( "context" + "crypto/sha256" "crypto/tls" "crypto/x509" + "encoding/hex" "fmt" "log/slog" "os" @@ -53,6 +55,7 @@ var defaultTLSPaths = tlsPaths{ // Plugin implements volume.VolumePluginWorkerPlane using the CSI Client. type Plugin struct { client *Client + driverName string stagingDirPrefix string // capsMu guards controllerCaps and nodeCaps. The maps are built by @@ -69,9 +72,10 @@ var _ volume.VolumePluginWorkerPlane = (*Plugin)(nil) // NewPlugin creates a new Plugin adapter. Call InitControllerCapabilities or // InitNodeCapabilities before use; until then, no optional RPC is reported as supported. -func NewPlugin(client *Client) *Plugin { +func NewPlugin(client *Client, driverName string) *Plugin { return &Plugin{ client: client, + driverName: driverName, stagingDirPrefix: filepath.Join(nodepath.BasePath, "staging"), } } @@ -286,10 +290,7 @@ func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest) return "", nil } - stagingPath, err := p.stagingDir(req.VolumeID) - if err != nil { - return "", err - } + stagingPath := p.stagingDir(req.VolumeID) if err := os.MkdirAll(stagingPath, 0750); err != nil { return "", fmt.Errorf("failed to create staging directory %q: %w", stagingPath, err) } @@ -302,7 +303,7 @@ func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest) PublishContext: req.PublishContext, } - _, err = p.client.NodeStageVolume(ctx, csiReq) + _, err := p.client.NodeStageVolume(ctx, csiReq) if status.Code(err) == codes.Unimplemented { slog.WarnContext(ctx, "CSI NodeStageVolume is unimplemented by driver; skipping staging", slog.String("volume_id", req.VolumeID)) p.disableNodeCap(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) @@ -337,20 +338,13 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error { return nil } - stagingPath, err := p.stagingDir(volumeID) - if err != nil { - // stageVolume rejects this ID too, so this build staged nothing. An - // older build may have, and that directory is left behind. Failing - // here would leave the actor that carries it unable to terminate. - slog.WarnContext(ctx, "Skipping unstage of a volume ID that cannot have been staged", slog.String("volume_id", volumeID), slog.Any("error", err)) - return nil - } + stagingPath := p.stagingDir(volumeID) req := &csi.NodeUnstageVolumeRequest{ VolumeId: volumeID, StagingTargetPath: stagingPath, } - _, err = p.client.NodeUnstageVolume(ctx, req) + _, err := p.client.NodeUnstageVolume(ctx, req) if status.Code(err) == codes.Unimplemented { slog.WarnContext(ctx, "CSI NodeUnstageVolume is unimplemented by driver; skipping unstaging", slog.String("volume_id", volumeID)) p.disableNodeCap(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) @@ -362,20 +356,21 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error { return nil } -// stagingDir returns the staging directory for volumeID. The ID is the -// driver's own string and may contain slashes, but it must name a directory -// below stagingDirPrefix. -func (p *Plugin) stagingDir(volumeID string) (string, error) { - if !filepath.IsLocal(volumeID) || filepath.Clean(volumeID) == "." { - return "", fmt.Errorf("volume ID %q does not name a directory below the staging directory", volumeID) - } - return filepath.Join(p.stagingDirPrefix, volumeID), nil +// stagingDir returns the staging directory for volumeID. The directory is named +// by a hash of the ID, which is the driver's own string, and is scoped to the +// driver because IDs are only unique within one. +func (p *Plugin) stagingDir(volumeID string) string { + sum := sha256.Sum256([]byte(volumeID)) + return filepath.Join(p.stagingDirPrefix, p.driverName, hex.EncodeToString(sum[:]), "mount") } -// removeStagingDir removes an empty staging directory, logging any failure. +// removeStagingDir removes an empty staging directory and its per-volume +// parent, logging any failure. func removeStagingDir(ctx context.Context, stagingPath string) { - if err := os.Remove(stagingPath); err != nil && !os.IsNotExist(err) { - slog.WarnContext(ctx, "failed to remove staging directory", slog.String("path", stagingPath), slog.Any("error", err)) + for _, dir := range []string{stagingPath, filepath.Dir(stagingPath)} { + if err := os.Remove(dir); err != nil && !os.IsNotExist(err) { + slog.WarnContext(ctx, "failed to remove staging directory", slog.String("path", dir), slog.Any("error", err)) + } } } @@ -440,7 +435,7 @@ func newCSIPlugin(ctx context.Context, getter CSIDriverConfigGetter, driverName if err != nil { return nil, fmt.Errorf("failed to initialize CSI client from endpoint %q: %w", endpoint, err) } - csiPlugin := NewPlugin(csiClient) + csiPlugin := NewPlugin(csiClient, driverName) // Verify CSI plugin reported name matches requested name. reportedName, err := csiPlugin.DriverName(ctx) diff --git a/internal/volume/csi/plugin_test.go b/internal/volume/csi/plugin_test.go index 1986321c46..5eb9b586d0 100644 --- a/internal/volume/csi/plugin_test.go +++ b/internal/volume/csi/plugin_test.go @@ -20,6 +20,7 @@ import ( "net" "os" "path/filepath" + "strings" "testing" "github.com/container-storage-interface/spec/lib/go/csi" @@ -220,7 +221,7 @@ func TestPlugin_CreateVolume(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() resp, err := plugin.CreateVolume(ctx, volume.CreateVolumeRequest{ @@ -247,7 +248,7 @@ func TestPlugin_DeleteVolume(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() err = plugin.DeleteVolume(ctx, "test-vol") @@ -267,7 +268,7 @@ func TestPlugin_AttachVolume(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() plugin.InitControllerCapabilities(ctx) @@ -326,7 +327,7 @@ func TestPlugin_DetachVolume(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() plugin.InitControllerCapabilities(ctx) @@ -377,7 +378,7 @@ func TestPlugin_MountVolume(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) tmpDir, err := os.MkdirTemp("", "csi-mount-test-*") if err != nil { t.Fatalf("failed to create temp dir: %v", err) @@ -399,7 +400,7 @@ func TestPlugin_MountVolume(t *testing.T) { } // Verify staging directory was created - stagingPath := filepath.Join(plugin.stagingDirPrefix, "test-vol") + stagingPath := plugin.stagingDir("test-vol") if _, err := os.Stat(stagingPath); os.IsNotExist(err) { t.Errorf("staging directory %q was not created", stagingPath) } @@ -418,32 +419,26 @@ func TestPlugin_MountVolume(t *testing.T) { } } -func TestPlugin_StagingDirStaysBelowPrefix(t *testing.T) { +func TestPlugin_StagingDir(t *testing.T) { tests := []struct { name string volumeID string - wantErr bool }{ {name: "plain", volumeID: "vol-1"}, {name: "slashes", volumeID: "projects/p/disks/d-1"}, - {name: "parent", volumeID: "../x", wantErr: true}, - {name: "parent after a name", volumeID: "a/../../x", wantErr: true}, - {name: "absolute", volumeID: "/x", wantErr: true}, - {name: "prefix itself", volumeID: ".", wantErr: true}, - {name: "empty", volumeID: "", wantErr: true}, + {name: "parent", volumeID: "../x"}, + {name: "absolute", volumeID: "/x"}, + {name: "dot", volumeID: "."}, + {name: "longest", volumeID: strings.Repeat("a", 256)}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - var stageCalled, unstageCalled int + var stagedAt string driver := &mockCSIDriver{ - nodeStageVolumeFunc: func(context.Context, *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) { - stageCalled++ + nodeStageVolumeFunc: func(_ context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) { + stagedAt = req.GetStagingTargetPath() return &csi.NodeStageVolumeResponse{}, nil }, - nodeUnstageVolumeFunc: func(context.Context, *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) { - unstageCalled++ - return &csi.NodeUnstageVolumeResponse{}, nil - }, } endpoint, cleanup := startMockCSIDriver(t, driver) defer cleanup() @@ -454,29 +449,25 @@ func TestPlugin_StagingDirStaysBelowPrefix(t *testing.T) { defer client.Close() root := t.TempDir() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) plugin.stagingDirPrefix = filepath.Join(root, "staging") ctx := context.Background() plugin.InitNodeCapabilities(ctx) - err = plugin.MountVolume(ctx, volume.MountVolumeRequest{VolumeID: tc.volumeID, TargetPath: filepath.Join(root, "target")}) - if gotErr := err != nil; gotErr != tc.wantErr { - t.Fatalf("MountVolume() error = %v, wantErr %v", err, tc.wantErr) + req := volume.MountVolumeRequest{VolumeID: tc.volumeID, TargetPath: filepath.Join(root, "target")} + if err := plugin.MountVolume(ctx, req); err != nil { + t.Fatalf("MountVolume() error = %v", err) } - // Nothing was staged for a rejected ID, so unmounting it must still - // succeed or the actor could never be terminated. - if err := plugin.UnmountVolume(ctx, tc.volumeID, filepath.Join(root, "target")); err != nil { - t.Fatalf("UnmountVolume() error = %v", err) + // One directory per driver and volume, below the prefix. + wantDepth := 3 + rel, err := filepath.Rel(plugin.stagingDirPrefix, stagedAt) + if err != nil || !filepath.IsLocal(rel) || len(strings.Split(rel, string(filepath.Separator))) != wantDepth { + t.Errorf("staging path %q is not ///mount", stagedAt) } - - wantCalls := 1 - if tc.wantErr { - wantCalls = 0 - } - if stageCalled != wantCalls || unstageCalled != wantCalls { - t.Errorf("NodeStageVolume calls = %d, NodeUnstageVolume calls = %d, want %d each", stageCalled, unstageCalled, wantCalls) + if err := plugin.UnmountVolume(ctx, tc.volumeID, req.TargetPath); err != nil { + t.Fatalf("UnmountVolume() error = %v", err) } - // Everything the plugin creates must be under the staging prefix. + // Unmount leaves only the driver directory, and nothing outside the prefix. entries, err := os.ReadDir(root) if err != nil { t.Fatal(err) @@ -486,8 +477,22 @@ func TestPlugin_StagingDirStaysBelowPrefix(t *testing.T) { t.Errorf("%q was created outside the staging directory", filepath.Join(root, e.Name())) } } + if _, err := os.Stat(filepath.Dir(stagedAt)); !os.IsNotExist(err) { + t.Errorf("per-volume directory %q was not removed, err: %v", filepath.Dir(stagedAt), err) + } }) } + + t.Run("volume IDs map to distinct directories per driver", func(t *testing.T) { + a := &Plugin{stagingDirPrefix: "/s", driverName: "d1"} + b := &Plugin{stagingDirPrefix: "/s", driverName: "d2"} + if a.stagingDir("a/b") == a.stagingDir("a//b") || a.stagingDir("a") == a.stagingDir("a/b") { + t.Error("different volume IDs share a staging directory") + } + if a.stagingDir("v") == b.stagingDir("v") { + t.Error("different drivers share a staging directory") + } + }) } func TestPlugin_UnmountVolume(t *testing.T) { @@ -501,7 +506,7 @@ func TestPlugin_UnmountVolume(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) tmpDir, err := os.MkdirTemp("", "csi-unmount-test-*") if err != nil { t.Fatalf("failed to create temp dir: %v", err) @@ -512,7 +517,7 @@ func TestPlugin_UnmountVolume(t *testing.T) { targetPath := filepath.Join(tmpDir, "target") // Pre-create staging directory to test cleanup - stagingPath := filepath.Join(plugin.stagingDirPrefix, "test-vol") + stagingPath := plugin.stagingDir("test-vol") if err := os.MkdirAll(stagingPath, 0750); err != nil { t.Fatalf("failed to create staging path: %v", err) } @@ -616,7 +621,7 @@ func TestPlugin_Capabilities_SkipAttachDetachWhenUnsupported(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() plugin.InitControllerCapabilities(ctx) @@ -677,7 +682,7 @@ func TestPlugin_Capabilities_SkipStageUnstageWhenUnsupported(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() plugin.InitNodeCapabilities(ctx) @@ -743,7 +748,7 @@ func TestPlugin_Capabilities_AllEnabledWhenQueryFailsOrEmpty(t *testing.T) { } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() plugin.InitControllerCapabilities(ctx) plugin.InitNodeCapabilities(ctx) @@ -779,7 +784,7 @@ func TestPlugin_Capabilities_DisableAttachDetachAfterUnimplemented(t *testing.T) } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) ctx := context.Background() plugin.InitControllerCapabilities(ctx) @@ -830,7 +835,7 @@ func TestPlugin_Capabilities_DisableStageUnstageAfterUnimplemented(t *testing.T) } defer client.Close() - plugin := NewPlugin(client) + plugin := NewPlugin(client, mockDriverName) tmpDir := t.TempDir() plugin.stagingDirPrefix = filepath.Join(tmpDir, "staging") targetPath := filepath.Join(tmpDir, "target") @@ -848,7 +853,7 @@ func TestPlugin_Capabilities_DisableStageUnstageAfterUnimplemented(t *testing.T) if plugin.SupportsNodeCapability(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) { t.Errorf("expected STAGE_UNSTAGE_VOLUME to be false after Unimplemented") } - stagingPath := filepath.Join(plugin.stagingDirPrefix, "test-vol") + stagingPath := plugin.stagingDir("test-vol") if _, err := os.Stat(stagingPath); !os.IsNotExist(err) { t.Errorf("expected staging directory %q to be removed, got err: %v", stagingPath, err) } From 3eee9c44fa03b5b5e51a15c4f6d0734fd9dbd4bc Mon Sep 17 00:00:00 2001 From: Eric Curtin Date: Tue, 10 Nov 2026 09:27:24 +0000 Subject: [PATCH 5/5] csi: pin staging layout in tests, log staged path per volume Signed-off-by: Eric Curtin --- internal/volume/csi/plugin.go | 3 + internal/volume/csi/plugin_test.go | 89 ++++++++++++++++++++++++------ 2 files changed, 76 insertions(+), 16 deletions(-) diff --git a/internal/volume/csi/plugin.go b/internal/volume/csi/plugin.go index c136ab5dfc..3c0940b49d 100644 --- a/internal/volume/csi/plugin.go +++ b/internal/volume/csi/plugin.go @@ -295,6 +295,8 @@ func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest) return "", fmt.Errorf("failed to create staging directory %q: %w", stagingPath, err) } + // Staging dirs are named by a hash, so log which volume this one is for. + slog.InfoContext(ctx, "Staging CSI volume", slog.String("volume_id", req.VolumeID), slog.String("staging_path", stagingPath)) csiReq := &csi.NodeStageVolumeRequest{ VolumeId: req.VolumeID, StagingTargetPath: stagingPath, @@ -370,6 +372,7 @@ func removeStagingDir(ctx context.Context, stagingPath string) { for _, dir := range []string{stagingPath, filepath.Dir(stagingPath)} { if err := os.Remove(dir); err != nil && !os.IsNotExist(err) { slog.WarnContext(ctx, "failed to remove staging directory", slog.String("path", dir), slog.Any("error", err)) + return // The parent cannot be empty either. } } } diff --git a/internal/volume/csi/plugin_test.go b/internal/volume/csi/plugin_test.go index 5eb9b586d0..7ec708e0cd 100644 --- a/internal/volume/csi/plugin_test.go +++ b/internal/volume/csi/plugin_test.go @@ -16,6 +16,8 @@ package csi import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "net" "os" @@ -420,6 +422,13 @@ func TestPlugin_MountVolume(t *testing.T) { } func TestPlugin_StagingDir(t *testing.T) { + // The layout is on-disk state shared across atelet builds, so pin it. + p := &Plugin{stagingDirPrefix: "/staging", driverName: "d.example.com"} + want := "/staging/d.example.com/d2e8363faaac7ae76def3b14091d8eb5755f6b92e9531627aeec833a8731cc49/mount" + if got := p.stagingDir("vol-1"); got != want { + t.Errorf("stagingDir(%q) = %q, want %q", "vol-1", got, want) + } + tests := []struct { name string volumeID string @@ -433,12 +442,20 @@ func TestPlugin_StagingDir(t *testing.T) { } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - var stagedAt string + var stagedAt, publishedFrom, unstagedAt string driver := &mockCSIDriver{ nodeStageVolumeFunc: func(_ context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) { stagedAt = req.GetStagingTargetPath() return &csi.NodeStageVolumeResponse{}, nil }, + nodePublishVolumeFunc: func(_ context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) { + publishedFrom = req.GetStagingTargetPath() + return &csi.NodePublishVolumeResponse{}, nil + }, + nodeUnstageVolumeFunc: func(_ context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) { + unstagedAt = req.GetStagingTargetPath() + return &csi.NodeUnstageVolumeResponse{}, nil + }, } endpoint, cleanup := startMockCSIDriver(t, driver) defer cleanup() @@ -454,35 +471,70 @@ func TestPlugin_StagingDir(t *testing.T) { ctx := context.Background() plugin.InitNodeCapabilities(ctx) + sum := sha256.Sum256([]byte(tc.volumeID)) + driverDir := filepath.Join(plugin.stagingDirPrefix, mockDriverName) + wantPath := filepath.Join(driverDir, hex.EncodeToString(sum[:]), "mount") + req := volume.MountVolumeRequest{VolumeID: tc.volumeID, TargetPath: filepath.Join(root, "target")} if err := plugin.MountVolume(ctx, req); err != nil { t.Fatalf("MountVolume() error = %v", err) } - // One directory per driver and volume, below the prefix. - wantDepth := 3 - rel, err := filepath.Rel(plugin.stagingDirPrefix, stagedAt) - if err != nil || !filepath.IsLocal(rel) || len(strings.Split(rel, string(filepath.Separator))) != wantDepth { - t.Errorf("staging path %q is not ///mount", stagedAt) + if stagedAt != wantPath || publishedFrom != wantPath { + t.Errorf("staged at %q, published from %q, want both %q", stagedAt, publishedFrom, wantPath) } if err := plugin.UnmountVolume(ctx, tc.volumeID, req.TargetPath); err != nil { t.Fatalf("UnmountVolume() error = %v", err) } - // Unmount leaves only the driver directory, and nothing outside the prefix. + if unstagedAt != wantPath { + t.Errorf("unstaged at %q, want %q", unstagedAt, wantPath) + } + // Unmount removes the per-volume directory, keeps the driver directory, + // and creates nothing outside the prefix. + if _, err := os.Stat(filepath.Dir(wantPath)); !os.IsNotExist(err) { + t.Errorf("per-volume directory %q was not removed, err: %v", filepath.Dir(wantPath), err) + } + if _, err := os.Stat(driverDir); err != nil { + t.Errorf("driver directory %q was not kept: %v", driverDir, err) + } entries, err := os.ReadDir(root) if err != nil { t.Fatal(err) } - for _, e := range entries { - if e.Name() != "staging" { - t.Errorf("%q was created outside the staging directory", filepath.Join(root, e.Name())) - } - } - if _, err := os.Stat(filepath.Dir(stagedAt)); !os.IsNotExist(err) { - t.Errorf("per-volume directory %q was not removed, err: %v", filepath.Dir(stagedAt), err) + if len(entries) != 1 || entries[0].Name() != "staging" { + t.Errorf("unexpected entries under %q: %v", root, entries) } }) } + t.Run("non-empty mount directory is kept", func(t *testing.T) { + endpoint, cleanup := startMockCSIDriver(t, &mockCSIDriver{}) + defer cleanup() + client, err := NewCSIClient(endpoint, nil) + if err != nil { + t.Fatalf("failed to create CSI client: %v", err) + } + defer client.Close() + + plugin := NewPlugin(client, mockDriverName) + plugin.stagingDirPrefix = filepath.Join(t.TempDir(), "staging") + ctx := context.Background() + plugin.InitNodeCapabilities(ctx) + + mount := plugin.stagingDir("vol-1") + if err := os.MkdirAll(mount, 0750); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(mount, "data"), nil, 0600); err != nil { + t.Fatal(err) + } + if err := plugin.unstageVolume(ctx, "vol-1"); err != nil { + t.Fatalf("unstageVolume() error = %v", err) + } + if _, err := os.Stat(filepath.Join(mount, "data")); err != nil { + t.Errorf("non-empty mount directory was removed: %v", err) + } + }) + t.Run("volume IDs map to distinct directories per driver", func(t *testing.T) { a := &Plugin{stagingDirPrefix: "/s", driverName: "d1"} b := &Plugin{stagingDirPrefix: "/s", driverName: "d2"} @@ -854,8 +906,10 @@ func TestPlugin_Capabilities_DisableStageUnstageAfterUnimplemented(t *testing.T) t.Errorf("expected STAGE_UNSTAGE_VOLUME to be false after Unimplemented") } stagingPath := plugin.stagingDir("test-vol") - if _, err := os.Stat(stagingPath); !os.IsNotExist(err) { - t.Errorf("expected staging directory %q to be removed, got err: %v", stagingPath, err) + for _, dir := range []string{stagingPath, filepath.Dir(stagingPath)} { + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Errorf("expected staging directory %q to be removed, got err: %v", dir, err) + } } if err := plugin.UnmountVolume(ctx, "test-vol", targetPath); err != nil { @@ -888,6 +942,9 @@ func TestNewCSIPlugin_CapabilityQueryFailureIsNotFatal(t *testing.T) { } defer plugin.client.Close() + if plugin.driverName != mockDriverName { + t.Errorf("driverName = %q, want %q", plugin.driverName, mockDriverName) + } if isController && !plugin.SupportsControllerCapability(csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME) { t.Errorf("expected PUBLISH_UNPUBLISH_VOLUME to be true") }