Fix stackoverflow error when installing big theme (#2399)

#### What type of PR is this?

/kind bug
/area core
/milestone 2.0

#### What this PR does / why we need it:

We will encounter a stackoverflow error when installing theme with large size. Please see the following screenshot:

![a8a3fadd7b06a071e80d4dd1899a244](https://user-images.githubusercontent.com/16865714/189171274-1940634e-a984-4ca6-857f-f8232f6a6ad3.jpg)

#### Special notes for your reviewer:

How to test?

1. Create a big theme installation package
2. Install it and see the result

#### Does this PR introduce a user-facing change?

```release-note
None
```

src/main/java/run/halo/app/core/extension/endpoint/ThemeEndpoint.java

@@ -8,7 +8,6 @@ import static org.springframework.web.reactive.function.server.RequestPredicates
 import io.swagger.v3.oas.annotations.media.Schema;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.SequenceInputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -17,13 +16,12 @@ import java.util.List;
 import java.util.function.Predicate;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipInputStream;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springdoc.core.fn.builders.schema.Builder;
 import org.springdoc.webflux.core.fn.SpringdocRouteBuilder;
-import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.io.FileSystemResource;
 import org.springframework.core.io.Resource;
-import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.multipart.FilePart;
 import org.springframework.http.codec.multipart.Part;
@@ -33,12 +31,15 @@ import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.FileSystemUtils;
 import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.BodyExtractors;
 import org.springframework.web.reactive.function.server.RouterFunction;
 import org.springframework.web.reactive.function.server.ServerRequest;
 import org.springframework.web.reactive.function.server.ServerResponse;
 import org.springframework.web.server.ServerWebInputException;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import run.halo.app.core.extension.Setting;
 import run.halo.app.core.extension.Theme;
 import run.halo.app.extension.ConfigMap;
@@ -46,6 +47,7 @@ import run.halo.app.extension.ReactiveExtensionClient;
 import run.halo.app.extension.Unstructured;
 import run.halo.app.infra.exception.ThemeInstallationException;
 import run.halo.app.infra.properties.HaloProperties;
+import run.halo.app.infra.utils.DataBufferUtils;
 import run.halo.app.infra.utils.FileUtils;
 import run.halo.app.infra.utils.YamlUnstructuredLoader;
 
@@ -55,6 +57,7 @@ import run.halo.app.infra.utils.YamlUnstructuredLoader;
  * @author guqing
  * @since 2.0.0
  */
+@Slf4j
 @Component
 public class ThemeEndpoint implements CustomEndpoint {
 
@@ -92,13 +95,21 @@ public class ThemeEndpoint implements CustomEndpoint {
     }
 
     Mono<ServerResponse> install(ServerRequest request) {
-        return request.bodyToMono(new ParameterizedTypeReference<MultiValueMap<String, Part>>() {
-            })
+        return request.body(BodyExtractors.toMultipartData())
             .flatMap(this::getZipFilePart)
-            .flatMap(file -> file.content()
-                .map(DataBuffer::asInputStream)
-                .reduce(SequenceInputStream::new)
-                .map(inputStream -> ThemeUtils.unzipThemeTo(inputStream, getThemeWorkDir())))
+            .map(file -> {
+                try {
+                    var is = DataBufferUtils.toInputStream(file.content());
+                    var themeWorkDir = getThemeWorkDir();
+                    if (log.isDebugEnabled()) {
+                        log.debug("Transferring {} into {}", file.filename(), themeWorkDir);
+                    }
+                    return ThemeUtils.unzipThemeTo(is, themeWorkDir);
+                } catch (IOException e) {
+                    throw Exceptions.propagate(e);
+                }
+            })
+            .subscribeOn(Schedulers.boundedElastic())
             .flatMap(this::persistent)
             .flatMap(theme -> ServerResponse.ok()
                 .contentType(MediaType.APPLICATION_JSON)
@@ -212,7 +223,6 @@ public class ThemeEndpoint implements CustomEndpoint {
 
                 Path themeManifestPath = resolveThemeManifest(themeTempWorkDir);
                 if (themeManifestPath == null) {
-                    FileSystemUtils.deleteRecursively(tempDirectory);
                     throw new IllegalArgumentException(
                         "It's an invalid zip format for the theme, manifest "
                             + "file [theme.yaml] is required.");

src/main/java/run/halo/app/infra/utils/DataBufferUtils.java

@@ -0,0 +1,42 @@
+package run.halo.app.infra.utils;
+
+import static org.springframework.core.io.buffer.DataBufferUtils.releaseConsumer;
+import static org.springframework.core.io.buffer.DataBufferUtils.write;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.io.buffer.DataBuffer;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+@Slf4j
+public final class DataBufferUtils {
+
+    private DataBufferUtils() {
+    }
+
+    public static InputStream toInputStream(Flux<DataBuffer> content) throws IOException {
+        var pos = new PipedOutputStream();
+        var pis = new PipedInputStream(pos);
+        write(content, pos)
+            .doOnComplete(() -> {
+                try {
+                    pos.close();
+                } catch (IOException ignored) {
+                    // Ignore the error
+                }
+            })
+            .subscribeOn(Schedulers.boundedElastic())
+            .subscribe(releaseConsumer(), error -> {
+                if (error instanceof IOException) {
+                    // Ignore the error
+                    return;
+                }
+                log.error("Failed to write DataBuffer into OutputStream", error);
+            });
+        return pis;
+    }
+}

src/main/java/run/halo/app/infra/utils/FileUtils.java

@@ -41,10 +41,7 @@ public abstract class FileUtils {
             // Resolve the entry path
             Path entryPath = targetPath.resolve(zipEntry.getName());
 
-            // Check directory
-            if (targetPath.normalize().startsWith(entryPath)) {
-                throw new IllegalArgumentException("Cannot unzip to a subdirectory of itself");
-            }
+            checkDirectoryTraversal(targetPath, entryPath);
 
             if (Files.notExists(entryPath.getParent())) {
                 Files.createDirectories(entryPath.getParent());

src/main/java/run/halo/app/infra/utils/HaloUtils.java

@@ -3,6 +3,7 @@ package run.halo.app.infra.utils;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.util.StreamUtils;
 
@@ -10,6 +11,7 @@ import org.springframework.util.StreamUtils;
  * @author guqing
  * @date 2022-04-12
  */
+@Slf4j
 public class HaloUtils {
 
     /**
@@ -28,4 +30,5 @@ public class HaloUtils {
                     location), e);
         }
     }
+
 }