feat(xpp/http): httptest — three-layer HTTP testing (tower/axum + wiremock aligned)

issues/http-streaming-conn-close-uaf.md

@@ -0,0 +1,70 @@
+# Issue: 流式响应 body 在连接关闭后写已释放的 ctx —— heap-use-after-free
+
+> 状态：fixed（`on_close` 回调 + `ConnLifetime` 引用计数信号）
+> 首次观测：PR #87 的 CI（2026-08-26）——`ubuntu-latest / openssl / C++11`（ASan）与
+> `Symbol visibility (shared build)` 两个 lane 同时失败
+
+## 现象
+
+`libxpp/xpp/http/client_test.cpp` 的 `ClientSendTest.ReadTimeoutTriggersError`：
+
+- ASan lane：`ERROR: AddressSanitizer: heap-use-after-free`，读 `xHttpCtxWrite`
+  （`libx/x/http/server.c:1005`），freed by `xHttpConnClose`（`server.c:650`）
+- 无 ASan 的 shared-build lane：同测试直接 SegFault
+
+两个 lane 是**同一个 bug**——一个被 ASan 捕获，一个裸奔。
+
+## 根因
+
+响应流式路径 `stream_channel_body`（`libxpp/xpp/http/server.h`）跨挂起点持有裸
+`xHttpCtx*`，而它只受 `ServerLifetime` 保护。`ServerLifetime::destroyed` 只在
+`Server::~Server` 置位，**覆盖不到连接级销毁**：
+
+1. 测试服务器流式返回 1MB body，中途停顿 2s（`mid_body_delay_ms`）
+2. 客户端 `read_timeout(1000)`，`send()` 收到 header 成功，但 `bytes()` 读一半超时
+   → 客户端断开连接
+3. 服务端 event loop 读到 `n==0` → `xHttpConnClose` → `xHttpStreamDestroy` 释放
+   stream（`xHttpCtx` 内嵌在 stream 里，`stream->ctx.internal_ = stream`）
+4. 但 spawn 出去的 handler 任务还挂起在 `stream_channel_body` 的 body channel read 上；
+   测试服务器稍后推下一个 chunk 时，continuation 拿着悬空的 `xHttpCtx*` 调
+   `xHttpCtxWrite` → UAF
+
+关键点：`xHttpConnClose` 路径**不会**触发 `on_done`（`srv_on_done_cb` 只在
+`conn_dispatch_request` 的正常完成路径调用），所以既有的 per-request 清理钩子
+完全感知不到连接异常关闭。
+
+## 修复
+
+三处小改动：
+
+1. **C API**（`libx/x/http/server.h`）：`xHttpRouteInfo` 新增 `on_close` 回调
+   （`xHttpCloseFunc`），语义是「stream 销毁前调用，覆盖所有 teardown 路径」。
+2. **C 实现**（`libx/x/http/server.c`）：`xHttpConnClose` 在 `xHttpStreamDestroy`
+   之前调用 `on_close`，arg 用 `route_info->arg`（**不是** `stream->user`，后者可能
+   已被先前的 `on_done` 释放）。
+3. **C++ 层**（`libxpp/xpp/http/server.h`）：引入 `ConnLifetime`（`Arc` + `bool
+   closed`），存于 `ServerImpl::m_conns`（`Arc<ConnMap>`，key = ctx 地址）。`on_close`
+   回调置 `closed=true`；spawn 任务与 `stream_channel_body` 持有该 Arc，每次碰 ctx
+   前先查 `closed`。
+
+为什么不给 stream 加引用计数：`xHttpCtxWrite` 不只访问 stream，还访问 `stream->conn`
+→ 引用计数必须同时延长 conn（socket/TLS/协议状态）的生命周期，把「连接关闭」的
+资源释放时序和「响应流写完」耦合，复杂且易泄漏。`on_close` 信号方案保持资源释放
+时序不变，只多一个通知，与既有 `ServerLifetime` 模式同构。
+
+## 验证
+
+- `http_client_test` 13/13 通过（ASan 构建，含 `ReadTimeoutTriggersError`）
+- `http_server_test` 16/16 通过
+- `xhttp_test`（C 层）173/173 通过
+
+## 复现
+
+```bash
+cmake -B build -G Ninja -DCMAKE_C_FLAGS="-fsanitize=address -fno-omit-frame-pointer" \
+      -DCMAKE_CXX_FLAGS="-fsanitize=address -fno-omit-frame-pointer"
+cmake --build build --target http_client_test
+./build/libxpp/xpp/http_client_test --gtest_filter=ClientSendTest.ReadTimeoutTriggersError
+```
+
+修复前：ASan 报 heap-use-after-free（`xHttpCtxWrite` → `stream_channel_body` lambda）。

libx/x/http/server.c

@@ -647,6 +647,13 @@ void xHttpConnClose(struct xHttpConn_ *conn) {
   }
 
   if (conn->stream) {
+    /* Notify the route before the stream (and the xHttpCtx embedded in it)
+     * is freed — the C++ layer uses this to stop in-flight response
+     * streaming. Use route_info->arg (not stream->user, which may already
+     * be freed by a prior on_done). */
+    if (conn->stream->route_info && conn->stream->route_info->on_close) {
+      conn->stream->route_info->on_close(&conn->stream->ctx, conn->stream->route_info->arg);
+    }
     xHttpStreamDestroy(conn->stream);
     conn->stream = NULL;
   }

libx/x/http/server.h

@@ -34,21 +34,39 @@ XDEF_HANDLE(xHttpServer);
  */
 XDEF_HANDLE(xHttpMux);
 
+/**
+ * @brief Callback invoked right before the connection's request stream is
+ *        torn down (client disconnect, error, or normal close).
+ *
+ * Unlike @ref xHttpDoneFunc, this fires on *every* path that ends the
+ * request's lifetime, including an abrupt close that never reaches
+ * @ref xHttpDoneFunc. The @p ctx is still valid during the callback but
+ * must not be used after it returns.
+ *
+ * @param ctx  Request context (valid only during the callback).
+ * @param arg  User-provided argument (the route's @p arg, NOT the
+ *             per-request user data, which may already be freed).
+ */
+typedef void (*xHttpCloseFunc)(xHttpCtx *ctx, void *arg);
+
 /**
  * @brief Route information returned by the resolver.
  *
  * Returned by @ref xHttpResolveFunc after the request headers are parsed.
  * The library calls @p on_request (if non-NULL) right after resolution,
  * streams the body via @p on_data (if non-NULL), and finally invokes
- * @p on_done when the request is fully received.
+ * @p on_done when the request is fully received. @p on_close (if non-NULL)
+ * is invoked immediately before the stream is destroyed, on every teardown
+ * path.
  *
  * All callbacks receive @p arg as the user-provided context.
  */
 XDEF_STRUCT(xHttpRouteInfo) {
-  xHttpInitFunc on_request; /**< Called once after headers (may be NULL) */
-  xHttpDataFunc on_data;    /**< Per body chunk callback (may be NULL)    */
-  xHttpDoneFunc on_done;    /**< Called when request is complete           */
-  void         *arg;        /**< User argument forwarded to callbacks      */
+  xHttpInitFunc  on_request; /**< Called once after headers (may be NULL) */
+  xHttpDataFunc  on_data;    /**< Per body chunk callback (may be NULL)    */
+  xHttpDoneFunc  on_done;    /**< Called when request is complete           */
+  xHttpCloseFunc on_close;   /**< Called before the stream is destroyed     */
+  void          *arg;        /**< User argument forwarded to callbacks      */
 };
 
 /**

libxpp/xpp/http/client_test.cpp

@@ -5,11 +5,11 @@
  *
  * client_test.cpp — Layer 3 integration tests for xpp::http::Client.
  *
- * Uses xpp::http::test::TestServer (loopback, no external network) to
+ * Uses xpp::http::test::Server (loopback, no external network) to
  * exercise Client::send end-to-end: request submission, push→pull
  * body bridge, header parsing, error mapping, and timeout.
  *
- * TestServer uses libx C API (xTcpListener + synchronous accept
+ * test::Server uses libx C API (xTcpListener + synchronous accept
  * callback), so no fibers are involved. client.send(req).await()
  * runs on the main thread via the non-fiber park() path
  * (xEventLoopRun), which is safe because no fiber switches occur
@@ -21,12 +21,13 @@
 #include <gtest/gtest.h>
 #include <xpp/event.h>
 #include <xpp/http/client.h>
-#include <xpp/http/test_server.h>
+#include <xpp/http/test/evil_server.h>
+#include <xpp/http/test/server.h>
 
 using namespace xpp;
 using namespace xpp::http;
 
-/* ── Helper: build a URL for the TestServer ─────────────────────── */
+/* ── Helper: build a URL for the test::Server ─────────────────────── */
 
 static std::string url_for(uint16_t port, const char *path = "/") {
   return "http://127.0.0.1:" + std::to_string(port) + path;
@@ -53,7 +54,7 @@ TEST(ClientTest, BuilderRejectsNoEventLoop) {
 }
 
 /* ───────────────────────────────────────────────────────────────────
- *  End-to-end send via TestServer
+ *  End-to-end send via test::Server
  * ─────────────────────────────────────────────────────────────────── */
 
 TEST(ClientSendTest, GetReturns200WithBody) {
@@ -66,7 +67,7 @@ TEST(ClientSendTest, GetReturns200WithBody) {
     {String::from_utf8("Content-Type").unwrap(), String::from_utf8("text/plain").unwrap()});
   spec.body = Bytes::from("hello");
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req =
     Request::builder().method(Method::Get).url(url_for(server.port()).c_str()).body().unwrap();
@@ -96,7 +97,7 @@ TEST(ClientSendTest, PostWithBodyRoundTrips) {
   spec.status = StatusCode::Ok;
   spec.body   = Bytes::from("ack");
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req = Request::builder()
                .method(Method::Post)
@@ -126,7 +127,7 @@ TEST(ClientSendTest, NotFoundReturns400LevelStatus) {
   spec.status = StatusCode::NotFound;
   spec.body   = Bytes::from("nope");
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req =
     Request::builder().method(Method::Get).url(url_for(server.port()).c_str()).body().unwrap();
@@ -154,7 +155,7 @@ TEST(ClientSendTest, TimeoutTriggersError) {
   spec.body     = Bytes::from("slow");
   spec.delay_ms = 200;
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req =
     Request::builder().method(Method::Get).url(url_for(server.port()).c_str()).body().unwrap();
@@ -178,7 +179,7 @@ TEST(ClientSendTest, PostBodyIsTransmitted) {
   spec.status            = StatusCode::Ok;
   spec.echo_request_body = true;
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req = Request::builder()
                .method(Method::Post)
@@ -211,7 +212,7 @@ TEST(ClientSendTest, LargeBodyRoundTrips) {
   spec.status            = StatusCode::Ok;
   spec.echo_request_body = true;
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   std::string payload(2 * 1024 * 1024, 'x');
   auto        req = Request::builder()
@@ -247,7 +248,7 @@ TEST(ClientSendTest, RedirectFollowed) {
   spec.body        = Bytes::from("redirected!");
   spec.headers.push({String::from_utf8("X-Final-Hop").unwrap(), String::from_utf8("yes").unwrap()});
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req = Request::builder()
                .method(Method::Get)
@@ -290,7 +291,7 @@ TEST(ClientSendTest, LargeBodyBackpressure) {
   spec.status = StatusCode::Ok;
   spec.body   = Bytes::from(std::string(8 * 1024 * 1024, 'y').c_str());
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req =
     Request::builder().method(Method::Get).url(url_for(server.port()).c_str()).body().unwrap();
@@ -317,12 +318,11 @@ TEST(ClientSendTest, MidBodyDisconnectReportsError) {
   // send() still resolves Ok (headers arrived — reqwest semantics), but
   // reading the body must surface an error instead of a silent truncated
   // EOF.
-  test::TestResponseSpec spec;
-  spec.status              = StatusCode::Ok;
-  spec.body                = Bytes::from(std::string(1024 * 1024, 'z').c_str());
-  spec.truncate_body_after = 64 * 1024;
+  test::EvilSpec spec;
+  spec.body      = Bytes::copy(std::string(1024 * 1024, 'z').c_str(), 1024 * 1024);
+  spec.send_only = 64 * 1024;
 
-  auto server = test::TestServer::start(spec);
+  test::EvilServer server(spec);
 
   auto req =
     Request::builder().method(Method::Get).url(url_for(server.port()).c_str()).body().unwrap();
@@ -353,7 +353,7 @@ TEST(ClientSendTest, ReadTimeoutTriggersError) {
   spec.body              = Bytes::from(std::string(1024 * 1024, 'r').c_str());
   spec.mid_body_delay_ms = 2000;
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req =
     Request::builder().method(Method::Get).url(url_for(server.port()).c_str()).body().unwrap();
@@ -378,7 +378,7 @@ TEST(ClientSendTest, CustomHeaderSent) {
   test::TestResponseSpec spec;
   spec.status = StatusCode::Ok;
 
-  auto server = test::TestServer::start(spec);
+  test::Server server(spec);
 
   auto req = Request::builder()
                .method(Method::Get)

libxpp/xpp/http/http_convenience_test.cpp

@@ -4,15 +4,15 @@
  * found in the LICENSE file.
  *
  * http_convenience_test.cpp — Phase 6: Client convenience methods +
- * top-level xpp::http::get/post/... against the local TestServer.
+ * top-level xpp::http::get/post/... against the local test::Server.
  */
 
 #include <string>
 
 #include <gtest/gtest.h>
 #include <xpp/event.h>
 #include <xpp/http/client.h>
-#include <xpp/http/test_server.h>
+#include <xpp/http/test/server.h>
 
 using namespace xpp;
 using namespace xpp::http;
@@ -35,8 +35,8 @@ TEST(ClientConvenienceTest, VerbsRoundTrip) {
   spec.echo_request_method = true;
   spec.echo_request_body   = true;
 
-  auto server = test::TestServer::start(spec);
-  auto client = Client::builder().build().unwrap();
+  test::Server server(spec);
+  auto         client = Client::builder().build().unwrap();
 
   // GET
   {
@@ -98,8 +98,8 @@ TEST(ClientConvenienceTest, UrlOverloadsCompileAndRoundTrip) {
 
   test::TestResponseSpec spec;
   spec.status = StatusCode::Ok;
-  auto server = test::TestServer::start(spec);
-  auto client = Client::builder().build().unwrap();
+  test::Server server(spec);
+  auto         client = Client::builder().build().unwrap();
 
   // const char*
   ASSERT_TRUE(client.get(url_for(server.port()).c_str()).await().is_ok());