diff --git a/.changelog/12050.txt b/.changelog/12050.txt
new file mode 100644
index 000000000..593378f45
--- /dev/null
+++ b/.changelog/12050.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+csi: provide `CSI_ENDPOINT` environment variable to plugin tasks
+```
diff --git a/client/allocrunner/taskrunner/plugin_supervisor_hook.go b/client/allocrunner/taskrunner/plugin_supervisor_hook.go
index 2398ed030..9adb8d53b 100644
--- a/client/allocrunner/taskrunner/plugin_supervisor_hook.go
+++ b/client/allocrunner/taskrunner/plugin_supervisor_hook.go
@@ -34,6 +34,8 @@ type csiPluginSupervisorHook struct {
 	runner     *TaskRunner
 	mountPoint string
 
+	caps *drivers.Capabilities
+
 	// eventEmitter is used to emit events to the task
 	eventEmitter ti.EventEmitter
 
@@ -48,6 +50,14 @@ type csiPluginSupervisorHook struct {
 	previousHealthState bool
 }
 
+type csiPluginSupervisorHookConfig struct {
+	clientStateDirPath string
+	events             ti.EventEmitter
+	runner             *TaskRunner
+	capabilities       *drivers.Capabilities
+	logger             hclog.Logger
+}
+
 // The plugin supervisor uses the PrestartHook mechanism to setup the requisite
 // mount points and configuration for the task that exposes a CSI plugin.
 var _ interfaces.TaskPrestartHook = &csiPluginSupervisorHook{}
@@ -61,8 +71,8 @@ var _ interfaces.TaskPoststartHook = &csiPluginSupervisorHook{}
 // with the catalog and to ensure any mounts are cleaned up.
 var _ interfaces.TaskStopHook = &csiPluginSupervisorHook{}
 
-func newCSIPluginSupervisorHook(csiRootDir string, eventEmitter ti.EventEmitter, runner *TaskRunner, logger hclog.Logger) *csiPluginSupervisorHook {
-	task := runner.Task()
+func newCSIPluginSupervisorHook(config *csiPluginSupervisorHookConfig) *csiPluginSupervisorHook {
+	task := config.runner.Task()
 
 	// The Plugin directory will look something like this:
 	// .
@@ -72,19 +82,21 @@ func newCSIPluginSupervisorHook(csiRootDir string, eventEmitter ti.EventEmitter,
 	//  {volume-id}/{usage-mode-hash}/ - Intermediary mount point that will be used by plugins that support NODE_STAGE_UNSTAGE capabilities.
 	// per-alloc/
 	//  {alloc-id}/{volume-id}/{usage-mode-hash}/ - Mount Point that will be bind-mounted into tasks that utilise the volume
-	pluginRoot := filepath.Join(csiRootDir, string(task.CSIPluginConfig.Type), task.CSIPluginConfig.ID)
+	pluginRoot := filepath.Join(config.clientStateDirPath, "csi",
+		string(task.CSIPluginConfig.Type), task.CSIPluginConfig.ID)
 
 	shutdownCtx, cancelFn := context.WithCancel(context.Background())
 
 	hook := &csiPluginSupervisorHook{
-		alloc:            runner.Alloc(),
-		runner:           runner,
-		logger:           logger,
+		alloc:            config.runner.Alloc(),
+		runner:           config.runner,
+		logger:           config.logger,
 		task:             task,
 		mountPoint:       pluginRoot,
+		caps:             config.capabilities,
 		shutdownCtx:      shutdownCtx,
 		shutdownCancelFn: cancelFn,
-		eventEmitter:     eventEmitter,
+		eventEmitter:     config.events,
 	}
 
 	return hook
@@ -119,6 +131,20 @@ func (h *csiPluginSupervisorHook) Prestart(ctx context.Context,
 		Readonly: false,
 	}
 
+	switch h.caps.FSIsolation {
+	case drivers.FSIsolationNone:
+		// Plugin tasks with no filesystem isolation won't have the
+		// plugin dir bind-mounted to their alloc dir, but we can
+		// provide them the path to the socket. These Nomad-only
+		// plugins will need to be aware of the csi directory layout
+		// in the client data dir
+		resp.Env = map[string]string{
+			"CSI_ENDPOINT": filepath.Join(h.mountPoint, "csi.sock")}
+	default:
+		resp.Env = map[string]string{
+			"CSI_ENDPOINT": filepath.Join(h.task.CSIPluginConfig.MountDir, "csi.sock")}
+	}
+
 	mounts := ensureMountpointInserted(h.runner.hookResources.getMounts(), configMount)
 	mounts = ensureMountpointInserted(mounts, devMount)
 
diff --git a/client/allocrunner/taskrunner/task_runner_hooks.go b/client/allocrunner/taskrunner/task_runner_hooks.go
index a92194d48..b3f44a8b9 100644
--- a/client/allocrunner/taskrunner/task_runner_hooks.go
+++ b/client/allocrunner/taskrunner/task_runner_hooks.go
@@ -3,7 +3,6 @@ package taskrunner
 import (
 	"context"
 	"fmt"
-	"path/filepath"
 	"sync"
 	"time"
 
@@ -72,7 +71,14 @@ func (tr *TaskRunner) initHooks() {
 
 	// If the task has a CSI stanza, add the hook.
 	if task.CSIPluginConfig != nil {
-		tr.runnerHooks = append(tr.runnerHooks, newCSIPluginSupervisorHook(filepath.Join(tr.clientConfig.StateDir, "csi"), tr, tr, hookLogger))
+		tr.runnerHooks = append(tr.runnerHooks, newCSIPluginSupervisorHook(
+			&csiPluginSupervisorHookConfig{
+				clientStateDirPath: tr.clientConfig.StateDir,
+				events:             tr,
+				runner:             tr,
+				capabilities:       tr.driverCapabilities,
+				logger:             hookLogger,
+			}))
 	}
 
 	// If Vault is enabled, add the hook