feat: 完成tcpmuxserver的解码逻辑、并添加相应的单元测试

Author: meethigher

src/main/java/top/meethigher/proxy/tcp/mux/Mux.java

@@ -27,7 +27,9 @@ public abstract class Mux {
     /**
      * 默认对称加密密钥
      */
-    protected static final String SECRET_DEFAULT = "1234567890abcdef";
+    public static final String SECRET_DEFAULT = "1234567890abcdef";
+
+    public static final short type = 5209;
 
     protected static final char[] ID_CHARACTERS = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ".toCharArray();
 
@@ -49,7 +51,7 @@ public Buffer aesBase64Encode(NetAddress netAddress) {
         String addr = netAddress.toString();
         SecretKey key = restoreKey(secret.getBytes(StandardCharsets.UTF_8));
         String encryptedAddr = encryptToBase64(addr.getBytes(StandardCharsets.UTF_8), key);
-        return TunnelMessageCodec.encode((short) 0, encryptedAddr.getBytes(StandardCharsets.UTF_8));
+        return TunnelMessageCodec.encode(type, encryptedAddr.getBytes(StandardCharsets.UTF_8));
     }
 
     /**
@@ -64,4 +66,5 @@ public NetAddress aesBase64Decode(Buffer buffer) {
         return NetAddress.parse(addr);
     }
 
+
 }

src/main/java/top/meethigher/proxy/tcp/mux/MuxMessageParser.java

@@ -0,0 +1,76 @@
+package top.meethigher.proxy.tcp.mux;
+
+import io.vertx.core.Handler;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.net.NetSocket;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageParser;
+
+/**
+ * 适用于TcpMux的通信消息解析器
+ * 只解析第一条数据
+ *
+ * @author <a href="https://meethigher.top">chenchuancheng</a>
+ * @since 2025/07/27 02:19
+ */
+public class MuxMessageParser extends TunnelMessageParser {
+
+    public static final class MuxMessage {
+        // muxclient发送给muxserver的backendServer配置信息
+        public final Buffer backendServerBuf;
+        // user发给muxclient的信息
+        public final Buffer payload;
+
+        public MuxMessage(Buffer backendServerBuf, Buffer payload) {
+            this.backendServerBuf = backendServerBuf;
+            this.payload = payload;
+        }
+    }
+
+
+    private static final Logger log = LoggerFactory.getLogger(MuxMessageParser.class);
+
+    protected final Handler<MuxMessage> muxMessageHandler;
+
+    public MuxMessageParser(Handler<MuxMessage> muxMessageHandler, NetSocket netSocket) {
+        // 用不到这个父级的handler，传null即可
+        super(null, netSocket);
+        this.muxMessageHandler = muxMessageHandler;
+    }
+
+
+    @Override
+    protected void parse() {
+        // 获取消息的预设总长度
+        int totalLength = buf.getInt(lengthFieldOffset);
+        // 校验预设总长度
+        if (buf.length() < totalLength) {
+            return;
+        }
+        // 校验最大长度
+        if (totalLength > maxLength) {
+            log.warn("too many bytes in length field, {} > {}, connection {} -- {} will be closed",
+                    totalLength, maxLength,
+                    netSocket.localAddress(),
+                    netSocket.remoteAddress());
+            netSocket.close();
+            return;
+        }
+        // 校验类型编码是否在预设范围内
+        if (totalLength >= (lengthFieldLength + typeFieldLength)) {
+            short code = buf.getShort(lengthFieldLength);
+            if (Mux.type == code) {
+
+            } else {
+                log.warn("invalid type, connection {} -- {} will be closed",
+                        netSocket.localAddress(),
+                        netSocket.remoteAddress());
+                netSocket.close();
+                return;
+            }
+
+        }
+        muxMessageHandler.handle(new MuxMessage(buf.getBuffer(0, totalLength), buf.getBuffer(totalLength, buf.length())));
+    }
+}

src/main/java/top/meethigher/proxy/tcp/mux/ReverseTcpProxyMuxServer.java

@@ -1,6 +1,12 @@
 package top.meethigher.proxy.tcp.mux;
 
 import io.vertx.core.Vertx;
+import io.vertx.core.net.NetClient;
+import io.vertx.core.net.NetServer;
+import io.vertx.core.net.NetSocket;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import top.meethigher.proxy.NetAddress;
 
 import java.util.concurrent.ThreadLocalRandom;
 
@@ -15,12 +21,54 @@
  */
 public class ReverseTcpProxyMuxServer extends Mux {
 
-//    protected final NetServer netServer;
-//    protected final NetClient netClient;
-//    protected final String name;
+    private static final Logger log = LoggerFactory.getLogger(ReverseTcpProxyMuxServer.class);
+    protected final NetServer netServer;
+    protected final NetClient netClient;
+    protected final String name;
 
-    public ReverseTcpProxyMuxServer(Vertx vertx, String secret) {
+    protected String host = "0.0.0.0";
+    protected int port = 997;
+
+    protected ReverseTcpProxyMuxServer(Vertx vertx, String secret, NetServer netServer, NetClient netClient, String name) {
         super(vertx, secret);
+        this.netServer = netServer;
+        this.netClient = netClient;
+        this.name = name;
+    }
+
+    protected void handleConnect(NetSocket src) {
+        src.pause();
+        log.debug("source {} -- {} connected", src.localAddress(), src.remoteAddress());
+        src.exceptionHandler(e -> log.error("source {} -- {} exception occurred", src.localAddress(), src.remoteAddress(), e))
+                .closeHandler(v -> log.debug("source {} -- {} closed", src.localAddress(), src.remoteAddress()));
+        src.handler(new MuxMessageParser(muxMsg -> this.bindMuxConnections(src, muxMsg), src));
+        src.resume();
+    }
+
+    /**
+     * 根据{@code MuxMessage }建立后端连接，并将数据连接和后端连接进行绑定
+     */
+    protected void bindMuxConnections(NetSocket src, MuxMessageParser.MuxMessage muxMsg) {
+        src.pause();
+        NetAddress backend = aesBase64Decode(muxMsg.backendServerBuf);
+        if (backend == null) {
+            log.warn("source {} -- {} exception occurred: failed to parsing the backendServer address from encrypted content:{}",
+                    src.localAddress(), src.remoteAddress(),
+                    muxMsg.backendServerBuf);
+            src.close();
+            return;
+        }
+        netClient.connect(backend.getPort(), backend.getHost())
+                .onFailure(e -> {
+                    log.error("source {} -- {} failed to connect to {}", src.localAddress(), src.remoteAddress(), backend, e);
+                    src.close();
+                })
+                .onSuccess(dst -> {
+                    dst.pause();
+                    log.debug("target {} -- {} connected",dst.localAddress(),dst.remoteAddress());
+
+                });
+
     }
 
     public static String generateName() {
@@ -42,4 +90,20 @@ public static String generateName() {
         }
     }
 
+    public void start() {
+        netServer.connectHandler(this::handleConnect)
+                .exceptionHandler(e -> log.error("{} socket errors happening before the connection is passed to the connectHandler", name, e))
+                .listen(port, host)
+                .onFailure(e -> log.error("{} start failed", name))
+                .onSuccess(v -> log.info("{} started on {}:{}", name, host, port));
+    }
+
+    public void stop() {
+        netServer.close()
+                .onSuccess(v -> log.info("{} closed", name))
+                .onFailure(e -> log.error("{} close failed", name, e));
+
+    }
+
+
 }

src/test/java/top/meethigher/proxy/tcp/mux/MuxMessageParserTest.java

@@ -0,0 +1,81 @@
+package top.meethigher.proxy.tcp.mux;
+
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.net.NetClient;
+import io.vertx.core.net.NetSocket;
+import io.vertx.core.parsetools.RecordParser;
+import org.junit.Test;
+import top.meethigher.proxy.NetAddress;
+
+import java.util.concurrent.locks.LockSupport;
+
+public class MuxMessageParserTest {
+
+
+    /**
+     * 我想验证，如果一次性发了一大串内容。RecordParser可以只读取前1个字节，就pause，当resume可以继续从第2个之后消费
+     * <p>
+     * 结论：不行
+     */
+    @Test
+    public void testRecordParserAndPause() {
+        Vertx vertx = Vertx.vertx();
+        vertx.createNetServer().connectHandler(socket -> {
+            socket.pause();
+            RecordParser parser = RecordParser.newFixed(1);
+            parser.handler(buf -> {
+                System.out.println(buf);
+                socket.pause();
+                vertx.setTimer(5000, id -> {
+                    socket.handler(buf1 -> {
+                        System.out.println(buf1);
+                    });
+                    socket.resume();
+                });
+            });
+            socket.handler(parser);
+            socket.resume();
+        }).listen(777).onFailure(e -> System.exit(1));
+
+        NetClient netClient = vertx.createNetClient();
+        netClient.connect(777, "127.0.0.1").onSuccess(socket -> {
+            socket.write("123456789");
+            vertx.setTimer(1000, id -> {
+                socket.write("abcdefg");
+            });
+        });
+
+        LockSupport.park();
+    }
+
+    @Test
+    public void name() {
+
+        NetSocket socket = null;
+
+        Vertx vertx = Vertx.vertx();
+        Mux mux = new Mux(vertx, Mux.SECRET_DEFAULT) {
+        };
+        NetAddress netAddress1 = new NetAddress("127.0.0.1", 8080);
+
+
+        Buffer buffer1 = mux.aesBase64Encode(netAddress1);
+
+
+        Buffer buffer2 = Buffer.buffer("halo wode");
+
+        Buffer buffer = Buffer.buffer().appendBuffer(buffer1).appendBuffer(buffer2);
+
+
+        MuxMessageParser parser = new MuxMessageParser(msg -> {
+            System.out.println("控制消息：" + mux.aesBase64Decode(msg.backendServerBuf));
+            System.out.println("用户消息：" + msg.payload);
+        }, socket);
+
+        parser.handle(buffer);
+
+
+        LockSupport.park();
+    }
+}