yeyinlaoying/neutrino-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

服务端收到UDP数据包转发至客户端调通

Browse Source
This commit is contained in:
aoshiguchen
2023-09-20 11:57:00 +08:00
parent 6880898321
commit fd306ce8c3
17 changed files with 365 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/dal/PortMappingMapper.java
+8
View File
@@ -74,6 +74,14 @@ public interface PortMappingMapper extends BaseMapper<PortMappingDO> {
);
}
default PortMappingDO findByLicenseIdAndServerPort(Integer licenseId, Integer serverPort) {
return this.selectOne(new LambdaQueryWrapper<PortMappingDO>()
.eq(PortMappingDO::getLicenseId, licenseId)
.eq(PortMappingDO::getServerPort, serverPort)
.last("limit 1")
);
}
default void updateOnlineStatus(Integer licenseId,Integer serverPort, Integer isOnline, Date updateTime) {
this.update(null, new LambdaUpdateWrapper<PortMappingDO>()
.eq(PortMappingDO::getLicenseId, licenseId)
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/TcpVisitorChannelHandler.java
+2 -1
View File
@@ -4,6 +4,7 @@ import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
import org.dromara.neutrinoproxy.server.proxy.domain.VisitorChannelAttachInfo;
import org.dromara.neutrinoproxy.server.service.FlowReportService;
import org.dromara.neutrinoproxy.server.util.ProxyUtil;
@@ -76,7 +77,7 @@ public class TcpVisitorChannelHandler extends SimpleChannelInboundHandler<ByteBu
visitorChannel.config().setOption(ChannelOption.AUTO_READ, false);
String visitorId = ProxyUtil.newVisitorId();
ProxyUtil.addVisitorChannelToCmdChannel(cmdChannel, visitorId, visitorChannel, sa.getPort());
ProxyUtil.addVisitorChannelToCmdChannel(NetworkProtocolEnum.TCP, cmdChannel, visitorId, visitorChannel, sa.getPort());
cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(visitorId).setData(lanInfo.getBytes()));
super.channelActive(ctx);
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/core/UdpVisitorChannelHandler.java
+61 -10
View File
@@ -6,8 +6,14 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import lombok.extern.slf4j.Slf4j;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
import org.dromara.neutrinoproxy.server.proxy.domain.VisitorChannelAttachInfo;
import org.dromara.neutrinoproxy.server.service.FlowReportService;
import org.dromara.neutrinoproxy.server.util.ProxyUtil;
import org.noear.solon.Solon;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
@@ -16,12 +22,46 @@ import java.nio.charset.StandardCharsets;
* @author: aoshiguchen
* @date: 2023/9/16
*/
@Slf4j
public class UdpVisitorChannelHandler extends SimpleChannelInboundHandler<DatagramPacket> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket datagramPacket) throws Exception {
System.out.println("channelId:" + ctx.channel().id().asLongText());
System.out.println("服务端接收到消息 \nsender:" + datagramPacket.sender().toString() + "内容\n" + datagramPacket.content().toString(StandardCharsets.UTF_8));
// 通知代理客户端
Channel visitorChannel = ctx.channel();
Channel proxyChannel = visitorChannel.attr(Constants.NEXT_CHANNEL).get();
if (null == proxyChannel) {
// 该端口还没有代理客户端
ctx.channel().close();
return;
}
String targetIp = proxyChannel.attr(Constants.TARGET_IP).get();
int targetPort = proxyChannel.attr(Constants.TARGET_PORT).get();
// 转发代理数据
byte[] bytes = new byte[datagramPacket.content().readableBytes()];
datagramPacket.content().readBytes(bytes);
String visitorId = ProxyUtil.getVisitorIdByChannel(visitorChannel);
proxyChannel.writeAndFlush(ProxyMessage.buildUdpTransferMessage(new ProxyMessage.UdpBaseInfo()
.setVisitorId(visitorId)
.setVisitorIp(datagramPacket.sender().getAddress().getHostAddress())
.setVisitorPort(datagramPacket.sender().getPort())
.setTargetIp(targetIp)
.setTargetPort(targetPort)
).setData(bytes));
// 增加流量计数
VisitorChannelAttachInfo visitorChannelAttachInfo = ProxyUtil.getAttachInfo(visitorChannel);
Solon.context().getBean(FlowReportService.class).addWriteByte(visitorChannelAttachInfo.getLicenseId(), bytes.length);
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("active channelId:" + ctx.channel().id().asLongText());
Channel visitorChannel = ctx.channel();
InetSocketAddress sa = (InetSocketAddress) visitorChannel.localAddress();
Channel cmdChannel = ProxyUtil.getCmdChannelByServerPort(sa.getPort());
@@ -43,16 +83,27 @@ public class UdpVisitorChannelHandler extends SimpleChannelInboundHandler<Datagr
String targetIp = targetInfo[0];
int targetPort = Integer.parseInt(targetInfo[1]);
// 转发代理数据
byte[] bytes = new byte[datagramPacket.content().readableBytes()];
datagramPacket.content().readBytes(bytes);
// 用户连接到代理服务器时，设置用户连接不可读，等待代理后端服务器连接成功后再改变为可读状态
visitorChannel.config().setOption(ChannelOption.AUTO_READ, false);
cmdChannel.writeAndFlush(ProxyMessage.buildUdpTransferMessage(
sa.getAddress().getHostAddress(),
sa.getPort(),
targetIp,
targetPort,
bytes
));
// UDP此处叫visitor似有不妥，与TCP不同
String visitorId = ProxyUtil.newVisitorId();
// 此处需要和tcp分开
ProxyUtil.addVisitorChannelToCmdChannel(NetworkProtocolEnum.UDP, cmdChannel, visitorId, visitorChannel, sa.getPort());
cmdChannel.writeAndFlush(ProxyMessage.buildUdpConnectMessage(new ProxyMessage.UdpBaseInfo()
.setVisitorId(visitorId)
.setServerPort(sa.getPort())
.setTargetIp(targetIp)
.setTargetPort(targetPort)
));
super.channelActive(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// 当出现异常就关闭连接
ctx.close();
log.error("[UDP Visitor Channel]VisitorChannel error", cause);
}
}
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/domain/VisitorChannelAttachInfo.java
+2
View File
@@ -23,6 +23,7 @@ package org.dromara.neutrinoproxy.server.proxy.domain;
import lombok.Data;
import lombok.experimental.Accessors;
import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
/**
*
@@ -32,6 +33,7 @@ import lombok.experimental.Accessors;
@Accessors(chain = true)
@Data
public class VisitorChannelAttachInfo {
private NetworkProtocolEnum protocol;
private String visitorId;
private String lanInfo;
private Integer serverPort;
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/enhance/HttpVisitorChannelHandler.java
+2 -1
View File
@@ -10,6 +10,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
import org.dromara.neutrinoproxy.server.proxy.domain.ProxyAttachment;
import org.dromara.neutrinoproxy.server.proxy.domain.VisitorChannelAttachInfo;
import org.dromara.neutrinoproxy.server.service.FlowReportService;
@@ -98,7 +99,7 @@ public class HttpVisitorChannelHandler extends SimpleChannelInboundHandler<ByteB
}
visitorId = ProxyUtil.newVisitorId();
ProxyUtil.addVisitorChannelToCmdChannel(cmdChannel, visitorId, ctx.channel(), serverPort);
ProxyUtil.addVisitorChannelToCmdChannel(NetworkProtocolEnum.HTTP, cmdChannel, visitorId, ctx.channel(), serverPort);
ProxyUtil.addProxyConnectAttachment(visitorId, proxyAttachment);
cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(visitorId).setData(lanInfo.getBytes()));
}
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/ProxyMessageConnectHandler.java
+6 -6
View File
@@ -34,14 +34,14 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler {
public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
String info = proxyMessage.getInfo();
if (StrUtil.isEmpty(info)) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "info不能为空!"));
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "info cannot be empty!"));
ctx.channel().close();
return;
}
String[] tokens = info.split("@");
if (tokens.length != 2) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "info格式有误!"));
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "info format error!"));
ctx.channel().close();
return;
}
@@ -50,18 +50,18 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler {
LicenseDO licenseDO = licenseService.findByKey(licenseKey);
if (null == licenseDO) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "license不存在!"));
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "the license notfound!"));
ctx.channel().close();
return;
}
if (EnableStatusEnum.DISABLE.getStatus().equals(licenseDO.getEnable())) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "当前license已被禁用!"));
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "the license invalid!"));
ctx.channel().close();
return;
}
UserDO userDO = userService.findById(licenseDO.getUserId());
if (null == userDO || EnableStatusEnum.DISABLE.getStatus().equals(userDO.getEnable())) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "当前license无效!"));
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "the license invalid!"));
ctx.channel().close();
return;
}
@@ -69,7 +69,7 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler {
Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseId(licenseDO.getId());
if (null == cmdChannel) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "服务端异常，指令通道不存在!"));
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "server error，cmd channel notfound!"));
ctx.channel().close();
return;
}
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/proxy/handler/UdpProxyMessageConnectHandler.java
+94
View File
@@ -0,0 +1,94 @@
package org.dromara.neutrinoproxy.server.proxy.handler;
import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import lombok.extern.slf4j.Slf4j;
import org.dromara.neutrinoproxy.core.*;
import org.dromara.neutrinoproxy.core.dispatcher.Match;
import org.dromara.neutrinoproxy.server.constant.EnableStatusEnum;
import org.dromara.neutrinoproxy.server.dal.PortMappingMapper;
import org.dromara.neutrinoproxy.server.dal.entity.LicenseDO;
import org.dromara.neutrinoproxy.server.dal.entity.PortMappingDO;
import org.dromara.neutrinoproxy.server.dal.entity.UserDO;
import org.dromara.neutrinoproxy.server.service.LicenseService;
import org.dromara.neutrinoproxy.server.service.UserService;
import org.dromara.neutrinoproxy.server.util.ProxyUtil;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
/**
* @author: aoshiguchen
* @date: 2023/9/19
*/
@Slf4j
@Match(type = Constants.ProxyDataTypeName.UDP_CONNECT)
@Component
public class UdpProxyMessageConnectHandler implements ProxyMessageHandler {
@Inject
private LicenseService licenseService;
@Inject
private UserService userService;
@Inject
private PortMappingMapper portMappingMapper;
@Override
public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) {
final Channel proxyChannel = ctx.channel();
final ProxyMessage.UdpBaseInfo udpBaseInfo = JSONObject.parseObject(proxyMessage.getInfo(), ProxyMessage.UdpBaseInfo.class);
final String licenseKey = new String(proxyMessage.getData());
log.info("[UDP connect]info:{} licenseKey:{}", proxyMessage.getInfo(), licenseKey);
LicenseDO licenseDO = licenseService.findByKey(licenseKey);
if (null == licenseDO) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "the license notfound!"));
ctx.channel().close();
return;
}
if (EnableStatusEnum.DISABLE.getStatus().equals(licenseDO.getEnable())) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "the license disabled!"));
ctx.channel().close();
return;
}
UserDO userDO = userService.findById(licenseDO.getUserId());
if (null == userDO || EnableStatusEnum.DISABLE.getStatus().equals(userDO.getEnable())) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "the license invalid!"));
ctx.channel().close();
return;
}
Channel cmdChannel = ProxyUtil.getCmdChannelByLicenseId(licenseDO.getId());
if (null == cmdChannel) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "server error，cmd channel notfound!"));
ctx.channel().close();
return;
}
Channel visitorChannel = ProxyUtil.getVisitorChannel(cmdChannel, udpBaseInfo.getVisitorId());
if (null == visitorChannel) {
return;
}
PortMappingDO portMappingDO = portMappingMapper.findByLicenseIdAndServerPort(licenseDO.getId(), udpBaseInfo.getServerPort());
if (null == portMappingDO || !EnableStatusEnum.ENABLE.getStatus().equals(portMappingDO.getEnable())) {
ctx.channel().writeAndFlush(ProxyMessage.buildErrMessage(ExceptionEnum.CONNECT_FAILED, "server error, port mapping notfound!"));
ctx.channel().close();
return;
}
ctx.channel().attr(Constants.VISITOR_ID).set(udpBaseInfo.getVisitorId());
ctx.channel().attr(Constants.LICENSE_ID).set(licenseDO.getId());
ctx.channel().attr(Constants.NEXT_CHANNEL).set(visitorChannel);
ctx.channel().attr(Constants.TARGET_IP).set(portMappingDO.getClientIp());
ctx.channel().attr(Constants.TARGET_PORT).set(portMappingDO.getClientPort());
visitorChannel.attr(Constants.NEXT_CHANNEL).set(ctx.channel());
// 代理客户端与后端服务器连接成功，修改用户连接为可读状态
visitorChannel.config().setOption(ChannelOption.AUTO_READ, true);
}
@Override
public String name() {
return ProxyDataTypeEnum.UDP_CONNECT.getDesc();
}
}
neutrino-proxy-server/src/main/java/org/dromara/neutrinoproxy/server/util/ProxyUtil.java
+12 -8
View File
@@ -5,6 +5,7 @@ import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.dromara.neutrinoproxy.core.ChannelAttribute;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.server.constant.NetworkProtocolEnum;
import org.dromara.neutrinoproxy.server.proxy.domain.CmdChannelAttachInfo;
import org.dromara.neutrinoproxy.server.proxy.domain.ProxyAttachment;
import org.dromara.neutrinoproxy.server.proxy.domain.ProxyMapping;
@@ -201,18 +202,21 @@ public class ProxyUtil {
* @param visitorId
* @param visitorChannel
*/
public static void addVisitorChannelToCmdChannel(Channel cmdChannel, String visitorId, Channel visitorChannel, Integer serverPort) {
public static void addVisitorChannelToCmdChannel(NetworkProtocolEnum protocol, Channel cmdChannel, String visitorId, Channel visitorChannel, Integer serverPort) {
InetSocketAddress sa = (InetSocketAddress) visitorChannel.localAddress();
String lanInfo = getClientLanInfoByServerPort(sa.getPort());
CmdChannelAttachInfo cmdChannelAttachInfo = getAttachInfo(cmdChannel);
setAttachInfo(visitorChannel, new VisitorChannelAttachInfo()
.setVisitorId(visitorId)
.setLanInfo(lanInfo)
.setServerPort(serverPort)
.setLicenseId(cmdChannelAttachInfo.getLicenseId())
.setIp(((InetSocketAddress)visitorChannel.remoteAddress()).getAddress().getHostAddress())
);
VisitorChannelAttachInfo attachInfo = new VisitorChannelAttachInfo()
.setProtocol(protocol)
.setVisitorId(visitorId)
.setLanInfo(lanInfo)
.setServerPort(serverPort)
.setLicenseId(cmdChannelAttachInfo.getLicenseId());
if (NetworkProtocolEnum.UDP != protocol) {
attachInfo.setIp(((InetSocketAddress)visitorChannel.remoteAddress()).getAddress().getHostAddress());
}
setAttachInfo(visitorChannel, attachInfo);
userChannelMapLock.writeLock().lock();
try {
cmdChannelAttachInfo.getVisitorChannelMap().put(visitorId, visitorChannel);