【fix】 添加socket5代理demo

wu-lazy-cloud-heartbeat-protocol-proxy/src/test/java/org/framework/lazy/cloud/network/heartbeat/protocol/socket5/Client2DestInboundHandler.java

@@ -0,0 +1,60 @@
+package org.framework.lazy.cloud.network.heartbeat.protocol.socket5;
+
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.util.ReferenceCountUtil;
+import lombok.extern.slf4j.Slf4j;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * @author kdyzm
+ * @date 2021-04-23
+ */
+@Slf4j
+public class Client2DestInboundHandler extends ChannelInboundHandlerAdapter {
+
+    private final ChannelFuture dstChannelFuture;
+
+    public Client2DestInboundHandler(ChannelFuture dstChannelFuture) {
+        this.dstChannelFuture = dstChannelFuture;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) {
+        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        log.info("转发客户端的请求到代理服务器");
+        if (dstChannelFuture.channel().isActive()) {
+            dstChannelFuture.channel().writeAndFlush(msg);
+        } else {
+            log.info("释放内存");
+            ReferenceCountUtil.release(msg);
+        }
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        log.info("客户端与代理服务器的连接已经断开，即将断开代理服务器和目标服务器的连接");
+        if (dstChannelFuture.channel().isActive()) {
+            if (ctx.channel().isActive()) {
+                ctx.channel().writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+            }
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        log.error("Client2DestInboundHandler exception", cause);
+        ctx.close();
+    }
+}

wu-lazy-cloud-heartbeat-protocol-proxy/src/test/java/org/framework/lazy/cloud/network/heartbeat/protocol/socket5/Dest2ClientInboundHandler.java

@@ -0,0 +1,53 @@
+package org.framework.lazy.cloud.network.heartbeat.protocol.socket5;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.util.ReferenceCountUtil;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * @author kdyzm
+ * @date 2021-04-24
+ */
+@Slf4j
+public class Dest2ClientInboundHandler extends ChannelInboundHandlerAdapter {
+
+    private final ChannelHandlerContext clientChannelHandlerContext;
+
+    public Dest2ClientInboundHandler(ChannelHandlerContext clientChannelHandlerContext) {
+        this.clientChannelHandlerContext = clientChannelHandlerContext;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) {
+        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        log.trace("开始写回客户端数据");
+        if (clientChannelHandlerContext.channel().isActive()) {
+            clientChannelHandlerContext.writeAndFlush(msg);
+        } else {
+            log.info("释放内存");
+            ReferenceCountUtil.release(msg);
+        }
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        log.trace("代理服务器和目标服务器的连接已经断开，即将断开客户端和代理服务器的连接");
+        if (clientChannelHandlerContext.channel().isActive()) {
+            clientChannelHandlerContext.channel().writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        log.error("Dest2ClientInboundHandler exception", cause);
+        ctx.close();
+    }
+}

wu-lazy-cloud-heartbeat-protocol-proxy/src/test/java/org/framework/lazy/cloud/network/heartbeat/protocol/socket5/Socks5CommandRequestHandler.java

@@ -0,0 +1,66 @@
+package org.framework.lazy.cloud.network.heartbeat.protocol.socket5;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.*;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.socksx.v5.*;
+import io.netty.util.ReferenceCountUtil;
+import lombok.extern.slf4j.Slf4j;
+
+import java.net.InetSocketAddress;
+
+@Slf4j
+public class Socks5CommandRequestHandler extends SimpleChannelInboundHandler<Socks5CommandRequest> {
+    private final EventLoopGroup group;
+
+    public Socks5CommandRequestHandler(EventLoopGroup group) {
+        this.group = group;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, Socks5CommandRequest request) throws Exception {
+        if (request.type() == Socks5CommandType.CONNECT) {
+            Bootstrap b = new Bootstrap();
+            b.group(group)
+                    .channel(NioSocketChannel.class)
+                    .option(ChannelOption.TCP_NODELAY, true)
+                    .handler(new ChannelInitializer<SocketChannel>() {
+                        @Override
+                        protected void initChannel(SocketChannel ch) throws Exception {
+                            ch.pipeline().addLast(new Dest2ClientInboundHandler(ctx));
+                        }
+                    });
+            log.info("准备连接目标服务器，ip={},port={}", request.dstAddr(), request.dstPort());
+            ChannelFuture f = b.connect(new InetSocketAddress(request.dstAddr(), request.dstPort()));
+            f.addListener((ChannelFutureListener) future -> {
+
+                if (future.isSuccess()) {
+                    log.info("目标服务器连接成功");
+                    //添加客户端转发请求到服务端的Handler
+                    ctx.pipeline().addLast(new Client2DestInboundHandler(future));
+                    DefaultSocks5CommandResponse commandResponse = new DefaultSocks5CommandResponse(Socks5CommandStatus.SUCCESS, request.dstAddrType());
+                    ctx.writeAndFlush(commandResponse);
+                    ctx.pipeline().remove(Socks5CommandRequestHandler.class);
+                    ctx.pipeline().remove(Socks5CommandRequestDecoder.class);
+                } else {
+                    log.error("连接目标服务器失败,address={},port={}", request.dstAddr(), request.dstPort());
+                    DefaultSocks5CommandResponse commandResponse = new DefaultSocks5CommandResponse(Socks5CommandStatus.FAILURE, request.dstAddrType());
+                    ctx.writeAndFlush(commandResponse);
+                    future.channel().close();
+                }
+
+            });
+        } else {
+            log.info("receive commandRequest type={}", request.type());
+            ReferenceCountUtil.retain(request);
+            ctx.fireChannelRead(request);
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        cause.printStackTrace();
+        ctx.close();
+    }
+}    

wu-lazy-cloud-heartbeat-protocol-proxy/src/test/java/org/framework/lazy/cloud/network/heartbeat/protocol/socket5/Socks5InitialRequestHandler.java

@@ -0,0 +1,34 @@
+package org.framework.lazy.cloud.network.heartbeat.protocol.socket5;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.socksx.v5.*;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class Socks5InitialRequestHandler extends SimpleChannelInboundHandler<Socks5InitialRequest> {
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, Socks5InitialRequest msg) throws Exception {
+        boolean failure = msg.decoderResult().isFailure();
+        if (failure) {
+            log.error("初始化socks5失败，请检查是否是socks5协议");
+//            ReferenceCountUtil.retain(msg);
+            ctx.fireChannelRead(msg);
+            return;
+        }
+        log.info("初始化socket连接");
+
+        // 不验证账号密码
+        Socks5InitialResponse socks5InitialResponse = new DefaultSocks5InitialResponse(Socks5AuthMethod.NO_AUTH);
+        ctx.writeAndFlush(socks5InitialResponse);
+
+        ctx.pipeline().remove(this);
+        ctx.pipeline().remove(Socks5InitialRequestDecoder.class);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        cause.printStackTrace();
+        ctx.close();
+    }
+}    

wu-lazy-cloud-heartbeat-protocol-proxy/src/test/java/org/framework/lazy/cloud/network/heartbeat/protocol/socket5/Socks5PasswordAuthRequestInboundHandler.java

@@ -0,0 +1,28 @@
+package org.framework.lazy.cloud.network.heartbeat.protocol.socket5;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.socksx.v5.*;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class Socks5PasswordAuthRequestInboundHandler extends SimpleChannelInboundHandler<DefaultSocks5PasswordAuthRequest> {
+
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, DefaultSocks5PasswordAuthRequest msg) throws Exception {
+        //认证成功
+
+        log.info("认证直接成功");
+        Socks5PasswordAuthResponse passwordAuthResponse = new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.SUCCESS);
+        ctx.writeAndFlush(passwordAuthResponse);
+        ctx.pipeline().remove(this);
+        ctx.pipeline().remove(Socks5PasswordAuthRequestDecoder.class);
+        return;
+
+        // 认证失败
+//        Socks5PasswordAuthResponse passwordAuthResponse = new DefaultSocks5PasswordAuthResponse(Socks5PasswordAuthStatus.FAILURE);
+//        //发送鉴权失败消息，完成后关闭channel
+//        ctx.writeAndFlush(passwordAuthResponse).addListener(ChannelFutureListener.CLOSE);
+    }
+}

wu-lazy-cloud-heartbeat-protocol-proxy/src/test/java/org/framework/lazy/cloud/network/heartbeat/protocol/socket5/Socks5ProxyServer.java

@@ -0,0 +1,71 @@
+package org.framework.lazy.cloud.network.heartbeat.protocol.socket5;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.socksx.v5.Socks5CommandRequestDecoder;
+import io.netty.handler.codec.socksx.v5.Socks5InitialRequestDecoder;
+import io.netty.handler.codec.socksx.v5.Socks5PasswordAuthRequestDecoder;
+import io.netty.handler.codec.socksx.v5.Socks5ServerEncoder;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class Socks5ProxyServer {
+    private final int port;
+
+    public Socks5ProxyServer(int port) {
+        this.port = port;
+    }
+
+    public void run() throws Exception {
+        EventLoopGroup bossGroup = new NioEventLoopGroup();
+        EventLoopGroup workerGroup = new NioEventLoopGroup();
+
+        EventLoopGroup group = new NioEventLoopGroup();
+        try {
+            ServerBootstrap b = new ServerBootstrap();
+            b.group(bossGroup, workerGroup)
+                    .channel(NioServerSocketChannel.class)
+                    .childHandler(new ChannelInitializer<SocketChannel>() {
+                        @Override
+                        public void initChannel(SocketChannel ch) throws Exception {
+
+                            //socks5响应最后一个encode
+                            ch.pipeline().addLast(Socks5ServerEncoder.DEFAULT);
+                            // 初始化连接
+                            ch.pipeline().addLast(new Socks5InitialRequestDecoder());
+                            ch.pipeline().addLast(new Socks5InitialRequestHandler());
+
+
+                            // 认证
+//                            ch.pipeline().addLast(new Socks5PasswordAuthRequestDecoder());
+//                            ch.pipeline().addLast(new Socks5PasswordAuthRequestInboundHandler());
+
+                            // 连接请求
+                            ch.pipeline().addLast(new Socks5CommandRequestDecoder());
+                            ch.pipeline().addLast(new Socks5CommandRequestHandler(group));
+                        }
+                    })
+                    .option(ChannelOption.SO_BACKLOG, 512)
+                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
+            ;
+
+            log.info("启动 socket:{}", port);
+            ChannelFuture f = b.bind(port).sync();
+            f.channel().closeFuture().sync();
+
+        } finally {
+            workerGroup.shutdownGracefully();
+            bossGroup.shutdownGracefully();
+        }
+    }
+
+    public static void main(String[] args) throws Exception {
+        new Socks5ProxyServer(1080).run();
+    }
+}    

wu-lazy-cloud-heartbeat-protocol-proxy/src/test/java/org/framework/lazy/cloud/network/heartbeat/protocol/socket5/Socks5ServerConnectHandler.java

@@ -0,0 +1,42 @@
+package org.framework.lazy.cloud.network.heartbeat.protocol.socket5;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+public class Socks5ServerConnectHandler extends SimpleChannelInboundHandler<Object> {
+    private final Channel relayChannel;
+
+    public Socks5ServerConnectHandler(Channel relayChannel) {
+        this.relayChannel = relayChannel;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
+        relayChannel.writeAndFlush(msg);
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        ctx.read();
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        if (relayChannel.isActive()) {
+            relayChannel.writeAndFlush(new io.netty.handler.codec.socksx.v5.DefaultSocks5CommandResponse(
+                    io.netty.handler.codec.socksx.v5.Socks5CommandStatus.FAILURE,
+                    io.netty.handler.codec.socksx.v5.Socks5AddressType.IPv4))
+                    .addListener(ChannelFutureListener.CLOSE);
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        cause.printStackTrace();
+        ctx.close();
+    }
+}