Exclude WebSocket request when serving console index (#4096)

#### What type of PR is this?

/kind bug
/area core

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

This PR excludes WebSocket request when serving console index and remove request predicate accept in ConsoleProxyFilter.

#### Which issue(s) this PR fixes:

Fixes https://github.com/halo-dev/halo/issues/4083

#### Special notes for your reviewer:

1. Start Console with dev environment
2. Start Halo with dev profile
3. Try to browse <http://localhost:8090/console> and check the log

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

```release-note
修复开发环境下访问 Console 出现错误的问题
```

application/src/main/java/run/halo/app/config/WebFluxConfig.java

@@ -37,6 +37,7 @@ import org.springframework.web.reactive.result.view.ViewResolutionResultHandler;
 import org.springframework.web.reactive.result.view.ViewResolver;
 import reactor.core.publisher.Mono;
 import run.halo.app.console.ConsoleProxyFilter;
+import run.halo.app.console.WebSocketRequestPredicate;
 import run.halo.app.core.extension.endpoint.CustomEndpoint;
 import run.halo.app.core.extension.endpoint.CustomEndpointsBuilder;
 import run.halo.app.infra.properties.HaloProperties;
@@ -101,7 +102,8 @@ public class WebFluxConfig implements WebFluxConfigurer {
     RouterFunction<ServerResponse> consoleIndexRedirection() {
         var consolePredicate = method(HttpMethod.GET)
             .and(path("/console/**").and(path("/console/assets/**").negate()))
-            .and(accept(MediaType.TEXT_HTML));
+            .and(accept(MediaType.TEXT_HTML))
+            .and(new WebSocketRequestPredicate().negate());
         return route(consolePredicate, this::serveConsoleIndex);
     }
 

application/src/main/java/run/halo/app/console/ConsoleProxyFilter.java

@@ -1,14 +1,12 @@
 package run.halo.app.console;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.HttpMethod;
-import org.springframework.http.MediaType;
 import org.springframework.security.web.server.util.matcher.AndServerWebExchangeMatcher;
-import org.springframework.security.web.server.util.matcher.MediaTypeServerWebExchangeMatcher;
 import org.springframework.security.web.server.util.matcher.NegatedServerWebExchangeMatcher;
 import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatcher;
 import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatchers;
-import org.springframework.web.reactive.function.BodyExtractors;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.server.ServerWebExchange;
 import org.springframework.web.server.WebFilter;
@@ -30,8 +28,6 @@ public class ConsoleProxyFilter implements WebFilter {
     public ConsoleProxyFilter(HaloProperties haloProperties) {
         this.proxyProperties = haloProperties.getConsole().getProxy();
         var consoleMatcher = ServerWebExchangeMatchers.pathMatchers(HttpMethod.GET, "/console/**");
-        consoleMatcher = new AndServerWebExchangeMatcher(consoleMatcher,
-            new MediaTypeServerWebExchangeMatcher(MediaType.TEXT_HTML));
         consoleMatcher = new AndServerWebExchangeMatcher(consoleMatcher,
             new NegatedServerWebExchangeMatcher(new WebSocketServerWebExchangeMatcher()));
         this.consoleMatcher = consoleMatcher;
@@ -53,8 +49,8 @@ public class ConsoleProxyFilter implements WebFilter {
                     .toUriString();
             })
             .doOnNext(uri -> {
-                if (log.isDebugEnabled()) {
-                    log.debug("Proxy {} to {}", uri, proxyProperties.getEndpoint());
+                if (log.isTraceEnabled()) {
+                    log.trace("Proxy {} to {}", uri, proxyProperties.getEndpoint());
                 }
             })
             .flatMap(uri -> webClient.get()
@@ -68,8 +64,8 @@ public class ConsoleProxyFilter implements WebFilter {
                     response.getCookies().putAll(clientResponse.cookies());
                     // set status code
                     response.setStatusCode(clientResponse.statusCode());
-                    var body = clientResponse.body(BodyExtractors.toDataBuffers());
-                    return exchange.getResponse().writeAndFlushWith(Mono.just(body));
+                    var body = clientResponse.bodyToFlux(DataBuffer.class);
+                    return exchange.getResponse().writeWith(body);
                 }));
     }
 }

application/src/main/java/run/halo/app/console/WebSocketRequestPredicate.java

@@ -0,0 +1,15 @@
+package run.halo.app.console;
+
+import static run.halo.app.console.WebSocketUtils.isWebSocketUpgrade;
+
+import org.springframework.web.reactive.function.server.RequestPredicate;
+import org.springframework.web.reactive.function.server.ServerRequest;
+
+public class WebSocketRequestPredicate implements RequestPredicate {
+
+    @Override
+    public boolean test(ServerRequest request) {
+        var httpHeaders = request.exchange().getRequest().getHeaders();
+        return isWebSocketUpgrade(httpHeaders);
+    }
+}

application/src/main/java/run/halo/app/console/WebSocketServerWebExchangeMatcher.java

@@ -1,5 +1,9 @@
 package run.halo.app.console;
 
+import static org.springframework.security.web.server.util.matcher.ServerWebExchangeMatcher.MatchResult.match;
+import static org.springframework.security.web.server.util.matcher.ServerWebExchangeMatcher.MatchResult.notMatch;
+import static run.halo.app.console.WebSocketUtils.isWebSocketUpgrade;
+
 import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatcher;
 import org.springframework.web.server.ServerWebExchange;
 import reactor.core.publisher.Mono;
@@ -7,14 +11,6 @@ import reactor.core.publisher.Mono;
 public class WebSocketServerWebExchangeMatcher implements ServerWebExchangeMatcher {
     @Override
     public Mono<MatchResult> matches(ServerWebExchange exchange) {
-        var headers = exchange.getRequest().getHeaders();
-        if (!headers.getConnection().contains("Upgrade")) {
-            return MatchResult.notMatch();
-        }
-        var upgrade = headers.getUpgrade();
-        if (!"websocket".equalsIgnoreCase(upgrade)) {
-            return MatchResult.notMatch();
-        }
-        return MatchResult.match();
+        return isWebSocketUpgrade(exchange.getRequest().getHeaders()) ? match() : notMatch();
     }
 }

application/src/main/java/run/halo/app/console/WebSocketUtils.java

@@ -0,0 +1,16 @@
+package run.halo.app.console;
+
+import org.springframework.http.HttpHeaders;
+
+public enum WebSocketUtils {
+    ;
+
+    public static boolean isWebSocketUpgrade(HttpHeaders headers) {
+        // See io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionUtil
+        // .isWebsocketUpgrade for more.
+        return headers.containsKey(HttpHeaders.UPGRADE)
+            && headers.getConnection().contains(HttpHeaders.UPGRADE)
+            && "websocket".equalsIgnoreCase(headers.getUpgrade());
+    }
+
+}

application/src/test/java/run/halo/app/console/WebSocketUtilsTest.java

@@ -0,0 +1,51 @@
+package run.halo.app.console;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpHeaders;
+
+class WebSocketUtilsTest {
+
+    @Nested
+    class IsWebSocketTest {
+
+        @Test
+        void shouldBeWebSocketIfHeadersContaining() {
+            var headers = new HttpHeaders();
+            headers.add("Connection", "Upgrade");
+            headers.add("Upgrade", "websocket");
+            assertTrue(WebSocketUtils.isWebSocketUpgrade(headers));
+        }
+
+        @Test
+        void shouldNotBeWebSocketIfHeaderValuesAreIncorrect() {
+            var headers = new HttpHeaders();
+            headers.add("Connection", "keep-alive");
+            headers.add("Upgrade", "websocket");
+            assertFalse(WebSocketUtils.isWebSocketUpgrade(headers));
+        }
+
+        @Test
+        void shouldNotBeWebSocketIfMissingUpgradeHeader() {
+            var headers = new HttpHeaders();
+            headers.add("Connection", "Upgrade");
+            assertFalse(WebSocketUtils.isWebSocketUpgrade(headers));
+        }
+
+        @Test
+        void shouldNotBeWebSocketIfMissingConnectionHeader() {
+            var headers = new HttpHeaders();
+            headers.add("Connection", "Upgrade");
+            assertFalse(WebSocketUtils.isWebSocketUpgrade(headers));
+        }
+
+        @Test
+        void shouldNotBeWebSocketIfMissingHeaders() {
+            var headers = new HttpHeaders();
+            assertFalse(WebSocketUtils.isWebSocketUpgrade(headers));
+        }
+    }
+}