From 9e11dd7a3ac1da816c3f146eec0ef4cb225783ae Mon Sep 17 00:00:00 2001 From: wwbmmm Date: Sat, 22 Aug 2026 16:02:38 +0800 Subject: [PATCH 1/2] Harden HTTP/2 receive path against unbounded resource growth The HTTP/2 (h2) parser advertised several receive-side limits that were never enforced, letting one hostile peer grow process memory without bound: - SETTINGS_MAX_CONCURRENT_STREAMS defaulted to unlimited and was never checked on the server; every odd stream id was accepted, so a peer could pin an unbounded number of streams per connection. The default is now a bounded 1024 and the server rejects streams beyond it with RST_STREAM(REFUSED_STREAM). - SETTINGS_MAX_HEADER_LIST_SIZE defaulted to unlimited and was never enforced on receive; decoded headers amplified by 1-byte HPACK indexed references ("HPACK bomb") and never-completed HEADERS/CONTINUATION fragments accumulated without bound. Both are now capped at a 1MB default. - The HPACK dynamic table size update was validated against the protocol default instead of the size this endpoint advertised, so a decoder configured with a smaller table could be forced to grow it beyond the capacity it was sized for. Updates above the advertised beyond the capacity it was sized for. Updates above the advertisedondi beyond the capacity it was sized for. Updates above the advertisg P beyETT beyond the capacity it was sized for. Updates above the advertisy wit beyond the capacity it was sized for. Updates above the advertisek_ beyond the capacity it was sized for. Updates above the advertisis de beyond the capacity it was sized for. Updates above the advertisedse beyond the capacity it wasfo beyond the capacity it was sized for. Updates above the adver continuing to parse whatever the peer keeps sending. Tests: new casesTests: new casesTests: new casesTests: new casesTests: nest cover each enforcement; the full brpc_hpack_unittest, brpc_http_rpc_protocol_unittest, brpc_h2_unsent_message_unittest and brpc_grpc_protocol_unittest suites pass. --- src/brpc/details/hpack.cpp | 12 +- src/brpc/details/hpack.h | 5 + src/brpc/http2.cpp | 10 +- src/brpc/http2.h | 16 ++- src/brpc/policy/http2_rpc_protocol.cpp | 89 ++++++++++++- src/brpc/policy/http2_rpc_protocol.h | 7 +- test/brpc_hpack_unittest.cpp | 28 +++++ test/brpc_http_rpc_protocol_unittest.cpp | 152 +++++++++++++++++++++++ 8 files changed, 308 insertions(+), 11 deletions(-) diff --git a/src/brpc/details/hpack.cpp b/src/brpc/details/hpack.cpp index 7a7bb366b9..107ec38605 100644 --- a/src/brpc/details/hpack.cpp +++ b/src/brpc/details/hpack.cpp @@ -659,7 +659,8 @@ inline ssize_t DecodeString(butil::IOBufBytesIterator& iter, std::string* out) { HPacker::HPacker() : _encode_table(nullptr) - , _decode_table(nullptr) { + , _decode_table(nullptr) + , _max_table_size(0) { CreateStaticTableOnceOrDie(); } @@ -677,6 +678,7 @@ HPacker::~HPacker() { int HPacker::Init(size_t max_table_size) { CHECK(!_encode_table); CHECK(!_decode_table); + _max_table_size = max_table_size; IndexTableOptions encode_table_options; encode_table_options.max_size = max_table_size; encode_table_options.start_index = s_static_table->end_index(); @@ -844,7 +846,13 @@ ssize_t HPacker::Decode(butil::IOBufBytesIterator& iter, Header* h) { if (read_bytes <= 0) { return read_bytes; } - if (max_size > H2Settings::DEFAULT_HEADER_TABLE_SIZE) { + if (max_size > _max_table_size) { + // RFC 7541 section 6.3: the new maximum size MUST be lower + // than or equal to the limit determined by the protocol using + // HPACK, i.e. the SETTINGS_HEADER_TABLE_SIZE this decoder + // advertised (what Init() was called with), not the protocol + // default. Growing past the Init-time size would also desync + // the fixed-capacity header queue from the byte accounting. LOG(ERROR) << "Invalid max_size=" << max_size; return -1; } diff --git a/src/brpc/details/hpack.h b/src/brpc/details/hpack.h index 05a5f30727..bd1da32114 100644 --- a/src/brpc/details/hpack.h +++ b/src/brpc/details/hpack.h @@ -132,6 +132,11 @@ class HPacker : public Describable { IndexTable* _encode_table; IndexTable* _decode_table; + // The max dynamic table size this decoder was initialized with, i.e. + // the SETTINGS_HEADER_TABLE_SIZE we advertised to the peer. Per RFC + // 7541 section 4.2/6.3 a dynamic table size update exceeding it must + // be treated as a decoding error. + size_t _max_table_size; }; // Lowercase the input string, a fast implementation. diff --git a/src/brpc/http2.cpp b/src/brpc/http2.cpp index 59099c2dd0..4b43e36359 100644 --- a/src/brpc/http2.cpp +++ b/src/brpc/http2.cpp @@ -23,14 +23,20 @@ namespace brpc { +// Out-of-class definitions for the in-class initialized integral constants: +// required when the constants are odr-used (e.g. passed by reference in +// ASSERT_EQ). +const uint32_t H2Settings::DEFAULT_MAX_CONCURRENT_STREAMS; +const uint32_t H2Settings::DEFAULT_MAX_HEADER_LIST_SIZE; + H2Settings::H2Settings() : header_table_size(DEFAULT_HEADER_TABLE_SIZE) , enable_push(false) - , max_concurrent_streams(std::numeric_limits::max()) + , max_concurrent_streams(DEFAULT_MAX_CONCURRENT_STREAMS) , stream_window_size(256 * 1024) , connection_window_size(1024 * 1024) , max_frame_size(DEFAULT_MAX_FRAME_SIZE) - , max_header_list_size(std::numeric_limits::max()) { + , max_header_list_size(DEFAULT_MAX_HEADER_LIST_SIZE) { } bool H2Settings::IsValid(bool log_error) const { diff --git a/src/brpc/http2.h b/src/brpc/http2.h index 69d30874fb..88ae8353cb 100644 --- a/src/brpc/http2.h +++ b/src/brpc/http2.h @@ -60,7 +60,12 @@ struct H2Settings { // for any limit that is exhausted with active streams. Servers SHOULD only // set a zero value for short durations; if a server does not wish to // accept requests, closing the connection is more appropriate. - // Default: unlimited + // The server enforces this limit: streams opened beyond it are rejected + // with RST_STREAM(REFUSED_STREAM). Set to + // std::numeric_limits::max() explicitly to restore the old + // unlimited (unenforced) behavior. + // Default: 1024 + static const uint32_t DEFAULT_MAX_CONCURRENT_STREAMS = 1024; uint32_t max_concurrent_streams; // Sender's initial window size (in octets) for stream-level flow control. @@ -92,7 +97,14 @@ struct H2Settings { // and value in octets plus an overhead of 32 octets for each header field. // For any given request, a lower limit than what is advertised MAY be // enforced. - // Default: unlimited. + // brpc enforces this limit on received header blocks to bound per-stream + // memory: HPACK indexed references would otherwise let a small + // HEADERS/CONTINUATION frame expand into an unbounded header list + // ("HPACK bomb"). Set to std::numeric_limits::max() to restore + // the old unlimited behavior (not recommended for servers exposed to + // untrusted peers). + // Default: 1MB + static const uint32_t DEFAULT_MAX_HEADER_LIST_SIZE = 1024 * 1024; uint32_t max_header_list_size; }; diff --git a/src/brpc/policy/http2_rpc_protocol.cpp b/src/brpc/policy/http2_rpc_protocol.cpp index af9ed0b151..24f62cde17 100644 --- a/src/brpc/policy/http2_rpc_protocol.cpp +++ b/src/brpc/policy/http2_rpc_protocol.cpp @@ -42,12 +42,22 @@ DEFINE_int32(h2_client_connection_window_size, 1024 * 1024, DEFINE_int32(h2_client_max_frame_size, H2Settings::DEFAULT_MAX_FRAME_SIZE, "Size of the largest frame payload that client is willing to receive"); +DEFINE_int32(h2_client_max_header_list_size, + H2Settings::DEFAULT_MAX_HEADER_LIST_SIZE, + "Maximum decoded size of a header list that the client accepts" + " on a received stream, 0 or negative means unlimited"); DEFINE_bool(h2_hpack_encode_name, false, "Encode name in HTTP2 headers with huffman encoding"); DEFINE_bool(h2_hpack_encode_value, false, "Encode value in HTTP2 headers with huffman encoding"); +DEFINE_bool(h2_ack_ignore_eovercrowded, false, + "Let h2 control-frame replies (PING/SETTINGS acks, RST_STREAM, " + "GOAWAY, WINDOW_UPDATE) bypass -socket_max_unwritten_bytes. " + "Dangerous: a peer flooding PING frames while withholding TCP " + "reads then grows this process's memory without bound."); + static bool CheckStreamWindowSize(const char*, int32_t val) { return val >= 0; } @@ -144,13 +154,19 @@ static int WriteAck(Socket* s, const void* data, size_t n) { butil::IOBuf sendbuf; sendbuf.append(data, n); Socket::WriteOptions wopt; - wopt.ignore_eovercrowded = true; + // These writes historically ignored EOVERCROWDED so control replies + // could always be queued, but that defeats the explicit memory bound of + // -socket_max_unwritten_bytes: a peer flooding PING/SETTINGS frames + // while withholding TCP reads makes this process buffer ack frames + // without limit. Respect the bound by default; callers treat a failed + // ack write as a connection error and close the overcrowded connection. + wopt.ignore_eovercrowded = FLAGS_h2_ack_ignore_eovercrowded; return s->Write(&sendbuf, &wopt); } static int WriteAck(Socket* s, butil::IOBuf* data) { Socket::WriteOptions wopt; - wopt.ignore_eovercrowded = true; + wopt.ignore_eovercrowded = FLAGS_h2_ack_ignore_eovercrowded; return s->Write(data, &wopt); } @@ -338,6 +354,10 @@ H2Context::H2Context(Socket* socket, const Server* server) // SETTINGS_INITIAL_WINDOW_SIZE defaults to 65535 until the peer sends a // different value. Larger requests are resumed by WINDOW_UPDATE. _remote_settings.stream_window_size = H2Settings::DEFAULT_INITIAL_WINDOW_SIZE; + // RFC 7540 section 6.5.2: the peer's SETTINGS_MAX_CONCURRENT_STREAMS + // defaults to unlimited until its SETTINGS frame is received. + _remote_settings.max_concurrent_streams = + std::numeric_limits::max(); if (server) { _unack_local_settings = server->options().h2_settings; } else { @@ -345,6 +365,10 @@ H2Context::H2Context(Socket* socket, const Server* server) _unack_local_settings.stream_window_size = FLAGS_h2_client_stream_window_size; _unack_local_settings.max_frame_size = FLAGS_h2_client_max_frame_size; _unack_local_settings.connection_window_size = FLAGS_h2_client_connection_window_size; + _unack_local_settings.max_header_list_size = + FLAGS_h2_client_max_header_list_size > 0 + ? (uint32_t)FLAGS_h2_client_max_header_list_size + : std::numeric_limits::max(); } #if defined(UNIT_TEST) // In ut, we hope _last_sent_stream_id run out quickly to test the correctness @@ -444,6 +468,16 @@ int H2Context::TryToInsertStream(int stream_id, H2StreamContext* ctx) { if (_goaway_stream_id >= 0 && stream_id > _goaway_stream_id) { return 1; } + // Enforce the SETTINGS_MAX_CONCURRENT_STREAMS value the server + // advertised. Use _unack_local_settings so that a client which never + // ACKs our SETTINGS frame cannot dodge the limit (_local_settings is + // only synchronized on ACK). Client-side streams are already bounded + // against the remote peer's setting before insertion (see + // H2UnsentRequest::AppendAndDestroySelf). + if (is_server_side() && + _pending_streams.size() >= _unack_local_settings.max_concurrent_streams) { + return 2; + } H2StreamContext*& sctx = _pending_streams[stream_id]; if (sctx == nullptr) { // Synchronize creation with SETTINGS_INITIAL_WINDOW_SIZE updates. @@ -555,7 +589,11 @@ ParseResult H2Context::Consume( LOG(WARNING) << "Fail to send GOAWAY to " << *_socket; return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); } - return MakeMessage(nullptr); + // https://tools.ietf.org/html/rfc7540#section-5.4.1 + // A connection error is unrecoverable: close the connection + // after sending GOAWAY instead of continuing to parse (and + // buffer) whatever the misbehaving peer keeps sending. + return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); } } else { return MakeParseError(PARSE_ERROR_NO_RESOURCE); @@ -610,6 +648,18 @@ H2ParseResult H2Context::OnHeaders( delete sctx; LOG(ERROR) << "Fail to insert existing stream_id=" << frame_head.stream_id; return MakeH2Error(H2_PROTOCOL_ERROR); + } else if (rc == 2) { + delete sctx; + LOG_EVERY_SECOND(WARNING) + << "Refused stream_id=" << frame_head.stream_id + << " since concurrent streams reached max_concurrent_streams=" + << _unack_local_settings.max_concurrent_streams + << " on " << *_socket; + // A stream error (RST_STREAM) rather than a connection error: + // RFC 7540 section 5.1.2 requires REFUSED_STREAM (or + // PROTOCOL_ERROR) for streams exceeding the advertised limit, + // and REFUSED_STREAM lets a compliant client retry later. + return MakeH2Error(H2_REFUSED_STREAM, frame_head.stream_id); } else if (rc > 0) { delete sctx; return MakeH2Error(H2_REFUSED_STREAM); @@ -695,6 +745,19 @@ H2ParseResult H2StreamContext::OnContinuation( butil::IOBufBytesIterator& it, const H2FrameHead& frame_head) { _parsed_length += FRAME_HEAD_SIZE + frame_head.payload_size; it.append_and_forward(&_remaining_header_fragment, frame_head.payload_size); + // A header block may span many CONTINUATION frames; ConsumeHeaders() + // drains complete fields, so the fragment only buffers one incomplete + // field, whose wire size never legitimately exceeds the decoded header + // list limit. Without this cap a never-completed field (e.g. a huge + // declared string length) accumulates unbounded memory. + if (_remaining_header_fragment.size() > + _conn_ctx->_unack_local_settings.max_header_list_size) { + LOG(ERROR) << "Accumulated header fragment exceeds" + " max_header_list_size=" + << _conn_ctx->_unack_local_settings.max_header_list_size + << ", stream_id=" << frame_head.stream_id; + return MakeH2Error(H2_ENHANCE_YOUR_CALM); + } const size_t size = _remaining_header_fragment.size(); butil::IOBufBytesIterator it2(_remaining_header_fragment); if (ConsumeHeaders(it2) < 0) { @@ -1130,6 +1193,10 @@ void H2Context::DeferWindowUpdate(int64_t size) { SaveUint32(winbuf + FRAME_HEAD_SIZE, conn_wu); if (WriteAck(_socket, winbuf, sizeof(winbuf)) != 0) { LOG(WARNING) << "Fail to send WINDOW_UPDATE"; + // Retry on a later DATA frame instead of silently losing + // the window bytes (the peer would stall otherwise). + _deferred_window_update.fetch_add( + conn_wu, butil::memory_order_relaxed); } } } @@ -1207,7 +1274,8 @@ H2StreamContext::H2StreamContext(bool read_body_progressively) , _stream_ended(false) , _remote_window_left(0) , _deferred_window_update(0) - , _correlation_id(INVALID_BTHREAD_ID.value) { + , _correlation_id(INVALID_BTHREAD_ID.value) + , _decoded_header_list_size(0) { header().set_version(2, 0); #ifndef NDEBUG get_h2_bvars()->h2_stream_context_count << 1; @@ -1240,6 +1308,13 @@ void H2StreamContext::SetState(H2StreamState state) { int H2StreamContext::ConsumeHeaders(butil::IOBufBytesIterator& it) { HPacker& hpacker = _conn_ctx->hpacker(); HttpHeader& h = header(); + // https://tools.ietf.org/html/rfc7540#section-10.5.1 + // Bound the cumulative decoded size of the header list. Without this + // check nothing limits header bytes (-max_body_size covers DATA only) + // and 1-byte HPACK indexed references to a large dynamic-table entry + // amplify a small HEADERS/CONTINUATION frame ~4000x ("HPACK bomb"). + const uint32_t max_header_list_size = + _conn_ctx->_unack_local_settings.max_header_list_size; while (it) { HPacker::Header pair; const int rc = hpacker.Decode(it, &pair); @@ -1249,6 +1324,12 @@ int H2StreamContext::ConsumeHeaders(butil::IOBufBytesIterator& it) { if (rc == 0) { break; } + _decoded_header_list_size += pair.name.size() + pair.value.size() + 32; + if (_decoded_header_list_size > max_header_list_size) { + LOG(ERROR) << "Decoded header list exceeds max_header_list_size=" + << max_header_list_size << ", stream_id=" << _stream_id; + return -1; + } const char* const name = pair.name.c_str(); bool matched = false; if (name[0] == ':') { // reserved names diff --git a/src/brpc/policy/http2_rpc_protocol.h b/src/brpc/policy/http2_rpc_protocol.h index 022f978c85..41c952d6f3 100644 --- a/src/brpc/policy/http2_rpc_protocol.h +++ b/src/brpc/policy/http2_rpc_protocol.h @@ -273,6 +273,10 @@ friend class H2Context; butil::atomic _remote_window_left; butil::atomic _deferred_window_update; uint64_t _correlation_id; + // Cumulative decoded size of the header list of this stream + // (name + value + 32 per field, RFC 7540 section 10.5.1), checked + // against the local max_header_list_size in ConsumeHeaders(). + uint64_t _decoded_header_list_size; butil::IOBuf _remaining_header_fragment; // Request body which cannot be sent yet due to remote flow control. // Accessed under H2Context::_stream_mutex. @@ -336,7 +340,8 @@ class H2Context : public Destroyable, public Describable { int AllocateClientStreamId(); bool RunOutStreams() const; // Try to map stream_id to ctx if stream_id does not exist before - // Returns 0 on success, -1 on exist, 1 on goaway. + // Returns 0 on success, -1 on exist, 1 on goaway, 2 on exceeding + // the local max_concurrent_streams limit (server side). int TryToInsertStream(int stream_id, H2StreamContext* ctx); size_t VolatilePendingStreamSize() const; bool PendingDataOvercrowded() const; diff --git a/test/brpc_hpack_unittest.cpp b/test/brpc_hpack_unittest.cpp index 191bf003da..7ae65d53bf 100644 --- a/test/brpc_hpack_unittest.cpp +++ b/test/brpc_hpack_unittest.cpp @@ -96,6 +96,34 @@ TEST_F(HPackTest, dynamic_table_size_update_before_header) { ASSERT_EQ("GET", h.value); } +TEST_F(HPackTest, dynamic_table_size_update_over_advertised_limit) { + // RFC 7541 section 4.2/6.3: a dynamic table size update must not exceed + // the max size this decoder advertised via SETTINGS_HEADER_TABLE_SIZE + // (i.e. what Init() was called with), not the protocol default of 4096. + // Previously the update was validated against the compile-time default, + // so a decoder initialized with a smaller table accepted a peer update + // that regrew the table beyond the queue it was sized for. + brpc::HPacker p; + ASSERT_EQ(0, p.Init(256)); + { + // Update to 256 (== advertised limit) is acceptable. + butil::IOBuf buf; + uint8_t ok[] = { 0x3F, 0xE1, 0x01 }; // size update to 256 + buf.append(ok, sizeof(ok)); + brpc::HPacker::Header h; + ASSERT_GE(p.Decode(&buf, &h), 0); + } + { + // Update to 4096 (> advertised 256) must be rejected as malformed, + // even though it equals the protocol default. + butil::IOBuf buf; + uint8_t bad[] = { 0x3F, 0xE1, 0x1F }; // size update to 4096 + buf.append(bad, sizeof(bad)); + brpc::HPacker::Header h; + ASSERT_EQ(-1, p.Decode(&buf, &h)); + } +} + TEST_F(HPackTest, integer_with_overlong_continuation) { brpc::HPacker p; ASSERT_EQ(0, p.Init(4096)); diff --git a/test/brpc_http_rpc_protocol_unittest.cpp b/test/brpc_http_rpc_protocol_unittest.cpp index df5c838f5b..f86a392a2e 100644 --- a/test/brpc_http_rpc_protocol_unittest.cpp +++ b/test/brpc_http_rpc_protocol_unittest.cpp @@ -61,6 +61,7 @@ DECLARE_int32(rpc_dump_max_requests_in_one_file); DECLARE_bool(allow_chunked_length); DECLARE_int32(max_connection_pool_size); DECLARE_uint64(max_body_size); +DECLARE_int64(socket_max_unwritten_bytes); extern bvar::CollectorSpeedLimit g_rpc_dump_sl; } @@ -1539,6 +1540,11 @@ TEST_F(HttpTest, http2_window_used_up_buffers_request) { char settingsbuf[brpc::policy::FRAME_HEAD_SIZE + 36]; brpc::H2Settings h2_settings; + // The fake server advertises an unlimited stream count so the test can + // fill the flow-control window with many streams: this test exercises + // WINDOW_UPDATE buffering, not SETTINGS_MAX_CONCURRENT_STREAMS (which + // defaults to a bounded value now). + h2_settings.max_concurrent_streams = std::numeric_limits::max(); const size_t nb = brpc::policy::SerializeH2Settings(h2_settings, settingsbuf + brpc::policy::FRAME_HEAD_SIZE); brpc::policy::SerializeFrameHead(settingsbuf, nb, brpc::policy::H2_FRAME_SETTINGS, 0, 0); butil::IOBuf buf; @@ -1600,6 +1606,152 @@ TEST_F(HttpTest, http2_settings) { ASSERT_TRUE(ctx->_remote_settings.stream_window_size == (1u << 29) - 1); } +TEST_F(HttpTest, http2_header_list_size_limit) { + // HPACK bomb regression: 1-byte indexed references to a large + // dynamic-table entry amplify a tiny HEADERS frame into an unbounded + // decoded header list. SETTINGS_MAX_HEADER_LIST_SIZE is advertise-but- + // never-enforce without the fix, so ConsumeHeaders accepts the overflow. + // The default must be bounded, not unlimited. + brpc::H2Settings default_settings; + ASSERT_EQ(brpc::H2Settings::DEFAULT_MAX_HEADER_LIST_SIZE, + default_settings.max_header_list_size); + + brpc::policy::H2Context* ctx = + new brpc::policy::H2Context(_socket.get(), nullptr); + CHECK_EQ(ctx->Init(), 0); + _socket->initialize_parsing_context(&ctx); + // Tiny limit so the amplification is visible in a few bytes. + ctx->_unack_local_settings.max_header_list_size = 4096; + + brpc::policy::H2StreamContext sctx(false); + sctx.Init(ctx, 1); + + // Literal header field with incremental indexing: name "x", value 3000 + // bytes, entry size = 3000 + 1 + 32 = 3033, which fits both the dynamic + // table (4096) and the decoded-list budget (4096). + butil::IOBuf buf; + const uint8_t literal_prefix[] = { 0x40, 0x01, 'x', 0x7F, 0xB9, 0x16 }; + buf.append(literal_prefix, sizeof(literal_prefix)); + buf.append(std::string(3000, 'v')); + { + butil::IOBufBytesIterator it(buf); + ASSERT_EQ(0, sctx.ConsumeHeaders(it)); + } + + // One more header referencing the table entry (dynamic index 62) costs + // another 3033 decoded bytes and must be refused: the decoded list would + // reach 6066 > 4096. + butil::IOBuf ref_buf; + const uint8_t indexed_ref[] = { 0xBE }; // indexed entry 62 + ref_buf.append(indexed_ref, sizeof(indexed_ref)); + { + butil::IOBufBytesIterator it(ref_buf); + ASSERT_LT(sctx.ConsumeHeaders(it), 0); + } + + // A list that stays within the limit keeps working. + brpc::policy::H2StreamContext sctx2(false); + sctx2.Init(ctx, 3); + butil::IOBuf ok_buf; + const uint8_t ok_prefix[] = { 0x40, 0x01, 'y', 0x01, 'v' }; + ok_buf.append(ok_prefix, sizeof(ok_prefix)); + { + butil::IOBufBytesIterator it(ok_buf); + ASSERT_EQ(0, sctx2.ConsumeHeaders(it)); + } +} + +TEST_F(HttpTest, http2_server_enforces_max_concurrent_streams) { + // The server advertises SETTINGS_MAX_CONCURRENT_STREAMS but never + // enforced it: every odd stream id was accepted, so a peer could pin an + // unbounded number of streams per connection. Pending streams beyond the + // limit must be refused. + // The default must be bounded, not unlimited. + brpc::H2Settings default_settings; + ASSERT_EQ(brpc::H2Settings::DEFAULT_MAX_CONCURRENT_STREAMS, + default_settings.max_concurrent_streams); + + brpc::policy::H2Context* ctx = + new brpc::policy::H2Context(_socket.get(), nullptr); + CHECK_EQ(ctx->Init(), 0); + _socket->initialize_parsing_context(&ctx); + ASSERT_TRUE(ctx->is_server_side()); + ctx->_unack_local_settings.max_concurrent_streams = 2; + + brpc::policy::H2StreamContext* s1 = new brpc::policy::H2StreamContext(false); + s1->Init(ctx, 1); + ASSERT_EQ(0, ctx->TryToInsertStream(1, s1)); + brpc::policy::H2StreamContext* s3 = new brpc::policy::H2StreamContext(false); + s3->Init(ctx, 3); + ASSERT_EQ(0, ctx->TryToInsertStream(3, s3)); + + // The third concurrent stream must be refused (rc=2 -> REFUSED_STREAM) + // instead of inserted. + brpc::policy::H2StreamContext s5(false); + s5.Init(ctx, 5); + ASSERT_EQ(2, ctx->TryToInsertStream(5, &s5)); + ASSERT_EQ(2u, ctx->VolatilePendingStreamSize()); + + // Closing a stream frees a slot again. + delete ctx->RemoveStreamAndDeferWU(1); + brpc::policy::H2StreamContext* s7 = new brpc::policy::H2StreamContext(false); + s7->Init(ctx, 7); + ASSERT_EQ(0, ctx->TryToInsertStream(7, s7)); +} + +TEST_F(HttpTest, h2_ping_ack_respects_socket_write_cap) { + // A peer flooding PING frames while withholding TCP reads must not make + // this side queue PONGs past -socket_max_unwritten_bytes. Use a dedicated + // pipe so the fixture pipe stays readable for other tests. + int fds[2]; + ASSERT_EQ(0, pipe(fds)); + brpc::SocketOptions options; + options.fd = fds[1]; + brpc::SocketId id; + ASSERT_EQ(0, brpc::Socket::Create(options, &id)); + brpc::SocketUniquePtr sock; + ASSERT_EQ(0, brpc::Socket::Address(id, &sock)); + + GFLAGS_NAMESPACE::FlagSaver flag_saver; + brpc::FLAGS_socket_max_unwritten_bytes = 64; + + // Write far more than the pipe capacity (nobody reads the other end), so + // bytes stay queued well past the cap and the socket is overcrowded. + butil::IOBuf big; + big.append(std::string(1024 * 1024, 'x')); + brpc::Socket::WriteOptions wopt; + wopt.ignore_eovercrowded = false; + ASSERT_EQ(0, sock->Write(&big, &wopt)); + ASSERT_TRUE(sock->_overcrowded); + + brpc::policy::H2Context* ctx = + new brpc::policy::H2Context(sock.get(), nullptr); + CHECK_EQ(ctx->Init(), 0); + sock->initialize_parsing_context(&ctx); + ctx->_conn_state = brpc::policy::H2_CONNECTION_READY; + + // The PING must not be answered by bypassing the write cap: when the + // connection is overcrowded the ack write fails, the connection errors + // out (RFC 7540 5.4.1) and no PONG is queued. Before the fix the ack + // write always succeeded (ignore_eovercrowded) and the connection kept + // asking for more data, so this parse ended with NOT_ENOUGH_DATA. + butil::IOBuf ping; + char pingbuf[brpc::policy::FRAME_HEAD_SIZE + 8]; + brpc::policy::SerializeFrameHead( + pingbuf, 8, brpc::policy::H2_FRAME_PING, 0, 0); + ping.append(pingbuf, sizeof(pingbuf)); + ping.append(std::string(8, '\0')); // opaque payload + const int64_t before = + sock->_unwritten_bytes.load(butil::memory_order_relaxed); + const brpc::ParseResult pr = + brpc::policy::ParseH2Message(&ping, sock.get(), false, nullptr); + ASSERT_EQ(before, + sock->_unwritten_bytes.load(butil::memory_order_relaxed)); + // The overcrowded connection must terminate parsing instead of accepting + // the PING and queueing a PONG past the socket write cap. + ASSERT_EQ(brpc::PARSE_ERROR_ABSOLUTELY_WRONG, pr.error()); +} + TEST_F(HttpTest, http2_goaway_with_debug_data) { // GOAWAY payload is Last-Stream-ID(4) | Error Code(4) | Debug Data(*). const uint32_t payload_size = 16; From 979491b6f36ee9a7173a3a9fa0d17b0b2f0dc718 Mon Sep 17 00:00:00 2001 From: wwbmmm Date: Sat, 22 Aug 2026 17:16:10 +0800 Subject: [PATCH 2/2] Reset h2 header-list budget per header block and cap initial HEADERS fragment Two follow-up fixes from code review of the HTTP/2 receive-path hardening: - The decoded header-list size counter was never reset when a new HEADERS block began, so trailers on the same stream inherited the budget of the initial header list and could be rejected even though they form a separate list (RFC 7540 section 10.5.1 applies per header block). Reset the counter at the start of OnHeaders; it stays cumulative across the CONTINUATION frames that complete the same block. - The _remaining_header_fragment size cap was only checked when CONTINUATION frames were processed, but a single oversized HEADERS frame appends its unconsumed tail into the fragment too. Apply the same cap in OnHeaders via a shared HeaderFragmentTooLarge() helper and fail with ENHANCE_YOUR_CALM. Tests: brpc_http_rpc_protocol_unittest gains two cases (h2_header_list_budget_resets_per_block and h2_oversized_single_headers_bloh2_oversized_single_headers_bloh2_oversized_single_headers_bloh2_oversized_single_headers_b_message_unittest and brpc_grpc_protocol_unittest suites pass. --- src/brpc/policy/http2_rpc_protocol.cpp | 23 +++++++- src/brpc/policy/http2_rpc_protocol.h | 4 ++ test/brpc_http_rpc_protocol_unittest.cpp | 75 ++++++++++++++++++++++++ 3 files changed, 100 insertions(+), 2 deletions(-) diff --git a/src/brpc/policy/http2_rpc_protocol.cpp b/src/brpc/policy/http2_rpc_protocol.cpp index 24f62cde17..2f4b599004 100644 --- a/src/brpc/policy/http2_rpc_protocol.cpp +++ b/src/brpc/policy/http2_rpc_protocol.cpp @@ -683,6 +683,11 @@ H2ParseResult H2Context::OnHeaders( return sctx->OnHeaders(it, frame_head, frag_size, pad_length); } +bool H2StreamContext::HeaderFragmentTooLarge() const { + return _remaining_header_fragment.size() > + _conn_ctx->_unack_local_settings.max_header_list_size; +} + H2ParseResult H2StreamContext::OnHeaders( butil::IOBufBytesIterator& it, const H2FrameHead& frame_head, uint32_t frag_size, uint8_t pad_length) { @@ -690,6 +695,12 @@ H2ParseResult H2StreamContext::OnHeaders( #if defined(BRPC_H2_STREAM_STATE) SetState(H2_STREAM_OPEN); #endif + // A new HEADERS block (which may be an initial request header set, or + // trailing headers on the same stream) starts here. The decoded header + // list budget is per header block (RFC 7540 section 10.5.1), so reset the + // counter; it stays cumulative across the CONTINUATION frames that finish + // this same block. + _decoded_header_list_size = 0; butil::IOBufBytesIterator it2(it, frag_size); if (ConsumeHeaders(it2) < 0) { LOG(ERROR) << "Invalid header, frag_size=" << frag_size @@ -701,6 +712,15 @@ H2ParseResult H2StreamContext::OnHeaders( if (it2.bytes_left()) { it.append_and_forward(&_remaining_header_fragment, it2.bytes_left()); + // A single HEADERS frame can carry more than max_header_list_size of + // an incomplete header field; cap it here just like CONTINUATION. + if (HeaderFragmentTooLarge()) { + LOG(ERROR) << "Accumulated header fragment exceeds" + " max_header_list_size=" + << _conn_ctx->_unack_local_settings.max_header_list_size + << ", stream_id=" << frame_head.stream_id; + return MakeH2Error(H2_ENHANCE_YOUR_CALM); + } } it.forward(pad_length); if (frame_head.flags & H2_FLAGS_END_HEADERS) { @@ -750,8 +770,7 @@ H2ParseResult H2StreamContext::OnContinuation( // field, whose wire size never legitimately exceeds the decoded header // list limit. Without this cap a never-completed field (e.g. a huge // declared string length) accumulates unbounded memory. - if (_remaining_header_fragment.size() > - _conn_ctx->_unack_local_settings.max_header_list_size) { + if (HeaderFragmentTooLarge()) { LOG(ERROR) << "Accumulated header fragment exceeds" " max_header_list_size=" << _conn_ctx->_unack_local_settings.max_header_list_size diff --git a/src/brpc/policy/http2_rpc_protocol.h b/src/brpc/policy/http2_rpc_protocol.h index 41c952d6f3..6c58f71aa5 100644 --- a/src/brpc/policy/http2_rpc_protocol.h +++ b/src/brpc/policy/http2_rpc_protocol.h @@ -244,6 +244,10 @@ class H2StreamContext : public HttpContext { uint32_t frag_size, uint8_t pad_length); H2ParseResult OnContinuation(butil::IOBufBytesIterator&, const H2FrameHead&); H2ParseResult OnResetStream(H2Error h2_error, const H2FrameHead&); + + // True if the accumulated HEADERS/CONTINUATION fragment has grown past the + // local max_header_list_size. Bound every place the fragment is appended. + bool HeaderFragmentTooLarge() const; uint64_t correlation_id() const { return _correlation_id; } void set_correlation_id(uint64_t cid) { _correlation_id = cid; } diff --git a/test/brpc_http_rpc_protocol_unittest.cpp b/test/brpc_http_rpc_protocol_unittest.cpp index f86a392a2e..c650b6df97 100644 --- a/test/brpc_http_rpc_protocol_unittest.cpp +++ b/test/brpc_http_rpc_protocol_unittest.cpp @@ -1661,6 +1661,81 @@ TEST_F(HttpTest, http2_header_list_size_limit) { } } +TEST_F(HttpTest, h2_header_list_budget_resets_per_block) { + // The decoded header-list budget is per header block (RFC 7540 + // section 10.5.1): trailing headers on the same stream start a fresh + // block. The counter must be reset at the start of a new block, otherwise + // legit trailers are rejected once the first block used most of the + // budget. + brpc::policy::H2Context* ctx = + new brpc::policy::H2Context(_socket.get(), nullptr); + CHECK_EQ(ctx->Init(), 0); + _socket->initialize_parsing_context(&ctx); + ctx->_unack_local_settings.max_header_list_size = 4096; + + brpc::policy::H2StreamContext* sctx = + new brpc::policy::H2StreamContext(false); + sctx->Init(ctx, 1); + + // First block: a 3000-byte header costs 3033 of the 4096 budget. + butil::IOBuf first; + const uint8_t p1[] = { 0x40, 0x01, 'x', 0x7F, 0xB9, 0x16 }; + first.append(p1, sizeof(p1)); + first.append(std::string(3000, 'v')); + butil::IOBufBytesIterator it1(first); + ASSERT_EQ(0, sctx->ConsumeHeaders(it1)); + + // Second block (trailers) on the same stream: a 2000-byte header (2033 + // bytes) exceeds the remaining budget (4096 - 3033 = 1063) but must be + // accepted because its counter starts at zero. + butil::IOBuf second; + const uint8_t p2[] = { 0x40, 0x01, 'x', 0x7F, 0xD1, 0x0E }; + second.append(p2, sizeof(p2)); + second.append(std::string(2000, 'v')); + butil::IOBufBytesIterator it2(second); + brpc::policy::H2FrameHead head; + head.payload_size = second.size(); + head.type = brpc::policy::H2_FRAME_HEADERS; + head.flags = 0x4; // H2_FLAGS_END_HEADERS + head.stream_id = 1; + const brpc::policy::H2ParseResult res = + sctx->OnHeaders(it2, head, second.size(), 0); + ASSERT_TRUE(res.is_ok()); + delete sctx; +} + +TEST_F(HttpTest, h2_oversized_single_headers_block_rejected) { + // A single HEADERS frame whose decoded header list exceeds + // max_header_list_size must be rejected at the block boundary (before + // any CONTINUATION), not accepted into _remaining_header_fragment. + brpc::policy::H2Context* ctx = + new brpc::policy::H2Context(_socket.get(), nullptr); + CHECK_EQ(ctx->Init(), 0); + _socket->initialize_parsing_context(&ctx); + ctx->_unack_local_settings.max_header_list_size = 64; + + brpc::policy::H2StreamContext* sctx = + new brpc::policy::H2StreamContext(false); + sctx->Init(ctx, 1); + + // One literal-with-incremental-indexing header with a 200-byte value: + // its decoded cost (200 + 1 + 32 = 233) exceeds the 64-byte budget. + butil::IOBuf payload; + const uint8_t p[] = { 0x40, 0x01, 'x', 0x7F, 0xC1, 0x01 }; + payload.append(p, sizeof(p)); + payload.append(std::string(200, 'v')); + butil::IOBufBytesIterator it(payload); + brpc::policy::H2FrameHead head; + head.payload_size = payload.size(); + head.type = brpc::policy::H2_FRAME_HEADERS; + head.flags = 0x4; // H2_FLAGS_END_HEADERS + head.stream_id = 1; + const brpc::policy::H2ParseResult res = + sctx->OnHeaders(it, head, payload.size(), 0); + ASSERT_FALSE(res.is_ok()); + delete sctx; +} + TEST_F(HttpTest, http2_server_enforces_max_concurrent_streams) { // The server advertises SETTINGS_MAX_CONCURRENT_STREAMS but never // enforced it: every odd stream id was accepted, so a peer could pin an