Harden HTTP/2 receive path against unbounded resource growth (#3474)

* 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.

* 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.

Author: wwbmmm

src/brpc/details/hpack.cpp

@@ -657,7 +657,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();
 }
 
@@ -675,6 +676,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();
@@ -842,7 +844,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;
             }

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.

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<uint32_t>::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<uint32_t>::max()) {
+    , max_header_list_size(DEFAULT_MAX_HEADER_LIST_SIZE) {
 }
 
 bool H2Settings::IsValid(bool log_error) const {

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<uint32_t>::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<uint32_t>::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;
 };
 

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,13 +354,21 @@ 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<uint32_t>::max();
     if (server) {
         _unack_local_settings = server->options().h2_settings;
     } else {
         _unack_local_settings.header_table_size = FLAGS_h2_client_header_table_size;
         _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<uint32_t>::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,13 +683,24 @@ 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) {
     _parsed_length += FRAME_HEAD_SIZE + frame_head.payload_size;
 #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

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<int64_t> _remote_window_left;
     butil::atomic<int64_t> _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;

test/brpc_hpack_unittest.cpp

@@ -77,6 +77,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));