fix(osc): exists horizontal overstep access data permission when swap table manual (#1405)

* fix Horizontal Permission on swap table manual

* modify code for comment

* Revert "modify code for comment"

This reverts commit dfa766b.

* modify code for comment

* modify error message

* dealloc flow instance

Author: krihy

server/odc-service/src/main/java/com/oceanbase/odc/service/onlineschemachange/OnlineSchemaChangeFlowableTask.java

@@ -102,6 +102,7 @@ protected Void start(Long taskId, TaskService taskService, DelegateExecution exe
         // for public cloud
         String uid = FlowTaskUtil.getCloudMainAccountId(execution);
         OnlineSchemaChangeParameters parameter = FlowTaskUtil.getOnlineSchemaChangeParameter(execution);
+        parameter.setFlowInstanceId(FlowTaskUtil.getFlowInstanceId(execution));
         ConnectionConfig connectionConfig = FlowTaskUtil.getConnectionConfig(execution);
         String schema = FlowTaskUtil.getSchemaName(execution);
         continueOnError = parameter.isContinueOnError();

server/odc-service/src/main/java/com/oceanbase/odc/service/onlineschemachange/OscService.java

@@ -16,6 +16,7 @@
 
 package com.oceanbase.odc.service.onlineschemachange;
 
+import java.util.Objects;
 import java.util.Optional;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,6 +33,7 @@
 import com.oceanbase.odc.core.shared.constant.ErrorCodes;
 import com.oceanbase.odc.core.shared.constant.ResourceType;
 import com.oceanbase.odc.core.shared.constant.TaskType;
+import com.oceanbase.odc.core.shared.exception.NotFoundException;
 import com.oceanbase.odc.metadb.connection.DatabaseEntity;
 import com.oceanbase.odc.metadb.connection.DatabaseRepository;
 import com.oceanbase.odc.metadb.schedule.ScheduleEntity;
@@ -40,6 +42,10 @@
 import com.oceanbase.odc.metadb.schedule.ScheduleTaskRepository;
 import com.oceanbase.odc.service.connection.ConnectionService;
 import com.oceanbase.odc.service.connection.model.ConnectionConfig;
+import com.oceanbase.odc.service.flow.factory.FlowFactory;
+import com.oceanbase.odc.service.flow.instance.FlowInstance;
+import com.oceanbase.odc.service.iam.HorizontalDataPermissionValidator;
+import com.oceanbase.odc.service.iam.auth.AuthenticationFacade;
 import com.oceanbase.odc.service.onlineschemachange.model.OnlineSchemaChangeParameters;
 import com.oceanbase.odc.service.onlineschemachange.model.OnlineSchemaChangeScheduleTaskResult;
 import com.oceanbase.odc.service.onlineschemachange.model.OscLockDatabaseUserInfo;
@@ -69,6 +75,13 @@ public class OscService {
     private ScheduleTaskRepository scheduleTaskRepository;
     @Autowired
     private ScheduleRepository scheduleRepository;
+    @Autowired
+    private AuthenticationFacade authenticationFacade;
+    @Autowired
+    private HorizontalDataPermissionValidator permissionValidator;
+    @Autowired
+    private FlowFactory flowFactory;
+
 
     @SkipAuthorize("internal authenticated")
     public OscLockDatabaseUserInfo getOscDatabaseInfo(@NonNull Long id) {
@@ -102,6 +115,21 @@ public OscSwapTableVO swapTable(@PathVariable Long scheduleTaskId) {
         OnlineSchemaChangeParameters oscParameters = JsonUtils.fromJson(scheduleEntity.get().getJobParametersJson(),
                 OnlineSchemaChangeParameters.class);
 
+        Optional<FlowInstance> optional = flowFactory.getFlowInstance(oscParameters.getFlowInstanceId());
+        FlowInstance flowInstance = optional.orElseThrow(
+                () -> new NotFoundException(ResourceType.ODC_FLOW_INSTANCE, "id", oscParameters.getFlowInstanceId()));
+        try {
+            permissionValidator.checkCurrentOrganization(flowInstance);
+        } finally {
+            flowInstance.dealloc();
+        }
+
+        // check user permission, only creator can swap table manual
+        PreConditions.validHasPermission(
+                Objects.equals(authenticationFacade.currentUserId(), scheduleEntity.get().getCreatorId()),
+                ErrorCodes.AccessDenied,
+                "no permission swap table.");
+
         OnlineSchemaChangeScheduleTaskResult result = JsonUtils.fromJson(scheduleTask.getResultJson(),
                 OnlineSchemaChangeScheduleTaskResult.class);
 

server/odc-service/src/main/java/com/oceanbase/odc/service/onlineschemachange/model/OnlineSchemaChangeParameters.java

@@ -40,6 +40,7 @@
  */
 @Data
 public class OnlineSchemaChangeParameters implements Serializable, TaskParameters {
+    private static final long serialVersionUID = 2870979595720162565L;
 
     private OnlineSchemaChangeSqlType sqlType;
 
@@ -56,6 +57,7 @@ public class OnlineSchemaChangeParameters implements Serializable, TaskParameter
 
     private List<String> lockUsers;
     private SwapTableType swapTableType;
+    private Long flowInstanceId;
 
     public boolean isContinueOnError() {
         return this.errorStrategy == TaskErrorStrategy.CONTINUE;