Added experimental options to limit incoming packets

src/main/java/ua/nanit/limbo/configuration/LimboConfig.java

@@ -71,6 +71,11 @@ public final class LimboConfig {
     private int bossGroupSize;
     private int workerGroupSize;
 
+    private boolean useTrafficLimits;
+    private int maxPacketSize;
+    private int maxPacketsPerSec;
+    private int maxBytesPerSec;
+
     public LimboConfig(Path root) {
         this.root = root;
     }
@@ -127,6 +132,11 @@ public final class LimboConfig {
         useEpoll = conf.node("netty", "useEpoll").getBoolean(true);
         bossGroupSize = conf.node("netty", "threads", "bossGroup").getInt(1);
         workerGroupSize = conf.node("netty", "threads", "workerGroup").getInt(4);
+
+        useTrafficLimits = conf.node("traffic", "enable").getBoolean(false);
+        maxPacketSize = conf.node("traffic", "packetSize").getInt(-1);
+        maxPacketsPerSec = conf.node("traffic", "packets").getInt(-1);
+        maxBytesPerSec = conf.node("traffic", "bytes").getInt(-1);
     }
 
     private BufferedReader getReader() throws IOException {
@@ -250,4 +260,20 @@ public final class LimboConfig {
     public int getWorkerGroupSize() {
         return workerGroupSize;
     }
+
+    public boolean isUseTrafficLimits() {
+        return useTrafficLimits;
+    }
+
+    public int getMaxPacketSize() {
+        return maxPacketSize;
+    }
+
+    public int getMaxPacketsPerSec() {
+        return maxPacketsPerSec;
+    }
+
+    public int getMaxBytesPerSec() {
+        return maxBytesPerSec;
+    }
 }

src/main/java/ua/nanit/limbo/connection/ClientChannelInitializer.java

@@ -21,10 +21,7 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.timeout.ReadTimeoutHandler;
-import ua.nanit.limbo.connection.pipeline.PacketDecoder;
+import ua.nanit.limbo.connection.pipeline.*;
-import ua.nanit.limbo.connection.pipeline.PacketEncoder;
-import ua.nanit.limbo.connection.pipeline.VarIntFrameDecoder;
-import ua.nanit.limbo.connection.pipeline.VarIntLengthEncoder;
 import ua.nanit.limbo.server.LimboServer;
 
 import java.util.concurrent.TimeUnit;
@@ -49,6 +46,15 @@ public class ClientChannelInitializer extends ChannelInitializer<Channel> {
                 TimeUnit.MILLISECONDS));
         pipeline.addLast("frame_decoder", new VarIntFrameDecoder());
         pipeline.addLast("frame_encoder", new VarIntLengthEncoder());
+
+        if (server.getConfig().isUseTrafficLimits()) {
+            pipeline.addLast("traffic_limit", new ChannelTrafficHandler(
+                    server.getConfig().getMaxPacketSize(),
+                    server.getConfig().getMaxPacketsPerSec(),
+                    server.getConfig().getMaxBytesPerSec()
+            ));
+        }
+
         pipeline.addLast("decoder", decoder);
         pipeline.addLast("encoder", encoder);
         pipeline.addLast("handler", connection);

src/main/java/ua/nanit/limbo/connection/pipeline/ChannelTrafficHandler.java

@@ -0,0 +1,77 @@
+package ua.nanit.limbo.connection.pipeline;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import org.jetbrains.annotations.NotNull;
+import ua.nanit.limbo.server.Logger;
+
+public class ChannelTrafficHandler extends ChannelInboundHandlerAdapter {
+
+    private final int packetSize;
+    private final int packetsPerSec;
+    private final int bytesPerSec;
+
+    private int packetsCounter;
+    private int bytesCounter;
+
+    private long lastPacket;
+
+    public ChannelTrafficHandler(int packetSize, int packetsPerSec, int bytesPerSec) {
+        this.packetSize = packetSize;
+        this.packetsPerSec = packetsPerSec;
+        this.bytesPerSec = bytesPerSec;
+    }
+
+    @Override
+    public void channelRead(@NotNull ChannelHandlerContext ctx, @NotNull Object msg) throws Exception {
+        if (msg instanceof ByteBuf) {
+            ByteBuf in = (ByteBuf) msg;
+            int bytes = in.readableBytes();
+
+            System.out.println(bytes + " bytes");
+
+            if (packetSize > 0 && bytes > packetSize) {
+                closeConnection(ctx, "Closed %s due too large packet size (%d bytes)", ctx.channel().remoteAddress(), bytes);
+                return;
+            }
+
+            if (!measureTraffic(ctx, bytes)) return;
+        }
+
+        super.channelRead(ctx, msg);
+    }
+
+    private boolean measureTraffic(ChannelHandlerContext ctx, int bytes) {
+        if (packetsPerSec < 0 && bytesPerSec < 0) return true;
+
+        long time = System.currentTimeMillis();
+
+        if (time - lastPacket >= 1000) {
+            bytesCounter = 0;
+            packetsCounter = 0;
+        }
+
+        packetsCounter++;
+        bytesCounter += bytes;
+
+        if (packetsPerSec > 0 && packetsCounter > packetsPerSec) {
+            closeConnection(ctx, "Closed %s due too frequent packet sending (%d per sec)", ctx.channel().remoteAddress(), packetsCounter);
+            return false;
+        }
+
+        if (bytesPerSec > 0 && bytesCounter > bytesPerSec) {
+            closeConnection(ctx, "Closed %s due too many bytes sent per second (%d per sec)", ctx.channel().remoteAddress(), bytesCounter);
+            return false;
+        }
+
+        lastPacket = time;
+
+        return true;
+    }
+
+    private void closeConnection(ChannelHandlerContext ctx, String reason, Object... args) {
+        ctx.close();
+        Logger.info(reason, args);
+    }
+}

src/main/java/ua/nanit/limbo/connection/pipeline/VarIntFrameDecoder.java

@@ -47,6 +47,7 @@ public class VarIntFrameDecoder extends ByteToMessageDecoder {
                 in.readerIndex(varIntEnd + 1);
             } else {
                 int minimumRead = bytesRead + readVarInt;
+
                 if (in.isReadable(minimumRead)) {
                     out.add(in.retainedSlice(varIntEnd + 1, readVarInt));
                     in.skipBytes(minimumRead);

src/main/resources/settings.yml

@@ -112,4 +112,20 @@ netty:
   # EventLoopGroup threads count
   threads:
     bossGroup: 1
-    workerGroup: 4
+    workerGroup: 4
+
+# [Experimental]
+# Options to check incoming traffic and kick potentially malicious connections.
+# Take into account that player can send many small packets, for example just moving mouse.
+traffic:
+  # If true, then additional handler will be added to channel pipeline
+  enable: false
+  # Max packet size in bytes
+  # Unlimited if -1
+  packetSize: 1024
+  # How many packets per second allowed for single connection
+  # Ignored if -1
+  packets: -1
+  # How many bytes per second allowed for single connection
+  # Ignored if -1
+  bytes: 2048