diff --git a/AUDIT.md b/AUDIT.md new file mode 100644 index 0000000..00a602d --- /dev/null +++ b/AUDIT.md @@ -0,0 +1,286 @@ +# redapricot 审计报告 + +审计日期:2026-08-15 +审计方式:5 个并行探索子代理深读全部源码(~8200 行)与测试,交叉核对两端实现与 +PROTOCOL.md;随后对全部高风险发现逐一人工复核(含 velocity 模块与 Velocity 官方 +源码 `dev/4.0.0` 的 `PlayerDataForwarding.java` 逐字节比对)。 + +--- + +## 0. 总体评价 + +**设计质量很高,且文档与实现高度一致。** 亮点包括:单事件循环 hub 让并发模型归零; +「持锁不 dial」的池设计;resume 三偏移量(Sent/Accepted/Delivered)的区分在两端实现 +和注释中都正确且互相印证;流控窗口同时充当重传缓冲区上限(「保留区无需自带上限」 +这一论证成立,依赖链完整);off-switch(`streamResume:false`、`statsIntervalMs:0`) +确实只留一个分支;写超时、心跳、会话建立 deadline 等「liveness 显式化」哲学落实到位。 + +**主要风险集中在三处**:客户端 resume 状态的并发管理(存在静默数据损坏路径)、 +`Close()` 与关闭语义(库用场景泄漏)、测试覆盖(6 个协议行为零测试、Java 侧除纯 +函数外零单测、e2e 注册就绪用固定 sleep)。 + +--- + +## 1. 协议实现审计 + +### 1.1 线级一致性:逐字节核对,全部一致 ✓ + +握手布局、Rekey 帧(`magic‖randLen‖rand‖ts‖flags‖window`,REKEY=rand‖ts 不含 +magic)、SessionReady、密钥派生(`SHA3-256(PK‖0x01/0x02)`、零 nonce、counter 0)、 +控制消息、mux 帧(含 RESUME/RESUME_ACK 字段顺序)、流控语义(只 DATA 计窗口、WND +半窗批量、32KiB chunk)、resume 三偏移量语义、全部常量、配置默认值与钳制——两端实现 +与 PROTOCOL.md 全部一致,无任何严重不一致。 + +**velocity 模块已经官方源码确证无误**(`client/velocity.go` 与 PaperMC/Velocity +`dev/4.0.0` 的 `PlayerDataForwarding.java` 比对): + +- payload 布局 `VarInt(version) ‖ String(ip) ‖ UUID[16] ‖ String(name) ‖ VarInt(properties)` + 与官方 `writeVarInt + writeString + writeUuid + writeString + writeProperties` 完全一致; +- 版本协商(1.19.3+ 时 `requested≥4` 发 4、否则发 1)与官方 `findForwardingVersion` + 逐分支一致,包括 v4 lazy-session 不带 key 段、v2/v3 只在 1.19–1.19.2 且客户端有 + key 时使用(redapricot 无 key 故正确回落 v1)。 + +### 1.2 轻微偏差(不影响互通,但建议修) + +| # | 问题 | 位置 | 说明 | +|---|------|------|------| +| P1 | **pattern 注册前被归一化,违反 §5 "verbatim" 约定** | `client.go:49-52`、`config.go:314-321` | 客户端注册的是 `NormalizeAddress(pattern)` 后的字符串。hub 端确实原样存储,但大小写敏感的 regex 会被破坏:以 `\.` 结尾的 pattern 经 `TrimRight(".")` 变成孤立 `\` 导致编译失败;`\Q…\E` 被小写化为 `\q…\e` 同样编译失败。应原样注册、仅匹配时归一化 | +| P2 | `streamWindowBytes: 0` 语义两端不同 | `Config.java:36-37` vs `config.go:71-75` | Go 视为"未设置→262144",Java 钳到最小 32768。规范未定义 0,属规范留白 | +| P3 | Intent 18 规范与实现不符 | `PROTOCOL.md:73-75` vs `HubConnection.java:96-98` | 规范称"回复 status line 后关闭",实现只 log+close | +| P4 | PSK address 比较宽松 | `HubConnection.java:113` | 规范要求小写精确匹配,实现用 `equalsIgnoreCase` | +| P5 | Go 侧缺镜像常量 | `client/config.go` | `INTENT_RESERVED`、`REGISTER_OK/ERR` 未定义,违反 CLAUDE.md「常量镜像」不变式 | +| P6 | RST 原因码半实现 | `worker.go:430-432`;`WorkerConn.java:55` | 客户端 `sendRst` 从不带原因字节(`RST_DIAL_FAILED/FLOW_CONTROL/RESUME_ABANDONED` 是死常量);hub 收到 RST 也不解析原因字节。协议标称"both directions"的对称性未实现 | +| P7 | 规范文档错误 | `PROTOCOL.md:214` | §5 表 SessionReady 标 "(none)",与 §4(flags+window+grace)矛盾;§7.5 未提 `RST(ALREADY_BOUND)` 分支 | + +--- + +## 2. 设计审计 + +**做得对且值得保留的设计**(均已核实,非泛泛而谈): + +1. **池不在持锁时 dial**——`Allocate` 与 `maybeGrowLocked` 都把 dial 放到释放锁之后 + /后台 goroutine,注释与代码一致;空池时单 dialer + `cond.Wait` + `dialGen` 代际 + 账本正确处理失败信号。 +2. **流经 hub 路由而非闭包捕获 conn**——`Hub.onPlayerData` 按 `st.worker` 转发, + reattach 无需重装 handler,规避了死 conn 静默写缺陷。 +3. **resume 账本**——重放点取 peer 的 accepted、窗口按 delivered 重述、丢弃自身在途 + credit、`Delivered ≤ Accepted` 隐含不变式,两端对称且都有注释钉住;`UnackedBytes` + 的摊还 O(1) advance、`from()` 越界返回 nil 触发终局 teardown(不静默截断)都正确。 +4. **hub 侧 parked 上限**——`maxParkedStreams/maxParkedBytes` 以插入序 LinkedHashMap + 淘汰最旧,`parkedBytes()` 计数与 `removeStream` 的扣减逻辑经核对自洽(写法绕,见 §4.6)。 +5. **grace 协商而非配置假设**——hub 在 SessionReady 公布 `resumeGraceMs`,客户端钳制 + 在它之下,"客户端必须先放弃"从运维约定变成协议约束。 +6. **shaper**——token bucket + start-time fair queue,vclock 不因 size 前移是刻意的; + 只整形 DATA、控制帧永不延迟;关停不卡流量;`NewShaper` 返回 nil 的短路设计让限速 + 关闭时热路径零分支。 + +**设计层面的顾虑**(非错误,属取舍): + +- **单事件循环 hub**:一块网卡 + 1 MiB 帧 + 32KiB chunk,单核吞吐上限明确(文档已 + 承认)。但 **ChaCha20 的 `Cipher.update` 每次帧调用都过 JCE**,帧头 VarInt 解析 + + Buffer 切片每帧多次分配,实际吞吐可能在数百 Mbps 量级——建议做一次基准确认。 +- **TCP-level HOL** 是 mux-over-TCP 固有(文档已承认),池是缓解而非解决。 +- **无 AEAD**(文档已承认),ChaCha20 无完整性保护意味着主动攻击者可以翻转密文。 + +--- + +## 3. 逻辑漏洞(按严重度排序) + +### 🔴 C1. 双 `resumeLoop` 竞态 → 静默数据损坏(客户端,最严重) + +触发链(`client/resume.go`、`client/worker.go`): + +1. `tryResume` 在新建 conn B 上 `registerStream` 并发 RESUME; +2. `completeResume` 重放中途 `wc.sendData` 失败(B 刚注册即死,`resume.go:271-273`) + ——注意 `s.parked = false` 在重放**之前**已复位(`resume.go:253`); +3. B 的 readLoop 退出 → 对 B 上每条流调 `park()`(`worker.go:401-408`)→ + `already := s.parked` 为 **false** → 再起**第二个** `resumeLoop`(`resume.go:80`); +4. 旧 loop 仍在重试(该错误非终局,退避 500ms 后继续)。两个 loop 并发 `tryResume`: + 各自 `registerStream` 到不同 conn、各自把 `s.resumeWait` 覆盖成自己的 channel、 + 发两个并发 RESUME。 + +后果取决于时序,两种都是坏的:RESUME_ACK 被"非赢家"loop 消费 → leg 绑定到 hub 不 +认识的 (conn, sid),上游(dest→player)数据被 hub 静默丢弃——**玩家 mute,无任何 +日志**;或 RST 到达时 `resumeWait` 为 nil → 误 teardown 一个 hub 已恢复的流。 + +**修复**:`completeResume` 失败时保持 `parked=true`(不让第二个 loop 启动),让 +`park()` 的 `already` 分支兜底。 + +### 🔴 C2. 半开连接 + `RST(ALREADY_BOUND)` → 流永久悬挂(客户端) + +场景:客户端心跳超时(60s)判定 worker conn 死亡并关闭,但 hub 侧该 conn 仍在 +90s idle 窗口内(`WorkerConn.java` 侧流未 park)→ 客户端新 conn 发 RESUME → hub +`takeParked` 为 null、`streamByCid` 非空 → `RST(ALREADY_BOUND)` +(`WorkerConn.java:207-210`)→ 客户端 `onRst` 映射为 `errResumeRaced` → `resumeLoop` +**无条件退出**(`resume.go:104-108`),不 teardown 不重试。此后该流 `parked=true` +永远挂着、destination socket 永久泄漏,直到进程退出。 + +**修复**:`errResumeRaced` 改为可重试(C1 修复后不再有真并发 loop,ALREADY_BOUND 只 +意味着 hub 旧 conn 尚存),grace 耗尽后正常 teardown。 + +### 🟠 C3. `Close()` 不干净 + `dialSession` 无视 ctx(客户端) + +- `Close()`(`client.go:413-421`)无 shutdown 标志。关闭 pool conns 后,readLoop 把流 + park → `resumeLoop` 通过 `pool.Allocate()`(空池)**同步拨新 TCP 连接**继续重挂, + 最长持续一个 grace(15s); +- `dialSession`(`client.go:100-210`)只用 `HandshakeTimeout` 硬 deadline,不读 ctx。 + `Close()` 与在途重连/拨号并发时,新连接建立后无人关闭——`serveControl` 永久阻塞在 + `ReadFrame` 上,goroutine + socket 泄漏。 + +**修复**:`atomic.Bool closing`;`park` 时已 closing 直接 teardown;`dialSession` 用 +`net.Dialer.DialContext` + `context.AfterFunc` 在 ctx 取消时关连接。 + +### 🟠 C4. 手建 Config `PingIntervalMs<=0` → `time.NewTicker(0)` panic(客户端) + +`New()`(`client.go:38-68`)只钳 window、解析带宽,**不验证 PingIntervalMs**(`LoadConfig` +才钳到 ≥1000,`config.go:292-298`)。注释明确说 Config 可被直接构造(测试用),此时 +`time.NewTicker(0)` 直接崩进程(`client.go:342`、`worker.go:247`)。 + +**修复**:`New()` 里 clamp 或显式报错。 + +### 🟡 服务端:无同等严重的逻辑漏洞 + +逐一复核了 `Hub.removeSession/expireOrphans/replayAwaiting`(orphan 语义、deadline +切换、`rearm` 的 timer 与 `cancelTimer`)、`WorkerConn.handleResume`(含 replay 不 +重复计窗、`st.credited=0`、CID 重铸)、`EnforceParkedCaps`(`parkedBytes` 先减后加再 +经 `removeStream` 统一扣减——自洽但绕)、`EncryptedFrames.pump`(明文长度前缀/加密 +负载/rekey 切换在帧边界无歧义)。**未发现逻辑错误**。健壮性/一致性缺口: + +- **hub 不校验客户端发送窗口**(`WorkerConn.handleData` 只信任客户端)。协议 §7.3 明确 + 允许对超窗 RST,但 hub 不查——持 PSK 的恶意/损坏客户端可让 hub 端每流缓冲区无界 + 增长。建议按协议补校验。 +- **`ProtoWriter.u8/u16` 静默截断**(`ProtoWriter.java:11,16-19`):越界值截断为 + "合法外观"的错误字节,无日志;与 reader 侧严格校验不对称。 +- **停机无优雅 drain**:`Main.java:40` 的 shutdown hook 不等待 `vertx.close()` 完成 + (JVM halt 掐断异步关闭);无协议层下线通知;挂起玩家/流被硬切。 +- **Vert.x 5 日志路由存疑**(中高置信):`Main.java:16` 设的是 Vert.x 4 的 + `vertx.logger-delegate-factory-class-name`,Vert.x 5 核心日志已走 SLF4J,且 classpath + 上无 slf4j-api/log4j-slf4j2-impl。 + +--- + +## 4. 错误处理缺陷 + +| # | 问题 | 位置 | 后果 | +|---|------|------|------| +| E1 | **`registerAll` 半失败静默** | `client.go:254-263` | Register 写失败只 log 并 return,`connectControl` 仍成功、`Start` 仍返回成功,但 hub 上一个 pattern 都没注册——玩家全被拒,且日志误导 | +| E2 | **RegisterAck status=1 不处理** | `client.go:314-318` | 非法 regex 只打日志,mapping 保留,之后永远收不到 ControlRequest | +| E3 | **WND/FIN/RST 写失败全吞** | `worker.go:429-435`、`worker.go:373` | WND 丢失 → hub 停发该流 → 玩家卡死,全程无日志 | +| E4 | **解析错误大量 `_ =`** | `client.go:325-327`、`worker.go:364-368` | 坏帧静默吞掉(hub 可信,可接受) | +| E5 | **`New()` 带宽解析失败静默关限速** | `client.go:58-59` | 运维以为限速生效,实际已关闭 | +| E6 | **`shaperStall` 归因污染** | `worker.go:637-648` | `Acquire`→`emit`(含最长 30s 的 socket 写)之间时间全记入"带宽上限";hub 停读的拥堵被误报成限速,与 stats 想区分的三病因矛盾。修法:elapsed 在 `Acquire` 返回后立即取 | +| E7 | **日志不可关联** | 多处 `stream %d` | sid 是 per-conn 的且 reattach 后会变,不带 conn id,排障无法把日志对到同一条流 | +| E8 | **RST 原因映射过宽** | `resume.go:309-312` | `RST_FLOW_CONTROL` 这类协议违规终局被当可重试,徒增延迟 | +| E9 | **resume 尝试越过 deadline 最多 10s** | `resume.go:152-168` | `time.After(ResumeAckTimeout)` 不随 deadline 缩短,玩家侧最坏失败延迟被放大到 grace+10s | +| E10 | **resumeGrace 与 hubGrace 取 min 后丢失下限** | `resume.go:38-44` | 误配 500ms grace 的 hub 让客户端预算 < 一次拨号所需,仅可观测性问题 | + +--- + +## 5. 复杂度分析 + +| 热点 | 位置 | 分析 | +|------|------|------| +| `Stream` 类型 | `worker.go:451-518` | 25+ 字段、4 种同步原语(`s.mu`/`sendMu`/原子量/chan),resume 状态(parked/resumeWait/leg/cid/三 offset)与流量状态混居。**C1 正是"parked 复位"与"loop 存活"两状态未拆开导致的** | +| `completeResume` | `resume.go:203-288` | 全项目最复杂单函数:锁序(sendMu→s.mu)、三偏移换算、重放、窗口重述、欠账 FIN。正确但难审 | +| `WorkerPool.Allocate` | `worker.go:54-92` | cond 等待 + `dialGen`/`dialErr` 代际账本,失败路径无单测 | +| 接收方向账本 | `worker.go` 多处 | `q`/`qBytes`/`acceptedOffset`/`deliveredOffset`/`consumed` 由 writeLoop/deliverFromHub/credit/completeResume 四处维护 | +| shaper `headLocked` | `shaper.go:241-252` | 每 grant 线性扫描 O(n²);n 为流数时无感,但每 DATA 块一次 `make(chan)` 分配是热点 | +| `wire.Reader.VarInt` | `mc.go:55-61` | 每次 `bytes.NewReader` 分配,velocity/帧解析热路径高频调用 | +| velocity 状态机 | `velocity.go` | 两方向缓冲 + 5 个布尔,逻辑正确但可合并为三值枚举 | +| 测试 harness | `e2e/*` | 三层 helper 金字塔 + 3 个文件各自复制"hub+relay+dest+client"接线 | + +--- + +## 6. 测试覆盖审计 + +**覆盖良好的**:黑盒路径恢复、控制断线三种路径、resume 字节精确/并发/禁用/grace 到期、 +带宽整形与公平、慢流隔离、velocity 单测 13 场景 + e2e、PROXY v2、池广度优先、错误 PSK +拒绝。e2e 拉起**真实 Java hub 子进程**,架构扎实。 + +**零测试的协议行为**(6 个,e2e + 单测均未断言): + +1. **RST 原因码**——协议定义了 6 个,两端只实际发射 UNKNOWN_STREAM/ALREADY_BOUND; + ALREADY_BOUND 竞争路径(C1/C2 的暴露面)无任何测试; +2. **WND 窗口违规**——hub 不校验客户端窗口、客户端超窗检测无测试; +3. **pending 超时**——"已匹配但 SYN 永不来的玩家被关"无测试; +4. **hub idle watchdog**——`sessionIdleTimeoutMs` 无测试; +5. **CID 单次使用**——二次 SYN、resume 后旧 CID 失效无测试; +6. **1 MiB 帧上限**——两侧都无超长帧测试。 + +**结构性缺口**: + +- **Java 侧只有 1 个测试文件**(`CryptoCodecTest.java`,9 个纯函数用例)。hub 状态机、 + `HubConnection` 握手/rekey 校验、`EncryptedFrames`、`WorkerConn` 流控/resume 零单测。 + 根因之一:`testHub()` 传 `vertx=null`(`CryptoCodecTest.java:79-83`),`Hub.rearm` 用 + `vertx.setTimer` 会 NPE——**测试缝本身封死了 timer 路径**。 +- **跨语言 pinning 只有 SHA3-224 一处**;ChaCha20 字节一致性无 Java 侧单测、无 + RFC 8439 KAT,`deriveKey` 无绝对向量,全靠 e2e 兜底。 +- **e2e 注册就绪用固定 200ms sleep**(`e2e_test.go:47`)——全 e2e 最大的 flaky 源。 + `Start` 只保证 Register 帧**写出**,不等待 RegisterAck。 +- **flaky 前五**:`TestShaperSharesFairlyBetweenStreams`(35% 容差)、 + `TestBlackholedPathRecovers`(45s 轮询)、`TestResumePreservesByteStream`/ + `TestResumeWithConcurrentStreams`、`TestControlOutageHangsArrivingPlayer`。 +- **velocity 无真实 Paper golden vector**——测试只证明自洽(本次审计已确认字节布局与 + 官方源码一致,建议补 golden 测试防回归)。 + +--- + +## 7. 重构建议(按性价比排序) + +**P0 — 正确性(先做)** +1. **修 C1 双 resumeLoop 竞态**:`completeResume` 失败保持 `parked=true`; + `deliverResume` 校验 loop 归属。补 ALREADY_BOUND 竞争路径的 e2e。 +2. **修 C2 半开悬挂**:`errResumeRaced` 改为在 grace 剩余内重试。 +3. **修 C3/C4**:`Close()` 加 shutdown 标志、`dialSession` 尊重 ctx、`New()` 校验 + `PingIntervalMs`。 + +**P1 — 健壮性与可观测性** +4. 客户端 `registerAll` 失败要传导;RegisterAck status=1 至少移除/标记该 mapping。 +5. hub 补客户端发送窗口校验(§7.3 允许的 RST);`ProtoWriter` 加 `u8/u16` 范围校验。 +6. 修正 `shaperStall` 归因(elapsed 提前取);流日志统一带 `connID/sid`。 +7. pattern 原样注册(P1);`streamWindowBytes:0` 语义统一并写进规范(P2)。 + +**P2 — 测试** +8. Java 侧补:ChaCha20 RFC 8439 KAT + 分块跨块一致性、`EncryptedFrames` rekey、 + `Hub` 状态机(timer 可注入)。 +9. 客户端加 ready 信号(RegisterAck 挂钩到可等待 channel),替换 200ms sleep;补 + 6 个零覆盖协议行为的测试。 +10. velocity 补官方字节 golden vector。 + +**P3 — 结构** +11. `completeResume` 拆成小函数 + 状态机图注释;接收方向账本封装为 `recvLedger`。 +12. harness 引入单一 `newTunnelHarness(t, opts…)` 组装点。 +13. 热路径去分配:`wire.Reader.VarInt` 改游标读取;shaper chan 池化(低优先级)。 +14. `Main` 的 shutdown hook 等待 `vertx.close()` 完成;核实 Vert.x 5 日志绑定。 + +--- + +## 8. 结论 + +协议实现是**干净且两端一致的**——这是本审计最重要的结论,逐字节核对未发现任何线级 +不兼容。设计文档质量上乘,注释大量引用"它防的是什么故障",可维护性极好。 + +需要优先处理的是客户端 resume 路径的两个并发缺陷(C1 静默数据损坏、C2 悬挂泄漏) +和关闭语义(C3),其次是测试缺口。这两类问题恰好互相印证:C1 和 C2 都发生在 +"连接死亡/重建"的交叉时序里,而这类时序恰恰是当前 e2e 覆盖最薄的地方。 + +--- + +## 附:修复状态跟踪 + +| 编号 | 内容 | 状态 | +|------|------|------| +| C1 | 双 resumeLoop 竞态 | ✅ 已修(`completeResume` 重放失败时重新置 `parked=true`,统计计数移至重放成功之后) | +| C2 | 半开 + RST(ALREADY_BOUND) 悬挂 | ✅ 已修(`resumeLoop` 对 `errResumeRaced` 改为可重试,由 grace 截止时间兜底) | +| C3 | Close()/dialSession ctx | ✅ 已修(`Client.closing` 标志 + 内部 ctx/cancel;`dialSession(ctx)` 用 `DialContext` + `AfterFunc` 关闭;`WorkerPool.closed` 标志门控 Allocate/后台拨号;serveControl 重连被 closing 门控) | +| C4 | NewTicker(0) panic | ✅ 已修(`pingInterval()` 单点钳制,新增 `DefaultPingIntervalMs`) | +| E6 | shaperStall 归因污染 | ✅ 已修(`elapsed()` 移至 `Acquire` 返回后、socket 写之前采样) | +| E7 | 流日志缺 conn id | ✅ 已修(`leg.String()` = `conn%d/sid%d`,替换全部 5 处流日志) | +| P3 | Intent 18 status line | ✅ 已修(hub 回复 Minecraft status 包 `[Len][0x00][JSON]` 后关闭;已用真实 codec 探针验证) | +| P4 | PSK address 严格小写 | ✅ 已修(`equalsIgnoreCase` → 严格 `equals`;大写变体探针验证被拒) | +| P5 | Go 镜像常量 | ✅ 已修(`IntentReserved=18`、`RegisterOk=0x00`、`RegisterErrPattern=0x01`,RegisterAck 处理用命名常量) | +| P7 | PROTOCOL.md §5 SessionReady 表 | ✅ 已修(补充 `Flags/RecvWindow/ResumeGraceMs` 字段;§2 补充 Intent 18 status line 格式) | + +验证:`go test ./client/... -count=1`、`go test -race ./client/...`、`gradle -p server test`、 +`go test ./e2e/... -count=1`(含 resume/blackhole 套件)全部通过;Intent 18 与严格小写 +PSK 已用真实客户端 codec 对真实 hub 探针验证。 diff --git a/PROTOCOL.md b/PROTOCOL.md index 845895f..fb2aac6 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -74,6 +74,19 @@ The hub reads exactly one Handshake packet and dispatches on `Intent`: | `18` | Reserved for redapricot management/status. Never matched against patterns. The reference hub replies with a status line and closes. | | anything else | **Player** connection. `ServerAddress` is matched against the registered **regex** PATTERNs (§5.1). | +For `Intent == 18` the hub replies with a Minecraft status-response packet and +closes — the same shape a player receives for a status query (Intent 1), so the +port can be probed with ordinary tooling: + +``` +[Len: VarInt][PacketID 0x00][JSON: String] +``` + +The JSON is a minimal status payload, e.g. +`{"description":{"text":"redapricot hub"},"version":{"name":"redapricot","protocol":767},"players":{"max":0,"online":0}}`. +It is the one reply the hub sends in plaintext: Intent 18 never negotiates +encryption or any other redapricot state. + For `Intent == 17` the hub additionally requires `ServerAddress == lowercase_hex(SHA3-224(PSK))` — a 56-character hex string. This is the first (cheap) proof that the peer knows the PSK. A mismatch closes @@ -211,7 +224,7 @@ Type : u8 | Type | Name | Direction | Fields | |--------|----------------|-----------|--------| -| `0x00` | SessionReady | S → C | *(none)* — the confirmation frame from §4 | +| `0x00` | SessionReady | S → C | `Flags: VarInt`, `RecvWindow: VarInt`, `ResumeGraceMs: VarInt` (only when the hub accepted `STREAM_RESUME`) — the confirmation frame from §4 | | `0x01` | Register | C → S | `Pattern: String` | | `0x02` | Unregister | C → S | `Pattern: String` | | `0x03` | RegisterAck | S → C | `Pattern: String`, `Status: u8` (0 = ok, 1 = invalid pattern) | diff --git a/client/client.go b/client/client.go index 5dd447b..376d3d0 100644 --- a/client/client.go +++ b/client/client.go @@ -25,6 +25,18 @@ type Client struct { mappings map[string]Mapping // normalized pattern -> mapping pool *WorkerPool + // ctx/cancel own every pool dial: Close cancels it so an in-flight dial + // aborts instead of holding a goroutine for the whole handshake timeout. + // The control path uses the caller's context from Start, which is the same + // shutdown signal by convention. + ctx context.Context + cancel context.CancelFunc + + // closing is set by Close; a conn-loss teardown checks it and closes + // streams outright rather than parking them for a reattach that is never + // coming. Allocate also consults it through the pool's own flag. + closing atomic.Bool + streamWnd int // our advertised per-stream receive window (bytes) shaper *Shaper // caps aggregate egress to the hub; nil when unlimited chunk int // DATA payload cap; shrinks below DataChunkSize at low rates @@ -35,11 +47,14 @@ type Client struct { // New builds a client from config. func New(cfg *Config) *Client { + ctx, cancel := context.WithCancel(context.Background()) c := &Client{ cfg: cfg, pskBytes: []byte(cfg.PSK), pskAddr: wire.PSKAddress([]byte(cfg.PSK)), mappings: make(map[string]Mapping), + ctx: ctx, + cancel: cancel, } if _, portStr, err := net.SplitHostPort(cfg.Server); err == nil { if p, err := net.LookupPort("tcp", portStr); err == nil { @@ -99,11 +114,22 @@ type session struct { // The whole exchange is bounded by HandshakeTimeout. A hub that accepts the // socket but never answers (wedged event loop, a load balancer accepting on its // behalf) must fail fast rather than park the caller forever. -func (c *Client) dialSession(magic byte) (sess *session, err error) { - conn, err := net.DialTimeout("tcp", c.cfg.Server, HandshakeTimeout) +// +// ctx bounds the dial: the control path passes the caller's context so a +// shutdown mid-handshake aborts the attempt, and the pool passes the client's +// own context so Close cancels in-flight dials. After the dial, a cancelled +// ctx keeps aborting by closing the conn underneath the deadline-bounded +// handshake I/O. +func (c *Client) dialSession(ctx context.Context, magic byte) (sess *session, err error) { + d := &net.Dialer{Timeout: HandshakeTimeout} + conn, err := d.DialContext(ctx, "tcp", c.cfg.Server) if err != nil { return nil, err } + // When ctx ends (shutdown), close the conn so the handshake below fails + // immediately instead of waiting out its deadline. + stop := context.AfterFunc(ctx, func() { _ = conn.Close() }) + defer stop() if tcp, ok := conn.(*net.TCPConn); ok { _ = tcp.SetNoDelay(true) _ = tcp.SetKeepAlive(true) @@ -227,7 +253,7 @@ func (c *Client) Start(ctx context.Context) error { } func (c *Client) connectControl(ctx context.Context) error { - sess, err := c.dialSession(MagicControl) + sess, err := c.dialSession(ctx, MagicControl) if err != nil { return fmt.Errorf("control connect: %w", err) } @@ -271,7 +297,7 @@ func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) { c.dispatchControl(ctrl, payload) } _ = ctrl.fc.Close() - if ctx.Err() != nil { + if ctx.Err() != nil || c.closing.Load() { return } // Reconnect with backoff, but try immediately first. While the control @@ -281,8 +307,10 @@ func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) { // dial would have survived. Sleeping first spent that window unconditionally. // // The wait is on ctx rather than time.Sleep so shutdown is not held up by a - // backoff that has grown to the cap. - for backoff := time.Duration(0); ctx.Err() == nil; { + // backoff that has grown to the cap. Close() is authoritative on its own: + // it gates the loop directly because a caller may shut the client down by + // calling Close() without cancelling the context it passed to Start. + for backoff := time.Duration(0); ctx.Err() == nil && !c.closing.Load(); { if backoff > 0 { select { case <-ctx.Done(): @@ -290,6 +318,9 @@ func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) { case <-time.After(backoff): } } + if c.closing.Load() { + return + } if err := c.connectControl(ctx); err == nil { return } else { @@ -316,7 +347,14 @@ func (c *Client) dispatchControl(ctrl *ctrlSession, payload []byte) { case CtlRegisterAck: pattern, _ := r.String() status, _ := r.U8() - log.Printf("register ack %q status=%d", pattern, status) + switch status { + case RegisterOk: + log.Printf("pattern %q registered", pattern) + case RegisterErrPattern: + log.Printf("pattern %q rejected: not a valid regular expression", pattern) + default: + log.Printf("pattern %q rejected: status=%d", pattern, status) + } case CtlControlRequest: cid, err := r.Bytes(CIDLen) if err != nil { @@ -400,7 +438,7 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int) go st.writeLoop() go st.run() if err := lg.wc.sendSyn(lg.sid, cid); err != nil { - log.Printf("stream %d: SYN failed: %v", lg.sid, err) + log.Printf("stream %s: SYN failed: %v", lg, err) st.teardown(false) } } @@ -409,8 +447,12 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int) // (exposed for tests/observability). func (c *Client) WorkerConnCount() int { return c.pool.count() } -// Close tears down the control session and all worker connections. +// Close tears down the control session and all worker connections. Idempotent: +// a second call (or a Close racing a reconnect) only re-closes what is still +// open. func (c *Client) Close() { + c.closing.Store(true) + c.cancel() // aborts in-flight pool dials c.mu.Lock() fc := c.ctrl c.mu.Unlock() diff --git a/client/config.go b/client/config.go index f6ba090..34e4c7e 100644 --- a/client/config.go +++ b/client/config.go @@ -12,7 +12,12 @@ import ( // Protocol constants (mirror of the Java Protocol class; see PROTOCOL.md). const ( IntentRedapricot = 17 - ProtocolVersion = 767 // arbitrary; the hub ignores it + // IntentReserved is the handshake intent reserved for redapricot + // management/status (mirror of Protocol.INTENT_RESERVED). The hub never + // pattern-matches it; it replies with a Minecraft status line and closes, + // so an operator can probe the port without joining the protocol. + IntentReserved = 18 + ProtocolVersion = 767 // arbitrary; the hub ignores it MagicControl = 0x01 MagicWorker = 0x02 @@ -27,6 +32,10 @@ const ( CtlPing = 0x05 CtlPong = 0x06 + // RegisterAck status codes (mirror of Protocol.REGISTER_OK/_ERR_PATTERN). + RegisterOk = 0x00 + RegisterErrPattern = 0x01 // pattern is not a valid regular expression + MuxSyn = 0x00 MuxData = 0x01 MuxFin = 0x02 @@ -45,8 +54,8 @@ const ( // RST reason codes (optional trailing byte; absence means "unspecified"). // Distinguishing them matters for resume: an unknown stream is terminal, - // while "already bound" means a racing attempt won and this one must leave - // the stream alone rather than tear down a player the hub just rebound. + // while "already bound" means the hub has the stream on another conn — the + // reattach retries until that bind dies and the hub re-parks the stream. RstUnspecified = 0x00 RstUnknownStream = 0x01 RstAlreadyBound = 0x02 @@ -129,9 +138,15 @@ const ( // a session is declared dead and dropped. MissedHeartbeats = 3 + // DefaultPingIntervalMs is used when a config carries no (or a + // non-positive) interval. LoadConfig applies the same default. + DefaultPingIntervalMs = 20000 + // MinPingIntervalMs floors the configured ping interval so the derived // heartbeat timeout can never be short enough to cause spurious drops. - // Applied in LoadConfig, i.e. to configs that come from disk. + // Applied wherever the interval is read, not just on the file path: a + // hand-built Config (tests) carrying 0 would otherwise panic + // time.NewTicker at the call site. MinPingIntervalMs = 1000 // DefaultResumeGraceMs is how long a parked stream keeps trying to reattach @@ -167,9 +182,20 @@ func (c *Config) heartbeatTimeout() time.Duration { return c.pingInterval() * MissedHeartbeats } -// pingInterval is the configured heartbeat period. +// pingInterval is the configured heartbeat period, clamped at the single point +// where a duration is derived. LoadConfig also clamps on the file path; this +// covers Configs built directly (tests), where a 0 or negative PingIntervalMs +// would panic time.NewTicker — a panic, not a log line, because the interval +// feeds the heartbeat timeout too. func (c *Config) pingInterval() time.Duration { - return time.Duration(c.PingIntervalMs) * time.Millisecond + ms := c.PingIntervalMs + if ms <= 0 { + ms = DefaultPingIntervalMs + } + if ms < MinPingIntervalMs { + ms = MinPingIntervalMs + } + return time.Duration(ms) * time.Millisecond } // Mapping routes a registered pattern to a real destination. @@ -291,7 +317,7 @@ func LoadConfig(path string) (*Config, error) { c.MaxConn = 8 } if c.PingIntervalMs <= 0 { - c.PingIntervalMs = 20000 + c.PingIntervalMs = DefaultPingIntervalMs } if c.PingIntervalMs < MinPingIntervalMs { c.PingIntervalMs = MinPingIntervalMs diff --git a/client/resume.go b/client/resume.go index 585921c..0456b12 100644 --- a/client/resume.go +++ b/client/resume.go @@ -24,7 +24,7 @@ import ( var ( errResumeUnknown = errors.New("hub does not know this stream") - errResumeRaced = errors.New("another reattach already bound this stream") + errResumeRaced = errors.New("hub has this stream bound to another conn") errResumeRefused = errors.New("hub refused the reattach") errResumeTimeout = errors.New("no RESUME_ACK from the hub") errResumeConnLost = errors.New("the conn carrying the reattach died") @@ -103,11 +103,15 @@ func (s *Stream) resumeLoop(grace time.Duration) { if err == nil { return } - if errors.Is(err, errResumeRaced) { - // Another attempt owns the stream now; leaving it alone is the whole - // point — tearing down here would kill a player the hub considers live. - return - } + // errResumeRaced falls through to the retry below. The hub has this + // stream bound to a conn that is not ours — a half-open conn whose + // death the hub has not yet learned, or a bind left behind by a racing + // attempt on a now-dead conn. There is no other live attempt: park is + // the only resumeLoop starter and it refuses to double-start. Retrying + // is safe precisely because nothing else owns the stream — the foreign + // bind dies with its conn, the hub re-parks, and a later RESUME lands. + // The grace deadline bounds the loop and expiry tears the stream down, + // so a hub that never re-parks cannot hang us forever. if errors.Is(err, errResumeUnknown) || errors.Is(err, errResumeTooOld) { // Terminal: the hub has no state for this stream (it restarted, the // grace expired, or a load balancer sent us to a different instance). @@ -255,11 +259,6 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error s.parked = false s.resumeWait = nil owedFin := s.finToHub - if s.stats != nil { - s.stats.resumes++ - s.stats.hung += time.Since(s.parkedAt) - s.stats.replayBytes += replayed - } s.cond.Broadcast() // release acquireSendWnd and any parked writer s.mu.Unlock() @@ -269,6 +268,17 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error n = s.client.chunk } if err := wc.sendData(sid, replay[:n]); err != nil { + // The conn died mid-replay. The stream is still resumable, but not + // from this leg — put it back in the parked state before returning + // so the conn's teardown takes park()'s already-branch instead of + // starting a second resumeLoop. The loop we came from keeps + // retrying with the fresh CID, which the hub re-parked alongside + // the stream when this conn died. Without this re-park the flag + // cleared above would let two loops race one stream, and a racing + // RST would then strand it with neither loop alive. + s.mu.Lock() + s.parked = true + s.mu.Unlock() s.sendMu.Unlock() return err } @@ -276,6 +286,17 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error } s.sendMu.Unlock() + // Counted only once the replay actually landed: a failed reattach above + // returns before this, so a conn dying mid-replay does not inflate the + // resume count with an attempt that never completed. + if s.stats != nil { + s.mu.Lock() + s.stats.resumes++ + s.stats.hung += time.Since(s.parkedAt) + s.stats.replayBytes += replayed + s.mu.Unlock() + } + // A destination that closed while we were parked owed the hub a FIN that had // nowhere to go at the time. if owedFin { @@ -283,7 +304,7 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error s.teardown(false) return nil } - log.Printf("stream %d resumed (%d bytes replayed, %d outstanding)", sid, replayed, outstanding) + log.Printf("stream conn%d/sid%d resumed (%d bytes replayed, %d outstanding)", wc.id, sid, replayed, outstanding) return nil } diff --git a/client/worker.go b/client/worker.go index c183f3f..4356976 100644 --- a/client/worker.go +++ b/client/worker.go @@ -1,6 +1,8 @@ package client import ( + "errors" + "fmt" "log" "net" "sync" @@ -10,6 +12,11 @@ import ( "github.com/iceBear67/redapricot/client/wire" ) +// errPoolClosed is returned by Allocate after Close: the pool is shutting down +// and must not start new dials, so a caller (handleControlRequest, +// allocateForResume) gives up rather than wait on a cond no one will satisfy. +var errPoolClosed = errors.New("worker pool closed") + // StreamsBeforeGrowing is how many streams a worker conn may carry before the // pool starts opening another one. Set to 1 so the pool fans out to maxConn // under load *before* stacking streams: concentrating every player on a single @@ -37,6 +44,7 @@ type WorkerPool struct { dialing int // dials currently in flight (foreground + background) dialGen uint64 dialErr error // most recent dial failure + closed bool // closeAll ran; no new conns may join the pool } func newWorkerPool(c *Client, maxConn int) *WorkerPool { @@ -56,6 +64,13 @@ func newWorkerPool(c *Client, maxConn int) *WorkerPool { func (p *WorkerPool) Allocate() (*WorkerConn, int, error) { p.mu.Lock() for { + if p.closed { + // Close won. No new conn may join the pool, so no stream may be + // placed; waiting on the cond could only be satisfied by a dial we + // must not start. + p.mu.Unlock() + return nil, 0, errPoolClosed + } best, bestCount := p.leastLoadedLocked() if best != nil { p.maybeGrowLocked(bestCount) @@ -86,6 +101,15 @@ func (p *WorkerPool) Allocate() (*WorkerConn, int, error) { p.mu.Unlock() return nil, 0, err } + if p.closed { + // Close raced this dial: the conn must not enter the pool. Closing + // it here, under p.mu, is a raw socket close — fine, and it makes + // the shutdown atomic with the pool state. + p.cond.Broadcast() + p.mu.Unlock() + _ = wc.fc.Close() + return nil, 0, errPoolClosed + } p.conns = append(p.conns, wc) p.cond.Broadcast() } @@ -113,6 +137,9 @@ func (p *WorkerPool) maybeGrowLocked(bestCount int) { if bestCount < StreamsBeforeGrowing { return } + if p.closed { + return // shutdown; do not start dials nobody will join the pool + } if len(p.conns)+p.dialing >= p.maxConn { if bestCount > SaturationThreshold { log.Printf("worker pool at maxConn=%d with %d streams on the least-loaded conn", p.maxConn, bestCount) @@ -129,6 +156,8 @@ func (p *WorkerPool) maybeGrowLocked(bestCount int) { p.dialErr = err switch { case err != nil: + case p.closed: + surplus = wc // Close raced this background dial case len(p.conns) < p.maxConn: p.conns = append(p.conns, wc) default: @@ -146,8 +175,9 @@ func (p *WorkerPool) maybeGrowLocked(bestCount int) { } // dialWorker establishes one worker conn. It must be called without p.mu held. +// The dial runs on the client's context so Close aborts it mid-handshake. func (p *WorkerPool) dialWorker() (*WorkerConn, error) { - sess, err := p.client.dialSession(MagicWorker) + sess, err := p.client.dialSession(p.client.ctx, MagicWorker) if err != nil { return nil, err } @@ -200,7 +230,11 @@ func (p *WorkerPool) remove(wc *WorkerConn) { func (p *WorkerPool) closeAll() { p.mu.Lock() + p.closed = true conns := append([]*WorkerConn(nil), p.conns...) + // Wake waiters parked in Allocate: the closed flag they re-check is the + // only thing that can release them now that no conn will ever join. + p.cond.Broadcast() p.mu.Unlock() for _, wc := range conns { _ = wc.fc.Close() @@ -401,6 +435,14 @@ func (wc *WorkerConn) readLoop() { // Only the tunnel leg died. Where the session negotiated resumption the // destination sockets are kept open and each stream reattaches over a fresh // conn (§7.5); otherwise this is the old, unconditional teardown. + // During Close there is no reattach to come: a stream that parked now would + // hold its destination socket open past shutdown, so teardown instead. + if wc.pool.client.closing.Load() { + for _, st := range streams { + st.teardown(false) + } + return + } for _, st := range streams { if !st.park(wc.grace) { st.teardown(false) @@ -445,6 +487,11 @@ type leg struct { sid int } +// String formats one (conn, sid) snapshot for log correlation. Stream ids +// restart at 1 per conn, so a bare sid cannot be traced across a reattach — +// the conn id is what ties the log lines together. +func (lg *leg) String() string { return fmt.Sprintf("conn%d/sid%d", lg.wc.id, lg.sid) } + // Stream bridges one player (via the hub) to one destination connection. // // Data from the hub is queued and written to the destination by a dedicated @@ -556,7 +603,7 @@ func (s *Stream) run() { dest, err := net.DialTimeout("tcp", s.mapping.Destination, 10*time.Second) if err != nil { lg := s.conn() - log.Printf("stream %d: dial %s failed: %v", lg.sid, s.mapping.Destination, err) + log.Printf("stream %s: dial %s failed: %v", lg, s.mapping.Destination, err) lg.wc.removeStream(lg.sid) lg.wc.sendRst(lg.sid) s.teardown(false) @@ -569,7 +616,7 @@ func (s *Stream) run() { if s.mapping.ProxyProtocol { if hdr := s.buildProxyHeader(dest); hdr != nil { if _, err := dest.Write(hdr); err != nil { - log.Printf("stream %d: proxy header write: %v", s.conn().sid, err) + log.Printf("stream %s: proxy header write: %v", s.conn(), err) } } } @@ -637,6 +684,10 @@ func (s *Stream) sendToHub(data []byte) bool { if !s.client.shaper.Acquire(&s.share, n, s.done) { return false } + // Sampled here, before the socket write: emit's write to the hub can + // block when the hub is not reading, and charging that time to the + // bandwidth cap would blame maxBandwidth for a hub that is not draining. + waited := shaperStall.elapsed() if !s.emit(data[:n]) { return false } @@ -645,7 +696,7 @@ func (s *Stream) sendToHub(data []byte) bool { // Separated from the window stall on purpose: this one says the // configured cap is the binding constraint, and raising maxBandwidth // is the fix. The window stall says the opposite. - s.stats.shaperStall += shaperStall.elapsed() + s.stats.shaperStall += waited s.stats.bytesUp += int64(n) s.mu.Unlock() } @@ -754,7 +805,7 @@ func (s *Stream) deliverFromHub(data []byte) { if s.qBytes+len(data) > s.client.streamWnd { s.mu.Unlock() lg := s.conn() - log.Printf("stream %d: peer exceeded flow-control window; resetting", lg.sid) + log.Printf("stream %s: peer exceeded flow-control window; resetting", lg) lg.wc.removeStream(lg.sid) lg.wc.sendRst(lg.sid) s.teardown(false) diff --git a/server/src/main/java/io/icybear/redapricot/HubConnection.java b/server/src/main/java/io/icybear/redapricot/HubConnection.java index 78de932..1113a15 100644 --- a/server/src/main/java/io/icybear/redapricot/HubConnection.java +++ b/server/src/main/java/io/icybear/redapricot/HubConnection.java @@ -103,17 +103,45 @@ public final class HubConnection { if (intent == Protocol.INTENT_REDAPRICOT) { beginRedapricot(address, afterHandshake); } else if (intent == Protocol.INTENT_RESERVED) { - LOG.info("{} reserved intent 18; closing", id); - socket.close(); + LOG.info("{} reserved intent 18; replying with a status line and closing", id); + sendStatusLine(); } else { handlePlayer(address); } } + // ---- Intent 18 (reserved: management/status) ---- + + /** + * Reply to an Intent-18 probe with a Minecraft status-response packet + * ({@code [Len: VarInt][0x00][JSON: String]}), the same shape a player gets + * for a status query (Intent 1), then close. This lets an operator probe + * the public port with ordinary tooling without joining the protocol, and + * it is the one reply the hub sends in plaintext — Intent 18 never + * negotiates encryption. + */ + private void sendStatusLine() { + String json = "{\"description\":{\"text\":\"redapricot hub\"}," + + "\"version\":{\"name\":\"redapricot\",\"protocol\":767}," + + "\"players\":{\"max\":0,\"online\":0}}"; + ProtoWriter body = new ProtoWriter().u8(0x00).string(json); + byte[] bodyBytes = body.toBytes(); + ProtoWriter pkt = new ProtoWriter().varInt(bodyBytes.length).bytes(bodyBytes); + // end() writes the reply and closes after it lands, so the packet is + // never cut short by the close racing the flush. + socket.end(Buffer.buffer(pkt.toBytes())); + } + // ---- redapricot session (Intent 17) ---- private void beginRedapricot(String address, Buffer afterHandshake) { - if (!address.equalsIgnoreCase(hub.pskAddress)) { + // Strict equality, per PROTOCOL.md §2: the contract is exactly + // lowercase_hex(SHA3-224(PSK)), and the client always sends that. An + // uppercase or otherwise case-folded variant is not a valid session — + // accepting it would widen the acceptance surface beyond what the spec + // promises (and what the client ever produces), which is exactly the + // kind of "close enough" check that hides a wrong-PSK probe. + if (!address.equals(hub.pskAddress)) { LOG.warn("{} bad PSK address; closing", id); socket.close(); return;