This is an automated email from the ASF dual-hosted git repository.
guohao pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new c1b013e [3.0-Triple] Fix unary wrapper overload method npe (#9153)
c1b013e is described below
commit c1b013e3da573c9d6804e82dbf3359345c07d48c
Author: earthchen <[email protected]>
AuthorDate: Thu Oct 28 22:02:34 2021 -0500
[3.0-Triple] Fix unary wrapper overload method npe (#9153)
* fix 9139
* fix not wrapper
* opt
* fix 9139
* fix unary condition
---
.../rpc/protocol/tri/AbstractServerStream.java | 139 +++++++++++++++++----
.../dubbo/rpc/protocol/tri/ServerStream.java | 3 +-
.../tri/TripleHttp2FrameServerHandler.java | 2 +-
.../dubbo/rpc/protocol/tri/UnaryServerStream.java | 33 ++---
4 files changed, 125 insertions(+), 52 deletions(-)
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractServerStream.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractServerStream.java
index 17dd26d..47de41d 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractServerStream.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractServerStream.java
@@ -23,6 +23,7 @@ import
org.apache.dubbo.common.serialize.MultipleSerialization;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.rpc.HeaderFilter;
import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.FrameworkServiceRepository;
import org.apache.dubbo.rpc.model.MethodDescriptor;
@@ -130,6 +131,12 @@ public abstract class AbstractServerStream extends
AbstractStream implements Str
return providerModel;
}
+ /**
+ * Build the RpcInvocation with metadata and execute headerFilter
+ *
+ * @param metadata request header
+ * @return RpcInvocation
+ */
protected RpcInvocation buildInvocation(Metadata metadata) {
RpcInvocation inv = new RpcInvocation(getUrl().getServiceModel(),
getMethodName(), getServiceDescriptor().getServiceName(),
@@ -139,48 +146,90 @@ public abstract class AbstractServerStream extends
AbstractStream implements Str
final Map<String, Object> attachments =
parseMetadataToAttachmentMap(metadata);
inv.setObjectAttachments(attachments);
+ invokeHeaderFilter(inv);
+ return inv;
+ }
+ /**
+ * Intercept the header to do some validation
+ * <p>
+ * for example, check the token or a user-defined permission check
operation
+ *
+ * @param inv RPC Invocation
+ * @throws RpcException maybe throw rpcException
+ */
+ protected void invokeHeaderFilter(RpcInvocation inv) throws RpcException {
for (HeaderFilter headerFilter : getHeaderFilters()) {
- inv = headerFilter.invoke(getInvoker(), inv);
+ headerFilter.invoke(getInvoker(), inv);
}
- return inv;
}
- protected Object[] deserializeRequest(byte[] data) {
+ /**
+ * For the unary method, there may be overloaded methods,
+ * so need to parse out the Wrapper from the data and continue
buildRpcInvocation
+ * <p>
+ * Also, to prevent serialization attacks, headerFilter needs to be
executed
+ *
+ * @param metadata request headers
+ * @param data request data
+ * @return RPC Invocation
+ */
+ protected RpcInvocation buildUnaryInvocation(Metadata metadata, byte[]
data) {
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
try {
if (getProviderModel() != null) {
ClassLoadUtil.switchContextLoader(getProviderModel().getServiceInterfaceClass().getClassLoader());
}
- if (getMethodDescriptor() == null ||
getMethodDescriptor().isNeedWrap()) {
- final TripleWrapper.TripleRequestWrapper wrapper = unpack(data,
- TripleWrapper.TripleRequestWrapper.class);
- if
(!getSerializeType().equals(convertHessianFromWrapper(wrapper.getSerializeType())))
{
-
transportError(GrpcStatus.fromCode(GrpcStatus.Code.INVALID_ARGUMENT)
- .withDescription("Received inconsistent serialization
type from client, " +
- "reject to deserialize! Expected:" +
getSerializeType() +
- " Actual:" +
convertHessianFromWrapper(wrapper.getSerializeType())));
+ // For the Wrapper method,the methodDescriptor needs to get from
data, so parse the request first
+ if (needDeserializeWrapper(getMethodDescriptor())) {
+ // the wrapper structure is first resolved without actual
deserialization
+ TripleWrapper.TripleRequestWrapper wrapper =
deserializeWrapperSetMdIfNeed(data);
+ if (wrapper == null) {
return null;
}
- if (getMethodDescriptor() == null) {
- final String[] paramTypes =
wrapper.getArgTypesList().toArray(new String[wrapper.getArgsCount()]);
- // wrapper mode the method can overload so maybe list
- for (MethodDescriptor descriptor : getMethodDescriptors())
{
- // params type is array
- if
(Arrays.equals(descriptor.getCompatibleParamSignatures(), paramTypes)) {
- method(descriptor);
- break;
- }
- }
- if (getMethodDescriptor() == null) {
-
transportError(GrpcStatus.fromCode(GrpcStatus.Code.UNIMPLEMENTED)
- .withDescription("Method :" + getMethodName() +
"[" + Arrays.toString(paramTypes) + "] " +
- "not found of service:" +
getServiceDescriptor().getServiceName()));
- return null;
- }
+ RpcInvocation inv = buildInvocation(metadata);
+ inv.setArguments(unwrapReq(getUrl(), wrapper,
getMultipleSerialization()));
+ return inv;
+ } else {
+ // Protobuf MethodDescriptor must not be null
+ RpcInvocation inv = buildInvocation(metadata);
+ inv.setArguments(new Object[]{unpack(data,
getMethodDescriptor().getParameterClasses()[0])});
+ return inv;
+ }
+ } catch (RpcException rpcException) {
+ // for catch exceptions in headerFilter
+ transportError(GrpcStatus.getStatus(rpcException,
rpcException.getMessage()));
+ return null;
+ } catch (Throwable throwable) {
+ LOGGER.warn("Decode request failed:", throwable);
+ transportError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("Decode request failed:" +
throwable.getMessage()));
+ return null;
+ } finally {
+ ClassLoadUtil.switchContextLoader(tccl);
+ }
+ }
+
+ /**
+ * Deserialize the stream request data
+ *
+ * @param data request data
+ * @return Deserialized object
+ */
+ protected Object[] deserializeRequest(byte[] data) {
+ ClassLoader tccl = Thread.currentThread().getContextClassLoader();
+ try {
+ if (getProviderModel() != null) {
+
ClassLoadUtil.switchContextLoader(getProviderModel().getServiceInterfaceClass().getClassLoader());
+ }
+ if (needDeserializeWrapper(getMethodDescriptor())) {
+ TripleWrapper.TripleRequestWrapper wrapper =
deserializeWrapperSetMdIfNeed(data);
+ if (wrapper == null) {
+ return null;
}
return unwrapReq(getUrl(), wrapper,
getMultipleSerialization());
} else {
+ // Protobuf MethodDescriptor must not be null
return new Object[]{unpack(data,
getMethodDescriptor().getParameterClasses()[0])};
}
} catch (Throwable throwable) {
@@ -193,6 +242,42 @@ public abstract class AbstractServerStream extends
AbstractStream implements Str
}
}
+ private boolean needDeserializeWrapper(MethodDescriptor md) {
+ if (md == null) {
+ return true;
+ }
+ return getMethodDescriptor().isNeedWrap();
+ }
+
+ private TripleWrapper.TripleRequestWrapper
deserializeWrapperSetMdIfNeed(byte[] data) {
+ final TripleWrapper.TripleRequestWrapper wrapper = unpack(data,
TripleWrapper.TripleRequestWrapper.class);
+ if
(!getSerializeType().equals(convertHessianFromWrapper(wrapper.getSerializeType())))
{
+
transportError(GrpcStatus.fromCode(GrpcStatus.Code.INVALID_ARGUMENT)
+ .withDescription("Received inconsistent serialization type
from client, " +
+ "reject to deserialize! Expected:" + getSerializeType() +
+ " Actual:" +
convertHessianFromWrapper(wrapper.getSerializeType())));
+ return null;
+ }
+ if (getMethodDescriptor() == null) {
+ final String[] paramTypes = wrapper.getArgTypesList().toArray(new
String[wrapper.getArgsCount()]);
+ // wrapper mode the method can overload so maybe list
+ for (MethodDescriptor descriptor : getMethodDescriptors()) {
+ // params type is array
+ if (Arrays.equals(descriptor.getCompatibleParamSignatures(),
paramTypes)) {
+ method(descriptor);
+ break;
+ }
+ }
+ if (getMethodDescriptor() == null) {
+
transportError(GrpcStatus.fromCode(GrpcStatus.Code.UNIMPLEMENTED)
+ .withDescription("Method :" + getMethodName() + "[" +
Arrays.toString(paramTypes) + "] " +
+ "not found of service:" +
getServiceDescriptor().getServiceName()));
+ return null;
+ }
+ }
+ return wrapper;
+ }
+
private Object[] unwrapReq(URL url, TripleWrapper.TripleRequestWrapper
wrap,
MultipleSerialization multipleSerialization) {
String serializeType =
convertHessianFromWrapper(wrap.getSerializeType());
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
index 5623b33..f5ff7c9 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
@@ -22,7 +22,6 @@ import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcInvocation;
-import org.apache.dubbo.rpc.model.MethodDescriptor;
public class ServerStream extends AbstractServerStream implements Stream {
protected ServerStream(URL url) {
@@ -107,7 +106,7 @@ public class ServerStream extends AbstractServerStream
implements Stream {
@Override
public void onMetadata(Metadata metadata, boolean endStream) {
super.onMetadata(metadata, endStream);
- if (getMethodDescriptor().getRpcType() ==
MethodDescriptor.RpcType.SERVER_STREAM) {
+ if (getMethodDescriptor().isServerStream()) {
return;
}
execute(() -> {
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java
index f450e47..ab232f7 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java
@@ -225,7 +225,7 @@ public class TripleHttp2FrameServerHandler extends
ChannelDuplexHandler {
}
}
- boolean isUnary = methodDescriptor != null &&
methodDescriptor.isUnary();
+ boolean isUnary = methodDescriptor == null ||
methodDescriptor.isUnary();
final AbstractServerStream stream =
AbstractServerStream.newServerStream(invoker.getUrl(), isUnary);
Channel channel = ctx.channel();
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
index 15be193..21c418a 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
@@ -53,31 +53,20 @@ public class UnaryServerStream extends AbstractServerStream
implements Stream {
@Override
public void onComplete() {
- if (getData() != null) {
- invoke();
- } else {
- onError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
- .withDescription("Missing request data"));
- }
+ execute(() -> {
+ if (getData() != null) {
+ invoke();
+ } else {
+ onError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("Missing request data"));
+ }
+ });
}
public void invoke() {
- RpcInvocation invocation;
- if (getMethodDescriptor().isNeedWrap()) {
- // For wrapper overload methods, the methodDescriptor needs to
get from data, so parse the request first
- final Object[] arguments = deserializeRequest(getData());
- if (arguments == null) {
- return;
- }
- invocation = buildInvocation(getHeaders());
- invocation.setArguments(arguments);
- } else {
- invocation = buildInvocation(getHeaders());
- final Object[] arguments = deserializeRequest(getData());
- if (arguments == null) {
- return;
- }
- invocation.setArguments(arguments);
+ RpcInvocation invocation = buildUnaryInvocation(getHeaders(),
getData());
+ if (invocation == null) {
+ return;
}
final Result result = getInvoker().invoke(invocation);
CompletionStage<Object> future =
result.thenApply(Function.identity());
