与子节点连接失败，正在自动返回主节点...
`) + c.Abort() + return + } + c.JSON(502, gin.H{"code": 502, "msg": "与子节点逆向隧道通信异常: " + err.Error()}) + c.Abort() + return + } + c.Abort() + return + } + + // 走普通 HTTP 直连代理 + targetURL := strings.TrimRight(node.URL, "/") + targetPath + if c.Request.URL.RawQuery != "" { + targetURL += "?" + c.Request.URL.RawQuery + } + + req, err := http.NewRequest(c.Request.Method, targetURL, c.Request.Body) + if err != nil { + c.JSON(500, gin.H{"code": 500, "msg": "Failed to create proxy request"}) + c.Abort() + return + } + + // 复制请求头 + req.Header = c.Request.Header.Clone() + + // 覆盖认证授权 Header 确保子节点鉴权通过 + if node.Token != "" { + req.Header.Set("Authorization", "Bearer "+node.Token) + } + // 移除 Cookie 头部防干扰 + req.Header.Del("Cookie") + req.Header.Set("X-Tunnel-Proxy", "true") + + resp, err := travelProxyClient.Do(req) + if err != nil { + // 如果是客户端自己主动取消了请求（例如连续刷新、关闭网页等），直接退出，不应视为子节点离线而执行退回主节点的操作 + if c.Request.Context().Err() == context.Canceled || strings.Contains(err.Error(), "context canceled") { + c.Abort() + return + } + if strings.Contains(c.GetHeader("Accept"), "text/html") { + c.SetCookie(constant.CookieActiveInterconnectNodeID, "", -1, "/", "", false, false) + c.Header("Content-Type", "text/html; charset=utf-8") + c.String(200, `与子节点连接失败，正在自动返回主节点...
`) + c.Abort() + return + } + c.JSON(502, gin.H{"code": 502, "msg": "无法连接至目标子节点: " + err.Error()}) + c.Abort() + return + } + defer resp.Body.Close() + + // 复制响应头 + for k, v := range resp.Header { + for _, vv := range v { + c.Writer.Header().Add(k, vv) + } + } + c.Status(resp.StatusCode) + io.Copy(c.Writer, resp.Body) + c.Abort() + } +} diff --git a/internal/models/interconnect.go b/internal/models/interconnect.go new file mode 100644 index 0000000..2a5a823 --- /dev/null +++ b/internal/models/interconnect.go @@ -0,0 +1,32 @@ +package models + +import ( + "github.com/engigu/baihu-panel/internal/constant" +) + +// NodeMetrics 表示互联节点的性能指标，使用 JSON 存储 +type NodeMetrics struct { + CPUPercent float64 `json:"cpu_percent"` + MemPercent float64 `json:"mem_percent"` + DiskPercent float64 `json:"disk_percent"` + TxBytes uint64 `json:"tx_bytes,omitempty"` + RxBytes uint64 `json:"rx_bytes,omitempty"` +} + +// InterconnectNode represents a connected remote panel +type InterconnectNode struct { + ID string `json:"id" gorm:"primaryKey;size:20"` + Name string `json:"name" gorm:"size:255;not null"` + URL string `json:"url" gorm:"size:255;not null"` + Token string `json:"token" gorm:"size:255"` + Remark string `json:"remark" gorm:"size:500"` + CreatedAt LocalTime `json:"created_at"` + UpdatedAt LocalTime `json:"updated_at"` + Status string `json:"status" gorm:"size:50"` // online / offline + Metrics NodeMetrics `json:"metrics" gorm:"serializer:json"` + LastHeartbeatAt *LocalTime `json:"last_heartbeat_at"` +} + +func (InterconnectNode) TableName() string { + return constant.TablePrefix + "interconnect_nodes" +} diff --git a/internal/router/api_routes.go b/internal/router/api_routes.go index 784107b..acbb79e 100644 --- a/internal/router/api_routes.go +++ b/internal/router/api_routes.go @@ -1,9 +1,9 @@ package router import ( - // "github.com/engigu/baihu-panel/internal/controllers" - "github.com/engigu/baihu-panel/internal/middleware" + "runtime" + "github.com/engigu/baihu-panel/internal/middleware" "github.com/gin-gonic/gin" ) @@ -13,6 +13,12 @@ func initPublicAPIRoutes(api *gin.RouterGroup, c *Controllers) { ctx.JSON(200, gin.H{"message": "pong"}) }) + api.GET("/debug/goroutines", func(ctx *gin.Context) { + buf := make([]byte, 1024*1024) + n := runtime.Stack(buf, true) + ctx.Data(200, "text/plain; charset=utf-8", buf[:n]) + }) + // Authentication routes (无需认证) auth := api.Group("/auth") { @@ -24,6 +30,11 @@ func initPublicAPIRoutes(api *gin.RouterGroup, c *Controllers) { // 公开的站点设置（无需认证） api.GET("/settings/public", c.Settings.GetPublicSiteSettings) + // 隧道模式 (被控端反向连入，使用独立 Token 做 WebSocket 鉴权) + api.GET("/interconnect/tunnel", c.Interconnect.HandleTunnel) + // 子节点主动上报监控数据 (无中间件鉴权，内部鉴权) + api.POST("/interconnect/report", c.Interconnect.ReportMonitorData) + // 内部使用的 API（仅限本地调用，无需 Bearer 认证） internalAPI := api.Group("/internal") internalAPI.Use(middleware.LocalhostOnly()) @@ -66,6 +77,7 @@ func initAuthorizedAPIRoutes(api *gin.RouterGroup, c *Controllers) { registerSystemWSRoutes(adminOnly, c) registerWebUIRoutes(adminOnly, c) registerMonitorRoutes(adminOnly, c) + registerInterconnectRoutes(adminOnly, c) } } @@ -83,6 +95,7 @@ func registerTaskRoutes(g *gin.RouterGroup, c *Controllers) { tasks.POST("", c.Task.CreateTask) tasks.GET("", c.Task.GetTasks) tasks.GET("/:id", c.Task.GetTask) + tasks.POST("/bulk_save", c.Task.BulkSaveTask) tasks.PUT("/:id", c.Task.UpdateTask) tasks.DELETE("/:id", c.Task.DeleteTask) tasks.POST("/batch-delete", c.Task.BatchDeleteTasks) @@ -273,7 +286,7 @@ func registerMonitorRoutes(g *gin.RouterGroup, c *Controllers) { monitor := g.Group("/monitor") { monitor.GET("", c.Monitor.GetSystemMonitor) - monitor.GET("/ws", c.Monitor.MonitorWS) + monitor.GET("/sse", c.Monitor.MonitorSSE) } } @@ -298,3 +311,22 @@ func registerWebUIRoutes(g *gin.RouterGroup, c *Controllers) { webuiGroup.DELETE("/:name", c.WebUI.DeleteWebUI) } } + +func registerInterconnectRoutes(g *gin.RouterGroup, c *Controllers) { + interconnect := g.Group("/interconnect") + { + interconnect.GET("/nodes", c.Interconnect.GetNodes) + interconnect.POST("/nodes", c.Interconnect.CreateNode) + interconnect.PUT("/nodes/:id", c.Interconnect.UpdateNode) + interconnect.DELETE("/nodes/:id", c.Interconnect.DeleteNode) + interconnect.GET("/nodes/:id/status", c.Interconnect.GetNodeStatus) + interconnect.POST("/sync/script", c.Interconnect.SyncScript) + interconnect.POST("/sync/env", c.Interconnect.SyncEnv) + interconnect.POST("/sync/task", c.Interconnect.SyncTask) + + interconnect.GET("/child/status", c.Interconnect.GetChildStatus) + + // 代理模式 (面板穿越) + interconnect.Any("/proxy/:node_id/*path", c.Interconnect.ProxyRequest) + } +} diff --git a/internal/router/events.go b/internal/router/events.go index 1af6f9c..ac68406 100644 --- a/internal/router/events.go +++ b/internal/router/events.go @@ -2,13 +2,13 @@ package router import ( // "fmt" - "time" // "github.com/engigu/baihu-panel/internal/constant" "github.com/engigu/baihu-panel/internal/eventbus" // "github.com/engigu/baihu-panel/internal/logger" // "github.com/engigu/baihu-panel/internal/models" "github.com/engigu/baihu-panel/internal/services" + "github.com/engigu/baihu-panel/internal/executor" ) func setupEventHandlers(subscribers ...eventbus.Subscriber) { @@ -21,12 +21,8 @@ func setupEventHandlers(subscribers ...eventbus.Subscriber) { } func startAppLogCleanup(appLogSvc *services.AppLogService) { - // 初始化时执行一次清理 - appLogSvc.CleanUp() - - // 定期清理（每隔1小时执行一次巡检） - ticker := time.NewTicker(1 * time.Hour) - for range ticker.C { + // 注册到内部系统定时器（并立即执行第一次） + executor.GetSysCron().AddJobWithRun("@every 1h", func() { appLogSvc.CleanUp() - } + }) } diff --git a/internal/router/register.go b/internal/router/register.go index 8065f3d..f2a1c9f 100644 --- a/internal/router/register.go +++ b/internal/router/register.go @@ -29,6 +29,7 @@ func RegisterControllers() *Controllers { // 创建任务执行服务（需要依赖注入） notifyService := services.NewNotificationService() appLogService := services.NewAppLogService() + interconnectService := services.NewInterconnectService() // 清理 task 运行状态的任务可以直接由 executorService 承担或在此处通过 Database 直接清理 // 简单期间，我们使用一个新方法 tasks.CleanupRunningTasks() 或者让 executorService 启动时清理 @@ -42,7 +43,7 @@ func RegisterControllers() *Controllers { // 初始化所有关注系统总线的服务 setupEventHandlers(appLogService, notifyService, loginLogService, systemWSManager) - go startAppLogCleanup(appLogService) + startAppLogCleanup(appLogService) // 初始化并返回控制器 return &Controllers{ @@ -65,6 +66,7 @@ func RegisterControllers() *Controllers { SystemWS: controllers.NewSystemWSController(), WebUI: controllers.NewWebUIController(services.NewWebUIService(settingsService)), Monitor: controllers.NewMonitorController(executorService), + Interconnect: controllers.NewInterconnectController(interconnectService), } } diff --git a/internal/router/router.go b/internal/router/router.go index 0156d44..eaf562d 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -32,6 +32,7 @@ type Controllers struct { SystemWS *controllers.SystemWSController WebUI *controllers.WebUIController Monitor *controllers.MonitorController + Interconnect *controllers.InterconnectController } func Setup(c *Controllers) *gin.Engine { @@ -40,6 +41,7 @@ func Setup(c *Controllers) *gin.Engine { } router := gin.New() router.Use(middleware.GinLogger(), middleware.GinRecovery()) + router.Use(middleware.TravelProxyMiddleware()) // 获取 URL 前缀 cfg := services.GetConfig() diff --git a/internal/services/agent_ws_service.go b/internal/services/agent_ws_service.go index eb3e744..b7ac129 100644 --- a/internal/services/agent_ws_service.go +++ b/internal/services/agent_ws_service.go @@ -7,6 +7,7 @@ import ( "github.com/engigu/baihu-panel/internal/constant" "github.com/engigu/baihu-panel/internal/database" + "github.com/engigu/baihu-panel/internal/executor" "github.com/engigu/baihu-panel/internal/logger" "github.com/engigu/baihu-panel/internal/models" @@ -78,7 +79,11 @@ func GetAgentWSManager() *AgentWSManager { ipFailCount: make(map[string]int), remoteWaiters: make(map[string]chan *models.AgentTaskResult), } - go agentWSManager.cleanupLoop() + // 启动时，先将所有 "online" 状态的 Agent 重置为 "offline" + NewAgentService().ResetAllAgentsToOffline() + + // 将清理任务注册到系统内部 Cron，每 30 秒执行一次 + executor.GetSysCron().AddJob("@every 30s", agentWSManager.cleanupLoop) }) return agentWSManager } @@ -282,66 +287,51 @@ func (m *AgentWSManager) OnlineCount() int { return len(m.connections) } -// cleanupLoop 清理超时连接 +// cleanupLoop 清理超时连接 (由 SysCron 每 30 秒调用一次) func (m *AgentWSManager) cleanupLoop() { - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() + defer func() { + if r := recover(); r != nil { + logger.Errorf("[AgentWS] cleanupLoop panic: %v", r) + } + }() - // 启动时，先将所有 "online" 状态的 Agent 重置为 "offline" - // 因为 WebSocket 连接在应用启动时是空的，所有 Agent 客观上都是离线状态 - // 等它们重新连接上来后，会变为 "online" - NewAgentService().ResetAllAgentsToOffline() + m.mu.Lock() + defer m.mu.Unlock() + now := time.Now() - for range ticker.C { - func() { - defer func() { - if r := recover(); r != nil { - logger.Errorf("[AgentWS] cleanupLoop panic: %v", r) - } - }() - m.mu.Lock() - now := time.Now() - - // 清理超时连接 - for agentID, conn := range m.connections { - if now.Sub(conn.LastPing) > 2*time.Minute { - // 减少 IP 连接计数 - if conn.IP != "" { - if count, ok := m.ipConnections[conn.IP]; ok && count > 0 { - m.ipConnections[conn.IP] = count - 1 - } - } - conn.Close() - delete(m.connections, agentID) - // 更新数据库状态 - database.DB.Model(&models.Agent{}).Where("id = ?", agentID).Update("status", constant.AgentStatusOffline) - logger.Infof("[AgentWS] Agent #%s 心跳超时，已断开", agentID) + // 清理超时连接 + for agentID, conn := range m.connections { + if now.Sub(conn.LastPing) > 2*time.Minute { + // 减少 IP 连接计数 + if conn.IP != "" { + if count, ok := m.ipConnections[conn.IP]; ok && count > 0 { + m.ipConnections[conn.IP] = count - 1 } } + conn.Close() + delete(m.connections, agentID) + // 更新数据库状态 + database.DB.Model(&models.Agent{}).Where("id = ?", agentID).Update("status", constant.AgentStatusOffline) + logger.Infof("[AgentWS] Agent #%s 心跳超时，已断开", agentID) + } + } - // 定期清理数据库中的过期状态（处理服务重启或异常终止的情况） - // 有些 Agent 虽然没有连接，但数据库状态可能是 "online" - cutoff := now.Add(-2 * time.Minute) - database.DB.Model(&models.Agent{}). - Where("status = ? AND last_seen < ?", constant.AgentStatusOnline, cutoff). - Update("status", constant.AgentStatusOffline) + // 定期清理数据库中的过期状态（处理服务重启或异常终止的情况） + cutoff := now.Add(-2 * time.Minute) + database.DB.Model(&models.Agent{}). + Where("status = ? AND last_seen < ?", constant.AgentStatusOnline, cutoff). + Update("status", constant.AgentStatusOffline) - // 清理过期的限流记录（超过 10 分钟未活动） - - // 清理过期的限流记录（超过 10 分钟未活动） - for ip, lastAttempt := range m.ipLastAttempt { - if now.Sub(lastAttempt) > 10*time.Minute { - delete(m.ipLastAttempt, ip) - delete(m.ipFailCount, ip) - // 只清理没有活跃连接的 IP 计数 - if m.ipConnections[ip] == 0 { - delete(m.ipConnections, ip) - } - } + // 清理过期的限流记录（超过 10 分钟未活动） + for ip, lastAttempt := range m.ipLastAttempt { + if now.Sub(lastAttempt) > 10*time.Minute { + delete(m.ipLastAttempt, ip) + delete(m.ipFailCount, ip) + // 只清理没有活跃连接的 IP 计数 + if m.ipConnections[ip] == 0 { + delete(m.ipConnections, ip) } - - m.mu.Unlock() - }() + } } } diff --git a/internal/services/interconnect_service.go b/internal/services/interconnect_service.go new file mode 100644 index 0000000..bd3adec --- /dev/null +++ b/internal/services/interconnect_service.go @@ -0,0 +1,85 @@ +package services + +import ( + "strings" + + "github.com/engigu/baihu-panel/internal/database" + "github.com/engigu/baihu-panel/internal/models" + "github.com/engigu/baihu-panel/internal/utils" +) + +type InterconnectService struct{} + +func NewInterconnectService() *InterconnectService { + return &InterconnectService{} +} + +func (s *InterconnectService) GetNodes() ([]*models.InterconnectNode, error) { + var nodes []*models.InterconnectNode + err := database.DB.Find(&nodes).Error + return nodes, err +} + +func (s *InterconnectService) GetNodeByID(id string) (*models.InterconnectNode, error) { + var node models.InterconnectNode + err := database.DB.Where("id = ?", id).First(&node).Error + return &node, err +} + +func (s *InterconnectService) CreateNode(name, url, token, remark string) (*models.InterconnectNode, error) { + nodeID := utils.GenerateID() + if url == "" { + url = "tunnel://" + nodeID + } + node := &models.InterconnectNode{ + ID: nodeID, + Name: name, + URL: url, + Token: strings.ToLower(token), + Remark: remark, + CreatedAt: models.Now(), + UpdatedAt: models.Now(), + } + err := database.DB.Create(node).Error + return node, err +} + +func (s *InterconnectService) UpdateNode(id, name, url, token, remark string) (*models.InterconnectNode, error) { + node, err := s.GetNodeByID(id) + if err != nil { + return nil, err + } + node.Name = name + if url != "" { + node.URL = url + } + node.Token = token + node.Remark = remark + node.UpdatedAt = models.Now() + + err = database.DB.Save(node).Error + return node, err +} + +func (s *InterconnectService) DeleteNode(id string) error { + return database.DB.Where("id = ?", id).Delete(&models.InterconnectNode{}).Error +} + +func (s *InterconnectService) GetNodeByToken(token string) (*models.InterconnectNode, error) { + var node models.InterconnectNode + err := database.DB.Where("token = ?", token).First(&node).Error + return &node, err +} + +func (s *InterconnectService) UpdateNodeMonitorData(id string, metrics models.NodeMetrics) error { + now := models.Now() + return database.DB.Model(&models.InterconnectNode{}). + Where("id = ?", id). + Select("status", "metrics", "last_heartbeat_at", "updated_at"). + Updates(models.InterconnectNode{ + Status: "online", + Metrics: metrics, + LastHeartbeatAt: &now, + UpdatedAt: now, + }).Error +} diff --git a/internal/services/monitor_service.go b/internal/services/monitor_service.go new file mode 100644 index 0000000..7a008ff --- /dev/null +++ b/internal/services/monitor_service.go @@ -0,0 +1,132 @@ +package services + +import ( + "math/rand" + "runtime" + "sync" + "time" + + "github.com/engigu/baihu-panel/internal/constant" + "github.com/shirou/gopsutil/v3/cpu" + "github.com/shirou/gopsutil/v3/disk" + "github.com/shirou/gopsutil/v3/host" + "github.com/shirou/gopsutil/v3/mem" +) + +type HostMetrics struct { + CPUPercent float64 + VMem *mem.VirtualMemoryStat + DiskUsage *disk.UsageStat + HostInfo *host.InfoStat +} + +type MonitorService struct { + hostMu sync.RWMutex + lastUpdate time.Time + metrics HostMetrics +} + +var ( + monitorServiceInstance *MonitorService + monitorServiceOnce sync.Once +) + +// GetMonitorService 获取系统监控服务单例 +func GetMonitorService() *MonitorService { + monitorServiceOnce.Do(func() { + monitorServiceInstance = &MonitorService{} + }) + return monitorServiceInstance +} + +// GetHostMetrics 获取并返回物理机状态（带有缓存和演示模式伪装） +func (ms *MonitorService) GetHostMetrics() HostMetrics { + ms.hostMu.Lock() + defer ms.hostMu.Unlock() + + // 缓存 2 秒 + if time.Since(ms.lastUpdate) < 2*time.Second && ms.metrics.VMem != nil { + return ms.metrics + } + + if constant.DemoMode { + ms.updateDemoMetrics() + return ms.metrics + } + + cpuPercents, _ := cpu.Percent(0, false) + if len(cpuPercents) > 0 { + ms.metrics.CPUPercent = cpuPercents[0] + } + ms.metrics.VMem, _ = mem.VirtualMemory() + ms.metrics.DiskUsage, _ = disk.Usage("/") + ms.metrics.HostInfo, _ = host.Info() + ms.lastUpdate = time.Now() + + // 提供默认值防空指针 + if ms.metrics.VMem == nil { + ms.metrics.VMem = &mem.VirtualMemoryStat{} + } + if ms.metrics.DiskUsage == nil { + ms.metrics.DiskUsage = &disk.UsageStat{} + } + if ms.metrics.HostInfo == nil { + ms.metrics.HostInfo = &host.InfoStat{} + } + + return ms.metrics +} + +type RuntimeMetrics struct { + NumGoroutine int + MemStats runtime.MemStats +} + +var ( + runtimeMu sync.RWMutex + lastRuntime time.Time + cachedRuntime RuntimeMetrics +) + +// GetRuntimeMetrics 获取 Go 运行时指标（缓存 2 秒，防止高并发下频繁触发 STW） +func (ms *MonitorService) GetRuntimeMetrics() RuntimeMetrics { + runtimeMu.Lock() + defer runtimeMu.Unlock() + + if time.Since(lastRuntime) < 2*time.Second && cachedRuntime.NumGoroutine > 0 { + return cachedRuntime + } + + cachedRuntime.NumGoroutine = runtime.NumGoroutine() + runtime.ReadMemStats(&cachedRuntime.MemStats) + lastRuntime = time.Now() + + return cachedRuntime +} + +func (ms *MonitorService) updateDemoMetrics() { + ms.metrics.CPUPercent = 10 + rand.Float64()*40 // 10% - 50% 的随机 CPU 波动 + + totalMem := uint64(8 * 1024 * 1024 * 1024) // 8GB 内存 + usedMem := uint64(float64(totalMem) * (0.3 + rand.Float64()*0.3)) // 30% - 60% 随机使用率 + ms.metrics.VMem = &mem.VirtualMemoryStat{ + Total: totalMem, + Used: usedMem, + UsedPercent: float64(usedMem) / float64(totalMem) * 100, + } + + totalDisk := uint64(500 * 1024 * 1024 * 1024) // 500GB 硬盘 + usedDisk := uint64(float64(totalDisk) * 0.45) // 固定 45% 使用率 + ms.metrics.DiskUsage = &disk.UsageStat{ + Total: totalDisk, + Used: usedDisk, + UsedPercent: float64(usedDisk) / float64(totalDisk) * 100, + } + + ms.metrics.HostInfo = &host.InfoStat{ + Platform: "Demo Environment", + OS: "linux", + Uptime: uint64(time.Now().Unix() - 1700000000), // 生成一个较长且持续增加的运行时间 + } + ms.lastUpdate = time.Now() +} diff --git a/internal/tunnel/client.go b/internal/tunnel/client.go new file mode 100644 index 0000000..a43892d --- /dev/null +++ b/internal/tunnel/client.go @@ -0,0 +1,254 @@ +package tunnel + +import ( + "crypto/tls" + "fmt" + "math/rand" + "net/http" + "net/url" + "sync" + "time" + + "github.com/engigu/baihu-panel/internal/constant" + "github.com/engigu/baihu-panel/internal/logger" + "github.com/engigu/baihu-panel/internal/services" + "github.com/gorilla/websocket" + "github.com/hashicorp/yamux" +) + +var ( + tunnelActive bool + clientConn *yamux.Session + clientMu sync.Mutex + stopCh chan struct{} + + // LocalEngine 保存全局的 Gin 引擎或 HTTP 处理器，用于接收主节点的隧道请求并进行纯内存函数路由 + LocalEngine http.Handler +) + +// SetLocalEngine 由 bootstrap 层注入 +func SetLocalEngine(engine http.Handler) { + LocalEngine = engine +} + +// StartClient 尝试连接到父节点面板（如果已配置） +func StartClient() { + clientMu.Lock() + if tunnelActive { + clientMu.Unlock() + return + } + tunnelActive = true + stopCh = make(chan struct{}) + clientMu.Unlock() + + go runClientLoop() +} + +// StopClient 停止后台的隧道客户端 +func StopClient() { + clientMu.Lock() + if !tunnelActive { + clientMu.Unlock() + return + } + tunnelActive = false + if stopCh != nil { + close(stopCh) + } + if clientConn != nil { + clientConn.Close() + } + clientMu.Unlock() +} + +// Init 从数据库加载配置并初始化对应的后台隧道服务 +func Init() { + siteConfig := services.NewSettingsService().GetSection(constant.SectionInterconnect) + role := siteConfig[constant.KeyInterconnectRole] + ApplyRole(role) +} + +// ApplyRole 根据互联角色动态启停背景协程服务 +func ApplyRole(role string) { + switch role { + case constant.InterconnectRoleMaster: + // 主控角色：无需后台轮询，等待子节点上报即可 + StopClient() + case constant.InterconnectRoleChild: + // 子节点：关闭可能存在的主控会话，启动连接守护 + CloseAllSessions() + StartClient() + default: + // 未开启或离线 + StopClient() + CloseAllSessions() + } +} + +// IsTunnelConnected 返回当前子节点隧道是否已连接到主节点 +func IsTunnelConnected() bool { + clientMu.Lock() + defer clientMu.Unlock() + if !tunnelActive || clientConn == nil { + return false + } + return !clientConn.IsClosed() +} + +func runClientLoop() { + settingsSvc := services.NewSettingsService() + + const ( + initialBackoff = 5 * time.Second + maxBackoff = 300 * time.Second + fatalBackoff = 60 * time.Second + ) + + backoff := initialBackoff + retryCount := 0 + var lastErrorMsg string + + for { + clientMu.Lock() + if !tunnelActive { + clientMu.Unlock() + return + } + clientMu.Unlock() + + siteConfig := settingsSvc.GetSection(constant.SectionInterconnect) + + role := siteConfig[constant.KeyInterconnectRole] + if role != constant.InterconnectRoleChild { + time.Sleep(10 * time.Second) + continue + } + + parentURL := siteConfig[constant.KeyInterconnectParentURL] + parentToken := siteConfig[constant.KeyInterconnectParentToken] + + if parentURL == "" || parentToken == "" { + time.Sleep(10 * time.Second) + continue + } + + u, err := url.Parse(parentURL) + if err != nil { + logger.Errorf("[Tunnel] 无效的主节点 URL: %v", err) + time.Sleep(10 * time.Second) + continue + } + + scheme := "ws" + if u.Scheme == "https" { + scheme = "wss" + } + + wsURL := fmt.Sprintf("%s://%s/api/v1/interconnect/tunnel", scheme, u.Host) + + header := http.Header{} + header.Set("Authorization", "Bearer "+parentToken) + + dialer := &websocket.Dialer{ + Proxy: http.ProxyFromEnvironment, + HandshakeTimeout: 45 * time.Second, + } + dialer.EnableCompression = true + if u.Scheme == "https" { + dialer.TLSClientConfig = &tls.Config{ + InsecureSkipVerify: true, + } + } + + conn, resp, err := dialer.Dial(wsURL, header) + if err != nil { + retryCount++ + errMsg := err.Error() + + // 特殊处理致命的认证或配置错误 (仅401作为致命错误，403使用指数退避) + isFatal := resp != nil && resp.StatusCode == 401 + if isFatal { + backoff = fatalBackoff + } + + // 先计算退避时间 (加上 0-8 秒随机抖动防止风暴) + jitter := time.Duration(rand.Intn(8000)) * time.Millisecond + sleepDuration := backoff + jitter + + // 日志记录 + if retryCount == 1 || errMsg != lastErrorMsg { + logger.Errorf("[Tunnel] 连接主节点隧道失败: %v, 下次重试将在 %v 后", err, sleepDuration.Round(time.Second)) + } else { + logger.Warnf("[Tunnel] 仍无法连接主节点，已尝试 %d 次, 下次重试将在 %v 后", retryCount, sleepDuration.Round(time.Second)) + } + + lastErrorMsg = errMsg + + time.Sleep(sleepDuration) + + // 指数级增加退避时间 (最大不超过 maxBackoff) + if !isFatal && backoff < maxBackoff { + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } + continue + } + + // 连接成功，重置退避状态 + retryCount = 0 + lastErrorMsg = "" + backoff = initialBackoff + + logger.Infof("[Tunnel] 已成功连接到主节点隧道: %s", wsURL) + + netConn := NetConn(conn) + trackedConn := &trafficCounterConn{Conn: netConn} + conf := yamux.DefaultConfig() + conf.EnableKeepAlive = true + + // 我们是子节点，接受来自主节点的请求。 + // 因此在这里，我们是 Yamux Server，负责监听传入的流。 + session, err := yamux.Server(trackedConn, conf) + if err != nil { + logger.Errorf("[Tunnel] 启动 Yamux Server 失败: %v", err) + trackedConn.Close() + time.Sleep(5 * time.Second) + continue + } + + // 如果已经存在上一个活跃的隧道连接，在重建前必须将其显式 Close()，以释放该会话在后台运行的 HTTP 监听与监听协程 + clientMu.Lock() + if clientConn != nil { + clientConn.Close() + } + clientConn = session + clientMu.Unlock() + + services.GetSystemWSManager().Broadcast(constant.EventInterconnectChildStatus, map[string]interface{}{ + "connected": true, + }) + + // 启动子节点主动上报服务 + StartReporter(parentURL, parentToken) + + // 启动本地 HTTP 代理服务 (阻塞直到会话关闭) + serveLocalProxy(session) + + // 断开时停止上报 + StopReporter() + + session.Close() + services.GetSystemWSManager().Broadcast(constant.EventInterconnectChildStatus, map[string]interface{}{ + "connected": false, + }) + logger.Infof("[Tunnel] 已从主节点隧道断开连接") + + // 断开后稍微缓冲一下再去重试 + time.Sleep(2 * time.Second) + } +} + + diff --git a/internal/tunnel/conn.go b/internal/tunnel/conn.go new file mode 100644 index 0000000..8c190c8 --- /dev/null +++ b/internal/tunnel/conn.go @@ -0,0 +1,87 @@ +package tunnel + +import ( + "io" + "net" + "time" + + "github.com/gorilla/websocket" +) + +// wsConn 包装 gorilla websocket 以实现 net.Conn 接口 +type wsConn struct { + conn *websocket.Conn + reader io.Reader +} + +// NetConn 将 WebSocket 连接转换为标准 net.Conn +func NetConn(conn *websocket.Conn) net.Conn { + return &wsConn{ + conn: conn, + } +} + +func (c *wsConn) Read(b []byte) (n int, err error) { + for { + if c.reader == nil { + _, r, err := c.conn.NextReader() + if err != nil { + return 0, err + } + c.reader = r + } + n, err = c.reader.Read(b) + if err == io.EOF { + c.reader = nil + if n > 0 { + return n, nil + } + continue + } + return n, err + } +} + +func (c *wsConn) Write(b []byte) (n int, err error) { + // Yamux 保证了对底层连接的序列化写入（不会出现并发写）， + // 因此我们这里不需要加互斥锁 (Mutex)。 + w, err := c.conn.NextWriter(websocket.BinaryMessage) + if err != nil { + return 0, err + } + n, err = w.Write(b) + + // 必须关闭 writer 以触发 WebSocket 的消息帧封包发送 + closeErr := w.Close() + if err == nil { + err = closeErr + } + return n, err +} + +func (c *wsConn) Close() error { + return c.conn.Close() +} + +func (c *wsConn) LocalAddr() net.Addr { + return c.conn.LocalAddr() +} + +func (c *wsConn) RemoteAddr() net.Addr { + return c.conn.RemoteAddr() +} + +func (c *wsConn) SetDeadline(t time.Time) error { + if err := c.conn.SetReadDeadline(t); err != nil { + return err + } + return c.conn.SetWriteDeadline(t) +} + +func (c *wsConn) SetReadDeadline(t time.Time) error { + return c.conn.SetReadDeadline(t) +} + +func (c *wsConn) SetWriteDeadline(t time.Time) error { + return c.conn.SetWriteDeadline(t) +} diff --git a/internal/tunnel/proxy.go b/internal/tunnel/proxy.go new file mode 100644 index 0000000..62fbf93 --- /dev/null +++ b/internal/tunnel/proxy.go @@ -0,0 +1,109 @@ +package tunnel + +import ( + "errors" + "net/http" + "net/http/httputil" + "strings" + "time" + + "github.com/engigu/baihu-panel/internal/logger" + + "github.com/gin-gonic/gin" + "github.com/hashicorp/yamux" +) + +// ProxyHTTP 以完全透明的流式转发将 HTTP 请求代理到远程节点 (主节点调用) +func ProxyHTTP(nodeID string, c *gin.Context, targetPath string) error { + sess := GetSession(nodeID) + if sess == nil { + return errors.New("node is offline or tunnel not established") + } + + if sess.Token == "" { + return errors.New("node token not found in session cache") + } + + // 获取真实客户端 IP，传递给子节点用于审计日志等 + clientIP := c.ClientIP() + + director := func(req *http.Request) { + req.URL.Scheme = "http" + req.URL.Host = "tunnel.local" + req.URL.Path = targetPath + // gin 和 httputil 默认会保留并透传 Query 参数 + + req.Header.Set("Authorization", "Bearer "+sess.Token) + // 删除父节点的 Cookie 避免干扰子节点的会话验证 + req.Header.Del("Cookie") + + // 强制加上该 Header，防止出现节点相互代理的死循环 + req.Header.Set("X-Tunnel-Proxy", "true") + + // 传递真实的客户端 IP 给子节点 + if clientIP != "" { + req.Header.Set("X-Forwarded-For", clientIP) + } + } + + proxy := &httputil.ReverseProxy{ + Director: director, + Transport: sess.Transport, // 直接复用 AddSession 时创建的全局连接池，由原生库处理 Context 取消 + ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) { + logger.Errorf("[Tunnel] 节点 %s 代理隧道请求发生错误: %v", nodeID, err) + w.WriteHeader(http.StatusBadGateway) + w.Write([]byte("Tunnel connection error: " + err.Error())) + }, + } + + proxy.ServeHTTP(c.Writer, c.Request) + return nil +} + +// serveLocalProxy 本地服务代理逻辑 (子节点调用) +func serveLocalProxy(session *yamux.Session) { + if LocalEngine == nil { + logger.Errorf("[Tunnel] 无法启动本地代理服务，因为 LocalEngine 尚未注入。") + return + } + + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // 安全规则 1：防止互相调用的死循环 + if strings.HasPrefix(r.URL.Path, "/api/v1/interconnect/proxy/") { + w.WriteHeader(http.StatusForbidden) + w.Write([]byte("Recursive proxy loops are not allowed via tunnel")) + return + } + + // 安全规则 2：严格限制只能访问本机的 /api/v1/ 接口 + if !strings.HasPrefix(r.URL.Path, "/api/v1/") { + w.WriteHeader(http.StatusForbidden) + w.Write([]byte("Tunnel access is strictly restricted to /api/v1/ endpoints")) + return + } + + // 确保保留 X-Tunnel-Proxy 请求头，以防后续逻辑需要判定 + r.Header.Set("X-Tunnel-Proxy", "true") + + // 纯内存函数调用，直接扔给 Gin 引擎，零网络开销！ + LocalEngine.ServeHTTP(w, r) + }) + + server := &http.Server{ + Handler: handler, + ReadTimeout: 15 * time.Second, + IdleTimeout: 15 * time.Second, + } + + // 监听会话关闭信号，主动关闭 http.Server 以彻底释放协程 + go func() { + <-session.CloseChan() + server.Close() + }() + + // yamux.Session 实现了标准的 net.Listener 接口！ + err := server.Serve(session) + if err != nil && err != http.ErrServerClosed && err != yamux.ErrSessionShutdown { + logger.Errorf("[Tunnel] Yamux 代理服务意外停止: %v", err) + } +} diff --git a/internal/tunnel/reporter.go b/internal/tunnel/reporter.go new file mode 100644 index 0000000..beff10b --- /dev/null +++ b/internal/tunnel/reporter.go @@ -0,0 +1,117 @@ +package tunnel + +import ( + "bytes" + "encoding/json" + "net/http" + "strings" + "sync" + "time" + + "github.com/engigu/baihu-panel/internal/executor" + "github.com/engigu/baihu-panel/internal/logger" + "github.com/engigu/baihu-panel/internal/services" + "github.com/robfig/cron/v3" +) + +var ( + reporterCronID cron.EntryID + isReporting bool + localTunnelURL string + localTunnelURLMutex sync.RWMutex +) + +// GetLocalTunnelURL 返回当前子节点被分配的隧道地址 +func GetLocalTunnelURL() string { + localTunnelURLMutex.RLock() + defer localTunnelURLMutex.RUnlock() + return localTunnelURL +} + +// StartReporter 启动子节点上报守护进程 +func StartReporter(parentURL, token string) { + if isReporting { + return + } + isReporting = true + + // 修整 parentURL 确保正确指向 /api/v1/interconnect/report + // 如果用户填的是 http://192.168.1.100:8000，我们要加上路径 + baseURL := strings.TrimRight(parentURL, "/") + // 注意兼容如果用户带了 /api/v1 + if !strings.HasSuffix(baseURL, "/api/v1") && !strings.Contains(baseURL, "/api/v1") { + baseURL = baseURL + "/api/v1" + } + reportURL := baseURL + "/interconnect/report" + + // 使用全局 SysCron 并立即上报一次，随后每 45 秒循环一次 + id, err := executor.GetSysCron().AddJobWithRun("@every 45s", func() { + reportMonitorData(reportURL, token) + }) + + if err == nil { + reporterCronID = id + } else { + logger.Warnf("[Tunnel] 无法将上报任务加入 SysCron: %v", err) + } +} + +// StopReporter 停止子节点上报守护进程 +func StopReporter() { + if isReporting { + if reporterCronID != 0 { + executor.GetSysCron().RemoveJob(reporterCronID) + reporterCronID = 0 + } + isReporting = false + } +} + +func reportMonitorData(reportURL, token string) { + // 调用单例监控服务获取底层系统状态 + metrics := services.GetMonitorService().GetHostMetrics() + + payload := map[string]interface{}{ + "cpu_percent": metrics.CPUPercent, + "mem_percent": metrics.VMem.UsedPercent, + "disk_percent": metrics.DiskUsage.UsedPercent, + "tx_bytes": GetTxBytes(), + "rx_bytes": GetRxBytes(), + } + + data, _ := json.Marshal(payload) + + req, err := http.NewRequest("POST", reportURL, bytes.NewBuffer(data)) + if err != nil { + return + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + + client := &http.Client{Timeout: 10 * time.Second} + resp, err := client.Do(req) + if err != nil { + logger.Warnf("[Tunnel] 上报监控数据失败: %v", err) + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + logger.Warnf("[Tunnel] 上报监控数据返回异常状态码: %d", resp.StatusCode) + return + } + + var res struct { + Data struct { + TunnelURL string `json:"tunnel_url"` + } `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&res); err == nil { + if res.Data.TunnelURL != "" { + localTunnelURLMutex.Lock() + localTunnelURL = res.Data.TunnelURL + localTunnelURLMutex.Unlock() + } + } +} diff --git a/internal/tunnel/server.go b/internal/tunnel/server.go new file mode 100644 index 0000000..e0a881f --- /dev/null +++ b/internal/tunnel/server.go @@ -0,0 +1,96 @@ +package tunnel + +import ( + "strings" + + "github.com/engigu/baihu-panel/internal/constant" + "github.com/engigu/baihu-panel/internal/database" + "github.com/engigu/baihu-panel/internal/logger" + "github.com/engigu/baihu-panel/internal/models" + "github.com/engigu/baihu-panel/internal/services" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "net/http" + "github.com/hashicorp/yamux" +) + +var upgrader = websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { + return true + }, + EnableCompression: true, // 启用协议级别的 Deflate 数据压缩 +} + +func HandleTunnel(c *gin.Context) { + authHeader := c.GetHeader("Authorization") + if authHeader == "" { + c.JSON(401, gin.H{"error": "missing authorization"}) + return + } + + tokenStr := strings.TrimSpace(strings.TrimPrefix(authHeader, "Bearer ")) + if tokenStr == "" { + c.JSON(401, gin.H{"error": "invalid token format"}) + return + } + + settingsSvc := services.NewSettingsService() + siteConfig := settingsSvc.GetSection(constant.SectionInterconnect) + role := siteConfig[constant.KeyInterconnectRole] + + if role != constant.InterconnectRoleMaster { + c.JSON(403, gin.H{"error": "interconnect master role not enabled on this server"}) + return + } + + var node models.InterconnectNode + if err := database.DB.Where("token = ?", tokenStr).First(&node).Error; err != nil { + c.JSON(401, gin.H{"error": "invalid interconnect token"}) + return + } + + nodeID := node.ID + + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + logger.Errorf("[Tunnel] 升级隧道连接失败 (NodeID: %s): %v", nodeID, err) + return + } + + // 将 WebSocket 包装为标准的 net.Conn + netConn := NetConn(conn) + + // 获取客户端真实 IP 和端口以更新连接地址，方便面板显示 + clientIP := c.ClientIP() + port := "unknown" + if parts := strings.Split(netConn.RemoteAddr().String(), ":"); len(parts) > 1 { + port = parts[len(parts)-1] + } + realURL := "tunnel://" + clientIP + ":" + port + + if node.URL != realURL { + database.DB.Model(&node).Update("url", realURL) + } + + // 我们是主节点（父节点），主动发起对子节点的请求。 + // 因此在 Yamux 协议中我们扮演 Client，而子节点扮演 Server。 + conf := yamux.DefaultConfig() + conf.EnableKeepAlive = true + + session, err := yamux.Client(netConn, conf) + if err != nil { + logger.Errorf("[Tunnel] 启动 Yamux Client 失败 (NodeID: %s): %v", nodeID, err) + netConn.Close() + return + } + + AddSession(nodeID, tokenStr, session) + logger.Infof("[Tunnel] 已建立来自子节点的隧道连接: %s (%s)", nodeID, node.Name) + + go func(s *yamux.Session) { + <-s.CloseChan() + logger.Infof("[Tunnel] 子节点隧道连接已断开 (NodeID: %s, IsClosed: %v)", nodeID, s.IsClosed()) + RemoveSession(nodeID, s) + }(session) +} diff --git a/internal/tunnel/session.go b/internal/tunnel/session.go new file mode 100644 index 0000000..f55437e --- /dev/null +++ b/internal/tunnel/session.go @@ -0,0 +1,91 @@ +package tunnel + +import ( + "context" + "net" + "net/http" + "sync" + + "github.com/engigu/baihu-panel/internal/database" + "github.com/engigu/baihu-panel/internal/models" + "github.com/hashicorp/yamux" +) + +// TunnelSession 表示一个基于 WebSocket 的 Yamux 活跃会话 +type TunnelSession struct { + NodeID string + Token string + Session *yamux.Session + Transport *http.Transport +} + +var ( + sessions = make(map[string]*TunnelSession) + sessionsMu sync.RWMutex +) + +func AddSession(nodeID string, token string, session *yamux.Session) *TunnelSession { + sessionsMu.Lock() + defer sessionsMu.Unlock() + + // 关闭已经存在的旧会话 + if old, exists := sessions[nodeID]; exists { + old.Session.Close() + old.Transport.CloseIdleConnections() + } + + tr := &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + return session.Open() + }, + DisableKeepAlives: true, + } + + sess := &TunnelSession{ + NodeID: nodeID, + Token: token, + Session: session, + Transport: tr, + } + sessions[nodeID] = sess + return sess +} + +func RemoveSession(nodeID string, session *yamux.Session) { + sessionsMu.Lock() + defer sessionsMu.Unlock() + + if sess, exists := sessions[nodeID]; exists && sess.Session == session { + sess.Session.Close() + sess.Transport.CloseIdleConnections() + delete(sessions, nodeID) + + // 节点下线时实时更新数据库状态 + database.DB.Model(&models.InterconnectNode{}). + Where("id = ?", nodeID). + Update("status", "offline") + } +} + +// CloseAllSessions 强制关闭所有现存的子节点会话（用于主控角色被取消时） +func CloseAllSessions() { + sessionsMu.Lock() + defer sessionsMu.Unlock() + + for _, sess := range sessions { + if sess.Session != nil { + sess.Session.Close() + } + if sess.Transport != nil { + sess.Transport.CloseIdleConnections() + } + } + // 清空所有的会话 + sessions = make(map[string]*TunnelSession) +} + +func GetSession(nodeID string) *TunnelSession { + sessionsMu.RLock() + defer sessionsMu.RUnlock() + return sessions[nodeID] +} diff --git a/internal/tunnel/traffic.go b/internal/tunnel/traffic.go new file mode 100644 index 0000000..1c9bdfe --- /dev/null +++ b/internal/tunnel/traffic.go @@ -0,0 +1,48 @@ +package tunnel + +import ( + "net" + "sync/atomic" +) + +var ( + txBytes uint64 + rxBytes uint64 +) + +// trafficCounterConn 包装了原生的 net.Conn，用于统计底层收发的真实物理字节数 +type trafficCounterConn struct { + net.Conn +} + +func (c *trafficCounterConn) Read(b []byte) (n int, err error) { + n, err = c.Conn.Read(b) + if n > 0 { + atomic.AddUint64(&rxBytes, uint64(n)) + } + return +} + +func (c *trafficCounterConn) Write(b []byte) (n int, err error) { + n, err = c.Conn.Write(b) + if n > 0 { + atomic.AddUint64(&txBytes, uint64(n)) + } + return +} + +// GetTxBytes 返回隧道累计发送的字节数 +func GetTxBytes() uint64 { + return atomic.LoadUint64(&txBytes) +} + +// GetRxBytes 返回隧道累计接收的字节数 +func GetRxBytes() uint64 { + return atomic.LoadUint64(&rxBytes) +} + +// ResetTrafficBytes 重置流量统计计数器 +func ResetTrafficBytes() { + atomic.StoreUint64(&txBytes, 0) + atomic.StoreUint64(&rxBytes, 0) +} diff --git a/web/src/App.vue b/web/src/App.vue index cb9c02d..23aa376 100644 --- a/web/src/App.vue +++ b/web/src/App.vue @@ -2,6 +2,13 @@ import { onMounted } from 'vue' import { RouterView } from 'vue-router' import { Toaster } from '@/components/ui/sonner' +import { LogOut } from 'lucide-vue-next' +import { activeInterconnectNodeId, setActiveInterconnectNodeId } from '@/api' + +function exitTravel() { + setActiveInterconnectNodeId('') + window.location.href = '/' +} onMounted(() => { // 全局应用设备差异化垂直抗锯齿 @@ -19,4 +26,18 @@ onMounted(() => {