From 7a0ec8ded93d3897af8056a96704750d19820ab6 Mon Sep 17 00:00:00 2001 From: W11 Date: Fri, 25 Sep 2026 01:21:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(node,agent):=20=E8=BD=AC=E5=8F=91=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=E4=B8=A4=E5=A4=84=E6=85=A2=E5=A4=B1=E8=B4=A5=E8=A1=A5?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E2=80=94=E2=80=94=E2=91=A0=E8=8A=82=E7=82=B9?= =?UTF-8?q?=E8=BD=AC=E5=8F=91=E6=8F=A1=E6=89=8B(=E5=91=8A=E7=9F=A5?= =?UTF-8?q?=E7=9B=AE=E6=A0=87+=E7=AD=89agent=E5=9B=9E=E6=89=A7)=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=20HandshakeTimeout=20=E9=BB=98=E8=AE=A45s,=20agent?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E5=8D=8A=E6=AD=BB=E6=97=B6=E8=A7=82=E7=9C=8B?= =?UTF-8?q?=E7=AB=AF=E4=B8=8D=E5=86=8D=E6=97=A0=E9=99=90=E6=8C=82=E8=B5=B7?= =?UTF-8?q?=E5=88=B0=E7=82=B9=E5=8D=B3=E6=96=AD=E3=80=81=E6=8F=A1=E6=89=8B?= =?UTF-8?q?=E6=88=90=E5=8A=9F=E5=8D=B3=E6=B8=85=E9=99=A4=E4=B8=8D=E7=BA=A6?= =?UTF-8?q?=E6=9D=9F=E6=95=B0=E6=8D=AE=E9=98=B6=E6=AE=B5;=20=E2=91=A1agent?= =?UTF-8?q?=20=E6=8B=A8=E5=86=85=E7=BD=91=E7=9B=AE=E6=A0=87=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=20DialTimeout=20=E9=BB=98=E8=AE=A45s(Agent=20?= =?UTF-8?q?=E5=8F=AF=E9=85=8D),=20=E7=9B=AE=E6=A0=87=E7=A6=BB=E7=BA=BF?= =?UTF-8?q?=E4=B8=94=E9=98=B2=E7=81=AB=E5=A2=99=E4=B8=A2=E5=8C=85=E6=97=B6?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E8=A3=B8=20Dial=20=E5=90=8A20s+,=20=E5=88=B0?= =?UTF-8?q?=E7=82=B9=E5=9B=9E=E6=89=A7=E5=A4=B1=E8=B4=A5=E8=A7=82=E7=9C=8B?= =?UTF-8?q?=E7=AB=AF=E7=AB=8B=E5=8D=B3=E6=96=AD=E5=BC=80;=20=E9=99=84=20e2?= =?UTF-8?q?e=20=E7=BA=A2=E7=BB=BF=E5=8F=8C=E5=90=91=E8=AF=81=E6=98=8E(?= =?UTF-8?q?=E5=8D=8A=E6=AD=BBagent/=E9=BB=91=E6=B4=9E=E6=8B=A8=E5=8F=B7=20?= =?UTF-8?q?=E4=B8=A4=E5=9C=BA=E6=99=AF=E6=92=A4=E4=BF=AE=E5=A4=8D=E5=BF=85?= =?UTF-8?q?=E7=BA=A2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent/agent.go | 23 ++++--- agent/forward.go | 6 +- e2e/handshake_test.go | 136 ++++++++++++++++++++++++++++++++++++++++ internal/node/node.go | 18 +++--- internal/node/tunnel.go | 7 +++ 5 files changed, 172 insertions(+), 18 deletions(-) create mode 100644 e2e/handshake_test.go diff --git a/agent/agent.go b/agent/agent.go index d1d2be4..796d563 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -28,6 +28,9 @@ import ( const ( defaultBackoff = 5 * time.Second defaultPing = 30 * time.Second + // defaultDialTimeout 内网目标拨号上限：目标离线且防火墙丢包(不回 RST)时 + // 裸 Dial 在 Windows 上要 SYN 重传 20s+, 观看端在节点侧只能干等。 + defaultDialTimeout = 5 * time.Second ) // Logger agent 日志接口。logrus.Logger 的 Debugf/Infof/Warnf/Errorf 签名 @@ -59,9 +62,10 @@ func (DefaultLogger) Errorf(format string, args ...any) { // Agent 被控端：隧道注册表。每条 Tunnel 携带自己的 NodeAddr+Token， // 由独立 worker（tunnelConn）连接登录各自节点；Agent 不再有集中连接与登录。 type Agent struct { - AgentId string - Backoff time.Duration // 永久隧道重连退避（默认 5s） - PingEvery time.Duration // 各隧道控制流心跳间隔（默认 30s） + AgentId string + Backoff time.Duration // 永久隧道重连退避（默认 5s） + PingEvery time.Duration // 各隧道控制流心跳间隔（默认 30s） + DialTimeout time.Duration // 内网目标拨号上限（默认 5s; 超时向节点回执失败, 观看端立即断开） // Logger 日志注入点。需要纳入宿主统一日志体系时设置 // （logrus 直接注入 logrus.StandardLogger() 即可）；未设置用 DefaultLogger。 @@ -82,12 +86,13 @@ type Agent struct { // New 创建被控端。节点地址与令牌随各隧道下发（Tunnel.NodeAddr/Token）。 func New(agentId string) *Agent { return &Agent{ - AgentId: agentId, - Backoff: defaultBackoff, - PingEvery: defaultPing, - Logger: DefaultLogger{}, - tunnels: make(map[string]*tunnelConn), - done: make(chan struct{}), + AgentId: agentId, + Backoff: defaultBackoff, + PingEvery: defaultPing, + DialTimeout: defaultDialTimeout, + Logger: DefaultLogger{}, + tunnels: make(map[string]*tunnelConn), + done: make(chan struct{}), } } diff --git a/agent/forward.go b/agent/forward.go index 7ca43bd..f399b1e 100644 --- a/agent/forward.go +++ b/agent/forward.go @@ -22,7 +22,11 @@ func (a *Agent) handleData(stream net.Conn) { return } - conn, err := net.Dial("tcp", t.TargetAddr()) + dialTimeout := a.DialTimeout + if dialTimeout <= 0 { // 零值 Agent 兜底 + dialTimeout = defaultDialTimeout + } + conn, err := net.DialTimeout("tcp", t.TargetAddr(), dialTimeout) if err != nil { a.logger().Warnf("agent 拨号本地目标失败 target=%s err=%s", t.TargetAddr(), err.Error()) rf := NewFrame(FrameVersion, CmdTarget, f.StreamID()) diff --git a/e2e/handshake_test.go b/e2e/handshake_test.go new file mode 100644 index 0000000..9d4625a --- /dev/null +++ b/e2e/handshake_test.go @@ -0,0 +1,136 @@ +// 转发链路"慢失败"测试：两处历史缺陷的回归锚—— +// 1. 节点转发握手无上限：agent 会话半死(收得到帧但永不应答)时观看端无限挂起 +// → HandshakeTimeout 到点必须切断观看端, 让其快速失败可重试; +// 2. agent 拨内网目标无上限：目标离线且防火墙丢包(不回 RST)时裸 Dial 吊 20s+ +// → DialTimeout 到点必须向节点回执失败, 观看端立即断开。 +package e2e + +import ( + "errors" + "fmt" + "net" + "strconv" + "testing" + "time" + + "git.zeroonesoft.cn/golib/zonat/agent" + "git.zeroonesoft.cn/golib/zonat/internal/node" + + "github.com/xtaci/smux" +) + +// deafAgent 手工模拟"半死 agent"：完成登录+注册隧道后, 对节点打开的数据流 +// (CmdTarget)永不应答。返回节点为该隧道分配的公网端口。 +func deafAgent(t *testing.T, n *node.Node, tunnelId, targetIp string, targetPort int) int { + t.Helper() + conn, err := net.DialTimeout("tcp", n.Addr(), 2*time.Second) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + + tok, err := node.SignToken(n.JwtSecret, "deaf-agent", time.Hour) + if err != nil { + t.Fatal(err) + } + lf := agent.NewFrame(agent.FrameVersion, agent.CmdLogin, 0) + if err := lf.Marshal(agent.Login{AgentId: "deaf-agent", Token: tok}); err != nil { + t.Fatal(err) + } + if err := agent.WriteFrame(conn, lf); err != nil { + t.Fatal(err) + } + rf, err := agent.ReadFrame(conn) + if err != nil { + t.Fatal(err) + } + ret := agent.Ret{} + if err := rf.Unmarshal(&ret); err != nil || ret.Code != 0 { + t.Fatalf("登录失败: %+v", ret) + } + + sess, err := smux.Client(conn, agent.SmuxConfig()) + if err != nil { + t.Fatal(err) + } + control, err := sess.OpenStream() + if err != nil { + t.Fatal(err) + } + reg := agent.NewFrame(agent.FrameVersion, agent.CmdRegisterTunnel, 0) + if err := reg.Marshal(agent.Tunnel{Id: tunnelId, TargetIp: targetIp, TargetPort: targetPort, TTLSec: 0}); err != nil { + t.Fatal(err) + } + if err := agent.WriteFrame(control, reg); err != nil { + t.Fatal(err) + } + rr, err := agent.ReadFrame(control) + if err != nil { + t.Fatal(err) + } + rret := agent.Ret{} + if err := rr.Unmarshal(&rret); err != nil || rret.Code != 0 { + t.Fatalf("注册失败: %+v", rret) + } + return rret.Port +} + +// assertViewerCutBy 断言观看端在时限内被节点断开(读到 EOF/重置, 而非超时)。 +func assertViewerCutBy(t *testing.T, port int, within time.Duration) { + t.Helper() + viewer, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), 2*time.Second) + if err != nil { + t.Fatal(err) + } + defer viewer.Close() + _ = viewer.SetReadDeadline(time.Now().Add(within)) + buf := make([]byte, 16) + _, err = viewer.Read(buf) + if err == nil { + t.Fatalf("观看端读到了数据, 不应发生") + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + t.Fatalf("观看端 %v 内未被节点断开(超时切断未生效)", within) + } +} + +// TestHandshakeTimeoutCutsViewer 半死 agent: 节点必须在握手超时内切断观看端 +// (修复前观看端无限挂起, 只能等 smux keepalive 判死整个会话 ~40s)。 +func TestHandshakeTimeoutCutsViewer(t *testing.T) { + target := startEchoServer(t) + n := node.New() + n.JwtSecret = "test-jwt-secret" + n.BindTunnel = "127.0.0.1" + n.HandshakeTimeout = 500 * time.Millisecond + if err := n.Start("127.0.0.1:0"); err != nil { + t.Fatal(err) + } + t.Cleanup(n.Stop) + + host, portStr, _ := net.SplitHostPort(target) + portNum, _ := strconv.Atoi(portStr) + listen := deafAgent(t, n, fmt.Sprintf("deaf-%d", time.Now().UnixNano()), host, portNum) + + assertViewerCutBy(t, listen, 3*time.Second) +} + +// TestAgentDialTimeoutCutsViewer 目标黑洞(不回 RST): agent 拨号到点回执失败, +// 观看端立即断开(修复前裸 Dial 在 Windows 吊 20s+, 观看端干等)。 +func TestAgentDialTimeoutCutsViewer(t *testing.T) { + // 本机网络若把 TEST-NET-3 判定为可达(隧道/虚拟网卡), 黑洞前提不成立 + if c, err := net.DialTimeout("tcp", "203.0.113.1:5900", 300*time.Millisecond); err == nil { + c.Close() + t.Skip("本机可将 TEST-NET-3 判定连通, 黑洞拨号场景不成立") + } + + n, a := startEnv(t) + a.DialTimeout = 300 * time.Millisecond + id := fmt.Sprintf("blackhole-%d", time.Now().UnixNano()) + port, err := a.RegisterTunnel(buildTunnel(t, n, a, id, "203.0.113.1:5900", 0)) + if err != nil { + t.Fatal(err) + } + + assertViewerCutBy(t, port, 3*time.Second) +} diff --git a/internal/node/node.go b/internal/node/node.go index ec28753..cbbbca5 100644 --- a/internal/node/node.go +++ b/internal/node/node.go @@ -24,25 +24,27 @@ const loginTimeout = 10 * time.Second // Node 穿透节点 type Node struct { - JwtSecret string // agent 登录 JWT 校验密钥（HS256 系） - BindTunnel string // 隧道监听绑定地址（默认 0.0.0.0） - SweepInterval time.Duration // TTL 清扫间隔（默认 30s） + JwtSecret string // agent 登录 JWT 校验密钥（HS256 系） + BindTunnel string // 隧道监听绑定地址（默认 0.0.0.0） + SweepInterval time.Duration // TTL 清扫间隔（默认 30s） + HandshakeTimeout time.Duration // 单条观看连接的 agent 拨号应答上限（默认 5s; 超时立即断开观看端, 数据阶段不受限） listener net.Listener mu sync.Mutex stopped bool - sessions map[*Session]struct{} // 该节点上的全部 agent 会话（一个 agent 多隧道=多会话） + sessions map[*Session]struct{} // 该节点上的全部 agent 会话（一个 agent 多隧道=多会话） tunnels map[string]*TunnelServer // tunnelId -> 隧道 } // New 创建节点 func New() *Node { return &Node{ - BindTunnel: "0.0.0.0", - SweepInterval: 30 * time.Second, - sessions: make(map[*Session]struct{}), - tunnels: make(map[string]*TunnelServer), + BindTunnel: "0.0.0.0", + SweepInterval: 30 * time.Second, + HandshakeTimeout: 5 * time.Second, + sessions: make(map[*Session]struct{}), + tunnels: make(map[string]*TunnelServer), } } diff --git a/internal/node/tunnel.go b/internal/node/tunnel.go index 99a0fee..cf5004e 100644 --- a/internal/node/tunnel.go +++ b/internal/node/tunnel.go @@ -58,6 +58,11 @@ func (t *TunnelServer) handleConn(conn net.Conn) { } defer stream.Close() + // 转发握手(告知目标+等 agent 回执)限时: agent 会话半死/目标拨号不回时 + // 观看端不再无限挂起, 到点立即断开让观看端快速失败可重试 + if hs := t.session.node.HandshakeTimeout; hs > 0 { + _ = stream.SetDeadline(time.Now().Add(hs)) + } tf := agent.NewFrame(agent.FrameVersion, agent.CmdTarget, 0) if err := tf.Marshal(t.tunnel); err != nil { return @@ -67,8 +72,10 @@ func (t *TunnelServer) handleConn(conn net.Conn) { } rf, err := agent.ReadFrame(stream) if err != nil { + slog.Warn("节点 agent拨号应答失败", "id", t.tunnel.Id, "err", err.Error()) return } + _ = stream.SetDeadline(time.Time{}) // 数据阶段不受握手超时约束 ret := agent.Ret{} if err := rf.Unmarshal(&ret); err != nil || ret.Code != 0 { slog.Warn("节点 agent拨号失败", "id", t.tunnel.Id, "msg", ret.Msg)