Skip to content

CSI staging directories are named by a hash of the volume ID #3180

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Eric Curtin (ericcurtin) wants to merge 5 commits into
base: main
Choose a base branch
from
Open

CSI staging directories are named by a hash of the volume ID #3180

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c641ca6
csi: reject a volume ID that escapes the staging directory
ericcurtin Oct 23, 2026
7322fad
csi: unstaging a rejected volume ID is a no-op, so Terminate still works
ericcurtin Oct 23, 2026
0c81411
csi: note an older build may have staged a rejected ID
ericcurtin Oct 24, 2026
1c6737e
csi: stage at a per-driver hash of the volume ID
ericcurtin Nov 9, 2026
3eee9c4
csi: pin staging layout in tests, log staged path per volume
ericcurtin Nov 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 25 additions & 7 deletions internal/volume/csi/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,10 @@ package csi

import (
"context"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"encoding/hex"
"fmt"
"log/slog"
"os"
Expand Down Expand Up @@ -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
Expand All @@ -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"),
}
}
Expand Down Expand Up @@ -286,11 +290,13 @@ func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest)
return "", nil
}

stagingPath := filepath.Join(p.stagingDirPrefix, req.VolumeID)
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)
}

// 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,
Expand Down Expand Up @@ -334,7 +340,7 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error {
return nil
}

stagingPath := filepath.Join(p.stagingDirPrefix, volumeID)
stagingPath := p.stagingDir(volumeID)
req := &csi.NodeUnstageVolumeRequest{
VolumeId: volumeID,
StagingTargetPath: stagingPath,
Expand All @@ -352,10 +358,22 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID string) error {
return nil
}

// removeStagingDir removes an empty staging directory, logging any failure.
// 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 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))
return // The parent cannot be empty either.
}
}
}
Comment on lines +361 to 378

@msau42 Michelle Au (msau42) Nov 10, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 With hashed names, nothing records which volume a staging directory belongs to, so logging staging_path with volume_id when staging would help debug leftovers.

removeStagingDir could skip the parent when removing the leaf fails, to avoid a second ENOTEMPTY warning.

@ericcurtin Eric Curtin (ericcurtin) Nov 10, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in 3eee9c4, thanks.


Expand Down Expand Up @@ -420,7 +438,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)
Expand Down
166 changes: 150 additions & 16 deletions internal/volume/csi/plugin_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,13 @@ package csi

import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"net"
"os"
"path/filepath"
"strings"
"testing"

"github.com/container-storage-interface/spec/lib/go/csi"
Expand Down Expand Up @@ -220,7 +223,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{
Expand All @@ -247,7 +250,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")
Expand All @@ -267,7 +270,7 @@ func TestPlugin_AttachVolume(t *testing.T) {
}
defer client.Close()

plugin := NewPlugin(client)
plugin := NewPlugin(client, mockDriverName)

ctx := context.Background()
plugin.InitControllerCapabilities(ctx)
Expand Down Expand Up @@ -326,7 +329,7 @@ func TestPlugin_DetachVolume(t *testing.T) {
}
defer client.Close()

plugin := NewPlugin(client)
plugin := NewPlugin(client, mockDriverName)

ctx := context.Background()
plugin.InitControllerCapabilities(ctx)
Expand Down Expand Up @@ -377,7 +380,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)
Expand All @@ -399,7 +402,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)
}
Expand All @@ -418,6 +421,132 @@ 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
}{
{name: "plain", volumeID: "vol-1"},
{name: "slashes", volumeID: "projects/p/disks/d-1"},
{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 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()
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, mockDriverName)
plugin.stagingDirPrefix = filepath.Join(root, "staging")
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)
}
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)
}
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)
}
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"}
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) {
driver := &mockCSIDriver{}
endpoint, cleanup := startMockCSIDriver(t, driver)
Expand All @@ -429,7 +558,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)
Expand All @@ -440,7 +569,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)
}
Expand Down Expand Up @@ -544,7 +673,7 @@ func TestPlugin_Capabilities_SkipAttachDetachWhenUnsupported(t *testing.T) {
}
defer client.Close()

plugin := NewPlugin(client)
plugin := NewPlugin(client, mockDriverName)
ctx := context.Background()

plugin.InitControllerCapabilities(ctx)
Expand Down Expand Up @@ -605,7 +734,7 @@ func TestPlugin_Capabilities_SkipStageUnstageWhenUnsupported(t *testing.T) {
}
defer client.Close()

plugin := NewPlugin(client)
plugin := NewPlugin(client, mockDriverName)
ctx := context.Background()

plugin.InitNodeCapabilities(ctx)
Expand Down Expand Up @@ -671,7 +800,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)
Expand Down Expand Up @@ -707,7 +836,7 @@ func TestPlugin_Capabilities_DisableAttachDetachAfterUnimplemented(t *testing.T)
}
defer client.Close()

plugin := NewPlugin(client)
plugin := NewPlugin(client, mockDriverName)
ctx := context.Background()
plugin.InitControllerCapabilities(ctx)

Expand Down Expand Up @@ -758,7 +887,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")
Expand All @@ -776,9 +905,11 @@ 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")
if _, err := os.Stat(stagingPath); !os.IsNotExist(err) {
t.Errorf("expected staging directory %q to be removed, got err: %v", stagingPath, err)
stagingPath := plugin.stagingDir("test-vol")
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 {
Expand Down Expand Up @@ -811,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")
}
Expand Down
Loading