From 984041f1a058ee70f8708908c0c398cd51369d4b Mon Sep 17 00:00:00 2001 From: W11 Date: Sat, 19 Sep 2026 18:22:51 +0800 Subject: [PATCH] =?UTF-8?q?init:=20=E8=87=AA=20zomaintain/backend/zonat=20?= =?UTF-8?q?=E5=B9=B3=E7=A7=BB=E7=8B=AC=E7=AB=8B=E6=88=90=E5=BA=93;=20modul?= =?UTF-8?q?e=20path=20=E6=94=B9=E4=B8=BA=20git.zeroonesoft.cn/golib/zonat;?= =?UTF-8?q?=20=E5=86=85=E9=83=A8=20import=20=E5=90=8C=E6=AD=A5=E9=87=8D?= =?UTF-8?q?=E5=86=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + README.md | 176 ++++++++++++++++++ agent/agent.go | 222 ++++++++++++++++++++++ agent/forward.go | 52 ++++++ agent/frame.go | 149 +++++++++++++++ agent/protocol.go | 75 ++++++++ agent/tunnel.go | 304 ++++++++++++++++++++++++++++++ cmd/agent/main.go | 139 ++++++++++++++ cmd/http-demo/main.go | 25 +++ cmd/node/main.go | 62 +++++++ cmd/tcp-echo/main.go | 30 +++ e2e/http_test.go | 110 +++++++++++ e2e/logger_test.go | 69 +++++++ e2e/tcp_test.go | 391 +++++++++++++++++++++++++++++++++++++++ go.mod | 7 + go.sum | 4 + internal/node/jwt.go | 60 ++++++ internal/node/node.go | 307 ++++++++++++++++++++++++++++++ internal/node/session.go | 137 ++++++++++++++ internal/node/tunnel.go | 107 +++++++++++ 20 files changed, 2427 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 agent/agent.go create mode 100644 agent/forward.go create mode 100644 agent/frame.go create mode 100644 agent/protocol.go create mode 100644 agent/tunnel.go create mode 100644 cmd/agent/main.go create mode 100644 cmd/http-demo/main.go create mode 100644 cmd/node/main.go create mode 100644 cmd/tcp-echo/main.go create mode 100644 e2e/http_test.go create mode 100644 e2e/logger_test.go create mode 100644 e2e/tcp_test.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/node/jwt.go create mode 100644 internal/node/node.go create mode 100644 internal/node/session.go create mode 100644 internal/node/tunnel.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e660fd9 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +bin/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..502ffe3 --- /dev/null +++ b/README.md @@ -0,0 +1,176 @@ +# zonat — 内网穿透轻量库 + +从完整版内网穿透项目（client + server + cloud 三端约 1.4 万行）裁剪出的轻量实现， +只服务远程桌面场景（VNC / RDP / webshell / 文件）。背景与裁剪依据见 +capricorn 仓库 `docs/remote-desktop-nat-analysis.md` 第八节。 + +**在 zomaintain 中的定位**：`backend/zonat` 是 backend/ 下的**共享库 module** +（module 名 `zonat`，非业务端）——被控场所侧宿主（网吧服务器 BS，即 `backend/server`） +内嵌 `agent/` 包建立穿透隧道；穿透节点用 `cmd/node` 独立部署在公网/机房。 +业务端之间仍是「零 import」，zonat 是唯一例外：桥接协议必须单一来源，防止两端漂移。 + +**打包约定**： + +- **单 Go module**（node + agent + wire 全在一起，单一来源，方便维护）：端到端测试可 + 进程内直接跑通；后续如需按端拆分再拆。 +- **宿主用 import 引用 agent（推荐）**：`agent/` 是公开包且自包含（5 个文件，仅依赖标准库 + + smux），宿主 go.mod `require zonat` + `replace zonat => ../zonat` 即可；构建加 `-trimpath` + 后，agent 日志路径稳定为 `zonat@版本/agent/agent.go` 模块形式（已实测，见下）。 +- **整目录拷贝为备用方案**：不依赖模块引用的场景（如临时放进某个项目调试）直接拷 `agent/` + 目录，日志走宿主项目内路径，同样零修改。 +- **日志注入**：agent 包不依赖任何具体日志实现，宿主注入 4 方法 `Logger` 接口 + （logrus 原生满足，slog 写几行适配器），agent 日志即归入宿主的格式/级别/输出。 + 唯一例外 `DefaultLogger` 是仅标准库的兜底（stderr，Info 级）。 + +## 架构 + +``` +控制面：cloud ──WS──► 场所服务器（内嵌 agent 库）──桥接帧命令注册隧道──► 节点（纯内存，无DB/无管理API） +数据面：浏览器 ──► 节点:隧道端口 ──smux(TCP桥接)──► agent ──LAN TCP──► VNC/RDP/3000 +``` + +**多节点模型**：每条 Tunnel 自带 `NodeAddr` + `Token`，由独立连接完成 +"登录 → smux → 控制流注册自身 → 数据流服务"；Agent 只是隧道注册表，不持有网络连接。 +因此一个 agent 的多条隧道可以分布在多个不同节点，各隧道的连接/重连/心跳互不影响。 + +相比完整版裁掉：KCP 桥接、UDP 隧道、http(s) 反向代理模式、协议嗅探与兜底站、 +管理 REST、数据库、P2P、限流统计、zonat cloud 整端。 +保留：帧协议（wire format 与完整版一致）、smux 多路复用（恒开，不再有开关）、 +TCP 桥接、心跳重连、TTL 空闲自动删除（对应完整版 AutoDelete）。 + +## 协议 + +- 帧格式：`ver(1) + cmd(1) + length(2,LE) + sid(4,LE) + data(JSON)`，与完整版桥接帧一致。 +- 命令字与完整版对齐：`CmdLogin=0`、`CmdPing=3`、`CmdTarget=4`（同 CmdTunnelLogin/Ping/Target）， + 新增 `CmdRegisterTunnel=10`、`CmdUnregisterTunnel=11`、`CmdTunnelClosed=12`（推送）。 +- **一条隧道一条桥接连接**（TCP + smux）：连接建立时先用该隧道的 Token 登录，再经控制流 + 注册自身。数据流由节点主动 OpenStream，首帧 CmdTarget 告知目标，agent 拨号本地 TCP + 目标后回执，之后纯字节管道（**无协议嗅探**）。 +- 隧道生命周期由 agent 注册：`port=0` 节点随机分配；`ttl>0` 空闲超时由节点清扫并推送 + CmdTunnelClosed；永久隧道（ttl=0）断线后各自自动重连重注册。 +- 同一 agentId 在一个节点上可有多个并发会话（每隧道一个）；同 ID 隧道重复注册按 + "隧道 ID + agentId" 粒度顶替（同 agent 更新配置，跨 agent 活隧道拒绝）。 +- 鉴权：登录帧带 **JWT**（HS256 系，节点用 `-jwt-secret` 密钥校验）。token 必须带 `exp`； + 若带 `agentId`/`sub` claim 则必须与登录 AgentId 一致。正式部署由 cloud 签发下发， + 每条隧道各自的 token 可不同（含不同有效期/不同节点密钥）。 + +## 目录 + +``` +zonat/ +├── agent/ 可拷贝分发：agent.go + tunnel.go + forward.go + frame.go + protocol.go（自包含，仅依赖标准库+smux） +├── cmd/node/ 节点入口（独立部署二进制；含 -print-token 手工签发） +├── cmd/agent/ 被控端独立二进制（测试/独立跑用；slog 适配器示例） +├── cmd/tcp-echo/ 测试目标：TCP 回显（模拟 VNC/RDP/3000） +├── cmd/http-demo/ 测试目标：HTTP 服务（webshell/文件场景） +├── internal/node/ 节点：JWT 登录 / 桥接会话 / 隧道注册表 / 纯 TCP 隧道监听 / TTL 清扫 +└── e2e/ 端到端测试（进程内，不依赖外部环境） +``` + +node 与 agent 共用同一份 wire 实现：节点侧 `import "git.zeroonesoft.cn/golib/zonat/agent"` 复用 +Frame/Tunnel/命令字（agent 包是协议的唯一来源，节点不另留副本，防止两端漂移）。 + +## 宿主接入 agent（如 backend/server） + +### 方式一：import 引用（推荐） + +宿主 go.mod（宿主在 `backend/<端>` 下，与本库平级，用相对路径 replace）： + +``` +require zonat v0.0.0 +replace zonat => ../zonat // 例：backend/server、backend/user +``` + +宿主代码： + +```go +import "git.zeroonesoft.cn/golib/zonat/agent" + +ag := agent.New(agentId) // 节点地址与令牌随各隧道下发 +ag.Logger = logrus.StandardLogger() // logrus 四个方法签名天然满足接口；不注入也有兜底 +ag.OnTunnelPort = func(id string, port int) { /* 保存/上报分配到的端口 */ } +go ag.Run() +// 每条隧道自带节点与 JWT，可指向不同节点 +port, _ := ag.RegisterTunnel(agent.Tunnel{ + Id: "vnc-1", NodeAddr: "node1.example.com:5212", Token: jwtFromCloud, + TargetIp: "127.0.0.1", TargetPort: 5900, TTLSec: 300, +}) +port2, _ := ag.RegisterTunnel(agent.Tunnel{ + Id: "web-1", NodeAddr: "node2.example.com:5212", Token: anotherJwt, + TargetIp: "127.0.0.1", TargetPort: 3000, +}) +``` + +`RegisterTunnel` 同步等待首次"连接+登录+注册"完成并返回节点分配的端口；之后由后台 +worker 维持：永久隧道（TTLSec=0）断线自动重连重注册，临时隧道（TTLSec>0）一次性。 + +**宿主构建必须加 `-trimpath`**，否则 agent 日志记的是构建机全路径 +（`D:/golib/zonat/agent/agent.go`）。实测对比（模拟宿主 Logger 记录调用点）： + +``` +不加 -trimpath： WARN D:/golib/zonat/agent/agent.go:119 +加 -trimpath： WARN zonat@v0.0.0/agent/agent.go:119 ← 模块名+包相对路径，跨机器稳定 +``` + +`@v0.0.0` 来自宿主 require 的版本号（tag 后即 `@v1.0.0`），Logger 里可顺手替换掉。 +**注意**：加了 `-trimpath` 后宿主自身日志路径也会变成 `server/internal/...` 模块形式， +且这两类路径都不含宿主 rootDir——宿主 Logger 若做过"削 rootDir、兜底取文件名"式处理， +需同步升级为**保留模块形式路径**（else 分支一行改动），否则退化成 `agent.go:119`。 + +### 方式二：整目录拷贝（备用） + +```bash +cp -r /d/zomaintain/backend/zonat/agent /d/zomaintain/backend/server/pkg/agent +cd /d/zomaintain/backend/server && go mod tidy # 只为拉一次 xtaci/smux +``` + +import 改为 `server/pkg/agent`。源码在宿主项目内，`runtime.Caller` 记录宿主内路径， +宿主现有 rootDir 削前缀**原样生效**（得到 `pkg/agent/agent.go:88`），无需 `-trimpath`。 +已用临时 server 模块验证零修改编译。升级纪律：从 `backend/zonat` 重新拷贝整目录， +不在副本里改协议。 + +## 节点部署（多台） + +```bash +# Windows 交叉编译 Linux 节点 +GOOS=linux GOARCH=amd64 go build -o bin/node-linux-amd64 ./cmd/node +# 本机 +go build -o bin/node.exe ./cmd/node +# 各节点：./node-linux-amd64 -addr :5212 -jwt-secret <各节点不同密钥> -tunnel-bind 0.0.0.0 +# 手工签发一张登录 token（正式部署由 cloud 签发）： +./node-linux-amd64 -jwt-secret <密钥> -print-token -agent -ttl 24h +``` + +## 验证 + +```bash +cd /d/zomaintain/backend/zonat +go test ./e2e/ -v # 13 个用例：TCP 回显（并发/大包/二进制安全）、HTTP GET/POST/keep-alive、 + # 单agent双隧道连两节点、注销、TTL 过期、ID 冲突（同agent顶替/跨agent拒绝）、 + # JWT 拒绝（错误签名/过期/agentId不一致）、踢会话重连重注册、日志注入 +go build ./... +``` + +手动验证： + +```bash +go build -o bin/ ./cmd/... +./bin/tcp-echo -addr 127.0.0.1:9000 & +./bin/node -addr :5212 -jwt-secret s3cret -tunnel-bind 127.0.0.1 & +TOK=$(./bin/node -jwt-secret s3cret -print-token -agent a1 -ttl 1h) +./bin/agent -node 127.0.0.1:5212 -token "$TOK" -id a1 -tunnel "id=demo,target=127.0.0.1:9000" +# agent 日志输出 "隧道端口 id=demo listen=" 后： +nc 127.0.0.1 # 输入任意内容，回显即穿透成功 +``` + +## 已知简化（P0 范围） + +- 无 TLS：节点 ↔ agent 明文（内网/机房部署假设）；JWT 解决的是"静态共享秘密 + 无过期"， + 不解决信道窃听——明文信道上 token 在 exp 前仍可被截获重放。如需加密后续加 smux 前置 TLS。 +- 每隧道一条连接：同一节点上的多条隧道不共享桥接连接（简单优先；如需收敛连接数， + 后续可在 agent 内按 NodeAddr 池化）。 +- 无限流/流量统计；连接数仅内存计数。 +- 临时隧道 TTL 过期由节点清扫推送，agent 不重试（与完整版临时通道语义一致：由上层重新申请）。 +- HTTP 隧道场景（webshell）验证走纯 tcp 模式；完整版的 http 反代模式不再实现。 +- 拷贝式分发下各宿主副本可能随时间漂移：纪律是升级一律从 `backend/zonat` 重新拷贝整目录， + 不在副本里改协议逻辑（与仓库"拷贝下沉"惯例一致，取舍已明确）。 diff --git a/agent/agent.go b/agent/agent.go new file mode 100644 index 0000000..d1d2be4 --- /dev/null +++ b/agent/agent.go @@ -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() +} diff --git a/agent/forward.go b/agent/forward.go new file mode 100644 index 0000000..7ca43bd --- /dev/null +++ b/agent/forward.go @@ -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() +} diff --git a/agent/frame.go b/agent/frame.go new file mode 100644 index 0000000..d630c1d --- /dev/null +++ b/agent/frame.go @@ -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 +} diff --git a/agent/protocol.go b/agent/protocol.go new file mode 100644 index 0000000..a4fad2c --- /dev/null +++ b/agent/protocol.go @@ -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, + } +} diff --git a/agent/tunnel.go b/agent/tunnel.go new file mode 100644 index 0000000..33a7809 --- /dev/null +++ b/agent/tunnel.go @@ -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() + } +} diff --git a/cmd/agent/main.go b/cmd/agent/main.go new file mode 100644 index 0000000..1fb668a --- /dev/null +++ b/cmd/agent/main.go @@ -0,0 +1,139 @@ +// zonat 被控端入口：跑在内网机器上，按隧道定义登录各自节点并注册。 +// 每条隧道自带节点地址与登录令牌（JWT），可分布在多个不同节点。 +// 隧道通过 -tunnel 参数下发（后续由 capricorn cloud 经 WS 下发，见分析文档 P1）： +// +// -tunnel id=web1,target=127.0.0.1:3000,node=10.0.0.2:5212,token=,port=0,ttl=0 +// -tunnel id=vnc1,target=127.0.0.1:5900,ttl=300 // node/token 取 -node/-token 缺省值 +package main + +import ( + "flag" + "fmt" + "log/slog" + "net" + "os" + "strconv" + "strings" + "time" + + "git.zeroonesoft.cn/golib/zonat/agent" +) + +// slogLogger 把 slog 适配为 agent.Logger 接口（独立二进制用； +// 宿主项目若是 logrus 可直接注入 logrus.StandardLogger()） +type slogLogger struct{ l *slog.Logger } + +func (s slogLogger) Debugf(format string, args ...any) { s.l.Debug(fmt.Sprintf(format, args...)) } +func (s slogLogger) Infof(format string, args ...any) { s.l.Info(fmt.Sprintf(format, args...)) } +func (s slogLogger) Warnf(format string, args ...any) { s.l.Warn(fmt.Sprintf(format, args...)) } +func (s slogLogger) Errorf(format string, args ...any) { s.l.Error(fmt.Sprintf(format, args...)) } + +func main() { + nodeAddr := flag.String("node", "127.0.0.1:5212", "缺省节点桥接地址（隧道未带 node= 时使用）") + token := flag.String("token", "", "缺省登录令牌/JWT（隧道未带 token= 时使用）") + id := flag.String("id", "", "agent 唯一标识（默认取主机名）") + backoff := flag.Duration("backoff", 5*time.Second, "永久隧道重连退避") + ping := flag.Duration("ping", 30*time.Second, "控制流心跳间隔") + var tunnels sliceFlag + flag.Var(&tunnels, "tunnel", "隧道定义 id=..,target=ip:port[,node=addr][,token=jwt][,port=0][,ttl=0]，可重复") + flag.Parse() + + if *id == "" { + host, err := os.Hostname() + if err != nil { + slog.Error("获取主机名失败", "err", err.Error()) + os.Exit(1) + } + *id = host + } + + a := agent.New(*id) + a.Logger = slogLogger{slog.Default()} + a.Backoff = *backoff + a.PingEvery = *ping + a.OnTunnelPort = func(tunnelId string, port int) { + if port > 0 { + slog.Info("隧道端口", "id", tunnelId, "listen", port) + } + } + for _, spec := range tunnels { + t, err := parseTunnel(spec, *nodeAddr, *token) + if err != nil { + slog.Error("隧道参数错误", "spec", spec, "err", err.Error()) + os.Exit(1) + } + if _, err := a.RegisterTunnel(t); err != nil { + slog.Warn("注册隧道失败（永久隧道会在后台持续重试）", "id", t.Id, "err", err.Error()) + } + } + + if err := a.Run(); err != nil { + slog.Error("agent 退出", "err", err.Error()) + os.Exit(1) + } +} + +// parseTunnel 解析一条隧道定义，node/token 未带时回落到全局缺省值 +func parseTunnel(spec, defNode, defToken string) (agent.Tunnel, error) { + t := agent.Tunnel{NodeAddr: defNode, Token: defToken} + for _, kv := range strings.Split(spec, ",") { + parts := strings.SplitN(kv, "=", 2) + if len(parts) != 2 { + return t, errBad("键值对缺少=:" + kv) + } + switch parts[0] { + case "id": + t.Id = parts[1] + case "node": + t.NodeAddr = parts[1] + case "token": + t.Token = parts[1] + case "target": + host, portStr, err := net.SplitHostPort(parts[1]) + if err != nil { + return t, err + } + port, err := strconv.Atoi(portStr) + if err != nil { + return t, err + } + t.TargetIp, t.TargetPort = host, port + case "port": + v, err := strconv.Atoi(parts[1]) + if err != nil { + return t, err + } + t.ListenPort = v + case "ttl": + v, err := strconv.Atoi(parts[1]) + if err != nil { + return t, err + } + t.TTLSec = v + default: + return t, errBad("未知键:" + parts[0]) + } + } + if t.Id == "" || t.TargetIp == "" || t.TargetPort == 0 { + return t, errBad("缺少 id 或 target") + } + if t.NodeAddr == "" || t.Token == "" { + return t, errBad("缺少 node 或 token（登录令牌/JWT）") + } + return t, nil +} + +func errBad(msg string) error { return &badSpec{msg} } + +type badSpec struct{ msg string } + +func (e *badSpec) Error() string { return e.msg } + +type sliceFlag []string + +func (f *sliceFlag) String() string { return strings.Join(*f, " ") } + +func (f *sliceFlag) Set(v string) error { + *f = append(*f, v) + return nil +} diff --git a/cmd/http-demo/main.go b/cmd/http-demo/main.go new file mode 100644 index 0000000..9bb08e5 --- /dev/null +++ b/cmd/http-demo/main.go @@ -0,0 +1,25 @@ +// 测试目标：HTTP 服务，验证纯 TCP 隧道承载 HTTP（webshell/文件服务场景）。 +package main + +import ( + "flag" + "io" + "log" + "net/http" +) + +func main() { + addr := flag.String("addr", "127.0.0.1:9001", "监听地址") + flag.Parse() + + mux := http.NewServeMux() + mux.HandleFunc("/hello", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"hello":"git.zeroonesoft.cn/golib/zonat","via":"tcp-tunnel"}`)) + }) + mux.HandleFunc("/echo", func(w http.ResponseWriter, r *http.Request) { + io.Copy(w, r.Body) + }) + log.Println("http-demo 监听", *addr) + log.Fatal(http.ListenAndServe(*addr, mux)) +} diff --git a/cmd/node/main.go b/cmd/node/main.go new file mode 100644 index 0000000..02e387a --- /dev/null +++ b/cmd/node/main.go @@ -0,0 +1,62 @@ +// zonat 节点入口：TCP 桥接监听 + 隧道注册表 + 纯 TCP 隧道监听。 +// 无管理 API、无数据库；隧道由 agent 经桥接控制流注册。 +// agent 登录用 JWT 校验（HS256）；token 正式部署由 cloud 签发， +// 手工/测试场景可用 -print-token 现场签发。 +package main + +import ( + "flag" + "fmt" + "log/slog" + "os" + "os/signal" + "syscall" + "time" + + "git.zeroonesoft.cn/golib/zonat/internal/node" +) + +func main() { + addr := flag.String("addr", ":5212", "桥接监听地址") + tunnelBind := flag.String("tunnel-bind", "0.0.0.0", "隧道监听绑定地址") + jwtSecret := flag.String("jwt-secret", "", "agent 登录 JWT 校验密钥（HS256）") + sweep := flag.Duration("sweep", 30*time.Second, "TTL 清扫间隔") + // 便捷签发：node -jwt-secret s3cret -print-token -agent a1 -ttl 24h + printToken := flag.Bool("print-token", false, "用 -jwt-secret 签发一张 agent 登录 token 后退出") + tokenAgent := flag.String("agent", "", "print-token 时写入 agentId claim") + tokenTTL := flag.Duration("ttl", 24*time.Hour, "print-token 签发的有效期") + flag.Parse() + + if *printToken { + if *jwtSecret == "" || *tokenTTL <= 0 { + slog.Error("print-token 需要 -jwt-secret 与正的 -ttl") + os.Exit(1) + } + tok, err := node.SignToken(*jwtSecret, *tokenAgent, *tokenTTL) + if err != nil { + slog.Error("签发失败", "err", err.Error()) + os.Exit(1) + } + fmt.Println(tok) + return + } + + if *jwtSecret == "" { + slog.Error("缺少 -jwt-secret（agent 登录 JWT 校验密钥）") + os.Exit(1) + } + + n := node.New() + n.JwtSecret = *jwtSecret + n.BindTunnel = *tunnelBind + n.SweepInterval = *sweep + if err := n.Start(*addr); err != nil { + slog.Error("节点启动失败", "err", err.Error()) + os.Exit(1) + } + + quit := make(chan os.Signal, 1) + signal.Notify(quit, os.Interrupt, syscall.SIGTERM) + <-quit + n.Stop() +} diff --git a/cmd/tcp-echo/main.go b/cmd/tcp-echo/main.go new file mode 100644 index 0000000..799500e --- /dev/null +++ b/cmd/tcp-echo/main.go @@ -0,0 +1,30 @@ +// 测试目标：TCP 回显服务，模拟 VNC/RDP/3000 等内网 TCP 目标。 +package main + +import ( + "flag" + "io" + "log" + "net" +) + +func main() { + addr := flag.String("addr", "127.0.0.1:9000", "监听地址") + flag.Parse() + + l, err := net.Listen("tcp", *addr) + if err != nil { + log.Fatal(err) + } + log.Println("tcp-echo 监听", l.Addr().String()) + for { + conn, err := l.Accept() + if err != nil { + log.Fatal(err) + } + go func(c net.Conn) { + defer c.Close() + io.Copy(c, c) + }(conn) + } +} diff --git a/e2e/http_test.go b/e2e/http_test.go new file mode 100644 index 0000000..bf3849f --- /dev/null +++ b/e2e/http_test.go @@ -0,0 +1,110 @@ +// HTTP 穿透测试：验证"隧道模式全用 tcp"成立——真实 HTTP 服务（含 keep-alive、 +// 大 body）经纯 TCP 隧道完整可用，无需 zonat 的 http 反向代理模式。 +package e2e + +import ( + "bytes" + "crypto/rand" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + "git.zeroonesoft.cn/golib/zonat/agent" + "git.zeroonesoft.cn/golib/zonat/internal/node" +) + +func startHTTPServer(t *testing.T) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("/hello", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]string{"hello": "zonat", "via": "tcp-tunnel"}) + }) + mux.HandleFunc("/echo", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "POST only", http.StatusMethodNotAllowed) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.Write(body) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv +} + +// registerHTTP 把 httptest.Server 登记为一条隧道，返回公网端口 +func registerHTTP(t *testing.T, n *node.Node, a *agent.Agent, srv *httptest.Server, id string) int { + t.Helper() + port, err := a.RegisterTunnel(buildTunnel(t, n, a, id, srv.Listener.Addr().String(), 0)) + if err != nil { + t.Fatal(err) + } + return port +} + +func TestHTTPGetPostThroughTunnel(t *testing.T) { + srv := startHTTPServer(t) + n, a := startEnv(t) + port := registerHTTP(t, n, a, srv, "http-tunnel") + base := "http://127.0.0.1:" + strconv.Itoa(port) + + // GET + resp, err := http.Get(base + "/hello") + if err != nil { + t.Fatal(err) + } + var body map[string]string + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK || body["hello"] != "zonat" { + t.Fatalf("GET /hello 异常: status=%d body=%v", resp.StatusCode, body) + } + + // POST 大 body（1MB，跨 smux 多帧） + payload := make([]byte, 1<<20) + if _, err := rand.Read(payload); err != nil { + t.Fatal(err) + } + resp2, err := http.Post(base+"/echo", "application/octet-stream", bytes.NewReader(payload)) + if err != nil { + t.Fatal(err) + } + got, err := io.ReadAll(resp2.Body) + resp2.Body.Close() + if err != nil { + t.Fatal(err) + } + if resp2.StatusCode != http.StatusOK || !bytes.Equal(got, payload) { + t.Fatalf("POST /echo 异常: status=%d len(got)=%d", resp2.StatusCode, len(got)) + } +} + +func TestHTTPKeepAliveThroughTunnel(t *testing.T) { + srv := startHTTPServer(t) + n, a := startEnv(t) + port := registerHTTP(t, n, a, srv, "http-ka") + + // 同一 client 复用 TCP 连接（keep-alive），10 次请求走同一条隧道流 + client := &http.Client{Timeout: 10 * time.Second} + for i := 0; i < 10; i++ { + resp, err := client.Get("http://127.0.0.1:" + strconv.Itoa(port) + "/hello") + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("第 %d 次请求异常: %d", i+1, resp.StatusCode) + } + } +} diff --git a/e2e/logger_test.go b/e2e/logger_test.go new file mode 100644 index 0000000..251e97c --- /dev/null +++ b/e2e/logger_test.go @@ -0,0 +1,69 @@ +// 日志注入测试：验证 agent 包可携带宿主项目的日志实现， +// 日志统一走注入的 Logger（嵌入到 capricorn 等项目时日志不再"各写各的"）。 +package e2e + +import ( + "fmt" + "strings" + "sync" + "testing" + "time" + + "git.zeroonesoft.cn/golib/zonat/agent" +) + +// captureLogger 捕获日志的假实现（模拟宿主项目的 logger） +type captureLogger struct { + mu sync.Mutex + lines []string +} + +func (c *captureLogger) Debugf(format string, args ...any) {} + +func (c *captureLogger) appendf(format string, args ...any) { + c.mu.Lock() + c.lines = append(c.lines, fmt.Sprintf(format, args...)) + c.mu.Unlock() +} + +func (c *captureLogger) Infof(format string, args ...any) { c.appendf(format, args...) } +func (c *captureLogger) Warnf(format string, args ...any) { c.appendf(format, args...) } +func (c *captureLogger) Errorf(format string, args ...any) { c.appendf(format, args...) } + +func (c *captureLogger) contains(sub string) bool { + c.mu.Lock() + defer c.mu.Unlock() + for _, l := range c.lines { + if strings.Contains(l, sub) { + return true + } + } + return false +} + +func TestLoggerInjection(t *testing.T) { + target := startEchoServer(t) + n := startNode(t) + + cap := &captureLogger{} + a := agent.New("agent-log") + a.Logger = cap + a.Backoff = 100 * time.Millisecond + a.PingEvery = 0 + go a.Run() + t.Cleanup(a.Close) + + if _, err := a.RegisterTunnel(buildTunnel(t, n, a, "log-tunnel", target, 0)); err != nil { + t.Fatal(err) + } + + // 注入的 logger 必须收到 agent 内部日志 + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cap.contains("已连接节点") { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("注入的 logger 未收到 agent 日志") +} diff --git a/e2e/tcp_test.go b/e2e/tcp_test.go new file mode 100644 index 0000000..263b302 --- /dev/null +++ b/e2e/tcp_test.go @@ -0,0 +1,391 @@ +// 端到端穿透测试：节点 + agent + 内网目标全部进程内启动， +// 验证 公网口 → 桥接(smux/TCP) → 内网目标 的完整链路，不依赖 capricorn。 +// 每条隧道自带节点地址与 JWT，独立连接登录；agent 不再有集中连接。 +package e2e + +import ( + "bytes" + "crypto/rand" + "encoding/binary" + "fmt" + "io" + "net" + "strconv" + "sync" + "testing" + "time" + + "git.zeroonesoft.cn/golib/zonat/agent" + "git.zeroonesoft.cn/golib/zonat/internal/node" +) + +// startEchoServer 启动 TCP 回显服务（模拟 VNC/RDP/3000 等内网 TCP 目标） +func startEchoServer(t *testing.T) string { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { l.Close() }) + go func() { + for { + conn, err := l.Accept() + if err != nil { + return + } + go func(c net.Conn) { + defer c.Close() + io.Copy(c, c) + }(conn) + } + }() + return l.Addr().String() +} + +// startNode 启动一个节点（随机端口，JWT 密钥 test-jwt-secret） +func startNode(t *testing.T) *node.Node { + t.Helper() + n := node.New() + n.JwtSecret = "test-jwt-secret" + n.BindTunnel = "127.0.0.1" + n.SweepInterval = 200 * time.Millisecond + if err := n.Start("127.0.0.1:0"); err != nil { + t.Fatal(err) + } + t.Cleanup(n.Stop) + return n +} + +// startAgent 启动一个 agent（不连接任何节点，隧道注册时才连） +func startAgent(t *testing.T, id string) *agent.Agent { + t.Helper() + a := agent.New(id) + a.Backoff = 100 * time.Millisecond + a.PingEvery = 200 * time.Millisecond + go a.Run() + t.Cleanup(a.Close) + return a +} + +// startEnv 单节点 + agent-1 +func startEnv(t *testing.T) (*node.Node, *agent.Agent) { + t.Helper() + n := startNode(t) + return n, startAgent(t, "agent-1") +} + +// mintToken 签发该节点可认的登录 JWT +func mintToken(t *testing.T, n *node.Node, agentId string) string { + t.Helper() + tok, err := node.SignToken(n.JwtSecret, agentId, time.Hour) + if err != nil { + t.Fatal(err) + } + return tok +} + +// buildTunnel 构造一条登记到指定节点的隧道定义 +func buildTunnel(t *testing.T, n *node.Node, a *agent.Agent, id, target string, ttl int) agent.Tunnel { + t.Helper() + host, portStr, _ := net.SplitHostPort(target) + portNum, _ := strconv.Atoi(portStr) + return agent.Tunnel{ + Id: id, NodeAddr: n.Addr(), Token: mintToken(t, n, a.AgentId), + TargetIp: host, TargetPort: portNum, TTLSec: ttl, + } +} + +func mustRegister(t *testing.T, n *node.Node, a *agent.Agent, target string) (id string, port int) { + t.Helper() + id = fmt.Sprintf("t-%d", time.Now().UnixNano()) + port, err := a.RegisterTunnel(buildTunnel(t, n, a, id, target, 0)) + if err != nil { + t.Fatal(err) + } + return id, port +} + +// roundTrip 建立一条穿透连接，发送 payload 并读回，返回实际回读内容 +func roundTrip(t *testing.T, port int, payload []byte) []byte { + t.Helper() + conn, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), 2*time.Second) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil { + t.Fatal(err) + } + if _, err := conn.Write(payload); err != nil { + t.Fatal(err) + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil { + t.Fatal(err) + } + return got +} + +func TestTCPEchoRoundTrip(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + _, port := mustRegister(t, n, a, target) + + payload := []byte("hello 内网穿透") + if got := roundTrip(t, port, payload); !bytes.Equal(got, payload) { + t.Fatalf("回显不匹配 got=%q want=%q", got, payload) + } +} + +// TestMultiNodeTunnels 本次多节点重构的验收用例： +// 一个 agent 的两条隧道分别登录两个不同节点，各自独立可用。 +func TestMultiNodeTunnels(t *testing.T) { + target1 := startEchoServer(t) + target2 := startEchoServer(t) + n1 := startNode(t) + n2 := startNode(t) + a := startAgent(t, "multi-1") + + p1, err := a.RegisterTunnel(buildTunnel(t, n1, a, "t-node1", target1, 0)) + if err != nil { + t.Fatal(err) + } + p2, err := a.RegisterTunnel(buildTunnel(t, n2, a, "t-node2", target2, 0)) + if err != nil { + t.Fatal(err) + } + if !a.Connected("t-node1") || !a.Connected("t-node2") { + t.Fatal("两条隧道应同时保持连接") + } + + if got := roundTrip(t, p1, []byte("via-node-1")); !bytes.Equal(got, []byte("via-node-1")) { + t.Fatal("节点1的隧道回显失败") + } + if got := roundTrip(t, p2, []byte("via-node-2")); !bytes.Equal(got, []byte("via-node-2")) { + t.Fatal("节点2的隧道回显失败") + } + + // 踢掉其中一个节点上的会话，只影响该节点的隧道 + n1.Kick("multi-1") + time.Sleep(300 * time.Millisecond) + if !a.Connected("t-node2") { + t.Fatal("节点2的隧道不应受节点1踢会话影响") + } +} + +func TestTCPEchoConcurrent(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + _, port := mustRegister(t, n, a, target) + + const workers, rounds = 20, 10 + var wg sync.WaitGroup + errCh := make(chan error, workers) + for i := 0; i < workers; i++ { + wg.Add(1) + go func(seed byte) { + defer wg.Done() + payload := bytes.Repeat([]byte{seed}, 1024) + for r := 0; r < rounds; r++ { + conn, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), 2*time.Second) + if err != nil { + errCh <- err + return + } + if _, err := conn.Write(payload); err != nil { + conn.Close() + errCh <- err + return + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil { + conn.Close() + errCh <- err + return + } + conn.Close() + if !bytes.Equal(got, payload) { + errCh <- fmt.Errorf("回显不匹配 seed=%d", seed) + return + } + } + }(byte(i + 1)) + } + wg.Wait() + select { + case err := <-errCh: + t.Fatal(err) + default: + } +} + +func TestTCPEchoLargePayload(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + _, port := mustRegister(t, n, a, target) + + // 4MB：远大于 smux 单帧(32KB)，验证分帧与重组 + payload := make([]byte, 4<<20) + if _, err := rand.Read(payload); err != nil { + t.Fatal(err) + } + if got := roundTrip(t, port, payload); !bytes.Equal(got, payload) { + t.Fatal("大包回显不匹配") + } +} + +func TestUnregisterStopsListener(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + id, port := mustRegister(t, n, a, target) + + if got := roundTrip(t, port, []byte("ok")); !bytes.Equal(got, []byte("ok")) { + t.Fatal("注销前回显失败") + } + if err := a.UnregisterTunnel(id); err != nil { + t.Fatal(err) + } + time.Sleep(200 * time.Millisecond) + if conn, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), 1*time.Second); err == nil { + conn.Close() + t.Fatal("注销后端口仍可连接") + } +} + +func TestTTLAutoDelete(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + + port, err := a.RegisterTunnel(buildTunnel(t, n, a, "ttl-tunnel", target, 1)) + if err != nil { + t.Fatal(err) + } + if got := roundTrip(t, port, []byte("x")); len(got) != 1 { + t.Fatal("TTL 隧道不可用") + } + + // ttl=1s + 清扫间隔 200ms，2s 后应已删除并收到关闭推送 + time.Sleep(2 * time.Second) + if conn, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), 1*time.Second); err == nil { + conn.Close() + t.Fatal("TTL 过期后端口仍可连接") + } +} + +func TestDuplicateTunnelId(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + tun := buildTunnel(t, n, a, "dup", target, 0) + if _, err := a.RegisterTunnel(tun); err != nil { + t.Fatal(err) + } + + // 同 agent 重复注册同 ID 视为更新配置：停旧起新，应成功 + if _, err := a.RegisterTunnel(tun); err != nil { + t.Fatalf("同 agent 重复注册应顶替成功: %v", err) + } + + // 其它 agent 的活隧道占用同 ID：应拒绝 + b := startAgent(t, "agent-2") + if _, err := b.RegisterTunnel(buildTunnel(t, n, b, "dup", target, 0)); err == nil { + t.Fatal("不同 agent 注册同一隧道ID应失败") + } +} + +// TestJWTAuthRejected JWT 认证拒绝：错误密钥签名 / 已过期 / agentId 不一致 +func TestJWTAuthRejected(t *testing.T) { + target := startEchoServer(t) + n := startNode(t) + + expired, err := node.SignToken(n.JwtSecret, "agent-x", -time.Minute) + if err != nil { + t.Fatal(err) + } + cases := []struct { + name string + token func(t *testing.T) string + }{ + {"错误密钥签名", func(t *testing.T) string { + tok, err := node.SignToken("wrong-secret", "agent-x", time.Hour) + if err != nil { + t.Fatal(err) + } + return tok + }}, + {"过期token", func(*testing.T) string { return expired }}, + {"agentId不一致", func(t *testing.T) string { return mintToken(t, n, "someone-else") }}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + a := startAgent(t, "agent-x") + host, portStr, _ := net.SplitHostPort(target) + portNum, _ := strconv.Atoi(portStr) + _, err := a.RegisterTunnel(agent.Tunnel{ + Id: "bad-jwt", NodeAddr: n.Addr(), Token: c.token(t), + TargetIp: host, TargetPort: portNum, + }) + if err == nil { + t.Fatal("JWT 校验未通过仍登录成功") + } + }) + } +} + +func TestAgentReRegisterAfterKick(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + + portCh := make(chan int, 8) + a.OnTunnelPort = func(id string, port int) { + if port > 0 { + portCh <- port + } + } + + if _, err := a.RegisterTunnel(buildTunnel(t, n, a, "keep", target, 0)); err != nil { + t.Fatal(err) + } + firstPort := <-portCh + + // 节点踢掉会话，该隧道的 worker 应自动重连并重注册 + if !n.Kick("agent-1") { + t.Fatal("Kick 失败") + } + + var newPort int + select { + case newPort = <-portCh: + case <-time.After(5 * time.Second): + t.Fatal("重连后未重注册隧道") + } + if newPort == firstPort { + t.Log("重注册端口与原端口相同（随机分配碰撞，可接受）") + } + if got := roundTrip(t, newPort, []byte("after-kick")); !bytes.Equal(got, []byte("after-kick")) { + t.Fatal("重连后穿透失败") + } +} + +// TestBinarySafePayload 首字节为 0x16（TLS 握手特征）等敏感字节的数据 +// 必须原样通过——完整版的嗅探劫持问题在本库中已不存在。 +func TestBinarySafePayload(t *testing.T) { + target := startEchoServer(t) + n, a := startEnv(t) + _, port := mustRegister(t, n, a, target) + + // 0x16 0x03 0x01 开头：zonat IsHttps 会把它劫持到兜底站 + payload := make([]byte, 0, 4096) + head := []byte{0x16, 0x03, 0x01, 0x00} + payload = append(payload, head...) + var l uint16 = 4000 + binary.BigEndian.PutUint16(payload[3:5], l) + rest := make([]byte, 4096-len(payload)) + if _, err := rand.Read(rest); err != nil { + t.Fatal(err) + } + payload = append(payload, rest...) + + if got := roundTrip(t, port, payload); !bytes.Equal(got, payload) { + t.Fatal("二进制载荷被篡改（疑似协议嗅探干扰）") + } +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..4950ef9 --- /dev/null +++ b/go.mod @@ -0,0 +1,7 @@ +module git.zeroonesoft.cn/golib/zonat + +go 1.24.0 + +require github.com/xtaci/smux v1.5.57 + +require github.com/golang-jwt/jwt/v5 v5.3.1 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..5e4bd57 --- /dev/null +++ b/go.sum @@ -0,0 +1,4 @@ +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/xtaci/smux v1.5.57 h1:N72VbGoSYxgcm6mPOYX0QzEZNVD3UI/JlVvAtXF+WrY= +github.com/xtaci/smux v1.5.57/go.mod h1:IGQ9QYrBphmb/4aTnLEcJby0TNr3NV+OslIOMrX825Q= diff --git a/internal/node/jwt.go b/internal/node/jwt.go new file mode 100644 index 0000000..afa4ae5 --- /dev/null +++ b/internal/node/jwt.go @@ -0,0 +1,60 @@ +// jwt.go agent 登录令牌的 JWT 校验与签发（HS256）。 +// 正式部署由签发方（如 cloud）生成 token 下发给各隧道；节点只做校验。 +// SignToken 供 cmd/node -print-token（运维手工签发）与测试使用。 +package node + +import ( + "errors" + "fmt" + "time" + + "github.com/golang-jwt/jwt/v5" +) + +// verifyToken 校验 agent 登录 JWT： +// - HS256/384/512 签名（显式白名单，拒绝 none 与其它算法） +// - 必须带 exp 且未过期 +// - 若带 agentId 或 sub claim，则必须与登录 AgentId 一致（防跨 agent 冒用） +func verifyToken(token, secret, agentId string) error { + if secret == "" { + return errors.New("节点未配置JWT密钥") + } + parsed, err := jwt.Parse(token, func(t *jwt.Token) (any, error) { + if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok { + return nil, fmt.Errorf("签名算法不允许:%v", t.Header["alg"]) + } + return []byte(secret), nil + }, jwt.WithValidMethods([]string{ + jwt.SigningMethodHS256.Alg(), + jwt.SigningMethodHS384.Alg(), + jwt.SigningMethodHS512.Alg(), + }), jwt.WithExpirationRequired()) + if err != nil { + return err + } + claims, ok := parsed.Claims.(jwt.MapClaims) + if !ok { + return errors.New("claims类型错误") + } + for _, k := range []string{"agentId", "sub"} { + if v, ok := claims[k].(string); ok && v != "" && v != agentId { + return fmt.Errorf("token %s 与登录agentId不一致", k) + } + } + return nil +} + +// SignToken 用节点密钥签发一张 agent 登录 JWT。 +// ttl<=0 会得到立即过期的 token（仅测试构造过期样例时有意义）。 +func SignToken(secret, agentId string, ttl time.Duration) (string, error) { + if secret == "" { + return "", errors.New("未配置JWT密钥") + } + now := time.Now() + claims := jwt.MapClaims{ + "agentId": agentId, + "iat": now.Unix(), + "exp": now.Add(ttl).Unix(), + } + return jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString([]byte(secret)) +} diff --git a/internal/node/node.go b/internal/node/node.go new file mode 100644 index 0000000..ec28753 --- /dev/null +++ b/internal/node/node.go @@ -0,0 +1,307 @@ +// Package node 轻量穿透节点（zonat server 的裁剪版）： +// TCP 桥接登录（JWT）→ smux 会话 → 隧道注册表（纯内存）→ 纯 TCP 隧道监听。 +// 相比 zonat server 裁掉：KCP 桥接、udp/http(s) 隧道、协议嗅探与兜底站、 +// 管理 REST、数据库、限流与流量统计。 +// 隧道生命周期由 agent 经控制流帧命令注册/注销，节点不再对外暴露任何管理端口。 +// 一个 agent 的多条隧道各持一条桥接连接（可分布多节点），因此同一 agentId +// 同时存在多个会话是常态；同名隧道顶替按"隧道 ID + agentId"粒度在注册时进行。 +package node + +import ( + "errors" + "fmt" + "log/slog" + "net" + "strconv" + "sync" + "time" + + "git.zeroonesoft.cn/golib/zonat/agent" +) + +// loginTimeout 登录握手读超时 +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） + + listener net.Listener + + mu sync.Mutex + stopped bool + 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), + } +} + +// Start 启动桥接监听与 TTL 清扫 +func (n *Node) Start(addr string) error { + l, err := net.Listen("tcp", addr) + if err != nil { + return err + } + n.listener = l + slog.Info("节点 桥接监听", "addr", l.Addr().String()) + go n.acceptLoop() + go n.sweepLoop() + return nil +} + +// Stop 停止节点，关闭所有隧道与会话 +func (n *Node) Stop() { + n.mu.Lock() + if n.stopped { + n.mu.Unlock() + return + } + n.stopped = true + sessions := make([]*Session, 0, len(n.sessions)) + for s := range n.sessions { + sessions = append(sessions, s) + } + tunnels := make([]*TunnelServer, 0, len(n.tunnels)) + for _, t := range n.tunnels { + tunnels = append(tunnels, t) + } + n.mu.Unlock() + + if n.listener != nil { + n.listener.Close() + } + for _, t := range tunnels { + t.close() + } + for _, s := range sessions { + s.close() + } +} + +func (n *Node) isStopped() bool { + n.mu.Lock() + defer n.mu.Unlock() + return n.stopped +} + +// Addr 桥接监听地址 +func (n *Node) Addr() string { + if n.listener == nil { + return "" + } + return n.listener.Addr().String() +} + +func (n *Node) acceptLoop() { + for { + conn, err := n.listener.Accept() + if err != nil { + if !n.isStopped() { + slog.Warn("节点 桥接接受连接错误", "err", err.Error()) + } + return + } + go n.handleConn(conn) + } +} + +// handleConn 登录握手（JWT 校验）+ 建立 smux 会话。 +// 新版 agent 每条隧道一条连接，同一 agentId 的多条会话并存。 +func (n *Node) handleConn(conn net.Conn) { + conn.SetReadDeadline(time.Now().Add(loginTimeout)) + f, err := agent.ReadFrame(conn) + if err != nil { + conn.Close() + return + } + if f.Cmd() != agent.CmdLogin { + conn.Close() + return + } + lg := agent.Login{} + if err := f.Unmarshal(&lg); err != nil || lg.AgentId == "" { + _ = writeRet(conn, agent.CmdLogin, 0, agent.Ret{Code: 1, Msg: "登录失败"}) + conn.Close() + return + } + if err := verifyToken(lg.Token, n.JwtSecret, lg.AgentId); err != nil { + slog.Warn("节点 agent登录拒绝", "agent", lg.AgentId, "err", err.Error()) + _ = writeRet(conn, agent.CmdLogin, 0, agent.Ret{Code: 1, Msg: "登录失败"}) + conn.Close() + return + } + if err := writeRet(conn, agent.CmdLogin, 0, agent.Ret{Code: 0, Msg: "登录成功"}); err != nil { + conn.Close() + return + } + // smux 自管超时，清除登录用的读超时 + conn.SetReadDeadline(time.Time{}) + + sess, err := smuxServer(conn) + if err != nil { + conn.Close() + return + } + + s := &Session{agentId: lg.AgentId, node: n, session: sess} + n.mu.Lock() + if n.stopped { + n.mu.Unlock() + conn.Close() + return + } + n.sessions[s] = struct{}{} + n.mu.Unlock() + slog.Info("节点 agent登录", "agent", lg.AgentId, "remote", conn.RemoteAddr().String()) + + s.loop() + n.detachSession(s) +} + +// detachSession 会话断开后：摘除会话，关闭其名下全部隧道 +func (n *Node) detachSession(s *Session) { + n.mu.Lock() + delete(n.sessions, s) + var mine []*TunnelServer + for id, t := range n.tunnels { + if t.session == s { + mine = append(mine, t) + delete(n.tunnels, id) + } + } + n.mu.Unlock() + for _, t := range mine { + t.close() + } + slog.Info("节点 agent会话结束", "agent", s.agentId, "关闭隧道", len(mine)) +} + +// Kick 踢掉指定 agent 的全部会话（运维/测试用）。 +// 新版 agent 每条隧道一条会话，Kick 会断其所有隧道，agent 侧各自重连重注册。 +func (n *Node) Kick(agentId string) bool { + n.mu.Lock() + var targets []*Session + for s := range n.sessions { + if s.agentId == agentId { + targets = append(targets, s) + } + } + n.mu.Unlock() + for _, s := range targets { + s.close() + } + return len(targets) > 0 +} + +// registerTunnel 为 agent 注册一条隧道并开始公网监听 +func (n *Node) registerTunnel(s *Session, t agent.Tunnel) (agent.Tunnel, error) { + if t.Id == "" { + return t, errors.New("隧道ID不能为空") + } + if t.TargetIp == "" || t.TargetPort <= 0 || t.TargetPort > 65535 { + return t, fmt.Errorf("目标地址错误:%s", t.TargetAddr()) + } + if t.ListenPort < 0 || t.ListenPort > 65535 { + return t, fmt.Errorf("监听端口错误:%d", t.ListenPort) + } + // 凭据只用于登录，节点侧不留存也不回传（CmdTarget 不携带） + t.NodeAddr, t.Token = "", "" + + n.mu.Lock() + defer n.mu.Unlock() + if n.stopped { + return t, errors.New("节点已停止") + } + // 同 ID 隧道已存在：同 agent 重登顶替（更新配置）或死会话残留可替换， + // 其它 agent 的活隧道拒绝 + if old, ok := n.tunnels[t.Id]; ok { + if old.session.agentId != s.agentId && !old.session.isDead() { + return t, fmt.Errorf("隧道已存在:%s", t.Id) + } + old.close() + old.session.close() // 顶替旧连接，避免旧 worker 与新 worker 争抢 + delete(n.tunnels, t.Id) + slog.Info("节点 同agent隧道顶替", "id", t.Id, "agent", s.agentId) + } + + ts := &TunnelServer{tunnel: t, session: s} + ts.touch() + l, err := net.Listen("tcp", net.JoinHostPort(n.BindTunnel, strconv.Itoa(t.ListenPort))) + if err != nil { + return t, err + } + ts.listener = l + _, portStr, _ := net.SplitHostPort(l.Addr().String()) + ts.tunnel.ListenPort, _ = strconv.Atoi(portStr) + + n.tunnels[t.Id] = ts + go ts.acceptLoop() + slog.Info("节点 隧道注册", "id", t.Id, "listen", ts.tunnel.ListenPort, + "target", t.TargetAddr(), "ttl", t.TTLSec) + return ts.tunnel, nil +} + +// unregisterTunnel 删除隧道（关闭公网监听） +func (n *Node) unregisterTunnel(id string) bool { + n.mu.Lock() + ts, ok := n.tunnels[id] + if ok { + delete(n.tunnels, id) + } + n.mu.Unlock() + if !ok { + return false + } + ts.close() + slog.Info("节点 隧道删除", "id", id) + return true +} + +func (n *Node) sweepLoop() { + ticker := time.NewTicker(n.SweepInterval) + defer ticker.Stop() + for range ticker.C { + if n.isStopped() { + return + } + n.sweepOnce() + } +} + +// sweepOnce 清扫 TTL 过期且空闲的隧道，并向所属 agent 推送关闭通知 +func (n *Node) sweepOnce() { + now := time.Now() + n.mu.Lock() + var expired []*TunnelServer + for id, ts := range n.tunnels { + ttl := time.Duration(ts.tunnel.TTLSec) * time.Second + if ttl > 0 && ts.connCount.Load() == 0 && now.Sub(ts.lastActiveTime()) > ttl { + expired = append(expired, ts) + delete(n.tunnels, id) + } + } + n.mu.Unlock() + for _, ts := range expired { + ts.close() + ts.session.pushClosed(ts.tunnel.Id) + slog.Info("节点 隧道TTL过期删除", "id", ts.tunnel.Id) + } +} + +func writeRet(conn net.Conn, cmd byte, sid uint32, ret agent.Ret) error { + f := agent.NewFrame(agent.FrameVersion, cmd, sid) + if err := f.Marshal(ret); err != nil { + return err + } + return agent.WriteFrame(conn, f) +} diff --git a/internal/node/session.go b/internal/node/session.go new file mode 100644 index 0000000..bc35a37 --- /dev/null +++ b/internal/node/session.go @@ -0,0 +1,137 @@ +package node + +import ( + "net" + "sync" + "sync/atomic" + + "git.zeroonesoft.cn/golib/zonat/agent" + + "github.com/xtaci/smux" +) + +// Session 一个 agent 连接的会话：smux 服务端 + 控制流处理。 +// 数据流由节点侧隧道监听主动 OpenStream；agent 主动打开的第一条流是控制流， +// 用于注册/注销隧道与心跳。 +type Session struct { + agentId string + node *Node + session *smux.Session + + control net.Conn // agent 打开的控制流 + controlMu sync.Mutex // 控制流写锁（handleControl 与 pushClosed 并发写） + dead atomic.Bool +} + +func (s *Session) isDead() bool { return s.dead.Load() } + +func (s *Session) close() { + s.dead.Store(true) + s.session.Close() +} + +// OpenStream 建立一条到 agent 的数据流 +func (s *Session) OpenStream() (net.Conn, error) { + return s.session.OpenStream() +} + +// loop 接受 agent 打开的控制流，直到会话结束 +func (s *Session) loop() { + for { + stream, err := s.session.AcceptStream() + if err != nil { + s.close() + return + } + // agent 打开的第一条流是控制流，pushClosed 依赖它； + // 赋值与写入共用 controlMu，避免与推送协程竞争 + s.controlMu.Lock() + if s.control == nil { + s.control = stream + } + s.controlMu.Unlock() + go s.handleControl(stream) + } +} + +// handleControl 处理控制流上的帧：注册/注销隧道、心跳。 +// 单条控制流内的帧顺序处理；应答与节点推送共用 controlMu 串行写入。 +func (s *Session) handleControl(stream net.Conn) { + defer stream.Close() + for { + f, err := agent.ReadFrame(stream) + if err != nil { + return + } + switch f.Cmd() { + case agent.CmdRegisterTunnel: + t := agent.Tunnel{} + ret := agent.Ret{Code: 1, Msg: "注册失败"} + if err := f.Unmarshal(&t); err != nil { + ret.Msg = err.Error() + } else if nt, err := s.node.registerTunnel(s, t); err != nil { + ret.Msg = err.Error() + } else { + ret = agent.Ret{Code: 0, Msg: "注册成功", Port: nt.ListenPort} + } + s.reply(stream, f, ret) + case agent.CmdUnregisterTunnel: + t := agent.Tunnel{} + ret := agent.Ret{Code: 1, Msg: "注销失败"} + if err := f.Unmarshal(&t); err != nil { + ret.Msg = err.Error() + } else if s.node.unregisterTunnel(t.Id) { + ret = agent.Ret{Code: 0, Msg: "注销成功"} + } else { + ret.Msg = "隧道不存在" + } + s.reply(stream, f, ret) + case agent.CmdPing: + s.writeControl(func() error { + return agent.WriteFrame(stream, agent.NewFrame(agent.FrameVersion, agent.CmdPing, f.StreamID())) + }) + case agent.CmdLogin, agent.CmdTarget, agent.CmdTunnelClosed: + // 不该出现在 agent→节点方向，忽略 + default: + s.reply(stream, f, agent.Ret{Code: 1, Msg: "未知命令"}) + } + } +} + +func (s *Session) reply(stream net.Conn, req agent.Frame, ret agent.Ret) { + s.writeControl(func() error { + f := agent.NewFrame(agent.FrameVersion, req.Cmd(), req.StreamID()) + if err := f.Marshal(ret); err != nil { + return err + } + return agent.WriteFrame(stream, f) + }) +} + +// pushClosed 节点主动通知 agent：隧道已被删除（TTL 过期等） +func (s *Session) pushClosed(tunnelId string) { + if s.control == nil { + return + } + s.writeControl(func() error { + f := agent.NewFrame(agent.FrameVersion, agent.CmdTunnelClosed, 0) + if err := f.Marshal(agent.Tunnel{Id: tunnelId}); err != nil { + return err + } + return agent.WriteFrame(s.control, f) + }) +} + +func (s *Session) writeControl(fn func() error) { + s.controlMu.Lock() + defer s.controlMu.Unlock() + if s.dead.Load() { + return + } + // 写失败意味着会话已死，由读循环感知并收尾 + _ = fn() +} + +func smuxServer(conn net.Conn) (*smux.Session, error) { + return smux.Server(conn, agent.SmuxConfig()) +} diff --git a/internal/node/tunnel.go b/internal/node/tunnel.go new file mode 100644 index 0000000..99a0fee --- /dev/null +++ b/internal/node/tunnel.go @@ -0,0 +1,107 @@ +package node + +import ( + "io" + "log/slog" + "net" + "sync/atomic" + "time" + + "git.zeroonesoft.cn/golib/zonat/agent" +) + +// TunnelServer 一条已注册隧道的公网侧监听。 +// 纯 TCP 字节管道：无协议嗅探、无 HTTP 代理、无证书，任何 TCP 协议原样通过。 +type TunnelServer struct { + tunnel agent.Tunnel + session *Session + listener net.Listener + + connCount atomic.Int64 + lastActive atomic.Int64 // unix nano + closed atomic.Bool +} + +// Port 公网监听端口 +func (t *TunnelServer) Port() int { return t.tunnel.ListenPort } + +func (t *TunnelServer) touch() { t.lastActive.Store(time.Now().UnixNano()) } + +func (t *TunnelServer) lastActiveTime() time.Time { + return time.Unix(0, t.lastActive.Load()) +} + +func (t *TunnelServer) acceptLoop() { + for { + conn, err := t.listener.Accept() + if err != nil { + return + } + if t.closed.Load() { + conn.Close() + return + } + go t.handleConn(conn) + } +} + +// handleConn 公网连接 → smux 新流 → 首帧告知目标 → agent 回执后双向管道。 +// 与 zonat 多路复用路径一致，仅去掉协议嗅探与限流。 +func (t *TunnelServer) handleConn(conn net.Conn) { + defer conn.Close() + if t.session.isDead() { + return + } + stream, err := t.session.OpenStream() + if err != nil { + return + } + defer stream.Close() + + tf := agent.NewFrame(agent.FrameVersion, agent.CmdTarget, 0) + if err := tf.Marshal(t.tunnel); err != nil { + return + } + if err := agent.WriteFrame(stream, tf); err != nil { + return + } + rf, err := agent.ReadFrame(stream) + if err != nil { + return + } + ret := agent.Ret{} + if err := rf.Unmarshal(&ret); err != nil || ret.Code != 0 { + slog.Warn("节点 agent拨号失败", "id", t.tunnel.Id, "msg", ret.Msg) + return + } + + // 公网侧 TCP 保活（smux 流自身由会话级 keepalive 保护） + if tc, ok := conn.(*net.TCPConn); ok { + tc.SetKeepAlive(true) + tc.SetKeepAlivePeriod(5 * time.Second) + } + t.connCount.Add(1) + t.touch() + defer func() { + t.connCount.Add(-1) + t.touch() + }() + + pipe(conn, stream) +} + +func (t *TunnelServer) close() { + if t.closed.CompareAndSwap(false, true) && t.listener != nil { + t.listener.Close() + } +} + +// pipe 双向拷贝，任一方向结束即关闭两端 +func pipe(a, b net.Conn) { + go func() { + io.Copy(b, a) + b.Close() + }() + io.Copy(a, b) + a.Close() +}