Skip to content

Commit 8204f6e

Browse files
yungyu16yungyu16
committed
fix: enhance token server request handling and add max frame length validation
1 parent a3f40ba commit 8204f6e

8 files changed

Lines changed: 478 additions & 58 deletions

File tree

‎sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/NettyTransportServer.java‎

Lines changed: 30 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -15,17 +15,12 @@
1515
*/
1616
package com.alibaba.csp.sentinel.cluster.server;
1717

18-
import java.util.ArrayList;
19-
import java.util.List;
20-
import java.util.concurrent.atomic.AtomicInteger;
21-
2218
import com.alibaba.csp.sentinel.cluster.server.codec.netty.NettyRequestDecoder;
2319
import com.alibaba.csp.sentinel.cluster.server.codec.netty.NettyResponseEncoder;
2420
import com.alibaba.csp.sentinel.cluster.server.connection.Connection;
2521
import com.alibaba.csp.sentinel.cluster.server.connection.ConnectionPool;
2622
import com.alibaba.csp.sentinel.cluster.server.handler.TokenServerHandler;
2723
import com.alibaba.csp.sentinel.log.RecordLog;
28-
2924
import io.netty.bootstrap.ServerBootstrap;
3025
import io.netty.buffer.PooledByteBufAllocator;
3126
import io.netty.channel.ChannelFuture;
@@ -42,7 +37,14 @@
4237
import io.netty.util.concurrent.GenericFutureListener;
4338
import io.netty.util.internal.SystemPropertyUtil;
4439

45-
import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.*;
40+
import java.util.ArrayList;
41+
import java.util.List;
42+
import java.util.concurrent.atomic.AtomicInteger;
43+
44+
import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.NETTY_MAX_FRAME_LENGTH;
45+
import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.SERVER_STATUS_OFF;
46+
import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.SERVER_STATUS_STARTED;
47+
import static com.alibaba.csp.sentinel.cluster.server.ServerConstants.SERVER_STATUS_STARTING;
4648

4749
/**
4850
* @author Eric Zhao
@@ -51,7 +53,7 @@
5153
public class NettyTransportServer implements ClusterTokenServer {
5254

5355
private static final int DEFAULT_EVENT_LOOP_THREADS = Math.max(1,
54-
SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
56+
SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
5557
private static final int MAX_RETRY_TIMES = 3;
5658
private static final int RETRY_SLEEP_MS = 2000;
5759

@@ -79,32 +81,32 @@ public void start() {
7981
this.bossGroup = new NioEventLoopGroup(1);
8082
this.workerGroup = new NioEventLoopGroup(DEFAULT_EVENT_LOOP_THREADS);
8183
b.group(bossGroup, workerGroup)
82-
.channel(NioServerSocketChannel.class)
83-
.option(ChannelOption.SO_BACKLOG, 128)
84-
.handler(new LoggingHandler(LogLevel.INFO))
85-
.childHandler(new ChannelInitializer<SocketChannel>() {
86-
@Override
87-
public void initChannel(SocketChannel ch) throws Exception {
88-
ChannelPipeline p = ch.pipeline();
89-
p.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2));
90-
p.addLast(new NettyRequestDecoder());
91-
p.addLast(new LengthFieldPrepender(2));
92-
p.addLast(new NettyResponseEncoder());
93-
p.addLast(new TokenServerHandler(connectionPool));
94-
}
95-
})
96-
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
97-
.childOption(ChannelOption.SO_SNDBUF, 32 * 1024)
98-
.childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
99-
.childOption(ChannelOption.SO_TIMEOUT, 10)
100-
.childOption(ChannelOption.TCP_NODELAY, true)
101-
.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
84+
.channel(NioServerSocketChannel.class)
85+
.option(ChannelOption.SO_BACKLOG, 128)
86+
.handler(new LoggingHandler(LogLevel.INFO))
87+
.childHandler(new ChannelInitializer<SocketChannel>() {
88+
@Override
89+
public void initChannel(SocketChannel ch) throws Exception {
90+
ChannelPipeline p = ch.pipeline();
91+
p.addLast(new LengthFieldBasedFrameDecoder(NETTY_MAX_FRAME_LENGTH, 0, 2, 0, 2));
92+
p.addLast(new NettyRequestDecoder());
93+
p.addLast(new LengthFieldPrepender(2));
94+
p.addLast(new NettyResponseEncoder());
95+
p.addLast(new TokenServerHandler(connectionPool));
96+
}
97+
})
98+
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
99+
.childOption(ChannelOption.SO_SNDBUF, 32 * 1024)
100+
.childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
101+
.childOption(ChannelOption.SO_TIMEOUT, 10)
102+
.childOption(ChannelOption.TCP_NODELAY, true)
103+
.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
102104
b.bind(port).addListener(new GenericFutureListener<ChannelFuture>() {
103105
@Override
104106
public void operationComplete(ChannelFuture future) {
105107
if (future.cause() != null) {
106108
RecordLog.info("[NettyTransportServer] Token server start failed (port=" + port + "), failedTimes: " + failedTimes.get(),
107-
future.cause());
109+
future.cause());
108110
currentState.compareAndSet(SERVER_STATUS_STARTING, SERVER_STATUS_OFF);
109111
int failCount = failedTimes.incrementAndGet();
110112
if (failCount > MAX_RETRY_TIMES) {

‎sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/ServerConstants.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,5 +27,10 @@ public final class ServerConstants {
2727

2828
public static final String DEFAULT_NAMESPACE = "default";
2929

30-
private ServerConstants() {}
30+
public static final int NETTY_MAX_FRAME_LENGTH = 1024;
31+
public static final int MAX_PARAM_AMOUNT = 512;
32+
public static final int MAX_PARAM_STRING_LENGTH = 1024;
33+
34+
private ServerConstants() {
35+
}
3136
}

‎sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/ParamFlowRequestDataDecoder.java‎

Lines changed: 56 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,16 @@
1616
package com.alibaba.csp.sentinel.cluster.server.codec.data;
1717

1818
import java.util.ArrayList;
19+
import java.util.Collections;
1920
import java.util.List;
2021

2122
import com.alibaba.csp.sentinel.cluster.ClusterConstants;
2223
import com.alibaba.csp.sentinel.cluster.codec.EntityDecoder;
2324
import com.alibaba.csp.sentinel.cluster.request.data.ParamFlowRequestData;
25+
import com.alibaba.csp.sentinel.cluster.server.ServerConstants;
2426

2527
import io.netty.buffer.ByteBuf;
28+
import io.netty.handler.codec.CorruptedFrameException;
2629

2730
/**
2831
* @author jialiang.linjl
@@ -31,61 +34,93 @@
3134
*/
3235
public class ParamFlowRequestDataDecoder implements EntityDecoder<ByteBuf, ParamFlowRequestData> {
3336

37+
private static final int REQUEST_HEADER_LENGTH = Long.BYTES + Integer.BYTES + Integer.BYTES;
38+
private static final int MIN_PARAM_LENGTH = Byte.BYTES + Byte.BYTES;
39+
3440
@Override
3541
public ParamFlowRequestData decode(ByteBuf source) {
36-
if (source.readableBytes() >= 16) {
37-
ParamFlowRequestData requestData = new ParamFlowRequestData()
42+
ensureReadable(source, REQUEST_HEADER_LENGTH);
43+
44+
ParamFlowRequestData requestData = new ParamFlowRequestData()
3845
.setFlowId(source.readLong())
3946
.setCount(source.readInt());
4047

41-
int amount = source.readInt();
42-
if (amount > 0) {
43-
List<Object> params = new ArrayList<>(amount);
44-
for (int i = 0; i < amount; i++) {
45-
decodeParam(source, params);
46-
}
48+
int amount = source.readInt();
49+
if (amount < 0 || amount > ServerConstants.MAX_PARAM_AMOUNT
50+
|| amount > source.readableBytes() / MIN_PARAM_LENGTH) {
51+
throw new CorruptedFrameException("Invalid parameter amount: " + amount);
52+
}
4753

48-
requestData.setParams(params);
49-
return requestData;
54+
List<Object> params;
55+
if (amount == 0) {
56+
params = Collections.emptyList();
57+
} else {
58+
params = new ArrayList<>(Math.min(amount, 16));
59+
for (int i = 0; i < amount; i++) {
60+
decodeParam(source, params);
5061
}
5162
}
52-
return null;
63+
if (source.isReadable()) {
64+
throw new CorruptedFrameException("Parameter flow payload contains trailing bytes: "
65+
+ source.readableBytes());
66+
}
67+
return requestData.setParams(params);
5368
}
5469

55-
private boolean decodeParam(ByteBuf source, List<Object> params) {
70+
private void decodeParam(ByteBuf source, List<Object> params) {
71+
ensureReadable(source, Byte.BYTES);
5672
byte paramType = source.readByte();
5773

5874
switch (paramType) {
5975
case ClusterConstants.PARAM_TYPE_INTEGER:
76+
ensureReadable(source, Integer.BYTES);
6077
params.add(source.readInt());
61-
return true;
78+
return;
6279
case ClusterConstants.PARAM_TYPE_STRING:
80+
ensureReadable(source, Integer.BYTES);
6381
int length = source.readInt();
82+
if (length < 0 || length > ServerConstants.MAX_PARAM_STRING_LENGTH
83+
|| length > source.readableBytes()) {
84+
throw new CorruptedFrameException("Invalid string parameter length: " + length);
85+
}
6486
byte[] bytes = new byte[length];
6587
source.readBytes(bytes);
6688
// TODO: take care of charset?
6789
params.add(new String(bytes));
68-
return true;
90+
return;
6991
case ClusterConstants.PARAM_TYPE_BOOLEAN:
92+
ensureReadable(source, Byte.BYTES);
7093
params.add(source.readBoolean());
71-
return true;
94+
return;
7295
case ClusterConstants.PARAM_TYPE_DOUBLE:
96+
ensureReadable(source, Double.BYTES);
7397
params.add(source.readDouble());
74-
return true;
98+
return;
7599
case ClusterConstants.PARAM_TYPE_LONG:
100+
ensureReadable(source, Long.BYTES);
76101
params.add(source.readLong());
77-
return true;
102+
return;
78103
case ClusterConstants.PARAM_TYPE_FLOAT:
104+
ensureReadable(source, Float.BYTES);
79105
params.add(source.readFloat());
80-
return true;
106+
return;
81107
case ClusterConstants.PARAM_TYPE_BYTE:
108+
ensureReadable(source, Byte.BYTES);
82109
params.add(source.readByte());
83-
return true;
110+
return;
84111
case ClusterConstants.PARAM_TYPE_SHORT:
112+
ensureReadable(source, Short.BYTES);
85113
params.add(source.readShort());
86-
return true;
114+
return;
87115
default:
88-
return false;
116+
throw new CorruptedFrameException("Unknown parameter type: " + paramType);
117+
}
118+
}
119+
120+
private void ensureReadable(ByteBuf source, int requiredBytes) {
121+
if (source.readableBytes() < requiredBytes) {
122+
throw new CorruptedFrameException("Incomplete parameter flow payload: required=" + requiredBytes
123+
+ ", actual=" + source.readableBytes());
89124
}
90125
}
91126
}

‎sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/codec/data/PingRequestDataDecoder.java‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import com.alibaba.csp.sentinel.cluster.codec.EntityDecoder;
1919

2020
import io.netty.buffer.ByteBuf;
21+
import io.netty.handler.codec.CorruptedFrameException;
2122

2223
/**
2324
* @author Eric Zhao
@@ -27,14 +28,19 @@ public class PingRequestDataDecoder implements EntityDecoder<ByteBuf, String> {
2728

2829
@Override
2930
public String decode(ByteBuf source) {
30-
if (source.readableBytes() >= 4) {
31-
int length = source.readInt();
32-
if (length > 0 && source.readableBytes() > 0) {
33-
byte[] bytes = new byte[length];
34-
source.readBytes(bytes);
35-
return new String(bytes);
36-
}
31+
if (source.readableBytes() < Integer.BYTES) {
32+
throw new CorruptedFrameException("Incomplete ping payload length");
3733
}
38-
return null;
34+
35+
int packetLen = source.readInt();
36+
int actualLength = source.readableBytes();
37+
if (packetLen < 0 || packetLen != actualLength) {
38+
throw new CorruptedFrameException("Invalid ping payload length: declared=" + packetLen
39+
+ ", actual=" + actualLength);
40+
}
41+
42+
byte[] bytes = new byte[packetLen];
43+
source.readBytes(bytes);
44+
return new String(bytes);
3945
}
4046
}

‎sentinel-cluster/sentinel-cluster-server-default/src/main/java/com/alibaba/csp/sentinel/cluster/server/handler/TokenServerHandler.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@
2929

3030
import io.netty.channel.ChannelHandlerContext;
3131
import io.netty.channel.ChannelInboundHandlerAdapter;
32+
import io.netty.handler.codec.CorruptedFrameException;
33+
import io.netty.handler.codec.TooLongFrameException;
3234

3335
/**
3436
* Netty server handler for Sentinel token server.
@@ -81,6 +83,14 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
8183
}
8284
}
8385

86+
@Override
87+
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
88+
if (!(cause instanceof CorruptedFrameException) && !(cause instanceof TooLongFrameException)) {
89+
RecordLog.warn("[TokenServerHandler] Unexpected exception", cause);
90+
}
91+
ctx.close();
92+
}
93+
8494
private void writeBadResponse(ChannelHandlerContext ctx, ClusterRequest request) {
8595
ClusterResponse<?> response = new ClusterResponse<>(request.getId(), request.getType(),
8696
ClusterConstants.RESPONSE_STATUS_BAD, null);

0 commit comments

Comments
 (0)