feat: enforce SINGLE_NODE_SINGLE_WRITER access mode in node publish

Return FAILED_PRECONDITION if a volume in NodePublish uses the
SINGLE_NODE_SINGLE_WRITER access mode and is already mounted elsewhere
on the node.

See the second table in this section for more details:
https://github.com/container-storage-interface/spec/blob/v1.7.0/spec.md#nodepublishvolume

pkg/hostpath/nodeserver.go

@@ -31,10 +31,13 @@ import (
 	"k8s.io/utils/mount"
 )
 
-const TopologyKeyNode = "topology.hostpath.csi/node"
+const (
+	TopologyKeyNode = "topology.hostpath.csi/node"
 
-func (hp *hostPath) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
+	failedPreconditionAccessModeConflict = "volume uses SINGLE_NODE_SINGLE_WRITER access mode and is already mounted at a different target path"
+)
 
+func (hp *hostPath) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
 	// Check arguments
 	if req.GetVolumeCapability() == nil {
 		return nil, status.Error(codes.InvalidArgument, "Volume capability missing in request")
@@ -82,6 +85,16 @@ func (hp *hostPath) NodePublishVolume(ctx context.Context, req *csi.NodePublishV
 		return nil, status.Error(codes.NotFound, err.Error())
 	}
 
+	if hasSingleNodeSingleWriterAccessMode(req) {
+		pvName, err := parsePVName(req)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+		if isMountedElsewhere(req, pvName, vol) {
+			return nil, status.Error(codes.FailedPrecondition, failedPreconditionAccessModeConflict)
+		}
+	}
+
 	if !ephemeralVolume {
 		if vol.Staged.Empty() {
 			return nil, status.Errorf(codes.FailedPrecondition, "volume %q must be staged before publishing", vol.VolID)
@@ -521,3 +534,30 @@ func makeFile(pathname string) error {
 	}
 	return nil
 }
+
+// hasSingleNodeSingleWriterAccessMode checks if the publish request uses the
+// SINGLE_NODE_SINGLE_WRITER access mode.
+func hasSingleNodeSingleWriterAccessMode(req *csi.NodePublishVolumeRequest) bool {
+	accessMode := req.GetVolumeCapability().GetAccessMode()
+	return accessMode != nil && accessMode.GetMode() == csi.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER
+}
+
+// parsePVName returns the name of the PersistentVolume to publish on the node.
+func parsePVName(req *csi.NodePublishVolumeRequest) (string, error) {
+	targetPathTokens := strings.Split(req.GetTargetPath(), "/")
+	if len(targetPathTokens) < 2 {
+		return "", fmt.Errorf("failed to parse pvc name from target path")
+	}
+	return targetPathTokens[len(targetPathTokens)-2], nil
+}
+
+// isMountedElsewhere checks if the volume to publish is mounted elsewhere on
+// the node.
+func isMountedElsewhere(req *csi.NodePublishVolumeRequest, pvName string, volume state.Volume) bool {
+	for _, pub := range volume.Published {
+		if strings.Contains(pub, pvName) && pub != req.GetTargetPath() {
+			return true
+		}
+	}
+	return false
+}