Merge pull request #5661 from sysown/fix/3.0.8-review-items

renecannao · web-flow · commit 5dc586fd6bcf · 2026-04-23T19:49:54.000+07:00
Fix GTID range validation for 3.0.8 review
diff --git a/include/PgSQL_Session.h b/include/PgSQL_Session.h
@@ -394,8 +394,9 @@ class PgSQL_Session : public Base_Session<PgSQL_Session, PgSQL_Data_Stream, PgSQ
 	// admin var off, result transfer already started, or a preflight failed).
 	bool handler_minus1_PoisonTransaction(PgSQL_Data_Stream* myds);
 	// While tx_poisoned, classify a 'Q' packet and either clear the poison
-	// and synthesize a ROLLBACK response (for ROLLBACK / COMMIT / ABORT /
-	// ROLLBACK TO SAVEPOINT) or reject with ERROR 25P02 (anything else).
+	// and synthesize a ROLLBACK response (for plain whole-transaction
+	// ROLLBACK / COMMIT / ABORT / END) or reject with ERROR 25P02
+	// (anything else, including ROLLBACK TO SAVEPOINT).
 	// Returns true if the packet was handled here. Increments the
 	// pgsql_tx_poisoned_{recovered,rejected_statements}_total counters.
 	bool handler_poisoned_simple_query(PtrSize_t* pkt);
@@ -481,12 +482,12 @@ class PgSQL_Session : public Base_Session<PgSQL_Session, PgSQL_Data_Stream, PgSQ
 	// the client, destroy the backend pool connection, and set this flag
 	// true instead of tearing down the client session. While this is true,
 	// the query intake path short-circuits before query rules:
-	//   ROLLBACK / ROLLBACK TO SAVEPOINT / ABORT -> synthesize
+	//   plain ROLLBACK / ABORT -> synthesize
 	//     CommandComplete('ROLLBACK') + ReadyForQuery('I'), clear flag.
-	//   COMMIT -> same ROLLBACK response + NoticeResponse carrying the
+	//   plain COMMIT / END -> same ROLLBACK response + NoticeResponse carrying the
 	//     "there is no transaction in progress" warning, clear flag.
-	//   anything else (including RELEASE SAVEPOINT) -> reply ERROR 25P02
-	//     + ReadyForQuery('E'), stay poisoned.
+	//   anything else (including ROLLBACK TO SAVEPOINT and RELEASE SAVEPOINT)
+	//     -> reply ERROR 25P02 + ReadyForQuery('E'), stay poisoned.
 	bool tx_poisoned{ false };
 
 #ifdef DEBUG
diff --git a/include/proxysql_gtid.h b/include/proxysql_gtid.h
@@ -19,6 +19,7 @@ class TrxId_Interval {
 		explicit TrxId_Interval(const trxid_t trxid);
 		explicit TrxId_Interval(const char* s);
 		explicit TrxId_Interval(const std::string& s);
+		static bool parse(const char* s, TrxId_Interval* out);
 
 		const bool contains(const TrxId_Interval& other);
 		const bool contains(trxid_t trxid);
@@ -53,4 +54,4 @@ class GTID_Set {
 		const std::string to_string(void);
 };
 
-#endif /* PROXYSQL_GTID */
+#endif /* PROXYSQL_GTID */
diff --git a/lib/GTID_Server_Data.cpp b/lib/GTID_Server_Data.cpp
@@ -330,6 +330,7 @@ bool GTID_Server_Data::read_next_gtid() {
 	char rec_msg[80];
 	if (strncmp(data+pos,(char *)"ST=",3)==0) {
 		// we are reading the bootstrap
+		bool invalid_msg = false;
 		char *bs = (char *)malloc(l+1-3); // length + 1 (null byte) - 3 (header)
 		memcpy(bs, data+pos+3, l-3);
 		bs[l-3] = '\0';
@@ -361,14 +362,31 @@ bool GTID_Server_Data::read_next_gtid() {
 							p++;
 						}
 					}
+					*p = '\0';
 				} else { // we are reading the trxid or trxid range
-					updated = gtid_executed.add((std::string)uuid_server, subtoken) || updated;
+					TrxId_Interval iv(trxid_t(0));
+					if (!TrxId_Interval::parse(subtoken, &iv)) {
+						invalid_msg = true;
+						break;
+					}
+					updated = gtid_executed.add((std::string)uuid_server, iv) || updated;
 				}
 			}
+			if (invalid_msg || j == 0 || j%2 != 0) {
+				invalid_msg = true;
+				break;
+			}
 		}
 		pos += l+1;
 		free(bs);
 
+		if (invalid_msg) {
+			proxy_warning("GTID: invalid bootstrap message from binlog reader on port %d for server %s:%d, disconnecting\n",
+				port, address, mysql_port);
+			active = false;
+			return false;
+		}
+
 		if (updated) {
 			events_read++;
 		}
@@ -406,19 +424,33 @@ bool GTID_Server_Data::read_next_gtid() {
 					ul = a-rec_msg-3;
 					strncpy(uuid_server,rec_msg+3,ul);
 					uuid_server[ul] = 0;
-					gtid_executed.add((std::string)uuid_server, a+1);
+					{
+						TrxId_Interval iv(trxid_t(0));
+						if (!TrxId_Interval::parse(a+1, &iv)) {
+							invalid_msg = true;
+							break;
+						}
+						gtid_executed.add((std::string)uuid_server, iv);
+					}
 					events_read++;
 					break;
 				case '4': // trxid range, reuse last UUID
-					gtid_executed.add((std::string)uuid_server, rec_msg+3);
+					{
+						TrxId_Interval iv(trxid_t(0));
+						if (!TrxId_Interval::parse(rec_msg+3, &iv)) {
+							invalid_msg = true;
+							break;
+						}
+						gtid_executed.add((std::string)uuid_server, iv);
+					}
 					events_read++;
 					break;
 				default:
 					invalid_msg = true;
 			}
 
 			if (invalid_msg) {
-				proxy_warning("GTID: unsupported message (%s) from binlog reader on port %d for server %s:%d, disconnecting\n",
+				proxy_warning("GTID: invalid or unsupported message (%s) from binlog reader on port %d for server %s:%d, disconnecting\n",
 					rec_msg, port, address, mysql_port);
 				active = false;
 				return false;
@@ -450,4 +482,3 @@ void * GTID_syncer_run() {
 	//sleep(1000);
 	return NULL;
 }
-
diff --git a/lib/PgSQL_Session.cpp b/lib/PgSQL_Session.cpp
@@ -458,11 +458,11 @@ PgSQL_Session::~PgSQL_Session() {
 // message.
 //
 // Response shape:
-//   * Recovery (ROLLBACK / ROLLBACK TO SAVEPOINT / ABORT / COMMIT):
+//   * Recovery (plain ROLLBACK / ABORT / COMMIT / END):
 //       CommandComplete('ROLLBACK') + ReadyForQuery('I'). For COMMIT also a
 //       preceding NoticeResponse with "there is no transaction in progress"
 //       — matches Postgres native behavior for COMMIT inside an aborted tx.
-//   * Rejection (anything else, incl. RELEASE SAVEPOINT):
+//   * Rejection (anything else, incl. ROLLBACK TO SAVEPOINT / RELEASE SAVEPOINT):
 //       ErrorResponse(25P02) + ReadyForQuery('E'), tx_poisoned stays true.
 bool PgSQL_Session::handler_poisoned_simple_query(PtrSize_t* pkt) {
 	if (pkt->size <= 5) {
diff --git a/lib/proxysql_gtid.cpp b/lib/proxysql_gtid.cpp
@@ -1,3 +1,5 @@
+#include <cerrno>
+#include <cctype>
 #include <cstdio>
 #include <cstdlib>
 #include <string>
@@ -18,23 +20,61 @@ TrxId_Interval::TrxId_Interval(const trxid_t _start, const trxid_t _end) {
 TrxId_Interval::TrxId_Interval(const trxid_t trxid) : TrxId_Interval(trxid, trxid) {
 }
 
+static bool parse_trxid_component(const char*& p, trxid_t& out) {
+	if (p == nullptr || !std::isdigit(static_cast<unsigned char>(*p))) {
+		return false;
+	}
+
+	errno = 0;
+	char* end = nullptr;
+	long long parsed = strtoll(p, &end, 10);
+	if (end == p || errno == ERANGE || parsed < 0) {
+		return false;
+	}
+
+	out = static_cast<trxid_t>(parsed);
+	p = end;
+	return true;
+}
+
+bool TrxId_Interval::parse(const char* s, TrxId_Interval* out) {
+	if (s == nullptr || out == nullptr) {
+		return false;
+	}
+
+	const char* p = s;
+	trxid_t _start = 0;
+	trxid_t _end = 0;
+
+	if (!parse_trxid_component(p, _start)) {
+		return false;
+	}
+
+	_end = _start;
+	if (*p == '-') {
+		p++;
+		if (!parse_trxid_component(p, _end)) {
+			return false;
+		}
+	}
+
+	if (*p != '\0') {
+		return false;
+	}
+
+	*out = TrxId_Interval(_start, _end);
+	return true;
+}
+
 // Initializes a trxid interval from a C string buffer, in [trxid]{-[trxid]} format.
 TrxId_Interval::TrxId_Interval(const char *s) {
 	start = 0;
 	end = 0;
 
-	if (s == nullptr) {
-		return;
-	}
-
-	trxid_t _start = 0, _end = 0;
-
-	if (sscanf(s, "%ld-%ld", &_start, &_end) == 2) {
-		start = _start;
-		end = _end;
-	} else if (sscanf(s, "%ld", &_start) == 1) {
-		start = _start;
-		end = _start;
+	TrxId_Interval iv(trxid_t(0));
+	if (parse(s, &iv)) {
+		start = iv.start;
+		end = iv.end;
 	}
 
 	if (start > end) {
@@ -208,7 +248,11 @@ bool GTID_Set::add(const std::string& uuid, const trxid_t& start, const trxid_t&
 
 // Adds a new trxid range for a given UUID, as a C string buffer. Returns true if the set was modified, false otherwise.
 bool GTID_Set::add(const std::string& uuid, const char *s) {
-	return add(uuid, TrxId_Interval(s));
+	TrxId_Interval iv(trxid_t(0));
+	if (!TrxId_Interval::parse(s, &iv)) {
+		return false;
+	}
+	return add(uuid, iv);
 }
 
 // Adds a new trxid range for a given UUID, as a string. Returns true if the set was modified, false otherwise.
@@ -252,4 +296,4 @@ const std::string GTID_Set::to_string(void) {
 	}
 
 	return out.str();
-}
+}
diff --git a/test/tap/tests/unit/gtid_server_data_unit-t.cpp b/test/tap/tests/unit/gtid_server_data_unit-t.cpp
@@ -193,6 +193,56 @@ static void test_unknown_message_disconnects() {
 	ok(sd.events_read == 1, "unknown: events_read NOT incremented");
 }
 
+/**
+ * @brief Malformed bootstrap trxid ranges disconnect without counting an event.
+ */
+static void test_malformed_bootstrap_disconnects() {
+	GTID_Server_Data sd(nullptr, (char *)"127.0.0.1", 0, 3306);
+
+	std::string msg = std::string("ST=") + UUID_A + ":abc\n";
+	stuff_buffer(sd, msg);
+
+	ok(sd.read_next_gtid() == false, "malformed ST: returns false");
+	ok(sd.active == false, "malformed ST: active set to false");
+	ok(sd.events_read == 0, "malformed ST: events_read NOT incremented");
+	ok(sd.gtid_exists((char *)UUID_A_STRIPPED, 0) == false, "malformed ST: trxid 0 was not added");
+}
+
+/**
+ * @brief Malformed I3 trxid ranges disconnect without counting an event.
+ */
+static void test_malformed_i3_disconnects() {
+	GTID_Server_Data sd(nullptr, (char *)"127.0.0.1", 0, 3306);
+
+	std::string msg = std::string("I3=") + UUID_A_STRIPPED + ":10-abc\n";
+	stuff_buffer(sd, msg);
+
+	ok(sd.read_next_gtid() == false, "malformed I3: returns false");
+	ok(sd.active == false, "malformed I3: active set to false");
+	ok(sd.events_read == 0, "malformed I3: events_read NOT incremented");
+	ok(sd.gtid_exists((char *)UUID_A_STRIPPED, 0) == false, "malformed I3: trxid 0 was not added");
+	ok(sd.gtid_exists((char *)UUID_A_STRIPPED, 10) == false, "malformed I3: trxid 10 was not added");
+}
+
+/**
+ * @brief Malformed I4 trxid ranges disconnect and preserve earlier event count.
+ */
+static void test_malformed_i4_disconnects() {
+	GTID_Server_Data sd(nullptr, (char *)"127.0.0.1", 0, 3306);
+
+	std::string msg1 = std::string("I3=") + UUID_B_STRIPPED + ":10-20\n";
+	stuff_buffer(sd, msg1);
+	sd.read_next_gtid();
+
+	std::string msg2 = "I4=30-40x\n";
+	stuff_buffer(sd, msg2);
+
+	ok(sd.read_next_gtid() == false, "malformed I4: returns false");
+	ok(sd.active == false, "malformed I4: active set to false");
+	ok(sd.events_read == 1, "malformed I4: events_read NOT incremented");
+	ok(sd.gtid_exists((char *)UUID_B_STRIPPED, 30) == false, "malformed I4: trxid 30 was not added");
+}
+
 /**
  * @brief Multiple messages in sequence: ST bootstrap, then I1, I3, I2, I4.
  */
@@ -287,7 +337,7 @@ static void test_incomplete_message() {
 }
 
 int main() {
-	plan(70);
+	plan(83);
 
 	test_bootstrap_single();            //  6 assertions
 	test_bootstrap_range();             //  8 assertions
@@ -297,6 +347,9 @@ int main() {
 	test_i3_range();                    //  8 assertions
 	test_i4_range_reuse_uuid();         //  7 assertions
 	test_unknown_message_disconnects(); //  5 assertions
+	test_malformed_bootstrap_disconnects(); // 4 assertions
+	test_malformed_i3_disconnects();    //  5 assertions
+	test_malformed_i4_disconnects();    //  4 assertions
 	test_mixed_sequence();              // 14 assertions
 	test_read_all_stops_on_unknown();   //  4 assertions
 	test_empty_buffer();                //  3 assertions
diff --git a/test/tap/tests/unit/gtid_set_unit-t.cpp b/test/tap/tests/unit/gtid_set_unit-t.cpp
@@ -77,6 +77,19 @@ static void test_add_string_range() {
 	ok(it->start == 18 && it->end == 30, "add string range: interval is [18,30]");
 }
 
+/**
+ * @brief add() with malformed string ranges must not create a trxid 0 interval.
+ */
+static void test_add_invalid_string_range() {
+	GTID_Set gs;
+
+	ok(!gs.add(UUID_A, "abc"), "add invalid C string range: rejects non-numeric input");
+	ok(!gs.add(UUID_A, "10-abc"), "add invalid C string range: rejects malformed end");
+	ok(!gs.add(UUID_A, "10-20x"), "add invalid C string range: rejects trailing characters");
+	ok(gs.map.empty(), "add invalid C string range: no UUID entry created");
+	ok(!gs.has_gtid(UUID_A, 0), "add invalid C string range: trxid 0 was not added");
+}
+
 /**
  * @brief add() with explicit start, end parameters.
  */
@@ -299,12 +312,13 @@ static void test_copy() {
 }
 
 int main() {
-	plan(62);
+	plan(67);
 
 	test_add_interval();				          // 8 assertions
 	test_add_trxid();                             // 2 assertions
 	test_add_cstring_range();                     // 5 assertions
 	test_add_string_range();                      // 3 assertions
+	test_add_invalid_string_range();              // 5 assertions
 	test_add_start_end();                         // 3 assertions
 	test_add_multi_uuid();                        // 3 assertions
 	test_add_consecutive_trxid();                 // 2 assertions
