MDEV-15566: System tablespace does not easily key rotate to unencrypted

Problem was that key rotation from encrypted to unecrypted was skipped
when encryption is disabled (i.e. set global innodb-encrypt-tables=OFF).

fil_crypt_needs_rotation
        If encryption is disabled (i.e. innodb-encrypt-tables=off)
	and there is tablespaces using default encryption (e.g.
	system tablespace) that are still encrypted state we need
	to rotate them from encrypted state to unencrypted state.

4 files changed, 156 insertions, 36 deletions

diff --git a/mysql-test/suite/encryption/r/innodb-remove-encryption.result b/mysql-test/suite/encryption/r/innodb-remove-encryption.result
new file mode 100644
index 00000000000..06f1c986a25
--- /dev/null
+++ b/mysql-test/suite/encryption/r/innodb-remove-encryption.result
@@ -0,0 +1,41 @@
+set global innodb_file_per_table=OFF;
+call mtr.add_suppression("mysqld: file-key-management-filename is not set");
+call mtr.add_suppression("Plugin 'file_key_management' init function returned error.");
+call mtr.add_suppression("Plugin 'file_key_management' registration as a ENCRYPTION failed.");
+flush tables;
+create table t1(a int not null primary key, b char(200)) engine=innodb;
+
+# Restart server with encryption
+# Wait until encryption threads have encrypted all tablespaces
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+mysql/innodb_table_stats
+mysql/innodb_index_stats
+./ibdata1
+# Success!
+SELECT * from t1;
+a	b
+# Now turn off encryption and wait for threads to decrypt all tablespaces
+SET GLOBAL innodb_encrypt_tables = off;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+mysql/innodb_table_stats
+mysql/innodb_index_stats
+./ibdata1
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+# Success!
+
+# Restart server with no encryption setup, there should be no crashes
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+mysql/innodb_table_stats
+mysql/innodb_index_stats
+./ibdata1
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+SELECT * from t1;
+a	b
+DROP TABLE t1;
diff --git a/mysql-test/suite/encryption/t/innodb-remove-encryption.test b/mysql-test/suite/encryption/t/innodb-remove-encryption.test
new file mode 100644
index 00000000000..24e00a00a02
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb-remove-encryption.test
@@ -0,0 +1,59 @@
+--source include/have_innodb.inc
+# Test uses restart
+--source include/not_embedded.inc
+--source filekeys_plugin.inc
+
+#
+# MDEV-15566: System tablespace does not easily key rotate to unencrypted
+#
+
+set global innodb_file_per_table=OFF;
+
+call mtr.add_suppression("mysqld: file-key-management-filename is not set");
+call mtr.add_suppression("Plugin 'file_key_management' init function returned error.");
+call mtr.add_suppression("Plugin 'file_key_management' registration as a ENCRYPTION failed.");
+flush tables;
+
+create table t1(a int not null primary key, b char(200)) engine=innodb;
+
+--echo
+--echo # Restart server with encryption
+-- let $restart_parameters=--plugin-load-add=$FILE_KEY_MANAGEMENT_SO --loose-file-key-management --loose-file-key-management-filename=$MYSQL_TEST_DIR/std_data/keys.txt --file-key-management-encryption-algorithm=aes_cbc --innodb-encrypt-tables=ON --innodb-encryption-threads=4 --innodb-tablespaces-encryption --innodb-encryption-rotate-key-age=15
+-- source include/restart_mysqld.inc
+
+--echo # Wait until encryption threads have encrypted all tablespaces
+
+--let $tables_count= `select count(*) from information_schema.tables where engine = 'InnoDB'`
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) = $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0 AND ROTATING_OR_FLUSHING = 0;
+--source include/wait_condition.inc
+
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+
+--echo # Success!
+
+SELECT * from t1;
+
+--echo # Now turn off encryption and wait for threads to decrypt all tablespaces
+SET GLOBAL innodb_encrypt_tables = off;
+
+--let $wait_condition=SELECT COUNT(*) = $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0 AND ROTATING_OR_FLUSHING = 0;
+--source include/wait_condition.inc
+
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+
+--echo # Success!
+
+--echo
+--echo # Restart server with no encryption setup, there should be no crashes
+--let $restart_parameters=--skip-file-key-management --innodb-encrypt-tables=OFF --innodb-encryption-threads=0 --innodb-tablespaces-encryption
+-- source include/restart_mysqld.inc
+
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+
+SELECT * from t1;
+DROP TABLE t1;
+
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 3095503cfc5..a9e887ad10c 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -1,6 +1,6 @@
 /*****************************************************************************
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (c) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2014, 2018, 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
@@ -114,17 +114,17 @@ extern my_bool srv_background_scrub_data_compressed;
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
-@param[in]	encrypt_mode		Encryption mode
+@param[in]	crypt_data		Encryption information
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
 @param[in]	rotate_key_age		when to rotate
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
-	fil_encryption_t	encrypt_mode,
-	uint			key_version,
-	uint			latest_key_version,
-	uint			rotate_key_age)
+	const fil_space_crypt_t*	crypt_data,
+	uint				key_version,
+	uint				latest_key_version,
+	uint				rotate_key_age)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /*********************************************************************
@@ -187,7 +187,8 @@ fil_crypt_get_latest_key_version(
 
 	if (crypt_data->is_key_found()) {
 
-		if (fil_crypt_needs_rotation(crypt_data->encryption,
+		if (fil_crypt_needs_rotation(
+				crypt_data,
 				crypt_data->min_key_version,
 				key_version,
 				srv_fil_crypt_rotate_key_age)) {
@@ -963,17 +964,17 @@ fil_crypt_get_key_state(
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
-@param[in]	encrypt_mode		Encryption mode
+@param[in]	crypt_data		Encryption information
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
 @param[in]	rotate_key_age		when to rotate
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
-	fil_encryption_t	encrypt_mode,
-	uint			key_version,
-	uint			latest_key_version,
-	uint			rotate_key_age)
+	const fil_space_crypt_t*	crypt_data,
+	uint				key_version,
+	uint				latest_key_version,
+	uint				rotate_key_age)
 {
 	if (key_version == ENCRYPTION_KEY_VERSION_INVALID) {
 		return false;
@@ -986,13 +987,20 @@ fil_crypt_needs_rotation(
 	}
 
 	if (latest_key_version == 0 && key_version != 0) {
-		if (encrypt_mode == FIL_ENCRYPTION_DEFAULT) {
+		if (crypt_data->encryption == FIL_ENCRYPTION_DEFAULT) {
 			/* this is rotation encrypted => unencrypted */
 			return true;
 		}
 		return false;
 	}
 
+	if (crypt_data->encryption == FIL_ENCRYPTION_DEFAULT
+	    && crypt_data->type == CRYPT_SCHEME_1
+	    && srv_encrypt_tables == 0 ) {
+		/* This is rotation encrypted => unencrypted */
+		return true;
+	}
+
 	/* this is rotation encrypted => encrypted,
 	* only reencrypt if key is sufficiently old */
 	if (key_version + rotate_key_age < latest_key_version) {
@@ -1272,9 +1280,10 @@ fil_crypt_space_needs_rotation(
 		}
 
 		bool need_key_rotation = fil_crypt_needs_rotation(
-			crypt_data->encryption,
+			crypt_data,
 			crypt_data->min_key_version,
-			key_state->key_version, key_state->rotate_key_age);
+			key_state->key_version,
+			key_state->rotate_key_age);
 
 		crypt_data->rotate_state.scrubbing.is_active =
 			btr_scrub_start_space(space->id, &state->scrub_data);
@@ -1862,9 +1871,10 @@ fil_crypt_rotate_page(
 			ut_ad(kv == 0);
 			ut_ad(page_get_space_id(frame) == 0);
 		} else if (fil_crypt_needs_rotation(
-				   crypt_data->encryption,
-				   kv, key_state->key_version,
-				   key_state->rotate_key_age)) {
+				crypt_data,
+				kv,
+				key_state->key_version,
+				key_state->rotate_key_age)) {
 
 			modified = true;
 
diff --git a/storage/xtradb/fil/fil0crypt.cc b/storage/xtradb/fil/fil0crypt.cc
index 3095503cfc5..a9e887ad10c 100644
--- a/storage/xtradb/fil/fil0crypt.cc
+++ b/storage/xtradb/fil/fil0crypt.cc
@@ -1,6 +1,6 @@
 /*****************************************************************************
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (c) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2014, 2018, 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
@@ -114,17 +114,17 @@ extern my_bool srv_background_scrub_data_compressed;
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
-@param[in]	encrypt_mode		Encryption mode
+@param[in]	crypt_data		Encryption information
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
 @param[in]	rotate_key_age		when to rotate
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
-	fil_encryption_t	encrypt_mode,
-	uint			key_version,
-	uint			latest_key_version,
-	uint			rotate_key_age)
+	const fil_space_crypt_t*	crypt_data,
+	uint				key_version,
+	uint				latest_key_version,
+	uint				rotate_key_age)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /*********************************************************************
@@ -187,7 +187,8 @@ fil_crypt_get_latest_key_version(
 
 	if (crypt_data->is_key_found()) {
 
-		if (fil_crypt_needs_rotation(crypt_data->encryption,
+		if (fil_crypt_needs_rotation(
+				crypt_data,
 				crypt_data->min_key_version,
 				key_version,
 				srv_fil_crypt_rotate_key_age)) {
@@ -963,17 +964,17 @@ fil_crypt_get_key_state(
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
-@param[in]	encrypt_mode		Encryption mode
+@param[in]	crypt_data		Encryption information
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
 @param[in]	rotate_key_age		when to rotate
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
-	fil_encryption_t	encrypt_mode,
-	uint			key_version,
-	uint			latest_key_version,
-	uint			rotate_key_age)
+	const fil_space_crypt_t*	crypt_data,
+	uint				key_version,
+	uint				latest_key_version,
+	uint				rotate_key_age)
 {
 	if (key_version == ENCRYPTION_KEY_VERSION_INVALID) {
 		return false;
@@ -986,13 +987,20 @@ fil_crypt_needs_rotation(
 	}
 
 	if (latest_key_version == 0 && key_version != 0) {
-		if (encrypt_mode == FIL_ENCRYPTION_DEFAULT) {
+		if (crypt_data->encryption == FIL_ENCRYPTION_DEFAULT) {
 			/* this is rotation encrypted => unencrypted */
 			return true;
 		}
 		return false;
 	}
 
+	if (crypt_data->encryption == FIL_ENCRYPTION_DEFAULT
+	    && crypt_data->type == CRYPT_SCHEME_1
+	    && srv_encrypt_tables == 0 ) {
+		/* This is rotation encrypted => unencrypted */
+		return true;
+	}
+
 	/* this is rotation encrypted => encrypted,
 	* only reencrypt if key is sufficiently old */
 	if (key_version + rotate_key_age < latest_key_version) {
@@ -1272,9 +1280,10 @@ fil_crypt_space_needs_rotation(
 		}
 
 		bool need_key_rotation = fil_crypt_needs_rotation(
-			crypt_data->encryption,
+			crypt_data,
 			crypt_data->min_key_version,
-			key_state->key_version, key_state->rotate_key_age);
+			key_state->key_version,
+			key_state->rotate_key_age);
 
 		crypt_data->rotate_state.scrubbing.is_active =
 			btr_scrub_start_space(space->id, &state->scrub_data);
@@ -1862,9 +1871,10 @@ fil_crypt_rotate_page(
 			ut_ad(kv == 0);
 			ut_ad(page_get_space_id(frame) == 0);
 		} else if (fil_crypt_needs_rotation(
-				   crypt_data->encryption,
-				   kv, key_state->key_version,
-				   key_state->rotate_key_age)) {
+				crypt_data,
+				kv,
+				key_state->key_version,
+				key_state->rotate_key_age)) {
 
 			modified = true;