This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 84457631 Bound total array allocation of one redis reply by
-redis_max_allocation_size (#3560)
84457631 is described below
commit 8445763165f3a1b381c54eb6fb1c1bc4f26ebb8e
Author: Weibing Wang <[email protected]>
AuthorDate: Wed Sep 23 10:07:20 2026 +0800
Bound total array allocation of one redis reply by
-redis_max_allocation_size (#3560)
* Bound total array allocation of one redis reply by
-redis_max_allocation_size
The redis reply parser capped each array allocation individually and the
nesting depth separately, but nothing bounded the two together: nesting a
maximally-sized array inside another committed depth * cap bytes from a
tiny reply. Thread a per-reply byte budget through
RedisReply::ConsumePartialIOBuf, save it in the root array when the
parsing suspends and restore it on resume so that withholding data cannot
reset it.
Also stop pre-resizing RedisCommandParser::_args from a declared RESP
array count; grow it as the arguments actually arrive so that a ~12-byte
incomplete command no longer commits up to the cap per connection.
* Initialize used_bytes in RedisReply::Reset()
Arrays built by user code (e.g. SetArray) never went through the reply
parser, so the new used_bytes field could hold an indeterminate value
that CopyFromDifferentArena would read. Initialize it in Reset(), which
every construction path goes through.
---------
Co-authored-by: brpc-oncall <brpc-oncall@localhost>
---
src/brpc/redis_command.cpp | 9 ++-
src/brpc/redis_reply.cpp | 46 ++++++++++++++--
src/brpc/redis_reply.h | 8 ++-
test/brpc_redis_unittest.cpp | 128 +++++++++++++++++++++++++++++++++++++++++++
4 files changed, 184 insertions(+), 7 deletions(-)
diff --git a/src/brpc/redis_command.cpp b/src/brpc/redis_command.cpp
index a833cb54..62b9bd1c 100644
--- a/src/brpc/redis_command.cpp
+++ b/src/brpc/redis_command.cpp
@@ -565,7 +565,10 @@ RedisCommandConsumeState
RedisCommandParser::ConsumeImpl(butil::IOBuf& buf,
_parsing_array = true;
_length = value;
_index = 0;
- _args.resize(value);
+ // NOTE: don't resize _args to `value' here. The count is just a
+ // declaration and the arguments may never arrive; growing _args
+ // lazily avoids committing up to -redis_max_allocation_size of a
+ // connection from a ~12-byte incomplete command.
return CONSUME_STATE_CONTINUE;
}
if (_index >= _length) {
@@ -599,6 +602,10 @@ RedisCommandConsumeState
RedisCommandParser::ConsumeImpl(butil::IOBuf& buf,
}
buf.cutn(d, len);
d[len] = '\0';
+ if (_args.size() <= (size_t)_index) {
+ // Grow _args as the arguments arrive instead of upfront.
+ _args.resize(_index + 1);
+ }
_args[_index].set(d, len);
if (_index == 0) {
// convert it to lowercase when it is command name
diff --git a/src/brpc/redis_reply.cpp b/src/brpc/redis_reply.cpp
index e93b9282..ee77b3e6 100644
--- a/src/brpc/redis_reply.cpp
+++ b/src/brpc/redis_reply.cpp
@@ -96,21 +96,41 @@ bool RedisReply::SerializeTo(butil::IOBufAppender*
appender) {
}
ParseError RedisReply::ConsumePartialIOBuf(butil::IOBuf& buf) {
- return ConsumePartialIOBuf(buf, 0);
+ // `used_bytes' accumulates the arena bytes charged to
+ // -redis_max_allocation_size for this reply. It is shared by the whole
+ // reply tree so that nested arrays draw from one budget instead of each
+ // getting a full one (which multiplied the cap by nesting depth). When
+ // the parsing suspends, the count is saved in the root array and restored
+ // on resume, otherwise a peer could reset the budget by withholding data.
+ uint32_t used_bytes = 0;
+ ParseError err = ConsumePartialIOBuf(buf, 0, &used_bytes);
+ if (err != PARSE_OK && _type == REDIS_REPLY_ARRAY &&
+ _data.array.last_index >= 0) {
+ // The parsing suspends on sub replies, remember the budget for resume.
+ _data.array.used_bytes = used_bytes;
+ }
+ return err;
}
-ParseError RedisReply::ConsumePartialIOBuf(butil::IOBuf& buf, int depth) {
+ParseError RedisReply::ConsumePartialIOBuf(butil::IOBuf& buf, int depth,
+ uint32_t* used_bytes) {
if (depth > FLAGS_redis_max_reply_depth) {
LOG(ERROR) << "redis reply exceeds max depth! max="
<< FLAGS_redis_max_reply_depth << ", actually=" << depth;
return PARSE_ERROR_ABSOLUTELY_WRONG;
}
if (_type == REDIS_REPLY_ARRAY && _data.array.last_index >= 0) {
+ if (depth == 0) {
+ // Resuming the root array of a suspended reply: restore the
+ // budget charged by previous calls.
+ *used_bytes = _data.array.used_bytes;
+ }
// The parsing was suspended while parsing sub replies,
// continue the parsing.
RedisReply* subs = (RedisReply*)_data.array.replies;
for (int i = _data.array.last_index; i < _length; ++i) {
- ParseError err = subs[i].ConsumePartialIOBuf(buf, depth + 1);
+ ParseError err = subs[i].ConsumePartialIOBuf(buf, depth + 1,
+ used_bytes);
if (err != PARSE_OK) {
return err;
}
@@ -268,12 +288,26 @@ ParseError RedisReply::ConsumePartialIOBuf(butil::IOBuf&
buf, int depth) {
<< max_count << ", actually=" << count;
return PARSE_ERROR_ABSOLUTELY_WRONG;
}
+ // Charge the allocation to the per-reply budget so that nested
+ // arrays cannot multiply -redis_max_allocation_size by depth.
+ const size_t need = sizeof(RedisReply) * count;
+ if (FLAGS_redis_max_allocation_size > 0 &&
+ (uint64_t)*used_bytes + need >
+ (uint64_t)FLAGS_redis_max_allocation_size) {
+ LOG(ERROR) << "accumulated array allocation exceeds max "
+ "allocation size! max="
+ << FLAGS_redis_max_allocation_size
+ << ", already=" << *used_bytes
+ << ", requesting=" << need;
+ return PARSE_ERROR_ABSOLUTELY_WRONG;
+ }
// FIXME(gejun): Call allocate_aligned instead.
- RedisReply* subs =
(RedisReply*)_arena->allocate(sizeof(RedisReply) * count);
+ RedisReply* subs = (RedisReply*)_arena->allocate(need);
if (subs == nullptr) {
LOG(FATAL) << "Fail to allocate RedisReply[" << count << "]";
return PARSE_ERROR_ABSOLUTELY_WRONG;
}
+ *used_bytes += need;
for (int64_t i = 0; i < count; ++i) {
new (&subs[i]) RedisReply(_arena);
}
@@ -286,7 +320,8 @@ ParseError RedisReply::ConsumePartialIOBuf(butil::IOBuf&
buf, int depth) {
// be continued in next calls by tracking _data.array.last_index.
_data.array.last_index = 0;
for (int64_t i = 0; i < count; ++i) {
- ParseError err = subs[i].ConsumePartialIOBuf(buf, depth + 1);
+ ParseError err = subs[i].ConsumePartialIOBuf(buf, depth + 1,
+ used_bytes);
if (err != PARSE_OK) {
return err;
}
@@ -404,6 +439,7 @@ void RedisReply::CopyFromDifferentArena(const RedisReply&
other) {
new (&subs[i]) RedisReply(_arena);
}
_data.array.last_index = other._data.array.last_index;
+ _data.array.used_bytes = other._data.array.used_bytes;
if (_data.array.last_index > 0) {
// incomplete state
for (int i = 0; i < _data.array.last_index; ++i) {
diff --git a/src/brpc/redis_reply.h b/src/brpc/redis_reply.h
index 4efb5b71..725e184e 100644
--- a/src/brpc/redis_reply.h
+++ b/src/brpc/redis_reply.h
@@ -146,7 +146,8 @@ private:
// by calling CopyFrom[Different|Same]Arena.
DISALLOW_COPY_AND_ASSIGN(RedisReply);
- ParseError ConsumePartialIOBuf(butil::IOBuf& buf, int depth);
+ ParseError ConsumePartialIOBuf(butil::IOBuf& buf, int depth,
+ uint32_t* used_bytes);
void FormatStringImpl(const char* fmt, va_list args, RedisReplyType type);
void SetStringImpl(const butil::StringPiece& str, RedisReplyType type);
@@ -158,6 +159,10 @@ private:
const char* long_str;
struct {
int32_t last_index; // >= 0 if previous parsing suspends on
replies.
+ // Arena bytes charged to -redis_max_allocation_size by the reply
+ // tree so far. Only read/written on the root array of a reply
+ // whose parsing was suspended, to keep the budget across resumes.
+ uint32_t used_bytes;
RedisReply* replies;
} array;
uint64_t padding[2]; // For swapping, must cover all bytes.
@@ -176,6 +181,7 @@ inline void RedisReply::Reset() {
_type = REDIS_REPLY_NIL;
_length = 0;
_data.array.last_index = -1;
+ _data.array.used_bytes = 0;
_data.array.replies = nullptr;
// _arena should not be reset because further memory allocation needs it.
}
diff --git a/test/brpc_redis_unittest.cpp b/test/brpc_redis_unittest.cpp
index 724b5e77..b1c57067 100644
--- a/test/brpc_redis_unittest.cpp
+++ b/test/brpc_redis_unittest.cpp
@@ -116,6 +116,20 @@ private:
int32_t _old_depth;
};
+class ScopedRedisMaxAllocationSize {
+public:
+ explicit ScopedRedisMaxAllocationSize(int32_t size)
+ : _old_size(brpc::FLAGS_redis_max_allocation_size) {
+ brpc::FLAGS_redis_max_allocation_size = size;
+ }
+ ~ScopedRedisMaxAllocationSize() {
+ brpc::FLAGS_redis_max_allocation_size = _old_size;
+ }
+
+private:
+ int32_t _old_size;
+};
+
class RedisTest : public testing::Test {
protected:
RedisTest() {}
@@ -921,6 +935,120 @@ TEST_F(RedisTest, redis_reply_rejects_deep_nested_arrays)
{
EXPECT_TRUE(valid_reply.is_array());
}
+TEST_F(RedisTest, redis_reply_rejects_nested_array_memory_amplification) {
+ // Each array allocation below is individually within -redis_max_
+ // allocation_size, yet the total for one reply must not exceed it.
+ // Otherwise the per-allocation cap is multiplied by nesting depth and a
+ // tiny reply commits depth * cap bytes.
+ ScopedRedisMaxAllocationSize scoped_size(1024);
+ const int32_t count_at_cap =
+ brpc::FLAGS_redis_max_allocation_size / sizeof(brpc::RedisReply);
+
+ // One maximal array nested inside another.
+ {
+ butil::IOBuf buf;
+ buf.append("*2\r\n");
+ buf.append("*" + std::to_string(count_at_cap) + "\r\n");
+ butil::Arena arena;
+ brpc::RedisReply reply(&arena);
+ EXPECT_EQ(brpc::PARSE_ERROR_ABSOLUTELY_WRONG,
+ reply.ConsumePartialIOBuf(buf));
+ }
+
+ // The budget must survive suspension: the outer header arrives first,
+ // the inner one only after a resume. A fresh budget per call would let
+ // the peer reset it by withholding data.
+ {
+ butil::IOBuf buf;
+ buf.append("*2\r\n");
+ butil::Arena arena;
+ brpc::RedisReply reply(&arena);
+ EXPECT_EQ(brpc::PARSE_ERROR_NOT_ENOUGH_DATA,
+ reply.ConsumePartialIOBuf(buf));
+ buf.append("*" + std::to_string(count_at_cap) + "\r\n");
+ EXPECT_EQ(brpc::PARSE_ERROR_ABSOLUTELY_WRONG,
+ reply.ConsumePartialIOBuf(buf));
+ }
+
+ // Feeding one array header per call must not bypass the budget either.
+ {
+ butil::IOBuf buf;
+ butil::Arena arena;
+ brpc::RedisReply reply(&arena);
+ // Each "*2" costs sizeof(RedisReply) * 2 = 64 bytes, so the 17th
+ // nested array exceeds the 1024-byte budget.
+ for (int i = 0; i < 20; ++i) {
+ buf.append("*2\r\n");
+ brpc::ParseError err = reply.ConsumePartialIOBuf(buf);
+ if (i < 16) {
+ EXPECT_EQ(brpc::PARSE_ERROR_NOT_ENOUGH_DATA, err);
+ } else {
+ EXPECT_EQ(brpc::PARSE_ERROR_ABSOLUTELY_WRONG, err);
+ break;
+ }
+ }
+ }
+
+ // Valid replies are unaffected: a nested reply well within the budget...
+ {
+ butil::IOBuf buf;
+ buf.append("*2\r\n*2\r\n:1\r\n:2\r\n:3\r\n");
+ butil::Arena arena;
+ brpc::RedisReply reply(&arena);
+ EXPECT_EQ(brpc::PARSE_OK, reply.ConsumePartialIOBuf(buf));
+ EXPECT_TRUE(reply.is_array());
+ EXPECT_EQ(2u, reply.size());
+ EXPECT_EQ(2u, reply[0].size());
+ EXPECT_EQ(1, reply[0][0].integer());
+ }
+ // ... and a single flat array exactly at the budget boundary.
+ {
+ butil::IOBuf buf;
+ buf.append("*" + std::to_string(count_at_cap) + "\r\n");
+ for (int i = 0; i < count_at_cap; ++i) {
+ buf.append(":7\r\n");
+ }
+ butil::Arena arena;
+ brpc::RedisReply reply(&arena);
+ EXPECT_EQ(brpc::PARSE_OK, reply.ConsumePartialIOBuf(buf));
+ EXPECT_TRUE(reply.is_array());
+ EXPECT_EQ((size_t)count_at_cap, reply.size());
+ }
+}
+
+TEST_F(RedisTest, command_parser_does_not_preallocate_declared_args) {
+ // A declared RESP array count must not commit memory before the arguments
+ // actually arrive: "*<count>\r\n" used to resize _args upfront, costing
+ // up to -redis_max_allocation_size per connection from ~12 bytes.
+ ScopedRedisMaxAllocationSize scoped_size(1024 * 1024);
+ const int32_t count_at_cap =
+ brpc::FLAGS_redis_max_allocation_size / sizeof(butil::StringPiece);
+
+ brpc::RedisCommandParser parser;
+ butil::Arena arena;
+ butil::IOBuf buf;
+ buf.append("*" + std::to_string(count_at_cap) + "\r\n$3\r\nget\r\n");
+ std::vector<butil::StringPiece> args;
+ EXPECT_EQ(brpc::PARSE_ERROR_NOT_ENOUGH_DATA,
+ parser.Consume(buf, &args, &arena));
+ // Only the argument that has arrived is stored.
+ EXPECT_EQ(1u, parser.ParsedArgsSize());
+
+ // Complete commands still parse into the full argument list.
+ {
+ brpc::RedisCommandParser parser2;
+ butil::Arena arena2;
+ butil::IOBuf buf2;
+ buf2.append("*3\r\n$3\r\nget\r\n$3\r\nfoo\r\n$3\r\nbar\r\n");
+ std::vector<butil::StringPiece> args2;
+ EXPECT_EQ(brpc::PARSE_OK, parser2.Consume(buf2, &args2, &arena2));
+ ASSERT_EQ(3u, args2.size());
+ EXPECT_EQ("get", args2[0].as_string());
+ EXPECT_EQ("foo", args2[1].as_string());
+ EXPECT_EQ("bar", args2[2].as_string());
+ }
+}
+
butil::Mutex s_mutex;
std::unordered_map<std::string, std::string> m;
std::unordered_map<std::string, int64_t> int_map;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
