init: 自 zomaintain/backend/zonat 平移独立成库; module path 改为 git.zeroonesoft.cn/golib/zonat; 内部 import 同步重写

2026-09-19 18:55:43 +08:00
commit 984041f1a0
20 changed files with 2427 additions and 0 deletions
@@ -0,0 +1,222 @@
+// Package agent 轻量穿透被控端（zonat client 的裁剪版），设计为**可嵌入公开包**：
+// 每条隧道自带节点地址与登录令牌，由独立连接完成
+// 登录节点 → smux 会话 → 控制流注册自身 → 数据流拨号本地 TCP 目标，
+// 因此一个 agent 的多条隧道可以分布在**多个不同节点**。
+// Agent 本身只是隧道的注册表与生命周期管理器，不持有网络连接、不做登录。
+// 相比 zonat client 裁掉：KCP 拨号、UDP 代理、普通(非复用)连接模式、
+// 用户级通道分发、服务安装等本机管理逻辑。
+// 永久隧道（TTLSec=0）断线各自自动重连重注册；临时隧道（TTLSec>0）
+// 一次性，连接断开或节点 TTL 过期推送后即消失，与 zonat 临时通道语义一致。
+//
+// 日志：包内不依赖任何具体日志实现，宿主项目通过注入 Logger 接口
+// 把 agent 日志纳入自己的日志体系（格式/级别/输出/文件路径归一）；
+// 不注入时使用仅标准库的兜底实现（stderr，Info 级）。
+// 注意：日志中的 file:line 归属调用点。宿主以 import 方式引用本包并加
+// -trimpath 构建时，路径为模块形式 zonat@版本/agent/agent.go（跨机器稳定）；
+// 以整目录拷贝方式放进宿主项目时，路径在宿主项目内，宿主现有 rootDir
+// 削前缀规则原样生效。两种方式都不需要在包内做任何路径处理。
+package agent
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+)
+
+const (
+	defaultBackoff = 5 * time.Second
+	defaultPing    = 30 * time.Second
+)
+
+// Logger agent 日志接口。logrus.Logger 的 Debugf/Infof/Warnf/Errorf 签名
+// 与此完全一致，可直接注入；slog 等其它实现写一个几行的适配器即可。
+type Logger interface {
+	Debugf(format string, args ...any)
+	Infof(format string, args ...any)
+	Warnf(format string, args ...any)
+	Errorf(format string, args ...any)
+}
+
+// DefaultLogger 未注入时的兜底日志：标准错误，Info 级，仅标准库实现。
+type DefaultLogger struct{}
+
+func (DefaultLogger) Debugf(string, ...any) {}
+
+func (DefaultLogger) Infof(format string, args ...any) {
+	log.Printf("INFO %s", fmt.Sprintf(format, args...))
+}
+
+func (DefaultLogger) Warnf(format string, args ...any) {
+	log.Printf("WARN %s", fmt.Sprintf(format, args...))
+}
+
+func (DefaultLogger) Errorf(format string, args ...any) {
+	log.Printf("ERROR %s", fmt.Sprintf(format, args...))
+}
+
+// Agent 被控端：隧道注册表。每条 Tunnel 携带自己的 NodeAddr+Token，
+// 由独立 worker（tunnelConn）连接登录各自节点；Agent 不再有集中连接与登录。
+type Agent struct {
+	AgentId   string
+	Backoff   time.Duration // 永久隧道重连退避（默认 5s）
+	PingEvery time.Duration // 各隧道控制流心跳间隔（默认 30s）
+
+	// Logger 日志注入点。需要纳入宿主统一日志体系时设置
+	// （logrus 直接注入 logrus.StandardLogger() 即可）；未设置用 DefaultLogger。
+	Logger Logger
+
+	// OnTunnelPort 隧道端口确定/变更回调：port>0 已注册，port<0 已关闭。
+	// 节点随机分配端口时，上层靠它拿到实际端口。多隧道场景下可能从
+	// 不同 worker 并发回调，宿主需自行保证线程安全。
+	OnTunnelPort func(id string, port int)
+
+	mu      sync.Mutex
+	closed  bool
+	tunnels map[string]*tunnelConn
+
+	done chan struct{} // Run 的阻塞点，Close 时关闭
+}
+
+// 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{}),
+	}
+}
+
+// logger 取当前日志实现（零值 Agent 未注入时兜底）
+func (a *Agent) logger() Logger {
+	a.mu.Lock()
+	l := a.Logger
+	a.mu.Unlock()
+	if l == nil {
+		return DefaultLogger{}
+	}
+	return l
+}
+
+// Run 阻塞直到 Close。隧道的连接在 RegisterTunnel 时即自行驱动，
+// 这里仅保留"go a.Run()"式宿主用法的阻塞点。
+func (a *Agent) Run() error {
+	<-a.done
+	return nil
+}
+
+// Close 停止 agent：断开并停止全部隧道 worker
+func (a *Agent) Close() {
+	a.mu.Lock()
+	if a.closed {
+		a.mu.Unlock()
+		return
+	}
+	a.closed = true
+	workers := make([]*tunnelConn, 0, len(a.tunnels))
+	for _, w := range a.tunnels {
+		workers = append(workers, w)
+	}
+	close(a.done)
+	a.mu.Unlock()
+	for _, w := range workers {
+		w.stop()
+	}
+}
+
+func (a *Agent) isClosed() bool {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.closed
+}
+
+// Connected 指定隧道是否已连接其节点
+func (a *Agent) Connected(id string) bool {
+	a.mu.Lock()
+	w := a.tunnels[id]
+	a.mu.Unlock()
+	if w == nil {
+		return false
+	}
+	w.mu.Lock()
+	s := w.session
+	w.mu.Unlock()
+	return s != nil && !s.IsClosed()
+}
+
+// RegisterTunnel 注册隧道：同步等待首次"连接+登录+注册"完成并返回节点分配端口，
+// 之后由后台 worker 维持——永久隧道（TTLSec=0）断线自动重连重注册，
+// 临时隧道（TTLSec>0）一次性。同 ID 重复注册视为更新配置（停旧 worker 起新）。
+func (a *Agent) RegisterTunnel(t Tunnel) (int, error) {
+	if t.Id == "" {
+		return 0, errors.New("隧道ID不能为空")
+	}
+	if t.NodeAddr == "" || t.Token == "" {
+		return 0, fmt.Errorf("隧道缺少节点地址或令牌 id=%s", t.Id)
+	}
+	a.mu.Lock()
+	if a.closed {
+		a.mu.Unlock()
+		return 0, errors.New("agent 已关闭")
+	}
+	old := a.tunnels[t.Id]
+	w := newTunnelConn(a, t)
+	a.tunnels[t.Id] = w
+	a.mu.Unlock()
+	if old != nil {
+		old.stop()
+	}
+
+	go w.run()
+	select {
+	case <-w.first:
+		return w.firstPort, w.firstErr
+	case <-time.After(registerWait):
+		return 0, fmt.Errorf("注册超时(节点 %s)，永久隧道将在后台继续重试", t.NodeAddr)
+	}
+}
+
+// UnregisterTunnel 注销隧道：尽力经该隧道的控制流通知节点删除，随后停止本地 worker
+func (a *Agent) UnregisterTunnel(id string) error {
+	a.mu.Lock()
+	w := a.tunnels[id]
+	a.mu.Unlock()
+	if w == nil {
+		return errors.New("隧道不存在:" + id)
+	}
+	err := w.unregister()
+	w.stop()
+	a.removeTunnelState(id)
+	return err
+}
+
+// notifyPort 隧道注册成功（含重连后重注册）时回调
+func (a *Agent) notifyPort(id string, port int) {
+	a.mu.Lock()
+	cb := a.OnTunnelPort
+	a.mu.Unlock()
+	if cb != nil {
+		cb(id, port)
+	}
+}
+
+func (a *Agent) removeTunnelState(id string) {
+	a.mu.Lock()
+	delete(a.tunnels, id)
+	cb := a.OnTunnelPort
+	a.mu.Unlock()
+	if cb != nil {
+		cb(id, -1)
+	}
+}
+
+func errText(err error) string {
+	if err == nil {
+		return ""
+	}
+	return err.Error()
+}
@@ -0,0 +1,52 @@
+package agent
+
+import (
+	"io"
+	"net"
+)
+
+// handleData 处理节点主动打开的数据流：
+// 首帧 CmdTarget 携带目标信息 → 拨号本地 TCP 目标 → 回执 → 双向管道。
+func (a *Agent) handleData(stream net.Conn) {
+	defer stream.Close()
+	f, err := ReadFrame(stream)
+	if err != nil {
+		return
+	}
+	if f.Cmd() != CmdTarget {
+		a.logger().Warnf("agent 数据流首帧命令错误 cmd=%d", f.Cmd())
+		return
+	}
+	t := Tunnel{}
+	if err := f.Unmarshal(&t); err != nil {
+		return
+	}
+
+	conn, err := net.Dial("tcp", t.TargetAddr())
+	if err != nil {
+		a.logger().Warnf("agent 拨号本地目标失败 target=%s err=%s", t.TargetAddr(), err.Error())
+		rf := NewFrame(FrameVersion, CmdTarget, f.StreamID())
+		_ = rf.Marshal(Ret{Code: 1, Msg: err.Error()})
+		_ = WriteFrame(stream, rf)
+		return
+	}
+	defer conn.Close()
+
+	rf := NewFrame(FrameVersion, CmdTarget, f.StreamID())
+	_ = rf.Marshal(Ret{Code: 0, Msg: "ok"})
+	if err := WriteFrame(stream, rf); err != nil {
+		return
+	}
+
+	pipe(conn, stream)
+}
+
+// pipe 双向拷贝，任一方向结束即关闭两端
+func pipe(a, b net.Conn) {
+	go func() {
+		io.Copy(b, a)
+		b.Close()
+	}()
+	io.Copy(a, b)
+	a.Close()
+}
@@ -0,0 +1,149 @@
+// frame.go 帧协议，wire format 与 zonat 桥接帧完全一致
+// （ver(1) + cmd(1) + length(2, LE) + sid(4, LE) + data）。
+// 本文件取自 zonat client/internal/frame 的本地修复副本，仅裁掉未用方法。
+package agent
+
+import (
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+)
+
+const FrameVersion byte = 1
+
+// 帧头部
+const (
+	sizeOfVer    = 1
+	sizeOfCmd    = 1
+	sizeOfLength = 2
+	sizeOfSid    = 4
+	headerSize   = sizeOfVer + sizeOfCmd + sizeOfSid + sizeOfLength
+)
+
+const (
+	frameBufSize = 0x10000
+)
+
+type Frame struct {
+	ver  byte
+	cmd  byte
+	sid  uint32
+	data []byte
+}
+
+func NewFrame(version byte, cmd byte, sid uint32) Frame {
+	return Frame{ver: version, cmd: cmd, sid: sid}
+}
+
+// ReadFrame 读取一帧。头部与数据均用 ReadFull，防止单次 Read 短读导致解析错乱，
+// 也不会把后续帧的数据读进缓冲丢弃。
+func ReadFrame(conn net.Conn) (Frame, error) {
+	f := NewFrame(0, 0, 0)
+	hdr := RawHeader{}
+	if _, err := io.ReadFull(conn, hdr[:]); err != nil {
+		return f, err
+	}
+
+	f.ver = hdr.Version()
+	f.cmd = hdr.Cmd()
+	f.sid = hdr.StreamID()
+
+	length := int(hdr.Length())
+	if length == 0 {
+		return f, nil
+	}
+	data := make([]byte, length)
+	if _, err := io.ReadFull(conn, data); err != nil {
+		return f, err
+	}
+	f.data = data
+	return f, nil
+}
+
+func WriteFrame(conn net.Conn, frame Frame) error {
+	hdr := RawHeader{}
+	hdr[0] = frame.ver
+	hdr[1] = frame.cmd
+	binary.LittleEndian.PutUint16(hdr[2:], uint16(len(frame.data)))
+	binary.LittleEndian.PutUint32(hdr[4:], frame.sid)
+
+	buf := hdr.Bytes()
+	buf = append(buf, frame.data...)
+	_, err := conn.Write(buf)
+	return err
+}
+
+func (f *Frame) Version() byte {
+	return f.ver
+}
+
+func (f *Frame) Cmd() byte {
+	return f.cmd
+}
+
+func (f *Frame) Length() uint16 {
+	return uint16(len(f.data))
+}
+
+func (f *Frame) Data() []byte {
+	return f.data
+}
+
+func (f *Frame) StreamID() uint32 {
+	return f.sid
+}
+
+func (f *Frame) SetData(data []byte) error {
+	if len(data) > frameBufSize-1 {
+		return fmt.Errorf("超出最大长度:%d", frameBufSize-1)
+	}
+	f.data = data
+	return nil
+}
+
+func (f *Frame) Marshal(v any) error {
+	buf, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	return f.SetData(buf)
+}
+
+func (f *Frame) Unmarshal(v any) error {
+	if len(f.data) == 0 {
+		return errors.New("没有数据")
+	}
+	return json.Unmarshal(f.data, v)
+}
+
+func (f *Frame) String() string {
+	return fmt.Sprintf("Version:%d Cmd:%d StreamID:%d Length:%d",
+		f.Version(), f.Cmd(), f.StreamID(), f.Length())
+}
+
+type RawHeader [headerSize]byte
+
+func (h RawHeader) Version() byte {
+	return h[0]
+}
+
+func (h RawHeader) Cmd() byte {
+	return h[1]
+}
+
+func (h RawHeader) Length() uint16 {
+	return binary.LittleEndian.Uint16(h[2:])
+}
+
+func (h RawHeader) StreamID() uint32 {
+	return binary.LittleEndian.Uint32(h[4:])
+}
+
+func (h RawHeader) Bytes() []byte {
+	buf := make([]byte, headerSize)
+	copy(buf, h[:])
+	return buf
+}
@@ -0,0 +1,75 @@
+// protocol.go 定义 zonat 桥接协议的命令字与消息结构。
+// 命令字取值与 zonat 桥接命令字对齐（CmdLogin 同 CmdTunnelLogin、CmdTarget 同 CmdTunnelTarget），
+// 帧格式不变，便于将来与 zonat 互通或平滑迁移。
+package agent
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/xtaci/smux"
+)
+
+// 帧命令字
+const (
+	// CmdLogin 登录（请求与应答共用同一命令字，同 zonat）
+	CmdLogin byte = 0
+	// CmdPing 心跳（请求与应答共用同一命令字，同 zonat CmdTunnelPing）
+	CmdPing byte = 3
+	// CmdTarget 数据流首帧：节点告知 agent 本条公网连接的目标（同 zonat CmdTunnelTarget）
+	CmdTarget byte = 4
+	// CmdRegisterTunnel agent→节点 注册隧道（应答共用同一命令字，Ret.Port 为分配端口）
+	CmdRegisterTunnel byte = 10
+	// CmdUnregisterTunnel agent→节点 注销隧道（应答共用同一命令字）
+	CmdUnregisterTunnel byte = 11
+	// CmdTunnelClosed 节点→agent 推送：隧道已被节点删除（TTL 过期、会话顶替等）
+	CmdTunnelClosed byte = 12
+)
+
+// Version 协议/程序版本
+const Version = "0.1.0"
+
+// Login 登录请求
+type Login struct {
+	AgentId string `json:"agentId"`
+	Token   string `json:"token"` // JWT，由签发方（如 cloud）生成，节点用密钥校验
+	Version string `json:"version"`
+}
+
+// Ret 通用应答（zonat JsonRet 的裁剪版）
+type Ret struct {
+	Code int    `json:"code"`
+	Msg  string `json:"msg"`
+	Port int    `json:"port,omitempty"` // 注册隧道应答：节点分配的监听端口
+}
+
+// Tunnel 一条隧道的定义（zonat models.Tunnel 的裁剪版）。
+// NodeAddr/Token 是 zonat 的多节点扩展：每条隧道自带目标节点与登录令牌，
+// 由独立连接登录注册，因此一个 agent 的多条隧道可分布在多个不同节点。
+// 这两个字段只用于 agent→节点的登录，节点侧不使用也不回传（CmdTarget 不携带）。
+type Tunnel struct {
+	Id         string `json:"id"`
+	NodeAddr   string `json:"nodeAddr,omitempty"` // 该隧道登录的节点桥接地址
+	Token      string `json:"token,omitempty"`    // 登录令牌（JWT），仅登录帧使用
+	ListenPort int    `json:"listenPort"`         // 0=节点随机分配
+	TargetIp   string `json:"targetIp"`
+	TargetPort int    `json:"targetPort"`
+	TTLSec     int    `json:"ttlSec"` // 空闲自动删除秒数，0=永久
+}
+
+// TargetAddr 隧道在内网的目标地址
+func (t Tunnel) TargetAddr() string {
+	return fmt.Sprintf("%s:%d", t.TargetIp, t.TargetPort)
+}
+
+// SmuxConfig smux 会话参数，与 zonat 保持一致
+func SmuxConfig() *smux.Config {
+	return &smux.Config{
+		Version:           2,
+		KeepAliveInterval: 10 * time.Second,
+		KeepAliveTimeout:  30 * time.Second,
+		MaxFrameSize:      32768,
+		MaxReceiveBuffer:  4194304,
+		MaxStreamBuffer:   65536,
+	}
+}
@@ -0,0 +1,304 @@
+// tunnel.go 一条隧道的节点连接 worker：每条隧道独立完成
+// 连接→登录（携带自身 Token）→smux→控制流注册自身→数据流服务，
+// 因此天然支持不同隧道连接不同节点。
+// 永久隧道断线按 Backoff 重连重注册；临时隧道一次性退出。
+package agent
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/xtaci/smux"
+)
+
+const (
+	loginTimeout = 10 * time.Second // 登录握手超时（含拨号）
+	registerWait = 30 * time.Second // RegisterTunnel 等待首次注册完成的上限
+)
+
+// errTunnelClosed 节点推送 CmdTunnelClosed：隧道已被节点删除（TTL 过期等）
+var errTunnelClosed = errors.New("隧道已被节点关闭")
+
+// tunnelConn 一条隧道的全部运行态：期望配置 + 到该隧道节点的桥接连接
+type tunnelConn struct {
+	ag *Agent
+	t   Tunnel
+
+	stopCh   chan struct{}
+	stopOnce sync.Once
+
+	firstOnce sync.Once
+	first     chan struct{} // 首次"连接+登录+注册"尝试结束后关闭
+	firstPort int
+	firstErr  error
+
+	everRegistered bool // 本次注册成功过至少一次（run 协程独占读写）
+
+	mu      sync.Mutex
+	conn    net.Conn
+	session *smux.Session
+	control net.Conn
+
+	ctrlMu sync.Mutex // 控制流"写请求+读应答"串行化
+}
+
+func newTunnelConn(ag *Agent, t Tunnel) *tunnelConn {
+	return &tunnelConn{
+		ag:     ag,
+		t:      t,
+		stopCh: make(chan struct{}),
+		first:  make(chan struct{}),
+	}
+}
+
+// markFirst 记录首次注册尝试的结果；channel close 保证读取方看到字段写入
+func (w *tunnelConn) markFirst(port int, err error) {
+	w.firstOnce.Do(func() {
+		w.firstPort, w.firstErr = port, err
+		close(w.first)
+	})
+}
+
+// run 重连主循环，阻塞直到停止/一次性结束
+func (w *tunnelConn) run() {
+	for {
+		port, err := w.connect()
+		w.markFirst(port, err)
+		if err == nil {
+			err = w.serve()
+		}
+		w.teardown()
+		if w.isStopped() || w.ag.isClosed() {
+			return
+		}
+		if errors.Is(err, errTunnelClosed) {
+			w.ag.removeTunnelState(w.t.Id)
+			return
+		}
+		if w.t.TTLSec > 0 {
+			// 临时隧道一次性：连接断开即结束（节点侧随会话移除）
+			if w.everRegistered {
+				w.ag.removeTunnelState(w.t.Id)
+			}
+			return
+		}
+		w.ag.logger().Warnf("agent 隧道连接断开 id=%s node=%s err=%s backoff=%s",
+			w.t.Id, w.t.NodeAddr, errText(err), w.ag.Backoff)
+		select {
+		case <-time.After(w.ag.Backoff):
+		case <-w.stopCh:
+			return
+		}
+	}
+}
+
+// connect 连接并登录该隧道的节点，经控制流注册自身，返回节点分配端口
+func (w *tunnelConn) connect() (int, error) {
+	conn, err := net.DialTimeout("tcp", w.t.NodeAddr, loginTimeout)
+	if err != nil {
+		return 0, err
+	}
+
+	// 登录（同 zonat：CmdLogin 帧 + Ret 应答）
+	lf := NewFrame(FrameVersion, CmdLogin, 0)
+	if err := lf.Marshal(Login{AgentId: w.ag.AgentId, Token: w.t.Token, Version: Version}); err != nil {
+		conn.Close()
+		return 0, err
+	}
+	if err := WriteFrame(conn, lf); err != nil {
+		conn.Close()
+		return 0, err
+	}
+	conn.SetReadDeadline(time.Now().Add(loginTimeout))
+	rf, err := ReadFrame(conn)
+	if err != nil {
+		conn.Close()
+		return 0, err
+	}
+	ret := Ret{}
+	if err := rf.Unmarshal(&ret); err != nil {
+		conn.Close()
+		return 0, err
+	}
+	if ret.Code != 0 {
+		conn.Close()
+		return 0, fmt.Errorf("登录失败:%s", ret.Msg)
+	}
+	conn.SetReadDeadline(time.Time{})
+
+	sess, err := smux.Client(conn, SmuxConfig())
+	if err != nil {
+		conn.Close()
+		return 0, err
+	}
+	control, err := sess.OpenStream()
+	if err != nil {
+		sess.Close()
+		return 0, err
+	}
+
+	w.mu.Lock()
+	w.conn, w.session, w.control = conn, sess, control
+	w.mu.Unlock()
+
+	// 注册自身；凭据只用于登录，不随注册/数据帧回发
+	wire := w.t
+	wire.NodeAddr, wire.Token = "", ""
+	f, err := w.callControl(func(control net.Conn) error {
+		rf := NewFrame(FrameVersion, CmdRegisterTunnel, 0)
+		if err := rf.Marshal(wire); err != nil {
+			return err
+		}
+		return WriteFrame(control, rf)
+	}, CmdRegisterTunnel)
+	if err != nil {
+		return 0, err
+	}
+	rret := Ret{}
+	if err := f.Unmarshal(&rret); err != nil {
+		return 0, err
+	}
+	if rret.Code != 0 {
+		return 0, errors.New(rret.Msg)
+	}
+	w.everRegistered = true
+	w.ag.notifyPort(w.t.Id, rret.Port)
+	w.ag.logger().Infof("agent 隧道已连接节点 id=%s node=%s listen=%d",
+		w.t.Id, w.t.NodeAddr, rret.Port)
+	if w.ag.PingEvery > 0 {
+		go w.pingLoop()
+	}
+	return rret.Port, nil
+}
+
+// serve 数据流循环（阻塞；出错即断线，由 run 重连）
+func (w *tunnelConn) serve() error {
+	w.mu.Lock()
+	sess := w.session
+	w.mu.Unlock()
+	if sess == nil {
+		return errors.New("未连接节点")
+	}
+	for {
+		stream, err := sess.AcceptStream()
+		if err != nil {
+			return err
+		}
+		go w.ag.handleData(stream)
+	}
+}
+
+func (w *tunnelConn) pingLoop() {
+	ticker := time.NewTicker(w.ag.PingEvery)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-w.stopCh:
+			return
+		case <-ticker.C:
+		}
+		_, err := w.callControl(func(control net.Conn) error {
+			return WriteFrame(control, NewFrame(FrameVersion, CmdPing, 0))
+		}, CmdPing)
+		if err != nil {
+			if errors.Is(err, errTunnelClosed) {
+				// 节点已删除本隧道：杀掉会话让 serve 退出，run 收尾
+				w.teardown()
+			}
+			return
+		}
+	}
+}
+
+// callControl 串行化控制流交互。等待应答期间收到本隧道的关闭推送即返回 errTunnelClosed。
+func (w *tunnelConn) callControl(write func(control net.Conn) error, wantCmd byte) (Frame, error) {
+	w.ctrlMu.Lock()
+	defer w.ctrlMu.Unlock()
+
+	w.mu.Lock()
+	control := w.control
+	w.mu.Unlock()
+	if control == nil {
+		return Frame{}, errors.New("未连接节点")
+	}
+	if err := write(control); err != nil {
+		return Frame{}, err
+	}
+	for {
+		f, err := ReadFrame(control)
+		if err != nil {
+			return Frame{}, err
+		}
+		if f.Cmd() == CmdTunnelClosed {
+			t := Tunnel{}
+			_ = f.Unmarshal(&t)
+			if t.Id == w.t.Id {
+				return Frame{}, errTunnelClosed
+			}
+			continue
+		}
+		if f.Cmd() == wantCmd {
+			return f, nil
+		}
+		// 其它命令忽略
+	}
+}
+
+// unregister 经控制流注销本隧道。传输失败视为成功：
+// 连接已断时节点会随会话移除隧道，无需再通知。
+func (w *tunnelConn) unregister() error {
+	w.mu.Lock()
+	connected := w.control != nil
+	w.mu.Unlock()
+	if !connected {
+		return nil
+	}
+	f, err := w.callControl(func(control net.Conn) error {
+		rf := NewFrame(FrameVersion, CmdUnregisterTunnel, 0)
+		if err := rf.Marshal(Tunnel{Id: w.t.Id}); err != nil {
+			return err
+		}
+		return WriteFrame(control, rf)
+	}, CmdUnregisterTunnel)
+	if err != nil {
+		return nil
+	}
+	ret := Ret{}
+	if err := f.Unmarshal(&ret); err != nil {
+		return nil
+	}
+	if ret.Code != 0 {
+		return errors.New(ret.Msg)
+	}
+	return nil
+}
+
+func (w *tunnelConn) stop() {
+	w.stopOnce.Do(func() { close(w.stopCh) })
+	w.teardown()
+}
+
+func (w *tunnelConn) isStopped() bool {
+	select {
+	case <-w.stopCh:
+		return true
+	default:
+		return false
+	}
+}
+
+func (w *tunnelConn) teardown() {
+	w.mu.Lock()
+	sess, conn := w.session, w.conn
+	w.session, w.conn, w.control = nil, nil, nil
+	w.mu.Unlock()
+	if sess != nil {
+		sess.Close()
+	}
+	if conn != nil {
+		conn.Close()
+	}
+}