From d1ff650f10fe3e3496a133ad5e22376c33a3e2ef Mon Sep 17 00:00:00 2001 From: nick Date: Wed, 4 Nov 2026 11:43:20 +0800 Subject: [PATCH] fix(redis): reject malformed RESP3 reply boundaries --- docs/features/redis/configuration.md | 4 +- docs/features/redis/configuration_CN.md | 4 +- include/arknet/redis/detail/resp.hpp | 10 +++ tests/redis_client.cpp | 81 +++++++++++++++++++++++++ tests/redis_resp.cpp | 40 ++++++++++++ 5 files changed, 137 insertions(+), 2 deletions(-) diff --git a/docs/features/redis/configuration.md b/docs/features/redis/configuration.md index 2bc77eb..1f10909 100644 --- a/docs/features/redis/configuration.md +++ b/docs/features/redis/configuration.md @@ -18,7 +18,9 @@ Configure `redis::connect_options` before `async_connect`: `host`, `port`, `user | `map`, `attribute` | Alternating key/value entries in `elements`, retaining all key types | | `null` | No payload | -`attributes` contains RESP3 attributes attached to the value. Inspect types before accessing data. A server error is a typed reply, not a transport failure: `bool(result)` checks `client_errc`, so applications must also inspect individual replies. Failed pipelines can contain replies received before the failure. +`attributes` contains RESP3 attributes attached to the value. Inspect types before accessing data. A server error is a typed reply, not a transport failure: `bool(result)` checks `client_errc`, so applications must also inspect individual replies. Failed or cancelled pipelines retain complete replies received before completion; those values remain valid after subsequent decoding and client destruction. Partial replies are never exposed. + +The decoder follows the [Redis protocol specification](https://redis.io/docs/latest/develop/reference/protocol-spec/): RESP3 double fractions require digits after the decimal point, and pushes are top-level messages. A malformed reply or a `response_limits` violation closes the connection with `client_errc::protocol_error` and fails outstanding requests without replay. ## Limits and deadlines diff --git a/docs/features/redis/configuration_CN.md b/docs/features/redis/configuration_CN.md index fe78196..28da32e 100644 --- a/docs/features/redis/configuration_CN.md +++ b/docs/features/redis/configuration_CN.md @@ -18,7 +18,9 @@ | `map`、`attribute` | `elements` 中交替排列键和值，保留键的类型 | | `null` | 无数据 | -`attributes` 保存附着在该值上的 RESP3 属性。读取数据前先检查类型。服务端错误是类型化回复，不是传输失败：`bool(result)` 检查 `client_errc`，应用仍需检查每一项回复。流水线失败时，结果可能保留失败前已经收到的回复。 +`attributes` 保存附着在该值上的 RESP3 属性。读取数据前先检查类型。服务端错误是类型化回复，不是传输失败：`bool(result)` 检查 `client_errc`，应用仍需检查每一项回复。流水线失败或取消时，结果保留完成前已经收到的完整回复；后续解码和客户端析构不会使这些值失效。不完整的回复不会暴露给调用者。 + +解码器遵循 [Redis 协议规范](https://redis.io/docs/latest/develop/reference/protocol-spec/)：RESP3 double 的小数点后必须有数字，推送必须位于消息顶层。回复格式错误或超出 `response_limits` 时，以 `client_errc::protocol_error` 关闭连接，让在途请求失败，不重放命令。 ## 限额与超时 diff --git a/include/arknet/redis/detail/resp.hpp b/include/arknet/redis/detail/resp.hpp index 0ca8437..8a817fc 100644 --- a/include/arknet/redis/detail/resp.hpp +++ b/include/arknet/redis/detail/resp.hpp @@ -139,6 +139,10 @@ class resp_decoder { return fail(resp_status::malformed); } + else if (byte == '>' && !frames_.empty()) + { + return fail(resp_status::malformed); + } else if (++values_ > limits_.max_values || frames_.size() >= limits_.max_depth) { return fail(resp_status::limit_exceeded); @@ -563,6 +567,12 @@ class resp_decoder { return fail(resp_status::malformed); } + const auto point = integral.find('.'); + if (point != std::string_view::npos && + (point + 1 == integral.size() || integral[point + 1] < '0' || integral[point + 1] > '9')) + { + return fail(resp_status::malformed); + } const auto parsed = std::from_chars(number.data(), number.data() + number.size(), value.real); if (parsed.ec != std::errc{} || parsed.ptr != number.data() + number.size() || !std::isfinite(value.real)) { diff --git a/tests/redis_client.cpp b/tests/redis_client.cpp index e0e319b..c0e38c3 100644 --- a/tests/redis_client.cpp +++ b/tests/redis_client.cpp @@ -269,6 +269,87 @@ DOCTEST_TEST_CASE("Redis disconnect and malformed replies fail outstanding comma } } +DOCTEST_TEST_CASE("Redis failed pipelines retain complete owning replies and retire completion captures") { + const std::string binary("a\0\r\nb", 5); + const std::vector> failures{{",1.\r\n+VALID\r\n", client_errc::protocol_error}, + {"*1\r\n>1\r\n+invalid\r\n+VALID\r\n", client_errc::protocol_error}, {"$3\r\nab", client_errc::disconnected}}; + for (const auto& [wire, error] : failures) + { + server endpoint([&](peer& remote) { + remote.hello(); + remote.expect({"ECHO", binary}); + remote.expect({"GET", "broken"}); + remote.expect({"PING"}); + remote.send("$5\r\n" + binary + "\r\n" + wire); + }); + auto connection = std::make_unique(); + check(connect(*connection, endpoint.port()), "connected"); + auto capture = std::make_shared(1); + const std::weak_ptr retained = capture; + std::atomic calls{}; + std::promise failed, following; + auto failed_future = failed.get_future(), following_future = following.get_future(); + connection->async_pipeline({{"ECHO", binary}, {"GET", "broken"}}, [&, capture](result reply) { + ++calls; + failed.set_value(std::move(reply)); + }); + capture.reset(); + connection->async_execute({"PING"}, [&](result reply) { following.set_value(std::move(reply)); }); + auto response = await(failed_future); + check(response.error == error && response.command_may_have_run, "failure preserves unknown write outcome"); + check(await(following_future).error == error, "later command fails without receiving a stray reply"); + check(connection->close_and_wait(), "retire failed connection"); + connection.reset(); + endpoint.finish(); + check(calls == 1 && retained.expired(), "complete once and release callback captures"); + check(response.replies.size() == 1 && response.replies[0].type == value_type::bulk_string && + response.replies[0].text == binary, + "complete prefix survives parser and client destruction"); + } +} + +DOCTEST_TEST_CASE("Redis cancellation preserves an owning pipeline prefix while later replies drain") { + const std::string binary("a\0\r\nb", 5); + std::promise release; + auto allowed = release.get_future(); + server endpoint([&](peer& remote) { + remote.hello(); + remote.expect({"ECHO", binary}); + remote.expect({"INCR", "cancelled"}); + remote.expect({"GET", "following"}); + remote.send("$5\r\n" + binary + "\r\n>1\r\n+prefix-ready\r\n"); + if (allowed.wait_for(5s) != std::future_status::ready) + { + throw std::runtime_error("Missing drain signal"); + } + remote.send(":22\r\n:33\r\n"); + }); + client connection; + std::promise prefix; + auto prefix_ready = prefix.get_future(); + connection.set_push_handler([&](auto) { prefix.set_value(); }); + check(connect(connection, endpoint.port()), "connected"); + std::atomic calls{}; + std::promise cancelled, following; + auto cancelled_future = cancelled.get_future(), following_future = following.get_future(); + auto id = connection.async_pipeline({{"ECHO", binary}, {"INCR", "cancelled"}}, [&](result reply) { + ++calls; + cancelled.set_value(std::move(reply)); + }); + connection.async_execute({"GET", "following"}, [&](result reply) { following.set_value(std::move(reply)); }); + await(prefix_ready); + connection.cancel(id); + auto response = await(cancelled_future); + check(response.error == client_errc::cancelled && response.command_may_have_run, "cancel sent pipeline"); + release.set_value(); + auto next = await(following_future); + check(next && next.replies.size() == 1 && next.replies[0].integer == 33, "remaining reply keeps its old slot"); + check(connection.close_and_wait(), "retire drained connection"); + endpoint.finish(); + check(calls == 1 && response.replies.size() == 1 && response.replies[0].text == binary, + "cancelled owning prefix survives subsequent decoding and shutdown"); +} + DOCTEST_TEST_CASE("Redis owner destruction inside a completion aborts later slots safely") { server endpoint([](peer& remote) { remote.hello(); diff --git a/tests/redis_resp.cpp b/tests/redis_resp.cpp index 266db94..4001ed4 100644 --- a/tests/redis_resp.cpp +++ b/tests/redis_resp.cpp @@ -90,6 +90,46 @@ DOCTEST_TEST_CASE("RESP3 represents scalars, maps, sets, pushes and verbatim val check(push.type == resp_type::push && push.elements[0].text == "invalidate", "push remains distinguishable"); } +DOCTEST_TEST_CASE("RESP3 doubles require digits after a decimal point") { + check(decode(",1.0\r\n").real == 1.0, "fractional digits"); + check(decode(",-1.25e+2\r\n").real == -125.0, "fraction and exponent"); + check(decode(",+1e-2\r\n").real == 0.01, "exponent without a fraction"); + for (const std::string wire : {",1.\r\n", ",-1.e+2\r\n", ",+0.E-1\r\n"}) + { + DOCTEST_INFO(wire); + resp_decoder decoder; + resp_value value; + value.text = "unchanged"; + check(decoder.append(wire + "+VALID\r\n"), "append invalid double and following reply"); + check(decoder.next(value) == resp_status::malformed && value.text == "unchanged", + "incomplete fraction is not a reply"); + check(decoder.next(value) == resp_status::malformed, "invalid double cannot expose the following reply"); + } +} + +DOCTEST_TEST_CASE("RESP3 pushes cannot be nested inside another value") { + for (const std::string aggregate : {"*1", "%1", "~1", "|1", ">2\r\n+outer", "*?", "%?", "~?"}) + { + DOCTEST_INFO(aggregate); + const auto wire = aggregate + "\r\n>1\r\n+invalid\r\n:1\r\n+VALID\r\n"; + for (std::size_t split = 0; split < wire.size(); ++split) + { + resp_decoder decoder; + resp_value value; + value.text = "unchanged"; + check(decoder.append(std::string_view(wire).substr(0, split)), "append prefix"); + auto status = decoder.next(value); + if (status == resp_status::need_more) + { + check(decoder.append(std::string_view(wire).substr(split)), "append suffix"); + status = decoder.next(value); + } + check(status == resp_status::malformed && value.text == "unchanged", "reject nested push"); + check(decoder.next(value) == resp_status::malformed, "nested push failure is sticky"); + } + } +} + DOCTEST_TEST_CASE("RESP3 attributes decorate the correct nested value without consuming a reply") { auto value = decode("|1\r\n+outer\r\n:1\r\n|1\r\n+other\r\n:2\r\n" "*2\r\n|1\r\n+ttl\r\n:9\r\n$1\r\na\r\n:5\r\n");