From 8204f6e5c7e5c83d477d184bb5071afcc45bc505 Mon Sep 17 00:00:00 2001 From: yungyu16 Date: Tue, 11 Nov 2025 13:30:45 +0800 Subject: [PATCH] fix: enhance token server request handling and add max frame length validation --- .../cluster/server/NettyTransportServer.java | 58 +++--- .../cluster/server/ServerConstants.java | 7 +- .../data/ParamFlowRequestDataDecoder.java | 77 +++++-- .../codec/data/PingRequestDataDecoder.java | 22 +- .../server/handler/TokenServerHandler.java | 10 + .../data/ParamFlowRequestDataDecoderTest.java | 192 ++++++++++++++++++ .../data/PingRequestDataDecoderTest.java | 105 ++++++++++ .../handler/TokenServerHandlerTest.java | 65 ++++++ 8 files changed, 478 insertions(+), 58 deletions(-) create mode 100644 sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoderTest.java create mode 100644 sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoderTest.java create mode 100644 sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandlerTest.java diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/NettyTransportServer.java b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/NettyTransportServer.java index 25468551f0..cf50d12cce 100644 --- a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/NettyTransportServer.java +++ b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/NettyTransportServer.java @@ -15,17 +15,12 @@ */ package com.alibaba.csp.sentinel.cluster.server; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; - import com.alibaba.csp.sentinel.cluster.server.codec.netty.NettyRequestDecoder; import com.alibaba.csp.sentinel.cluster.server.codec.netty.NettyResponseEncoder; import com.alibaba.csp.sentinel.cluster.server.connection.Connection; import com.alibaba.csp.sentinel.cluster.server.connection.ConnectionPool; import com.alibaba.csp.sentinel.cluster.server.handler.TokenServerHandler; import com.alibaba.csp.sentinel.log.RecordLog; - import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.PooledByteBufAllocator; import io.netty.channel.ChannelFuture; @@ -42,7 +37,14 @@ import io.netty.util.concurrent.GenericFutureListener; import io.netty.util.internal.SystemPropertyUtil; -import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.*; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.NETTY_MAX_FRAME_LENGTH; +import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.SERVER_STATUS_OFF; +import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.SERVER_STATUS_STARTED; +import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.SERVER_STATUS_STARTING; /** * @author Eric Zhao @@ -51,7 +53,7 @@ public class NettyTransportServer implements ClusterTokenServer { private static final int DEFAULT_EVENT_LOOP_THREADS = Math.max(1, - SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2)); + SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2)); private static final int MAX_RETRY_TIMES = 3; private static final int RETRY_SLEEP_MS = 2000; @@ -79,32 +81,32 @@ public void start() { this.bossGroup = new NioEventLoopGroup(1); this.workerGroup = new NioEventLoopGroup(DEFAULT_EVENT_LOOP_THREADS); b.group(bossGroup, workerGroup) - .channel(NioServerSocketChannel.class) - .option(ChannelOption.SO_BACKLOG, 128) - .handler(new LoggingHandler(LogLevel.INFO)) - .childHandler(new ChannelInitializer() { - @Override - public void initChannel(SocketChannel ch) throws Exception { - ChannelPipeline p = ch.pipeline(); - p.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2)); - p.addLast(new NettyRequestDecoder()); - p.addLast(new LengthFieldPrepender(2)); - p.addLast(new NettyResponseEncoder()); - p.addLast(new TokenServerHandler(connectionPool)); - } - }) - .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) - .childOption(ChannelOption.SO_SNDBUF, 32 * 1024) - .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) - .childOption(ChannelOption.SO_TIMEOUT, 10) - .childOption(ChannelOption.TCP_NODELAY, true) - .childOption(ChannelOption.SO_RCVBUF, 32 * 1024); + .channel(NioServerSocketChannel.class) + .option(ChannelOption.SO_BACKLOG, 128) + .handler(new LoggingHandler(LogLevel.INFO)) + .childHandler(new ChannelInitializer() { + @Override + public void initChannel(SocketChannel ch) throws Exception { + ChannelPipeline p = ch.pipeline(); + p.addLast(new LengthFieldBasedFrameDecoder(NETTY_MAX_FRAME_LENGTH, 0, 2, 0, 2)); + p.addLast(new NettyRequestDecoder()); + p.addLast(new LengthFieldPrepender(2)); + p.addLast(new NettyResponseEncoder()); + p.addLast(new TokenServerHandler(connectionPool)); + } + }) + .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) + .childOption(ChannelOption.SO_SNDBUF, 32 * 1024) + .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) + .childOption(ChannelOption.SO_TIMEOUT, 10) + .childOption(ChannelOption.TCP_NODELAY, true) + .childOption(ChannelOption.SO_RCVBUF, 32 * 1024); b.bind(port).addListener(new GenericFutureListener() { @Override public void operationComplete(ChannelFuture future) { if (future.cause() != null) { RecordLog.info("[NettyTransportServer] Token server start failed (port=" + port + "), failedTimes: " + failedTimes.get(), - future.cause()); + future.cause()); currentState.compareAndSet(SERVER_STATUS_STARTING, SERVER_STATUS_OFF); int failCount = failedTimes.incrementAndGet(); if (failCount > MAX_RETRY_TIMES) { diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/ServerConstants.java b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/ServerConstants.java index 6569cef8cd..04da6b5d01 100644 --- a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/ServerConstants.java +++ b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/ServerConstants.java @@ -27,5 +27,10 @@ public final class ServerConstants { public static final String DEFAULT_NAMESPACE = "default"; - private ServerConstants() {} + public static final int NETTY_MAX_FRAME_LENGTH = 1024; + public static final int MAX_PARAM_AMOUNT = 512; + public static final int MAX_PARAM_STRING_LENGTH = 1024; + + private ServerConstants() { + } } diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoder.java b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoder.java index fd4eab6f6f..03dc209a40 100644 --- a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoder.java +++ b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoder.java @@ -16,13 +16,16 @@ package com.alibaba.csp.sentinel.cluster.server.codec.data; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import com.alibaba.csp.sentinel.cluster.ClusterConstants; import com.alibaba.csp.sentinel.cluster.codec.EntityDecoder; import com.alibaba.csp.sentinel.cluster.request.data.ParamFlowRequestData; +import com.alibaba.csp.sentinel.cluster.server.ServerConstants; import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.CorruptedFrameException; /** * @author jialiang.linjl @@ -31,61 +34,93 @@ */ public class ParamFlowRequestDataDecoder implements EntityDecoder { + private static final int REQUEST_HEADER_LENGTH = Long.BYTES + Integer.BYTES + Integer.BYTES; + private static final int MIN_PARAM_LENGTH = Byte.BYTES + Byte.BYTES; + @Override public ParamFlowRequestData decode(ByteBuf source) { - if (source.readableBytes() >= 16) { - ParamFlowRequestData requestData = new ParamFlowRequestData() + ensureReadable(source, REQUEST_HEADER_LENGTH); + + ParamFlowRequestData requestData = new ParamFlowRequestData() .setFlowId(source.readLong()) .setCount(source.readInt()); - int amount = source.readInt(); - if (amount > 0) { - List params = new ArrayList<>(amount); - for (int i = 0; i < amount; i++) { - decodeParam(source, params); - } + int amount = source.readInt(); + if (amount < 0 || amount > ServerConstants.MAX_PARAM_AMOUNT + || amount > source.readableBytes() / MIN_PARAM_LENGTH) { + throw new CorruptedFrameException("Invalid parameter amount: " + amount); + } - requestData.setParams(params); - return requestData; + List params; + if (amount == 0) { + params = Collections.emptyList(); + } else { + params = new ArrayList<>(Math.min(amount, 16)); + for (int i = 0; i < amount; i++) { + decodeParam(source, params); } } - return null; + if (source.isReadable()) { + throw new CorruptedFrameException("Parameter flow payload contains trailing bytes: " + + source.readableBytes()); + } + return requestData.setParams(params); } - private boolean decodeParam(ByteBuf source, List params) { + private void decodeParam(ByteBuf source, List params) { + ensureReadable(source, Byte.BYTES); byte paramType = source.readByte(); switch (paramType) { case ClusterConstants.PARAM_TYPE_INTEGER: + ensureReadable(source, Integer.BYTES); params.add(source.readInt()); - return true; + return; case ClusterConstants.PARAM_TYPE_STRING: + ensureReadable(source, Integer.BYTES); int length = source.readInt(); + if (length < 0 || length > ServerConstants.MAX_PARAM_STRING_LENGTH + || length > source.readableBytes()) { + throw new CorruptedFrameException("Invalid string parameter length: " + length); + } byte[] bytes = new byte[length]; source.readBytes(bytes); // TODO: take care of charset? params.add(new String(bytes)); - return true; + return; case ClusterConstants.PARAM_TYPE_BOOLEAN: + ensureReadable(source, Byte.BYTES); params.add(source.readBoolean()); - return true; + return; case ClusterConstants.PARAM_TYPE_DOUBLE: + ensureReadable(source, Double.BYTES); params.add(source.readDouble()); - return true; + return; case ClusterConstants.PARAM_TYPE_LONG: + ensureReadable(source, Long.BYTES); params.add(source.readLong()); - return true; + return; case ClusterConstants.PARAM_TYPE_FLOAT: + ensureReadable(source, Float.BYTES); params.add(source.readFloat()); - return true; + return; case ClusterConstants.PARAM_TYPE_BYTE: + ensureReadable(source, Byte.BYTES); params.add(source.readByte()); - return true; + return; case ClusterConstants.PARAM_TYPE_SHORT: + ensureReadable(source, Short.BYTES); params.add(source.readShort()); - return true; + return; default: - return false; + throw new CorruptedFrameException("Unknown parameter type: " + paramType); + } + } + + private void ensureReadable(ByteBuf source, int requiredBytes) { + if (source.readableBytes() < requiredBytes) { + throw new CorruptedFrameException("Incomplete parameter flow payload: required=" + requiredBytes + + ", actual=" + source.readableBytes()); } } } diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoder.java b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoder.java index cef33119bb..69e9ff3aca 100644 --- a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoder.java +++ b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoder.java @@ -18,6 +18,7 @@ import com.alibaba.csp.sentinel.cluster.codec.EntityDecoder; import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.CorruptedFrameException; /** * @author Eric Zhao @@ -27,14 +28,19 @@ public class PingRequestDataDecoder implements EntityDecoder { @Override public String decode(ByteBuf source) { - if (source.readableBytes() >= 4) { - int length = source.readInt(); - if (length > 0 && source.readableBytes() > 0) { - byte[] bytes = new byte[length]; - source.readBytes(bytes); - return new String(bytes); - } + if (source.readableBytes() < Integer.BYTES) { + throw new CorruptedFrameException("Incomplete ping payload length"); } - return null; + + int packetLen = source.readInt(); + int actualLength = source.readableBytes(); + if (packetLen < 0 || packetLen != actualLength) { + throw new CorruptedFrameException("Invalid ping payload length: declared=" + packetLen + + ", actual=" + actualLength); + } + + byte[] bytes = new byte[packetLen]; + source.readBytes(bytes); + return new String(bytes); } } diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandler.java b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandler.java index 44f5d6691c..a0c1618eb4 100644 --- a/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandler.java +++ b/sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandler.java @@ -29,6 +29,8 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.TooLongFrameException; /** * Netty server handler for Sentinel token server. @@ -81,6 +83,14 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception } } + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + if (!(cause instanceof CorruptedFrameException) && !(cause instanceof TooLongFrameException)) { + RecordLog.warn("[TokenServerHandler] Unexpected exception", cause); + } + ctx.close(); + } + private void writeBadResponse(ChannelHandlerContext ctx, ClusterRequest request) { ClusterResponse response = new ClusterResponse<>(request.getId(), request.getType(), ClusterConstants.RESPONSE_STATUS_BAD, null); diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoderTest.java b/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoderTest.java new file mode 100644 index 0000000000..2ea2e2e025 --- /dev/null +++ b/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoderTest.java @@ -0,0 +1,192 @@ +/* + * Copyright 1999-2026 Alibaba Group Holding Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.alibaba.csp.sentinel.cluster.server.codec.data; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; + +import com.alibaba.csp.sentinel.cluster.ClusterConstants; +import com.alibaba.csp.sentinel.cluster.request.data.ParamFlowRequestData; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.CorruptedFrameException; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Test cases for {@link ParamFlowRequestDataDecoder}. + */ +public class ParamFlowRequestDataDecoderTest { + + private final ParamFlowRequestDataDecoder decoder = new ParamFlowRequestDataDecoder(); + + @Test + public void testDecodeAllSupportedParamTypes() { + ByteBuf source = requestHeader(8) + .writeByte(ClusterConstants.PARAM_TYPE_INTEGER).writeInt(1) + .writeByte(ClusterConstants.PARAM_TYPE_STRING).writeInt(8) + .writeBytes("Sentinel".getBytes(StandardCharsets.UTF_8)) + .writeByte(ClusterConstants.PARAM_TYPE_BOOLEAN).writeBoolean(true) + .writeByte(ClusterConstants.PARAM_TYPE_DOUBLE).writeDouble(2.5D) + .writeByte(ClusterConstants.PARAM_TYPE_LONG).writeLong(3L) + .writeByte(ClusterConstants.PARAM_TYPE_FLOAT).writeFloat(4.5F) + .writeByte(ClusterConstants.PARAM_TYPE_BYTE).writeByte(5) + .writeByte(ClusterConstants.PARAM_TYPE_SHORT).writeShort(6); + try { + ParamFlowRequestData result = decoder.decode(source); + + assertThat(result.getFlowId()).isEqualTo(1L); + assertThat(result.getCount()).isEqualTo(2); + assertThat(result.getParams()).containsExactly(1, "Sentinel", true, 2.5D, 3L, 4.5F, (byte)5, (short)6); + assertThat(source.isReadable()).isFalse(); + } finally { + source.release(); + } + } + + @Test + public void testDecodeEmptyParams() { + ByteBuf source = requestHeader(0); + try { + ParamFlowRequestData result = decoder.decode(source); + + assertThat(result.getParams()).isEqualTo(Collections.emptyList()); + assertThat(source.isReadable()).isFalse(); + } finally { + source.release(); + } + } + + @Test + public void testRejectNegativeAmount() { + assertMalformed(requestHeader(-1)); + } + + @Test + public void testRejectAmountAboveLimit() { + ByteBuf source = requestHeader(513); + for (int i = 0; i < 513; i++) { + source.writeByte(ClusterConstants.PARAM_TYPE_BYTE).writeByte(1); + } + assertMalformed(source); + } + + @Test + public void testRejectAmountThatCannotFitRemainingBytes() { + assertMalformed(requestHeader(2) + .writeByte(ClusterConstants.PARAM_TYPE_BYTE) + .writeByte(1)); + } + + @Test + public void testRejectMissingParamType() { + assertMalformed(requestHeader(1)); + } + + @Test + public void testRejectTruncatedPrimitiveValues() { + for (TruncatedParam param : Arrays.asList( + new TruncatedParam((byte)ClusterConstants.PARAM_TYPE_INTEGER, 3), + new TruncatedParam((byte)ClusterConstants.PARAM_TYPE_BOOLEAN, 0), + new TruncatedParam((byte)ClusterConstants.PARAM_TYPE_DOUBLE, 7), + new TruncatedParam((byte)ClusterConstants.PARAM_TYPE_LONG, 7), + new TruncatedParam((byte)ClusterConstants.PARAM_TYPE_FLOAT, 3), + new TruncatedParam((byte)ClusterConstants.PARAM_TYPE_BYTE, 0), + new TruncatedParam((byte)ClusterConstants.PARAM_TYPE_SHORT, 1))) { + assertMalformed(requestHeader(1) + .writeByte(param.type) + .writeZero(param.valueBytes)); + } + } + + @Test + public void testRejectNegativeStringLength() { + assertMalformed(requestHeader(1) + .writeByte(ClusterConstants.PARAM_TYPE_STRING) + .writeInt(-1)); + } + + @Test + public void testRejectStringLengthAboveLimit() { + assertMalformed(requestHeader(1) + .writeByte(ClusterConstants.PARAM_TYPE_STRING) + .writeInt(1025) + .writeZero(1025)); + } + + @Test + public void testRejectStringLengthGreaterThanRemainingBytes() { + assertMalformed(requestHeader(1) + .writeByte(ClusterConstants.PARAM_TYPE_STRING) + .writeInt(4) + .writeZero(3)); + } + + @Test + public void testRejectUnknownParamType() { + assertMalformed(requestHeader(1) + .writeByte(Byte.MAX_VALUE) + .writeByte(1)); + } + + @Test + public void testRejectTrailingBytesWithEmptyParams() { + assertMalformed(requestHeader(0).writeByte(1)); + } + + @Test + public void testRejectTrailingBytesAfterDeclaredParams() { + assertMalformed(requestHeader(1) + .writeByte(ClusterConstants.PARAM_TYPE_BYTE) + .writeByte(1) + .writeByte(2)); + } + + @Test + public void testRejectTruncatedRequestHeader() { + assertMalformed(Unpooled.buffer().writeZero(15)); + } + + private ByteBuf requestHeader(int amount) { + return Unpooled.buffer() + .writeLong(1L) + .writeInt(2) + .writeInt(amount); + } + + private void assertMalformed(ByteBuf source) { + try { + assertThatThrownBy(() -> decoder.decode(source)) + .isInstanceOf(CorruptedFrameException.class); + } finally { + source.release(); + } + } + + private static final class TruncatedParam { + private final byte type; + private final int valueBytes; + + private TruncatedParam(byte type, int valueBytes) { + this.type = type; + this.valueBytes = valueBytes; + } + } +} diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoderTest.java b/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoderTest.java new file mode 100644 index 0000000000..59fc28bead --- /dev/null +++ b/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoderTest.java @@ -0,0 +1,105 @@ +/* + * Copyright 1999-2026 Alibaba Group Holding Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.alibaba.csp.sentinel.cluster.server.codec.data; + +import java.nio.charset.StandardCharsets; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.CorruptedFrameException; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Test cases for {@link PingRequestDataDecoder}. + */ +public class PingRequestDataDecoderTest { + + private final PingRequestDataDecoder decoder = new PingRequestDataDecoder(); + + @Test + public void testDecodeValidNamespace() { + ByteBuf source = pingPayload("default"); + try { + assertThat(decoder.decode(source)).isEqualTo("default"); + assertThat(source.isReadable()).isFalse(); + } finally { + source.release(); + } + } + + @Test + public void testDecodeEmptyNamespace() { + ByteBuf source = Unpooled.buffer().writeInt(0); + try { + assertThat(decoder.decode(source)).isEmpty(); + assertThat(source.isReadable()).isFalse(); + } finally { + source.release(); + } + } + + @Test + public void testRejectNmapProbeBeforeAllocation() { + ByteBuf source = Unpooled.buffer() + .writeInt(0x01000000) + .writeZero(21); + try { + assertThatThrownBy(() -> decoder.decode(source)) + .isInstanceOf(CorruptedFrameException.class); + } finally { + source.release(); + } + } + + @Test + public void testRejectNegativeLength() { + assertMalformed(Unpooled.buffer().writeInt(-1)); + } + + @Test + public void testRejectDeclaredLengthGreaterThanRemainingBytes() { + assertMalformed(Unpooled.buffer().writeInt(4).writeZero(3)); + } + + @Test + public void testRejectDeclaredLengthLessThanRemainingBytes() { + assertMalformed(Unpooled.buffer().writeInt(2).writeZero(3)); + } + + @Test + public void testRejectTruncatedLengthField() { + assertMalformed(Unpooled.buffer().writeZero(3)); + } + + private void assertMalformed(ByteBuf source) { + try { + assertThatThrownBy(() -> decoder.decode(source)) + .isInstanceOf(CorruptedFrameException.class); + } finally { + source.release(); + } + } + + private ByteBuf pingPayload(String namespace) { + byte[] bytes = namespace.getBytes(StandardCharsets.UTF_8); + return Unpooled.buffer() + .writeInt(bytes.length) + .writeBytes(bytes); + } +} diff --git a/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandlerTest.java b/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandlerTest.java new file mode 100644 index 0000000000..323b00bb90 --- /dev/null +++ b/sentinel-cluster/sentinel-cluster-server-default/src/test/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandlerTest.java @@ -0,0 +1,65 @@ +/* + * Copyright 1999-2026 Alibaba Group Holding Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.alibaba.csp.sentinel.cluster.server.handler; + +import com.alibaba.csp.sentinel.cluster.server.connection.ConnectionPool; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.TooLongFrameException; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +/** + * Test cases for {@link TokenServerHandler}. + */ +public class TokenServerHandlerTest { + + @Test + public void testCloseConnectionOnCorruptedFrame() { + assertExceptionClosesConnection(new CorruptedFrameException("malformed")); + } + + @Test + public void testCloseConnectionOnOversizedFrame() { + assertExceptionClosesConnection(new TooLongFrameException("oversized")); + } + + private void assertExceptionClosesConnection(Throwable cause) { + EmbeddedChannel channel = new EmbeddedChannel(new IgnoreChannelInactiveHandler(), + new TokenServerHandler(mock(ConnectionPool.class))); + try { + channel.pipeline().fireExceptionCaught(cause); + channel.runPendingTasks(); + + assertThat(channel.isActive()).isFalse(); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static final class IgnoreChannelInactiveHandler extends ChannelInboundHandlerAdapter { + + @Override + public void channelInactive(ChannelHandlerContext ctx) { + // EmbeddedChannel has no remote address; stop before the production connection cleanup. + } + } +}