diff --git a/internal/cephfs/util.go b/internal/cephfs/util.go
index f530a1a7d..e67bf9e8c 100644
--- a/internal/cephfs/util.go
+++ b/internal/cephfs/util.go
@@ -19,6 +19,7 @@ package cephfs
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 
 	"github.com/ceph/ceph-csi/internal/util"
@@ -26,6 +27,7 @@ import (
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	klog "k8s.io/klog/v2"
 )
 
 type volumeID string
@@ -35,6 +37,7 @@ func execCommandErr(ctx context.Context, program string, args ...string) error {
 	return err
 }
 
+// nolint:unparam //  todo:program values has to be revisited later
 func execCommandJSON(ctx context.Context, v interface{}, program string, args ...string) error {
 	stdout, _, err := util.ExecCommand(ctx, program, args...)
 	if err != nil {
@@ -69,6 +72,32 @@ func (cs *ControllerServer) validateCreateVolumeRequest(req *csi.CreateVolumeReq
 		}
 	}
 
+	if req.VolumeContentSource != nil {
+		volumeSource := req.VolumeContentSource
+		switch volumeSource.Type.(type) {
+		case *csi.VolumeContentSource_Snapshot:
+			snapshot := req.VolumeContentSource.GetSnapshot()
+			// CSI spec requires returning NOT_FOUND when the volumeSource is missing/incorrect.
+			if snapshot == nil {
+				return status.Error(codes.NotFound, "volume Snapshot cannot be empty")
+			}
+			if snapshot.GetSnapshotId() == "" {
+				return status.Error(codes.NotFound, "volume Snapshot ID cannot be empty")
+			}
+		case *csi.VolumeContentSource_Volume:
+			// CSI spec requires returning NOT_FOUND when the volumeSource is missing/incorrect.
+			vol := req.VolumeContentSource.GetVolume()
+			if vol == nil {
+				return status.Error(codes.NotFound, "volume cannot be empty")
+			}
+			if vol.GetVolumeId() == "" {
+				return status.Error(codes.NotFound, "volume ID cannot be empty")
+			}
+
+		default:
+			return status.Error(codes.InvalidArgument, "unsupported volume data source")
+		}
+	}
 	return nil
 }
 
@@ -97,3 +126,35 @@ func (cs *ControllerServer) validateExpandVolumeRequest(req *csi.ControllerExpan
 
 	return nil
 }
+
+func genSnapFromOptions(ctx context.Context, req *csi.CreateSnapshotRequest) (snap *cephfsSnapshot, err error) {
+	cephfsSnap := &cephfsSnapshot{}
+	cephfsSnap.RequestName = req.GetName()
+	snapOptions := req.GetParameters()
+
+	cephfsSnap.Monitors, cephfsSnap.ClusterID, err = getMonsAndClusterID(ctx, snapOptions)
+	if err != nil {
+		return nil, err
+	}
+	if namePrefix, ok := snapOptions["snapshotNamePrefix"]; ok {
+		cephfsSnap.NamePrefix = namePrefix
+	}
+	return cephfsSnap, nil
+}
+
+func getMonsAndClusterID(ctx context.Context, options map[string]string) (monitors, clusterID string, err error) {
+	var ok bool
+
+	if clusterID, ok = options["clusterID"]; !ok {
+		err = errors.New("clusterID must be set")
+		return
+	}
+
+	if monitors, err = util.Mons(csiConfigFile, clusterID); err != nil {
+		klog.Errorf(util.Log(ctx, "failed getting mons (%s)"), err)
+		err = fmt.Errorf("failed to fetch monitor list using clusterID (%s): %w", clusterID, err)
+		return
+	}
+
+	return
+}