MDEV-39762: Slave Overflow on Malformed Query_compressed_log_event

A replica can crash when a compressed event from its master carries a corrupted uncompressed length. A compressed Query event whose length is near 4GB overwrites the IO thread's stack with the event's content. A compressed rows event of the same shape makes the replica ask for an allocation of 4GB. A debug build that gets that memory then fails an assertion in binlog_buf_uncompress(). On a build where a ulong is 32 bits wide, both events overwrite the IO thread's stack. The IO thread uncompresses a compressed event before writing the event to the relay log, and offers the uncompress function 4096 bytes of its own stack to hold the result. The function computes how large the event will be once uncompressed, and that size decides where the uncompressed event goes. An event that fits in the stack buffer is uncompressed there. A larger event is uncompressed into an allocation the function makes at that size. With either buffer, the function tells zlib that the room available is the uncompressed length the event declared. query_event_uncompress() never bounded the length read out of the event. The function added the header length to that value, rounded the sum up for alignment, and cast the result to uint32. The cast dropped the high bits of a sum above 4GB, so a length near 4GB produced a size of a few dozen bytes. That size fit the stack buffer, so content of any size went onto the IO thread's stack. row_log_event_uncompress() computes that sum in a ulong, which wraps where a ulong is 32 bits wide. The Query_compressed_log_event constructor and Rows_log_event::uncompress_buf() size their allocations from the same unbounded length. Reject an uncompressed length above MAX_MAX_ALLOWED_PACKET at every point the length is read, before any size is computed from it. No master writes a larger length, because max_allowed_packet is capped at 1GB. query_event_uncompress() also keeps that size in a size_t, so the allocation is made from the whole computed value. Both functions return the error code for the IO thread to report, so a length over the limit stops the IO thread with ER_TOO_BIG_FOR_UNCOMPRESS rather than the generic uncompress error. A compressed event with a corrupted uncompressed length now stops the replica's IO thread with an error instead of crashing the replica. Alexandra Raveala wrote the original patch and its first regression test, contributed through PR 5413. Brandon Nesterenko saw the work through for the 10.6 release, replacing the test with one that makes the master write the corrupt length and reworking how the IO thread reports the error. Co-authored-by: Brandon Nesterenko <brandon.nesterenko@mariadb.com> Reviewed-by: Kristian Nielsen <knielsen@knielsen-hq.org>
2026-10-23 18:49:29 +03:00 · 2026-08-07 11:29:16 -06:00
parent 9e4948c2cf
commit f17a3ea600
5 changed files with 318 additions and 43 deletions
--- a/mysql-test/suite/rpl/r/rpl_compressed_log_event_overflow.result
+++ b/mysql-test/suite/rpl/r/rpl_compressed_log_event_overflow.result
@@ -0,0 +1,78 @@
+include/master-slave.inc
+[connection master]
+connection slave;
+include/stop_slave.inc
+CHANGE MASTER TO master_use_gtid= slave_pos;
+include/start_slave.inc
+call mtr.add_suppression("Uncompressed data size too large");
+call mtr.add_suppression("Relay log write failure: could not queue event from master");
+#
+# Initialize test data
+#
+connection master;
+create table t1 (a int);
+create table t2 (a longtext);
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+#
+# Test Case 1: a compressed Query event
+#
+connection master;
+set @saved_dbug= @@session.debug_dbug;
+set @@session.debug_dbug= "+d,binlog_compress_corrupt_len";
+set @saved_binlog_format= @@session.binlog_format;
+set @@session.binlog_format= STATEMENT;
+set @@session.binlog_format= @saved_binlog_format;
+set @@session.debug_dbug= @saved_dbug;
+connection slave;
+# Waiting for the IO thread to reject the compressed Query event
+include/wait_for_slave_io_error.inc [errno=1256, 1595]
+# Ensure the replica rejected the event on its recorded length
+include/assert_grep.inc [The IO thread reported an oversized uncompressed length]
+# Move the replica past the corrupt event
+include/stop_slave_sql.inc
+include/start_slave.inc
+# Ensure replication resumes
+connection master;
+insert into t1 values (1);
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/diff_tables.inc [master:t1,slave:t1]
+#
+# Test Case 2: a compressed Write_rows event
+#
+connection master;
+set @saved_dbug= @@session.debug_dbug;
+set @@session.debug_dbug= "+d,binlog_compress_corrupt_len";
+set @saved_binlog_format= @@session.binlog_format;
+set @@session.binlog_format= ROW;
+insert into t2 values (repeat('a', 65536));
+set @@session.binlog_format= @saved_binlog_format;
+set @@session.debug_dbug= @saved_dbug;
+connection slave;
+# Waiting for the IO thread to reject the compressed Write_rows event
+include/wait_for_slave_io_error.inc [errno=1256, 1595]
+# Ensure the replica rejected the event on its recorded length
+include/assert_grep.inc [The IO thread reported an oversized uncompressed length]
+# Move the replica past the corrupt event
+include/stop_slave_sql.inc
+include/start_slave.inc
+# Ensure replication resumes
+connection master;
+insert into t1 values (2);
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/diff_tables.inc [master:t1,slave:t1]
+#
+# Cleanup
+#
+connection master;
+drop table t1, t2;
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/rpl_end.inc
+# End of rpl_compressed_log_event_overflow.test
--- a/mysql-test/suite/rpl/t/rpl_compressed_log_event_overflow-master.opt
+++ b/mysql-test/suite/rpl/t/rpl_compressed_log_event_overflow-master.opt
@@ -0,0 +1 @@
+--log-bin-compress=1
--- a/mysql-test/suite/rpl/t/rpl_compressed_log_event_overflow.test
+++ b/mysql-test/suite/rpl/t/rpl_compressed_log_event_overflow.test
@@ -0,0 +1,160 @@
+#
+#   This test verifies that a replica rejects a compressed event whose
+# recorded uncompressed length is larger than the biggest packet the
+# server accepts. A compressed Query event and a compressed Write_rows
+# event are both covered, because the IO thread uncompresses each with
+# its own function. MDEV-39762 reported that the compressed Query event
+# crashed the replica.
+#
+#   A debug injection makes the master write each of those events with
+# a corrupted length field. The event stays in the master's binary log
+# and is resent as it is, so each case moves gtid_slave_pos past it
+# before checking that replication resumes. The sibling .opt file turns
+# on log_bin_compress, without which the master writes no compressed
+# events at all.
+#
+# References:
+#
+#   * MDEV-39762: Slave Overflow on Malformed Query_compressed_log_event
+#
+
+--source include/have_debug.inc
+# Each test case sets the binlog format that its event type needs
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+# Test requires slave_pos, as each case recovers by moving gtid_slave_pos
+# past the corrupt event.
+--connection slave
+--source include/stop_slave.inc
+CHANGE MASTER TO master_use_gtid= slave_pos;
+--source include/start_slave.inc
+
+call mtr.add_suppression("Uncompressed data size too large");
+call mtr.add_suppression("Relay log write failure: could not queue event from master");
+
+--echo #
+--echo # Initialize test data
+--echo #
+--connection master
+# t1 tracks what replicates. t2 takes the corrupt events, and the replica
+# never receives its rows.
+create table t1 (a int);
+create table t2 (a longtext);
+
+# The content must uncompress to far more than the 4096 byte stack buffer
+# that queue_event() passes to the uncompress functions.
+--let $long_value= `SELECT REPEAT('a', 65536)`
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+
+--echo #
+--echo # Test Case 1: a compressed Query event
+--echo #
+--connection master
+set @saved_dbug= @@session.debug_dbug;
+set @@session.debug_dbug= "+d,binlog_compress_corrupt_len";
+set @saved_binlog_format= @@session.binlog_format;
+set @@session.binlog_format= STATEMENT;
+# In statement format the compressed content is the query text itself, so
+# the value has to appear in the statement rather than come from repeat().
+--disable_query_log
+--eval insert into t2 values ('$long_value')
+--enable_query_log
+set @@session.binlog_format= @saved_binlog_format;
+set @@session.debug_dbug= @saved_dbug;
+--let $corrupt_gtid_pos= `SELECT @@gtid_binlog_pos`
+
+--connection slave
+--echo # Waiting for the IO thread to reject the compressed Query event
+# The IO thread reports ER_TOO_BIG_FOR_UNCOMPRESS (1256) and then
+# overwrites it with ER_SLAVE_RELAY_LOG_WRITE_FAILURE (1595), so the
+# wait can sample either errno.
+--let $slave_io_errno= 1256, 1595
+--source include/wait_for_slave_io_error.inc
+
+--echo # Ensure the replica rejected the event on its recorded length
+--let $assert_text= The IO thread reported an oversized uncompressed length
+--let $assert_file= $MYSQLTEST_VARDIR/log/mysqld.2.err
+--let $assert_select= Uncompressed data size too large
+--let $assert_match= Uncompressed data size too large; the maximum size is 1073741824
+--source include/assert_grep.inc
+
+--echo # Move the replica past the corrupt event
+--source include/stop_slave_sql.inc
+--disable_query_log
+--eval SET GLOBAL gtid_slave_pos= '$corrupt_gtid_pos'
+--enable_query_log
+--source include/start_slave.inc
+
+--echo # Ensure replication resumes
+--connection master
+insert into t1 values (1);
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+--let $diff_tables= master:t1,slave:t1
+--source include/diff_tables.inc
+
+--echo #
+--echo # Test Case 2: a compressed Write_rows event
+--echo #
+--connection master
+set @saved_dbug= @@session.debug_dbug;
+set @@session.debug_dbug= "+d,binlog_compress_corrupt_len";
+set @saved_binlog_format= @@session.binlog_format;
+set @@session.binlog_format= ROW;
+# In row format the compressed content is the record, so the statement
+# that writes it can stay short.
+insert into t2 values (repeat('a', 65536));
+set @@session.binlog_format= @saved_binlog_format;
+set @@session.debug_dbug= @saved_dbug;
+--let $corrupt_gtid_pos= `SELECT @@gtid_binlog_pos`
+
+--connection slave
+--echo # Waiting for the IO thread to reject the compressed Write_rows event
+# The IO thread reports ER_TOO_BIG_FOR_UNCOMPRESS (1256) and then
+# overwrites it with ER_SLAVE_RELAY_LOG_WRITE_FAILURE (1595), so the
+# wait can sample either errno.
+--let $slave_io_errno= 1256, 1595
+--source include/wait_for_slave_io_error.inc
+
+--echo # Ensure the replica rejected the event on its recorded length
+--let $assert_text= The IO thread reported an oversized uncompressed length
+--let $assert_file= $MYSQLTEST_VARDIR/log/mysqld.2.err
+--let $assert_select= Uncompressed data size too large
+--let $assert_match= Uncompressed data size too large; the maximum size is 1073741824
+--source include/assert_grep.inc
+
+--echo # Move the replica past the corrupt event
+--source include/stop_slave_sql.inc
+--disable_query_log
+--eval SET GLOBAL gtid_slave_pos= '$corrupt_gtid_pos'
+--enable_query_log
+--source include/start_slave.inc
+
+--echo # Ensure replication resumes
+--connection master
+insert into t1 values (2);
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+--let $diff_tables= master:t1,slave:t1
+--source include/diff_tables.inc
+
+--echo #
+--echo # Cleanup
+--echo #
+--connection master
+drop table t1, t2;
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+
+--source include/rpl_end.inc
+--echo # End of rpl_compressed_log_event_overflow.test
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -338,6 +338,25 @@ int binlog_buf_compress(const uchar *src, uchar *dst, uint32 len, uint32 *comlen
    dst[1]= uchar(len);
    lenlen= 1;
  }
+
+  /*
+    Write the length of an event no replica can uncompress, so that a
+    replica's handling of one can be tested. Reaching a value this large
+    takes four length bytes, and the encoding above spends four only on
+    content of 16MB or more, so the injection sets the encoding as well
+    as the value. The compressed content still goes where those four
+    bytes place it, leaving the event well formed apart from the length
+    it claims.
+  */
+  DBUG_EXECUTE_IF("binlog_compress_corrupt_len",
+                  {
+                    dst[1]= 0xFF;
+                    dst[2]= 0xFF;
+                    dst[3]= 0xFF;
+                    dst[4]= 0xFC;
+                    lenlen= 4;
+                  });
+
  dst[0]= 0x80 | (lenlen & 0x07);

  uLongf tmplen= (uLongf)*comlen - BINLOG_COMPRESSED_HEADER_LEN - lenlen - 1;
@@ -360,7 +379,7 @@ int binlog_buf_compress(const uchar *src, uchar *dst, uint32 len, uint32 *comlen
      2) If *is_malloc is retuened as false, then 'dst' reuses the passed-in
         'buf'.

-   return zero if successful, non-zero otherwise.
+   return zero if successful, otherwise the error code the caller reports.
 */

 int
@@ -375,8 +394,8 @@ query_event_uncompress(const Format_description_log_event *description_event,
  uchar *new_dst;

  // bad event
-  if (src_len < len)
-    return 1;
+  if (unlikely(src_len < len))
+    return ER_BINLOG_UNCOMPRESS_ERROR;

  DBUG_ASSERT((uchar)src[EVENT_TYPE_OFFSET] == QUERY_COMPRESSED_EVENT);

@@ -388,30 +407,33 @@ query_event_uncompress(const Format_description_log_event *description_event,

  tmp+= common_header_len;
  // bad event
-  if (end <= tmp)
-    return 1;
+  if (unlikely(end <= tmp))
+    return ER_BINLOG_UNCOMPRESS_ERROR;

  uint db_len= (uint)tmp[Q_DB_LEN_OFFSET];
  uint16 status_vars_len= uint2korr(tmp + Q_STATUS_VARS_LEN_OFFSET);

  tmp+= post_header_len + status_vars_len + db_len + 1;
  // bad event
-  if (end <= tmp)
-    return 1;
+  if (unlikely(end <= tmp))
+    return ER_BINLOG_UNCOMPRESS_ERROR;

  int32 comp_len= (int32)(len - (tmp - src) -
                          (contain_checksum ? BINLOG_CHECKSUM_LEN : 0));
  uint32 un_len=  binlog_get_uncompress_len(tmp);

  // bad event 
-  if (comp_len < 0 || un_len == 0)
-    return 1;
+  if (unlikely(comp_len < 0 || un_len == 0))
+    return ER_BINLOG_UNCOMPRESS_ERROR;
+
+  if (unlikely(un_len > MAX_MAX_ALLOWED_PACKET))
+    return ER_TOO_BIG_FOR_UNCOMPRESS;

  *newlen= (ulong)(tmp - src) + un_len;
  if (contain_checksum)
    *newlen+= BINLOG_CHECKSUM_LEN;
-  
-  uint32 alloc_size= (uint32)ALIGN_SIZE(*newlen);
+
+  size_t alloc_size= ALIGN_SIZE(*newlen);
  
  if (alloc_size <= buf_size) 
    new_dst= buf;
@@ -419,7 +441,7 @@ query_event_uncompress(const Format_description_log_event *description_event,
  {
    new_dst= (uchar *) my_malloc(PSI_INSTRUMENT_ME, alloc_size, MYF(MY_WME));
    if (!new_dst)
-      return 1;
+      return ER_BINLOG_UNCOMPRESS_ERROR;
    *is_malloc= true;
  }

@@ -432,7 +454,7 @@ query_event_uncompress(const Format_description_log_event *description_event,
      *is_malloc= false;
      my_free(new_dst);
    }
-    return 1;
+    return ER_BINLOG_UNCOMPRESS_ERROR;
  }

  new_dst[EVENT_TYPE_OFFSET]= QUERY_EVENT;
@@ -459,8 +481,8 @@ row_log_event_uncompress(const Format_description_log_event *description_event,
  uchar *new_dst= NULL;
  const uchar *end= tmp + len;

-  if (src_len < len)
-    return 1;                                   // bad event
+  if (unlikely(src_len < len))
+    return ER_BINLOG_UNCOMPRESS_ERROR;

  DBUG_ASSERT(LOG_EVENT_IS_ROW_COMPRESSED(type));

@@ -475,8 +497,8 @@ row_log_event_uncompress(const Format_description_log_event *description_event,
      which includes length bytes
    */

-    if (end - tmp <= 2)
-      return 1;                                 // bad event
+    if (unlikely(end - tmp <= 2))
+      return ER_BINLOG_UNCOMPRESS_ERROR;

    uint16 var_header_len= uint2korr(tmp);
    DBUG_ASSERT(var_header_len >= 2);
@@ -495,8 +517,8 @@ row_log_event_uncompress(const Format_description_log_event *description_event,
      (type - WRITE_ROWS_COMPRESSED_EVENT_V1 + WRITE_ROWS_EVENT_V1);
  }

-  if (end <= tmp)
-    return 1;                                   //bad event
+  if (unlikely(end <= tmp))
+    return ER_BINLOG_UNCOMPRESS_ERROR;

  ulong m_width= net_field_length((uchar **)&tmp);
  tmp+= (m_width + 7) / 8;
@@ -506,17 +528,20 @@ row_log_event_uncompress(const Format_description_log_event *description_event,
    tmp+= (m_width + 7) / 8;
  }

-  if (end <= tmp)
-    return 1;                                   //bad event
+  if (unlikely(end <= tmp))
+    return ER_BINLOG_UNCOMPRESS_ERROR;

  uint32 un_len= binlog_get_uncompress_len(tmp);
-  if (un_len == 0)
-    return 1;                                   //bad event
+  if (unlikely(un_len == 0))
+    return ER_BINLOG_UNCOMPRESS_ERROR;
+
+  if (unlikely(un_len > MAX_MAX_ALLOWED_PACKET))
+    return ER_TOO_BIG_FOR_UNCOMPRESS;

  int32 comp_len= (int32)(len - (tmp - src) -
                          (contain_checksum ? BINLOG_CHECKSUM_LEN : 0));
-  if (comp_len <=0)
-    return 1;                                   //bad event
+  if (unlikely(comp_len <= 0))
+    return ER_BINLOG_UNCOMPRESS_ERROR;

  *newlen= ulong(tmp - src) + un_len;
  if (contain_checksum)
@@ -533,7 +558,7 @@ row_log_event_uncompress(const Format_description_log_event *description_event,
  {
    new_dst= (uchar*) my_malloc(PSI_INSTRUMENT_ME, alloc_size, MYF(MY_WME));
    if (!new_dst)
-      return 1;
+      return ER_BINLOG_UNCOMPRESS_ERROR;
    *is_malloc= true;
  }

@@ -545,7 +570,7 @@ row_log_event_uncompress(const Format_description_log_event *description_event,
  {
    if (*is_malloc)
      my_free(new_dst);
-    return 1;
+    return ER_BINLOG_UNCOMPRESS_ERROR;
  }

  new_dst[EVENT_TYPE_OFFSET]= type;
@@ -1838,7 +1863,7 @@ Query_compressed_log_event::Query_compressed_log_event(const uchar *buf,
  if (query)
  {
    uint32 un_len= binlog_get_uncompress_len((uchar*) query);
-    if (!un_len)
+    if (unlikely(!un_len || un_len > MAX_MAX_ALLOWED_PACKET))
    {
      query= 0;
      return;
@@ -3539,8 +3564,12 @@ Rows_log_event::Rows_log_event(const uchar *buf, uint event_len,
 void Rows_log_event::uncompress_buf()
 {
  uint32 un_len= binlog_get_uncompress_len(m_rows_buf);
-  if (!un_len)
+  if (unlikely(!un_len || un_len > MAX_MAX_ALLOWED_PACKET))
+  {
+    /* my_bitmap_free() nulls m_cols.bitmap, which is_valid() rejects. */
+    my_bitmap_free(&m_cols);
    return;
+  }

  uchar *new_buf= (uchar*) my_malloc(PSI_INSTRUMENT_ME, ALIGN_SIZE(un_len),
                                     MYF(MY_WME));
@@ -3563,7 +3592,7 @@ void Rows_log_event::uncompress_buf()
      my_free(new_buf);
    }
  }
-  m_cols.bitmap= 0; // catch it in is_valid
+  my_bitmap_free(&m_cols); // catch it in is_valid
 }

 Rows_log_event::~Rows_log_event()
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -7405,13 +7405,14 @@ dbug_gtid_accept:
  */
  case QUERY_COMPRESSED_EVENT:
    inc_pos= event_len;
-    if (query_event_uncompress(rli->relay_log.description_event_for_queue,
-                               checksum_alg == BINLOG_CHECKSUM_ALG_CRC32,
-                               buf, event_len, new_buf_arr, sizeof(new_buf_arr),
-                               &is_malloc, &new_buf, &event_len))
+    if ((error=
+         query_event_uncompress(rli->relay_log.description_event_for_queue,
+                                checksum_alg == BINLOG_CHECKSUM_ALG_CRC32,
+                                buf, event_len, new_buf_arr,
+                                sizeof(new_buf_arr),
+                                &is_malloc, &new_buf, &event_len)))
    {
      char  llbuf[22];
-      error = ER_BINLOG_UNCOMPRESS_ERROR;
      error_msg.append(STRING_WITH_LEN("binlog uncompress error, master log_pos: "));
      llstr(mi->master_log_pos, llbuf);
      error_msg.append(llbuf, strlen(llbuf));
@@ -7429,14 +7430,14 @@ dbug_gtid_accept:
  case DELETE_ROWS_COMPRESSED_EVENT_V1:
    inc_pos = event_len;
    {
-      if (row_log_event_uncompress(rli->relay_log.description_event_for_queue,
-                                   checksum_alg == BINLOG_CHECKSUM_ALG_CRC32,
-                                   buf, event_len, new_buf_arr,
-                                   sizeof(new_buf_arr),
-                                   &is_malloc, &new_buf, &event_len))
+      if ((error=
+           row_log_event_uncompress(rli->relay_log.description_event_for_queue,
+                                    checksum_alg == BINLOG_CHECKSUM_ALG_CRC32,
+                                    buf, event_len, new_buf_arr,
+                                    sizeof(new_buf_arr),
+                                    &is_malloc, &new_buf, &event_len)))
      {
        char  llbuf[22];
-        error = ER_BINLOG_UNCOMPRESS_ERROR;
        error_msg.append(STRING_WITH_LEN("binlog uncompress error, master log_pos: "));
        llstr(mi->master_log_pos, llbuf);
        error_msg.append(llbuf, strlen(llbuf));
@@ -7818,8 +7819,14 @@ err:
    handle_slave_io() prints it on return.
  */
  if (unlikely(error) && error != ER_SLAVE_RELAY_LOG_WRITE_FAILURE)
-    mi->report(ERROR_LEVEL, error, NULL, ER_DEFAULT(error),
-               error_msg.ptr());
+  {
+    if (error == ER_TOO_BIG_FOR_UNCOMPRESS)
+      mi->report(ERROR_LEVEL, error, error_msg.c_ptr(), ER_DEFAULT(error),
+                 MAX_MAX_ALLOWED_PACKET);
+    else
+      mi->report(ERROR_LEVEL, error, NULL, ER_DEFAULT(error),
+                 error_msg.ptr());
+  }

  if (unlikely(is_malloc))
    my_free((void *)new_buf);