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..2f4b599004 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); @@ -633,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) { @@ -640,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 @@ -651,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) { @@ -695,6 +765,18 @@ 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 (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); + } const size_t size = _remaining_header_fragment.size(); butil::IOBufBytesIterator it2(_remaining_header_fragment); if (ConsumeHeaders(it2) < 0) { @@ -1130,6 +1212,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 +1293,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 +1327,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 +1343,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..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; } @@ -273,6 +277,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 +344,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..c650b6df97 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,227 @@ 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, 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 + // 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;