Auto close idle frontend connection for Proxy

RELEASE-NOTES.md

@@ -1,5 +1,3 @@
-## Release 5.5.4-SNAPSHOT
-
 ## Release 5.5.3
 
 ### CVE
@@ -9,6 +7,9 @@
 1. Fix CVE-2024-7254 [#36153](https://github.com/apache/shardingsphere/pull/36153)
 1. Fix CVE-2015-5237, CVE-2024-7254, CVE-2022-3171, CVE-2021-22569, CVE-2021-22570 [#37888](https://github.com/apache/shardingsphere/pull/37888)
 1. Fix CVE-2024-12798, CVE-2024-12801, CVE-2025-11226 [#37936](https://github.com/apache/shardingsphere/pull/37936)
+1. Fix CVE-2023-39017 [#38039](https://github.com/apache/shardingsphere/pull/38039)
+1. Fix CVE-2024-22399, CVE-2021-32824, CVE-2025-5222, CVE-2016-1000027 [#38040](https://github.com/apache/shardingsphere/pull/38040)
+1. Fix CVE-2023-2976, CVE-2024-29131, CVE-2025-27821 [#38042](https://github.com/apache/shardingsphere/pull/38042)
 
 ### Metadata Storage Changes
 
@@ -21,6 +22,7 @@
 1. Remove configuration property key `system-log-level` - [#35493](https://github.com/apache/shardingsphere/pull/35493)
 1. Change ShardingSphere SQL log topic from `ShardingSphere-SQL` to `org.apache.shardingsphere.sql` - [#37022](https://github.com/apache/shardingsphere/pull/37022)
 1. Add temporary config key `instance-connection-enabled` - [#37694](https://github.com/apache/shardingsphere/pull/37694)
+1. Add property config key `proxy-frontend-connection-idle-timeout` - [#38045](https://github.com/apache/shardingsphere/pull/38045)
 
 ### New Features
 
@@ -30,6 +32,7 @@
 1. Decouple registry center types as pluggable - [#36087](https://github.com/apache/shardingsphere/pull/36087)
 1. Proxy: Support Firebird Proxy - [#35937](https://github.com/apache/shardingsphere/pull/35937)
 1. JDBC: Support ZooKeeper and ETCD URL format - [#37037](https://github.com/apache/shardingsphere/pull/37037)
+1. Proxy: Auto close idle frontend connection for Proxy - [#38045](https://github.com/apache/shardingsphere/pull/38045)
 
 ### Enhancements
 

infra/common/src/main/java/org/apache/shardingsphere/infra/config/props/ConfigurationPropertyKey.java

@@ -95,6 +95,11 @@ public enum ConfigurationPropertyKey implements TypedPropertyKey {
      */
     PROXY_FRONTEND_MAX_CONNECTIONS("proxy-frontend-max-connections", "0", int.class, false),
 
+    /**
+     * Proxy frontend connection idle timeout in seconds.
+     */
+    PROXY_FRONTEND_CONNECTION_IDLE_TIMEOUT("proxy-frontend-connection-idle-timeout", "28800", long.class, false),
+
     /**
      * Proxy default start port.
      */

proxy/backend/core/src/test/java/org/apache/shardingsphere/proxy/backend/handler/distsql/ral/queryable/variable/ShowDistVariablesExecutorTest.java

@@ -64,7 +64,7 @@ void assertExecute() {
         executor.setConnectionContext(new DistSQLConnectionContext(mock(QueryContext.class), 1,
                 mock(DatabaseType.class), mock(DatabaseConnectionManager.class), mock(ExecutorStatementManager.class)));
         Collection<LocalDataQueryResultRow> actual = executor.getRows(mock(ShowDistVariablesStatement.class), contextManager);
-        assertThat(actual.size(), is(22));
+        assertThat(actual.size(), is(23));
         LocalDataQueryResultRow row = actual.iterator().next();
         assertThat(row.getCell(1), is("agent_plugins_enabled"));
         assertThat(row.getCell(2), is("false"));

proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/FrontendChannelInboundHandler.java

@@ -21,11 +21,15 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleStateEvent;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.shardingsphere.authentication.result.AuthenticationResult;
 import org.apache.shardingsphere.database.protocol.constant.CommonConstants;
+import org.apache.shardingsphere.infra.executor.sql.process.Process;
 import org.apache.shardingsphere.infra.executor.sql.process.ProcessEngine;
+import org.apache.shardingsphere.infra.executor.sql.process.ProcessRegistry;
 import org.apache.shardingsphere.infra.metadata.user.Grantee;
+import org.apache.shardingsphere.infra.session.connection.ConnectionContext;
 import org.apache.shardingsphere.proxy.backend.session.ConnectionSession;
 import org.apache.shardingsphere.proxy.frontend.exception.ExpectedExceptions;
 import org.apache.shardingsphere.proxy.frontend.executor.ConnectionThreadExecutorGroup;
@@ -99,6 +103,31 @@ private boolean authenticate(final ChannelHandlerContext context, final ByteBuf
         return false;
     }
 
+    @Override
+    public void userEventTriggered(final ChannelHandlerContext ctx, final Object event) throws Exception {
+        if (event instanceof IdleStateEvent) {
+            if (isIdle()) {
+                ConnectionContext connectionContext = connectionSession.getConnectionContext();
+                Grantee grantee = null == connectionContext ? null : connectionContext.getGrantee();
+                String databaseName = null == connectionContext ? null : connectionContext.getCurrentDatabaseName().orElse("NONE");
+                log.info("Connection {} (processId: {}) will be closed due to receiving an IdleStateEvent as it is idle. Grantee: {}, database name: {}",
+                        connectionSession.getConnectionId(), connectionSession.getProcessId(), grantee, databaseName);
+                ctx.close();
+            } else {
+                log.info("Received IdleStateEvent, but connection {} (processId: {}) is not idle, ignore.", connectionSession.getConnectionId(), connectionSession.getProcessId());
+            }
+        }
+        super.userEventTriggered(ctx, event);
+    }
+
+    private boolean isIdle() {
+        if (null == connectionSession.getProcessId()) {
+            return true;
+        }
+        Process process = ProcessRegistry.getInstance().get(connectionSession.getProcessId());
+        return null == process || process.isIdle();
+    }
+
     @Override
     public void channelInactive(final ChannelHandlerContext context) {
         context.fireChannelInactive();

proxy/frontend/core/src/main/java/org/apache/shardingsphere/proxy/frontend/netty/ServerHandlerInitializer.java

@@ -20,12 +20,17 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
+import io.netty.handler.timeout.IdleStateHandler;
 import lombok.RequiredArgsConstructor;
 import org.apache.shardingsphere.database.connector.core.spi.DatabaseTypedSPILoader;
 import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
 import org.apache.shardingsphere.database.protocol.codec.PacketCodec;
+import org.apache.shardingsphere.infra.config.props.ConfigurationPropertyKey;
+import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
 import org.apache.shardingsphere.proxy.frontend.spi.DatabaseProtocolFrontendEngine;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * Server handler initializer.
  */
@@ -42,7 +47,15 @@ protected void initChannel(final Channel socketChannel) {
         pipeline.addLast(new PacketCodec(databaseProtocolFrontendEngine.getCodecEngine()));
         pipeline.addLast(new FrontendChannelLimitationInboundHandler(databaseProtocolFrontendEngine));
         pipeline.addLast(ProxyFlowControlHandler.class.getSimpleName(), new ProxyFlowControlHandler());
+        addIdleStateHandlerIfNeeded(pipeline);
         pipeline.addLast(FrontendChannelInboundHandler.class.getSimpleName(), new FrontendChannelInboundHandler(databaseProtocolFrontendEngine, socketChannel));
         databaseProtocolFrontendEngine.initChannel(socketChannel);
     }
+
+    private void addIdleStateHandlerIfNeeded(final ChannelPipeline pipeline) {
+        long idleTimeout = ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData().getProps().getValue(ConfigurationPropertyKey.PROXY_FRONTEND_CONNECTION_IDLE_TIMEOUT);
+        if (0 < idleTimeout) {
+            pipeline.addLast(new IdleStateHandler(0, 0, idleTimeout, TimeUnit.SECONDS));
+        }
+    }
 }

proxy/frontend/core/src/test/java/org/apache/shardingsphere/proxy/frontend/netty/FrontendChannelInboundHandlerTest.java

@@ -22,6 +22,7 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.timeout.IdleStateEvent;
 import lombok.SneakyThrows;
 import org.apache.shardingsphere.authentication.result.AuthenticationResult;
 import org.apache.shardingsphere.authentication.result.AuthenticationResultBuilder;
@@ -30,7 +31,9 @@
 import org.apache.shardingsphere.database.exception.core.exception.SQLDialectException;
 import org.apache.shardingsphere.database.protocol.packet.DatabasePacket;
 import org.apache.shardingsphere.database.protocol.payload.PacketPayload;
+import org.apache.shardingsphere.infra.executor.sql.process.Process;
 import org.apache.shardingsphere.infra.executor.sql.process.ProcessEngine;
+import org.apache.shardingsphere.infra.executor.sql.process.ProcessRegistry;
 import org.apache.shardingsphere.infra.metadata.database.rule.RuleMetaData;
 import org.apache.shardingsphere.infra.metadata.user.Grantee;
 import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
@@ -71,15 +74,15 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockStatic;
-import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(AutoMockExtension.class)
-@StaticMockSettings(ProxyContext.class)
+@StaticMockSettings({ProxyContext.class, ProcessRegistry.class})
 class FrontendChannelInboundHandlerTest {
 
     private static final int CONNECTION_ID = 1;
@@ -266,6 +269,29 @@ void assertChannelWritabilityChangedWhenNotWritable() throws Exception {
         verify(databaseConnectionManager, never()).getConnectionResourceLock();
     }
 
+    @Test
+    void assertUserEventTriggeredWithIdleStateEventAndIdle() throws Exception {
+        channel.register();
+        ProcessRegistry processRegistry = mock(ProcessRegistry.class);
+        when(ProcessRegistry.getInstance()).thenReturn(processRegistry);
+        channel.pipeline().fireUserEventTriggered(IdleStateEvent.ALL_IDLE_STATE_EVENT);
+        assertThat(channel.isOpen(), is(false));
+    }
+
+    @Test
+    void assertUserEventTriggeredWithIdleStateEventAndNotIdle() throws Exception {
+        channel.register();
+        ProcessRegistry processRegistry = mock(ProcessRegistry.class);
+        when(ProcessRegistry.getInstance()).thenReturn(processRegistry);
+        String processId = "foo_id";
+        connectionSession.setProcessId(processId);
+        Process process = mock(Process.class);
+        when(process.isIdle()).thenReturn(false);
+        when(processRegistry.get(processId)).thenReturn(process);
+        channel.pipeline().fireUserEventTriggered(IdleStateEvent.ALL_IDLE_STATE_EVENT);
+        assertThat(channel.isOpen(), is(true));
+    }
+
     @SneakyThrows(ReflectiveOperationException.class)
     private AtomicBoolean getAuthenticated() {
         return (AtomicBoolean) Plugins.getMemberAccessor().get(FrontendChannelInboundHandler.class.getDeclaredField("authenticated"), frontendChannelInboundHandler);

proxy/frontend/core/src/test/java/org/apache/shardingsphere/proxy/frontend/netty/ServerHandlerInitializerTest.java

@@ -19,33 +19,44 @@
 
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.timeout.IdleStateHandler;
 import org.apache.shardingsphere.database.connector.core.type.DatabaseType;
 import org.apache.shardingsphere.database.protocol.codec.PacketCodec;
 import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
+import org.apache.shardingsphere.infra.config.props.ConfigurationPropertyKey;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
 import org.apache.shardingsphere.test.infra.framework.extension.mock.AutoMockExtension;
 import org.apache.shardingsphere.test.infra.framework.extension.mock.ConstructionMockSettings;
+import org.apache.shardingsphere.test.infra.framework.extension.mock.StaticMockSettings;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(AutoMockExtension.class)
 @ConstructionMockSettings(FrontendChannelInboundHandler.class)
+@StaticMockSettings(ProxyContext.class)
 class ServerHandlerInitializerTest {
 
     @Test
     void assertInitChannel() {
         SocketChannel channel = mock(SocketChannel.class);
         ChannelPipeline pipeline = mock(ChannelPipeline.class);
         when(channel.pipeline()).thenReturn(pipeline);
+        ContextManager contextManager = mock(ContextManager.class, RETURNS_DEEP_STUBS);
+        when(ProxyContext.getInstance().getContextManager()).thenReturn(contextManager);
+        when(contextManager.getMetaDataContexts().getMetaData().getProps().getValue(ConfigurationPropertyKey.PROXY_FRONTEND_CONNECTION_IDLE_TIMEOUT)).thenReturn(28800L);
         new ServerHandlerInitializer(TypedSPILoader.getService(DatabaseType.class, "FIXTURE")).initChannel(channel);
         verify(pipeline).addLast(any(ChannelAttrInitializer.class));
         verify(pipeline).addLast(any(PacketCodec.class));
         verify(pipeline).addLast(any(FrontendChannelLimitationInboundHandler.class));
+        verify(pipeline).addLast(any(IdleStateHandler.class));
         verify(pipeline).addLast(eq(ProxyFlowControlHandler.class.getSimpleName()), any(ProxyFlowControlHandler.class));
         verify(pipeline).addLast(eq(FrontendChannelInboundHandler.class.getSimpleName()), any(FrontendChannelInboundHandler.class));
     }

test/e2e/sql/src/test/resources/cases/ral/dataset/empty_rules/show_dist_variables.xml

@@ -32,6 +32,7 @@
     <row values="persist_schemas_to_repository_enabled| true"/>
     <row values="proxy_backend_query_fetch_size| -1" />
     <row values="proxy_default_port| 3307" />
+    <row values="proxy_frontend_connection_idle_timeout| 28800" />
     <row values="proxy_frontend_database_protocol_type| " />
     <row values="proxy_frontend_executor_size| 0" />
     <row values="proxy_frontend_flush_threshold| 128" />