feat: 1. protobuf封装tcp通信编码规范 2. 自定义tcp解码逻辑

(cherry picked from commit 6d860af)

Author: meethigher

src/main/java/top/meethigher/proxy/tcp/tunnel/codec/TunnelMessageCodec.java

@@ -0,0 +1,47 @@
+package top.meethigher.proxy.tcp.tunnel.codec;
+
+import io.vertx.core.buffer.Buffer;
+
+/**
+ * 自定义消息编解码
+ * <pre>
+ * |  4 字节（消息长度） |  2 字节（消息类型） |  变长（消息体）  |
+ * |      0x0010      |      0x0001       |   {"id":1, "msg":"Hello"}  |
+ * </pre>
+ *
+ * @author <a href="https://meethigher.top">chenchuancheng</a>
+ * @since 2025/04/04 23:12
+ */
+public class TunnelMessageCodec {
+
+
+    public static Buffer encode(short type, byte[] body) {
+        int totalLength = 4 + 2 + body.length;
+        Buffer buffer = Buffer.buffer();
+        buffer.appendInt(totalLength);
+        buffer.appendShort(type);
+        buffer.appendBytes(body);
+        return buffer;
+    }
+
+    public static DecodedMessage decode(Buffer buffer) {
+        int totalLength = buffer.getInt(0);
+        short type = buffer.getShort(4);
+        byte[] body = buffer.getBytes(6, buffer.length());
+        return new DecodedMessage(totalLength, type, body);
+    }
+
+    public static class DecodedMessage {
+        public final int totalLength;
+        public final short type;
+        public final byte[] body;
+
+        public DecodedMessage(int totalLength, short type, byte[] body) {
+            this.totalLength = totalLength;
+            this.type = type;
+            this.body = body;
+        }
+    }
+
+
+}

src/main/java/top/meethigher/proxy/tcp/tunnel/codec/TunnelMessageParser.java

@@ -0,0 +1,63 @@
+package top.meethigher.proxy.tcp.tunnel.codec;
+
+import io.vertx.core.Handler;
+import io.vertx.core.buffer.Buffer;
+
+/**
+ * 自定义消息结构解析器
+ * [4字节长度+2字节类型+protobuf变长消息体]
+ *
+ * @author <a href="https://meethigher.top">chenchuancheng</a>
+ * @since 2025/04/04 02:52
+ */
+public class TunnelMessageParser implements Handler<Buffer> {
+
+    private Buffer buf = Buffer.buffer();
+
+    /**
+     * 预设长度起始位置
+     */
+    private final int lengthFieldOffset = 0;
+
+    /**
+     * 预设长度占用的字节数
+     */
+    private final int lengthFieldLength = 4;
+    /**
+     * 消息类型起始位置
+     */
+    private final int typeFieldOffset = 4;
+
+    /**
+     * 消息类型占用的字节数
+     */
+    private final int typeFieldLength = 2;
+
+    /**
+     * 消息体起始位置
+     */
+    private final int bodyFieldOffset = 6;
+
+    private final Handler<Buffer> outputHandler;
+
+    public TunnelMessageParser(Handler<Buffer> outputHandler) {
+        this.outputHandler = outputHandler;
+    }
+
+    @Override
+    public void handle(Buffer buffer) {
+        buf.appendBuffer(buffer);
+        if (buf.length() < lengthFieldLength) {
+            return;
+        } else {
+            int totalLength = buf.getInt(lengthFieldOffset);
+            if (buf.length() < totalLength) {
+                return;
+            } else {
+                outputHandler.handle(buf.getBuffer(0, totalLength));
+                buf = buf.getBuffer(totalLength, buf.length());
+            }
+        }
+
+    }
+}

src/main/java/top/meethigher/proxy/tcp/tunnel/codec/TunnelMessageType.java

@@ -0,0 +1,40 @@
+package top.meethigher.proxy.tcp.tunnel.codec;
+
+/**
+ * 消息类型
+ *
+ * @author <a href="https://meethigher.top">chenchuancheng</a>
+ * @since 2025/04/04 23:11
+ */
+public enum TunnelMessageType {
+    // 客户端 → 服务端
+    AUTH(0x0001),
+    HEARTBEAT(0x0003),
+    OPEN_PORT(0x0005),
+    CONNECT_PORT_ACK(0x0008),
+
+    // 服务端 → 客户端
+    AUTH_ACK(0x0002),
+    HEARTBEAT_ACK(0x0004),
+    OPEN_PORT_ACK(0x0006),
+    CONNECT_PORT(0x0007);
+
+    private final int code;
+
+    TunnelMessageType(int code) {
+        this.code = code;
+    }
+
+    public short code() {
+        return (short) code;
+    }
+
+    public static TunnelMessageType fromCode(short code) {
+        for (TunnelMessageType type : values()) {
+            if (type.code() == code) {
+                return type;
+            }
+        }
+        throw new IllegalArgumentException("Unknown message type: " + code);
+    }
+}

src/main/java/top/meethigher/proxy/tcp/tunnel/handler/AuthHandler.java

@@ -0,0 +1,47 @@
+package top.meethigher.proxy.tcp.tunnel.handler;
+
+import io.vertx.core.net.NetSocket;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageCodec;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageType;
+import top.meethigher.proxy.tcp.tunnel.proto.TunnelMessage;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class AuthHandler extends TunnelHandler {
+
+
+    private static final Logger log = LoggerFactory.getLogger(AuthHandler.class);
+
+    public AuthHandler(NetSocket netSocket, TunnelMessageType type, TunnelMessageType respType, byte[] bodyBytes) {
+        super(netSocket, type, respType, bodyBytes);
+    }
+
+    @Override
+    protected boolean doHandler() {
+        TunnelMessage.AuthAck.Builder builder = TunnelMessage.AuthAck.newBuilder();
+        AtomicBoolean success = new AtomicBoolean(false);
+        String msg = "";
+        try {
+            TunnelMessage.Auth parsed = TunnelMessage.Auth.parseFrom(bodyBytes);
+            String token = parsed.getToken();
+            if ("token".equals(token)) {
+                success.set(true);
+            } else {
+                msg = "auth failed";
+            }
+        } catch (Exception e) {
+            msg = e.getMessage();
+            log.error("Error parsing message: {}", type, e);
+        } finally {
+
+            netSocket.write(TunnelMessageCodec.encode(respType.code(), builder.setSuccess(success.get()).setMessage(msg).build().toByteArray())).onSuccess(v -> {
+                if (!success.get()) {
+                    netSocket.close();
+                }
+            });
+        }
+        return success.get();
+    }
+}

src/main/java/top/meethigher/proxy/tcp/tunnel/handler/HeartbeatHandler.java

@@ -0,0 +1,19 @@
+package top.meethigher.proxy.tcp.tunnel.handler;
+
+import io.vertx.core.net.NetSocket;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageCodec;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageType;
+import top.meethigher.proxy.tcp.tunnel.proto.TunnelMessage;
+
+public class HeartbeatHandler extends TunnelHandler {
+    public HeartbeatHandler(NetSocket netSocket, TunnelMessageType type, TunnelMessageType respType, byte[] bodyBytes) {
+        super(netSocket, type, respType, bodyBytes);
+    }
+
+    @Override
+    protected boolean doHandler() {
+        TunnelMessage.HeartbeatAck ack = TunnelMessage.HeartbeatAck.newBuilder().setTimestamp(System.currentTimeMillis()).build();
+        netSocket.write(TunnelMessageCodec.encode(respType.code(), ack.toByteArray()));
+        return true;
+    }
+}

src/main/java/top/meethigher/proxy/tcp/tunnel/handler/OpenPortHandler.java

@@ -0,0 +1,36 @@
+package top.meethigher.proxy.tcp.tunnel.handler;
+
+import io.vertx.core.net.NetSocket;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageCodec;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageType;
+import top.meethigher.proxy.tcp.tunnel.proto.TunnelMessage;
+
+public class OpenPortHandler extends TunnelHandler {
+    private static final Logger log = LoggerFactory.getLogger(OpenPortHandler.class);
+
+    public OpenPortHandler(NetSocket netSocket, TunnelMessageType type, TunnelMessageType respType, byte[] bodyBytes) {
+        super(netSocket, type, respType, bodyBytes);
+    }
+
+    @Override
+    protected boolean doHandler() {
+        TunnelMessage.OpenPortAck.Builder builder = TunnelMessage.OpenPortAck.newBuilder();
+        boolean success = false;
+        String msg = "";
+        try {
+            TunnelMessage.OpenPort parsed = TunnelMessage.OpenPort.parseFrom(bodyBytes);
+            int port = parsed.getPort();
+            log.info("开启 {} 端口", port);
+            success = true;
+
+        } catch (Exception e) {
+            msg = e.getMessage();
+            log.error("");
+        } finally {
+            netSocket.write(TunnelMessageCodec.encode(respType.code(), builder.setSuccess(success).setMessage(msg).build().toByteArray()));
+        }
+        return success;
+    }
+}

src/main/java/top/meethigher/proxy/tcp/tunnel/handler/TunnelHandler.java

@@ -0,0 +1,35 @@
+package top.meethigher.proxy.tcp.tunnel.handler;
+
+import io.vertx.core.net.NetSocket;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageType;
+
+public abstract class TunnelHandler {
+
+    private static final Logger log = LoggerFactory.getLogger(TunnelHandler.class);
+    protected final NetSocket netSocket;
+
+    protected final TunnelMessageType type;
+
+    protected final TunnelMessageType respType;
+
+    protected final byte[] bodyBytes;
+
+    public TunnelHandler(NetSocket netSocket, TunnelMessageType type, TunnelMessageType respType, byte[] bodyBytes) {
+        this.netSocket = netSocket;
+        this.type = type;
+        this.respType = respType;
+        this.bodyBytes = bodyBytes;
+    }
+
+    public void handler() {
+        boolean result = doHandler();
+        log.debug("received message type = {}, handle result = {}", type, result);
+    }
+
+
+    protected abstract boolean doHandler();
+
+
+}