diff --git a/pkg/driver/node/mounter/credential_provider.go b/pkg/driver/node/mounter/credential_provider.go
index b72d2dfb..3e94dc26 100644
--- a/pkg/driver/node/mounter/credential_provider.go
+++ b/pkg/driver/node/mounter/credential_provider.go
@@ -20,6 +20,8 @@ import (
 	k8sv1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/klog/v2"
 	k8sstrings "k8s.io/utils/strings"
+
+	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/volumecontext"
 )
 
 const hostPluginDirEnv = "HOST_PLUGIN_DIR"
@@ -34,15 +36,6 @@ const (
 	AuthenticationSourcePod         AuthenticationSource = "pod"
 )
 
-const (
-	VolumeCtxAuthenticationSource = "authenticationSource"
-	VolumeCtxSTSRegion            = "stsRegion"
-	VolumeCtxServiceAccountName   = "csi.storage.k8s.io/serviceAccount.name"
-	VolumeCtxServiceAccountTokens = "csi.storage.k8s.io/serviceAccount.tokens"
-	VolumeCtxPodNamespace         = "csi.storage.k8s.io/pod.namespace"
-	VolumeCtxPodUID               = "csi.storage.k8s.io/pod.uid"
-)
-
 const (
 	// This is to ensure only owner/group can read the file and no one else.
 	serviceAccountTokenPerm = 0440
@@ -96,7 +89,7 @@ func (c *CredentialProvider) Provide(ctx context.Context, volumeID string, volum
 		return nil, status.Error(codes.InvalidArgument, "Missing volume context")
 	}
 
-	authenticationSource := volumeContext[VolumeCtxAuthenticationSource]
+	authenticationSource := volumeContext[volumecontext.AuthenticationSource]
 	switch authenticationSource {
 	case AuthenticationSourcePod:
 		return c.provideFromPod(ctx, volumeID, volumeContext, mountpointArgs)
@@ -129,7 +122,7 @@ func (c *CredentialProvider) provideFromDriver() (*MountCredentials, error) {
 func (c *CredentialProvider) provideFromPod(ctx context.Context, volumeID string, volumeContext map[string]string, mountpointArgs []string) (*MountCredentials, error) {
 	klog.V(4).Infof("NodePublishVolume: Using pod identity")
 
-	tokensJson := volumeContext[VolumeCtxServiceAccountTokens]
+	tokensJson := volumeContext[volumecontext.CSIServiceAccountTokens]
 	if tokensJson == "" {
 		klog.Error("`authenticationSource` configured to `pod` but no service account tokens are received. Please make sure to enable `podInfoOnMountCompat`, see " + podLevelCredentialsDocsPage)
 		return nil, status.Error(codes.InvalidArgument, "Missing service account tokens")
@@ -161,7 +154,7 @@ func (c *CredentialProvider) provideFromPod(ctx context.Context, volumeID string
 		defaultRegion = region
 	}
 
-	podID := volumeContext[VolumeCtxPodUID]
+	podID := volumeContext[volumecontext.CSIPodUID]
 	if podID == "" {
 		return nil, status.Error(codes.InvalidArgument, "Missing Pod info. Please make sure to enable `podInfoOnMountCompat`, see "+podLevelCredentialsDocsPage)
 	}
@@ -174,8 +167,8 @@ func (c *CredentialProvider) provideFromPod(ctx context.Context, volumeID string
 	hostPluginDir := hostPluginDirWithDefault()
 	hostTokenPath := path.Join(hostPluginDir, c.tokenFilename(podID, volumeID))
 
-	podNamespace := volumeContext[VolumeCtxPodNamespace]
-	podServiceAccount := volumeContext[VolumeCtxServiceAccountName]
+	podNamespace := volumeContext[volumecontext.CSIPodNamespace]
+	podServiceAccount := volumeContext[volumecontext.CSIServiceAccountName]
 	cacheKey := podNamespace + "/" + podServiceAccount
 
 	return &MountCredentials{
@@ -222,8 +215,8 @@ func (c *CredentialProvider) tokenFilename(podID string, volumeID string) string
 }
 
 func (c *CredentialProvider) findPodServiceAccountRole(ctx context.Context, volumeContext map[string]string) (string, error) {
-	podNamespace := volumeContext[VolumeCtxPodNamespace]
-	podServiceAccount := volumeContext[VolumeCtxServiceAccountName]
+	podNamespace := volumeContext[volumecontext.CSIPodNamespace]
+	podServiceAccount := volumeContext[volumecontext.CSIServiceAccountName]
 	if podNamespace == "" || podServiceAccount == "" {
 		klog.Error("`authenticationSource` configured to `pod` but no pod info found. Please make sure to enable `podInfoOnMountCompat`, see " + podLevelCredentialsDocsPage)
 		return "", status.Error(codes.InvalidArgument, "Missing Pod info. Please make sure to enable `podInfoOnMountCompat`, see "+podLevelCredentialsDocsPage)
@@ -253,7 +246,7 @@ func (c *CredentialProvider) findPodServiceAccountRole(ctx context.Context, volu
 //
 // It returns an error if all of them fails.
 func (c *CredentialProvider) stsRegion(volumeContext map[string]string, mountpointArgs []string) (string, error) {
-	region := volumeContext[VolumeCtxSTSRegion]
+	region := volumeContext[volumecontext.STSRegion]
 	if region != "" {
 		klog.V(5).Infof("NodePublishVolume: Pod-level: Detected STS region %s from volume context", region)
 		return region, nil
diff --git a/pkg/driver/node/node.go b/pkg/driver/node/node.go
index fe5e888d..0dccb01d 100644
--- a/pkg/driver/node/node.go
+++ b/pkg/driver/node/node.go
@@ -31,11 +31,11 @@ import (
 
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/targetpath"
+	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/volumecontext"
 )
 
 const (
-	volumeCtxBucketName = "bucketName"
-	defaultKubeletPath  = "/var/lib/kubelet"
+	defaultKubeletPath = "/var/lib/kubelet"
 )
 
 var kubeletPath = getKubeletPath()
@@ -68,8 +68,8 @@ func NewS3NodeServer(nodeID string, mounter mounter.Mounter, credentialProvider
 
 func (ns *S3NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
 	volumeContext := req.GetVolumeContext()
-	if volumeContext[mounter.VolumeCtxAuthenticationSource] == mounter.AuthenticationSourcePod {
-		podID := volumeContext[mounter.VolumeCtxPodUID]
+	if volumeContext[volumecontext.AuthenticationSource] == mounter.AuthenticationSourcePod {
+		podID := volumeContext[volumecontext.CSIPodUID]
 		volumeID := req.GetVolumeId()
 		if podID != "" && volumeID != "" {
 			err := ns.credentialProvider.CleanupToken(volumeID, podID)
@@ -96,7 +96,7 @@ func (ns *S3NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePubl
 
 	volumeContext := req.GetVolumeContext()
 
-	bucket, ok := volumeContext[volumeCtxBucketName]
+	bucket, ok := volumeContext[volumecontext.BucketName]
 	if !ok {
 		return nil, status.Error(codes.InvalidArgument, "Bucket name not provided")
 	}
@@ -295,7 +295,7 @@ func (ns *S3NodeServer) isValidVolumeCapabilities(volCaps []*csi.VolumeCapabilit
 // with sensitive fields removed.
 func logSafeNodePublishVolumeRequest(req *csi.NodePublishVolumeRequest) *csi.NodePublishVolumeRequest {
 	safeVolumeContext := maps.Clone(req.VolumeContext)
-	delete(safeVolumeContext, mounter.VolumeCtxServiceAccountTokens)
+	delete(safeVolumeContext, volumecontext.CSIServiceAccountTokens)
 
 	return &csi.NodePublishVolumeRequest{
 		VolumeId:          req.VolumeId,
diff --git a/pkg/driver/node/volumecontext/volume_context.go b/pkg/driver/node/volumecontext/volume_context.go
new file mode 100644
index 00000000..adc11ed1
--- /dev/null
+++ b/pkg/driver/node/volumecontext/volume_context.go
@@ -0,0 +1,13 @@
+// Package volumecontext provides utilities for accessing volume context passed via CSI RPC.
+package volumecontext
+
+const (
+	BucketName           = "bucketName"
+	AuthenticationSource = "authenticationSource"
+	STSRegion            = "stsRegion"
+
+	CSIServiceAccountName   = "csi.storage.k8s.io/serviceAccount.name"
+	CSIServiceAccountTokens = "csi.storage.k8s.io/serviceAccount.tokens"
+	CSIPodNamespace         = "csi.storage.k8s.io/pod.namespace"
+	CSIPodUID               = "csi.storage.k8s.io/pod.uid"
+)