Fixes #2793 and #1493: Connection issues in compressed connections and in mixed compressed/uncompressed connections.

lib/MySQL_Protocol.cpp

@@ -1551,13 +1551,15 @@ bool MySQL_Protocol::process_pkt_handshake_response(unsigned char *pkt, unsigned
 		//(*myds)->switching_auth_stage=2;
 		charset=(*myds)->tmp_charset;
 		proxy_debug(PROXY_DEBUG_MYSQL_PROTOCOL,2,"Session=%p , DS=%p . Encrypted: %d , switching_auth: %d, auth_plugin_id: %d\n", (*myds)->sess, (*myds), (*myds)->encrypted, (*myds)->switching_auth_stage, auth_plugin_id);
+		capabilities = (*myds)->myconn->options.client_flag;
 		goto __do_auth;
 	}
 
 	capabilities     = CPY4(pkt);
 	(*myds)->myconn->options.client_flag = capabilities;
 	pkt     += sizeof(uint32_t);
 	max_pkt  = CPY4(pkt);
+	(*myds)->myconn->options.max_allowed_pkt = max_pkt;
 	pkt     += sizeof(uint32_t);
 	charset  = *(uint8_t *)pkt;
 	if ( (*myds)->encrypted == false ) { // client wants to use SSL
@@ -2002,7 +2004,6 @@ bool MySQL_Protocol::process_pkt_handshake_response(unsigned char *pkt, unsigned
 
 	if (ret==true) {
 
-		(*myds)->myconn->options.max_allowed_pkt=max_pkt;
 		(*myds)->DSS=STATE_CLIENT_HANDSHAKE;
 
 		if (!userinfo->username) // if set already, ignore

lib/mysql_data_stream.cpp

@@ -880,6 +880,7 @@ int MySQL_Data_Stream::buffer2array() {
 			proxy_debug(PROXY_DEBUG_PKT_ARRAY, 5, "Session=%p . Reading the header of a new compressed packet\n", sess);
  			memcpy(&queueIN.hdr,queue_r_ptr(queueIN), sizeof(mysql_hdr));
 			queue_r(queueIN,sizeof(mysql_hdr));
+			pkt_sid=queueIN.hdr.pkt_id;
 			queueIN.pkt.size=queueIN.hdr.pkt_length+sizeof(mysql_hdr)+3;
 			queueIN.pkt.ptr=l_alloc(queueIN.pkt.size);
 			memcpy(queueIN.pkt.ptr, &queueIN.hdr, sizeof(mysql_hdr)); // immediately copy the header into the packet

test/tap/tap/utils.cpp

@@ -1,4 +1,6 @@
 #include <mysql.h>
+#include <unistd.h>
+#include <sys/wait.h>
 
 #include "tap.h"
 #include "utils.h"
@@ -102,3 +104,96 @@ int get_server_version(MYSQL *mysql, std::string& version) {
 
 	return 0;
 }
+
+// Pipes definition
+constexpr uint8_t NUM_PIPES = 3;
+constexpr uint8_t PARENT_WRITE_PIPE = 0;
+constexpr uint8_t PARENT_READ_PIPE  = 1;
+constexpr uint8_t PARENT_ERR_PIPE   = 2;
+int pipes[NUM_PIPES][2];
+// Pipe selection
+constexpr uint8_t READ_FD  = 0;
+constexpr uint8_t WRITE_FD = 1;
+// Parent pipes
+const auto& PARENT_READ_FD  = pipes[PARENT_READ_PIPE][READ_FD];
+const auto& PARENT_READ_ERR = pipes[PARENT_ERR_PIPE][READ_FD];
+const auto& PARENT_WRITE_FD = pipes[PARENT_WRITE_PIPE][WRITE_FD];
+// Child pipes
+const auto& CHILD_READ_FD   = pipes[PARENT_WRITE_PIPE][READ_FD];
+const auto& CHILD_WRITE_FD  = pipes[PARENT_READ_PIPE][WRITE_FD];
+const auto& CHILD_WRITE_ERR = pipes[PARENT_ERR_PIPE][WRITE_FD];
+
+int execvp(const std::string& cmd, const std::vector<const char*>& argv, std::string& result) {
+	int err = 0;
+	std::string result_ = "";
+	std::vector<const char*> _argv = argv;
+
+	// Append null to end of _argv for extra safety
+	_argv.push_back(nullptr);
+
+	int outfd[2];
+	int infd[2];
+
+	// Pipes for parent to write and read
+	pipe(pipes[PARENT_READ_PIPE]);
+	pipe(pipes[PARENT_WRITE_PIPE]);
+	pipe(pipes[PARENT_ERR_PIPE]);
+
+	pid_t child_pid = fork();
+	if(child_pid == 0) {
+		// Copy the pipe descriptors
+		dup2(CHILD_READ_FD, STDIN_FILENO);
+		dup2(CHILD_WRITE_FD, STDOUT_FILENO);
+		dup2(CHILD_WRITE_ERR, STDERR_FILENO);
+
+		// Close no longer needed pipes
+		close(CHILD_READ_FD);
+		close(CHILD_WRITE_FD);
+		close(CHILD_WRITE_ERR);
+
+		close(PARENT_READ_FD);
+		close(PARENT_READ_ERR);
+		close(PARENT_WRITE_FD);
+
+		char** args = const_cast<char**>(_argv.data());
+		err = execvp(cmd.c_str(), args);
+
+		if (err) {
+			exit(errno);
+		} else {
+			exit(0);
+		}
+	} else {
+		char buffer[128];
+		int count;
+
+		// Close no longer needed pipes
+		close(CHILD_READ_FD);
+		close(CHILD_WRITE_FD);
+		close(CHILD_WRITE_ERR);
+
+		if (err == 0) {
+			// Read from child’s stdout
+			count = read(PARENT_READ_FD, buffer, sizeof(buffer));
+			while (count > 0) {
+				buffer[count] = 0;
+				result_ += buffer;
+				count = read(PARENT_READ_FD, buffer, sizeof(buffer));
+			}
+		} else {
+			// Read from child’s stderr
+			count = read(PARENT_READ_ERR, buffer, sizeof(buffer));
+			while (count > 0) {
+				buffer[count] = 0;
+				result_ += buffer;
+				count = read(PARENT_READ_ERR, buffer, sizeof(buffer));
+			}
+		}
+
+		waitpid(child_pid, &err, 0);
+	}
+
+	result = result_;
+
+	return err;
+}

test/tap/tap/utils.h

@@ -3,6 +3,7 @@
 
 #include <mysql.h>
 #include <string>
+#include <vector>
 
 #define MYSQL_QUERY(mysql, query) \
 	do { \
@@ -27,5 +28,15 @@ int select_config_file(MYSQL* mysql, std::string& resultset);
 }
 #endif
 
+/**
+ * @brief Execute the given comand, and stores it's output.
+ *
+ * @param file File to be executed.
+ * @param argv Arguments to be given to the executable.
+ * @param result The output of the file execution. If the execution succeed it contains `stdout` output,
+ *  in case of failure `stderr` contents are returned.
+ * @return int Zero in case of success, or the errno returned by `execvp` in case of failure.
+ */
+int execvp(const std::string& file, const std::vector<const char*>& argv, std::string& result);
 
 #endif // #define UTILS_H

test/tap/tests/Makefile

@@ -103,3 +103,9 @@ aurora: aurora.cpp $(TAP_LIBDIR)/libtap.a
 
 test_tokenizer-t: test_tokenizer-t.cpp $(TAP_LIBDIR)/libtap.a
 	g++ test_tokenizer-t.cpp $(INCLUDEDIRS) $(LDIRS) $(OPT) -std=c++11 $(MYLIBS) -lproxysql -ltap -Wl,--no-as-needed -ldl -lpthread -o test_tokenizer-t -DGITVERSION=\"$(GIT_VERSION)\"
+
+1493_mixed_compression: reg_test_1493-mixed_compression-t.cpp
+	g++ -DDEBUG test_mixed_compression-t.cpp $(INCLUDEDIRS) $(LDIRS) $(OPT) -std=c++11 $(OBJ) $(MYLIBS) -ltap -ldl $(STATIC_LIBS) -o reg_test_1493-mixed_compression-t -DGITVERSION=\"$(GIT_VERSION)\"
+
+2793_compression: reg_test_2793-compression-t.cpp
+	g++ -DDEBUG reg_test_2793-compression-t.cpp $(INCLUDEDIRS) $(LDIRS) $(OPT) -std=c++11 $(OBJ) $(MYLIBS) -ltap -ldl $(STATIC_LIBS) -o reg_test_2793-compression-t -DGITVERSION=\"$(GIT_VERSION)\"

test/tap/tests/reg_test_1493-mixed_compression-t.cpp

@@ -0,0 +1,86 @@
+/**
+ * @file reg_test_1493-mixed_compression-t.cpp
+ * @brief This test is a regression test for issue #1493.
+ * @date 2020-05-14
+ */
+
+#include <vector>
+#include <string>
+#include <stdio.h>
+
+#include <mysql.h>
+#include <mysql/mysqld_error.h>
+
+#include "tap.h"
+#include "command_line.h"
+#include "utils.h"
+
+using std::string;
+
+int main(int argc, char** argv) {
+	CommandLine cl;
+
+	if (cl.getEnv()) {
+		diag("Failed to get the required environmental variables.");
+		return -1;
+	}
+
+	plan(2);
+
+	MYSQL* proxysql_admin = mysql_init(NULL);
+
+	// Initialize connections
+	if (!proxysql_admin) {
+		fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(proxysql_admin));
+		return -1;
+	}
+
+	// Connnect to local proxysql
+	if (!mysql_real_connect(proxysql_admin, cl.host, cl.admin_username, cl.admin_password, NULL, cl.admin_port, NULL, 0)) {
+		fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(proxysql_admin));
+		return -1;
+	}
+
+	const char* disable_select_query_rules =
+		"UPDATE mysql_query_rules SET active=0 WHERE match_digest='^SELECT'";
+	const char* enable_select_query_rules =
+		"UPDATE mysql_query_rules SET active=1 WHERE match_digest='^SELECT'";
+	const char* update_mysql_query_rules =
+		"INSERT INTO mysql_query_rules (active, username, match_digest, destination_hostgroup, apply, cache_ttl, comment) "
+		"VALUES (1,'root','^SELECT.*', 1, 1, 1000000, 'test_mixed_compression_rule')";
+	const char* delete_mysql_query_rule =
+		"DELETE FROM mysql_query_rules WHERE "
+		"comment='test_mixed_compression_rule'";
+	const char* load_mysql_queries_runtime =
+		"LOAD MYSQL QUERY RULES TO RUNTIME";
+
+	// Setup config - query_rules
+	MYSQL_QUERY(proxysql_admin, disable_select_query_rules);
+	MYSQL_QUERY(proxysql_admin, update_mysql_query_rules);
+	MYSQL_QUERY(proxysql_admin, load_mysql_queries_runtime);
+
+	// Mixed compressed / uncompressed queries test #1493
+	const char* mysql_client = "mysql";
+	std::string tg_port = std::string("-P") + std::to_string(cl.port);
+	std::string name = std::string("-u") + cl.username;
+	std::string pass = std::string("-p") + cl.password;
+
+	std::vector<const char*> n_auth_cargs = { "mysql", name.c_str(), pass.c_str(), "-h", cl.host, tg_port.c_str(), "-C", "-e", "select 1", "--default-auth=mysql_native_password" };
+	std::vector<const char*> n_auth_args = { "mysql", name.c_str(), pass.c_str(), "-h", cl.host, tg_port.c_str(), "-e", "select 1", "--default-auth=mysql_native_password" };
+
+	// Query the mysql server in a compressed connection
+	std::string result = "";
+	int query_res = execvp(mysql_client, n_auth_cargs, result);
+	ok(query_res == 0 && result != "", "Native auth compressed query should be executed correctly.");
+
+	// Now query again using a uncompressed connection
+	query_res = execvp(mysql_client, n_auth_args, result);
+	ok(query_res == 0 && result != "", "Native auth uncompressed query should be executed correctly.");
+
+	// Teardown config
+	MYSQL_QUERY(proxysql_admin, delete_mysql_query_rule);
+	MYSQL_QUERY(proxysql_admin, enable_select_query_rules);
+	MYSQL_QUERY(proxysql_admin, load_mysql_queries_runtime);
+
+	return exit_status();
+}

test/tap/tests/reg_test_2793-compression-t.cpp

@@ -0,0 +1,38 @@
+/**
+ * @file reg_test_2793-compression-t.cpp
+ * @brief This test is a regression test for issue #2793.
+ * @date 2020-05-14
+ */
+
+#include <vector>
+#include <string>
+#include <stdio.h>
+#include <mysql.h>
+#include <mysql/mysqld_error.h>
+
+#include "tap.h"
+#include "command_line.h"
+#include "utils.h"
+
+using std::string;
+
+int main(int argc, char** argv) {
+	CommandLine cl;
+
+	if (cl.getEnv()) {
+		diag("Failed to get the required environmental variables.");
+		return -1;
+	}
+
+	plan(1);
+
+	const char* mysql_client = "mysql";
+	std::vector<const char*> cargs = { "mysql", "-uroot", "-proot", "-h", "127.0.0.1", "-P6033", "-C", "-e", "select 1" };
+
+	// Query the mysql server in a compressed connection
+	std::string result = "";
+	int query_res = execvp(mysql_client, cargs, result);
+	ok(query_res == 0 && result != "", "Compressed query should be executed correctly.");
+
+	return exit_status();
+}