csi: check for empty arguments on CSI endpoint (#8027)

Some of the CSI RPC endpoints were missing validation that the ID or
the Volume definition was present. This could result in nonsense
`CSIVolume` structs being written to raft during registration. This
changeset corrects that bug and adds validation checks to present
nicer error messages to operators in some other cases.

nomad/csi_endpoint.go

@@ -186,6 +186,10 @@ func (v *CSIVolume) Get(args *structs.CSIVolumeGetRequest, reply *structs.CSIVol
 	metricsStart := time.Now()
 	defer metrics.MeasureSince([]string{"nomad", "volume", "get"}, metricsStart)
 
+	if args.ID == "" {
+		return fmt.Errorf("missing volume ID")
+	}
+
 	opts := blockingOptions{
 		queryOpts: &args.QueryOptions,
 		queryMeta: &reply.QueryMeta,
@@ -266,6 +270,10 @@ func (v *CSIVolume) Register(args *structs.CSIVolumeRegisterRequest, reply *stru
 		return structs.ErrPermissionDenied
 	}
 
+	if args.Volumes == nil || len(args.Volumes) == 0 {
+		return fmt.Errorf("missing volume definition")
+	}
+
 	// This is the only namespace we ACL checked, force all the volumes to use it.
 	// We also validate that the plugin exists for each plugin, and validate the
 	// capabilities when the plugin has a controller.
@@ -318,6 +326,10 @@ func (v *CSIVolume) Deregister(args *structs.CSIVolumeDeregisterRequest, reply *
 		return structs.ErrPermissionDenied
 	}
 
+	if len(args.VolumeIDs) == 0 {
+		return fmt.Errorf("missing volume IDs")
+	}
+
 	resp, index, err := v.srv.raftApply(structs.CSIVolumeDeregisterRequestType, args)
 	if err != nil {
 		v.logger.Error("csi raft apply failed", "error", err, "method", "deregister")
@@ -351,6 +363,10 @@ func (v *CSIVolume) Claim(args *structs.CSIVolumeClaimRequest, reply *structs.CS
 		return structs.ErrPermissionDenied
 	}
 
+	if args.VolumeID == "" {
+		return fmt.Errorf("missing volume ID")
+	}
+
 	// COMPAT(1.0): the NodeID field was added after 0.11.0 and so we
 	// need to ensure it's been populated during upgrades from 0.11.0
 	// to later patch versions. Remove this block in 1.0
@@ -564,6 +580,10 @@ func (v *CSIPlugin) Get(args *structs.CSIPluginGetRequest, reply *structs.CSIPlu
 	metricsStart := time.Now()
 	defer metrics.MeasureSince([]string{"nomad", "plugin", "get"}, metricsStart)
 
+	if args.ID == "" {
+		return fmt.Errorf("missing plugin ID")
+	}
+
 	opts := blockingOptions{
 		queryOpts: &args.QueryOptions,
 		queryMeta: &reply.QueryMeta,
@@ -616,6 +636,10 @@ func (v *CSIPlugin) Delete(args *structs.CSIPluginDeleteRequest, reply *structs.
 	metricsStart := time.Now()
 	defer metrics.MeasureSince([]string{"nomad", "plugin", "delete"}, metricsStart)
 
+	if args.ID == "" {
+		return fmt.Errorf("missing plugin ID")
+	}
+
 	resp, index, err := v.srv.raftApply(structs.CSIPluginDeleteRequestType, args)
 	if err != nil {
 		v.logger.Error("csi raft apply failed", "error", err, "method", "delete")