[fix] 添加客户端渗透客户端流量记录

wu-lazy-cloud-heartbeat-client/src/main/java/org/framework/lazy/cloud/network/heartbeat/client/netty/socket/NettyClientPermeateClientRealSocket.java

@@ -150,6 +150,8 @@ public class NettyClientPermeateClientRealSocket {
                 ChannelAttributeKeyUtils.buildVisitorId(transferChannel, visitorId);
                 ChannelAttributeKeyUtils.buildClientId(transferChannel, clientId);
 
+                ChannelAttributeKeyUtils.buildVisitorPort(transferChannel, visitorPort);
+
 
 
             } else {

wu-lazy-cloud-heartbeat-client/src/main/java/org/framework/lazy/cloud/network/heartbeat/client/netty/socket/NettyClientPermeateClientVisitorSocket.java

@@ -14,6 +14,7 @@ import org.framework.lazy.cloud.network.heartbeat.common.NettyVisitorPortContext
 import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.HandleChannelTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.socket.PermeateVisitorSocket;
+import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
 
 import java.util.List;
 
@@ -72,9 +73,11 @@ public class NettyClientPermeateClientVisitorSocket implements PermeateVisitorSo
             try {
                  bootstrap.bind(visitorPort).sync().addListener((ChannelFutureListener) future -> {
                     if (future.isSuccess()) {
+                        Channel channel = future.channel();
+                        ChannelAttributeKeyUtils.buildVisitorPort(channel,visitorPort);
                         // 这里时异步处理
                         log.info("客户端:[{}]访客端口:[{}] 开启", clientId, visitorPort);
-                        NettyVisitorPortContext.pushVisitorChannel(visitorPort, future.channel());
+                        NettyVisitorPortContext.pushVisitorChannel(visitorPort, channel);
 
                     } else {
                         log.error("客户端:[{}]访客端口:[{}]绑定失败", clientId, visitorPort);

wu-lazy-cloud-heartbeat-client/src/main/java/org/framework/lazy/cloud/network/heartbeat/client/netty/socket/NettyClientPermeateServerVisitorSocket.java

@@ -13,6 +13,7 @@ import org.framework.lazy.cloud.network.heartbeat.common.NettyClientVisitorConte
 import org.framework.lazy.cloud.network.heartbeat.common.NettyVisitorPortContext;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.HandleChannelTypeAdvanced;
 import org.framework.lazy.cloud.network.heartbeat.common.socket.PermeateVisitorSocket;
+import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
 
 import java.util.List;
 
@@ -73,7 +74,8 @@ public class NettyClientPermeateServerVisitorSocket implements PermeateVisitorSo
                     if (future.isSuccess()) {
                         // 这里时异步处理
                         log.info("客户端:[{}]访客端口:[{}] 开启", clientId, visitorPort);
-
+                        Channel channel = future.channel();
+                        ChannelAttributeKeyUtils.buildVisitorPort(channel,visitorPort);
                     } else {
                         log.error("客户端:[{}]访客端口:[{}]绑定失败", clientId, visitorPort);
                     }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/advanced/ServerHandleReportClientPermeateServerTransferTypeAdvanced.java

@@ -36,7 +36,7 @@ public class ServerHandleReportClientPermeateServerTransferTypeAdvanced extends
         String clientId = new String(msg.getClientId());
         Integer visitorPort = Integer.valueOf(new String(msg.getVisitorPort()));
         byte[] visitorId = msg.getVisitorId();
-        log.info("【客户端渗透服务端】访客ID:【{}】 访客端口:[{}] 接收到客户端:[{}] 接收服务端数据大小:[{}]", new String(visitorId), visitorPort, clientId, msg.getData().length);
+        log.debug("【客户端渗透服务端】访客ID:【{}】 访客端口:[{}] 接收到客户端:[{}] 接收服务端数据大小:[{}]", new String(visitorId), visitorPort, clientId, msg.getData().length);
         log.debug("【客户端渗透服务端】访客ID:【{}】接收到客户端:[{}] 接收服务端数据大小:[{}] 接收服务端数据:[{}]", new String(visitorId), clientId, msg.getData().length, new String(msg.getData()));
         // 将数据转发访客通道
 

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/advanced/ServerHandleReportClientTransferClientPermeateChannelConnectionSuccessfulTypeAdvanced.java

@@ -7,6 +7,7 @@ import org.framework.lazy.cloud.network.heartbeat.common.advanced.server.Abstrac
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
 import org.springframework.stereotype.Component;
 
+import java.util.Arrays;
 import java.util.List;
 
 /**
@@ -31,6 +32,7 @@ public class ServerHandleReportClientTransferClientPermeateChannelConnectionSucc
         byte[] clientId = nettyProxyMsg.getClientId();// 目标客户端ID
 
         ChannelAttributeKeyUtils.buildClientId(transferChannel, clientId);
+        ChannelAttributeKeyUtils.buildVisitorPort(transferChannel, Integer.parseInt(new String(msgVisitorPort)));
         ChannelAttributeKeyUtils.buildVisitorId(transferChannel, msgVisitorId);
         // 绑定访客通道
         NettyTransferChannelContext.pushVisitor(transferChannel,msgVisitorId);

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/advanced/ServerHandleReportClientTransferClientPermeateChannelInitSuccessfulTypeAdvanced.java

@@ -30,7 +30,7 @@ public class ServerHandleReportClientTransferClientPermeateChannelInitSuccessful
         byte[] msgClientTargetPort = nettyProxyMsg.getClientTargetPort();
         byte[] clientId = nettyProxyMsg.getClientId();
 
-
+        ChannelAttributeKeyUtils.buildVisitorPort(transferChannel, Integer.parseInt(new String(msgVisitorPort)));
         // next translation
         Channel nextTransferChannel = NettyTransferChannelContext.getVisitor(msgVisitorId);
         ChannelAttributeKeyUtils.buildTransferNextChannel(nextTransferChannel,transferChannel);

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/advanced/ServerHandleReportClientTransferClientResponseTypeAdvanced.java

@@ -4,9 +4,13 @@ import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.MessageType;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.server.AbstractHandleReportClientTransferClientResponseTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.ServerChannelFlow;
 import org.springframework.stereotype.Component;
+import org.wu.framework.spring.utils.SpringContextHolder;
 
 /**
  * 上报客户端渗透客户端数据传输结果
@@ -27,7 +31,7 @@ public class ServerHandleReportClientTransferClientResponseTypeAdvanced extends
         byte[] msgVisitorPort = nettyProxyMsg.getVisitorPort();
         byte[] msgClientTargetIp = nettyProxyMsg.getClientTargetIp();
         byte[] msgClientTargetPort = nettyProxyMsg.getClientTargetPort();
-        byte[] clientId = nettyProxyMsg.getClientId();
+        String clientId = nettyProxyMsg.getClientIdString();
         // 下发客户端初始化成功
         Channel nextChannel = ChannelAttributeKeyUtils.getTransferNextChannel(channel);
 
@@ -42,8 +46,19 @@ public class ServerHandleReportClientTransferClientResponseTypeAdvanced extends
         clientConnectTagetNettyProxyMsg.setType(MessageType.DISTRIBUTE_CLIENT_PERMEATE_CLIENT_TRANSFER_RESPONSE);
         if (nextChannel != null) {
             nextChannel.writeAndFlush(clientConnectTagetNettyProxyMsg);
+
+            //记录出口流量
+            ChannelFlowAdapter channelFlowAdapter = SpringContextHolder.getBean(ChannelFlowAdapter.class);
+            ServerChannelFlow serverChannelFlow = ServerChannelFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.OUT_FLOW)
+                    .port(Integer.parseInt(new String(msgVisitorPort)))
+                    .clientId(clientId)
+                    .flow(clientConnectTagetNettyProxyMsg.getData().length)
+                    .build();
+            channelFlowAdapter.asyncHandler(channel, serverChannelFlow);
         }else {
-            log.error("can not find the client:【】 channel",clientId);
+            log.error("can not find the client:【{}】 channel",clientId);
         }
 
 

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/advanced/ServerHandleReportClientTransferClientTypeAdvanced.java

@@ -4,9 +4,13 @@ import io.netty.channel.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.MessageType;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelFlowAdapter;
 import org.framework.lazy.cloud.network.heartbeat.common.advanced.server.AbstractHandleReportClientTransferClientTypeAdvanced;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.ServerChannelFlow;
 import org.springframework.stereotype.Component;
+import org.wu.framework.spring.utils.SpringContextHolder;
 
 /**
  * 上报客户端渗透客户端数据传输
@@ -27,9 +31,10 @@ public class ServerHandleReportClientTransferClientTypeAdvanced extends Abstract
         byte[] msgVisitorPort = nettyProxyMsg.getVisitorPort();
         byte[] msgClientTargetIp = nettyProxyMsg.getClientTargetIp();
         byte[] msgClientTargetPort = nettyProxyMsg.getClientTargetPort();
-        byte[] clientId = nettyProxyMsg.getClientId();
+        String clientId = nettyProxyMsg.getClientIdString();
         // 下发客户端初始化成功
         Channel nextChannel = ChannelAttributeKeyUtils.getTransferNextChannel(channel);
+        Integer visitorPort = ChannelAttributeKeyUtils.getVisitorPort(channel);
 
 
         NettyProxyMsg clientConnectTagetNettyProxyMsg = new NettyProxyMsg();
@@ -42,11 +47,18 @@ public class ServerHandleReportClientTransferClientTypeAdvanced extends Abstract
         clientConnectTagetNettyProxyMsg.setType(MessageType.DISTRIBUTE_CLIENT_PERMEATE_CLIENT_TRANSFER_REQUEST);
         if (nextChannel != null) {
             nextChannel.writeAndFlush(clientConnectTagetNettyProxyMsg);
+            // 记录进口数据
+            ChannelFlowAdapter channelFlowAdapter = SpringContextHolder.getBean(ChannelFlowAdapter.class);
+            ServerChannelFlow serverChannelFlow = ServerChannelFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.IN_FLOW)
+                    .port(visitorPort)
+                    .clientId(clientId)
+                    .flow(clientConnectTagetNettyProxyMsg.getData().length)
+                    .build();
+            channelFlowAdapter.asyncHandler(channel, serverChannelFlow);
         }else {
-            log.error("can not find the client:【】 channel",clientId);
+            log.error("can not find the client:【{}】 channel",clientId);
         }
-
-
-
     }
 }

wu-lazy-cloud-heartbeat-server/src/main/java/org/framework/lazy/cloud/network/heartbeat/server/netty/handler/NettyClientPermeateServerRealHandler.java

@@ -9,7 +9,11 @@ import lombok.extern.slf4j.Slf4j;
 import org.framework.lazy.cloud.network.heartbeat.common.MessageType;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyByteBuf;
 import org.framework.lazy.cloud.network.heartbeat.common.NettyProxyMsg;
+import org.framework.lazy.cloud.network.heartbeat.common.adapter.ChannelFlowAdapter;
+import org.framework.lazy.cloud.network.heartbeat.common.enums.ChannelFlowEnum;
 import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeKeyUtils;
+import org.framework.lazy.cloud.network.heartbeat.server.netty.flow.ServerChannelFlow;
+import org.wu.framework.spring.utils.SpringContextHolder;
 
 /**
  * 客户端渗透服务端
@@ -17,6 +21,7 @@ import org.framework.lazy.cloud.network.heartbeat.common.utils.ChannelAttributeK
 @Slf4j
 public class NettyClientPermeateServerRealHandler extends SimpleChannelInboundHandler<NettyByteBuf> {
 
+
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         // 根据访客ID 确认真实通道 读写打开
@@ -38,13 +43,31 @@ public class NettyClientPermeateServerRealHandler extends SimpleChannelInboundHa
         log.debug("【客户端渗透服务端】服务端真实服务数据:{}", new String(bytes));
         Channel nextChannel = ChannelAttributeKeyUtils.getNextChannel(channel);
         // 消息下发到客户端
+        String clientId = ChannelAttributeKeyUtils.getClientId(channel);
+        Integer visitorPort = ChannelAttributeKeyUtils.getVisitorPort(channel);
+
+        if (nextChannel != null) {
+            NettyProxyMsg nettyMsg = new NettyProxyMsg();
+            nettyMsg.setType(MessageType.DISTRIBUTE_CLIENT_PERMEATE_SERVER_TRANSFER);
+            nettyMsg.setData(bytes);
+
+            nextChannel.writeAndFlush(nettyMsg);
+            ChannelFlowAdapter channelFlowAdapter = SpringContextHolder.getBean(ChannelFlowAdapter.class);
+            // 记录出口数据
+            ServerChannelFlow serverChannelFlow = ServerChannelFlow
+                    .builder()
+                    .channelFlowEnum(ChannelFlowEnum.OUT_FLOW)
+                    .port(visitorPort)
+                    .clientId(clientId)
+                    .flow(nettyMsg.getData().length)
+                    .build();
+            channelFlowAdapter.asyncHandler(channel, serverChannelFlow);
+        } else {
+            log.error("we can not find next channel  for transfer  with client permeate server user client_id:{} ", clientId);
+        }
+
 
-        NettyProxyMsg nettyMsg = new NettyProxyMsg();
-        nettyMsg.setType(MessageType.DISTRIBUTE_CLIENT_PERMEATE_SERVER_TRANSFER);
-        nettyMsg.setData(bytes);
 
-        nextChannel.writeAndFlush(nettyMsg);
-        // 记录出口数据
     }