MDEV-14874 innodb_encrypt_log corrupts the log when the LSN crosses 32-bit boundary

This bug affects both writing and reading encrypted redo log in
MariaDB 10.1, starting from version 10.1.3 which added support for
innodb_encrypt_log. That is, InnoDB crash recovery and Mariabackup
will sometimes fail when innodb_encrypt_log is used.

MariaDB 10.2 or Mariabackup 10.2 or later versions are not affected.

log_block_get_start_lsn(): Remove. This function would cause trouble if
a log segment that is being read is crossing a 32-bit boundary of the LSN,
because this function does not allow the most significant 32 bits of the
LSN to change.

log_blocks_crypt(), log_encrypt_before_write(), log_decrypt_after_read():
Add the parameter "lsn" for the start LSN of the block.

log_blocks_encrypt(): Remove (unused function).

extra/mariabackup/xtrabackup.cc

@@ -2698,7 +2698,7 @@ xtrabackup_scan_log_recs(
 
 	if (srv_encrypt_log) {
 		log_encrypt_before_write(scanned_checkpoint_no,
-			log_sys->buf, write_size);
+					 log_sys->buf, start_lsn, write_size);
 	}
 
 	if (ds_write(dst_log_file, log_sys->buf, write_size)) {

mysql-test/suite/mariabackup/huge_lsn.opt

@@ -0,0 +1,6 @@
+--innodb-encrypt-log=ON
+--plugin-load-add=$FILE_KEY_MANAGEMENT_SO
+--loose-file-key-management
+--loose-file-key-management-filekey=FILE:$MTR_SUITE_DIR/filekeys-data.key
+--loose-file-key-management-filename=$MTR_SUITE_DIR/filekeys-data.enc
+--loose-file-key-management-encryption-algorithm=aes_cbc

mysql-test/suite/mariabackup/huge_lsn.result

@@ -0,0 +1,18 @@
+#
+# MDEV-13416 mariabackup fails with EFAULT "Bad Address"
+#
+call mtr.add_suppression("InnoDB: New log files created");
+FOUND /InnoDB: .*started; log sequence number 17596481010700/ in mysqld.1.err
+CREATE TABLE t(i INT) ENGINE INNODB;
+INSERT INTO t VALUES(1);
+# xtrabackup backup
+INSERT INTO t VALUES(2);
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT * FROM t;
+i
+1
+DROP TABLE t;

mysql-test/suite/mariabackup/huge_lsn.test

@@ -0,0 +1,51 @@
+--source include/not_embedded.inc
+
+--echo #
+--echo # MDEV-13416 mariabackup fails with EFAULT "Bad Address"
+--echo #
+
+let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
+let MYSQLD_DATADIR=`select @@datadir`;
+call mtr.add_suppression("InnoDB: New log files created");
+
+--source include/shutdown_mysqld.inc
+
+perl;
+my $file= "$ENV{MYSQLD_DATADIR}/ibdata1";
+open(FILE, "+<", $file) or die "Unable to open $file\n";
+binmode FILE;
+my $ps= $ENV{INNODB_PAGE_SIZE};
+my $page;
+die "Unable to read $file" unless sysread(FILE, $page, $ps) == $ps;
+substr($page,26,8) = pack("NN", 4096, ~1024);
+substr($page,0,4)=pack("N",0xdeadbeef);
+substr($page,$ps-8,4)=pack("N",0xdeadbeef);
+sysseek(FILE, 0, 0) || die "Unable to rewind $file\n";
+syswrite(FILE, $page, $ps)==$ps || die "Unable to write $file\n";
+close(FILE) || die "Unable to close $file\n";
+EOF
+
+--remove_files_wildcard $MYSQLD_DATADIR ib_logfile*
+
+--source include/start_mysqld.inc
+let SEARCH_FILE= $MYSQLTEST_VARDIR/log/mysqld.1.err;
+--let SEARCH_PATTERN= InnoDB: .*started; log sequence number 17596481010700
+--source include/search_pattern_in_file.inc
+
+CREATE TABLE t(i INT) ENGINE INNODB;
+INSERT INTO t VALUES(1);
+
+echo # xtrabackup backup;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$targetdir;
+--enable_result_log
+INSERT INTO t VALUES(2);
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+--source include/restart_and_restore.inc
+--enable_result_log
+SELECT * FROM t;
+DROP TABLE t;
+rmdir $targetdir;

storage/innobase/include/log0crypt.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2018, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -73,6 +73,8 @@ log_encrypt_before_write(
 /*=====================*/
 	ib_uint64_t	next_checkpoint_no,	/*!< in: log group to be flushed */
 	byte* 		block,			/*!< in/out: pointer to a log block */
+	lsn_t		lsn,			/*!< in: log sequence number of
+						the start of the buffer */
 	const ulint 	size);			/*!< in: size of log blocks */
 
 /********************************************************
@@ -83,6 +85,8 @@ void
 log_decrypt_after_read(
 /*===================*/
 	byte*		frame,	/*!< in/out: log segment */
+	lsn_t lsn,		/*!< in: log sequence number of the start
+				of the buffer */
 	const ulint	size);	/*!< in: log segment size */
 
 /* Error codes for crypt info */

storage/innobase/log/log0crypt.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2018, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -70,22 +70,6 @@ struct crypt_info_t {
 
 static std::deque<crypt_info_t> crypt_info;
 
-/*********************************************************************//**
-Get a log block's start lsn.
-@return a log block's start lsn */
-static inline
-lsn_t
-log_block_get_start_lsn(
-/*====================*/
-	lsn_t lsn,			/*!< in: checkpoint lsn */
-	ulint log_block_no)		/*!< in: log block number */
-{
-	lsn_t start_lsn =
-		(lsn & (lsn_t)0xffffffff00000000ULL) |
-		(((log_block_no - 1) & (lsn_t)0x3fffffff) << 9);
-	return start_lsn;
-}
-
 /*********************************************************************//**
 Get crypt info from checkpoint.
 @return a crypt info or NULL if not present. */
@@ -162,6 +146,8 @@ Crypt_result
 log_blocks_crypt(
 /*=============*/
 	const byte* block,	/*!< in: blocks before encrypt/decrypt*/
+	lsn_t lsn,		/*!< in: log sequence number of the start
+				of the buffer */
 	ulint size,		/*!< in: size of block */
 	byte* dst_block,	/*!< out: blocks after encrypt/decrypt */
 	int what,		/*!< in: encrypt or decrypt*/
@@ -171,21 +157,18 @@ log_blocks_crypt(
 	Crypt_result rc = MY_AES_OK;
 	uint dst_len;
 	byte aes_ctr_counter[MY_AES_BLOCK_SIZE];
-	byte is_encrypt= what == ENCRYPTION_FLAG_ENCRYPT;
-	lsn_t lsn = is_encrypt ? log_sys->lsn : srv_start_lsn;
 
 	const uint src_len = OS_FILE_LOG_BLOCK_SIZE - LOG_BLOCK_HDR_SIZE;
-	for (ulint i = 0; i < size ; i += OS_FILE_LOG_BLOCK_SIZE) {
+	for (ulint i = 0; i < size ; i += OS_FILE_LOG_BLOCK_SIZE,
+	     lsn += OS_FILE_LOG_BLOCK_SIZE) {
 		ulint log_block_no = log_block_get_hdr_no(log_block);
-		lsn_t log_block_start_lsn = log_block_get_start_lsn(
-			lsn, log_block_no);
 
 		const crypt_info_t* info = crypt_info == NULL ? get_crypt_info(log_block) :
 			crypt_info;
 #ifdef DEBUG_CRYPT
 		fprintf(stderr,
 			"%s %lu chkpt: %lu key: %u lsn: %lu\n",
-			is_encrypt ? "crypt" : "decrypt",
+			what == ENCRYPTION_FLAG_ENCRYPT ? "crypt" : "decrypt",
 			log_block_no,
 			log_block_get_checkpoint_no(log_block),
 			info ? info->key_version : 0,
@@ -214,7 +197,7 @@ log_blocks_crypt(
 		// (1-byte, only 5 bits are used). "+" means concatenate.
 		bzero(aes_ctr_counter, MY_AES_BLOCK_SIZE);
 		memcpy(aes_ctr_counter, info->crypt_nonce, 3);
-		mach_write_to_8(aes_ctr_counter + 3, log_block_start_lsn);
+		mach_write_to_8(aes_ctr_counter + 3, lsn);
 		mach_write_to_4(aes_ctr_counter + 11, log_block_no);
 		bzero(aes_ctr_counter + 15, 1);
 
@@ -459,19 +442,6 @@ add_crypt_info(
 	return true;
 }
 
-/*********************************************************************//**
-Encrypt log blocks. */
-UNIV_INTERN
-Crypt_result
-log_blocks_encrypt(
-/*===============*/
-	const byte* block,		/*!< in: blocks before encryption */
-	const ulint size,		/*!< in: size of blocks, must be multiple of a log block */
-	byte* dst_block)		/*!< out: blocks after encryption */
-{
-	return log_blocks_crypt(block, size, dst_block, ENCRYPTION_FLAG_ENCRYPT, NULL);
-}
-
 /*********************************************************************//**
 Set next checkpoint's key version to latest one, and generate current
 key. Key version 0 means no encryption. */
@@ -523,6 +493,8 @@ log_encrypt_before_write(
 /*=====================*/
 	ib_uint64_t next_checkpoint_no,	/*!< in: log group to be flushed */
 	byte* block,			/*!< in/out: pointer to a log block */
+	lsn_t lsn,			/*!< in: log sequence number of
+					the start of the buffer */
 	const ulint size)		/*!< in: size of log blocks */
 {
 	ut_ad(size % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -541,7 +513,8 @@ log_encrypt_before_write(
 	byte* dst_frame = (byte*)malloc(size);
 
 	//encrypt log blocks content
-	Crypt_result result = log_blocks_crypt(block, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT, NULL);
+	Crypt_result result = log_blocks_crypt(
+		block, lsn, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		ut_ad(block[0] == dst_frame[0]);
@@ -561,13 +534,16 @@ void
 log_decrypt_after_read(
 /*===================*/
 	byte* frame,	        /*!< in/out: log segment */
+	lsn_t lsn,		/*!< in: log sequence number of the start
+				of the buffer */
 	const ulint size)	/*!< in: log segment size */
 {
 	ut_ad(size % OS_FILE_LOG_BLOCK_SIZE == 0);
 	byte* dst_frame = (byte*)malloc(size);
 
 	// decrypt log blocks content
-	Crypt_result result = log_blocks_crypt(frame, size, dst_frame, ENCRYPTION_FLAG_DECRYPT, NULL);
+	Crypt_result result = log_blocks_crypt(
+		frame, lsn, size, dst_frame, ENCRYPTION_FLAG_DECRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		memcpy(frame, dst_frame, size);

storage/innobase/log/log0log.cc

@@ -1380,7 +1380,7 @@ loop:
 		ut_a(next_offset / UNIV_PAGE_SIZE <= ULINT_MAX);
 
 		log_encrypt_before_write(log_sys->next_checkpoint_no,
-					 buf, write_len);
+					 buf, start_lsn, write_len);
 
 		fil_io(OS_FILE_WRITE | OS_FILE_LOG, true, group->space_id, 0,
 		       (ulint) (next_offset / UNIV_PAGE_SIZE),
@@ -2335,7 +2335,7 @@ loop:
 		log_block_get_checksum(buf), source_offset);
 #endif
 
-	log_decrypt_after_read(buf, len);
+	log_decrypt_after_read(buf, start_lsn, len);
 
 #ifdef DEBUG_CRYPT
 	fprintf(stderr, "AFTER DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx\n",
@@ -2576,7 +2576,8 @@ loop:
 	MONITOR_INC(MONITOR_LOG_IO);
 
 	//TODO (jonaso): This must be dead code??
-	log_encrypt_before_write(log_sys->next_checkpoint_no, buf, len);
+	log_encrypt_before_write(log_sys->next_checkpoint_no,
+				 buf, start_lsn, len);
 
 	fil_io(OS_FILE_WRITE | OS_FILE_LOG, false, group->archive_space_id,
 	       (ulint) (next_offset / UNIV_PAGE_SIZE),

storage/xtradb/include/log0crypt.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2018, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -73,6 +73,8 @@ log_encrypt_before_write(
 /*=====================*/
 	ib_uint64_t	next_checkpoint_no,	/*!< in: log group to be flushed */
 	byte* 		block,			/*!< in/out: pointer to a log block */
+	lsn_t		lsn,			/*!< in: log sequence number of
+						the start of the buffer */
 	const ulint 	size);			/*!< in: size of log blocks */
 
 /********************************************************
@@ -83,6 +85,8 @@ void
 log_decrypt_after_read(
 /*===================*/
 	byte*		frame,	/*!< in/out: log segment */
+	lsn_t lsn,		/*!< in: log sequence number of the start
+				of the buffer */
 	const ulint	size);	/*!< in: log segment size */
 
 /* Error codes for crypt info */

storage/xtradb/log/log0crypt.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2018, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -69,22 +69,6 @@ struct crypt_info_t {
 
 static std::deque<crypt_info_t> crypt_info;
 
-/*********************************************************************//**
-Get a log block's start lsn.
-@return a log block's start lsn */
-static inline
-lsn_t
-log_block_get_start_lsn(
-/*====================*/
-	lsn_t lsn,			/*!< in: checkpoint lsn */
-	ulint log_block_no)		/*!< in: log block number */
-{
-	lsn_t start_lsn =
-		(lsn & (lsn_t)0xffffffff00000000ULL) |
-		(((log_block_no - 1) & (lsn_t)0x3fffffff) << 9);
-	return start_lsn;
-}
-
 /*********************************************************************//**
 Get crypt info from checkpoint.
 @return a crypt info or NULL if not present. */
@@ -161,6 +145,8 @@ Crypt_result
 log_blocks_crypt(
 /*=============*/
 	const byte* block,	/*!< in: blocks before encrypt/decrypt*/
+	lsn_t lsn,		/*!< in: log sequence number of the start
+				of the buffer */
 	ulint size,		/*!< in: size of block */
 	byte* dst_block,	/*!< out: blocks after encrypt/decrypt */
 	int what,		/*!< in: encrypt or decrypt*/
@@ -170,21 +156,18 @@ log_blocks_crypt(
 	Crypt_result rc = MY_AES_OK;
 	uint dst_len;
 	byte aes_ctr_counter[MY_AES_BLOCK_SIZE];
-	byte is_encrypt= what == ENCRYPTION_FLAG_ENCRYPT;
-	lsn_t lsn = is_encrypt ? log_sys->lsn : srv_start_lsn;
 
 	const uint src_len = OS_FILE_LOG_BLOCK_SIZE - LOG_BLOCK_HDR_SIZE;
-	for (ulint i = 0; i < size ; i += OS_FILE_LOG_BLOCK_SIZE) {
+	for (ulint i = 0; i < size ; i += OS_FILE_LOG_BLOCK_SIZE,
+	     lsn += OS_FILE_LOG_BLOCK_SIZE) {
 		ulint log_block_no = log_block_get_hdr_no(log_block);
-		lsn_t log_block_start_lsn = log_block_get_start_lsn(
-			lsn, log_block_no);
 
 		const crypt_info_t* info = crypt_info == NULL ? get_crypt_info(log_block) :
 			crypt_info;
 #ifdef DEBUG_CRYPT
 		fprintf(stderr,
 			"%s %lu chkpt: %lu key: %u lsn: %lu\n",
-			is_encrypt ? "crypt" : "decrypt",
+			what == ENCRYPTION_FLAG_ENCRYPT ? "crypt" : "decrypt",
 			log_block_no,
 			log_block_get_checkpoint_no(log_block),
 			info ? info->key_version : 0,
@@ -213,7 +196,7 @@ log_blocks_crypt(
 		// (1-byte, only 5 bits are used). "+" means concatenate.
 		bzero(aes_ctr_counter, MY_AES_BLOCK_SIZE);
 		memcpy(aes_ctr_counter, info->crypt_nonce, 3);
-		mach_write_to_8(aes_ctr_counter + 3, log_block_start_lsn);
+		mach_write_to_8(aes_ctr_counter + 3, lsn);
 		mach_write_to_4(aes_ctr_counter + 11, log_block_no);
 		bzero(aes_ctr_counter + 15, 1);
 
@@ -458,19 +441,6 @@ add_crypt_info(
 	return true;
 }
 
-/*********************************************************************//**
-Encrypt log blocks. */
-UNIV_INTERN
-Crypt_result
-log_blocks_encrypt(
-/*===============*/
-	const byte* block,		/*!< in: blocks before encryption */
-	const ulint size,		/*!< in: size of blocks, must be multiple of a log block */
-	byte* dst_block)		/*!< out: blocks after encryption */
-{
-	return log_blocks_crypt(block, size, dst_block, ENCRYPTION_FLAG_ENCRYPT, NULL);
-}
-
 /*********************************************************************//**
 Set next checkpoint's key version to latest one, and generate current
 key. Key version 0 means no encryption. */
@@ -522,6 +492,8 @@ log_encrypt_before_write(
 /*=====================*/
 	ib_uint64_t next_checkpoint_no,	/*!< in: log group to be flushed */
 	byte* block,			/*!< in/out: pointer to a log block */
+	lsn_t lsn,			/*!< in: log sequence number of
+					the start of the buffer */
 	const ulint size)		/*!< in: size of log blocks */
 {
 	ut_ad(size % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -540,7 +512,8 @@ log_encrypt_before_write(
 	byte* dst_frame = (byte*)malloc(size);
 
 	//encrypt log blocks content
-	Crypt_result result = log_blocks_crypt(block, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT, NULL);
+	Crypt_result result = log_blocks_crypt(
+		block, lsn, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		ut_ad(block[0] == dst_frame[0]);
@@ -560,13 +533,16 @@ void
 log_decrypt_after_read(
 /*===================*/
 	byte* frame,	        /*!< in/out: log segment */
+	lsn_t lsn,		/*!< in: log sequence number of the start
+				of the buffer */
 	const ulint size)	/*!< in: log segment size */
 {
 	ut_ad(size % OS_FILE_LOG_BLOCK_SIZE == 0);
 	byte* dst_frame = (byte*)malloc(size);
 
 	// decrypt log blocks content
-	Crypt_result result = log_blocks_crypt(frame, size, dst_frame, ENCRYPTION_FLAG_DECRYPT, NULL);
+	Crypt_result result = log_blocks_crypt(
+		frame, lsn, size, dst_frame, ENCRYPTION_FLAG_DECRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		memcpy(frame, dst_frame, size);

storage/xtradb/log/log0log.cc

@@ -1490,7 +1490,7 @@ loop:
 		ut_a(next_offset / UNIV_PAGE_SIZE <= ULINT_MAX);
 
 		log_encrypt_before_write(log_sys->next_checkpoint_no,
-					 buf, write_len);
+					 buf, start_lsn, write_len);
 
 #ifdef DEBUG_CRYPT
 		fprintf(stderr, "WRITE: block: %lu checkpoint: %lu %.8lx %.8lx\n",
@@ -2582,7 +2582,7 @@ loop:
 		log_block_get_checksum(buf), source_offset);
 #endif
 
-	log_decrypt_after_read(buf, len);
+	log_decrypt_after_read(buf, start_lsn, len);
 
 #ifdef DEBUG_CRYPT
 	fprintf(stderr, "AFTER DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx\n",
@@ -2890,7 +2890,8 @@ loop:
 	MONITOR_INC(MONITOR_LOG_IO);
 
 	//TODO (jonaso): This must be dead code??
-	log_encrypt_before_write(log_sys->next_checkpoint_no, buf, len);
+	log_encrypt_before_write(log_sys->next_checkpoint_no,
+				 buf, start_lsn, len);
 
 	fil_io(OS_FILE_WRITE | OS_FILE_LOG, false, group->archive_space_id,
 	       0,