diff --git a/mysql-test/suite/rpl/r/rpl_compressed_log_event_overflow.result b/mysql-test/suite/rpl/r/rpl_compressed_log_event_overflow.result new file mode 100644 index 0000000000000..6107508c9fc9a --- /dev/null +++ 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 diff --git 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 new file mode 100644 index 0000000000000..c949694b21bde --- /dev/null +++ b/mysql-test/suite/rpl/t/rpl_compressed_log_event_overflow-master.opt @@ -0,0 +1 @@ +--log-bin-compress=1 diff --git a/mysql-test/suite/rpl/t/rpl_compressed_log_event_overflow.test b/mysql-test/suite/rpl/t/rpl_compressed_log_event_overflow.test new file mode 100644 index 0000000000000..29d51a823a79e --- /dev/null +++ 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 diff --git a/sql/log_event.cc b/sql/log_event.cc index f72423718db59..3f2c9f4e9581d 100644 --- 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; @@ -1823,7 +1848,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; @@ -3501,8 +3526,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)); @@ -3525,7 +3554,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() diff --git a/sql/slave.cc b/sql/slave.cc index 9ca95dee93851..ec2991dc45480 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -7106,13 +7106,14 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len) */ 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)); @@ -7130,14 +7131,14 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len) 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)); @@ -7517,8 +7518,14 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len) 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);