flik/GoTunnel
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add plugins
All checks were successful
Build Multi-Platform Binaries / build (push) Successful in 11m9s
Details

Browse Source
This commit is contained in:
Flik
2025-12-26 11:24:23 +08:00
parent d56fdafc1e
commit 4623a7f031
27 changed files with 2090 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
internal/server/tunnel/server.go
View File

@@ -9,6 +9,7 @@ import (
"time"
"github.com/gotunnel/internal/server/db"
"github.com/gotunnel/pkg/plugin"
"github.com/gotunnel/pkg/protocol"
"github.com/gotunnel/pkg/proxy"
"github.com/gotunnel/pkg/relay"
@@ -16,28 +17,37 @@ import (
"github.com/hashicorp/yamux"
)
// 服务端常量
const (
authTimeout = 10 * time.Second
heartbeatTimeout = 10 * time.Second
udpBufferSize = 65535
)
// Server 隧道服务端
type Server struct {
clientStore db.ClientStore
bindAddr string
bindPort int
token string
heartbeat int
hbTimeout int
portManager *utils.PortManager
clients map[string]*ClientSession
mu sync.RWMutex
tlsConfig *tls.Config
clientStore db.ClientStore
bindAddr string
bindPort int
token string
heartbeat int
hbTimeout int
portManager *utils.PortManager
clients map[string]*ClientSession
mu sync.RWMutex
tlsConfig *tls.Config
pluginRegistry *plugin.Registry
}
// ClientSession 客户端会话
type ClientSession struct {
ID string
Session *yamux.Session
Rules []protocol.ProxyRule
Listeners map[int]net.Listener
LastPing time.Time
mu sync.Mutex
ID string
Session *yamux.Session
Rules []protocol.ProxyRule
Listeners map[int]net.Listener
UDPConns map[int]*net.UDPConn // UDP 连接
LastPing time.Time
mu sync.Mutex
}
// NewServer 创建服务端
@@ -59,6 +69,11 @@ func (s *Server) SetTLSConfig(config *tls.Config) {
s.tlsConfig = config
}
// SetPluginRegistry 设置插件注册表
func (s *Server) SetPluginRegistry(registry *plugin.Registry) {
s.pluginRegistry = registry
}
// Run 启动服务端
func (s *Server) Run() error {
addr := fmt.Sprintf("%s:%d", s.bindAddr, s.bindPort)
@@ -95,7 +110,7 @@ func (s *Server) Run() error {
func (s *Server) handleConnection(conn net.Conn) {
defer conn.Close()
conn.SetReadDeadline(time.Now().Add(10 * time.Second))
conn.SetReadDeadline(time.Now().Add(authTimeout))
msg, err := protocol.ReadMessage(conn)
if err != nil {
@@ -148,6 +163,7 @@ func (s *Server) setupClientSession(conn net.Conn, clientID string, rules []prot
Session: session,
Rules: rules,
Listeners: make(map[int]net.Listener),
UDPConns: make(map[int]*net.UDPConn),
LastPing: time.Now(),
}
@@ -169,7 +185,10 @@ func (s *Server) setupClientSession(conn net.Conn, clientID string, rules []prot
// sendAuthResponse 发送认证响应
func (s *Server) sendAuthResponse(conn net.Conn, success bool, message string) error {
resp := protocol.AuthResponse{Success: success, Message: message}
msg, _ := protocol.NewMessage(protocol.MsgTypeAuthResp, resp)
msg, err := protocol.NewMessage(protocol.MsgTypeAuthResp, resp)
if err != nil {
return err
}
return protocol.WriteMessage(conn, msg)
}
@@ -182,7 +201,10 @@ func (s *Server) sendProxyConfig(session *yamux.Session, rules []protocol.ProxyR
defer stream.Close()
cfg := protocol.ProxyConfig{Rules: rules}
msg, _ := protocol.NewMessage(protocol.MsgTypeProxyConfig, cfg)
msg, err := protocol.NewMessage(protocol.MsgTypeProxyConfig, cfg)
if err != nil {
return err
}
return protocol.WriteMessage(stream, msg)
}
@@ -203,6 +225,10 @@ func (s *Server) unregisterClient(cs *ClientSession) {
ln.Close()
s.portManager.Release(port)
}
for port, conn := range cs.UDPConns {
conn.Close()
s.portManager.Release(port)
}
cs.mu.Unlock()
delete(s.clients, cs.ID)
@@ -211,6 +237,18 @@ func (s *Server) unregisterClient(cs *ClientSession) {
// startProxyListeners 启动代理监听
func (s *Server) startProxyListeners(cs *ClientSession) {
for _, rule := range cs.Rules {
ruleType := rule.Type
if ruleType == "" {
ruleType = "tcp"
}
// UDP 单独处理
if ruleType == "udp" {
s.startUDPListener(cs, rule)
continue
}
// TCP 类型
if err := s.portManager.Reserve(rule.RemotePort, cs.ID); err != nil {
log.Printf("[Server] Port %d error: %v", rule.RemotePort, err)
continue
@@ -227,15 +265,12 @@ func (s *Server) startProxyListeners(cs *ClientSession) {
cs.Listeners[rule.RemotePort] = ln
cs.mu.Unlock()
ruleType := rule.Type
if ruleType == "" {
ruleType = "tcp"
}
switch ruleType {
case "socks5", "http":
log.Printf("[Server] %s proxy %s on :%d",
ruleType, rule.Name, rule.RemotePort)
case "socks5":
log.Printf("[Server] SOCKS5 proxy %s on :%d", rule.Name, rule.RemotePort)
go s.acceptProxyServerConns(cs, ln, rule)
case "http", "https":
log.Printf("[Server] HTTP proxy %s on :%d", rule.Name, rule.RemotePort)
go s.acceptProxyServerConns(cs, ln, rule)
default:
log.Printf("[Server] TCP proxy %s: :%d -> %s:%d",
@@ -259,8 +294,23 @@ func (s *Server) acceptProxyConns(cs *ClientSession, ln net.Listener, rule proto
// acceptProxyServerConns 接受 SOCKS5/HTTP 代理连接
func (s *Server) acceptProxyServerConns(cs *ClientSession, ln net.Listener, rule protocol.ProxyRule) {
dialer := proxy.NewTunnelDialer(cs.Session)
proxyServer := proxy.NewServer(rule.Type, dialer)
// 优先使用插件系统
if s.pluginRegistry != nil {
if handler, err := s.pluginRegistry.Get(rule.Type); err == nil {
handler.Init(rule.PluginConfig)
for {
conn, err := ln.Accept()
if err != nil {
return
}
go handler.HandleConn(conn, dialer)
}
}
}
// 回退到内置 proxy 实现
proxyServer := proxy.NewServer(rule.Type, dialer)
for {
conn, err := ln.Accept()
if err != nil {
@@ -309,13 +359,12 @@ func (s *Server) heartbeatLoop(cs *ClientSession) {
}
cs.mu.Unlock()
stream, err := cs.Session.Open()
if err != nil {
return
// 发送心跳并等待响应
if s.sendHeartbeat(cs) {
cs.mu.Lock()
cs.LastPing = time.Now()
cs.mu.Unlock()
}
msg := &protocol.Message{Type: protocol.MsgTypeHeartbeat}
protocol.WriteMessage(stream, msg)
stream.Close()
case <-cs.Session.CloseChan():
return
@@ -323,6 +372,31 @@ func (s *Server) heartbeatLoop(cs *ClientSession) {
}
}
// sendHeartbeat 发送心跳并等待响应
func (s *Server) sendHeartbeat(cs *ClientSession) bool {
stream, err := cs.Session.Open()
if err != nil {
return false
}
defer stream.Close()
// 设置读写超时
stream.SetDeadline(time.Now().Add(heartbeatTimeout))
msg := &protocol.Message{Type: protocol.MsgTypeHeartbeat}
if err := protocol.WriteMessage(stream, msg); err != nil {
return false
}
// 等待心跳响应
resp, err := protocol.ReadMessage(stream)
if err != nil {
return false
}
return resp.Type == protocol.MsgTypeHeartbeatAck
}
// GetClientStatus 获取客户端状态
func (s *Server) GetClientStatus(clientID string) (online bool, lastPing string) {
s.mu.RLock()
@@ -341,17 +415,22 @@ func (s *Server) GetAllClientStatus() map[string]struct {
Online bool
LastPing string
} {
// 先复制客户端引用，避免嵌套锁
s.mu.RLock()
defer s.mu.RUnlock()
clients := make([]*ClientSession, 0, len(s.clients))
for _, cs := range s.clients {
clients = append(clients, cs)
}
s.mu.RUnlock()
result := make(map[string]struct {
Online bool
LastPing string
})
for id, cs := range s.clients {
for _, cs := range clients {
cs.mu.Lock()
result[id] = struct {
result[cs.ID] = struct {
Online bool
LastPing string
}{
@@ -364,8 +443,9 @@ func (s *Server) GetAllClientStatus() map[string]struct {
}
// ReloadConfig 重新加载配置
// 注意: 当前版本不支持热重载，需要重启服务
func (s *Server) ReloadConfig() error {
return nil
return fmt.Errorf("hot reload not supported, please restart the server")
}
// GetBindAddr 获取绑定地址
@@ -377,3 +457,87 @@ func (s *Server) GetBindAddr() string {
func (s *Server) GetBindPort() int {
return s.bindPort
}
// startUDPListener 启动 UDP 监听
func (s *Server) startUDPListener(cs *ClientSession, rule protocol.ProxyRule) {
if err := s.portManager.Reserve(rule.RemotePort, cs.ID); err != nil {
log.Printf("[Server] UDP port %d error: %v", rule.RemotePort, err)
return
}
addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf(":%d", rule.RemotePort))
if err != nil {
log.Printf("[Server] UDP resolve error: %v", err)
s.portManager.Release(rule.RemotePort)
return
}
conn, err := net.ListenUDP("udp", addr)
if err != nil {
log.Printf("[Server] UDP listen %d error: %v", rule.RemotePort, err)
s.portManager.Release(rule.RemotePort)
return
}
cs.mu.Lock()
cs.UDPConns[rule.RemotePort] = conn
cs.mu.Unlock()
log.Printf("[Server] UDP proxy %s: :%d -> %s:%d",
rule.Name, rule.RemotePort, rule.LocalIP, rule.LocalPort)
go s.handleUDPConn(cs, conn, rule)
}
// handleUDPConn 处理 UDP 连接
func (s *Server) handleUDPConn(cs *ClientSession, conn *net.UDPConn, rule protocol.ProxyRule) {
buf := make([]byte, udpBufferSize)
for {
n, clientAddr, err := conn.ReadFromUDP(buf)
if err != nil {
return
}
// 封装 UDP 数据包发送到客户端
packet := protocol.UDPPacket{
RemotePort: rule.RemotePort,
ClientAddr: clientAddr.String(),
Data: buf[:n],
}
go s.sendUDPPacket(cs, conn, clientAddr, packet)
}
}
// sendUDPPacket 发送 UDP 数据包到客户端
func (s *Server) sendUDPPacket(cs *ClientSession, conn *net.UDPConn, clientAddr *net.UDPAddr, packet protocol.UDPPacket) {
stream, err := cs.Session.Open()
if err != nil {
return
}
defer stream.Close()
msg, err := protocol.NewMessage(protocol.MsgTypeUDPData, packet)
if err != nil {
return
}
if err := protocol.WriteMessage(stream, msg); err != nil {
return
}
// 等待客户端响应
respMsg, err := protocol.ReadMessage(stream)
if err != nil {
return
}
if respMsg.Type == protocol.MsgTypeUDPData {
var respPacket protocol.UDPPacket
if err := respMsg.ParsePayload(&respPacket); err != nil {
return
}
conn.WriteToUDP(respPacket.Data, clientAddr)
}
}