MDEV-9422: Checksum errors on restart when killing busy instance that uses encrypted XtraDB tables

Analysis:

-- InnoDB has n (>0) redo-log files.
-- In the first page of redo-log there is 2 checkpoint records on fixed location (checkpoint is not encrypted)
-- On every checkpoint record there is up to 5 crypt_keys containing the keys used for encryption/decryption
-- On crash recovery we read all checkpoints on every file
-- Recovery starts by reading from the latest checkpoint forward
-- Problem is that latest checkpoint might not always contain the key we need to decrypt all the
   redo-log blocks (see MDEV-9422 for one example)
-- Furthermore, there is no way to identify is the log block corrupted or encrypted

For example checkpoint can contain following keys :

write chk: 4 [ chk key ]: [ 5 1 ] [ 4 1 ] [ 3 1 ] [ 2 1 ] [ 1 1 ]

so over time we could have a checkpoint

write chk: 13 [ chk key ]: [ 14 1 ] [ 13 1 ] [ 12 1 ] [ 11 1 ] [ 10 1 ]

killall -9 mysqld causes crash recovery and on crash recovery we read as
many checkpoints as there is log files, e.g.

read [ chk key ]: [ 13 1 ] [ 12 1 ] [ 11 1 ] [ 10 1 ] [ 9 1 ]
read [ chk key ]: [ 14 1 ] [ 13 1 ] [ 12 1 ] [ 11 1 ] [ 10 1 ] [ 9 1 ]

This is problematic, as we could still scan log blocks e.g. from checkpoint 4 and we do
not know anymore the correct key.

CRYPT INFO: for checkpoint 14 search 4
CRYPT INFO: for checkpoint 13 search 4
CRYPT INFO: for checkpoint 12 search 4
CRYPT INFO: for checkpoint 11 search 4
CRYPT INFO: for checkpoint 10 search 4
CRYPT INFO: for checkpoint 9 search 4 (NOTE: NOT FOUND)

For every checkpoint, code generated a new encrypted key based on key
from encryption plugin and random numbers. Only random numbers are
stored on checkpoint.

Fix: Generate only one key for every log file. If checkpoint contains only
one key, use that key to encrypt/decrypt all log blocks. If checkpoint
contains more than one key (this is case for databases created
using MariaDB server version 10.1.0 - 10.1.12 if log encryption was
used). If looked checkpoint_no is found from keys on checkpoint we use
that key to decrypt the log block. For encryption we use always the
first key. If the looked checkpoint_no is not found from keys on checkpoint
we use the first key.

Modified code also so that if log is not encrypted, we do not generate
any empty keys. If we have a log block and no keys is found from
checkpoint we assume that log block is unencrypted. Log corruption or
missing keys is found by comparing log block checksums. If we have
a keys but current log block checksum is correct we again assume
log block to be unencrypted. This is because current implementation
stores checksum only before encryption and new checksum after
encryption but before disk write is not stored anywhere.

14 files changed, 237 insertions, 67 deletions

diff --git a/mysql-test/suite/encryption/r/innodb-log-encrypt-crash.result b/mysql-test/suite/encryption/r/innodb-log-encrypt-crash.result
new file mode 100644
index 00000000000..5310fb6ace2
--- /dev/null
+++ b/mysql-test/suite/encryption/r/innodb-log-encrypt-crash.result
@@ -0,0 +1,19 @@
+call mtr.add_suppression("InnoDB: New log files created, LSN=.*");
+call mtr.add_suppression("InnoDB: Creating foreign key constraint system tables.");
+call mtr.add_suppression("InnoDB: Error: Table .*");
+CREATE TABLE t1 (
+pk bigint auto_increment,
+col_int int,
+col_int_key int,
+col_char char(12),
+col_char_key char(12),
+primary key (pk),
+key (`col_int_key` ),
+key (`col_char_key` )
+) ENGINE=InnoDB;
+CREATE TABLE t2 LIKE t1;
+INSERT INTO t1 VALUES (NULL,1,1,'foo','foo'),(NULL,2,2,'bar','bar'),(NULL,3,3,'baz','baz'),(NULL,4,4,'qux','qux');
+INSERT INTO t2 
+SELECT NULL, a1.col_int, a1.col_int_key, a1.col_char, a1.col_char_key 
+FROM t1 a1, t1 a2, t1 a3, t1 a4, t1 a5, t1 a6, t1 a7, t1 a8, t1 a9, t1 a10;
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/encryption/r/innodb-log-encrypt.result b/mysql-test/suite/encryption/r/innodb-log-encrypt.result
index 3e281efd08a..655e3023f7a 100644
--- a/mysql-test/suite/encryption/r/innodb-log-encrypt.result
+++ b/mysql-test/suite/encryption/r/innodb-log-encrypt.result
@@ -8,7 +8,7 @@ begin
 declare current_num int;
 set current_num = 0;
 while current_num < repeat_count do
-insert into t1 values(current_num, substring(MD5(RAND()), -64), REPEAT('secredsecredsecred',10));
+insert into t1 values(current_num, substring(MD5(RAND()), -64), REPEAT('privatejanprivate',10));
 set current_num = current_num + 1;
 end while;
 end//
@@ -22,34 +22,34 @@ select count(*) from t1;
 count(*)
 2000
 # ibdata1 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in ibdata1
+NOT FOUND /privatejanprivate/ in ibdata1
 # t1 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in t1.ibd
+NOT FOUND /privatejanprivate/ in t1.ibd
 # log0 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in ib_logfile0
+NOT FOUND /privatejanprivate/ in ib_logfile0
 # log1 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in ib_logfile1
+NOT FOUND /privatejanprivate/ in ib_logfile1
 # Restart mysqld --innodb_encrypt_log=0
-insert into t1 values(5000, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5001, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5002, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5003, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5004, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
+insert into t1 values(5000, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5001, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5002, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5003, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5004, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
 # ibdata1 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in ibdata1
+NOT FOUND /privatejanprivate/ in ibdata1
 # t1 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in t1.ibd
+NOT FOUND /privatejanprivate/ in t1.ibd
 # log0 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in ib_logfile0
+NOT FOUND /privatejanprivate/ in ib_logfile0
 # log1 yes on expecting NOT FOUND
-NOT FOUND /secredsecred/ in ib_logfile1
+NOT FOUND /privatejanprivate/ in ib_logfile1
 # ibdata1 yes on expecting NOT FOUND
-NOT FOUND /notsecred/ in ibdata1
+NOT FOUND /publicmessage/ in ibdata1
 # t1 yes on expecting NOT FOUND
-NOT FOUND /notsecred/ in t1.ibd
+NOT FOUND /publicmessage/ in t1.ibd
 # log0 no on expecting FOUND/NOTFOUND depending where insert goes
-FOUND /notsecred/ in ib_logfile0
+FOUND /publicmessage/ in ib_logfile0
 # log1 no on expecting FOUND/NOTFOUND depending where insert goes
-NOT FOUND /notsecred/ in ib_logfile1
+NOT FOUND /publicmessage/ in ib_logfile1
 drop procedure innodb_insert_proc;
 drop table t1;
diff --git a/mysql-test/suite/encryption/t/innodb-log-encrypt-crash.opt b/mysql-test/suite/encryption/t/innodb-log-encrypt-crash.opt
new file mode 100644
index 00000000000..e76aa060879
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb-log-encrypt-crash.opt
@@ -0,0 +1,6 @@
+--innodb-encrypt-log=ON
+--plugin-load-add=$FILE_KEY_MANAGEMENT_SO
+--loose-file-key-management
+--loose-file-key-management-filename=$MYSQL_TEST_DIR/std_data/logkey.txt
+--file-key-management-encryption-algorithm=aes_cbc
+--innodb-buffer-pool-size=128M
diff --git a/mysql-test/suite/encryption/t/innodb-log-encrypt-crash.test b/mysql-test/suite/encryption/t/innodb-log-encrypt-crash.test
new file mode 100644
index 00000000000..8bb5f9f2c49
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb-log-encrypt-crash.test
@@ -0,0 +1,38 @@
+-- source include/have_innodb.inc
+-- source include/not_embedded.inc
+-- source filekeys_plugin.inc
+
+call mtr.add_suppression("InnoDB: New log files created, LSN=.*");
+call mtr.add_suppression("InnoDB: Creating foreign key constraint system tables.");
+call mtr.add_suppression("InnoDB: Error: Table .*");
+
+#
+# MDEV-9422: Checksum errors on restart when killing busy instance that uses encrypted XtraDB tables
+#
+
+CREATE TABLE t1 (
+  pk bigint auto_increment,
+  col_int int,
+  col_int_key int,
+  col_char char(12),
+  col_char_key char(12),
+  primary key (pk),
+  key (`col_int_key` ),
+  key (`col_char_key` )
+) ENGINE=InnoDB;
+CREATE TABLE t2 LIKE t1;
+
+INSERT INTO t1 VALUES (NULL,1,1,'foo','foo'),(NULL,2,2,'bar','bar'),(NULL,3,3,'baz','baz'),(NULL,4,4,'qux','qux');
+INSERT INTO t2 
+  SELECT NULL, a1.col_int, a1.col_int_key, a1.col_char, a1.col_char_key 
+    FROM t1 a1, t1 a2, t1 a3, t1 a4, t1 a5, t1 a6, t1 a7, t1 a8, t1 a9, t1 a10;
+
+--exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--shutdown_server 0
+--source include/wait_until_disconnected.inc
+
+--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/encryption/t/innodb-log-encrypt.test b/mysql-test/suite/encryption/t/innodb-log-encrypt.test
index b2abfadccc2..7c2e6f847b1 100644
--- a/mysql-test/suite/encryption/t/innodb-log-encrypt.test
+++ b/mysql-test/suite/encryption/t/innodb-log-encrypt.test
@@ -28,7 +28,7 @@ begin
   declare current_num int;
   set current_num = 0;
   while current_num < repeat_count do
-    insert into t1 values(current_num, substring(MD5(RAND()), -64), REPEAT('secredsecredsecred',10));
+    insert into t1 values(current_num, substring(MD5(RAND()), -64), REPEAT('privatejanprivate',10));
     set current_num = current_num + 1;
   end while;
 end//
@@ -43,13 +43,15 @@ set autocommit=1;
 update t1 set c1 = c1 +1;
 select count(*) from t1;
 
+-- source include/restart_mysqld.inc
+
 --let $MYSQLD_DATADIR=`select @@datadir`
 --let ib1_IBD = $MYSQLD_DATADIR/ibdata1
 --let t1_IBD = $MYSQLD_DATADIR/test/t1.ibd
 --let log0 = $MYSQLD_DATADIR/ib_logfile0
 --let log1 = $MYSQLD_DATADIR/ib_logfile1
 --let SEARCH_RANGE = 10000000
---let SEARCH_PATTERN=secredsecred
+--let SEARCH_PATTERN=privatejanprivate
 
 --echo # ibdata1 yes on expecting NOT FOUND
 -- let SEARCH_FILE=$ib1_IBD
@@ -68,13 +70,13 @@ select count(*) from t1;
 -- let $restart_parameters=--innodb_encrypt_log=0
 -- source include/restart_mysqld.inc
 
-insert into t1 values(5000, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5001, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5002, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5003, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
-insert into t1 values(5004, substring(MD5(RAND()), -64), REPEAT('notsecred',10));
+insert into t1 values(5000, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5001, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5002, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5003, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
+insert into t1 values(5004, substring(MD5(RAND()), -64), REPEAT('publicmessage',10));
 
---let SEARCH_PATTERN=secredsecred
+--let SEARCH_PATTERN=privatejanprivate
 --echo # ibdata1 yes on expecting NOT FOUND
 -- let SEARCH_FILE=$ib1_IBD
 -- source include/search_pattern_in_file.inc
@@ -88,7 +90,7 @@ insert into t1 values(5004, substring(MD5(RAND()), -64), REPEAT('notsecred',10))
 -- let SEARCH_FILE=$log1
 -- source include/search_pattern_in_file.inc
 
---let SEARCH_PATTERN=notsecred
+--let SEARCH_PATTERN=publicmessage
 --echo # ibdata1 yes on expecting NOT FOUND
 -- let SEARCH_FILE=$ib1_IBD
 -- source include/search_pattern_in_file.inc
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index e4a06fcc532..292953854f7 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -498,6 +498,18 @@ use these free frames to read in pages when we start applying the
 log records to the database. */
 extern ulint	recv_n_pool_free_frames;
 
+/******************************************************//**
+Checks the 4-byte checksum to the trailer checksum field of a log
+block.  We also accept a log block in the old format before
+InnoDB-3.23.52 where the checksum field contains the log block number.
+@return TRUE if ok, or if the log block may be in the format of InnoDB
+version predating 3.23.52 */
+ibool
+log_block_checksum_is_ok_or_old_format(
+/*===================================*/
+	const byte*	block,	/*!< in: pointer to a log block */
+	bool            print_err); /*!< in print error ? */
+
 #ifndef UNIV_NONINL
 #include "log0recv.ic"
 #endif
diff --git a/storage/innobase/log/log0crypt.cc b/storage/innobase/log/log0crypt.cc
index e90533c2e76..00de0252d6e 100644
--- a/storage/innobase/log/log0crypt.cc
+++ b/storage/innobase/log/log0crypt.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2015, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
 
 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
@@ -156,12 +156,21 @@ log_blocks_crypt(
 			info ? info->key_version : 0,
 			log_block_start_lsn);
 #endif
+		/* If no key is found from checkpoint assume the log_block
+		to be unencrypted. If checkpoint contains the encryption key
+		compare log_block current checksum, if checksum matches,
+		block can't be encrypted. */
 		if (info == NULL ||
-		    info->key_version == UNENCRYPTED_KEY_VER) {
+		    info->key_version == UNENCRYPTED_KEY_VER ||
+			(log_block_checksum_is_ok_or_old_format(log_block, false) &&
+			 what == ENCRYPTION_FLAG_DECRYPT)) {
 			memcpy(dst_block, log_block, OS_FILE_LOG_BLOCK_SIZE);
 			goto next;
 		}
 
+		ut_ad(what == ENCRYPTION_FLAG_DECRYPT ? !log_block_checksum_is_ok_or_old_format(log_block, false) :
+			log_block_checksum_is_ok_or_old_format(log_block, false));
+
 		// Assume log block header is not encrypted
 		memcpy(dst_block, log_block, LOG_BLOCK_HDR_SIZE);
 
diff --git a/storage/innobase/log/log0log.cc b/storage/innobase/log/log0log.cc
index fd3646a1691..31104b395c1 100644
--- a/storage/innobase/log/log0log.cc
+++ b/storage/innobase/log/log0log.cc
@@ -2,7 +2,7 @@
 
 Copyright (c) 1995, 2015, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2009, Google Inc.
-Copyright (c) 2013, SkySQL Ab. All Rights Reserved.
+Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -52,6 +52,9 @@ Created 12/9/1995 Heikki Tuuri
 #include "trx0roll.h"
 #include "srv0mon.h"
 
+/* Used for debugging */
+// #define DEBUG_CRYPT 1
+
 /*
 General philosophy of InnoDB redo-logs:
 
@@ -2358,8 +2361,24 @@ loop:
 	       (ulint) (source_offset % UNIV_PAGE_SIZE),
 		len, buf, NULL, 0);
 
+#ifdef DEBUG_CRYPT
+	fprintf(stderr, "BEFORE DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx offset %lu\n",
+		log_block_get_hdr_no(buf),
+			log_block_get_checkpoint_no(buf),
+			log_block_calc_checksum(buf),
+		log_block_get_checksum(buf), source_offset);
+#endif
+
 	log_decrypt_after_read(buf, len);
 
+#ifdef DEBUG_CRYPT
+	fprintf(stderr, "AFTER DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx\n",
+			log_block_get_hdr_no(buf),
+			log_block_get_checkpoint_no(buf),
+			log_block_calc_checksum(buf),
+			log_block_get_checksum(buf));
+#endif
+
 	start_lsn += len;
 	buf += len;
 
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index c84e8277f07..6e09143fb96 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -923,11 +923,11 @@ block.  We also accept a log block in the old format before
 InnoDB-3.23.52 where the checksum field contains the log block number.
 @return TRUE if ok, or if the log block may be in the format of InnoDB
 version predating 3.23.52 */
-static
 ibool
 log_block_checksum_is_ok_or_old_format(
 /*===================================*/
-	const byte*	block)	/*!< in: pointer to a log block */
+	const byte*	block,	/*!< in: pointer to a log block */
+	bool            print_err) /*!< in print error ? */
 {
 #ifdef UNIV_LOG_DEBUG
 	return(TRUE);
@@ -950,11 +950,13 @@ log_block_checksum_is_ok_or_old_format(
 		return(TRUE);
 	}
 
-	fprintf(stderr, "BROKEN: block: %lu checkpoint: %lu %.8lx %.8lx\n",
-		log_block_get_hdr_no(block),
-		log_block_get_checkpoint_no(block),
-		log_block_calc_checksum(block),
-		log_block_get_checksum(block));
+	if (print_err) {
+		fprintf(stderr, "BROKEN: block: %lu checkpoint: %lu %.8lx %.8lx\n",
+			log_block_get_hdr_no(block),
+			log_block_get_checkpoint_no(block),
+			log_block_calc_checksum(block),
+			log_block_get_checksum(block));
+	}
 
 	return(FALSE);
 }
@@ -2686,12 +2688,12 @@ recv_scan_log_recs(
 		log_block_convert_lsn_to_no(scanned_lsn));
 		*/
 		if (no != log_block_convert_lsn_to_no(scanned_lsn)
-		    || !log_block_checksum_is_ok_or_old_format(log_block)) {
+		    || !log_block_checksum_is_ok_or_old_format(log_block, true)) {
 			log_crypt_err_t log_crypt_err;
 
 			if (no == log_block_convert_lsn_to_no(scanned_lsn)
 			    && !log_block_checksum_is_ok_or_old_format(
-				    log_block)) {
+				    log_block, true)) {
 				fprintf(stderr,
 					"InnoDB: Log block no %lu at"
 					" lsn " LSN_PF " has\n"
diff --git a/storage/xtradb/include/log0recv.h b/storage/xtradb/include/log0recv.h
index b23a140aac2..8fc5daaef1d 100644
--- a/storage/xtradb/include/log0recv.h
+++ b/storage/xtradb/include/log0recv.h
@@ -43,7 +43,8 @@ UNIV_INTERN
 ibool
 log_block_checksum_is_ok_or_old_format(
 /*===================================*/
-	const byte*	block);	/*!< in: pointer to a log block */
+	const byte*	block,	/*!< in: pointer to a log block */
+	bool		print_err); /*!< in print error ? */
 
 /*******************************************************//**
 Calculates the new value for lsn when more data is added to the log. */
diff --git a/storage/xtradb/log/log0crypt.cc b/storage/xtradb/log/log0crypt.cc
index e90533c2e76..852148899e9 100644
--- a/storage/xtradb/log/log0crypt.cc
+++ b/storage/xtradb/log/log0crypt.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (C) 2014, 2015, MariaDB Corporation. All Rights Reserved.
+Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
 
 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
@@ -36,6 +36,8 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 
 #include "my_crypt.h"
 
+/* Used for debugging */
+// #define DEBUG_CRYPT 1
 #define UNENCRYPTED_KEY_VER	0
 
 /* If true, enable redo log encryption. */
@@ -97,16 +99,24 @@ get_crypt_info(
 {
 	/* so that no one is modifying array while we search */
 	ut_ad(mutex_own(&(log_sys->mutex)));
+	size_t items = crypt_info.size();
 
 	/* a log block only stores 4-bytes of checkpoint no */
 	checkpoint_no &= 0xFFFFFFFF;
-	for (size_t i = 0; i < crypt_info.size(); i++) {
+	for (size_t i = 0; i < items; i++) {
 		struct crypt_info_t* it = &crypt_info[i];
 
 		if (it->checkpoint_no == checkpoint_no) {
 			return it;
 		}
 	}
+
+	/* If checkpoint contains more than one key and we did not
+	find the correct one use the first one. */
+	if (items) {
+		return (&crypt_info[0]);
+	}
+
 	return NULL;
 }
 
@@ -131,7 +141,8 @@ log_blocks_crypt(
 	const byte* block,	/*!< in: blocks before encrypt/decrypt*/
 	ulint size,		/*!< in: size of block */
 	byte* dst_block,	/*!< out: blocks after encrypt/decrypt */
-	int what)		/*!< in: encrypt or decrypt*/
+	int what,		/*!< in: encrypt or decrypt*/
+	const crypt_info_t* crypt_info) /*!< in: crypt info or NULL */
 {
 	byte *log_block = (byte*)block;
 	Crypt_result rc = MY_AES_OK;
@@ -146,7 +157,8 @@ log_blocks_crypt(
 		lsn_t log_block_start_lsn = log_block_get_start_lsn(
 			lsn, log_block_no);
 
-		const crypt_info_t* info = get_crypt_info(log_block);
+		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",
@@ -156,12 +168,21 @@ log_blocks_crypt(
 			info ? info->key_version : 0,
 			log_block_start_lsn);
 #endif
+		/* If no key is found from checkpoint assume the log_block
+		to be unencrypted. If checkpoint contains the encryption key
+		compare log_block current checksum, if checksum matches,
+		block can't be encrypted. */
 		if (info == NULL ||
-		    info->key_version == UNENCRYPTED_KEY_VER) {
+		    info->key_version == UNENCRYPTED_KEY_VER ||
+			(log_block_checksum_is_ok_or_old_format(log_block, false) &&
+			 what == ENCRYPTION_FLAG_DECRYPT)) {
 			memcpy(dst_block, log_block, OS_FILE_LOG_BLOCK_SIZE);
 			goto next;
 		}
 
+		ut_ad(what == ENCRYPTION_FLAG_DECRYPT ? !log_block_checksum_is_ok_or_old_format(log_block, false) :
+			log_block_checksum_is_ok_or_old_format(log_block, false));
+
 		// Assume log block header is not encrypted
 		memcpy(dst_block, log_block, LOG_BLOCK_HDR_SIZE);
 
@@ -292,7 +313,7 @@ log_blocks_encrypt(
 	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);
+	return log_blocks_crypt(block, size, dst_block, ENCRYPTION_FLAG_ENCRYPT, NULL);
 }
 
 /*********************************************************************//**
@@ -355,14 +376,16 @@ log_encrypt_before_write(
 		return;
 	}
 
-	if (info->key_version == UNENCRYPTED_KEY_VER) {
+	/* If the key is not encrypted or user has requested not to
+	encrypt, do not change log block. */
+	if (info->key_version == UNENCRYPTED_KEY_VER || !srv_encrypt_log) {
 		return;
 	}
 
 	byte* dst_frame = (byte*)malloc(size);
 
 	//encrypt log blocks content
-	Crypt_result result = log_blocks_crypt(block, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT);
+	Crypt_result result = log_blocks_crypt(block, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		ut_ad(block[0] == dst_frame[0]);
@@ -388,7 +411,7 @@ log_decrypt_after_read(
 	byte* dst_frame = (byte*)malloc(size);
 
 	// decrypt log blocks content
-	Crypt_result result = log_blocks_crypt(frame, size, dst_frame, ENCRYPTION_FLAG_DECRYPT);
+	Crypt_result result = log_blocks_crypt(frame, size, dst_frame, ENCRYPTION_FLAG_DECRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		memcpy(frame, dst_frame, size);
diff --git a/storage/xtradb/log/log0log.cc b/storage/xtradb/log/log0log.cc
index 853afe70100..36531f3c6f4 100644
--- a/storage/xtradb/log/log0log.cc
+++ b/storage/xtradb/log/log0log.cc
@@ -2,6 +2,7 @@
 
 Copyright (c) 1995, 2015, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2009, Google Inc.
+Copyright (C) 2014, 2016, MariaDB Corporation. All Rights Reserved.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -33,10 +34,13 @@ Created 12/9/1995 Heikki Tuuri
 #include "config.h"
 #ifdef HAVE_ALLOCA_H
 #include "alloca.h"
-#elif defined(HAVE_MALLOC_H) 
+#elif defined(HAVE_MALLOC_H)
 #include "malloc.h"
 #endif
 
+/* Used for debugging */
+// #define DEBUG_CRYPT 1
+
 #include "log0log.h"
 
 #ifdef UNIV_NONINL
@@ -1394,7 +1398,6 @@ log_group_file_header_flush(
 Stores a 4-byte checksum to the trailer checksum field of a log block
 before writing it to a log file. This checksum is used in recovery to
 check the consistency of a log block. */
-static
 void
 log_block_store_checksum(
 /*=====================*/
@@ -1512,6 +1515,14 @@ loop:
 		log_encrypt_before_write(log_sys->next_checkpoint_no,
 					 buf, write_len);
 
+#ifdef DEBUG_CRYPT
+		fprintf(stderr, "WRITE: block: %lu checkpoint: %lu %.8lx %.8lx\n",
+			log_block_get_hdr_no(buf),
+			log_block_get_checkpoint_no(buf),
+			log_block_calc_checksum(buf),
+			log_block_get_checksum(buf));
+#endif
+
 		fil_io(OS_FILE_WRITE | OS_FILE_LOG, true, group->space_id, 0,
 		       (ulint) (next_offset / UNIV_PAGE_SIZE),
 		       (ulint) (next_offset % UNIV_PAGE_SIZE), write_len, buf,
@@ -2320,7 +2331,10 @@ log_checkpoint(
 	* the checkpoint info has been written and THEN blocks will be encrypted
 	* with new key
 	*/
-	log_crypt_set_ver_and_key(log_sys->next_checkpoint_no + 1);
+	if (srv_encrypt_log) {
+		log_crypt_set_ver_and_key(log_sys->next_checkpoint_no + 1);
+	}
+
 	log_groups_write_checkpoint_info();
 
 	MONITOR_INC(MONITOR_NUM_CHECKPOINT);
@@ -2585,8 +2599,24 @@ loop:
 		mutex_enter(&log_sys->mutex);
 	}
 
+#ifdef DEBUG_CRYPT
+	fprintf(stderr, "BEFORE DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx offset %lu\n",
+		log_block_get_hdr_no(buf),
+			log_block_get_checkpoint_no(buf),
+			log_block_calc_checksum(buf),
+		log_block_get_checksum(buf), source_offset);
+#endif
+
 	log_decrypt_after_read(buf, len);
 
+#ifdef DEBUG_CRYPT
+	fprintf(stderr, "AFTER DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx\n",
+			log_block_get_hdr_no(buf),
+			log_block_get_checkpoint_no(buf),
+			log_block_calc_checksum(buf),
+			log_block_get_checksum(buf));
+#endif
+
 	if (release_mutex) {
 		mutex_exit(&log_sys->mutex);
 	}
diff --git a/storage/xtradb/log/log0online.cc b/storage/xtradb/log/log0online.cc
index 92f03f0e6a9..51a9fa8f6c5 100644
--- a/storage/xtradb/log/log0online.cc
+++ b/storage/xtradb/log/log0online.cc
@@ -905,7 +905,7 @@ log_online_is_valid_log_seg(
 	const byte* log_block)	/*!< in: read log data */
 {
 	ibool checksum_is_ok
-		= log_block_checksum_is_ok_or_old_format(log_block);
+		= log_block_checksum_is_ok_or_old_format(log_block, true);
 
 	if (!checksum_is_ok) {
 
diff --git a/storage/xtradb/log/log0recv.cc b/storage/xtradb/log/log0recv.cc
index c9bf5cf3f9e..f98adbbca08 100644
--- a/storage/xtradb/log/log0recv.cc
+++ b/storage/xtradb/log/log0recv.cc
@@ -2,7 +2,7 @@
 
 Copyright (c) 1997, 2015, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2013, 2015, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2013, 2016, MariaDB Corporation. All Rights Reserved.
 
 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
@@ -43,7 +43,7 @@ Created 9/20/1997 Heikki Tuuri
 #include "config.h"
 #ifdef HAVE_ALLOCA_H
 #include "alloca.h"
-#elif defined(HAVE_MALLOC_H) 
+#elif defined(HAVE_MALLOC_H)
 #include "malloc.h"
 #endif
 
@@ -932,7 +932,8 @@ UNIV_INTERN
 ibool
 log_block_checksum_is_ok_or_old_format(
 /*===================================*/
-	const byte*	block)	/*!< in: pointer to a log block */
+	const byte*	block,		/*!< in: pointer to a log block */
+	bool		print_err)	/*!< in print if error found */
 {
 #ifdef UNIV_LOG_DEBUG
 	return(TRUE);
@@ -1015,11 +1016,13 @@ log_block_checksum_is_ok_or_old_format(
 		return(TRUE);
 	}
 
-	fprintf(stderr, "BROKEN: block: %lu checkpoint: %lu %.8lx %.8lx\n",
-		log_block_get_hdr_no(block),
-		log_block_get_checkpoint_no(block),
-		log_block_calc_checksum(block),
-		log_block_get_checksum(block));
+	if (print_err) {
+		fprintf(stderr, "BROKEN: block: %lu checkpoint: %lu %.8lx %.8lx\n",
+			log_block_get_hdr_no(block),
+			log_block_get_checkpoint_no(block),
+			log_block_calc_checksum(block),
+			log_block_get_checksum(block));
+	}
 
 	return(FALSE);
 }
@@ -2734,6 +2737,7 @@ recv_scan_log_recs(
 	ibool		finished;
 	ulint		data_len;
 	ibool		more_data;
+	bool		maybe_encrypted=false;
 
 	ut_ad(start_lsn % OS_FILE_LOG_BLOCK_SIZE == 0);
 	ut_ad(len % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -2748,6 +2752,8 @@ recv_scan_log_recs(
 	*err = DB_SUCCESS;
 
 	do {
+		log_crypt_err_t log_crypt_err;
+
 		no = log_block_get_hdr_no(log_block);
 		/*
 		fprintf(stderr, "Log block header no %lu\n", no);
@@ -2755,13 +2761,13 @@ recv_scan_log_recs(
 		fprintf(stderr, "Scanned lsn no %lu\n",
 		log_block_convert_lsn_to_no(scanned_lsn));
 		*/
+
 		if (no != log_block_convert_lsn_to_no(scanned_lsn)
-		    || !log_block_checksum_is_ok_or_old_format(log_block)) {
-			log_crypt_err_t log_crypt_err;
+		    || !log_block_checksum_is_ok_or_old_format(log_block, true)) {
 
 			if (no == log_block_convert_lsn_to_no(scanned_lsn)
 			    && !log_block_checksum_is_ok_or_old_format(
-				    log_block)) {
+				    log_block, true)) {
 				fprintf(stderr,
 					"InnoDB: Log block no %lu at"
 					" lsn " LSN_PF " has\n"
@@ -2775,12 +2781,14 @@ recv_scan_log_recs(
 						log_block));
 			}
 
+			maybe_encrypted = log_crypt_block_maybe_encrypted(log_block,
+					&log_crypt_err);
+
 			/* Garbage or an incompletely written log block */
 
 			finished = TRUE;
 
-			if (log_crypt_block_maybe_encrypted(log_block,
-					&log_crypt_err)) {
+			if (maybe_encrypted) {
 				/* Log block maybe encrypted finish processing*/
 				log_crypt_print_error(log_crypt_err);
 				*err = DB_ERROR;
@@ -2790,12 +2798,13 @@ recv_scan_log_recs(
 			/* Stop if we encounter a garbage log block */
 			if (!srv_force_recovery) {
 				fputs("InnoDB: Set innodb_force_recovery"
-				      " to ignore this error.\n", stderr);
+					" to ignore this error.\n", stderr);
 				*err = DB_ERROR;
 				return (TRUE);
 			}
 
 			break;
+
 		}
 
 		if (log_block_get_flush_bit(log_block)) {