Merge bitcoin/bitcoin#36123: http: throttle per-connection reads while a request is in flight

3d1004cb9b http: throttle per-connection reads while a request is in flight (Matthew Zipkin)

Pull request description:

  This patches a memory exhaustion scenario found while auditing the new http server with kimi-k3. A shallow version of this scenario was addressed in #35735 (See  https://github.com/bitcoin/bitcoin/pull/35735#discussion_r3720177656 and https://github.com/bitcoin/bitcoin/pull/35735#issuecomment-5217000202) but a OOM vector still remained.

  On master when the sever is busy handling a request from a client, it will still read data from that client and "queue up" the next request. In #35735 we handled the scenario where that additional incoming data was an invalid HTTP request by not attempting to parse the data. However, we didn't add a size limit.

  A misbehaving client could block its request queue with something like `waitforblock` and then flood the server with nonsense data without any limit.

  The solution in this patch is to not even read from the socket at all if we are busy with a request. Similar to the intent of #35735, the kernel will buffer incoming data until backpressure kicks in and the TCP window drops to 0.

  If unaddressed, the attack vector is still limited to authenticated clients: unauthenticated REST requests don't block for very long, so the server *should* be able to drain the receive buffer.

ACKs for top commit:
  jeanpablojp:
    tACK 3d1004cb9b
  frankomosh:
    ACK 3d1004cb9b
  hodlinator:
    ACK 3d1004cb9b
  winterrdog:
    tACK 3d1004cb9b
  sedited:
    ACK 3d1004cb9b

Tree-SHA512: 56f7678a9ab6789aa542c1f252df0b6ccf9137cb426ff915a0a3fe8285200fdb62b7a47c476ed8617c3592e7a7eac18158cd8c0dac309cdcf4e5fd887e016209

test/functional/interface_http.py

@@ -153,6 +153,7 @@ class HTTPBasicsTest (BitcoinTestFramework):
         self.check_invalid_http_version()
         self.check_whitespace_in_headers()
         self.check_connection_limit()
+        self.check_pipelined_data_is_throttled()
 
 
     def check_default_connection(self):
@@ -699,5 +700,84 @@ class HTTPBasicsTest (BitcoinTestFramework):
                 client.close_sock()
 
 
+    def check_pipelined_data_is_throttled(self):
+        self.log.info("Check that pipelined data is throttled while a request is in flight")
+        self.restart_node(0, extra_args=["-rpcservertimeout=0"])
+
+        conn = BitcoinHTTPConnection(self.node)
+
+        # A blocking RPC request: the server reads it fully and it enters the
+        # worker pool as "in flight" (m_req_busy) until a new block arrives.
+        tip_height = self.node.getblockcount()
+        conn.post_raw('/', f'{{"method": "waitforblockheight", "params": [{tip_height + 1}]}}')
+
+        # Flood the same connection with big pipelined requests:
+        # Large garbage submitblock (just under MAX_BODY_SIZE each, including HTTP/jsonrpc overhead)
+        garbage_block = "0" * (MAX_BODY_SIZE - 100)
+        body = f'{{"method": "submitblock", "params": ["{garbage_block}"]}}'
+        flood = (
+            f'POST / HTTP/1.1\r\nAuthorization: Basic {str_to_b64str(conn.authpair)}\r\n'
+            f'Content-Length: {len(body)}\r\n\r\n' +
+            body
+        ).encode("ascii")
+
+        # Non-blocking send: When the server stops reading from the buffer
+        # due to TCP backpressure, Python will raise an error. If the socket
+        # was set to blocking, we would have to wait for an ambiguous timeout.
+        conn.conn.sock.setblocking(False)
+
+        # Kernel socket buffer sizes vary widely across platforms,
+        # so we can't rely on counting sent() bytes to determine if the
+        # server is actually draining its end of the socket.
+        # When the server is busy, a continuous flood from the client SHOULD,
+        # at some point, stall indefinitely. An unpatched server will continue
+        # to accept data from the socket, at some rate, indefinitely.
+
+        # If send() is blocked for this many seconds, we assume the server
+        # is behaving correctly.
+        STALL_TIMEOUT = 5
+        # If send() continues to progress for this many seconds, we assume
+        # the server is vulnerable to memory exhaustion.
+        PROGRESS_TIMEOUT = 10
+
+        sent = 0
+        stuck_since = None
+        start = time.monotonic()
+        while True:
+            try:
+                sent += conn.conn.sock.send(flood[sent % len(flood):])
+                # Progress: the server is still reading
+                stuck_since = None
+                self.log.debug(f"sent: {sent}")
+                assert sent <= len(flood) * 10, (
+                    f"Server accepted {sent} bytes of pipelined data while a "
+                    "request was still in flight: the receive buffer is not throttled")
+            except BlockingIOError:
+                # The kernel send buffer is full (EAGAIN).
+                # That's good, but we still need to determine if we are
+                # feeling backpressure from the server or the client-side buffer.
+                if stuck_since is None:
+                    stuck_since = time.monotonic()
+                elif time.monotonic() - stuck_since > STALL_TIMEOUT:
+                    # No progress: the server has stopped reading.
+                    break
+            if stuck_since is None and time.monotonic() - start > PROGRESS_TIMEOUT:
+                # Continuous progress: the server is still draining the
+                # receive buffer while a request is in flight.
+                raise AssertionError(
+                    f"Server kept reading pipelined data ({sent} bytes) while a "
+                    f"request was still in flight for {PROGRESS_TIMEOUT}s.")
+            time.sleep(0.05)
+
+        self.log.info(f"Pipelined flood stalled after {sent} bytes; no progress for {STALL_TIMEOUT}s.")
+
+        # Unblock the client request queue.
+        conn.conn.sock.settimeout(10)
+        generated_block = self.generate(self.node, 1, sync_fun=self.no_op)[0]
+        # First reply is for the blocking request.
+        response = conn.recv_raw().decode()
+        assert generated_block in response
+
+
 if __name__ == '__main__':
     HTTPBasicsTest(__file__).main()