websocket sync handshakes

shenyu-admin/src/main/java/org/apache/shenyu/admin/config/properties/WebsocketSyncProperties.java

@@ -40,6 +40,11 @@ public class WebsocketSyncProperties {
      */
     private String allowOrigins;
 
+    /**
+     * WebSocket sync token.
+     */
+    private String token;
+
     /**
      * Gets the value of enabled.
      *
@@ -91,4 +96,22 @@ public String getAllowOrigins() {
     public void setAllowOrigins(final String allowOrigins) {
         this.allowOrigins = allowOrigins;
     }
+
+    /**
+     * get token.
+     *
+     * @return token
+     */
+    public String getToken() {
+        return token;
+    }
+
+    /**
+     * set token.
+     *
+     * @param token token
+     */
+    public void setToken(final String token) {
+        this.token = token;
+    }
 }

shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/websocket/WebsocketConfigurator.java

@@ -26,13 +26,21 @@
 import org.apache.shenyu.admin.config.properties.WebsocketSyncProperties;
 import org.apache.shenyu.admin.spring.SpringBeanUtils;
 import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.exception.ShenyuException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.web.servlet.ServletContextInitializer;
 import org.springframework.context.annotation.Configuration;
 
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
 import static org.apache.tomcat.websocket.server.Constants.BINARY_BUFFER_SIZE_SERVLET_CONTEXT_INIT_PARAM;
 import static org.apache.tomcat.websocket.server.Constants.TEXT_BUFFER_SIZE_SERVLET_CONTEXT_INIT_PARAM;
 
@@ -52,6 +60,7 @@ public class WebsocketConfigurator extends ServerEndpointConfig.Configurator imp
 
     @Override
     public void modifyHandshake(final ServerEndpointConfig sec, final HandshakeRequest request, final HandshakeResponse response) {
+        checkSyncToken(request);
         HttpSession httpSession = (HttpSession) request.getHttpSession();
         sec.getUserProperties().put(WebsocketListener.CLIENT_IP_NAME, httpSession.getAttribute(WebsocketListener.CLIENT_IP_NAME));
         sec.getUserProperties().put(Constants.CLIENT_PORT_NAME, httpSession.getAttribute(Constants.CLIENT_PORT_NAME));
@@ -61,7 +70,7 @@ public void modifyHandshake(final ServerEndpointConfig sec, final HandshakeReque
 
     @Override
     public boolean checkOrigin(final String originHeaderValue) {
-        final WebsocketSyncProperties bean = SpringBeanUtils.getInstance().getBean(WebsocketSyncProperties.class);
+        final WebsocketSyncProperties bean = getWebsocketSyncProperties();
         if (StringUtils.isNotEmpty(bean.getAllowOrigins())) {
             String[] split = StringUtils.split(bean.getAllowOrigins(), ";");
             for (String configAllow : split) {
@@ -77,12 +86,47 @@ public boolean checkOrigin(final String originHeaderValue) {
 
     @Override
     public void onStartup(final ServletContext servletContext) {
-        int messageMaxSize = websocketSyncProperties.getMessageMaxSize();
+        int messageMaxSize = getWebsocketSyncProperties().getMessageMaxSize();
         if (messageMaxSize > 0) {
             servletContext.setInitParameter(TEXT_BUFFER_SIZE_SERVLET_CONTEXT_INIT_PARAM,
                     String.valueOf(messageMaxSize));
             servletContext.setInitParameter(BINARY_BUFFER_SIZE_SERVLET_CONTEXT_INIT_PARAM,
                     String.valueOf(messageMaxSize));
         }
     }
+
+    private void checkSyncToken(final HandshakeRequest request) {
+        String configuredToken = getWebsocketSyncProperties().getToken();
+        if (StringUtils.isBlank(configuredToken)) {
+            throw new ShenyuException("websocket sync token is not configured");
+        }
+        String requestToken = getHeader(request.getHeaders(), Constants.X_SHENYU_SYNC_TOKEN);
+        if (StringUtils.isBlank(requestToken) || !isSameToken(configuredToken, requestToken)) {
+            throw new ShenyuException("websocket sync token is invalid");
+        }
+    }
+
+    private WebsocketSyncProperties getWebsocketSyncProperties() {
+        return Optional.ofNullable(websocketSyncProperties)
+                .orElseGet(() -> SpringBeanUtils.getInstance().getBean(WebsocketSyncProperties.class));
+    }
+
+    private boolean isSameToken(final String configuredToken, final String requestToken) {
+        return MessageDigest.isEqual(
+                configuredToken.getBytes(StandardCharsets.UTF_8),
+                requestToken.getBytes(StandardCharsets.UTF_8));
+    }
+
+    private String getHeader(final Map<String, List<String>> headers, final String name) {
+        return Optional.ofNullable(headers)
+                .orElse(Collections.emptyMap())
+                .entrySet()
+                .stream()
+                .filter(entry -> StringUtils.equalsIgnoreCase(entry.getKey(), name))
+                .map(Map.Entry::getValue)
+                .filter(values -> !values.isEmpty())
+                .map(values -> values.get(0))
+                .findFirst()
+                .orElse(null);
+    }
 }

shenyu-admin/src/main/resources/application.yml

@@ -70,6 +70,7 @@ shenyu:
     websocket:
       enabled: true
       messageMaxSize: 10240
+      token: ${SHENYU_SYNC_WEBSOCKET_TOKEN:}
       allowOrigins: ws://localhost:9095;ws://localhost:9195;
 #    apollo:
 #      meta: http://localhost:8080

shenyu-admin/src/test/java/org/apache/shenyu/admin/config/properties/WebsocketSyncPropertiesTest.java

@@ -43,9 +43,11 @@ public void testWebsocketSyncPropertiesSetValue() {
         WebsocketSyncProperties websocketSyncProperties = getContext().getBean(WebsocketSyncProperties.class);
         websocketSyncProperties.setMessageMaxSize(0);
         websocketSyncProperties.setAllowOrigins("allowOrigins");
+        websocketSyncProperties.setToken("token");
         assertThat(websocketSyncProperties.isEnabled(), comparesEqualTo(false));
         Assertions.assertEquals(websocketSyncProperties.getMessageMaxSize(), 0);
         Assertions.assertEquals(websocketSyncProperties.getAllowOrigins(), "allowOrigins");
+        Assertions.assertEquals(websocketSyncProperties.getToken(), "token");
     }
 
     @Configuration

shenyu-admin/src/test/java/org/apache/shenyu/admin/listener/websocket/WebsocketConfiguratorTest.java

@@ -25,18 +25,22 @@
 import org.apache.shenyu.admin.config.properties.WebsocketSyncProperties;
 import org.apache.shenyu.admin.spring.SpringBeanUtils;
 import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.exception.ShenyuException;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.context.ApplicationContext;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import static org.apache.tomcat.websocket.server.Constants.BINARY_BUFFER_SIZE_SERVLET_CONTEXT_INIT_PARAM;
 import static org.apache.tomcat.websocket.server.Constants.TEXT_BUFFER_SIZE_SERVLET_CONTEXT_INIT_PARAM;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -64,11 +68,13 @@ void setUp() {
 
     @Test
     void testModifyHandshake() {
+        websocketSyncProperties.setToken("websocket-sync-token");
         ServerEndpointConfig sec = mock(ServerEndpointConfig.class);
         Map<String, Object> userProperties = new HashMap<>();
         when(sec.getUserProperties()).thenReturn(userProperties);
 
         HandshakeRequest request = mock(HandshakeRequest.class);
+        when(request.getHeaders()).thenReturn(Collections.singletonMap(Constants.X_SHENYU_SYNC_TOKEN, List.of("websocket-sync-token")));
         HttpSession httpSession = mock(HttpSession.class);
         when(request.getHttpSession()).thenReturn(httpSession);
         when(httpSession.getAttribute(WebsocketListener.CLIENT_IP_NAME)).thenReturn("192.168.1.1");
@@ -85,11 +91,13 @@ void testModifyHandshake() {
 
     @Test
     void testModifyHandshakePutsAllAttributes() {
+        websocketSyncProperties.setToken("websocket-sync-token");
         ServerEndpointConfig sec = mock(ServerEndpointConfig.class);
         Map<String, Object> userProperties = new HashMap<>();
         when(sec.getUserProperties()).thenReturn(userProperties);
 
         HandshakeRequest request = mock(HandshakeRequest.class);
+        when(request.getHeaders()).thenReturn(Collections.singletonMap(Constants.X_SHENYU_SYNC_TOKEN, List.of("websocket-sync-token")));
         HttpSession httpSession = mock(HttpSession.class);
         when(request.getHttpSession()).thenReturn(httpSession);
 
@@ -104,6 +112,59 @@ void testModifyHandshakePutsAllAttributes() {
         assertTrue(userProperties.containsKey(Constants.SHENYU_NAMESPACE_ID));
     }
 
+    @Test
+    void testModifyHandshakeUsesSpringBeanWhenConfiguratorIsNotAutowired() {
+        websocketSyncProperties.setToken("websocket-sync-token");
+        ReflectionTestUtils.setField(websocketConfigurator, "websocketSyncProperties", null);
+        ServerEndpointConfig sec = mock(ServerEndpointConfig.class);
+        Map<String, Object> userProperties = new HashMap<>();
+        when(sec.getUserProperties()).thenReturn(userProperties);
+
+        HandshakeRequest request = mock(HandshakeRequest.class);
+        when(request.getHeaders()).thenReturn(Collections.singletonMap(Constants.X_SHENYU_SYNC_TOKEN, List.of("websocket-sync-token")));
+        when(request.getHttpSession()).thenReturn(mock(HttpSession.class));
+
+        HandshakeResponse response = mock(HandshakeResponse.class);
+        websocketConfigurator.modifyHandshake(sec, request, response);
+
+        assertTrue(userProperties.containsKey(WebsocketListener.CLIENT_IP_NAME));
+        assertTrue(userProperties.containsKey(Constants.CLIENT_PORT_NAME));
+        assertTrue(userProperties.containsKey(Constants.SHENYU_NAMESPACE_ID));
+    }
+
+    @Test
+    void testModifyHandshakeRejectsMissingToken() {
+        websocketSyncProperties.setToken("websocket-sync-token");
+        ServerEndpointConfig sec = mock(ServerEndpointConfig.class);
+        HandshakeRequest request = mock(HandshakeRequest.class);
+        when(request.getHeaders()).thenReturn(Collections.emptyMap());
+        HandshakeResponse response = mock(HandshakeResponse.class);
+
+        assertThrows(ShenyuException.class, () -> websocketConfigurator.modifyHandshake(sec, request, response));
+    }
+
+    @Test
+    void testModifyHandshakeRejectsInvalidToken() {
+        websocketSyncProperties.setToken("websocket-sync-token");
+        ServerEndpointConfig sec = mock(ServerEndpointConfig.class);
+        HandshakeRequest request = mock(HandshakeRequest.class);
+        when(request.getHeaders()).thenReturn(Collections.singletonMap(Constants.X_SHENYU_SYNC_TOKEN, List.of("invalid-token")));
+        HandshakeResponse response = mock(HandshakeResponse.class);
+
+        assertThrows(ShenyuException.class, () -> websocketConfigurator.modifyHandshake(sec, request, response));
+    }
+
+    @Test
+    void testModifyHandshakeRejectsBlankConfiguredToken() {
+        websocketSyncProperties.setToken("");
+        ServerEndpointConfig sec = mock(ServerEndpointConfig.class);
+        HandshakeRequest request = mock(HandshakeRequest.class);
+        when(request.getHeaders()).thenReturn(Collections.singletonMap(Constants.X_SHENYU_SYNC_TOKEN, List.of("websocket-sync-token")));
+        HandshakeResponse response = mock(HandshakeResponse.class);
+
+        assertThrows(ShenyuException.class, () -> websocketConfigurator.modifyHandshake(sec, request, response));
+    }
+
     @Test
     void testCheckOriginAllowedWhenAllowOriginsEmpty() {
         websocketSyncProperties.setAllowOrigins("");

shenyu-bootstrap/src/main/resources/application.yml

@@ -223,6 +223,7 @@ shenyu:
   sync:
     websocket:
       urls: ws://localhost:9095/websocket
+      token: ${SHENYU_SYNC_WEBSOCKET_TOKEN:}
       allowOrigin: ws://localhost:9195
 #    apollo:
 #      appId: shenyu

shenyu-common/src/main/java/org/apache/shenyu/common/constant/Constants.java

@@ -713,6 +713,11 @@ public interface Constants {
      */
     String X_ACCESS_TOKEN = "X-Access-Token";
 
+    /**
+     * X-Shenyu-Sync-Token.
+     */
+    String X_SHENYU_SYNC_TOKEN = "X-Shenyu-Sync-Token";
+
     /**
      * X-API-KEY; AI proxy key header.
      */

shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml

@@ -31,6 +31,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/mysql/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -51,6 +52,7 @@ services:
     environment:
       - TZ=Asia/Beijing
       - shenyu.sync.websocket.urls=ws://shenyu-admin:9095/websocket
+      - shenyu.sync.websocket.token=shenyu-sync-token
       - shenyu.sync.websocket.allowOrigin=ws://localhost:9195
     healthcheck:
       test: [ "CMD-SHELL", "wget -q -O - http://shenyu-bootstrap:9195/actuator/health | grep UP || exit 1" ]

shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml

@@ -63,6 +63,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/mysql/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -86,6 +87,7 @@ services:
     environment:
       - TZ=Asia/Beijing
       - shenyu.sync.websocket.urls=ws://shenyu-admin:9095/websocket
+      - shenyu.sync.websocket.token=shenyu-sync-token
       - shenyu.sync.websocket.allowOrigin=ws://localhost:9195
     healthcheck:
       test: [ "CMD-SHELL", "wget -q -O - http://shenyu-bootstrap:9195/actuator/health | grep UP || exit 1" ]

shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml

@@ -62,6 +62,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/opengauss/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -85,6 +86,7 @@ services:
     environment:
       - TZ=Asia/Beijing
       - shenyu.sync.websocket.urls=ws://shenyu-admin:9095/websocket
+      - shenyu.sync.websocket.token=shenyu-sync-token
       - shenyu.sync.websocket.allowOrigin=ws://localhost:9195
     healthcheck:
       test: [ "CMD-SHELL", "wget -q -O - http://shenyu-bootstrap:9195/actuator/health | grep UP || exit 1" ]

shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml

@@ -62,6 +62,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/postgres/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -85,6 +86,7 @@ services:
     environment:
       - TZ=Asia/Beijing
       - shenyu.sync.websocket.urls=ws://shenyu-admin:9095/websocket
+      - shenyu.sync.websocket.token=shenyu-sync-token
       - shenyu.sync.websocket.allowOrigin=ws://localhost:9195
     healthcheck:
       test: [ "CMD-SHELL", "wget -q -O - http://shenyu-bootstrap:9195/actuator/health | grep UP || exit 1" ]

shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml

@@ -64,6 +64,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/mysql/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -96,6 +97,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/mysql/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -119,6 +121,7 @@ services:
     environment:
       - TZ=Asia/Beijing
       - shenyu.sync.websocket.urls=ws://shenyu-admin-1:9095/websocket,ws://shenyu-admin-2:9095/websocket
+      - shenyu.sync.websocket.token=shenyu-sync-token
       - shenyu.sync.websocket.allowOrigin=ws://localhost:9195
     healthcheck:
       test: [ "CMD-SHELL", "wget -q -O - http://shenyu-bootstrap:9195/actuator/health | grep UP || exit 1" ]

shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml

@@ -83,6 +83,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/mysql/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -119,6 +120,7 @@ services:
       - shenyu.sync.websocket.enabled=true
       - shenyu.sync.websocket.messageMaxSize=10240
       - shenyu.sync.websocket.allowOrigins=ws://localhost:9095;ws://localhost:9195;
+      - shenyu.sync.websocket.token=shenyu-sync-token
     volumes:
       - /tmp/shenyu-e2e/mysql/driver:/opt/shenyu-admin/ext-lib
     healthcheck:
@@ -144,6 +146,7 @@ services:
     environment:
       - TZ=Asia/Beijing
       - shenyu.sync.websocket.urls=ws://shenyu-admin-1:9095/websocket,ws://shenyu-admin-2:9095/websocket
+      - shenyu.sync.websocket.token=shenyu-sync-token
       - shenyu.sync.websocket.allowOrigin=ws://localhost:9195
     healthcheck:
       test: [ "CMD-SHELL", "wget -q -O - http://shenyu-bootstrap:9195/actuator/health | grep UP || exit 1" ]