MDEV-8410: Changing file-key-management to example-key-management causes crash and no real error

MDEV-8409: Changing file-key-management-encryption-algorithm causes crash and no real info why

Analysis: Both bugs has two different error cases. Firstly, at startup
when server reads latest checkpoint but requested key_version,
key management plugin or encryption algorithm or method is not found
leading corrupted log entry. Secondly, similarly when reading system
tablespace if requested key_version, key management plugin or encryption
algorithm or method is not found leading buffer pool page corruption.

Fix: Firsly, when reading checkpoint at startup check if the log record
may be encrypted and if we find that it could be encrypted, print error
message and do not start server. Secondly, if page is buffer pool seems
corrupted but we find out that there is crypt_info, print additional
error message before asserting.

storage/innobase/buf/buf0buf.cc

@@ -4272,6 +4272,46 @@ buf_mark_space_corrupt(
 	return(ret);
 }
 
+/********************************************************************//**
+Check if page is maybe compressed, encrypted or both when we encounter
+corrupted page. Note that we can't be 100% sure if page is corrupted
+or decrypt/decompress just failed.
+*/
+static
+void
+buf_page_check_corrupt(
+/*===================*/
+	const buf_page_t*	bpage)	/*!< in/out: buffer page read from disk */
+{
+	ulint zip_size = buf_page_get_zip_size(bpage);
+	byte* dst_frame = (zip_size) ? bpage->zip.data :
+		((buf_block_t*) bpage)->frame;
+	unsigned key_version =
+		mach_read_from_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
+	bool page_compressed = fil_page_is_compressed(dst_frame);
+	bool page_compressed_encrypted = fil_page_is_compressed_encrypted(dst_frame);
+	ulint space_id = mach_read_from_4(
+		dst_frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
+	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space_id);
+	fil_space_t* space = fil_space_found_by_id(space_id);
+
+	if (key_version != 0 ||
+		(crypt_data && crypt_data->type != CRYPT_SCHEME_UNENCRYPTED) ||
+		page_compressed || page_compressed_encrypted) {
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"Maybe corruption: Block space_id %lu in file %s maybe corrupted\n"
+			"Reason could be that key_version in page %u \n"
+			"or in crypt_data %p could not be found,\n"
+			"key management plugin is not found or\n"
+			"used encryption algorithm or method does not match.\n"
+			"Page compressed %d, compressed and encrypted %d.\n",
+			space_id, space ? space->name : "NULL",
+			key_version,
+			crypt_data,
+			page_compressed, page_compressed_encrypted);
+	}
+}
+
 /********************************************************************//**
 Completes an asynchronous read or write request of a file page to or from
 the buffer pool.
@@ -4438,6 +4478,8 @@ buf_page_io_complete(
 				    && buf_mark_space_corrupt(bpage)) {
 					return(false);
 				} else {
+					buf_page_check_corrupt(bpage);
+
 					fputs("InnoDB: Ending processing"
 					      " because of"
 					      " a corrupt database page.\n",

storage/innobase/include/log0crypt.h

+/*****************************************************************************
+
+Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
+Copyright (C) 2014, 2015, 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
+Foundation; version 2 of the License.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+
+*****************************************************************************/
 /**************************************************//**
 @file include/log0crypt.h
 Innodb log encrypt/decrypt
 
 Created 11/25/2013 Minli Zhu
+Modified           Jan Lindström jan.lindstrom@mariadb.com
 *******************************************************/
 #ifndef log0crypt_h
 #define log0crypt_h
@@ -22,7 +41,7 @@ UNIV_INTERN
 void
 log_crypt_set_ver_and_key(
 /*======================*/
-	ib_uint64_t next_checkpoint_no);
+	ib_uint64_t next_checkpoint_no);/*!< in: next checkpoint no */
 
 
 /*********************************************************************//**
@@ -43,17 +62,17 @@ UNIV_INTERN
 bool
 log_crypt_read_checkpoint_buf(
 /*===========================*/
-	const byte*	buf);			/*!< in: checkpoint buffer */
+	const byte*	buf);		/*!< in: checkpoint buffer */
 
 /********************************************************
 Encrypt one or more log block before it is flushed to disk */
 UNIV_INTERN
 void
 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 */
-	const ulint size);		/*!< in: size of log blocks */
+/*=====================*/
+	ib_uint64_t	next_checkpoint_no,	/*!< in: log group to be flushed */
+	byte* 		block,			/*!< in/out: pointer to a log block */
+	const ulint 	size);			/*!< in: size of log blocks */
 
 /********************************************************
 Decrypt a specified log segment after they are read from a log file to a buffer.
@@ -61,8 +80,41 @@ Decrypt a specified log segment after they are read from a log file to a buffer.
 UNIV_INTERN
 void
 log_decrypt_after_read(
-/*==========================*/
-	byte* frame,	/*!< in/out: log segment */
-	const ulint size);	/*!< in: log segment size */
+/*===================*/
+	byte*		frame,	/*!< in/out: log segment */
+	const ulint	size);	/*!< in: log segment size */
+
+/* Error codes for crypt info */
+typedef enum {
+	LOG_UNENCRYPTED = 0,
+	LOG_CRYPT_KEY_NOT_FOUND = 1,
+	LOG_DECRYPT_MAYBE_FAILED = 2
+} log_crypt_err_t;
+
+/********************************************************
+Check is the checkpoint information encrypted. This check
+is based on fact has log group crypt info and based
+on this crypt info was the key version different from
+unencrypted key version. There is no realible way to
+distinguish encrypted log block from corrupted log block,
+but if log block corruption is found this function is
+used to find out if log block is maybe encrypted but
+encryption key, key management plugin or encryption
+algorithm does not match.
+@return TRUE, if log block may be encrypted */
+UNIV_INTERN
+ibool
+log_crypt_block_maybe_encrypted(
+/*============================*/
+	const byte* 		log_block, 	/*!< in: log block */
+	log_crypt_err_t* 	err_info); 	/*!< out: error info */
+
+/********************************************************
+Print crypt error message to error log */
+UNIV_INTERN
+void
+log_crypt_print_error(
+/*==================*/
+	log_crypt_err_t 	err_info); 	/*!< out: error info */
 
 #endif  // log0crypt.h

storage/innobase/log/log0crypt.cc

@@ -86,6 +86,9 @@ log_block_get_start_lsn(
 	return start_lsn;
 }
 
+/*********************************************************************//**
+Get crypt info from checkpoint.
+@return a crypt info or NULL if not present. */
 static
 const crypt_info_t*
 get_crypt_info(
@@ -107,6 +110,9 @@ get_crypt_info(
 	return NULL;
 }
 
+/*********************************************************************//**
+Get crypt info from log block
+@return a crypt info or NULL if not present. */
 static
 const crypt_info_t*
 get_crypt_info(
@@ -239,14 +245,23 @@ init_crypt_key(
 	return true;
 }
 
-static bool mysort(const crypt_info_t& i,
-		   const crypt_info_t& j)
+/*********************************************************************//**
+Compare function for checkpoint numbers
+@return true if first checkpoint is larger than second one */
+static
+bool
+mysort(const crypt_info_t& i,
+       const crypt_info_t& j)
 {
 	return i.checkpoint_no > j.checkpoint_no;
 }
 
+/*********************************************************************//**
+Add crypt info to set if it is not already present
+@return true if successfull, false if not- */
 static
-bool add_crypt_info(crypt_info_t* info)
+bool
+add_crypt_info(crypt_info_t* info)
 {
 	/* so that no one is searching array while we modify it */
 	ut_ad(mutex_own(&(log_sys->mutex)));
@@ -332,7 +347,7 @@ Encrypt one or more log block before it is flushed to disk */
 UNIV_INTERN
 void
 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 */
 	const ulint size)		/*!< in: size of log blocks */
@@ -369,7 +384,7 @@ Decrypt a specified log segment after they are read from a log file to a buffer.
 */
 void
 log_decrypt_after_read(
-/*==========================*/
+/*===================*/
 	byte* frame,	        /*!< in/out: log segment */
 	const ulint size)	/*!< in: log segment size */
 {
@@ -499,3 +514,74 @@ log_crypt_read_checkpoint_buf(
 	return true;
 }
 
+/********************************************************
+Check is the checkpoint information encrypted. This check
+is based on fact has log group crypt info and based
+on this crypt info was the key version different from
+unencrypted key version. There is no realible way to
+distinguish encrypted log block from corrupted log block,
+but if log block corruption is found this function is
+used to find out if log block is maybe encrypted but
+encryption key, key management plugin or encryption
+algorithm does not match.
+@return TRUE, if log block may be encrypted */
+UNIV_INTERN
+ibool
+log_crypt_block_maybe_encrypted(
+/*============================*/
+	const byte* 		log_block,	/*!< in: log block */
+	log_crypt_err_t* 	err_info)	/*!< out: error info */
+{
+	ibool maybe_encrypted = FALSE;
+	const crypt_info_t* crypt_info;
+
+	*err_info = LOG_UNENCRYPTED;
+	crypt_info = get_crypt_info(log_block);
+
+	if (crypt_info &&
+	    crypt_info->key_version != UNENCRYPTED_KEY_VER) {
+		byte mysqld_key[MY_AES_BLOCK_SIZE] = {0};
+		uint keylen= sizeof(mysqld_key);
+
+		/* Log block contains crypt info and based on key
+		version block could be encrypted. */
+		*err_info = LOG_DECRYPT_MAYBE_FAILED;
+		maybe_encrypted = TRUE;
+
+		if (encryption_key_get(LOG_DEFAULT_ENCRYPTION_KEY,
+				       crypt_info->key_version, mysqld_key, &keylen)) {
+			*err_info = LOG_CRYPT_KEY_NOT_FOUND;
+		}
+	}
+
+	return (maybe_encrypted);
+}
+
+/********************************************************
+Print crypt error message to error log */
+UNIV_INTERN
+void
+log_crypt_print_error(
+/*==================*/
+	log_crypt_err_t 	err_info) 	/*!< out: error info */
+{
+	switch(err_info) {
+	case LOG_CRYPT_KEY_NOT_FOUND:
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"Redo log crypto: getting mysqld crypto key "
+			"from key version failed. Reason could be that "
+			"requested key version is not found or required "
+			"encryption key management plugin is not found.");
+		break;
+	case LOG_DECRYPT_MAYBE_FAILED:
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"Redo log crypto: failed to decrypt log block. "
+			"Reason could be that requested key version is "
+			"not found, required encryption key management "
+			"plugin is not found or configured encryption "
+			"algorithm and/or method does not match.");
+		break;
+	default:
+		ut_error; /* Real bug */
+	}
+}

storage/innobase/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, SkySQL Ab. All Rights Reserved.
+Copyright (c) 2013, 2015, 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 @@ Created 9/20/1997 Heikki Tuuri
 #include "log0recv.ic"
 #endif
 
+#include "log0crypt.h"
+
 #include "mem0mem.h"
 #include "buf0buf.h"
 #include "buf0flu.h"
@@ -2704,8 +2706,9 @@ recv_scan_log_recs(
 	lsn_t*		contiguous_lsn,	/*!< in/out: it is known that all log
 					groups contain contiguous log data up
 					to this lsn */
-	lsn_t*		group_scanned_lsn)/*!< out: scanning succeeded up to
+	lsn_t*		group_scanned_lsn,/*!< out: scanning succeeded up to
 					this lsn */
+	dberr_t*	err)		/*!< out: error code or DB_SUCCESS */
 {
 	const byte*	log_block;
 	ulint		no;
@@ -2724,6 +2727,7 @@ recv_scan_log_recs(
 	log_block = buf;
 	scanned_lsn = start_lsn;
 	more_data = FALSE;
+	*err = DB_SUCCESS;
 
 	do {
 		no = log_block_get_hdr_no(log_block);
@@ -2735,6 +2739,7 @@ recv_scan_log_recs(
 		*/
 		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;
 
 			if (no == log_block_convert_lsn_to_no(scanned_lsn)
 			    && !log_block_checksum_is_ok_or_old_format(
@@ -2756,6 +2761,14 @@ recv_scan_log_recs(
 
 			finished = TRUE;
 
+			if (log_crypt_block_maybe_encrypted(log_block,
+					&log_crypt_err)) {
+				/* Log block maybe encrypted */
+				log_crypt_print_error(log_crypt_err);
+				*err = DB_ERROR;
+				return (TRUE);
+			}
+
 			/* Crash if we encounter a garbage log block */
 			if (!srv_force_recovery) {
 				fputs("InnoDB: Set innodb_force_recovery"
@@ -2944,14 +2957,16 @@ recv_group_scan_log_recs(
 	lsn_t*		contiguous_lsn,	/*!< in/out: it is known that all log
 					groups contain contiguous log data up
 					to this lsn */
-	lsn_t*		group_scanned_lsn)/*!< out: scanning succeeded up to
+	lsn_t*		group_scanned_lsn,/*!< out: scanning succeeded up to
 					this lsn */
+	dberr_t*	err)		/*!< out: error code or DB_SUCCESS */
 {
 	ibool	finished;
 	lsn_t	start_lsn;
 	lsn_t	end_lsn;
 
 	finished = FALSE;
+	*err = DB_SUCCESS;
 
 	start_lsn = *contiguous_lsn;
 
@@ -2966,7 +2981,13 @@ recv_group_scan_log_recs(
 			- (recv_n_pool_free_frames * srv_buf_pool_instances))
 			* UNIV_PAGE_SIZE,
 			TRUE, log_sys->buf, RECV_SCAN_SIZE,
-			start_lsn, contiguous_lsn, group_scanned_lsn);
+			start_lsn, contiguous_lsn, group_scanned_lsn,
+			err);
+
+		if (*err != DB_SUCCESS) {
+			break;
+		}
+
 		start_lsn = end_lsn;
 	}
 
@@ -3174,6 +3195,7 @@ recv_recovery_from_checkpoint_start_func(
 		up_to_date_group = max_cp_group;
 	} else {
 		ulint	capacity;
+		dberr_t err;
 
 		/* Try to recover the remaining part from logs: first from
 		the logs of the archived group */
@@ -3193,8 +3215,9 @@ recv_recovery_from_checkpoint_start_func(
 		}
 
 		recv_group_scan_log_recs(group, &contiguous_lsn,
-					 &group_scanned_lsn);
-		if (recv_sys->scanned_lsn < checkpoint_lsn) {
+			&group_scanned_lsn, &err);
+
+		if (err != DB_SUCCESS || recv_sys->scanned_lsn < checkpoint_lsn) {
 
 			mutex_exit(&(log_sys->mutex));
 
@@ -3226,9 +3249,15 @@ recv_recovery_from_checkpoint_start_func(
 #ifdef UNIV_LOG_ARCHIVE
 		lsn_t	old_scanned_lsn	= recv_sys->scanned_lsn;
 #endif /* UNIV_LOG_ARCHIVE */
+		dberr_t err;
 
 		recv_group_scan_log_recs(group, &contiguous_lsn,
-					 &group_scanned_lsn);
+			&group_scanned_lsn, &err);
+
+		if (err != DB_SUCCESS) {
+			return (err);
+		}
+
 		group->scanned_lsn = group_scanned_lsn;
 
 #ifdef UNIV_LOG_ARCHIVE
@@ -3707,6 +3736,7 @@ log_group_recover_from_archive_file(
 	os_offset_t	file_size;
 	int		input_char;
 	char		name[10000];
+	dberr_t		err;
 
 	ut_a(0);
 
@@ -3851,7 +3881,11 @@ log_group_recover_from_archive_file(
 			(buf_pool_get_n_pages()
 			- (recv_n_pool_free_frames * srv_buf_pool_instances))
 			* UNIV_PAGE_SIZE, TRUE, buf, len, start_lsn,
-			&dummy_lsn, &scanned_lsn);
+			&dummy_lsn, &scanned_lsn, &err);
+
+		if (err != DB_SUCCESS) {
+			return (FALSE);
+		}
 
 		if (scanned_lsn == file_end_lsn) {
 

storage/xtradb/buf/buf0buf.cc

@@ -4320,6 +4320,46 @@ buf_mark_space_corrupt(
 	return(ret);
 }
 
+/********************************************************************//**
+Check if page is maybe compressed, encrypted or both when we encounter
+corrupted page. Note that we can't be 100% sure if page is corrupted
+or decrypt/decompress just failed.
+*/
+static
+void
+buf_page_check_corrupt(
+/*===================*/
+	const buf_page_t*	bpage)	/*!< in/out: buffer page read from disk */
+{
+	ulint zip_size = buf_page_get_zip_size(bpage);
+	byte* dst_frame = (zip_size) ? bpage->zip.data :
+		((buf_block_t*) bpage)->frame;
+	unsigned key_version =
+		mach_read_from_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
+	bool page_compressed = fil_page_is_compressed(dst_frame);
+	bool page_compressed_encrypted = fil_page_is_compressed_encrypted(dst_frame);
+	ulint space_id = mach_read_from_4(
+		dst_frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
+	fil_space_crypt_t* crypt_data = fil_space_get_crypt_data(space_id);
+	fil_space_t* space = fil_space_found_by_id(space_id);
+
+	if (key_version != 0 ||
+		(crypt_data && crypt_data->type != CRYPT_SCHEME_UNENCRYPTED) ||
+		page_compressed || page_compressed_encrypted) {
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"Maybe corruption: Block space_id %lu in file %s maybe corrupted\n"
+			"Reason could be that key_version in page %u \n"
+			"or in crypt_data %p could not be found,\n"
+			"key management plugin is not found or\n"
+			"used encryption algorithm or method does not match.\n"
+			"Page compressed %d, compressed and encrypted %d.\n",
+			space_id, space ? space->name : "NULL",
+			key_version,
+			crypt_data,
+			page_compressed, page_compressed_encrypted);
+	}
+}
+
 /********************************************************************//**
 Completes an asynchronous read or write request of a file page to or from
 the buffer pool.
@@ -4505,6 +4545,8 @@ buf_page_io_complete(
 				    && buf_mark_space_corrupt(bpage)) {
 					return(false);
 				} else {
+					buf_page_check_corrupt(bpage);
+
 					fputs("InnoDB: Ending processing"
 					      " because of"
 					      " a corrupt database page.\n",

storage/xtradb/include/log0crypt.h

+/*****************************************************************************
+
+Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
+Copyright (C) 2014, 2015, 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
+Foundation; version 2 of the License.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+
+*****************************************************************************/
 /**************************************************//**
 @file include/log0crypt.h
 Innodb log encrypt/decrypt
 
 Created 11/25/2013 Minli Zhu
+Modified           Jan Lindström jan.lindstrom@mariadb.com
 *******************************************************/
 #ifndef log0crypt_h
 #define log0crypt_h
@@ -22,7 +41,7 @@ UNIV_INTERN
 void
 log_crypt_set_ver_and_key(
 /*======================*/
-	ib_uint64_t next_checkpoint_no);
+	ib_uint64_t next_checkpoint_no);/*!< in: next checkpoint no */
 
 
 /*********************************************************************//**
@@ -43,17 +62,17 @@ UNIV_INTERN
 bool
 log_crypt_read_checkpoint_buf(
 /*===========================*/
-	const byte*	buf);			/*!< in: checkpoint buffer */
+	const byte*	buf);		/*!< in: checkpoint buffer */
 
 /********************************************************
 Encrypt one or more log block before it is flushed to disk */
 UNIV_INTERN
 void
 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 */
-	const ulint size);		/*!< in: size of log blocks */
+/*=====================*/
+	ib_uint64_t	next_checkpoint_no,	/*!< in: log group to be flushed */
+	byte* 		block,			/*!< in/out: pointer to a log block */
+	const ulint 	size);			/*!< in: size of log blocks */
 
 /********************************************************
 Decrypt a specified log segment after they are read from a log file to a buffer.
@@ -61,8 +80,41 @@ Decrypt a specified log segment after they are read from a log file to a buffer.
 UNIV_INTERN
 void
 log_decrypt_after_read(
-/*==========================*/
-	byte* frame,	/*!< in/out: log segment */
-	const ulint size);	/*!< in: log segment size */
+/*===================*/
+	byte*		frame,	/*!< in/out: log segment */
+	const ulint	size);	/*!< in: log segment size */
+
+/* Error codes for crypt info */
+typedef enum {
+	LOG_UNENCRYPTED = 0,
+	LOG_CRYPT_KEY_NOT_FOUND = 1,
+	LOG_DECRYPT_MAYBE_FAILED = 2
+} log_crypt_err_t;
+
+/********************************************************
+Check is the checkpoint information encrypted. This check
+is based on fact has log group crypt info and based
+on this crypt info was the key version different from
+unencrypted key version. There is no realible way to
+distinguish encrypted log block from corrupted log block,
+but if log block corruption is found this function is
+used to find out if log block is maybe encrypted but
+encryption key, key management plugin or encryption
+algorithm does not match.
+@return TRUE, if log block may be encrypted */
+UNIV_INTERN
+ibool
+log_crypt_block_maybe_encrypted(
+/*============================*/
+	const byte* 		log_block, 	/*!< in: log block */
+	log_crypt_err_t* 	err_info); 	/*!< out: error info */
+
+/********************************************************
+Print crypt error message to error log */
+UNIV_INTERN
+void
+log_crypt_print_error(
+/*==================*/
+	log_crypt_err_t 	err_info); 	/*!< out: error info */
 
 #endif  // log0crypt.h

storage/xtradb/log/log0crypt.cc

@@ -86,6 +86,9 @@ log_block_get_start_lsn(
 	return start_lsn;
 }
 
+/*********************************************************************//**
+Get crypt info from checkpoint.
+@return a crypt info or NULL if not present. */
 static
 const crypt_info_t*
 get_crypt_info(
@@ -107,6 +110,9 @@ get_crypt_info(
 	return NULL;
 }
 
+/*********************************************************************//**
+Get crypt info from log block
+@return a crypt info or NULL if not present. */
 static
 const crypt_info_t*
 get_crypt_info(
@@ -239,14 +245,23 @@ init_crypt_key(
 	return true;
 }
 
-static bool mysort(const crypt_info_t& i,
-		   const crypt_info_t& j)
+/*********************************************************************//**
+Compare function for checkpoint numbers
+@return true if first checkpoint is larger than second one */
+static
+bool
+mysort(const crypt_info_t& i,
+       const crypt_info_t& j)
 {
 	return i.checkpoint_no > j.checkpoint_no;
 }
 
+/*********************************************************************//**
+Add crypt info to set if it is not already present
+@return true if successfull, false if not- */
 static
-bool add_crypt_info(crypt_info_t* info)
+bool
+add_crypt_info(crypt_info_t* info)
 {
 	/* so that no one is searching array while we modify it */
 	ut_ad(mutex_own(&(log_sys->mutex)));
@@ -332,7 +347,7 @@ Encrypt one or more log block before it is flushed to disk */
 UNIV_INTERN
 void
 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 */
 	const ulint size)		/*!< in: size of log blocks */
@@ -369,7 +384,7 @@ Decrypt a specified log segment after they are read from a log file to a buffer.
 */
 void
 log_decrypt_after_read(
-/*==========================*/
+/*===================*/
 	byte* frame,	        /*!< in/out: log segment */
 	const ulint size)	/*!< in: log segment size */
 {
@@ -499,3 +514,74 @@ log_crypt_read_checkpoint_buf(
 	return true;
 }
 
+/********************************************************
+Check is the checkpoint information encrypted. This check
+is based on fact has log group crypt info and based
+on this crypt info was the key version different from
+unencrypted key version. There is no realible way to
+distinguish encrypted log block from corrupted log block,
+but if log block corruption is found this function is
+used to find out if log block is maybe encrypted but
+encryption key, key management plugin or encryption
+algorithm does not match.
+@return TRUE, if log block may be encrypted */
+UNIV_INTERN
+ibool
+log_crypt_block_maybe_encrypted(
+/*============================*/
+	const byte* 		log_block,	/*!< in: log block */
+	log_crypt_err_t* 	err_info)	/*!< out: error info */
+{
+	ibool maybe_encrypted = FALSE;
+	const crypt_info_t* crypt_info;
+
+	*err_info = LOG_UNENCRYPTED;
+	crypt_info = get_crypt_info(log_block);
+
+	if (crypt_info &&
+	    crypt_info->key_version != UNENCRYPTED_KEY_VER) {
+		byte mysqld_key[MY_AES_BLOCK_SIZE] = {0};
+		uint keylen= sizeof(mysqld_key);
+
+		/* Log block contains crypt info and based on key
+		version block could be encrypted. */
+		*err_info = LOG_DECRYPT_MAYBE_FAILED;
+		maybe_encrypted = TRUE;
+
+		if (encryption_key_get(LOG_DEFAULT_ENCRYPTION_KEY,
+				       crypt_info->key_version, mysqld_key, &keylen)) {
+			*err_info = LOG_CRYPT_KEY_NOT_FOUND;
+		}
+	}
+
+	return (maybe_encrypted);
+}
+
+/********************************************************
+Print crypt error message to error log */
+UNIV_INTERN
+void
+log_crypt_print_error(
+/*==================*/
+	log_crypt_err_t 	err_info) 	/*!< out: error info */
+{
+	switch(err_info) {
+	case LOG_CRYPT_KEY_NOT_FOUND:
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"Redo log crypto: getting mysqld crypto key "
+			"from key version failed. Reason could be that "
+			"requested key version is not found or required "
+			"encryption key management plugin is not found.");
+		break;
+	case LOG_DECRYPT_MAYBE_FAILED:
+		ib_logf(IB_LOG_LEVEL_ERROR,
+			"Redo log crypto: failed to decrypt log block. "
+			"Reason could be that requested key version is "
+			"not found, required encryption key management "
+			"plugin is not found or configured encryption "
+			"algorithm and/or method does not match.");
+		break;
+	default:
+		ut_error; /* Real bug */
+	}
+}

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, SkySQL Ab. All Rights Reserved.
+Copyright (c) 2013, 2015, 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 @@ Created 9/20/1997 Heikki Tuuri
 #include "log0recv.ic"
 #endif
 
+#include "log0crypt.h"
+
 #include "config.h"
 #ifdef HAVE_ALLOCA_H
 #include "alloca.h"
@@ -2774,8 +2776,9 @@ recv_scan_log_recs(
 	lsn_t*		contiguous_lsn,	/*!< in/out: it is known that all log
 					groups contain contiguous log data up
 					to this lsn */
-	lsn_t*		group_scanned_lsn)/*!< out: scanning succeeded up to
+	lsn_t*		group_scanned_lsn,/*!< out: scanning succeeded up to
 					this lsn */
+	dberr_t*	err)		/*!< out: error code or DB_SUCCESS */
 {
 	const byte*	log_block;
 	ulint		no;
@@ -2794,6 +2797,7 @@ recv_scan_log_recs(
 	log_block = buf;
 	scanned_lsn = start_lsn;
 	more_data = FALSE;
+	*err = DB_SUCCESS;
 
 	do {
 		no = log_block_get_hdr_no(log_block);
@@ -2805,6 +2809,7 @@ recv_scan_log_recs(
 		*/
 		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;
 
 			if (no == log_block_convert_lsn_to_no(scanned_lsn)
 			    && !log_block_checksum_is_ok_or_old_format(
@@ -2826,6 +2831,14 @@ recv_scan_log_recs(
 
 			finished = TRUE;
 
+			if (log_crypt_block_maybe_encrypted(log_block,
+					&log_crypt_err)) {
+				/* Log block maybe encrypted */
+				log_crypt_print_error(log_crypt_err);
+				*err = DB_ERROR;
+				return (TRUE);
+			}
+
 			/* Crash if we encounter a garbage log block */
 			if (!srv_force_recovery) {
 				fputs("InnoDB: Set innodb_force_recovery"
@@ -3014,14 +3027,16 @@ recv_group_scan_log_recs(
 	lsn_t*		contiguous_lsn,	/*!< in/out: it is known that all log
 					groups contain contiguous log data up
 					to this lsn */
-	lsn_t*		group_scanned_lsn)/*!< out: scanning succeeded up to
+	lsn_t*		group_scanned_lsn,/*!< out: scanning succeeded up to
 					this lsn */
+	dberr_t*	err)		/*!< out: error code or DB_SUCCESS */
 {
 	ibool	finished;
 	lsn_t	start_lsn;
 	lsn_t	end_lsn;
 
 	finished = FALSE;
+	*err = DB_SUCCESS;
 
 	start_lsn = *contiguous_lsn;
 
@@ -3036,7 +3051,13 @@ recv_group_scan_log_recs(
 			- (recv_n_pool_free_frames * srv_buf_pool_instances))
 			* UNIV_PAGE_SIZE,
 			TRUE, log_sys->buf, RECV_SCAN_SIZE,
-			start_lsn, contiguous_lsn, group_scanned_lsn);
+			start_lsn, contiguous_lsn, group_scanned_lsn,
+			err);
+
+		if (*err != DB_SUCCESS) {
+			break;
+		}
+
 		start_lsn = end_lsn;
 	}
 
@@ -3267,6 +3288,7 @@ recv_recovery_from_checkpoint_start_func(
 		up_to_date_group = max_cp_group;
 	} else {
 		ulint	capacity;
+		dberr_t err;
 
 		/* Try to recover the remaining part from logs: first from
 		the logs of the archived group */
@@ -3286,8 +3308,9 @@ recv_recovery_from_checkpoint_start_func(
 		}
 
 		recv_group_scan_log_recs(group, &contiguous_lsn,
-					 &group_scanned_lsn);
-		if (recv_sys->scanned_lsn < checkpoint_lsn) {
+			&group_scanned_lsn, &err);
+
+		if (err != DB_SUCCESS || recv_sys->scanned_lsn < checkpoint_lsn) {
 
 			mutex_exit(&(log_sys->mutex));
 
@@ -3319,9 +3342,15 @@ recv_recovery_from_checkpoint_start_func(
 #ifdef UNIV_LOG_ARCHIVE
 		lsn_t	old_scanned_lsn	= recv_sys->scanned_lsn;
 #endif /* UNIV_LOG_ARCHIVE */
+		dberr_t err;
 
 		recv_group_scan_log_recs(group, &contiguous_lsn,
-					 &group_scanned_lsn);
+			&group_scanned_lsn, &err);
+
+		if (err != DB_SUCCESS) {
+			return (err);
+		}
+
 		group->scanned_lsn = group_scanned_lsn;
 
 #ifdef UNIV_LOG_ARCHIVE
@@ -3801,6 +3830,7 @@ log_group_recover_from_archive_file(
 	os_offset_t	file_size;
 	int		input_char;
 	char		name[OS_FILE_MAX_PATH];
+	dberr_t		err;
 
 	ut_a(0);
 
@@ -3946,7 +3976,11 @@ log_group_recover_from_archive_file(
 			(buf_pool_get_n_pages()
 			- (recv_n_pool_free_frames * srv_buf_pool_instances))
 			* UNIV_PAGE_SIZE, TRUE, buf, len, start_lsn,
-			&dummy_lsn, &scanned_lsn);
+			&dummy_lsn, &scanned_lsn, &err);
+
+		if (err != DB_SUCCESS) {
+			return(FALSE);
+		}
 
 		if (scanned_lsn == file_end_lsn) {