feat: add openconnect

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)
+	}
+}
+
+

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)
+}

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)
+	}
+}

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()
+		}
+	}
+}

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)
+}

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]
+}

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)
+}