feat: 1. TunnelServer接收到未知消息或者未注册处理Handler的消息时，主动将连接关闭 2. 重新调整消息事件TunnelMessageType

(cherry picked from commit 2e7730121f7d4656509b46076e26ce96cc5be63b)

Author: meethigher

src/main/java/top/meethigher/proxy/tcp/tunnel/ReverseTcpProxyTunnelClient.java

@@ -31,15 +31,20 @@ public class ReverseTcpProxyTunnelClient extends TunnelClient {
     protected static final long HEARTBEAT_DELAY_DEFAULT = 5000;// 毫秒
     protected static final long MIN_DELAY_DEFAULT = 1000;// 毫秒
     protected static final long MAX_DELAY_DEFAULT = 64000;// 毫秒
-    protected static final String TOKEN_DEFAULT = "123456789";
+    protected static final String SECRET_DEFAULT = "123456789";
     protected static final char[] ID_CHARACTERS = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ".toCharArray();
 
 
     protected final long heartbeatDelay;
-    protected final String token;
+    protected final String secret;
     protected final String name;
 
 
+    protected String localHost = "127.0.0.1";
+    protected int localPort = 2222;
+    protected int remotePort = 2222;
+
+
     protected static String generateName() {
         final String prefix = "ReverseTcpProxyTunnelClient-";
         try {
@@ -61,76 +66,88 @@ protected static String generateName() {
 
     protected ReverseTcpProxyTunnelClient(Vertx vertx, NetClient netClient,
                                           long minDelay, long maxDelay, long heartbeatDelay,
-                                          String token, String name) {
+                                          String secret, String name) {
         super(vertx, netClient, minDelay, maxDelay);
         this.heartbeatDelay = heartbeatDelay;
-        this.token = token;
+        this.secret = secret;
         this.name = name;
         addMessageHandler();
     }
 
+    public ReverseTcpProxyTunnelClient localHost(String localHost) {
+        this.localHost = localHost;
+        return this;
+    }
+
+    public ReverseTcpProxyTunnelClient localPort(int localPort) {
+        this.localPort = localPort;
+        return this;
+    }
+
+    public ReverseTcpProxyTunnelClient remotePort(int remotePort) {
+        this.remotePort = remotePort;
+        return this;
+    }
+
     /**
      * 注册内网穿透的监听逻辑
      */
     protected void addMessageHandler() {
         // 监听连接成功事件
-        this.onConnected((vertx, netSocket, buffer) -> emit(TunnelMessageType.AUTH, TunnelMessage.Auth
-                .newBuilder()
-                .setToken(token)
-                .build()
-                .toByteArray()));
-        // 监听授权响应事件
-        this.on(TunnelMessageType.AUTH_ACK, new AbstractTunnelHandler() {
+        this.onConnected((vertx, netSocket, buffer) -> netSocket.write(encode(TunnelMessageType.OPEN_DATA_PORT,
+                TunnelMessage.OpenDataPort.newBuilder()
+                        .setSecret(secret)
+                        .setPort(remotePort)
+                        .build().toByteArray())));
+
+        // 监听授权与开通数据端口事件
+        this.on(TunnelMessageType.OPEN_DATA_PORT_ACK, new AbstractTunnelHandler() {
             @Override
             protected boolean doHandle(Vertx vertx, NetSocket netSocket, TunnelMessageType type, byte[] bodyBytes) {
                 boolean result = false;
                 try {
-                    TunnelMessage.AuthAck ack = TunnelMessage.AuthAck.parseFrom(bodyBytes);
-                    result = ack.getSuccess();
-                    if (result) {
-                        vertx.setTimer(heartbeatDelay, id -> emit(TunnelMessageType.HEARTBEAT,
-                                TunnelMessage.Heartbeat.newBuilder()
-                                        .setTimestamp(System.currentTimeMillis())
-                                        .build()
-                                        .toByteArray()));
+                    TunnelMessage.OpenDataPortAck parsed = TunnelMessage.OpenDataPortAck.parseFrom(bodyBytes);
+                    if (parsed.getSuccess()) {
+                        // 如果认证 + 开通端口成功，那么就需要进行长连接保持，并开启定期心跳。
+                        vertx.setTimer(heartbeatDelay, id -> netSocket.write(encode(TunnelMessageType.HEARTBEAT,
+                                TunnelMessage.Heartbeat.newBuilder().setTimestamp(System.currentTimeMillis()).build().toByteArray())));
+                    } else {
+                        // 如果认证失败，服务端会主动关闭 tcp 连接
+                        log.warn("{} error : {}", TunnelMessageType.OPEN_DATA_PORT_ACK, parsed.getMessage());
                     }
                 } catch (Exception e) {
                 }
                 return result;
             }
         });
-        // 监听心跳响应事件
+
+        // 监听心跳事件
         this.on(TunnelMessageType.HEARTBEAT_ACK, new AbstractTunnelHandler() {
             @Override
             protected boolean doHandle(Vertx vertx, NetSocket netSocket, TunnelMessageType type, byte[] bodyBytes) {
-                try {
-                    vertx.setTimer(heartbeatDelay, id -> emit(TunnelMessageType.HEARTBEAT,
-                            TunnelMessage.Heartbeat.newBuilder()
-                                    .setTimestamp(System.currentTimeMillis())
-                                    .build()
-                                    .toByteArray()));
-                } catch (Exception e) {
-                }
+                // 只要收到心跳，就证明没问题。不用去解析内容。直接开启下一波心跳计划即可。
+                vertx.setTimer(heartbeatDelay, id -> netSocket.write(encode(TunnelMessageType.HEARTBEAT,
+                        TunnelMessage.Heartbeat.newBuilder().setTimestamp(System.currentTimeMillis()).build().toByteArray())));
                 return true;
             }
         });
     }
 
-    public static ReverseTcpProxyTunnelClient create(Vertx vertx, NetClient netClient, long minDelay, long maxDelay, long heartbeatDelay, String token, String name) {
-        return new ReverseTcpProxyTunnelClient(vertx, netClient, minDelay, maxDelay, heartbeatDelay, token, name);
+    public static ReverseTcpProxyTunnelClient create(Vertx vertx, NetClient netClient, long minDelay, long maxDelay, long heartbeatDelay, String secret, String name) {
+        return new ReverseTcpProxyTunnelClient(vertx, netClient, minDelay, maxDelay, heartbeatDelay, secret, name);
     }
 
-    public static ReverseTcpProxyTunnelClient create(Vertx vertx, NetClient netClient, String token) {
-        return new ReverseTcpProxyTunnelClient(vertx, netClient, MIN_DELAY_DEFAULT, MAX_DELAY_DEFAULT, HEARTBEAT_DELAY_DEFAULT, token, generateName());
+    public static ReverseTcpProxyTunnelClient create(Vertx vertx, NetClient netClient, String secret) {
+        return new ReverseTcpProxyTunnelClient(vertx, netClient, MIN_DELAY_DEFAULT, MAX_DELAY_DEFAULT, HEARTBEAT_DELAY_DEFAULT, secret, generateName());
     }
 
 
     public static ReverseTcpProxyTunnelClient create(Vertx vertx, NetClient netClient) {
-        return new ReverseTcpProxyTunnelClient(vertx, netClient, MIN_DELAY_DEFAULT, MAX_DELAY_DEFAULT, HEARTBEAT_DELAY_DEFAULT, TOKEN_DEFAULT, generateName());
+        return new ReverseTcpProxyTunnelClient(vertx, netClient, MIN_DELAY_DEFAULT, MAX_DELAY_DEFAULT, HEARTBEAT_DELAY_DEFAULT, SECRET_DEFAULT, generateName());
     }
 
     public static ReverseTcpProxyTunnelClient create(Vertx vertx) {
-        return new ReverseTcpProxyTunnelClient(vertx, vertx.createNetClient(), MIN_DELAY_DEFAULT, MAX_DELAY_DEFAULT, HEARTBEAT_DELAY_DEFAULT, TOKEN_DEFAULT, generateName());
+        return new ReverseTcpProxyTunnelClient(vertx, vertx.createNetClient(), MIN_DELAY_DEFAULT, MAX_DELAY_DEFAULT, HEARTBEAT_DELAY_DEFAULT, SECRET_DEFAULT, generateName());
     }
 
 

src/main/java/top/meethigher/proxy/tcp/tunnel/ReverseTcpProxyTunnelServer.java

@@ -33,20 +33,20 @@ public class ReverseTcpProxyTunnelServer extends TunnelServer {
 
     private static final Logger log = LoggerFactory.getLogger(ReverseTcpProxyTunnelServer.class);
     protected static final char[] ID_CHARACTERS = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ".toCharArray();
-    protected static final String TOKEN_DEFAULT = "123456789";
+    protected static final String SECRECT_TOKEN = "123456789";
 
 
     protected String host = "0.0.0.0";
     protected int port = 44444;
     protected Set<NetSocket> authedSockets = new HashSet<>(); // 授权成功的socket列表
 
-    protected final String token;
+    protected final String secret;
     protected final String name;
     protected final Handler<NetSocket> connectHandler;
 
-    public ReverseTcpProxyTunnelServer(Vertx vertx, NetServer netServer, String token, String name) {
+    public ReverseTcpProxyTunnelServer(Vertx vertx, NetServer netServer, String secret, String name) {
         super(vertx, netServer);
-        this.token = token;
+        this.secret = secret;
         this.name = name;
         this.connectHandler = socket -> {
             socket.pause();
@@ -71,79 +71,46 @@ protected void addMessageHandler() {
         // 监听连接成功事件
         this.onConnected((vertx1, netSocket, buffer) -> log.debug("{} connected", netSocket.remoteAddress()));
 
-        // 监听授权消息
-        this.on(TunnelMessageType.AUTH, new AbstractTunnelHandler() {
+        // 监听授权与开通端口事件
+        this.on(TunnelMessageType.OPEN_DATA_PORT, new AbstractTunnelHandler() {
             @Override
             protected boolean doHandle(Vertx vertx, NetSocket netSocket, TunnelMessageType type, byte[] bodyBytes) {
+                // 如果授权通过，并且成功开通端口。则返回成功；否则则返回失败，并关闭连接
                 boolean result = false;
                 try {
-                    TunnelMessage.Auth parsed = TunnelMessage.Auth.parseFrom(bodyBytes);
-                    if (token.equals(parsed.getToken())) {
-                        result = true;
-                        netSocket.write(encode(TunnelMessageType.AUTH_ACK,
-                                        TunnelMessage.AuthAck.newBuilder()
-                                                .setSuccess(result)
-                                                .setMessage("success")
-                                                .build().toByteArray()))
-                                .onComplete(ar -> {
-                                    authedSockets.add(netSocket);
-                                });
+                    TunnelMessage.OpenDataPort parsed = TunnelMessage.OpenDataPort.parseFrom(bodyBytes);
+                    if (secret.equals(parsed.getSecret())) {
+                        final int tPort = parsed.getPort();
+                        
                     } else {
-                        netSocket.write(encode(TunnelMessageType.AUTH_ACK,
-                                        TunnelMessage.AuthAck.newBuilder()
-                                                .setSuccess(result)
-                                                .setMessage("failure")
-                                                .build().toByteArray()))
-                                .onComplete(ar -> {
-                                    // 鉴权失败，服务端主动关闭连接
-                                    netSocket.close();
-                                });
+                        netSocket.write(encode(TunnelMessageType.OPEN_DATA_PORT_ACK,
+                                TunnelMessage.OpenDataPortAck
+                                        .newBuilder()
+                                        .setSuccess(result)
+                                        .setMessage("your secret is incorrect!")
+                                        .build().toByteArray())).onComplete(ar -> netSocket.close());
                     }
                 } catch (Exception e) {
                 }
                 return result;
             }
         });
-
-        // 监听心跳
-        this.on(TunnelMessageType.HEARTBEAT, new AbstractTunnelHandler() {
-            @Override
-            protected boolean doHandle(Vertx vertx, NetSocket netSocket, TunnelMessageType type, byte[] bodyBytes) {
-                if (authedSockets.contains(netSocket)) {
-                    netSocket.write(encode(TunnelMessageType.HEARTBEAT_ACK,
-                            TunnelMessage.HeartbeatAck.newBuilder().setTimestamp(System.currentTimeMillis()).build().toByteArray()));
-                    return true;
-                } else {
-                    // 未经授权的连接，直接关闭
-                    netSocket.close();
-                    return false;
-                }
-            }
-        });
-
-        // 监听开通端口请求
-        this.on(TunnelMessageType.OPEN_PORT, new AbstractTunnelHandler() {
-            @Override
-            protected boolean doHandle(Vertx vertx, NetSocket netSocket, TunnelMessageType type, byte[] bodyBytes) {
-                return false;
-            }
-        });
     }
 
-    public static ReverseTcpProxyTunnelServer create(Vertx vertx, NetServer netServer, String token, String name) {
-        return new ReverseTcpProxyTunnelServer(vertx, netServer, token, name);
+    public static ReverseTcpProxyTunnelServer create(Vertx vertx, NetServer netServer, String secret, String name) {
+        return new ReverseTcpProxyTunnelServer(vertx, netServer, secret, name);
     }
 
-    public static ReverseTcpProxyTunnelServer create(Vertx vertx, NetServer netServer, String token) {
-        return new ReverseTcpProxyTunnelServer(vertx, netServer, token, generateName());
+    public static ReverseTcpProxyTunnelServer create(Vertx vertx, NetServer netServer, String secret) {
+        return new ReverseTcpProxyTunnelServer(vertx, netServer, secret, generateName());
     }
 
     public static ReverseTcpProxyTunnelServer create(Vertx vertx, NetServer netServer) {
-        return new ReverseTcpProxyTunnelServer(vertx, netServer, TOKEN_DEFAULT, generateName());
+        return new ReverseTcpProxyTunnelServer(vertx, netServer, SECRECT_TOKEN, generateName());
     }
 
     public static ReverseTcpProxyTunnelServer create(Vertx vertx) {
-        return new ReverseTcpProxyTunnelServer(vertx, vertx.createNetServer(), TOKEN_DEFAULT, generateName());
+        return new ReverseTcpProxyTunnelServer(vertx, vertx.createNetServer(), SECRECT_TOKEN, generateName());
     }
 
 

src/main/java/top/meethigher/proxy/tcp/tunnel/Tunnel.java

@@ -1,7 +1,10 @@
 package top.meethigher.proxy.tcp.tunnel;
 
+import io.vertx.core.Vertx;
 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.TunnelMessageCodec;
 import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageParser;
 import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageType;
@@ -19,6 +22,13 @@
  */
 public abstract class Tunnel {
 
+    private static final Logger log = LoggerFactory.getLogger(Tunnel.class);
+    protected final Vertx vertx;
+
+    protected Tunnel(Vertx vertx) {
+        this.vertx = vertx;
+    }
+
     /**
      * 监听消息类型 {@code TunnelMessageType} 以及触发的动作 {@code TunnelHandler}
      */
@@ -27,16 +37,55 @@ public abstract class Tunnel {
     /**
      * 监听socket连接成功触发的动作
      */
-    public abstract void onConnected(TunnelHandler tunnelHandler);
+    public void onConnected(TunnelHandler tunnelHandler) {
+        tunnelHandlers.put(null, tunnelHandler);
+    }
 
     /**
      * 监听 {@code TunnelMessageType} 事件
      */
-    public abstract void on(TunnelMessageType type, TunnelHandler tunnelHandler);
+    public void on(TunnelMessageType type, TunnelHandler tunnelHandler) {
+        tunnelHandlers.put(type, tunnelHandler);
+    }
 
+    /**
+     * 将请求体按照{@code TunnelMessageCodec} 规范进行编码
+     *
+     * @param type      消息类型
+     * @param bodyBytes 请求体
+     * @return 编码结果
+     */
     public Buffer encode(TunnelMessageType type, byte[] bodyBytes) {
         return TunnelMessageCodec.encode(type.code(), bodyBytes);
     }
 
-    public abstract TunnelMessageParser decode(NetSocket socket);
+    /**
+     * 将请求体按照 {@code TunnelMessageCodec } 进行解码
+     * <p>
+     * 若传输过来的数据，不符合要求，就会将连接关闭，该操作适合 Server 端的安全防护，Client 端完全没必要这么严格。因此，Client 如果使用该方法的话，建议重写
+     *
+     * @param socket 连接
+     * @return 解码器
+     */
+    public TunnelMessageParser decode(final NetSocket socket) {
+        return new TunnelMessageParser(buffer -> {
+            TunnelMessageCodec.DecodedMessage decodedMessage = TunnelMessageCodec.decode(buffer);
+            TunnelMessageType type = TunnelMessageType.fromCode(decodedMessage.type);
+            for (TunnelMessageType tunnelMessageType : tunnelHandlers.keySet()) {
+                if (type == tunnelMessageType) {
+                    TunnelHandler tunnelHandler = tunnelHandlers.get(tunnelMessageType);
+                    if (tunnelHandler != null) {
+                        tunnelHandler.handle(vertx, socket, buffer);
+                    } else {
+                        log.debug("no tunnel handler for {}, connection {} will be closed", type, socket.remoteAddress());
+                        socket.close();
+                    }
+                    return;
+                }
+            }
+            log.debug("no tunnel handler for {}, connection {} will be closed", type, socket.remoteAddress());
+            socket.close();
+            return;
+        }, socket);
+    }
 }

src/main/java/top/meethigher/proxy/tcp/tunnel/TunnelClient.java

@@ -27,8 +27,6 @@ public abstract class TunnelClient extends Tunnel {
      */
     protected long reconnectDelay;
 
-    protected final Vertx vertx;
-
 
     protected final NetClient netClient;
 
@@ -51,24 +49,13 @@ public abstract class TunnelClient extends Tunnel {
 
 
     protected TunnelClient(Vertx vertx, NetClient netClient, long minDelay, long maxDelay) {
-        this.vertx = vertx;
+        super(vertx);
         this.netClient = netClient;
         this.minDelay = minDelay;
         this.maxDelay = maxDelay;
         setReconnectDelay(this.minDelay);
     }
 
-
-    @Override
-    public void onConnected(TunnelHandler tunnelHandler) {
-        tunnelHandlers.put(null, tunnelHandler);
-    }
-
-    @Override
-    public void on(TunnelMessageType type, TunnelHandler tunnelHandler) {
-        tunnelHandlers.put(type, tunnelHandler);
-    }
-
     @Override
     public TunnelMessageParser decode(final NetSocket socket) {
         return new TunnelMessageParser(buffer -> {
@@ -80,6 +67,7 @@ public TunnelMessageParser decode(final NetSocket socket) {
                     if (tunnelHandler != null) {
                         tunnelHandler.handle(vertx, socket, buffer);
                     }
+                    return;
                 }
             }
         }, socket);

src/main/java/top/meethigher/proxy/tcp/tunnel/TunnelDataServer.java

@@ -0,0 +1,4 @@
+package top.meethigher.proxy.tcp.tunnel;
+
+public abstract class TunnelDataServer {
+}