initial

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java

@@ -28,6 +28,7 @@
 import io.rsocket.RSocket;
 import io.rsocket.exceptions.ConnectionErrorException;
 import io.rsocket.exceptions.Exceptions;
+import io.rsocket.fragmentation.FragmentationUtils;
 import io.rsocket.frame.CancelFrameFlyweight;
 import io.rsocket.frame.ErrorFrameFlyweight;
 import io.rsocket.frame.FrameHeaderFlyweight;
@@ -88,6 +89,7 @@ class RSocketRequester implements RSocket {
   private final IntObjectMap<Subscription> senders;
   private final IntObjectMap<Processor<Payload, Payload>> receivers;
   private final UnboundedProcessor<ByteBuf> sendProcessor;
+  private final int mtu;
   private final RequesterLeaseHandler leaseHandler;
   private final ByteBufAllocator allocator;
   private final KeepAliveFramesAcceptor keepAliveFramesAcceptor;
@@ -99,6 +101,7 @@ class RSocketRequester implements RSocket {
       PayloadDecoder payloadDecoder,
       Consumer<Throwable> errorConsumer,
       StreamIdSupplier streamIdSupplier,
+      int mtu,
       int keepAliveTickPeriod,
       int keepAliveAckTimeout,
       @Nullable KeepAliveHandler keepAliveHandler,
@@ -108,6 +111,7 @@ class RSocketRequester implements RSocket {
     this.payloadDecoder = payloadDecoder;
     this.errorConsumer = errorConsumer;
     this.streamIdSupplier = streamIdSupplier;
+    this.mtu = mtu;
     this.leaseHandler = leaseHandler;
     this.senders = new SynchronizedIntObjectHashMap<>();
     this.receivers = new SynchronizedIntObjectHashMap<>();
@@ -186,6 +190,11 @@ private Mono<Void> handleFireAndForget(Payload payload) {
       return Mono.error(err);
     }
 
+    if (!FragmentationUtils.isValid(this.mtu, payload)) {
+      payload.release();
+      return Mono.error(new IllegalArgumentException("Too big Payload size"));
+    }
+
     final int streamId = streamIdSupplier.nextStreamId(receivers);
 
     return UnicastMonoEmpty.newInstance(
@@ -210,6 +219,11 @@ private Mono<Payload> handleRequestResponse(final Payload payload) {
       return Mono.error(err);
     }
 
+    if (!FragmentationUtils.isValid(this.mtu, payload)) {
+      payload.release();
+      return Mono.error(new IllegalArgumentException("Too big Payload size"));
+    }
+
     int streamId = streamIdSupplier.nextStreamId(receivers);
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
 
@@ -255,6 +269,11 @@ private Flux<Payload> handleRequestStream(final Payload payload) {
       return Flux.error(err);
     }
 
+    if (!FragmentationUtils.isValid(this.mtu, payload)) {
+      payload.release();
+      return Flux.error(new IllegalArgumentException("Too big Payload size"));
+    }
+
     int streamId = streamIdSupplier.nextStreamId(receivers);
 
     final UnboundedProcessor<ByteBuf> sendProcessor = this.sendProcessor;
@@ -317,6 +336,12 @@ private Flux<Payload> handleChannel(Flux<Payload> request) {
         (s, flux) -> {
           Payload payload = s.get();
           if (payload != null) {
+              if (!FragmentationUtils.isValid(mtu, payload)) {
+                  payload.release();
+                  final IllegalArgumentException t = new IllegalArgumentException("Too big Payload size");
+                  errorConsumer.accept(t);
+                  return Mono.error(t);
+              }
             return handleChannel(payload, flux.skip(1));
           } else {
             return flux;
@@ -341,6 +366,16 @@ protected void hookOnSubscribe(Subscription subscription) {
 
           @Override
           protected void hookOnNext(Payload payload) {
+              if (!FragmentationUtils.isValid(mtu, payload)) {
+                  payload.release();
+                  cancel();
+                  final IllegalArgumentException t = new IllegalArgumentException("Too big Payload size");
+                  errorConsumer.accept(t);
+                  // no need to send any errors.
+                  sendProcessor.onNext(CancelFrameFlyweight.encode(allocator, streamId));
+                  receiver.onError(t);
+                  return ;
+              }
             final ByteBuf frame =
                 PayloadFrameFlyweight.encode(allocator, streamId, false, false, true, payload);
 

rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationUtils.java

@@ -0,0 +1,33 @@
+package io.rsocket.fragmentation;
+
+import io.netty.buffer.ByteBuf;
+import io.rsocket.Payload;
+import io.rsocket.frame.FrameHeaderFlyweight;
+import io.rsocket.frame.FrameLengthFlyweight;
+
+public final class FragmentationUtils {
+    public static boolean isValid(int mtu, Payload payload) {
+        return payload.hasMetadata() ? isValid(mtu, payload.data(), payload.metadata()) : isValid(mtu, payload.metadata());
+    }
+
+    public static boolean isValid(int mtu, ByteBuf data) {
+        return mtu > 0
+                || (((FrameHeaderFlyweight.size()
+                + data.readableBytes()
+                + FrameLengthFlyweight.FRAME_LENGTH_SIZE)
+                & ~FrameLengthFlyweight.FRAME_LENGTH_MASK)
+                == 0);
+    }
+
+    public static boolean isValid(int mtu, ByteBuf data, ByteBuf metadata) {
+        return mtu > 0
+                || (((FrameHeaderFlyweight.size()
+                + FrameLengthFlyweight.FRAME_LENGTH_SIZE
+                + FrameHeaderFlyweight.size()
+                + data.readableBytes()
+                + metadata.readableBytes())
+                & ~FrameLengthFlyweight.FRAME_LENGTH_MASK)
+                == 0);
+    }
+
+}

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameFragmenter.java

@@ -20,6 +20,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
 import io.netty.util.ReferenceCountUtil;
+import io.rsocket.Payload;
 import io.rsocket.frame.*;
 import java.util.function.Consumer;
 import org.reactivestreams.Publisher;

rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterSubscribersTest.java

@@ -67,6 +67,7 @@ void setUp() {
             StreamIdSupplier.clientSupplier(),
             0,
             0,
+            0,
             null,
             RequesterLeaseHandler.None);
   }

rsocket-transport-netty/src/test/resources/logback-test.xml

@@ -23,8 +23,8 @@
         </encoder>
     </appender>
 
-    <logger name="io.rsocket.transport.netty" level="INFO"/>
-    <logger name="io.rsocket.FrameLogger" level="INFO"/>
+    <logger name="io.rsocket.transport.netty" level="DEBUG"/>
+    <logger name="io.rsocket.FrameLogger" level="DEBUG"/>
     <logger name="io.rsocket.fragmentation.FragmentationDuplexConnection" level="INFO"/>
 
     <root level="INFO">