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

服务端、客户端认证逻辑调整

Browse Source
This commit is contained in:
aoshiguchen
2023-06-03 03:19:45 +08:00
parent 5a65eebfd3
commit 4c828f814c
7 changed files with 242 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/config/ProxyConfiguration.java
+7 -2
View File
@@ -31,8 +31,13 @@ public class ProxyConfiguration implements LifecycleBean {
Solon.context().wrapAndPut(Dispatcher.class, dispatcher);
}
@Bean("bootstrap")
public Bootstrap bootstrap() {
@Bean("cmdTunnelBootstrap")
public Bootstrap cmdTunnelBootstrap() {
return new Bootstrap();
}
@Bean("proxyTunnelBootstrap")
public Bootstrap proxyTunnelBootstrap() {
return new Bootstrap();
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/CmdChannelHandler.java
+77
View File
@@ -0,0 +1,77 @@
package org.dromara.neutrinoproxy.client.core;
import org.dromara.neutrinoproxy.client.util.ProxyUtil;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
import org.dromara.neutrinoproxy.core.dispatcher.Dispatcher;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.noear.solon.Solon;
/**
* 处理与服务端之间的数据传输
* @author: aoshiguchen
* @date: 2022/6/16
*/
@Slf4j
public class CmdChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ProxyMessage proxyMessage) throws Exception {
if (ProxyMessage.TYPE_HEARTBEAT != proxyMessage.getType()) {
log.info("Client CmdChannel recieved proxy message, type is {}", proxyMessage.getType());
}
Solon.context().getBean(Dispatcher.class).dispatch(ctx, proxyMessage);
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
Channel realServerChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
if (realServerChannel != null) {
realServerChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
log.info("Client CmdChannel 与服务端断开连接");
ProxyUtil.setCmdChannel(null);
ProxyUtil.clearRealServerChannels();
super.channelInactive(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
log.error("Client CmdChannel Error channelId:{}", ctx.channel().id().asLongText(), cause);
ctx.close();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if(evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent)evt;
switch (event.state()) {
case READER_IDLE:
// 读超时，断开连接
// log.info("读超时");
// ctx.channel().close();
break;
case WRITER_IDLE:
ctx.channel().writeAndFlush(ProxyMessage.buildHeartbeatMessage());
break;
case ALL_IDLE:
log.info("Client CmdChannel 读写超时");
ctx.close();
break;
}
}
}
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ClientChannelHandler.java → neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyChannelHandler.java
+13 -22
View File
@@ -1,15 +1,15 @@
package org.dromara.neutrinoproxy.client.core;
import org.dromara.neutrinoproxy.client.util.ProxyUtil;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
import org.dromara.neutrinoproxy.core.dispatcher.Dispatcher;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.dromara.neutrinoproxy.client.util.ProxyUtil;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
import org.dromara.neutrinoproxy.core.dispatcher.Dispatcher;
import org.noear.solon.Solon;
/**
@@ -18,13 +18,13 @@ import org.noear.solon.Solon;
* @date: 2022/6/16
*/
@Slf4j
public class ClientChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
public class ProxyChannelHandler extends SimpleChannelInboundHandler<ProxyMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ProxyMessage proxyMessage) throws Exception {
if (ProxyMessage.TYPE_HEARTBEAT != proxyMessage.getType()) {
log.info("recieved proxy message, type is {}", proxyMessage.getType());
log.info("Client ProxyChannel recieved proxy message, type is {}", proxyMessage.getType());
}
Solon.context().getBean(Dispatcher.class).dispatch(ctx, proxyMessage);
}
@@ -41,17 +41,10 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler<ProxyMessa
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
// 控制连接
if (ProxyUtil.getCmdChannel() == ctx.channel()) {
log.info("与服务端断开连接");
ProxyUtil.setCmdChannel(null);
ProxyUtil.clearRealServerChannels();
} else {
// 数据传输连接
Channel realServerChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
if (realServerChannel != null && realServerChannel.isActive()) {
realServerChannel.close();
}
// 数据传输连接
Channel realServerChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get();
if (realServerChannel != null && realServerChannel.isActive()) {
realServerChannel.close();
}
ProxyUtil.removeProxyChanel(ctx.channel());
@@ -60,10 +53,8 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler<ProxyMessa
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
if (ctx.channel().isActive()) {
ctx.channel().close();
}
log.error("Client ProxyChannel Error channelId:{}", ctx.channel().id().asLongText(), cause);
ctx.close();
}
@Override
@@ -81,7 +72,7 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler<ProxyMessa
break;
case ALL_IDLE:
log.info("读写超时");
ctx.channel().close();
ctx.close();
break;
}
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/ProxyClientService.java
+137 -113
View File
@@ -1,7 +1,6 @@
package org.dromara.neutrinoproxy.client.core;
import cn.hutool.core.util.StrUtil;
import io.netty.handler.logging.LoggingHandler;
import org.dromara.neutrinoproxy.client.config.ProxyConfig;
import org.dromara.neutrinoproxy.client.util.ProxyUtil;
import org.dromara.neutrinoproxy.core.ProxyMessage;
@@ -26,13 +25,14 @@ import javax.net.ssl.SSLEngine;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.security.KeyStore;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* 客户端服务
* 代理客户端服务
* @author: aoshiguchen
* @date: 2022/6/16
*/
@@ -41,15 +41,17 @@ import java.util.concurrent.TimeUnit;
public class ProxyClientService {
@Inject
private ProxyConfig proxyConfig;
@Inject("bootstrap")
private Bootstrap bootstrap;
@Inject("cmdTunnelBootstrap")
private Bootstrap cmdTunnelBootstrap;
@Inject("proxyTunnelBootstrap")
private Bootstrap proxyTunnelBootstrap;
@Inject("realServerBootstrap")
private Bootstrap realServerBootstrap;
private volatile Channel channel;
/**
* 重连间隔（秒）
*/
private static final long RECONNECT_INTERVAL_SECONDS = 6;
private static final long RECONNECT_INTERVAL_SECONDS = 5;
/**
* 重连次数
*/
@@ -58,6 +60,7 @@ public class ProxyClientService {
// * 启用重连服务
// */
// private volatile boolean reconnectServiceEnable = false;
private NioEventLoopGroup workerGroup;
/**
* 重连服务执行器
*/
@@ -66,8 +69,8 @@ public class ProxyClientService {
@Init
public void init() {
this.reconnectExecutor.scheduleWithFixedDelay(this::reconnect, 10, RECONNECT_INTERVAL_SECONDS, TimeUnit.SECONDS);
this.workerGroup = new NioEventLoopGroup(proxyConfig.getClient().getThreadCount());
NioEventLoopGroup workerGroup = new NioEventLoopGroup(proxyConfig.getClient().getThreadCount());
realServerBootstrap.group(workerGroup);
realServerBootstrap.channel(NioSocketChannel.class);
realServerBootstrap.handler(new ChannelInitializer<SocketChannel>() {
@@ -78,17 +81,10 @@ public class ProxyClientService {
}
});
bootstrap.group(workerGroup);
bootstrap.channel(NioSocketChannel.class);
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
/**
* TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。（一个连接会设置MSS参数，因此，TCP/IP希望每次都能够以MSS尺寸的数据块来发送数据）。
* Nagle算法就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块。
*/
bootstrap.option(ChannelOption.TCP_NODELAY, true);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
proxyTunnelBootstrap.group(workerGroup);
proxyTunnelBootstrap.channel(NioSocketChannel.class);
proxyTunnelBootstrap.remoteAddress(InetSocketAddress.createUnresolved(proxyConfig.getClient().getServerIp(), proxyConfig.getClient().getServerPort()));
proxyTunnelBootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
@@ -101,115 +97,143 @@ public class ProxyClientService {
proxyConfig.getProtocol().getLengthAdjustment(), proxyConfig.getProtocol().getInitialBytesToStrip()));
ch.pipeline().addLast(new ProxyMessageEncoder());
ch.pipeline().addLast(new IdleStateHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds()));
ch.pipeline().addLast(new ClientChannelHandler());
ch.pipeline().addLast(new ProxyChannelHandler());
}
});
try {
this.start();
} catch (Exception e) {
// 启动连不上也做一下重连，因此先catch异常
log.error("启动异常", e);
}
}
cmdTunnelBootstrap.group(workerGroup);
cmdTunnelBootstrap.channel(NioSocketChannel.class);
// cmdTunnelBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
// cmdTunnelBootstrap.option(ChannelOption.SO_KEEPALIVE, true);
// /**
// * TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。（一个连接会设置MSS参数，因此，TCP/IP希望每次都能够以MSS尺寸的数据块来发送数据）。
// * Nagle算法就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块。
// */
// cmdTunnelBootstrap.option(ChannelOption.TCP_NODELAY, true);
cmdTunnelBootstrap.remoteAddress(InetSocketAddress.createUnresolved(proxyConfig.getClient().getServerIp(), proxyConfig.getClient().getServerPort()));
public void start() {
if (StrUtil.isEmpty(proxyConfig.getClient().getServerIp())) {
log.error("not found server-ip config.");
Solon.stop();
return;
}
if (null == proxyConfig.getClient().getServerPort()) {
log.error("not found server-port config.");
Solon.stop();
return;
}
if (null != proxyConfig.getClient().getSslEnable() && proxyConfig.getClient().getSslEnable()
&& StrUtil.isEmpty(proxyConfig.getClient().getJksPath())) {
log.error("not found jks-path config.");
Solon.stop();
return;
}
if (StrUtil.isEmpty(proxyConfig.getClient().getLicenseKey())) {
log.error("not found license-key config.");
Solon.stop();
return;
}
if (null == channel || !channel.isActive()) {
try {
connectProxyServer();
} catch (Exception e) {
log.error("client start error", e);
cmdTunnelBootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
if (proxyConfig.getClient().getSslEnable()) {
ch.pipeline().addLast(createSslHandler());
}
} else {
channel.writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey(), ProxyUtil.getClientId()));
// ch.pipeline().addFirst(new LoggingHandler(ProxyClientService.class));
ch.pipeline().addLast(new ProxyMessageDecoder(proxyConfig.getProtocol().getMaxFrameLength(),
proxyConfig.getProtocol().getLengthFieldOffset(), proxyConfig.getProtocol().getLengthFieldLength(),
proxyConfig.getProtocol().getLengthAdjustment(), proxyConfig.getProtocol().getInitialBytesToStrip()));
ch.pipeline().addLast(new ProxyMessageEncoder());
ch.pipeline().addLast(new IdleStateHandler(proxyConfig.getProtocol().getReadIdleTime(), proxyConfig.getProtocol().getWriteIdleTime(), proxyConfig.getProtocol().getAllIdleTimeSeconds()));
ch.pipeline().addLast(new CmdChannelHandler());
}
});
try {
this.start();
} catch (Exception e) {
// 启动连不上也做一下重连，因此先catch异常
log.error("[客户端指令隧道] 启动异常", e);
}
}
/**
* 连接代理服务器
*/
private void connectProxyServer() throws InterruptedException {
bootstrap.connect(proxyConfig.getClient().getServerIp(), proxyConfig.getClient().getServerPort())
.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
channel = future.channel();
// 连接成功，向服务器发送客户端认证信息（licenseKey）
ProxyUtil.setCmdChannel(future.channel());
future.channel().writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey(), ProxyUtil.getClientId()));
log.info("连接代理服务成功. channelId:{}", future.channel().id().asLongText());
// reconnectServiceEnable = true;
reconnectCount = 0;
} else {
log.info("连接代理服务失败!");
}
}
}).sync();
public void start() {
if (StrUtil.isEmpty(proxyConfig.getClient().getServerIp())) {
log.error("not found server-ip config.");
Solon.stop();
return;
}
private ChannelHandler createSslHandler() {
try {
InputStream jksInputStream = FileUtil.getInputStream(proxyConfig.getClient().getJksPath());
SSLContext clientContext = SSLContext.getInstance("TLS");
final KeyStore ks = KeyStore.getInstance("JKS");
ks.load(jksInputStream, proxyConfig.getClient().getKeyStorePassword().toCharArray());
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(ks);
TrustManager[] trustManagers = tmf.getTrustManagers();
clientContext.init(null, trustManagers, null);
SSLEngine sslEngine = clientContext.createSSLEngine();
sslEngine.setUseClientMode(true);
return new SslHandler(sslEngine);
} catch (Exception e) {
log.error("创建SSL处理器失败", e);
e.printStackTrace();
}
return null;
if (null == proxyConfig.getClient().getServerPort()) {
log.error("not found server-port config.");
Solon.stop();
return;
}
protected synchronized void reconnect() {
// if (!reconnectServiceEnable) {
// return;
// }
if (null != channel) {
if (channel.isActive()) {
return;
}
channel.close();
}
log.info("客户端重连 seq:{}", ++reconnectCount);
if (null != proxyConfig.getClient().getSslEnable() && proxyConfig.getClient().getSslEnable()
&& StrUtil.isEmpty(proxyConfig.getClient().getJksPath())) {
log.error("not found jks-path config.");
Solon.stop();
return;
}
if (StrUtil.isEmpty(proxyConfig.getClient().getLicenseKey())) {
log.error("not found license-key config.");
Solon.stop();
return;
}
if (null == channel || !channel.isActive()) {
try {
connectProxyServer();
} catch (Exception e) {
log.error("重连异常", e);
log.error("client start error", e);
}
} else {
channel.writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey(), ProxyUtil.getClientId()));
}
}
/**
* 连接代理服务器
*/
private void connectProxyServer() throws InterruptedException {
cmdTunnelBootstrap.connect()
.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
channel = future.channel();
// 连接成功，向服务器发送客户端认证信息（licenseKey）
ProxyUtil.setCmdChannel(future.channel());
future.channel().writeAndFlush(ProxyMessage.buildAuthMessage(proxyConfig.getClient().getLicenseKey(), ProxyUtil.getClientId()));
log.info("[客户端指令隧道] 连接代理服务成功. channelId:{}", future.channel().id().asLongText());
// reconnectServiceEnable = true;
reconnectCount = 0;
} else {
log.info("[客户端指令隧道] 连接代理服务失败!");
}
}
}).sync();
}
private ChannelHandler createSslHandler() {
try {
InputStream jksInputStream = FileUtil.getInputStream(proxyConfig.getClient().getJksPath());
SSLContext clientContext = SSLContext.getInstance("TLS");
final KeyStore ks = KeyStore.getInstance("JKS");
ks.load(jksInputStream, proxyConfig.getClient().getKeyStorePassword().toCharArray());
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(ks);
TrustManager[] trustManagers = tmf.getTrustManagers();
clientContext.init(null, trustManagers, null);
SSLEngine sslEngine = clientContext.createSSLEngine();
sslEngine.setUseClientMode(true);
return new SslHandler(sslEngine);
} catch (Exception e) {
log.error("创建SSL处理器失败", e);
e.printStackTrace();
}
return null;
}
protected synchronized void reconnect() {
// if (!reconnectServiceEnable) {
// return;
// }
if (null != channel) {
if (channel.isActive()) {
return;
}
channel.close();
}
log.info("[客户端指令隧道] 客户端重连 seq:{}", ++reconnectCount);
try {
connectProxyServer();
} catch (Exception e) {
log.error("[客户端指令隧道] 重连异常", e);
}
}
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/core/RealServerChannelHandler.java
+3 -1
View File
@@ -22,6 +22,7 @@
package org.dromara.neutrinoproxy.client.core;
import lombok.extern.slf4j.Slf4j;
import org.dromara.neutrinoproxy.client.util.ProxyUtil;
import org.dromara.neutrinoproxy.core.Constants;
import org.dromara.neutrinoproxy.core.ProxyMessage;
@@ -36,6 +37,7 @@ import io.netty.channel.SimpleChannelInboundHandler;
* @author: aoshiguchen
* @date: 2022/6/16
*/
@Slf4j
public class RealServerChannelHandler extends SimpleChannelInboundHandler<ByteBuf> {
@@ -85,6 +87,6 @@ public class RealServerChannelHandler extends SimpleChannelInboundHandler<ByteBu
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
log.error("Client ProxyChannel Error", cause);
}
}
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/handler/ProxyMessageConnectHandler.java
+3 -3
View File
@@ -21,8 +21,8 @@ import org.noear.solon.annotation.Inject;
@Match(type = Constants.ProxyDataTypeName.CONNECT)
@Component
public class ProxyMessageConnectHandler implements ProxyMessageHandler {
@Inject("bootstrap")
private Bootstrap bootstrap;
@Inject("proxyTunnelBootstrap")
private Bootstrap proxyTunnelBootstrap;
@Inject("realServerBootstrap")
private Bootstrap realServerBootstrap;
@Inject
@@ -48,7 +48,7 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler {
realServerChannel.config().setOption(ChannelOption.AUTO_READ, false);
// 获取连接
ProxyUtil.borrowProxyChanel(bootstrap, new ProxyChannelBorrowListener() {
ProxyUtil.borrowProxyChanel(proxyTunnelBootstrap, new ProxyChannelBorrowListener() {
@Override
public void success(Channel channel) {
neutrino-proxy-client/src/main/java/org/dromara/neutrinoproxy/client/util/ProxyUtil.java
+2 -4
View File
@@ -61,16 +61,14 @@ public class ProxyUtil {
private static String clientId;
private static final String CLIENT_ID_FILE = ".NEUTRINO_PROXY_CLIENT_ID";
public static void borrowProxyChanel(Bootstrap bootstrap, final ProxyChannelBorrowListener borrowListener) {
public static void borrowProxyChanel(Bootstrap proxyTunnelBootstrap, final ProxyChannelBorrowListener borrowListener) {
Channel channel = proxyChannelPool.poll();
if (null != channel) {
borrowListener.success(channel);
return;
}
String serverIp = Solon.cfg().get("neutrino.proxy.client.server-ip");
Integer serverPort = Solon.cfg().getInt("neutrino.proxy.client.server-port", 9000);
bootstrap.connect(serverIp, serverPort).addListener((ChannelFutureListener) future -> {
proxyTunnelBootstrap.connect().addListener((ChannelFutureListener) future -> {
if (future.isSuccess()) {
borrowListener.success(future.channel());
} else {