Check for ownership on migrate cancel and status

Signed-off-by: Grant Griffiths <[email protected]>

api/server/sdk/volume_migrate.go

@@ -162,6 +162,28 @@ func (s *VolumeServer) volumeMigrate(
 	}, nil
 }
 
+func (s *VolumeServer) checkMigrationPermissions(ctx context.Context, taskId string) error {
+	// Inspect migration to get VolumeIds
+	resp, err := s.driver(ctx).CloudMigrateStatus(&api.CloudMigrateStatusRequest{
+		TaskId: taskId,
+	})
+	if err != nil {
+		return status.Errorf(codes.Internal, "Failed to get migration information : %v", err)
+	}
+
+	// Check that a user has access to all volumes being migrated
+	for _, cluster := range resp.Info {
+		for _, migrateInfo := range cluster.List {
+			if err := checkAccessFromDriverForVolumeId(ctx, s.driver(ctx),
+				migrateInfo.GetLocalVolumeId(), api.Ownership_Read); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
 // Cancel or stop a ongoing migration
 func (s *VolumeServer) Cancel(
 	ctx context.Context,
@@ -170,12 +192,16 @@ func (s *VolumeServer) Cancel(
 	if s.cluster() == nil || s.driver(ctx) == nil {
 		return nil, status.Error(codes.Unavailable, "Resource has not been initialized")
 	}
-
 	if req.GetRequest() == nil {
 		return nil, status.Errorf(codes.InvalidArgument, "Must supply valid request")
 	} else if len(req.GetRequest().GetTaskId()) == 0 {
 		return nil, status.Errorf(codes.InvalidArgument, "Must supply valid Task ID")
 	}
+
+	// Check if the user has access to all volumes associated with the TaskID
+	if err := s.checkMigrationPermissions(ctx, req.GetRequest().GetTaskId()); err != nil {
+		return nil, err
+	}
 	err := s.driver(ctx).CloudMigrateCancel(req.GetRequest())
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Cannot stop migration for %s : %v",
@@ -184,6 +210,29 @@ func (s *VolumeServer) Cancel(
 	return &api.SdkCloudMigrateCancelResponse{}, nil
 }
 
+func (s *VolumeServer) filterStatusResponseForPermissions(
+	ctx context.Context,
+	resp *api.CloudMigrateStatusResponse) *api.CloudMigrateStatusResponse {
+	// Filter out volumes the user doesn't have access to
+	for i, cluster := range resp.Info {
+		for j, migrateInfo := range cluster.List {
+			if err := checkAccessFromDriverForVolumeId(ctx, s.driver(ctx),
+				migrateInfo.GetLocalVolumeId(), api.Ownership_Read); err != nil {
+				// Remove entry if the user does not have access.
+				cluster.List = append(cluster.List[:j], cluster.List[j+1:]...)
+			}
+		}
+
+		// If we've removed all volume entries for a cluster,
+		// remove that cluster from the response.
+		if len(cluster.List) == 0 {
+			delete(resp.Info, i)
+		}
+	}
+
+	return resp
+}
+
 // Status of ongoing migration
 func (s *VolumeServer) Status(
 	ctx context.Context,
@@ -197,6 +246,10 @@ func (s *VolumeServer) Status(
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Cannot get status of migration : %v", err)
 	}
+
+	// Filter out volumes we don't have access to
+	resp = s.filterStatusResponseForPermissions(ctx, resp)
+
 	return &api.SdkCloudMigrateStatusResponse{
 		Result: resp,
 	}, nil

api/server/sdk/volume_migrate_test.go

@@ -301,14 +301,23 @@ func TestVolumeMigrate_CancelSuccess(t *testing.T) {
 	s := newTestServer(t)
 	defer s.Stop()
 
+	taskId := "1"
 	req := &api.SdkCloudMigrateCancelRequest{
 		Request: &api.CloudMigrateCancelRequest{
-			TaskId: "1"},
+			TaskId: taskId,
+		},
 	}
 
+	resp := &api.CloudMigrateStatusResponse{}
+	s.MockDriver().EXPECT().
+		CloudMigrateStatus(&api.CloudMigrateStatusRequest{
+			TaskId: taskId,
+		}).
+		Return(resp, nil)
+
 	s.MockDriver().EXPECT().
 		CloudMigrateCancel(&api.CloudMigrateCancelRequest{
-			TaskId: "1",
+			TaskId: taskId,
 		}).
 		Return(nil)
 	// Setup client
@@ -361,9 +370,10 @@ func TestVolumeMigrate_StatusSucess(t *testing.T) {
 	req := &api.SdkCloudMigrateStatusRequest{
 		Request: &api.CloudMigrateStatusRequest{},
 	}
+	vId := "VID"
 	info := &api.CloudMigrateInfo{
 		ClusterId:       "Source",
-		LocalVolumeId:   "VID",
+		LocalVolumeId:   vId,
 		LocalVolumeName: "VNAME",
 		RemoteVolumeId:  "RID",
 		CloudbackupId:   "CBKUPID",
@@ -383,6 +393,14 @@ func TestVolumeMigrate_StatusSucess(t *testing.T) {
 	s.MockDriver().EXPECT().
 		CloudMigrateStatus(&api.CloudMigrateStatusRequest{}).
 		Return(resp, nil)
+
+	inspectResp := &api.Volume{
+		Id: vId,
+	}
+	s.MockDriver().EXPECT().
+		Inspect([]string{vId}).
+		Return([]*api.Volume{inspectResp}, nil)
+
 	// Setup client
 	c := api.NewOpenStorageMigrateClient(s.Conn())
 	r, err := c.Status(context.Background(), req)