qing/blackbird
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(libmsg): security audit round 4 - eliminate residual race + atomic server_fd

Browse Source 
- P1-3: introduce ConnectionLock (mutex + atomic active) to close the residual
  CWE-362 race where _remove_client closes fd before reply/broadcast acquires
  the lock; senders now check active after locking and abort if fd was closed
- P2-7: make _server_fd std::atomic<int> to remove the data-race UB between
  stop() and _event_loop
- P2-8: validate on_message is non-null in MsgServer::start()
- P3-2: distinguish missing _route vs wrong-typed _route in RpcServer
- P3-3: set TCP_NODELAY/SO_SNDTIMEO before registering fd in _clients
- doc: add SO_SNDTIMEO precondition note to pmsg_send
This commit is contained in:
qing committed 2026-08-07 21:27:05 +08:00
1 parent 63c22a3d60
commit e2f05f9a43
4 files changed
+240 -82
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/security/libmsg_SECURITY_AUDIT_20260806.md
+142 -22
View File
@@ -1,6 +1,6 @@
# libmsg 模块安全审计报告 # libmsg 模块安全审计报告
**审计日期**: 2026-08-07（第三轮审计） **审计日期**: 2026-08-07（第四轮审计）
**首次审计**: 2026-08-06 **首次审计**: 2026-08-06
**审计范围**: `libmsg/inc/us/MsgServer.hpp`、`libmsg/inc/us/RpcServer.hpp`、`libmsg/inc/us/Msg.hpp`、`libmsg/inc/us/RpcClient.hpp`、`libmsg/inc/us/AsyncQueue.hpp`、`libmsg/inc/us/Msg_Basic.hpp`、`libmsg/src/Msg_Basic.cpp` **审计范围**: `libmsg/inc/us/MsgServer.hpp`、`libmsg/inc/us/RpcServer.hpp`、`libmsg/inc/us/Msg.hpp`、`libmsg/inc/us/RpcClient.hpp`、`libmsg/inc/us/AsyncQueue.hpp`、`libmsg/inc/us/Msg_Basic.hpp`、`libmsg/src/Msg_Basic.cpp`
**代码版本**: main 分支（基于 e92ef05 进行修复） **代码版本**: main 分支（基于 e92ef05 进行修复）
@@ -26,12 +26,15 @@
| 第一轮 (2026-08-06) | P0 | 1 | 1 | 0 | 100% | | 第一轮 (2026-08-06) | P0 | 1 | 1 | 0 | 100% |
| 第一轮 (2026-08-06) | P1 | 1 | 1 | 0 | 100% | | 第一轮 (2026-08-06) | P1 | 1 | 1 | 0 | 100% |
| 第一轮 (2026-08-06) | P2 | 1 | 1 | 0 | 100% | | 第一轮 (2026-08-06) | P2 | 1 | 1 | 0 | 100% |
| **第二轮 (2026-08-07)** | **P0** | **1** | **1** | **0** | **100%** | | 第二轮 (2026-08-07) | P0 | 1 | 1 | 0 | 100% |
| **第二轮 (2026-08-07)** | **P2** | **4** | **4** | **0** | **100%** | | 第二轮 (2026-08-07) | P2 | 4 | 4 | 0 | 100% |
| **第三轮 (2026-08-07)** | **P1** | **1** | **1** | **0** | **100%** | | 第三轮 (2026-08-07) | P1 | 1 | 1 | 0 | 100% |
| **第三轮 (2026-08-07)** | **P2** | **1** | **1** | **0** | **100%** | | 第三轮 (2026-08-07) | P2 | 1 | 1 | 0 | 100% |
| **第三轮 (2026-08-07)** | **P3** | **1** | **1** | **0** | **100%** | | 第三轮 (2026-08-07) | P3 | 1 | 1 | 0 | 100% |
| **合计** | **—** | **11** | **11** | **0** | **100%** | | **第四轮 (2026-08-07)** | **P1** | **1** | **1** | **0** | **100%** |
| **第四轮 (2026-08-07)** | **P2** | **2** | **2** | **0** | **100%** |
| **第四轮 (2026-08-07)** | **P3** | **2** | **2** | **0** | **100%** |
| **合计** | **—** | **16** | **16** | **0** | **100%** |
--- ---
@@ -271,21 +274,124 @@ Python 版 `pymsg.recv()` 在接收载荷后调用 `payload.decode("utf-8")` 进
--- ---
## 🟠 P1 — 高危问题（第四轮）
### P1-3: reply/broadcast 与 _remove_client 之间存在残余竞态，fd 复用可致数据发往错误客户端 ✅ 已修复（第四轮）
- **CWE**: CWE-362 (Race Condition)
- **OWASP**: A01:2021
- **位置**: `MsgServer.hpp` `reply()` / `broadcast()` / `_remove_client()`
- **置信度**: 85%
**问题描述**:
第一轮 P0-1 修复确保了"发送期间不会被 close"（reply/broadcast 先获取 send_lock → _remove_client 等待），但未覆盖反向场景：**_remove_client 先获取 send_lock 并 close(fd) 后，reply/broadcast 后获取 send_lock 时 fd 已关闭**。如果 OS 将该 fd 复用给新 accept 的连接，reply/broadcast 的 `pmsg_send()` 会向复用的 fd 写入数据，导致消息到达错误客户端。
**触发场景**:
1. reply/broadcast 在 `_clients_mtx` 下获取了 `send_lock` 的 shared_ptr 副本
2. 释放 `_clients_mtx`
3. `_remove_client()` 获取 `_clients_mtx`，erase fd，获取 `send_lock` 副本，释放 `_clients_mtx`
4. `_remove_client()` 先于 reply/broadcast 获取 `send_lock`
5. `_remove_client()` `close(fd)`，释放 `send_lock`
6. OS 将 fd 复用给新 accept 的连接
7. reply/broadcast 获取 `send_lock`，调用 `pmsg_send(fd, msg)` → 数据到达新连接
步骤 2-4 的窗口期虽然很短（仅几条指令），但在高并发场景（大量 broadcast + 连接断开同时发生）下可被触发。
**修复方案**（第四轮修复）:
引入 `ConnectionLock` 结构，将 `send_lock`（mutex）和 `active`（atomic\<bool\>）组合在一起：
```cpp
struct ConnectionLock {
std::mutex send_mtx;
std::atomic<bool> active{true};
};
```
- `reply()`/`broadcast()` 在获取 `send_mtx` 后检查 `active`：若为 false 则放弃发送（fd 已被关闭）
- `_remove_client()`/`_close_all_clients()` 在获取 `send_mtx` 后先设置 `active=false`，再 `shutdown()`+`close(fd)`
这样无论谁先获取锁：
- 若 reply/broadcast 先获取锁：`active=true`，正常发送，`_remove_client` 等待发送完成后再 close
- 若 `_remove_client` 先获取锁：设置 `active=false`，close(fd)，reply/broadcast 后获取锁时检查 `active=false` 直接 return
---
## 🟡 P2 — 中危问题（第四轮）
### P2-7: _server_fd 非原子访问构成数据竞争（UB） ✅ 已修复（第四轮）
- **CWE**: CWE-362 (Race Condition) / CWE-662 (Improper Synchronization)
- **位置**: `MsgServer.hpp` `_server_fd` 成员
- **置信度**: 90%
**问题描述**:
`_server_fd` 是普通 `int`，在主线程（`_event_loop`）读、`stop()` 线程写，构成 C++ 标准定义的数据竞争（Undefined Behavior）。虽然在实际 64 位系统上 int 读写通常是原子的（最坏情况读取旧值导致 accept 失败），但严格来说是 UB，编译器优化可能产生意外行为。
**修复方案**（第四轮修复）:
将 `_server_fd` 改为 `std::atomic<int>`，所有读写使用 `.load()` / `.store()` / `.exchange()`。`stop()` 使用 `exchange(-1)` 原子地认领 fd 并关闭。
---
### P2-8: start() 未校验 on_message 回调非空 ✅ 已修复（第四轮）
- **CWE**: CWE-20 (Improper Input Validation)
- **位置**: `MsgServer.hpp` `start()`
- **置信度**: 85%
**问题描述**:
`start(on_message)` 接受 `std::function` 参数但未校验是否为空。若传入空回调（默认构造的 `std::function`），后续 `_on_message(fd, addr, msg)` 会抛出 `std::bad_function_call`，该异常在 worker 线程中被 catch 但在同步模式下会导致连接断开；在 `_dispatch`（RpcServer）中会导致单个请求失败。虽然不会崩溃服务，但行为不符合预期且错误信息不明确。
**修复方案**（第四轮修复）:
`start()` 开头添加校验：`if (!on_message) throw MsgException("on_message callback must not be null");`
---
## 🟢 P3 — 优化建议（第四轮）
### P3-2: RpcServer _route 非 string 类型时错误信息不准确 ✅ 已修复（第四轮）
- **CWE**: CWE-209 (Information Exposure Through an Error Message)
- **位置**: `RpcServer.hpp` `_dispatch()`
- **置信度**: 70%
**问题描述**:
`_dispatch()` 中检查 `_route` 字段时，若字段存在但不是 string 类型（如数字、数组），错误信息统一返回 `"missing _route"`，误导客户端以为字段缺失而非类型错误。
**修复方案**（第四轮修复）:
区分两种情况：字段不存在返回 `"missing _route"`，类型不符返回 `"_route must be a string"`。
### P3-3: _accept_client 中 setsockopt 在加入 _clients 之后执行 ✅ 已修复（第四轮）
- **位置**: `MsgServer.hpp` `_accept_client()`
- **置信度**: 60%
**问题描述**:
原实现先将 `client_fd` 加入 `_clients` 和 `_send_locks`（持锁），释放锁后再设置 `TCP_NODELAY` 和 `SO_SNDTIMEO`。虽然新连接此时尚未收到消息，不会有 reply/broadcast 发送数据，但最佳实践是先完成所有 socket 配置再暴露给其他线程。
**修复方案**（第四轮修复）:
调整顺序：先设置 `TCP_NODELAY` 和 `SO_SNDTIMEO`，再持锁加入 `_clients`。
---
## ✅ 已有安全机制（更新后） ## ✅ 已有安全机制（更新后）
| 机制 | 位置 | 说明 | | 机制 | 位置 | 说明 |
|------|------|------| |------|------|------|
| 消息大小上限 64 MiB | `Msg_Basic.hpp` `kMaxMsgSize` | 防止超大消息耗尽内存 | | 消息大小上限 64 MiB | `Msg_Basic.hpp` `kMaxMsgSize` | 防止超大消息耗尽内存 |
| deadline 模式 recv 超时 | `Msg.hpp` `recvFrameDeadline` | header+payload 共享总超时，防 Slowloris | | deadline 模式 recv 超时 | `Msg.hpp` `recvFrameDeadline` | header+payload 共享总超时，防 Slowloris |
| **新增**: 入口参数边界校验 | `Msg.hpp/MsgServer.hpp/AsyncQueue.hpp/Rpc*.hpp` | timeout/port/配置项全范围校验，拒绝非法输入 | | 入口参数边界校验 | `Msg.hpp/MsgServer.hpp/AsyncQueue.hpp/Rpc*.hpp` | timeout/port/配置项全范围校验，拒绝非法输入 |
| **新增**: UTF-8 载荷校验 | `Msg.hpp` `isValidUtf8()` | 与 Python decode('utf-8') 对齐，拒绝非法字节序列 | | UTF-8 载荷校验 | `Msg.hpp` `isValidUtf8()` | 与 Python decode('utf-8') 对齐，拒绝非法字节序列 |
| 最大连接数限制 | `MsgServer.hpp` `max_connections=1000` | 超限拒绝新连接 | | 最大连接数限制 | `MsgServer.hpp` `max_connections=1000` | 超限拒绝新连接 |
| 有界任务队列 | `MsgServer.hpp` `max_queue_size=1000` | 队列满时断开连接，防 OOM | | 有界任务队列 | `MsgServer.hpp` `max_queue_size=1000` | 队列满时断开连接，防 OOM |
| per-connection 发送锁 | `MsgServer.hpp` `_send_locks` | reply/broadcast 线程安全，防帧交错 | | per-connection 发送锁 + active 标志 | `MsgServer.hpp` `ConnectionLock` | reply/broadcast 线程安全防帧交错 + active 检查防 fd 复用后发送 |
| **新增**: 服务端发送超时 SO_SNDTIMEO | `MsgServer.hpp` `_accept_client()` | 慢客户端不读数据时 send 超时抛异常，防拒绝服务（P1-2） | | 服务端发送超时 SO_SNDTIMEO | `MsgServer.hpp` `_accept_client()` | 慢客户端不读数据时 send 超时抛异常，防拒绝服务（P1-2） |
| **修复**: reply() 失败也标记 failed | `MsgServer.hpp` `reply()` | 与 broadcast 一致，失败连接由主循环清理（P2-6） | | reply() 失败标记 failed | `MsgServer.hpp` `reply()` | 与 broadcast 一致，失败连接由主循环清理（P2-6） |
| **修复**: RpcClient success 类型校验 | `RpcClient.hpp` `call()` | 非 bool 时抛 RpcError，而非未捕获的 boost 异常（P3-1） | | RpcClient success 类型校验 | `RpcClient.hpp` `call()` | 非 bool 时抛 RpcError，而非未捕获的 boost 异常（P3-1） |
| **修复**: _close_all_clients 批量关闭也持锁 | `MsgServer.hpp` `_close_all_clients()` | 修复批量关闭路径的竞态（P0-2） | | _close_all_clients 批量关闭也持锁 | `MsgServer.hpp` `_close_all_clients()` | 修复批量关闭路径的竞态（P0-2） |
| **新增**: ConnectionLock active 标志 | `MsgServer.hpp` `ConnectionLock::active` | reply/broadcast 获取锁后检查 active，彻底消除残余竞态（P1-3） |
| **新增**: _server_fd atomic 访问 | `MsgServer.hpp` `std::atomic<int> _server_fd` | 消除 stop() 与 _event_loop 的数据竞争 UB（P2-7） |
| **新增**: start() 空回调校验 | `MsgServer.hpp` `start()` | 防止 bad_function_call 异常（P2-8） |
| **新增**: _accept_client 先配置后注册 | `MsgServer.hpp` `_accept_client()` | socket 选项在加入 _clients 前设置完整（P3-3） |
| 同一连接消息串行 | `MsgServer.hpp` `_busy` 集合 | 回调执行中暂停该连接 select 监听 | | 同一连接消息串行 | `MsgServer.hpp` `_busy` 集合 | 回调执行中暂停该连接 select 监听 |
| JSON 解析深度限制 | `RpcServer.hpp` `opts.max_depth=32` | 防深度嵌套 JSON 导致栈溢出 | | JSON 解析深度限制 | `RpcServer.hpp` `opts.max_depth=32` | 防深度嵌套 JSON 导致栈溢出 |
| 异常信息脱敏 | `RpcServer.hpp` `_sanitize_error()` | 过滤绝对路径，完整异常由 on_error 钩子记录 | | 异常信息脱敏 | `RpcServer.hpp` `_sanitize_error()` | 过滤绝对路径，完整异常由 on_error 钩子记录 |
@@ -312,11 +418,24 @@ Python 版 `pymsg.recv()` 在接收载荷后调用 `payload.decode("utf-8")` 进
| L-7 | AsyncQueue 关闭时 flush 失败数据只记 on_error | P3 | _flush_all() 中 handler 失败后数据不重试，符合非关键路径设计，但应确保 on_error 有持久化/告警机制。 | | L-7 | AsyncQueue 关闭时 flush 失败数据只记 on_error | P3 | _flush_all() 中 handler 失败后数据不重试，符合非关键路径设计，但应确保 on_error 有持久化/告警机制。 |
| L-8 | 半连接（只发部分帧头）仍可占用一个 worker 线程 recv_timeout 秒 | P2 | workers>0 时消息接收在 worker 线程完成，慢连接可占用 worker 最多 recv_timeout（默认10秒）。数量受 max_connections 与 recv_timeout 有界约束，已缓解。可用更小 recv_timeout 或独立接收线程池进一步收紧。 | | L-8 | 半连接（只发部分帧头）仍可占用一个 worker 线程 recv_timeout 秒 | P2 | workers>0 时消息接收在 worker 线程完成，慢连接可占用 worker 最多 recv_timeout（默认10秒）。数量受 max_connections 与 recv_timeout 有界约束，已缓解。可用更小 recv_timeout 或独立接收线程池进一步收紧。 |
| L-9 | pmsg_send 发送端不校验 UTF-8，与 Python 版 encode('utf-8') 不对称 | P3 | 接收端已在 pmsg_recv 校验 UTF-8（P2-4），非法载荷会被远端断开。发送端未校验，若业务传入二进制可能被远端拒绝。若存在纯二进制传输场景，发送端校验会破坏兼容，故暂不强制，仅记录。 | | L-9 | pmsg_send 发送端不校验 UTF-8，与 Python 版 encode('utf-8') 不对称 | P3 | 接收端已在 pmsg_recv 校验 UTF-8（P2-4），非法载荷会被远端断开。发送端未校验，若业务传入二进制可能被远端拒绝。若存在纯二进制传输场景，发送端校验会破坏兼容，故暂不强制，仅记录。 |
| L-10 | pmsg_send 公共 API 依赖 SO_SNDTIMEO | P3 | sendAll() 在 send() 返回 EAGAIN 时抛超时异常，但前提是 socket 已设置 SO_SNDTIMEO。已添加文档警告，MsgClient 和 MsgServer 内部均已设置。直接使用 pmsg_send 的外部调用方需自行设置。 |
| L-11 | MsgServer 未设置 TCP keepalive | P3 | 客户端崩溃（不发送 FIN）时连接会残留直到 select 检测异常或 send 失败。建议生产环境设置 SO_KEEPALIVE 以更快检测死连接。 |
| L-12 | _remove_client 同步模式下无总超时限制 | P3 | 同步模式下 _handle_client 中 pmsg_recv 有 recv_timeout 限制，但回调执行无超时。慢回调会阻塞事件循环。建议将慢操作移至线程池模式或业务层自行限时。 |
--- ---
## 变更记录 ## 变更记录
### 2026-08-07（第四轮审计修复）
**代码变更文件**:
| 文件 | 变更内容 |
|------|----------|
| [MsgServer.hpp](file:///Users/qingsong/Documents/blackbird/libmsg/inc/us/MsgServer.hpp) | **P1修复**: 引入 `ConnectionLock` 结构（mutex + atomic active），`reply()`/`broadcast()` 获取锁后检查 `active`，`_remove_client()`/`_close_all_clients()` close 前设置 `active=false`，彻底消除残余竞态（P1-3）；**P2修复**: `_server_fd` 改为 `std::atomic<int>`，所有访问使用 `.load()`/`.store()`/`.exchange()`（P2-7）；**P2修复**: `start()` 校验 `on_message` 非空（P2-8）；**P3修复**: `_accept_client()` 中 setsockopt 移到加入 `_clients` 之前（P3-3） |
| [RpcServer.hpp](file:///Users/qingsong/Documents/blackbird/libmsg/inc/us/RpcServer.hpp) | **P3修复**: `_dispatch()` 区分 `_route` 字段缺失和类型不符的错误信息（P3-2） |
| [Msg.hpp](file:///Users/qingsong/Documents/blackbird/libmsg/inc/us/Msg.hpp) | **文档**: `pmsg_send` 添加 SO_SNDTIMEO 前置条件警告（L-10） |
### 2026-08-07（第三轮审计修复） ### 2026-08-07（第三轮审计修复）
**代码变更文件**: **代码变更文件**:
@@ -342,18 +461,19 @@ Python 版 `pymsg.recv()` 在接收载荷后调用 `payload.decode("utf-8")` 进
## 审计结论 ## 审计结论
第四轮审计（含第三轮新发现）累计发现 **11 个问题，全部已修复（100%）**。 四轮审计累计发现 **16 个问题，全部已修复（100%）**。
- **第一轮（2026-08-06）**: 3 个问题（P0 竞态、P1 Slowloris、P2 脱敏正则）已修复。 - **第一轮（2026-08-06）**: 3 个问题（P0 竞态、P1 Slowloris、P2 脱敏正则）已修复。
- **第二轮（2026-08-07）**: 5 个问题（P0 批量关闭竞态回归、4 个 P2 健壮性）已修复。 - **第二轮（2026-08-07）**: 5 个问题（P0 批量关闭竞态回归、4 个 P2 健壮性）已修复。
- **第三轮（2026-08-07）**: 3 个新问题（P1 服务端发送超时缺失、P2 reply 失败未清理、P3 RpcClient success 类型校验）已修复。 - **第三轮（2026-08-07）**: 3 个新问题（P1 服务端发送超时缺失、P2 reply 失败未清理、P3 RpcClient success 类型校验）已修复。
- **第四轮（2026-08-07）**: 5 个新问题（P1 reply/broadcast 残余竞态、P2 _server_fd 数据竞争、P2 start 空回调校验、P3 _route 错误信息、P3 setsockopt 顺序）已修复。
结合三轮审计结果，libmsg 核心通信层的安全状况良好： 结合四轮审计结果，libmsg 核心通信层的安全状况良好：
- **内存安全**: 无缓冲区溢出、无 use-after-free（fd 生命周期通过 mutex + shared_ptr 严格管理） - **内存安全**: 无缓冲区溢出、无 use-after-free（fd 生命周期通过 ConnectionLock + shared_ptr + atomic active 严格管理）
- **DoS 防护**: 64MiB 消息上限、deadline 模式接收超时（防Slowloris）、**服务端发送超时 SO_SNDTIMEO（防慢客户端耗尽资源）**、max_connections 连接数限制、max_queue_size 队列上限、poll_timeout 防忙轮询、全量参数边界校验 - **DoS 防护**: 64MiB 消息上限、deadline 模式接收超时（防Slowloris）、服务端发送超时 SO_SNDTIMEO（防慢客户端耗尽资源）、max_connections 连接数限制、max_queue_size 队列上限、poll_timeout 防忙轮询、全量参数边界校验
- **竞态安全**: 所有 fd 关闭路径均持有 per-connection send_lock，彻底消除 CWE-362 竞态 - **竞态安全**: 所有 fd 关闭路径均持有 per-connection send_mtx 并设置 active=false，reply/broadcast 获取锁后检查 active，**彻底消除 CWE-362 竞态（含正向和反向场景）**；_server_fd 使用 atomic 消除数据竞争 UB
- **协议健壮性**: UTF-8 收发端校验与 Python 版对齐、RpcClient 响应字段类型校验统一抛 RpcError - **协议健壮性**: UTF-8 收发端校验与 Python 版对齐、RpcClient 响应字段类型校验统一抛 RpcError、RpcServer _route 类型校验
- **异常安全**: worker 线程异常隔离、AsyncQueue fail-safe、Handler 异常不崩溃服务、RPC 错误脱敏、reply/broadcast 失败连接统一清理 - **异常安全**: worker 线程异常隔离、AsyncQueue fail-safe、Handler 异常不崩溃服务、RPC 错误脱敏、reply/broadcast 失败连接统一清理、start 空回调校验
遗留的 L-1~L-9 项为架构层面的已知限制/设计决策，非代码缺陷，可根据实际部署需求决定是否增强。 遗留的 L-1~L-12 项为架构层面的已知限制/设计决策，非代码缺陷，可根据实际部署需求决定是否增强。
libmsg/inc/us/Msg.hpp
+3
View File
@@ -286,6 +286,9 @@ inline int openConnection(const std::string& addr, int port, double timeout) {
// ============================================================================ // ============================================================================
/// 以 pymsg 协议发送一条字符串消息（长度前缀 + UTF-8 载荷）。 /// 以 pymsg 协议发送一条字符串消息（长度前缀 + UTF-8 载荷）。
/// 注意：sendAll() 依赖 socket 已设置的 SO_SNDTIMEO，否则慢客户端不读数据时
/// send() 会永久阻塞。MsgClient 和 MsgServer 内部均已设置 SO_SNDTIMEO，
/// 直接使用此公共 API 时需自行设置。
inline void pmsg_send(int sockfd, const std::string &data) { inline void pmsg_send(int sockfd, const std::string &data) {
size_t payload_size = data.size(); size_t payload_size = data.size();
if (payload_size > msg_constants::kMaxMsgSize) { if (payload_size > msg_constants::kMaxMsgSize) {
libmsg/inc/us/MsgServer.hpp
+91 -59
View File
@@ -75,6 +75,20 @@ struct MsgServerConfig {
double poll_timeout = 0.1; ///< select 轮询超时（秒），影响关闭响应速度 double poll_timeout = 0.1; ///< select 轮询超时（秒），影响关闭响应速度
}; };
// ============================================================================
// ConnectionLock — 每连接发送锁 + 活跃标志
// ============================================================================
/// 每连接的发送锁和活跃标志。
///
/// active 标志用于防止 CWE-362 竞态：当 _remove_client() 先获取 send_mtx
/// 并 close(fd) 后，reply/broadcast 后获取 send_mtx 时通过 active==false
/// 得知 fd 已关闭，立即放弃发送，避免向 OS 复用的 fd 发送数据。
struct ConnectionLock {
std::mutex send_mtx;
std::atomic<bool> active{true};
};
// ============================================================================ // ============================================================================
// MsgServer // MsgServer
// ============================================================================ // ============================================================================
@@ -122,38 +136,41 @@ public:
* 回调内可安全调用 reply()/broadcast()/stop()。 * 回调内可安全调用 reply()/broadcast()/stop()。
*/ */
void start(MessageCallback on_message) { void start(MessageCallback on_message) {
if (!on_message) {
throw MsgException("MsgServer::start(): on_message callback must not be null");
}
_on_message = std::move(on_message); _on_message = std::move(on_message);
_running = true; _running = true;
// 创建监听 socket // 创建监听 socket
_server_fd = ::socket(AF_INET, SOCK_STREAM, 0); _server_fd.store(::socket(AF_INET, SOCK_STREAM, 0));
if (_server_fd < 0) { if (_server_fd.load() < 0) {
throw MsgConnectionException(std::string("socket() failed: ") + ::strerror(errno)); throw MsgConnectionException(std::string("socket() failed: ") + ::strerror(errno));
} }
int optval = 1; int optval = 1;
::setsockopt(_server_fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)); ::setsockopt(_server_fd.load(), SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
struct sockaddr_in serv_addr {}; struct sockaddr_in serv_addr {};
serv_addr.sin_family = AF_INET; serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(static_cast<uint16_t>(_port)); serv_addr.sin_port = htons(static_cast<uint16_t>(_port));
if (::inet_pton(AF_INET, _addr.c_str(), &serv_addr.sin_addr) <= 0) { if (::inet_pton(AF_INET, _addr.c_str(), &serv_addr.sin_addr) <= 0) {
::close(_server_fd); int fd = _server_fd.exchange(-1);
_server_fd = -1; if (fd >= 0) ::close(fd);
throw MsgConnectionException("invalid bind address: " + _addr); throw MsgConnectionException("invalid bind address: " + _addr);
} }
if (::bind(_server_fd, reinterpret_cast<struct sockaddr*>(&serv_addr), sizeof(serv_addr)) < 0) { if (::bind(_server_fd.load(), reinterpret_cast<struct sockaddr*>(&serv_addr), sizeof(serv_addr)) < 0) {
int saved_errno = errno; int saved_errno = errno;
::close(_server_fd); int fd = _server_fd.exchange(-1);
_server_fd = -1; if (fd >= 0) ::close(fd);
throw MsgConnectionException(std::string("bind() failed: ") + ::strerror(saved_errno)); throw MsgConnectionException(std::string("bind() failed: ") + ::strerror(saved_errno));
} }
if (::listen(_server_fd, _config.listen_backlog) < 0) { if (::listen(_server_fd.load(), _config.listen_backlog) < 0) {
int saved_errno = errno; int saved_errno = errno;
::close(_server_fd); int fd = _server_fd.exchange(-1);
_server_fd = -1; if (fd >= 0) ::close(fd);
throw MsgConnectionException(std::string("listen() failed: ") + ::strerror(saved_errno)); throw MsgConnectionException(std::string("listen() failed: ") + ::strerror(saved_errno));
} }
@@ -175,13 +192,15 @@ public:
_running = false; _running = false;
_shutdown_pool(pool_threads); _shutdown_pool(pool_threads);
_close_all_clients(); _close_all_clients();
if (_server_fd >= 0) { ::close(_server_fd); _server_fd = -1; } int fd = _server_fd.exchange(-1);
if (fd >= 0) { ::shutdown(fd, SHUT_RDWR); ::close(fd); }
throw; throw;
} }
_shutdown_pool(pool_threads); _shutdown_pool(pool_threads);
_close_all_clients(); _close_all_clients();
if (_server_fd >= 0) { ::close(_server_fd); _server_fd = -1; } int fd = _server_fd.exchange(-1);
if (fd >= 0) { ::shutdown(fd, SHUT_RDWR); ::close(fd); }
} }
// ---- 停止 -------------------------------------------------------------- // ---- 停止 --------------------------------------------------------------
@@ -191,10 +210,10 @@ public:
void stop() noexcept { void stop() noexcept {
bool expected = true; bool expected = true;
if (!_running.compare_exchange_strong(expected, false)) return; if (!_running.compare_exchange_strong(expected, false)) return;
if (_server_fd >= 0) { int fd = _server_fd.exchange(-1);
::shutdown(_server_fd, SHUT_RDWR); if (fd >= 0) {
::close(_server_fd); ::shutdown(fd, SHUT_RDWR);
_server_fd = -1; ::close(fd);
} }
} }
@@ -206,14 +225,17 @@ public:
* 如果连接已断开则静默忽略（fail-safe，不抛异常）。 * 如果连接已断开则静默忽略（fail-safe，不抛异常）。
*/ */
void reply(int sockfd, const std::string& msg) { void reply(int sockfd, const std::string& msg) {
std::shared_ptr<std::mutex> send_lock; std::shared_ptr<ConnectionLock> conn_lock;
{ {
std::lock_guard<std::mutex> lock(_clients_mtx); std::lock_guard<std::mutex> lock(_clients_mtx);
auto it = _send_locks.find(sockfd); auto it = _send_locks.find(sockfd);
if (it == _send_locks.end()) return; // 连接已断开 if (it == _send_locks.end()) return; // 连接已断开
send_lock = it->second; conn_lock = it->second;
} }
std::lock_guard<std::mutex> lock(*send_lock); std::lock_guard<std::mutex> lock(conn_lock->send_mtx);
// 检查 active：防止 _remove_client 先获取 send_mtx 并 close(fd) 后，
// reply 向已关闭/被 OS 复用的 fd 发送数据（CWE-362）
if (!conn_lock->active.load()) return;
try { try {
pmsg_send(sockfd, msg); pmsg_send(sockfd, msg);
} catch (const MsgException&) { } catch (const MsgException&) {
@@ -230,8 +252,8 @@ public:
* @param except_fd 排除的 fd（通常是发送者自己），默认 -1 表示不排除。 * @param except_fd 排除的 fd（通常是发送者自己），默认 -1 表示不排除。
*/ */
void broadcast(const std::string& msg, int except_fd = -1) { void broadcast(const std::string& msg, int except_fd = -1) {
// 先拷贝一份 fd 列表，避免持锁期间发送 // 先拷贝一份 fd + conn_lock 列表，避免持锁期间发送
std::vector<std::pair<int, std::shared_ptr<std::mutex>>> targets; std::vector<std::pair<int, std::shared_ptr<ConnectionLock>>> targets;
{ {
std::lock_guard<std::mutex> lock(_clients_mtx); std::lock_guard<std::mutex> lock(_clients_mtx);
targets.reserve(_clients.size()); targets.reserve(_clients.size());
@@ -243,8 +265,10 @@ public:
} }
} }
} }
for (auto& [fd, send_lock] : targets) { for (auto& [fd, conn_lock] : targets) {
std::lock_guard<std::mutex> slock(*send_lock); std::lock_guard<std::mutex> slock(conn_lock->send_mtx);
// 检查 active：防止 _remove_client 先 close(fd) 后向复用 fd 发送（CWE-362）
if (!conn_lock->active.load()) continue;
try { try {
pmsg_send(fd, msg); pmsg_send(fd, msg);
} catch (const MsgException&) { } catch (const MsgException&) {
@@ -354,9 +378,10 @@ private:
FD_ZERO(&exceptfds); FD_ZERO(&exceptfds);
int maxfd = -1; int maxfd = -1;
if (_server_fd >= 0) { int srv_fd = _server_fd.load();
FD_SET(_server_fd, &readfds); if (srv_fd >= 0) {
maxfd = _server_fd; FD_SET(srv_fd, &readfds);
maxfd = srv_fd;
} }
std::vector<int> watched_clients; std::vector<int> watched_clients;
@@ -396,8 +421,9 @@ private:
} }
} }
// 处理新连接 // 处理新连接（重新加载 srv_fd，防止 stop() 期间关闭）
if (_server_fd >= 0 && FD_ISSET(_server_fd, &readfds)) { srv_fd = _server_fd.load();
if (srv_fd >= 0 && FD_ISSET(srv_fd, &readfds)) {
_accept_client(); _accept_client();
} }
@@ -418,28 +444,15 @@ private:
void _accept_client() { void _accept_client() {
struct sockaddr_in client_addr {}; struct sockaddr_in client_addr {};
socklen_t addrlen = sizeof(client_addr); socklen_t addrlen = sizeof(client_addr);
int client_fd = ::accept(_server_fd, reinterpret_cast<struct sockaddr*>(&client_addr), &addrlen); int srv_fd = _server_fd.load();
int client_fd = ::accept(srv_fd, reinterpret_cast<struct sockaddr*>(&client_addr), &addrlen);
if (client_fd < 0) { if (client_fd < 0) {
// EINVAL/EBADF: server_fd 已被 stop() 关闭；其他错误忽略 // EINVAL/EBADF: server_fd 已被 stop() 关闭；其他错误忽略
return; return;
} }
if (client_fd >= FD_SETSIZE) { ::close(client_fd); return; } if (client_fd >= FD_SETSIZE) { ::close(client_fd); return; }
// 检查连接数上限 // 禁用 Nagle 算法和设置发送超时（在加入 _clients 之前，确保连接配置完整）
{
std::lock_guard<std::mutex> lock(_clients_mtx);
if (static_cast<int>(_clients.size()) >= _config.max_connections) {
::close(client_fd);
return;
}
char ip_str[INET_ADDRSTRLEN];
::inet_ntop(AF_INET, &client_addr.sin_addr, ip_str, sizeof(ip_str));
std::string addr = std::string(ip_str) + ":" + std::to_string(ntohs(client_addr.sin_port));
_clients[client_fd] = addr;
_send_locks[client_fd] = std::make_shared<std::mutex>();
}
// 禁用 Nagle 算法，降低小包延迟
int flag = 1; int flag = 1;
::setsockopt(client_fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag)); ::setsockopt(client_fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));
@@ -452,6 +465,20 @@ private:
sndtv.tv_usec = static_cast<long>( sndtv.tv_usec = static_cast<long>(
(kServerSendTimeout - static_cast<double>(sndtv.tv_sec)) * 1'000'000.0); (kServerSendTimeout - static_cast<double>(sndtv.tv_sec)) * 1'000'000.0);
::setsockopt(client_fd, SOL_SOCKET, SO_SNDTIMEO, &sndtv, sizeof(sndtv)); ::setsockopt(client_fd, SOL_SOCKET, SO_SNDTIMEO, &sndtv, sizeof(sndtv));
// 检查连接数上限并注册
{
std::lock_guard<std::mutex> lock(_clients_mtx);
if (static_cast<int>(_clients.size()) >= _config.max_connections) {
::close(client_fd);
return;
}
char ip_str[INET_ADDRSTRLEN];
::inet_ntop(AF_INET, &client_addr.sin_addr, ip_str, sizeof(ip_str));
std::string addr = std::string(ip_str) + ":" + std::to_string(ntohs(client_addr.sin_port));
_clients[client_fd] = addr;
_send_locks[client_fd] = std::make_shared<ConnectionLock>();
}
} }
void _handle_client(int fd) { void _handle_client(int fd) {
@@ -520,28 +547,32 @@ private:
} }
void _remove_client(int fd) { void _remove_client(int fd) {
std::shared_ptr<std::mutex> send_lock; std::shared_ptr<ConnectionLock> conn_lock;
{ {
std::lock_guard<std::mutex> clk(_clients_mtx); std::lock_guard<std::mutex> clk(_clients_mtx);
if (_clients.erase(fd) == 0) return; // 已被移除，防止 double-close if (_clients.erase(fd) == 0) return; // 已被移除，防止 double-close
auto it = _send_locks.find(fd); auto it = _send_locks.find(fd);
if (it != _send_locks.end()) { if (it != _send_locks.end()) {
send_lock = it->second; conn_lock = it->second;
_send_locks.erase(it); _send_locks.erase(it);
} }
} }
// 持有 send_lock 后再关闭 fd：防止 broadcast/reply 在 fd 关闭后向 OS // 持有 send_mtx 后设置 active=false 再 close(fd)：
// 复用的新连接发送数据（CWE-362）。 // 1. 防止 broadcast/reply 在 fd 关闭后向 OS 复用的新连接发送数据（CWE-362）
if (send_lock) { // 2. active=false 让 reply/broadcast 在获取锁后得知 fd 已关闭，放弃发送
std::lock_guard<std::mutex> slk(*send_lock); if (conn_lock) {
std::lock_guard<std::mutex> slk(conn_lock->send_mtx);
conn_lock->active.store(false);
::shutdown(fd, SHUT_RDWR);
::close(fd); ::close(fd);
} else { } else {
::shutdown(fd, SHUT_RDWR);
::close(fd); ::close(fd);
} }
} }
void _close_all_clients() { void _close_all_clients() {
std::vector<std::pair<int, std::shared_ptr<std::mutex>>> to_close; std::vector<std::pair<int, std::shared_ptr<ConnectionLock>>> to_close;
{ {
std::lock_guard<std::mutex> clk(_clients_mtx); std::lock_guard<std::mutex> clk(_clients_mtx);
to_close.reserve(_clients.size()); to_close.reserve(_clients.size());
@@ -556,11 +587,12 @@ private:
_clients.clear(); _clients.clear();
_send_locks.clear(); _send_locks.clear();
} }
// 持有每个连接的 send_lock 后再 close(fd)：防止 stop()/析构时 // 持有每个连接的 send_mtx 后设置 active=false 再 close(fd)：
// reply/broadcast 正在发送，导致数据发到 OS 复用的新连接（CWE-362） // 防止 stop()/析构时 reply/broadcast 正在发送，导致数据发到 OS 复用的新连接（CWE-362）
for (auto& [fd, send_lock] : to_close) { for (auto& [fd, conn_lock] : to_close) {
if (send_lock) { if (conn_lock) {
std::lock_guard<std::mutex> slk(*send_lock); std::lock_guard<std::mutex> slk(conn_lock->send_mtx);
conn_lock->active.store(false);
::shutdown(fd, SHUT_RDWR); ::shutdown(fd, SHUT_RDWR);
::close(fd); ::close(fd);
} else { } else {
@@ -579,14 +611,14 @@ private:
std::string _addr; std::string _addr;
int _port; int _port;
MsgServerConfig _config; MsgServerConfig _config;
int _server_fd; std::atomic<int> _server_fd; ///< 监听 socket fd（atomic 防止 stop() 与 _event_loop 数据竞争）
std::atomic<bool> _running; std::atomic<bool> _running;
MessageCallback _on_message; MessageCallback _on_message;
// 客户端管理 // 客户端管理
mutable std::mutex _clients_mtx; mutable std::mutex _clients_mtx;
std::unordered_map<int, std::string> _clients; ///< fd -> "ip:port" std::unordered_map<int, std::string> _clients; ///< fd -> "ip:port"
std::unordered_map<int, std::shared_ptr<std::mutex>> _send_locks; ///< fd -> 发送锁 std::unordered_map<int, std::shared_ptr<ConnectionLock>> _send_locks; ///< fd -> 发送锁+活跃标志
// 同连接串行 // 同连接串行
std::mutex _busy_mtx; std::mutex _busy_mtx;
libmsg/inc/us/RpcServer.hpp
+4 -1
View File
@@ -132,9 +132,12 @@ private:
if (auto it = req.find("_route"); it != req.end() && it->value().is_string()) { if (auto it = req.find("_route"); it != req.end() && it->value().is_string()) {
route = it->value().as_string().c_str(); route = it->value().as_string().c_str();
req.erase(it); // 从业务参数中移除 _route req.erase(it); // 从业务参数中移除 _route
} else { } else if (it == req.end()) {
_reply_error(fd, "missing _route"); _reply_error(fd, "missing _route");
return; return;
} else {
_reply_error(fd, "_route must be a string");
return;
} }
// 查找 handler // 查找 handler