@@ -1087,42 +1087,33 @@ def multiselect_filtering(self, model, **params):
1087
1087
}
1088
1088
1089
1089
def import_services (self , ** kwargs ):
1090
- file = kwargs ["file" ]
1091
- base_path = vs .file_path / "services"
1092
- base_path .mkdir (parents = True , exist_ok = True )
1093
-
1094
- filepath = base_path / file .filename
1095
- file .save (str (filepath ))
1096
-
1097
- def is_safe_path (base_path , target_path ):
1098
- return base_path in target_path .parents or base_path == target_path
1099
-
1100
- with open_tar (filepath ) as tar_file :
1101
- for member in tar_file .getmembers ():
1102
- member_path = Path (base_path ) / member .name
1103
- member_path = member_path .resolve ()
1104
-
1105
- if not is_safe_path (base_path .resolve (), member_path ):
1106
- raise ValueError (f"Unsafe path detected: { member .name } " )
1107
-
1108
- tar_file .extractall (path = base_path )
1109
-
1110
- folder_name = tar_file .getmembers ()[0 ].name
1111
- status = self .migration_import (
1112
- folder = "services" ,
1113
- name = folder_name ,
1114
- import_export_types = ["service" , "workflow_edge" ],
1115
- service_import = True ,
1116
- skip_pool_update = True ,
1117
- skip_model_update = True ,
1118
- )
1119
-
1120
- rmtree (base_path / folder_name , ignore_errors = True )
1121
-
1122
- if "Error during import" in status :
1123
- raise Exception (status )
1124
-
1125
- return status
1090
+ file = kwargs ["file" ]
1091
+ service_path = vs .file_path / "services"
1092
+ filepath = service_path / file .filename
1093
+ service_path .mkdir (parents = True , exist_ok = True )
1094
+ file .save (str (filepath ))
1095
+ with open_tar (filepath ) as tar_file :
1096
+ for member in tar_file .getmembers ():
1097
+ member_path = (Path (service_path ) / member .name ).resolve ()
1098
+ if service_path not in member_path .parents :
1099
+ raise ValueError (
1100
+ "Unsafe path detected when importing service archive "
1101
+ f"(User: { current_user } - Path: { member_path } )"
1102
+ )
1103
+ tar_file .extractall (path = vs .file_path / "services" )
1104
+ folder_name = tar_file .getmembers ()[0 ].name
1105
+ status = self .migration_import (
1106
+ folder = "services" ,
1107
+ name = folder_name ,
1108
+ import_export_types = ["service" , "workflow_edge" ],
1109
+ service_import = True ,
1110
+ skip_pool_update = True ,
1111
+ skip_model_update = True ,
1112
+ )
1113
+ rmtree (vs .file_path / "services" / folder_name , ignore_errors = True )
1114
+ if "Error during import" in status :
1115
+ raise Exception (status )
1116
+ return status
1126
1117
1127
1118
def import_topology (self , ** kwargs ):
1128
1119
file = kwargs ["file" ]
0 commit comments