natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Reject if response do not match any request (#11882)

This commit is contained in:
Albumen Kevin 2023-03-25 09:37:31 +08:00
parent 62df60a046
commit 35f00f0395
4 changed files with 46 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java
View File

@ -42,6 +42,8 @@ import org.apache.dubbo.remoting.transport.ExceedPayloadLimitException;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* ExchangeCodec.
@ -167,7 +169,7 @@ public class ExchangeCodec extends TelnetCodec {
data = decodeEventData(channel, CodecSupport.deserialize(channel.getUrl(), new ByteArrayInputStream(eventPayload), proto), eventPayload);
}
} else {
data = decodeResponseData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto), getRequestData(id));
data = decodeResponseData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto), getRequestData(channel, res, id));
}
res.setResult(data);
} else {
@ -209,16 +211,21 @@ public class ExchangeCodec extends TelnetCodec {
}
}
protected Object getRequestData(long id) {
protected Object getRequestData(Channel channel, Response response, long id) {
DefaultFuture future = DefaultFuture.getFuture(id);
if (future == null) {
return null;
if (future != null) {
Request req = future.getRequest();
if (req != null) {
return req.getData();
}
}
Request req = future.getRequest();
if (req == null) {
return null;
}
return req.getData();
logger.warn("The timeout response finally returned at "
+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
+ ", response status is " + response.getStatus() + ", response id is " + response.getId()
+ (channel == null ? "" : ", channel: " + channel.getLocalAddress()
+ " -> " + channel.getRemoteAddress()) + ", please check provider side for detailed result.");
throw new IllegalArgumentException("Failed to find any request match the response, response id: " + id);
}
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
@ -427,7 +434,7 @@ public class ExchangeCodec extends TelnetCodec {
try {
if (eventBytes != null) {
int dataLen = eventBytes.length;
int threshold = ConfigurationUtils.getSystemConfiguration(channel.getUrl().getScopeModel()).getInt("deserialization.event.size", 50);
int threshold = ConfigurationUtils.getSystemConfiguration(channel.getUrl().getScopeModel()).getInt("deserialization.event.size", 15);
if (dataLen > threshold) {
throw new IllegalArgumentException("Event data too long, actual size " + threshold + ", threshold " + threshold + " rejected for security consideration.");
}

17
dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/codec/ExchangeCodecTest.java
View File

@ -30,11 +30,13 @@ import org.apache.dubbo.remoting.buffer.ChannelBuffers;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.remoting.exchange.Response;
import org.apache.dubbo.remoting.exchange.codec.ExchangeCodec;
import org.apache.dubbo.remoting.exchange.support.DefaultFuture;
import org.apache.dubbo.remoting.telnet.codec.TelnetCodec;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@ -137,6 +139,8 @@ public class ExchangeCodecTest extends TelnetCodecTest {
@Test
public void test_Decode_Error_Length() throws IOException {
DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null);
byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, 0x02, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
Person person = new Person();
byte[] request = getRequestBytes(person, header);
@ -148,6 +152,8 @@ public class ExchangeCodecTest extends TelnetCodecTest {
Assertions.assertEquals(person, obj.getResult());
//only decode necessary bytes
Assertions.assertEquals(request.length, buffer.readerIndex());
future.cancel();
}
@Test
@ -226,6 +232,7 @@ public class ExchangeCodecTest extends TelnetCodecTest {
@Test
public void test_Decode_Return_Response_Person() throws IOException {
DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null);
//00000010-response/oneway/hearbeat=false/hessian |20-stats=ok|id=0|length=0
byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, 2, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
Person person = new Person();
@ -235,6 +242,8 @@ public class ExchangeCodecTest extends TelnetCodecTest {
Assertions.assertEquals(20, obj.getStatus());
Assertions.assertEquals(person, obj.getResult());
System.out.println(obj);
future.cancel();
}
@Test //The status input has a problem, and the read information is wrong when the serialization is serialized.
@ -324,6 +333,7 @@ public class ExchangeCodecTest extends TelnetCodecTest {
@Test
public void test_Header_Response_NoSerializationFlag() throws IOException {
DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null);
//00000010-response/oneway/hearbeat=false/noset |20-stats=ok|id=0|length=0
byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, (byte) 0x02, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
Person person = new Person();
@ -333,10 +343,13 @@ public class ExchangeCodecTest extends TelnetCodecTest {
Assertions.assertEquals(20, obj.getStatus());
Assertions.assertEquals(person, obj.getResult());
System.out.println(obj);
future.cancel();
}
@Test
public void test_Header_Response_Heartbeat() throws IOException {
DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null);
//00000010-response/oneway/hearbeat=true |20-stats=ok|id=0|length=0
byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, 0x02, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
Person person = new Person();
@ -346,6 +359,8 @@ public class ExchangeCodecTest extends TelnetCodecTest {
Assertions.assertEquals(20, obj.getStatus());
Assertions.assertEquals(person, obj.getResult());
System.out.println(obj);
future.cancel();
}
@Test
@ -371,6 +386,7 @@ public class ExchangeCodecTest extends TelnetCodecTest {
@Test
public void test_Encode_Response() throws IOException {
DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(1001), 100000, null);
ChannelBuffer encodeBuffer = ChannelBuffers.dynamicBuffer(1024);
Channel channel = getClientSideChannel(url);
Response response = new Response();
@ -396,6 +412,7 @@ public class ExchangeCodecTest extends TelnetCodecTest {
// encode response verson ??
// Assertions.assertEquals(response.getProtocolVersion(), obj.getVersion());
future.cancel();
}
@Test

4
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java
View File

@ -101,12 +101,12 @@ public class DubboCodec extends ExchangeCodec {
DecodeableRpcResult result;
if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, DEFAULT_DECODE_IN_IO_THREAD)) {
result = new DecodeableRpcResult(channel, res, is,
(Invocation) getRequestData(id), proto);
(Invocation) getRequestData(channel, res, id), proto);
result.decode();
} else {
result = new DecodeableRpcResult(channel, res,
new UnsafeByteArrayInputStream(readMessageData(is)),
(Invocation) getRequestData(id), proto);
(Invocation) getRequestData(channel, res, id), proto);
}
data = result;
}

13
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCountCodecTest.java
View File

@ -22,6 +22,7 @@ import org.apache.dubbo.remoting.buffer.ChannelBuffer;
import org.apache.dubbo.remoting.buffer.ChannelBuffers;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.remoting.exchange.Response;
import org.apache.dubbo.remoting.exchange.support.DefaultFuture;
import org.apache.dubbo.remoting.exchange.support.MultiMessage;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.RpcInvocation;
@ -32,7 +33,9 @@ import org.apache.dubbo.rpc.protocol.dubbo.support.DemoService;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import static org.apache.dubbo.rpc.Constants.INPUT_KEY;
import static org.apache.dubbo.rpc.Constants.OUTPUT_KEY;
@ -45,23 +48,25 @@ public class DubboCountCodecTest {
ChannelBuffer buffer = ChannelBuffers.buffer(1024);
Channel channel = new MockChannel();
Assertions.assertEquals(Codec2.DecodeResult.NEED_MORE_INPUT, dubboCountCodec.decode(channel, buffer));
List<DefaultFuture> futures = new ArrayList<>();
for (int i = 0; i < 10; i++) {
Request request = new Request(1);
Request request = new Request(i);
futures.add(DefaultFuture.newFuture(channel, request, 1000, null));
RpcInvocation rpcInvocation = new RpcInvocation(null, "echo", DemoService.class.getName(), "", new Class<?>[]{String.class}, new String[]{"yug"});
request.setData(rpcInvocation);
dubboCountCodec.encode(channel, buffer, request);
}
for (int i = 0; i < 10; i++) {
Response response = new Response(1);
Response response = new Response(i);
AppResponse appResponse = new AppResponse(i);
response.setResult(appResponse);
dubboCountCodec.encode(channel, buffer, response);
}
MultiMessage multiMessage = (MultiMessage) dubboCountCodec.decode(channel, buffer);
Assertions.assertEquals(multiMessage.size(), 20);
Assertions.assertEquals(20, multiMessage.size());
int requestCount = 0;
int responseCount = 0;
Iterator iterator = multiMessage.iterator();
@ -79,6 +84,8 @@ public class DubboCountCodecTest {
}
Assertions.assertEquals(requestCount, 10);
Assertions.assertEquals(responseCount, 10);
futures.forEach(DefaultFuture::cancel);
}
}