MDEV-8817: Failing assertion: new_state->key_version != ENCRYPTION_KEY_VERSION_INVALID

Analysis: Problem sees to be the fact that we allow creating or altering
table to use encryption_key_id that does not exists in case where
original table is not encrypted currently. Secondly we should not
do key rotation to tables that are not encrypted or tablespaces
that can't be found from tablespace cache.

Fix: Do not allow creating unencrypted table with nondefault encryption key
and do not rotate tablespaces that are not encrypted (FIL_SPACE_ENCRYPTION_OFF)
or can't be found from tablespace cache.

7 files changed, 176 insertions, 22 deletions

diff --git a/mysql-test/suite/encryption/r/innodb-encryption-alter.result b/mysql-test/suite/encryption/r/innodb-encryption-alter.result
new file mode 100644
index 00000000000..62880015e29
--- /dev/null
+++ b/mysql-test/suite/encryption/r/innodb-encryption-alter.result
@@ -0,0 +1,13 @@
+SET GLOBAL innodb_file_format = `Barracuda`;
+SET GLOBAL innodb_file_per_table = ON;
+SET GLOBAL innodb_encrypt_tables = ON;
+SET GLOBAL innodb_encryption_threads = 4;
+CREATE TABLE t1 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=NO ENCRYPTION_KEY_ID=4;
+ERROR HY000: Can't create table `test`.`t1` (errno: 140 "Wrong create options")
+CREATE TABLE t2 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=NO ENCRYPTION_KEY_ID=1;
+CREATE TABLE t1 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=YES ENCRYPTION_KEY_ID=4;
+ALTER TABLE t2 ENCRYPTION_KEY_ID=4;
+ERROR HY000: Can't create table `test`.`#sql-temporary` (errno: 140 "Wrong create options")
+ALTER TABLE t1 ENCRYPTION_KEY_ID=99;
+ERROR HY000: Can't create table `test`.`#sql-temporary` (errno: 140 "Wrong create options")
+drop table t1,t2;
diff --git a/mysql-test/suite/encryption/t/encrypt_and_grep.test b/mysql-test/suite/encryption/t/encrypt_and_grep.test
index 43816784ebe..2a5fcbcebf8 100644
--- a/mysql-test/suite/encryption/t/encrypt_and_grep.test
+++ b/mysql-test/suite/encryption/t/encrypt_and_grep.test
@@ -27,7 +27,7 @@ insert t3 values (repeat('dummy', 42));
 
 --echo # Wait max 10 min for key encryption threads to encrypt all spaces
 --let $wait_timeout= 600
---let $wait_condition=SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0
+--let $wait_condition=SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0
 --source include/wait_condition.inc
 
 --sleep 5
@@ -55,7 +55,7 @@ SET GLOBAL innodb_encrypt_tables = off;
 
 --echo # Wait max 10 min for key encryption threads to decrypt all spaces
 --let $wait_timeout= 600
---let $wait_condition=SELECT COUNT(*) = 2 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+--let $wait_condition=SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
 --source include/wait_condition.inc
 --sleep 5
 
@@ -82,7 +82,7 @@ SET GLOBAL innodb_encrypt_tables = on;
 
 --echo # Wait max 10 min for key encryption threads to encrypt all spaces
 --let $wait_timeout= 600
---let $wait_condition=SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+--let $wait_condition=SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
 --source include/wait_condition.inc
 --sleep 5
 
@@ -105,4 +105,4 @@ SET GLOBAL innodb_encrypt_tables = on;
 
 --echo # TODO: add shutdown + grep tests
 
-drop table t1, t2, t3;
\ No newline at end of file
+drop table t1, t2, t3;
diff --git a/mysql-test/suite/encryption/t/innodb-encryption-alter.test b/mysql-test/suite/encryption/t/innodb-encryption-alter.test
new file mode 100644
index 00000000000..0eea28f22dd
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb-encryption-alter.test
@@ -0,0 +1,85 @@
+-- source include/have_innodb.inc
+-- source include/have_file_key_management_plugin.inc
+
+#
+# MDEV-8817: Failing assertion: new_state->key_version != ENCRYPTION_KEY_VERSION_INVALID
+#
+
+--disable_query_log
+let $innodb_file_format_orig = `SELECT @@innodb_file_format`;
+let $innodb_file_per_table_orig = `SELECT @@innodb_file_per_table`;
+let $encrypt_tables = `SELECT @@innodb_encrypt_tables`;
+let $threads = `SELECT @@innodb_encryption_threads`;
+--enable_query_log
+
+SET GLOBAL innodb_file_format = `Barracuda`;
+SET GLOBAL innodb_file_per_table = ON;
+SET GLOBAL innodb_encrypt_tables = ON;
+SET GLOBAL innodb_encryption_threads = 4;
+
+--error 1005
+CREATE TABLE t1 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=NO ENCRYPTION_KEY_ID=4;
+CREATE TABLE t2 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=NO ENCRYPTION_KEY_ID=1;
+CREATE TABLE t1 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=YES ENCRYPTION_KEY_ID=4;
+--replace_regex /#sql-[0-9a-f_]*/#sql-temporary/
+--error 1005
+ALTER TABLE t2 ENCRYPTION_KEY_ID=4;
+--replace_regex /#sql-[0-9a-f_]*/#sql-temporary/
+--error 1005
+ALTER TABLE t1 ENCRYPTION_KEY_ID=99;
+
+--disable_warnings
+--disable_query_log
+let $i = 400;
+while ($i)
+{
+INSERT INTO t1 values(NULL, substring(MD5(RAND()), -128));
+dec $i;
+}
+commit;
+INSERT INTO t2 select * from t1;
+
+--disable_abort_on_error
+
+--connect (con1,localhost,root,,test)
+--connect (con2,localhost,root,,test)
+
+let $i = 50;
+while ($i)
+{
+connection con1;
+send ALTER TABLE t1 ENCRYPTED=NO ENCRYPTION_KEY_ID=1;
+connection con2;
+send ALTER TABLE t1 ENCRYPTED=YES ENCRYPTION_KEY_ID=4;
+connection default;
+send ALTER TABLE t2 ENCRYPTED=NO ENCRYPTION_KEY_ID=1;
+connection con1;
+--reap;
+ALTER TABLE t1 ENCRYPTED=NO ENCRYPTION_KEY_ID=1;
+connection con2;
+--reap
+ALTER TABLE t1 ENCRYPTED=YES ENCRYPTION_KEY_ID=4;
+connection default;
+--reap
+ALTER TABLE t2 ENCRYPTED=YES ENCRYPTION_KEY_ID=1;
+ALTER TABLE t1 ENCRYPTED=NO ENCRYPTION_KEY_ID=1;
+dec $i;
+}
+
+connection default;
+--disconnect con1
+--disconnect con2
+
+--enable_abort_on_error
+--enable_warnings
+--enable_query_log
+
+drop table t1,t2;
+
+# reset system
+--disable_query_log
+EVAL SET GLOBAL innodb_file_per_table = $innodb_file_per_table_orig;
+EVAL SET GLOBAL innodb_file_format = $innodb_file_format_orig;
+EVAL SET GLOBAL innodb_encrypt_tables = $encrypt_tables;
+EVAL SET GLOBAL innodb_encryption_threads = $threads;
+--enable_query_log
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index bb0ca88d06d..100f304bf6f 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -199,7 +199,6 @@ fil_space_create_crypt_data(
 	if (encrypt_mode == FIL_SPACE_ENCRYPTION_OFF ||
 		(!srv_encrypt_tables && encrypt_mode == FIL_SPACE_ENCRYPTION_DEFAULT)) {
 		crypt_data->type = CRYPT_SCHEME_UNENCRYPTED;
-		crypt_data->min_key_version = 0;
 	} else {
 		crypt_data->type = CRYPT_SCHEME_1;
 		crypt_data->min_key_version = encryption_key_get_latest_version(key_id);
@@ -210,8 +209,8 @@ fil_space_create_crypt_data(
 	crypt_data->locker = crypt_data_scheme_locker;
 	my_random_bytes(crypt_data->iv, sizeof(crypt_data->iv));
 	crypt_data->encryption = encrypt_mode;
-	crypt_data->key_id = key_id;
 	crypt_data->inited = true;
+	crypt_data->key_id = key_id;
 	return crypt_data;
 }
 
@@ -964,6 +963,13 @@ fil_crypt_get_key_state(
 		new_state->key_version =
 			encryption_key_get_latest_version(new_state->key_id);
 		new_state->rotate_key_age = srv_fil_crypt_rotate_key_age;
+
+		if (new_state->key_version == ENCRYPTION_KEY_VERSION_INVALID) {
+			ib_logf(IB_LOG_LEVEL_ERROR,
+				"Used key_id %u can't be found from key file.",
+				new_state->key_id);
+		}
+
 		ut_a(new_state->key_version != ENCRYPTION_KEY_VERSION_INVALID);
 		ut_a(new_state->key_version != ENCRYPTION_KEY_NOT_ENCRYPTED);
 	} else {
@@ -1302,6 +1308,11 @@ fil_crypt_space_needs_rotation(
 			break;
 		}
 
+		/* No need to rotate space if encryption is disabled */
+		if (crypt_data->encryption == FIL_SPACE_ENCRYPTION_OFF) {
+			break;
+		}
+
 		if (crypt_data->key_id != key_state->key_id) {
 			key_state->key_id= crypt_data->key_id;
 			fil_crypt_get_key_state(key_state);
@@ -1549,13 +1560,16 @@ fil_crypt_find_space_to_rotate(
 	}
 
 	while (!state->should_shutdown() && state->space != ULINT_UNDEFINED) {
-
-		if (fil_crypt_space_needs_rotation(state, key_state, recheck)) {
-			ut_ad(key_state->key_id);
-			/* init state->min_key_version_found before
-			* starting on a space */
-			state->min_key_version_found = key_state->key_version;
-			return true;
+		fil_space_t* space = fil_space_found_by_id(state->space);
+
+		if (space) {
+			if (fil_crypt_space_needs_rotation(state, key_state, recheck)) {
+				ut_ad(key_state->key_id);
+				/* init state->min_key_version_found before
+				* starting on a space */
+				state->min_key_version_found = key_state->key_version;
+				return true;
+			}
 		}
 
 		state->space = fil_get_next_space_safe(state->space);
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 6bd932491d3..dbc1dd2a463 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -11504,6 +11504,20 @@ ha_innobase::check_table_options(
 		}
 	}
 
+	/* Do not allow creating unencrypted table with nondefault
+	encryption key */
+	if ((encrypt == FIL_SPACE_ENCRYPTION_OFF ||
+	    (encrypt == FIL_SPACE_ENCRYPTION_DEFAULT && !srv_encrypt_tables)) &&
+		options->encryption_key_id != FIL_DEFAULT_ENCRYPTION_KEY) {
+		push_warning_printf(
+			thd, Sql_condition::WARN_LEVEL_WARN,
+			HA_WRONG_CREATE_OPTION,
+			"InnoDB: Incorrect ENCRYPTION_KEY_ID %u when encryption is disabled",
+			(uint)options->encryption_key_id
+		);
+		return "ENCRYPTION_KEY_ID";
+	}
+
 	/* Check atomic writes requirements */
 	if (awrites == ATOMIC_WRITES_ON ||
 		(awrites == ATOMIC_WRITES_DEFAULT && srv_use_atomic_writes)) {
diff --git a/storage/xtradb/fil/fil0crypt.cc b/storage/xtradb/fil/fil0crypt.cc
index bb0ca88d06d..100f304bf6f 100644
--- a/storage/xtradb/fil/fil0crypt.cc
+++ b/storage/xtradb/fil/fil0crypt.cc
@@ -199,7 +199,6 @@ fil_space_create_crypt_data(
 	if (encrypt_mode == FIL_SPACE_ENCRYPTION_OFF ||
 		(!srv_encrypt_tables && encrypt_mode == FIL_SPACE_ENCRYPTION_DEFAULT)) {
 		crypt_data->type = CRYPT_SCHEME_UNENCRYPTED;
-		crypt_data->min_key_version = 0;
 	} else {
 		crypt_data->type = CRYPT_SCHEME_1;
 		crypt_data->min_key_version = encryption_key_get_latest_version(key_id);
@@ -210,8 +209,8 @@ fil_space_create_crypt_data(
 	crypt_data->locker = crypt_data_scheme_locker;
 	my_random_bytes(crypt_data->iv, sizeof(crypt_data->iv));
 	crypt_data->encryption = encrypt_mode;
-	crypt_data->key_id = key_id;
 	crypt_data->inited = true;
+	crypt_data->key_id = key_id;
 	return crypt_data;
 }
 
@@ -964,6 +963,13 @@ fil_crypt_get_key_state(
 		new_state->key_version =
 			encryption_key_get_latest_version(new_state->key_id);
 		new_state->rotate_key_age = srv_fil_crypt_rotate_key_age;
+
+		if (new_state->key_version == ENCRYPTION_KEY_VERSION_INVALID) {
+			ib_logf(IB_LOG_LEVEL_ERROR,
+				"Used key_id %u can't be found from key file.",
+				new_state->key_id);
+		}
+
 		ut_a(new_state->key_version != ENCRYPTION_KEY_VERSION_INVALID);
 		ut_a(new_state->key_version != ENCRYPTION_KEY_NOT_ENCRYPTED);
 	} else {
@@ -1302,6 +1308,11 @@ fil_crypt_space_needs_rotation(
 			break;
 		}
 
+		/* No need to rotate space if encryption is disabled */
+		if (crypt_data->encryption == FIL_SPACE_ENCRYPTION_OFF) {
+			break;
+		}
+
 		if (crypt_data->key_id != key_state->key_id) {
 			key_state->key_id= crypt_data->key_id;
 			fil_crypt_get_key_state(key_state);
@@ -1549,13 +1560,16 @@ fil_crypt_find_space_to_rotate(
 	}
 
 	while (!state->should_shutdown() && state->space != ULINT_UNDEFINED) {
-
-		if (fil_crypt_space_needs_rotation(state, key_state, recheck)) {
-			ut_ad(key_state->key_id);
-			/* init state->min_key_version_found before
-			* starting on a space */
-			state->min_key_version_found = key_state->key_version;
-			return true;
+		fil_space_t* space = fil_space_found_by_id(state->space);
+
+		if (space) {
+			if (fil_crypt_space_needs_rotation(state, key_state, recheck)) {
+				ut_ad(key_state->key_id);
+				/* init state->min_key_version_found before
+				* starting on a space */
+				state->min_key_version_found = key_state->key_version;
+				return true;
+			}
 		}
 
 		state->space = fil_get_next_space_safe(state->space);
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index 2f48bf06736..6cdfcbf927a 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -11987,6 +11987,20 @@ ha_innobase::check_table_options(
 		}
 	}
 
+	/* Do not allow creating unencrypted table with nondefault
+	encryption key */
+	if ((encrypt == FIL_SPACE_ENCRYPTION_OFF ||
+	    (encrypt == FIL_SPACE_ENCRYPTION_DEFAULT && !srv_encrypt_tables)) &&
+		options->encryption_key_id != FIL_DEFAULT_ENCRYPTION_KEY) {
+		push_warning_printf(
+			thd, Sql_condition::WARN_LEVEL_WARN,
+			HA_WRONG_CREATE_OPTION,
+			"InnoDB: Incorrect ENCRYPTION_KEY_ID %u when encryption is disabled",
+			(uint)options->encryption_key_id
+		);
+		return "ENCRYPTION_KEY_ID";
+	}
+
 	/* Check atomic writes requirements */
 	if (awrites == ATOMIC_WRITES_ON ||
 		(awrites == ATOMIC_WRITES_DEFAULT && srv_use_atomic_writes)) {