diff --git a/block/provider/attach_volume.go b/block/provider/attach_volume.go
index 2ab4108..cd7649a 100644
--- a/block/provider/attach_volume.go
+++ b/block/provider/attach_volume.go
@@ -56,13 +56,21 @@ func (vpcs *VPCSession) AttachVolume(volumeAttachmentRequest provider.VolumeAtta
 	}
 	var volumeAttachResult *models.VolumeAttachment
 	var varp *provider.VolumeAttachmentResponse
-	// If it is Non IKS environment then remove the IKSVolumeAttachment field from request struct which contains clusterID.
-	// TO-DO : Enhance this check. Put it in right place
+	// For non-IKS clusters, strip the IKS cluster ID from the request.
+	// Also determine whether this request targets a Bare Metal Server.
+	isBMS := false
 	if !vpcs.Config.VPCConfig.IsIKS {
 		volumeAttachmentRequest.IKSVolumeAttachment = nil
+		isBMS = volumeAttachmentRequest.BMSVolumeAttachment != nil
 	}
 	volumeAttachment := models.NewVolumeAttachment(volumeAttachmentRequest)
 
+	// Choose the correct attachment manager: BMS SDP API or VSI/IKS API.
+	attachMgr := vpcs.APIClientVolAttachMgr
+	if isBMS {
+		attachMgr = vpcs.getBMSVolAttachMgr()
+	}
+
 	err = vpcs.APIRetry.FlexyRetry(vpcs.Logger, func() (error, bool) {
 		// First , check if volume is already attached or attaching to given instance
 		vpcs.Logger.Info("Checking if volume is already attached by other thread")
@@ -73,13 +81,13 @@ func (vpcs *VPCSession) AttachVolume(volumeAttachmentRequest provider.VolumeAtta
 			return nil, true // stop retry volume already attached
 		}
 		//Try attaching volume if it's not already attached or there is error in getting current volume attachment
-		vpcs.Logger.Info("Attaching volume from VPC provider...", zap.Bool("IKSEnabled?", vpcs.Config.VPCConfig.IsIKS))
-		volumeAttachResult, err = vpcs.APIClientVolAttachMgr.AttachVolume(&volumeAttachment, vpcs.Logger)
+		vpcs.Logger.Info("Attaching volume from VPC provider...", zap.Bool("IKSEnabled?", vpcs.Config.VPCConfig.IsIKS), zap.Bool("isBMS?", isBMS))
+		volumeAttachResult, err = attachMgr.AttachVolume(&volumeAttachment, vpcs.Logger)
 		// Keep retry, until we get the proper volumeAttachResult object
 		if err != nil {
 			return err, skipRetryForObviousErrors(err, vpcs.Config.VPCConfig.IsIKS)
 		}
-		varp = volumeAttachResult.ToVolumeAttachmentResponse(vpcs.Config.VPCConfig.VPCBlockProviderType)
+		varp = volumeAttachResult.ToVolumeAttachmentResponse(vpcs.Config.VPCConfig.VPCBlockProviderType, isBMS)
 		return err, true // stop retry as no error
 	})
 
diff --git a/block/provider/bms_storage_access.go b/block/provider/bms_storage_access.go
new file mode 100644
index 0000000..ad180f0
--- /dev/null
+++ b/block/provider/bms_storage_access.go
@@ -0,0 +1,112 @@
+/**
+ * Copyright 2025 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package provider ...
+package provider
+
+// bms_storage_access.go implements the BMS-specific VPC API calls needed by
+// the CSI controller's ControllerPublishVolume STEP 4:
+//
+//  1. RotateBMSEncryptedSecret — POST .../storage_access/encrypted_secret/rotate
+//     body: {"key":{"id":keyID}}
+//     VPC re-encrypts the NVMe-oF PSK with the RSA public key registered as
+//     VPC key {keyID}.
+//
+//  2. GetBMSServer — GET .../bare_metal_servers/{serverID}
+//     Returns BOTH the host NQN (nvme_qualified_name) AND the encrypted PSK
+//     (storage_access.encrypted_secret) in one round-trip, per the VPC BMS
+//     Block Storage SDP User Guide.
+//
+// All methods delegate to getBMSVolAttachMgr() which returns the
+// lazily-initialised BMS VolumeAttachService.
+
+import (
+	"time"
+
+	"github.com/IBM/ibmcloud-volume-interface/lib/metrics"
+	"github.com/IBM/ibmcloud-volume-interface/lib/provider"
+	"go.uber.org/zap"
+)
+
+// RotateBMSEncryptedSecret calls
+// POST v1/bare_metal_servers/{serverID}/storage_access/encrypted_secret/rotate
+// body: {"key":{"id":keyID}}
+//
+// Tells VPC to re-encrypt the NVMe-oF PSK with the RSA public key registered
+// as VPC key {keyID}.  Must be called before GetBMSServer so that
+// the returned ciphertext is encrypted with this cluster's key.
+func (vpcs *VPCSession) RotateBMSEncryptedSecret(serverID, keyID string) error {
+	vpcs.Logger.Debug("Entry of RotateBMSEncryptedSecret...")
+	defer vpcs.Logger.Debug("Exit from RotateBMSEncryptedSecret...")
+	defer metrics.UpdateDurationFromStart(vpcs.Logger, "RotateBMSEncryptedSecret", time.Now())
+
+	if err := isValidServiceSession(vpcs); err != nil {
+		return err
+	}
+
+	vpcs.Logger.Info("RotateBMSEncryptedSecret: calling VPC API", // pragma: allowlist secret
+		zap.String("serverID", serverID), zap.String("keyID", keyID))
+
+	if err := vpcs.getBMSVolAttachMgr().RotateEncryptedSecret(serverID, keyID, vpcs.Logger); err != nil { // pragma: allowlist secret
+		vpcs.Logger.Error("RotateBMSEncryptedSecret: API call failed", // pragma: allowlist secret
+			zap.String("serverID", serverID), zap.Error(err))
+		return err
+	}
+
+	vpcs.Logger.Info("RotateBMSEncryptedSecret: succeeded", zap.String("serverID", serverID)) // pragma: allowlist secret
+	return nil
+}
+
+// GetBMSServer calls GET v1/bare_metal_servers/{serverID} and returns both
+// the host NQN and the RSA-encrypted PSK ciphertext in a single round-trip.
+//
+// Per the VPC BMS Block Storage SDP User Guide:
+//   nvme_qualified_name            → HOSTNQN  (initiator NQN of this BMS)
+//   storage_access.encrypted_secret → PSK ciphertext encrypted with this cluster's RSA key
+//
+// Call RotateBMSEncryptedSecret first so the ciphertext is encrypted with
+// this cluster's RSA public key before calling GetBMSServer.
+func (vpcs *VPCSession) GetBMSServer(serverID string) (*provider.BMSServerInfo, error) {
+	vpcs.Logger.Debug("Entry of GetBMSServer...")
+	defer vpcs.Logger.Debug("Exit from GetBMSServer...")
+	defer metrics.UpdateDurationFromStart(vpcs.Logger, "GetBMSServer", time.Now())
+
+	if err := isValidServiceSession(vpcs); err != nil {
+		return nil, err
+	}
+
+	vpcs.Logger.Info("GetBMSServer: calling VPC API", zap.String("serverID", serverID))
+
+	resp, err := vpcs.getBMSVolAttachMgr().GetBMSServer(serverID, vpcs.Logger)
+	if err != nil {
+		vpcs.Logger.Error("GetBMSServer: API call failed",
+			zap.String("serverID", serverID), zap.Error(err))
+		return nil, err
+	}
+
+	info := &provider.BMSServerInfo{
+		HostNQN: resp.NVMeQualifiedName,
+	}
+	if resp.StorageAccess != nil {
+		info.EncryptedSecret = resp.StorageAccess.EncryptedSecret // pragma: allowlist secret
+	}
+
+	vpcs.Logger.Info("GetBMSServer: succeeded",
+		zap.String("serverID", serverID),
+		zap.String("hostNQN", info.HostNQN),
+		zap.Bool("encryptedSecretPresent", info.EncryptedSecret != ""))
+	return info, nil
+}
diff --git a/block/provider/detach_volume.go b/block/provider/detach_volume.go
index 6e0c52c..0db64dc 100644
--- a/block/provider/detach_volume.go
+++ b/block/provider/detach_volume.go
@@ -47,6 +47,15 @@ func (vpcs *VPCSession) DetachVolume(volumeAttachmentTemplate provider.VolumeAtt
 		return nil, err
 	}
 
+	// Determine whether this request targets a Bare Metal Server.
+	isBMS := !vpcs.Config.VPCConfig.IsIKS && volumeAttachmentTemplate.BMSVolumeAttachment != nil
+
+	// Choose the correct attachment manager.
+	detachMgr := vpcs.APIClientVolAttachMgr
+	if isBMS {
+		detachMgr = vpcs.getBMSVolAttachMgr()
+	}
+
 	var response *http.Response
 	var volumeAttachment models.VolumeAttachment
 
@@ -59,8 +68,8 @@ func (vpcs *VPCSession) DetachVolume(volumeAttachmentTemplate provider.VolumeAtt
 			vpcs.Logger.Info("Found volume attachment", zap.Reflect("currentVolAttachment", currentVolAttachment))
 			volumeAttachment := models.NewVolumeAttachment(volumeAttachmentTemplate)
 			volumeAttachment.ID = currentVolAttachment.VPCVolumeAttachment.ID
-			vpcs.Logger.Info("Detaching volume from VPC provider...")
-			response, err = vpcs.APIClientVolAttachMgr.DetachVolume(&volumeAttachment, vpcs.Logger) //nolint:bodyclose
+			vpcs.Logger.Info("Detaching volume from VPC provider...", zap.Bool("isBMS?", isBMS))
+			response, err = detachMgr.DetachVolume(&volumeAttachment, vpcs.Logger) //nolint:bodyclose
 
 			if err != nil {
 				return err, skipRetryForObviousErrors(err, vpcs.Config.VPCConfig.IsIKS) // Retry in case of all errors
diff --git a/block/provider/get_volume_attachment.go b/block/provider/get_volume_attachment.go
index db4d09f..7774c93 100644
--- a/block/provider/get_volume_attachment.go
+++ b/block/provider/get_volume_attachment.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/IBM/ibmcloud-volume-interface/lib/provider"
 	userError "github.com/IBM/ibmcloud-volume-vpc/common/messages"
+	"github.com/IBM/ibmcloud-volume-vpc/common/vpcclient/instances"
 	"github.com/IBM/ibmcloud-volume-vpc/common/vpcclient/models"
 	"go.uber.org/zap"
 )
@@ -42,32 +43,36 @@ func (vpcs *VPCSession) GetVolumeAttachment(volumeAttachmentRequest provider.Vol
 	if err != nil {
 		return nil, err
 	}
+
+	// Determine manager and BMS flag the same way as attach/detach.
+	isBMS := !vpcs.Config.VPCConfig.IsIKS && volumeAttachmentRequest.BMSVolumeAttachment != nil
+	mgr := vpcs.APIClientVolAttachMgr
+	if isBMS {
+		mgr = vpcs.getBMSVolAttachMgr()
+	}
+
 	var volumeAttachmentResponse *provider.VolumeAttachmentResponse
 	volumeAttachment := models.NewVolumeAttachment(volumeAttachmentRequest)
 	if len(volumeAttachment.ID) > 0 {
 		//Get volume attachments by ID if it is specified
-		volumeAttachmentResponse, err = vpcs.getVolumeAttachmentByID(volumeAttachment)
+		volumeAttachmentResponse, err = vpcs.getVolumeAttachmentByID(volumeAttachment, mgr, isBMS)
 	} else {
 		// Get volume attachment by Volume ID. This is inefficient operation which requires iteration over volume attachment list
-		volumeAttachmentResponse, err = vpcs.getVolumeAttachmentByVolumeID(volumeAttachment)
+		volumeAttachmentResponse, err = vpcs.getVolumeAttachmentByVolumeID(volumeAttachment, mgr, isBMS)
 	}
 	vpcs.Logger.Info("Volume attachment response", zap.Reflect("volumeAttachmentResponse", volumeAttachmentResponse), zap.Error(err))
 	return volumeAttachmentResponse, err
 }
 
-func (vpcs *VPCSession) getVolumeAttachmentByID(volumeAttachmentRequest models.VolumeAttachment) (*provider.VolumeAttachmentResponse, error) {
+func (vpcs *VPCSession) getVolumeAttachmentByID(volumeAttachmentRequest models.VolumeAttachment, mgr instances.VolumeAttachManager, isBMS bool) (*provider.VolumeAttachmentResponse, error) {
 	vpcs.Logger.Debug("Entry of getVolumeAttachmentByID()")
 	defer vpcs.Logger.Debug("Exit from getVolumeAttachmentByID()")
 	vpcs.Logger.Info("Getting VolumeAttachment from VPC provider...")
 	var err error
 	var volumeAttachmentResult *models.VolumeAttachment
-	/*err = retry(vpcs.Logger, func() error {
-		volumeAttachmentResult, err = vpcs.APIClientVolAttachMgr.GetVolumeAttachment(&volumeAttachmentRequest, vpcs.Logger)
-		return err
-	})*/
 
 	err = vpcs.APIRetry.FlexyRetry(vpcs.Logger, func() (error, bool) {
-		volumeAttachmentResult, err = vpcs.APIClientVolAttachMgr.GetVolumeAttachment(&volumeAttachmentRequest, vpcs.Logger)
+		volumeAttachmentResult, err = mgr.GetVolumeAttachment(&volumeAttachmentRequest, vpcs.Logger)
 		// Keep retry, until we get the proper volumeAttachmentRequest object
 		if err != nil {
 			return err, skipRetryForObviousErrors(err, vpcs.Config.VPCConfig.IsIKS)
@@ -81,19 +86,19 @@ func (vpcs *VPCSession) getVolumeAttachmentByID(volumeAttachmentRequest models.V
 		return nil, userErr
 	}
 
-	volumeAttachmentResponse := volumeAttachmentResult.ToVolumeAttachmentResponse(vpcs.Config.VPCConfig.VPCBlockProviderType)
+	volumeAttachmentResponse := volumeAttachmentResult.ToVolumeAttachmentResponse(vpcs.Config.VPCConfig.VPCBlockProviderType, isBMS)
 	vpcs.Logger.Info("Successfully retrieved volume attachment", zap.Reflect("volumeAttachmentResponse", volumeAttachmentResponse))
 	return volumeAttachmentResponse, err
 }
 
-func (vpcs *VPCSession) getVolumeAttachmentByVolumeID(volumeAttachmentRequest models.VolumeAttachment) (*provider.VolumeAttachmentResponse, error) {
+func (vpcs *VPCSession) getVolumeAttachmentByVolumeID(volumeAttachmentRequest models.VolumeAttachment, mgr instances.VolumeAttachManager, isBMS bool) (*provider.VolumeAttachmentResponse, error) {
 	vpcs.Logger.Debug("Entry of getVolumeAttachmentByVolumeID()")
 	defer vpcs.Logger.Debug("Exit from getVolumeAttachmentByVolumeID()")
 	vpcs.Logger.Info("Getting VolumeAttachmentList from VPC provider...")
 	var volumeAttachmentList *models.VolumeAttachmentList
 	var err error
 	err = vpcs.APIRetry.FlexyRetry(vpcs.Logger, func() (error, bool) {
-		volumeAttachmentList, err = vpcs.APIClientVolAttachMgr.ListVolumeAttachments(&volumeAttachmentRequest, vpcs.Logger)
+		volumeAttachmentList, err = mgr.ListVolumeAttachments(&volumeAttachmentRequest, vpcs.Logger)
 		// Keep retry, until we get the proper volumeAttachmentRequest object
 		if err != nil {
 			return err, skipRetryForObviousErrors(err, vpcs.Config.VPCConfig.IsIKS)
@@ -111,12 +116,12 @@ func (vpcs *VPCSession) getVolumeAttachmentByVolumeID(volumeAttachmentRequest mo
 		// Check if volume ID is matching with requested volume ID
 		if volumeAttachmentItem.Volume.ID == volumeAttachmentRequest.Volume.ID {
 			vpcs.Logger.Info("Successfully found volume attachment", zap.Reflect("volumeAttachment", volumeAttachmentItem))
-			volumeResponse := volumeAttachmentItem.ToVolumeAttachmentResponse(vpcs.Config.VPCConfig.VPCBlockProviderType)
+			volumeResponse := volumeAttachmentItem.ToVolumeAttachmentResponse(vpcs.Config.VPCConfig.VPCBlockProviderType, isBMS)
 			vpcs.Logger.Info("Successfully fetched volume attachment from VPC provider", zap.Reflect("volumeResponse", volumeResponse))
 			return volumeResponse, nil
 		}
 	}
-	// No volume attahment found in the  list. So return error
+	// No volume attachment found in the list. So return error
 	userErr := userError.GetUserError(string(userError.VolumeAttachFindFailed), errors.New("no VolumeAttachment Found"), volumeAttachmentRequest.Volume.ID, *volumeAttachmentRequest.InstanceID)
 	vpcs.Logger.Error("Volume attachment not found", zap.Error(err))
 	return nil, userErr
diff --git a/block/provider/session.go b/block/provider/session.go
index a02bf73..af962fb 100644
--- a/block/provider/session.go
+++ b/block/provider/session.go
@@ -18,6 +18,8 @@
 package provider
 
 import (
+	"sync"
+
 	"github.com/IBM/ibmcloud-volume-interface/lib/provider"
 	vpcconfig "github.com/IBM/ibmcloud-volume-vpc/block/vpcconfig"
 	"github.com/IBM/ibmcloud-volume-vpc/common/vpcclient/instances"
@@ -25,6 +27,14 @@ import (
 	"go.uber.org/zap"
 )
 
+// bmsOnceState holds the sync.Once state for lazy BMS manager init.
+// It is heap-allocated so that VPCSession remains copyable (sync.Once must
+// not be copied after first use, but VPCSession is copied in the IKS path).
+type bmsOnceState struct {
+	once sync.Once
+	mgr  instances.VolumeAttachManager
+}
+
 // VPCSession implements lib.Session
 type VPCSession struct {
 	provider.DefaultVolumeProvider
@@ -39,6 +49,23 @@ type VPCSession struct {
 	Logger                *zap.Logger
 	APIRetry              FlexyRetry
 	SessionError          error
+
+	// bmsState is heap-allocated so that copying VPCSession (as the IKS path
+	// does) does not violate the sync.Once noCopy constraint.
+	// Only initialised on the first BMS request; nil on IKS clusters.
+	bmsState *bmsOnceState
+}
+
+// getBMSVolAttachMgr returns the lazily-initialised BMS volume-attachment
+// manager.  It is safe for concurrent use via sync.Once.
+func (vpcs *VPCSession) getBMSVolAttachMgr() instances.VolumeAttachManager {
+	if vpcs.bmsState == nil {
+		vpcs.bmsState = &bmsOnceState{}
+	}
+	vpcs.bmsState.once.Do(func() {
+		vpcs.bmsState.mgr = vpcs.Apiclient.BMSVolumeAttachService()
+	})
+	return vpcs.bmsState.mgr
 }
 
 const (
diff --git a/block/provider/wait_for_attach_volume.go b/block/provider/wait_for_attach_volume.go
index 27734df..4ac19dd 100644
--- a/block/provider/wait_for_attach_volume.go
+++ b/block/provider/wait_for_attach_volume.go
@@ -23,10 +23,12 @@ import (
 	"github.com/IBM/ibmcloud-volume-interface/lib/metrics"
 	"github.com/IBM/ibmcloud-volume-interface/lib/provider"
 	userError "github.com/IBM/ibmcloud-volume-vpc/common/messages"
+	"github.com/IBM/ibmcloud-volume-vpc/common/vpcclient/models"
 	"go.uber.org/zap"
 )
 
-// WaitForAttachVolume waits for volume to be attached to node. e.g waits till status becomes attached
+// WaitForAttachVolume waits for volume to be attached to node. e.g waits till status becomes attached.
+// For BMS SDP volumes the terminal status is "available"; for VSI volumes it is "attached".
 func (vpcs *VPCSession) WaitForAttachVolume(volumeAttachmentTemplate provider.VolumeAttachmentRequest) (*provider.VolumeAttachmentResponse, error) {
 	vpcs.Logger.Debug("Entry of WaitForAttachVolume method...")
 	defer vpcs.Logger.Debug("Exit from WaitForAttachVolume method...")
@@ -44,6 +46,13 @@ func (vpcs *VPCSession) WaitForAttachVolume(volumeAttachmentTemplate provider.Vo
 		return nil, err
 	}
 
+	// BMS SDP volumes report "available" when ready; VSI volumes report "attached".
+	isBMS := !vpcs.Config.VPCConfig.IsIKS && volumeAttachmentTemplate.BMSVolumeAttachment != nil
+	terminalStatus := StatusAttached
+	if isBMS {
+		terminalStatus = models.VolumeAvailable
+	}
+
 	var currentVolAttachment *provider.VolumeAttachmentResponse
 	err = vpcs.APIRetry.FlexyRetryWithCustomGap(vpcs.Logger, func() (error, bool) {
 		currentVolAttachment, err = vpcs.GetVolumeAttachment(volumeAttachmentTemplate)
@@ -52,11 +61,11 @@ func (vpcs *VPCSession) WaitForAttachVolume(volumeAttachmentTemplate provider.Vo
 			// considering that vpcs.GetVolumeAttachment already re-tried
 			return err, true
 		}
-		// Stop retry in case of volume is attached
-		return err, currentVolAttachment != nil && currentVolAttachment.Status == StatusAttached
+		// Stop retry once the volume reaches its terminal attached status
+		return err, currentVolAttachment != nil && currentVolAttachment.Status == terminalStatus
 	})
-	// Success case, checks are required in case of timeout happened and volume is still not attached state
-	if err == nil && (currentVolAttachment != nil && currentVolAttachment.Status == StatusAttached) {
+	// Success case, checks are required in case of timeout happened and volume is still not in attached state
+	if err == nil && (currentVolAttachment != nil && currentVolAttachment.Status == terminalStatus) {
 		return currentVolAttachment, nil
 	}
 
diff --git a/common/vpcclient/instances/bms_storage_access.go b/common/vpcclient/instances/bms_storage_access.go
new file mode 100644
index 0000000..2fe1ec3
--- /dev/null
+++ b/common/vpcclient/instances/bms_storage_access.go
@@ -0,0 +1,133 @@
+/**
+ * Copyright 2025 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package instances ...
+package instances
+
+// bms_storage_access.go implements the BMS-specific VPC API methods as
+// methods on VolumeAttachService (reusing the same client, pathPrefix and
+// error handler already wired by NewBMSVolumeAttachmentManager).
+//
+// IKSVolumeAttachService stubs return errors because these operations are
+// BMS-only and must never be called via the IKS path.
+//
+// Paths (relative to vs.pathPrefix = "v1/bare_metal_servers"):
+//   POST /{instance-id}/storage_access/encrypted_secret/rotate
+//   GET  /{instance-id}
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/IBM/ibmcloud-volume-interface/lib/metrics"
+	util "github.com/IBM/ibmcloud-volume-interface/lib/utils"
+	"github.com/IBM/ibmcloud-volume-vpc/common/vpcclient/client"
+	"github.com/IBM/ibmcloud-volume-vpc/common/vpcclient/models"
+	"go.uber.org/zap"
+)
+
+// ─── VolumeAttachService (BMS path) ──────────────────────────────────────────
+
+// RotateEncryptedSecret calls
+// POST v1/bare_metal_servers/{serverID}/storage_access/encrypted_secret/rotate
+// body: {"key":{"id":keyID}}
+func (vs *VolumeAttachService) RotateEncryptedSecret(serverID, keyID string, ctxLogger *zap.Logger) error {
+	methodName := "VolumeAttachService.RotateEncryptedSecret"
+	defer util.TimeTracker(methodName, time.Now())
+	defer metrics.UpdateDurationFromStart(ctxLogger, methodName, time.Now())
+
+	operation := &client.Operation{
+		Name:        "RotateEncryptedSecret",
+		Method:      "POST",
+		PathPattern: vs.pathPrefix + bmsRotateSecretSuffix,
+	}
+
+	reqBody := &models.BMSStorageAccessRotateRequest{
+		Key: models.BMSKeyRef{ID: keyID},
+	}
+
+	apiErr := vs.receiverError
+	operationRequest := vs.client.NewRequest(operation)
+
+	ctxLogger.Info("Equivalent curl command and payload details",
+		zap.Reflect("URL", operationRequest.URL()),
+		zap.Reflect("Payload", reqBody),
+		zap.Reflect("Operation", operation),
+		zap.String("serverID", serverID))
+
+	_, err := operationRequest.
+		PathParameter(instanceIDParam, serverID).
+		JSONBody(reqBody).
+		JSONError(apiErr).
+		Invoke()
+	if err != nil {
+		return err
+	}
+
+	ctxLogger.Info("RotateEncryptedSecret: succeeded", zap.String("serverID", serverID)) // pragma: allowlist secret
+	return nil
+}
+
+// GetBMSServer calls
+// GET v1/bare_metal_servers/{serverID}
+// and returns the HOSTNQN + encrypted_secret in one round-trip.
+func (vs *VolumeAttachService) GetBMSServer(serverID string, ctxLogger *zap.Logger) (*models.BMSServerResponse, error) {
+	methodName := "VolumeAttachService.GetBMSServer"
+	defer util.TimeTracker(methodName, time.Now())
+	defer metrics.UpdateDurationFromStart(ctxLogger, methodName, time.Now())
+
+	operation := &client.Operation{
+		Name:        "GetBMSServer",
+		Method:      "GET",
+		PathPattern: vs.pathPrefix + bmsServerSuffix,
+	}
+
+	var server models.BMSServerResponse
+	apiErr := vs.receiverError
+	operationRequest := vs.client.NewRequest(operation)
+
+	ctxLogger.Info("Equivalent curl command and payload details",
+		zap.Reflect("URL", operationRequest.URL()),
+		zap.Reflect("Operation", operation),
+		zap.String("serverID", serverID))
+
+	_, err := operationRequest.
+		PathParameter(instanceIDParam, serverID).
+		JSONSuccess(&server).
+		JSONError(apiErr).
+		Invoke()
+	if err != nil {
+		return nil, err
+	}
+
+	ctxLogger.Info("GetBMSServer: succeeded",
+		zap.String("serverID", serverID),
+		zap.String("hostNQN", server.NVMeQualifiedName),
+		zap.Bool("encryptedSecretPresent", server.StorageAccess != nil && server.StorageAccess.EncryptedSecret != ""))
+	return &server, nil
+}
+
+// ─── IKSVolumeAttachService stubs (BMS operations not supported via IKS) ─────
+
+// RotateEncryptedSecret is not supported via the IKS path.
+func (vs *IKSVolumeAttachService) RotateEncryptedSecret(serverID, keyID string, ctxLogger *zap.Logger) error {
+	return fmt.Errorf("RotateEncryptedSecret is not supported via the IKS volume attach path")
+}
+
+// GetBMSServer is not supported via the IKS path.
+func (vs *IKSVolumeAttachService) GetBMSServer(serverID string, ctxLogger *zap.Logger) (*models.BMSServerResponse, error) {
+	return nil, fmt.Errorf("GetBMSServer is not supported via the IKS volume attach path")
+}
diff --git a/common/vpcclient/instances/constants.go b/common/vpcclient/instances/constants.go
index 0902547..e416f92 100644
--- a/common/vpcclient/instances/constants.go
+++ b/common/vpcclient/instances/constants.go
@@ -25,12 +25,20 @@ const (
 	instanceIDvolumeAttachmentPath = instanceIDPath + "/" + volumeAttachmentPath
 	instanceIDattachmentIDPath     = instanceIDvolumeAttachmentPath + "/{" + attachmentIDParam + "}"
 
+	// BMS paths — used by RotateEncryptedSecret and GetBMSServer.
+	// All operations share the same server-ID path parameter (instanceIDParam).
+	bmsServerSuffix       = instanceIDPath                               // GET  /bare_metal_servers/{id}
+	bmsRotateSecretSuffix = instanceIDPath + "/storage_access/encrypted_secret/rotate" // POST
+
 	// VpcPathPrefix  VPC URL path prefix
 	VpcPathPrefix = "v1/instances"
 
 	// IksPathPrefix  IKS URL path prefix
 	IksPathPrefix = "v2/storage/vpc/"
 
+	// BMSPathPrefix  Bare Metal Server URL path prefix for SDP volume attachments
+	BMSPathPrefix = "v1/bare_metal_servers"
+
 	// IksClusterQueryKey ...
 	IksClusterQueryKey = "cluster"
 
diff --git a/common/vpcclient/instances/fakes/volume_attach_service.go b/common/vpcclient/instances/fakes/volume_attach_service.go
index 7fd7900..0e045fb 100644
--- a/common/vpcclient/instances/fakes/volume_attach_service.go
+++ b/common/vpcclient/instances/fakes/volume_attach_service.go
@@ -39,6 +39,20 @@ type VolumeAttachService struct {
 		result1 *http.Response
 		result2 error
 	}
+	GetBMSServerStub        func(string, *zap.Logger) (*models.BMSServerResponse, error)
+	getBMSServerMutex       sync.RWMutex
+	getBMSServerArgsForCall []struct {
+		arg1 string
+		arg2 *zap.Logger
+	}
+	getBMSServerReturns struct {
+		result1 *models.BMSServerResponse
+		result2 error
+	}
+	getBMSServerReturnsOnCall map[int]struct {
+		result1 *models.BMSServerResponse
+		result2 error
+	}
 	GetVolumeAttachmentStub        func(*models.VolumeAttachment, *zap.Logger) (*models.VolumeAttachment, error)
 	getVolumeAttachmentMutex       sync.RWMutex
 	getVolumeAttachmentArgsForCall []struct {
@@ -67,6 +81,19 @@ type VolumeAttachService struct {
 		result1 *models.VolumeAttachmentList
 		result2 error
 	}
+	RotateEncryptedSecretStub        func(string, string, *zap.Logger) error
+	rotateEncryptedSecretMutex       sync.RWMutex
+	rotateEncryptedSecretArgsForCall []struct {
+		arg1 string
+		arg2 string
+		arg3 *zap.Logger
+	}
+	rotateEncryptedSecretReturns struct {
+		result1 error
+	}
+	rotateEncryptedSecretReturnsOnCall map[int]struct {
+		result1 error
+	}
 	invocations      map[string][][]interface{}
 	invocationsMutex sync.RWMutex
 }
@@ -201,6 +228,71 @@ func (fake *VolumeAttachService) DetachVolumeReturnsOnCall(i int, result1 *http.
 	}{result1, result2}
 }
 
+func (fake *VolumeAttachService) GetBMSServer(arg1 string, arg2 *zap.Logger) (*models.BMSServerResponse, error) {
+	fake.getBMSServerMutex.Lock()
+	ret, specificReturn := fake.getBMSServerReturnsOnCall[len(fake.getBMSServerArgsForCall)]
+	fake.getBMSServerArgsForCall = append(fake.getBMSServerArgsForCall, struct {
+		arg1 string
+		arg2 *zap.Logger
+	}{arg1, arg2})
+	stub := fake.GetBMSServerStub
+	fakeReturns := fake.getBMSServerReturns
+	fake.recordInvocation("GetBMSServer", []interface{}{arg1, arg2})
+	fake.getBMSServerMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2)
+	}
+	if specificReturn {
+		return ret.result1, ret.result2
+	}
+	return fakeReturns.result1, fakeReturns.result2
+}
+
+func (fake *VolumeAttachService) GetBMSServerCallCount() int {
+	fake.getBMSServerMutex.RLock()
+	defer fake.getBMSServerMutex.RUnlock()
+	return len(fake.getBMSServerArgsForCall)
+}
+
+func (fake *VolumeAttachService) GetBMSServerCalls(stub func(string, *zap.Logger) (*models.BMSServerResponse, error)) {
+	fake.getBMSServerMutex.Lock()
+	defer fake.getBMSServerMutex.Unlock()
+	fake.GetBMSServerStub = stub
+}
+
+func (fake *VolumeAttachService) GetBMSServerArgsForCall(i int) (string, *zap.Logger) {
+	fake.getBMSServerMutex.RLock()
+	defer fake.getBMSServerMutex.RUnlock()
+	argsForCall := fake.getBMSServerArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2
+}
+
+func (fake *VolumeAttachService) GetBMSServerReturns(result1 *models.BMSServerResponse, result2 error) {
+	fake.getBMSServerMutex.Lock()
+	defer fake.getBMSServerMutex.Unlock()
+	fake.GetBMSServerStub = nil
+	fake.getBMSServerReturns = struct {
+		result1 *models.BMSServerResponse
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *VolumeAttachService) GetBMSServerReturnsOnCall(i int, result1 *models.BMSServerResponse, result2 error) {
+	fake.getBMSServerMutex.Lock()
+	defer fake.getBMSServerMutex.Unlock()
+	fake.GetBMSServerStub = nil
+	if fake.getBMSServerReturnsOnCall == nil {
+		fake.getBMSServerReturnsOnCall = make(map[int]struct {
+			result1 *models.BMSServerResponse
+			result2 error
+		})
+	}
+	fake.getBMSServerReturnsOnCall[i] = struct {
+		result1 *models.BMSServerResponse
+		result2 error
+	}{result1, result2}
+}
+
 func (fake *VolumeAttachService) GetVolumeAttachment(arg1 *models.VolumeAttachment, arg2 *zap.Logger) (*models.VolumeAttachment, error) {
 	fake.getVolumeAttachmentMutex.Lock()
 	ret, specificReturn := fake.getVolumeAttachmentReturnsOnCall[len(fake.getVolumeAttachmentArgsForCall)]
@@ -331,6 +423,69 @@ func (fake *VolumeAttachService) ListVolumeAttachmentsReturnsOnCall(i int, resul
 	}{result1, result2}
 }
 
+func (fake *VolumeAttachService) RotateEncryptedSecret(arg1 string, arg2 string, arg3 *zap.Logger) error {
+	fake.rotateEncryptedSecretMutex.Lock()
+	ret, specificReturn := fake.rotateEncryptedSecretReturnsOnCall[len(fake.rotateEncryptedSecretArgsForCall)]
+	fake.rotateEncryptedSecretArgsForCall = append(fake.rotateEncryptedSecretArgsForCall, struct {
+		arg1 string
+		arg2 string
+		arg3 *zap.Logger
+	}{arg1, arg2, arg3})
+	stub := fake.RotateEncryptedSecretStub
+	fakeReturns := fake.rotateEncryptedSecretReturns
+	fake.recordInvocation("RotateEncryptedSecret", []interface{}{arg1, arg2, arg3})
+	fake.rotateEncryptedSecretMutex.Unlock()
+	if stub != nil {
+		return stub(arg1, arg2, arg3)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *VolumeAttachService) RotateEncryptedSecretCallCount() int {
+	fake.rotateEncryptedSecretMutex.RLock()
+	defer fake.rotateEncryptedSecretMutex.RUnlock()
+	return len(fake.rotateEncryptedSecretArgsForCall)
+}
+
+func (fake *VolumeAttachService) RotateEncryptedSecretCalls(stub func(string, string, *zap.Logger) error) {
+	fake.rotateEncryptedSecretMutex.Lock()
+	defer fake.rotateEncryptedSecretMutex.Unlock()
+	fake.RotateEncryptedSecretStub = stub
+}
+
+func (fake *VolumeAttachService) RotateEncryptedSecretArgsForCall(i int) (string, string, *zap.Logger) {
+	fake.rotateEncryptedSecretMutex.RLock()
+	defer fake.rotateEncryptedSecretMutex.RUnlock()
+	argsForCall := fake.rotateEncryptedSecretArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+}
+
+func (fake *VolumeAttachService) RotateEncryptedSecretReturns(result1 error) {
+	fake.rotateEncryptedSecretMutex.Lock()
+	defer fake.rotateEncryptedSecretMutex.Unlock()
+	fake.RotateEncryptedSecretStub = nil
+	fake.rotateEncryptedSecretReturns = struct {
+		result1 error
+	}{result1}
+}
+
+func (fake *VolumeAttachService) RotateEncryptedSecretReturnsOnCall(i int, result1 error) {
+	fake.rotateEncryptedSecretMutex.Lock()
+	defer fake.rotateEncryptedSecretMutex.Unlock()
+	fake.RotateEncryptedSecretStub = nil
+	if fake.rotateEncryptedSecretReturnsOnCall == nil {
+		fake.rotateEncryptedSecretReturnsOnCall = make(map[int]struct {
+			result1 error
+		})
+	}
+	fake.rotateEncryptedSecretReturnsOnCall[i] = struct {
+		result1 error
+	}{result1}
+}
+
 func (fake *VolumeAttachService) Invocations() map[string][][]interface{} {
 	fake.invocationsMutex.RLock()
 	defer fake.invocationsMutex.RUnlock()
@@ -338,10 +493,14 @@ func (fake *VolumeAttachService) Invocations() map[string][][]interface{} {
 	defer fake.attachVolumeMutex.RUnlock()
 	fake.detachVolumeMutex.RLock()
 	defer fake.detachVolumeMutex.RUnlock()
+	fake.getBMSServerMutex.RLock()
+	defer fake.getBMSServerMutex.RUnlock()
 	fake.getVolumeAttachmentMutex.RLock()
 	defer fake.getVolumeAttachmentMutex.RUnlock()
 	fake.listVolumeAttachmentsMutex.RLock()
 	defer fake.listVolumeAttachmentsMutex.RUnlock()
+	fake.rotateEncryptedSecretMutex.RLock()
+	defer fake.rotateEncryptedSecretMutex.RUnlock()
 	copiedInvocations := map[string][][]interface{}{}
 	for key, value := range fake.invocations {
 		copiedInvocations[key] = value
diff --git a/common/vpcclient/instances/volume_mounter_service.go b/common/vpcclient/instances/volume_mounter_service.go
index 01c3972..16bee78 100644
--- a/common/vpcclient/instances/volume_mounter_service.go
+++ b/common/vpcclient/instances/volume_mounter_service.go
@@ -37,6 +37,21 @@ type VolumeAttachManager interface {
 	ListVolumeAttachments(*models.VolumeAttachment, *zap.Logger) (*models.VolumeAttachmentList, error)
 	// Delete the volume
 	DetachVolume(*models.VolumeAttachment, *zap.Logger) (*http.Response, error)
+
+	// RotateEncryptedSecret calls
+	// POST v1/bare_metal_servers/{serverID}/storage_access/encrypted_secret/rotate
+	// body: {"key":{"id":keyID}}
+	// BMS-only: tells VPC to re-encrypt the NVMe-oF PSK with the RSA public key
+	// registered as VPC key {keyID}.  No-op / returns error on non-BMS managers.
+	RotateEncryptedSecret(serverID, keyID string, ctxLogger *zap.Logger) error
+
+	// GetBMSServer calls GET v1/bare_metal_servers/{serverID} and returns the
+	// fields the CSI controller needs in one round-trip:
+	//   nvme_qualified_name          → HOSTNQN (initiator NQN of the server)
+	//   storage_access.encrypted_secret → RSA-encrypted PSK ciphertext
+	// BMS-only: returns error on non-BMS managers.
+	GetBMSServer(serverID string, ctxLogger *zap.Logger) (*models.BMSServerResponse, error)
+
 }
 
 // VolumeAttachService ...
@@ -81,3 +96,20 @@ func NewIKSVolumeAttachmentManager(clientIn client.SessionClient) VolumeAttachMa
 		receiverError: &err,
 	}
 }
+
+// NewBMSVolumeAttachmentManager creates a VolumeAttachManager that targets the
+// Bare Metal Server SDP volume-attachment API (v1/bare_metal_servers/…).
+// It reuses VolumeAttachService because the BMS REST resource has the same
+// path structure as the VSI resource: /<serverID>/volume_attachments[/<id>].
+func NewBMSVolumeAttachmentManager(clientIn client.SessionClient) VolumeAttachManager {
+	err := models.Error{}
+	return &VolumeAttachService{
+		client:        clientIn,
+		pathPrefix:    BMSPathPrefix,
+		receiverError: &err,
+		populatePathPrefixParameters: func(request *client.Request, volumeAttachmentTemplate *models.VolumeAttachment) *client.Request {
+			request.PathParameter(instanceIDParam, *volumeAttachmentTemplate.InstanceID)
+			return request
+		},
+	}
+}
diff --git a/common/vpcclient/models/bms_storage_access.go b/common/vpcclient/models/bms_storage_access.go
new file mode 100644
index 0000000..e0e8e38
--- /dev/null
+++ b/common/vpcclient/models/bms_storage_access.go
@@ -0,0 +1,72 @@
+/**
+ * Copyright 2025 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package models ...
+package models
+
+// BMSStorageAccessRotateRequest is the request body for
+// POST v1/bare_metal_servers/{serverID}/storage_access/encrypted_secret/rotate
+//
+// Example curl:
+//
+//	curl -X POST \
+//	  '.../v1/bare_metal_servers/{id}/storage_access/encrypted_secret/rotate?version=...&generation=2' \
+//	  -H 'Authorization: Bearer TOKEN' \
+//	  -H 'Content-Type: application/json' \
+//	  -d '{"key":{"id":"r134-e2f1a01f-bf35-444b-ba20-e3633cc201cd"}}'
+//
+// VPC re-encrypts the NVMe-oF PSK with the RSA public key registered as the
+// VPC key resource identified by Key.ID and stores the ciphertext.
+type BMSStorageAccessRotateRequest struct {
+	Key BMSKeyRef `json:"key"`
+}
+
+// BMSKeyRef references a VPC key resource by ID.
+type BMSKeyRef struct {
+	ID string `json:"id"`
+}
+
+// BMSStorageAccessResponse is the response from
+// GET v1/bare_metal_servers/{serverID}/storage_access
+// It is also embedded inside BMSServerResponse as the storage_access sub-object.
+type BMSStorageAccessResponse struct {
+	// EncryptedSecret is the base64-encoded RSA ciphertext of the NVMe-oF PSK.
+	// Encrypted with the public key identified by the key ID used in the
+	// preceding rotate call.  Safe to store in etcd via publishContext because
+	// it is useless without the rsa_private_key held only by the node DaemonSet.
+	EncryptedSecret string `json:"encrypted_secret"`
+}
+
+// BMSServerResponse is the relevant subset of the response from
+// GET v1/bare_metal_servers/{id}
+//
+// According to the VPC BMS Block Storage SDP User Guide:
+//
+//	nvme_qualified_name          → HOSTNQN: the host/initiator NQN used in ALL nvme commands
+//	storage_access.encrypted_secret → PSK ciphertext (same as GET /storage_access)
+//
+// Both fields are returned in a single API call, so the controller can get
+// the host NQN and the encrypted PSK together after one rotate + one GET.
+type BMSServerResponse struct {
+	// NVMeQualifiedName is the HOSTNQN — the initiator NQN of the bare metal
+	// server itself.  Stable for the lifetime of the server.
+	// Used as: nvme connect --hostnqn <NVMeQualifiedName>
+	NVMeQualifiedName string `json:"nvme_qualified_name,omitempty"`
+
+	// StorageAccess embeds the storage_access sub-object which contains the
+	// encrypted PSK ciphertext (same data as GET /storage_access).
+	StorageAccess *BMSStorageAccessResponse `json:"storage_access,omitempty"`
+}
diff --git a/common/vpcclient/models/constants.go b/common/vpcclient/models/constants.go
index 70a8acd..c3b8a7d 100644
--- a/common/vpcclient/models/constants.go
+++ b/common/vpcclient/models/constants.go
@@ -44,4 +44,14 @@ const (
 
 	// VolumeAttached ...
 	VolumeAttached = "attached"
+
+	// VolumeAvailable is the terminal attach status for BMS SDP volumes
+	VolumeAvailable = "available"
+
+	// GTypeBMS ...
+	GTypeBMS = "bms"
+
+	// GTypeBMSDevicePrefix is the NVMe-oF device path prefix used for BMS SDP volumes.
+	// Full path is GTypeBMSDevicePrefix + <full-volume-uuid>
+	GTypeBMSDevicePrefix = "/dev/disk/by-id/nvme-uuid."
 )
diff --git a/common/vpcclient/models/volume_attachment.go b/common/vpcclient/models/volume_attachment.go
index 42cc074..dc52ce0 100644
--- a/common/vpcclient/models/volume_attachment.go
+++ b/common/vpcclient/models/volume_attachment.go
@@ -28,15 +28,20 @@ type Device struct {
 	ID string `json:"id"`
 }
 
+// BMSVolumeIP holds an IP address entry returned in the BMS volume attachment IPs list
+type BMSVolumeIP struct {
+	Address string `json:"address,omitempty"`
+}
+
 // VolumeAttachment for riaas client
 type VolumeAttachment struct {
 	ID   string `json:"id,omitempty"`
 	Href string `json:"href,omitempty"`
 	Name string `json:"name,omitempty"`
-	// Status of volume attachment named - attaching , attached, detaching
+	// Status of volume attachment named - attaching, attached, detaching (VSI) or available (BMS)
 	Status string `json:"status,omitempty"`
 	Type   string `json:"type,omitempty"` //boot, data
-	// InstanceID this volume is attached to
+	// InstanceID this volume is attached to (also used as BMS server ID)
 	InstanceID *string    `json:"-"`
 	ClusterID  *string    `json:"clusterID,omitempty"`
 	Device     *Device    `json:"device,omitempty"`
@@ -44,6 +49,16 @@ type VolumeAttachment struct {
 	CreatedAt  *time.Time `json:"created_at,omitempty"`
 	// If set to true, when deleting the instance the volume will also be deleted
 	DeleteVolumeOnInstanceDelete bool `json:"delete_volume_on_instance_delete,omitempty"`
+
+	// BMS SDP-specific fields (only populated for bare metal server attachments)
+	// Protocol is the host protocol, e.g. "nvme_tcp"
+	Protocol string `json:"interface_protocol,omitempty"`
+	// DeleteVolumeOnBMSDelete mirrors the VSI delete-on-delete flag for BMS
+	DeleteVolumeOnBMSDelete bool `json:"delete_volume_on_bare_metal_server_delete,omitempty"`
+	// NVMeQualifiedName is the NQN for the volume on BMS
+	NVMeQualifiedName string `json:"nvme_qualified_name,omitempty"`
+	// IPs is the list of NVMe-oF target IPs returned by the BMS attach API
+	IPs []BMSVolumeIP `json:"ips,omitempty"`
 }
 
 // VolumeAttachmentList ...
@@ -68,11 +83,16 @@ func NewVolumeAttachment(volumeAttachmentRequest provider.VolumeAttachmentReques
 	if volumeAttachmentRequest.IKSVolumeAttachment != nil {
 		va.ClusterID = volumeAttachmentRequest.IKSVolumeAttachment.ClusterID
 	}
+	if volumeAttachmentRequest.BMSVolumeAttachment != nil {
+		va.Protocol = volumeAttachmentRequest.BMSVolumeAttachment.Protocol
+	}
 	return va
 }
 
-// ToVolumeAttachmentResponse converts VolumeAttachment VolumeAttachmentResponse
-func (va *VolumeAttachment) ToVolumeAttachmentResponse(providerType string) *provider.VolumeAttachmentResponse {
+// ToVolumeAttachmentResponse converts VolumeAttachment to VolumeAttachmentResponse.
+// Pass isBMS=true when the attachment was made via the BMS SDP API so that
+// the correct terminal status and device-path prefix are applied.
+func (va *VolumeAttachment) ToVolumeAttachmentResponse(providerType string, isBMS bool) *provider.VolumeAttachmentResponse {
 	varp := &provider.VolumeAttachmentResponse{
 		VolumeAttachmentRequest: provider.VolumeAttachmentRequest{
 			VolumeID: va.Volume.ID,
@@ -91,14 +111,34 @@ func (va *VolumeAttachment) ToVolumeAttachmentResponse(providerType string) *pro
 		varp.InstanceID = *va.InstanceID
 	}
 
-	//Set DevicePath
-	if va.Status == VolumeAttached && va.Device != nil && va.Device.ID != "" {
-		if providerType == GTypeG2 {
-			if len(va.Device.ID) >= GTypeG2DeviceIDLength {
-				varp.VolumeAttachmentRequest.VPCVolumeAttachment.DevicePath = GTypeG2DevicePrefix + va.Device.ID[:GTypeG2DeviceIDLength]
+	if isBMS {
+		// BMS SDP: terminal status is "available"; device path uses NVMe UUID prefix
+		// with the full volume UUID (no truncation).
+		if va.Status == VolumeAvailable && va.Volume != nil && va.Volume.ID != "" {
+			varp.VolumeAttachmentRequest.VPCVolumeAttachment.DevicePath = GTypeBMSDevicePrefix + va.Volume.ID
+		}
+		// Propagate NVMe-oF fields needed by the CSI controller to build publishContext
+		if va.NVMeQualifiedName != "" {
+			bmsResp := &provider.BMSVolumeAttachmentResponse{
+				SubsystemNQN: va.NVMeQualifiedName,
+			}
+			for _, ip := range va.IPs {
+				if ip.Address != "" {
+					bmsResp.TargetIPs = append(bmsResp.TargetIPs, ip.Address)
+				}
+			}
+			varp.BMSResponse = bmsResp
+		}
+	} else {
+		// VSI path: terminal status is "attached"; device path uses virtio or classic prefix.
+		if va.Status == VolumeAttached && va.Device != nil && va.Device.ID != "" {
+			if providerType == GTypeG2 {
+				if len(va.Device.ID) >= GTypeG2DeviceIDLength {
+					varp.VolumeAttachmentRequest.VPCVolumeAttachment.DevicePath = GTypeG2DevicePrefix + va.Device.ID[:GTypeG2DeviceIDLength]
+				}
+			} else { //GC
+				varp.VolumeAttachmentRequest.VPCVolumeAttachment.DevicePath = GTypeClassicDevicePrefix + va.Device.ID
 			}
-		} else { //GC
-			varp.VolumeAttachmentRequest.VPCVolumeAttachment.DevicePath = GTypeClassicDevicePrefix + va.Device.ID
 		}
 	}
 	return varp
diff --git a/common/vpcclient/riaas/fakes/regional_api.go b/common/vpcclient/riaas/fakes/regional_api.go
index af0b304..765f465 100644
--- a/common/vpcclient/riaas/fakes/regional_api.go
+++ b/common/vpcclient/riaas/fakes/regional_api.go
@@ -10,6 +10,16 @@ import (
 )
 
 type RegionalAPI struct {
+	BMSVolumeAttachServiceStub        func() instances.VolumeAttachManager
+	bMSVolumeAttachServiceMutex       sync.RWMutex
+	bMSVolumeAttachServiceArgsForCall []struct {
+	}
+	bMSVolumeAttachServiceReturns struct {
+		result1 instances.VolumeAttachManager
+	}
+	bMSVolumeAttachServiceReturnsOnCall map[int]struct {
+		result1 instances.VolumeAttachManager
+	}
 	IKSVolumeAttachServiceStub        func() instances.VolumeAttachManager
 	iKSVolumeAttachServiceMutex       sync.RWMutex
 	iKSVolumeAttachServiceArgsForCall []struct {
@@ -65,6 +75,59 @@ type RegionalAPI struct {
 	invocationsMutex sync.RWMutex
 }
 
+func (fake *RegionalAPI) BMSVolumeAttachService() instances.VolumeAttachManager {
+	fake.bMSVolumeAttachServiceMutex.Lock()
+	ret, specificReturn := fake.bMSVolumeAttachServiceReturnsOnCall[len(fake.bMSVolumeAttachServiceArgsForCall)]
+	fake.bMSVolumeAttachServiceArgsForCall = append(fake.bMSVolumeAttachServiceArgsForCall, struct {
+	}{})
+	stub := fake.BMSVolumeAttachServiceStub
+	fakeReturns := fake.bMSVolumeAttachServiceReturns
+	fake.recordInvocation("BMSVolumeAttachService", []interface{}{})
+	fake.bMSVolumeAttachServiceMutex.Unlock()
+	if stub != nil {
+		return stub()
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	return fakeReturns.result1
+}
+
+func (fake *RegionalAPI) BMSVolumeAttachServiceCallCount() int {
+	fake.bMSVolumeAttachServiceMutex.RLock()
+	defer fake.bMSVolumeAttachServiceMutex.RUnlock()
+	return len(fake.bMSVolumeAttachServiceArgsForCall)
+}
+
+func (fake *RegionalAPI) BMSVolumeAttachServiceCalls(stub func() instances.VolumeAttachManager) {
+	fake.bMSVolumeAttachServiceMutex.Lock()
+	defer fake.bMSVolumeAttachServiceMutex.Unlock()
+	fake.BMSVolumeAttachServiceStub = stub
+}
+
+func (fake *RegionalAPI) BMSVolumeAttachServiceReturns(result1 instances.VolumeAttachManager) {
+	fake.bMSVolumeAttachServiceMutex.Lock()
+	defer fake.bMSVolumeAttachServiceMutex.Unlock()
+	fake.BMSVolumeAttachServiceStub = nil
+	fake.bMSVolumeAttachServiceReturns = struct {
+		result1 instances.VolumeAttachManager
+	}{result1}
+}
+
+func (fake *RegionalAPI) BMSVolumeAttachServiceReturnsOnCall(i int, result1 instances.VolumeAttachManager) {
+	fake.bMSVolumeAttachServiceMutex.Lock()
+	defer fake.bMSVolumeAttachServiceMutex.Unlock()
+	fake.BMSVolumeAttachServiceStub = nil
+	if fake.bMSVolumeAttachServiceReturnsOnCall == nil {
+		fake.bMSVolumeAttachServiceReturnsOnCall = make(map[int]struct {
+			result1 instances.VolumeAttachManager
+		})
+	}
+	fake.bMSVolumeAttachServiceReturnsOnCall[i] = struct {
+		result1 instances.VolumeAttachManager
+	}{result1}
+}
+
 func (fake *RegionalAPI) IKSVolumeAttachService() instances.VolumeAttachManager {
 	fake.iKSVolumeAttachServiceMutex.Lock()
 	ret, specificReturn := fake.iKSVolumeAttachServiceReturnsOnCall[len(fake.iKSVolumeAttachServiceArgsForCall)]
@@ -341,6 +404,8 @@ func (fake *RegionalAPI) VolumeServiceReturnsOnCall(i int, result1 vpcvolume.Vol
 func (fake *RegionalAPI) Invocations() map[string][][]interface{} {
 	fake.invocationsMutex.RLock()
 	defer fake.invocationsMutex.RUnlock()
+	fake.bMSVolumeAttachServiceMutex.RLock()
+	defer fake.bMSVolumeAttachServiceMutex.RUnlock()
 	fake.iKSVolumeAttachServiceMutex.RLock()
 	defer fake.iKSVolumeAttachServiceMutex.RUnlock()
 	fake.loginMutex.RLock()
diff --git a/common/vpcclient/riaas/riaas.go b/common/vpcclient/riaas/riaas.go
index fa09c56..f95b2e4 100644
--- a/common/vpcclient/riaas/riaas.go
+++ b/common/vpcclient/riaas/riaas.go
@@ -38,6 +38,7 @@ type RegionalAPI interface {
 	VolumeService() vpcvolume.VolumeManager
 	VolumeAttachService() instances.VolumeAttachManager
 	IKSVolumeAttachService() instances.VolumeAttachManager
+	BMSVolumeAttachService() instances.VolumeAttachManager
 	SnapshotService() vpcvolume.SnapshotManager
 }
 
@@ -108,6 +109,11 @@ func (s *Session) IKSVolumeAttachService() instances.VolumeAttachManager {
 	return instances.NewIKSVolumeAttachmentManager(s.client)
 }
 
+// BMSVolumeAttachService returns the VolumeAttachService for managing BMS SDP volume attachments
+func (s *Session) BMSVolumeAttachService() instances.VolumeAttachManager {
+	return instances.NewBMSVolumeAttachmentManager(s.client)
+}
+
 // SnapshotService returns the Snapshot service for managing snapshot
 func (s *Session) SnapshotService() vpcvolume.SnapshotManager {
 	return vpcvolume.NewSnapshotManager(s.client)
diff --git a/go.mod b/go.mod
index bca4ae8..9bbd90d 100644
--- a/go.mod
+++ b/go.mod
@@ -1,10 +1,10 @@
 module github.com/IBM/ibmcloud-volume-vpc
 
-go 1.26.3
+go 1.26.9
 
 require (
 	github.com/IBM-Cloud/ibm-cloud-cli-sdk v1.10.5
-	github.com/IBM/ibmcloud-volume-interface v1.2.21
+	github.com/IBM/ibmcloud-volume-interface v1.2.24-0.20261030133322-0a24a5f8ac70
 	github.com/IBM/secret-common-lib v1.1.15
 	github.com/IBM/secret-utils-lib v1.1.16
 	github.com/fatih/structs v1.1.0
diff --git a/go.sum b/go.sum
index 72a6b39..f6f468a 100644
--- a/go.sum
+++ b/go.sum
@@ -4,8 +4,8 @@ github.com/IBM-Cloud/ibm-cloud-cli-sdk v1.10.5 h1:yIiwzsycjtQhLSqkhE6W/tOZT6T3Lc
 github.com/IBM-Cloud/ibm-cloud-cli-sdk v1.10.5/go.mod h1:Ass3v/R9GvOn8wQ7zEDF4/0oWUPo0fRpvilD8xOAO7w=
 github.com/IBM/go-sdk-core/v5 v5.17.4 h1:VGb9+mRrnS2HpHZFM5hy4J6ppIWnwNrw0G+tLSgcJLc=
 github.com/IBM/go-sdk-core/v5 v5.17.4/go.mod h1:KsAAI7eStAWwQa4F96MLy+whYSh39JzNjklZRbN/8ns=
-github.com/IBM/ibmcloud-volume-interface v1.2.21 h1:7w7ZAY3F5DLfzaV4sLoC/4S0AYXZrUPHanrlBHaNb4c=
-github.com/IBM/ibmcloud-volume-interface v1.2.21/go.mod h1:lpon+MgOK9F0Q8AxZDG/dWXc9vJiiYfPri9ToskBA08=
+github.com/IBM/ibmcloud-volume-interface v1.2.24-0.20261030133322-0a24a5f8ac70 h1:0IZ8tCDkYXERdhwvMjx41rAuFcbU7aGq74S8Ag5sk9I=
+github.com/IBM/ibmcloud-volume-interface v1.2.24-0.20261030133322-0a24a5f8ac70/go.mod h1:OVtko4KazT4Z0Yd4ZzJdFd0/xLDm7OS/ZIFCOxqGXys=
 github.com/IBM/secret-common-lib v1.1.15 h1:r2oQE7UAjHiPHhUUWJmI1NYYmuTRHWENtFolTSbXMwI=
 github.com/IBM/secret-common-lib v1.1.15/go.mod h1:bJO2iazBFnfjj6jhfgy/T4FQQQE/aIxio1CrNZ83+0U=
 github.com/IBM/secret-utils-lib v1.1.16 h1:RIainy11ICJVWjgi9FWSvBsl8xOwhzLvTKBI8AWq7wI=
diff --git a/iks/provider/session.go b/iks/provider/session.go
index 1948df4..ed61293 100644
--- a/iks/provider/session.go
+++ b/iks/provider/session.go
@@ -18,6 +18,7 @@
 package provider
 
 import (
+	"fmt"
 	"net/http"
 
 	"github.com/IBM/ibmcloud-volume-interface/lib/provider"
@@ -93,3 +94,14 @@ func (vpcIks *IksVpcSession) WaitForDetachVolume(volumeAttachmentRequest provide
 	defer vpcIks.IksSession.Logger.Debug("Exit from IksVpcSession.WaitForDetachVolume method...")
 	return vpcIks.IksSession.WaitForDetachVolume(volumeAttachmentRequest)
 }
+
+// RotateBMSEncryptedSecret is not supported via the IKS dual session path.
+func (vpcIks *IksVpcSession) RotateBMSEncryptedSecret(serverID, keyID string) error {
+	return fmt.Errorf("RotateBMSEncryptedSecret is not supported via the IKS session path")
+}
+
+// GetBMSServer is not supported via the IKS dual session path.
+func (vpcIks *IksVpcSession) GetBMSServer(serverID string) (*provider.BMSServerInfo, error) {
+	return nil, fmt.Errorf("GetBMSServer is not supported via the IKS session path")
+}
+
diff --git a/pkg/ibmcloudprovider/fake_volume_provider.go b/pkg/ibmcloudprovider/fake_volume_provider.go
index 8f33b11..d62ce0a 100644
--- a/pkg/ibmcloudprovider/fake_volume_provider.go
+++ b/pkg/ibmcloudprovider/fake_volume_provider.go
@@ -47,13 +47,13 @@ const (
 	IAMURL = "test-iam-url"
 
 	// IAMClientID ...
-	IAMClientID = "test-iam_client_id"
+	IAMClientID = "test-iam_client_id" // pragma: allowlist secret
 
 	// IAMClientSecret ...
-	IAMClientSecret = "test-iam_client_secret"
+	IAMClientSecret = "test-iam_client_secret" // pragma: allowlist secret
 
 	// IAMAPIKey ...
-	IAMAPIKey = "test-iam_api_key" // #nosec G101
+	IAMAPIKey = "test-iam_api_key" // #nosec G101 // pragma: allowlist secret
 
 	// RefreshToken ...
 	RefreshToken = "test-refresh_token"
@@ -127,3 +127,13 @@ func (ficp *FakeIBMCloudStorageProvider) GetConfig() *config.Config {
 func (ficp *FakeIBMCloudStorageProvider) GetClusterID() string {
 	return ficp.ClusterID
 }
+
+// GetBMSPrivKey returns nil for test use (non-BMS fake provider).
+func (ficp *FakeIBMCloudStorageProvider) GetBMSPrivKey() []byte {
+	return nil
+}
+
+// GetBMSKeyID returns empty string for test use (non-BMS fake provider).
+func (ficp *FakeIBMCloudStorageProvider) GetBMSKeyID() string {
+	return ""
+}
diff --git a/pkg/ibmcloudprovider/ibm_volume_provider.go b/pkg/ibmcloudprovider/ibm_volume_provider.go
index b82f914..bc80ab0 100644
--- a/pkg/ibmcloudprovider/ibm_volume_provider.go
+++ b/pkg/ibmcloudprovider/ibm_volume_provider.go
@@ -30,4 +30,13 @@ type CloudProviderInterface interface {
 	GetProviderSession(ctx context.Context, logger *zap.Logger) (provider.Session, error)
 	GetConfig() *config.Config
 	GetClusterID() string
+	// GetBMSPrivKey returns the PEM-encoded RSA private key loaded at startup
+	// from the k8s secret "storage-bm-secret" (key "rsa_private_key").
+	// Returns nil on non-BMS nodes where the secret does not exist.
+	GetBMSPrivKey() []byte
+	// GetBMSKeyID returns the VPC SSH Key resource ID loaded at startup from
+	// the k8s secret "storage-bm-secret" (key "key_id").
+	// Used by the controller in STEP 4 to call the rotate API.
+	// Returns empty string on non-BMS clusters.
+	GetBMSKeyID() string
 }
diff --git a/pkg/ibmcloudprovider/volume_provider.go b/pkg/ibmcloudprovider/volume_provider.go
index 6d6d198..7804d0b 100644
--- a/pkg/ibmcloudprovider/volume_provider.go
+++ b/pkg/ibmcloudprovider/volume_provider.go
@@ -35,12 +35,60 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	// bmsSecretName is the k8s Secret created by the cluster admin that holds
+	// the RSA key pair used for BMS NVMe-oF/TCP PSK encryption.
+	//
+	// Expected secret fields (create once per cluster):
+	//
+	//   key_id          – VPC SSH Key resource ID returned by:
+	//                       ibmcloud is key-create <name> --key-material @rsa_public_key.pub
+	//                     The controller passes this ID to the rotate API so VPC
+	//                     knows which stored public key to encrypt the PSK with.
+	//
+	//   rsa_private_key – PEM-encoded RSA private key.  Read by the node
+	//                     DaemonSet to RSA-OAEP decrypt the ciphertext received
+	//                     via publishContext at NodeStageVolume time.
+	//
+	//   rsa_public_key  – PEM-encoded RSA public key (stored for reference only).
+	//                     The driver does NOT read this field at runtime; it must
+	//                     be registered with VPC via `ibmcloud is key-create`
+	//                     before creating BMS volume attachments so VPC can
+	//                     encrypt the PSK ciphertext.
+	bmsSecretName = "storage-bm-secret"
+
+	// bmsSecretPrivKeyField is the Secret data key that holds the
+	// PEM-encoded RSA private key (used by the node DaemonSet only).
+	bmsSecretPrivKeyField = "rsa_private_key"
+
+	// bmsSecretKeyIDField is the Secret data key that holds the VPC SSH Key
+	// resource ID used by the controller to call the rotate API.
+	// The corresponding RSA public key must already be registered in VPC as an
+	// SSH key with this ID so VPC can encrypt the PSK for this cluster.
+	bmsSecretKeyIDField = "key_id"
+
+	// bmsSecretPubKeyField names the public-key field stored in the secret for
+	// operator reference.  The driver does not read this field; it is used only
+	// during one-time VPC SSH Key registration.
+	bmsSecretPubKeyField = "rsa_public_key" //nolint:deadcode,unused
+)
+
 // IBMCloudStorageProvider Provider
 type IBMCloudStorageProvider struct {
 	ProviderName   string
 	ProviderConfig *config.Config
 	Registry       registry.Providers
 	ClusterID      string
+	// BMSPrivKey is the PEM-encoded RSA private key read once at startup from
+	// the k8s secret "storage-bm-secret".  Nil on non-BMS clusters where the
+	// secret does not exist — all BMS code paths guard on this being non-nil.
+	BMSPrivKey []byte
+	// BMSKeyID is the VPC SSH Key resource ID read once at startup from the
+	// k8s secret "storage-bm-secret" (key "key_id").
+	// Used by the controller in STEP 4 to call the rotate API so VPC encrypts
+	// the PSK with the cluster's RSA public key.
+	// Empty on non-BMS clusters.
+	BMSKeyID string
 }
 
 var _ CloudProviderInterface = &IBMCloudStorageProvider{}
@@ -98,11 +146,40 @@ func NewIBMCloudStorageProvider(clusterVolumeLabel string, k8sClient *k8s_utils.
 		providerName = conf.VPC.VPCBlockProviderName
 	}
 
+	// Read BMS credentials from "storage-bm-secret".  Both reads are non-fatal:
+	// if the secret does not exist (non-BMS cluster) the fields stay at their
+	// zero values and all BMS code paths are skipped automatically.
+	//
+	// rsa_private_key — node DaemonSet only: decrypt PSK at NodeStageVolume
+	// key_id          — controller only:     passed in the rotate API body
+	var bmsPrivKey []byte
+	var bmsKeyID string
+
+	if privKeyPEM, err := k8s_utils.GetSecretData(*k8sClient, bmsSecretName, bmsSecretPrivKeyField); err != nil {
+		logger.Warn("NewIBMCloudStorageProvider: BMS private key not found — BMS NVMe attach will be disabled on node",
+			zap.String("secret", bmsSecretName), zap.String("key", bmsSecretPrivKeyField), zap.Error(err))
+	} else {
+		bmsPrivKey = []byte(privKeyPEM)
+		logger.Info("NewIBMCloudStorageProvider: BMS private key loaded successfully",
+			zap.String("secret", bmsSecretName))
+	}
+
+	if keyID, err := k8s_utils.GetSecretData(*k8sClient, bmsSecretName, bmsSecretKeyIDField); err != nil {
+		logger.Warn("NewIBMCloudStorageProvider: BMS key_id not found — encrypted_secret rotate will be disabled on controller",
+			zap.String("secret", bmsSecretName), zap.String("key", bmsSecretKeyIDField), zap.Error(err))
+	} else {
+		bmsKeyID = keyID
+		logger.Info("NewIBMCloudStorageProvider: BMS key_id loaded successfully",
+			zap.String("secret", bmsSecretName))
+	}
+
 	cloudProvider := &IBMCloudStorageProvider{
 		ProviderName:   providerName,
 		ProviderConfig: conf,
 		Registry:       registry,
 		ClusterID:      clusterInfo.ClusterID,
+		BMSPrivKey:     bmsPrivKey,
+		BMSKeyID:       bmsKeyID,
 	}
 	logger.Info("Successfully read provider configuration")
 	return cloudProvider, nil
@@ -154,6 +231,18 @@ func (icp *IBMCloudStorageProvider) GetClusterID() string {
 	return icp.ClusterID
 }
 
+// GetBMSPrivKey returns the PEM-encoded RSA private key loaded at startup from
+// the k8s secret "storage-bm-secret".  Returns nil on non-BMS nodes.
+func (icp *IBMCloudStorageProvider) GetBMSPrivKey() []byte {
+	return icp.BMSPrivKey
+}
+
+// GetBMSKeyID returns the VPC SSH Key resource ID loaded at startup from
+// the k8s secret "storage-bm-secret".  Returns empty string on non-BMS clusters.
+func (icp *IBMCloudStorageProvider) GetBMSKeyID() string {
+	return icp.BMSKeyID
+}
+
 // CorrectEndpointURL corrects endpoint url if it is of form "http://"
 func getEndpointURL(url string, logger *zap.Logger) string {
 	if strings.Contains(url, "http://") {