From 2b333ebbb0ab2dc7f53b42795d46ac135f28ef35 Mon Sep 17 00:00:00 2001 From: robobun Date: Sun, 12 Jul 2026 16:54:17 +0000 Subject: [PATCH 1/4] sql(mysql): validate response sequence ids to prevent cross-query result delivery process_packets tracked the incoming sequence id but never validated it, and the connection's expected sequence id was never reset when a new command (seq 0) was written. Residual bytes buffered after a completed command's terminator were routed to the next queued query, which resolved with rows the server never produced for it. Validate header.sequence_id against the expected value in the Connected state and fail the connection (ERR_MYSQL_PACKETS_OUT_OF_ORDER) on mismatch, matching libmysql's CR_NET_PACKETS_OUT_OF_ORDER. Reset the expected id to 1 at every ready-for-next-command transition (auth OK, result-set terminator, ERR packet, prepared-statement completion). --- src/sql/mysql/protocol/AnyMySQLError.rs | 1 + src/sql_jsc/mysql/MySQLConnection.rs | 18 +- .../mysql/protocol/any_mysql_error_jsc.rs | 1 + test/js/sql/sql-mysql-sequence-desync.test.ts | 220 ++++++++++++++++++ 4 files changed, 239 insertions(+), 1 deletion(-) create mode 100644 test/js/sql/sql-mysql-sequence-desync.test.ts diff --git a/src/sql/mysql/protocol/AnyMySQLError.rs b/src/sql/mysql/protocol/AnyMySQLError.rs index 0dc341ec5e32..8de2797f17f6 100644 --- a/src/sql/mysql/protocol/AnyMySQLError.rs +++ b/src/sql/mysql/protocol/AnyMySQLError.rs @@ -42,6 +42,7 @@ pub enum Error { InvalidEOFPacket, InvalidErrorPacket, UnexpectedPacket, + PacketsOutOfOrder, ShortRead, UnknownError, InvalidState, diff --git a/src/sql_jsc/mysql/MySQLConnection.rs b/src/sql_jsc/mysql/MySQLConnection.rs index ceabd1322dcc..5fb7b8b52ecb 100644 --- a/src/sql_jsc/mysql/MySQLConnection.rs +++ b/src/sql_jsc/mysql/MySQLConnection.rs @@ -589,7 +589,16 @@ impl MySQLConnection { }); reader.skip(PacketHeader::SIZE as isize); - // Update sequence id + // Command-phase responses restart at seq 1 after each seq-0 command + // packet; a mismatch means residual bytes from the previous exchange + // would be routed to the next queued query (CR_NET_PACKETS_OUT_OF_ORDER). + if self.status == ConnectionState::Connected && header.sequence_id != self.sequence_id { + debug!( + "packet out of order: expected seq {}, got {}", + self.sequence_id, header.sequence_id + ); + return Err(AnyMySQLError::PacketsOutOfOrder); + } self.sequence_id = header.sequence_id.wrapping_add(1); // Process packet based on connection state @@ -789,6 +798,8 @@ impl MySQLConnection { self.status_flags = ok.status_flags; self.flags.insert(ConnectionFlags::IS_READY_FOR_QUERY); self.queue.mark_as_ready_for_query(); + // Next command is sent at seq 0; its first response must be seq 1. + self.sequence_id = 1; self.advance(); } @@ -986,6 +997,7 @@ impl MySQLConnection { self.flags.insert(ConnectionFlags::IS_READY_FOR_QUERY); self.queue.mark_as_ready_for_query(); self.queue.mark_current_request_as_finished(request); + self.sequence_id = 1; // R-2: `on_error_packet` is `&self`; route through the // audited `js_connection_ref()` container_of accessor (one // centralised unsafe). `*self` sits inside the parent's @@ -1123,6 +1135,7 @@ impl MySQLConnection { self.queue.mark_as_ready_for_query(); self.queue.mark_as_prepared(); statement.reset(); + self.sequence_id = 1; self.advance(); } } @@ -1254,6 +1267,7 @@ impl MySQLConnection { }; self.queue.mark_as_ready_for_query(); self.queue.mark_current_request_as_finished(request); + self.sequence_id = 1; // R-2: `on_error_packet` is `&self`; `js_connection_ref()` is // the audited container_of accessor (one centralised unsafe). @@ -1306,6 +1320,7 @@ impl MySQLConnection { if is_last_result { self.queue.mark_as_ready_for_query(); self.queue.mark_current_request_as_finished(request); + self.sequence_id = 1; } // Short-lived borrow via the audited accessor; dropped before the @@ -1380,6 +1395,7 @@ impl MySQLConnection { self.flags.insert(ConnectionFlags::IS_READY_FOR_QUERY); self.queue.mark_as_ready_for_query(); self.queue.mark_current_request_as_finished(request); + self.sequence_id = 1; // R-2: `on_error_packet` is `&self`; route through the audited // `js_connection_ref()` container_of accessor. `*self` lives diff --git a/src/sql_jsc/mysql/protocol/any_mysql_error_jsc.rs b/src/sql_jsc/mysql/protocol/any_mysql_error_jsc.rs index c8e975d8a2d1..a882d78c84fa 100644 --- a/src/sql_jsc/mysql/protocol/any_mysql_error_jsc.rs +++ b/src/sql_jsc/mysql/protocol/any_mysql_error_jsc.rs @@ -107,6 +107,7 @@ pub(crate) fn mysql_error_to_js( "InvalidEOFPacket" => b"ERR_MYSQL_INVALID_EOF_PACKET", "InvalidErrorPacket" => b"ERR_MYSQL_INVALID_ERROR_PACKET", "UnexpectedPacket" => b"ERR_MYSQL_UNEXPECTED_PACKET", + "PacketsOutOfOrder" => b"ERR_MYSQL_PACKETS_OUT_OF_ORDER", "ConnectionTimedOut" => b"ERR_MYSQL_CONNECTION_TIMEOUT", "IdleTimeout" => b"ERR_MYSQL_IDLE_TIMEOUT", "LifetimeTimeout" => b"ERR_MYSQL_LIFETIME_TIMEOUT", diff --git a/test/js/sql/sql-mysql-sequence-desync.test.ts b/test/js/sql/sql-mysql-sequence-desync.test.ts new file mode 100644 index 000000000000..66c3fa69292d --- /dev/null +++ b/test/js/sql/sql-mysql-sequence-desync.test.ts @@ -0,0 +1,220 @@ +// Fault-injection test: requires a server that refuses / drops / sends malformed +// frames, which a healthy container will not do on demand. DO NOT COPY THIS +// PATTERN — anything a real server can produce belongs in describeWithContainer. +// All wire-protocol bytes come from test/js/sql/wire-frames.ts; do not inline +// Buffer.alloc frame construction here. + +import { SQL } from "bun"; +import { expect, test } from "bun:test"; +import { + listeningServer, + mysqlColumnDefinition, + mysqlHandshakeV10, + mysqlLenencInt, + mysqlOkPacket, + mysqlRawPacket, + mysqlReadPackets, + mysqlTextResultSetRow, +} from "./wire-frames"; + +// Build a single-column text result set starting at `seq` (column-count packet, +// one ColumnDefinition41, one row, OK-with-0xFE terminator). Returns the bytes +// and the sequence id that would follow the terminator. +function textResultSet(seq: number, columnName: string, value: string): { bytes: Buffer; nextSeq: number } { + const parts = [ + mysqlRawPacket(seq, mysqlLenencInt(1)), + mysqlColumnDefinition(seq + 1, { name: columnName, type: 0xfd /* VAR_STRING */ }), + mysqlTextResultSetRow(seq + 2, [value]), + mysqlOkPacket(seq + 3, 0xfe), + ]; + return { bytes: Buffer.concat(parts), nextSeq: seq + 4 }; +} + +type Settled = { state: "ok"; value: unknown } | { state: "rej"; code: string }; +function track(q: Promise, into: Record, key: string) { + into[key] = "unsettled"; + q.then( + v => (into[key] = { state: "ok", value: v }), + e => (into[key] = { state: "rej", code: String(e?.code ?? e) }), + ); + return q; +} + +// Two concurrent simple queries A and B on a max:1 pool (B queues client-side). +// The server answers A with its result set PLUS an unsolicited "GHOST" result +// set in the SAME write. The ghost packets carry the continuation sequence ids +// (5..8); a real answer to B would restart at seq 1 after B's COM_QUERY (seq 0). +// Without sequence-id validation, A's terminator advances the queue, B's +// COM_QUERY is sent, and the trailing ghost bytes in the buffer are routed to +// B, which resolves with rows the server never produced for it. +test("MySQL residual bytes after a completed result set are not delivered to the next queued query", async () => { + const a = textResultSet(1, "a", "Arow"); + const ghost = textResultSet(a.nextSeq, "g", "GHOST"); + let seenQueries = 0; + const sockets: import("node:net").Socket[] = []; + + const { port, server } = await listeningServer(socket => { + sockets.push(socket); + let buffered = Buffer.alloc(0); + let authed = false; + socket.on("error", () => {}); + socket.write(mysqlHandshakeV10()); + socket.on("data", chunk => { + buffered = Buffer.concat([buffered, chunk]); + buffered = mysqlReadPackets(buffered, (_seq, payload) => { + if (!authed) { + authed = true; + socket.write(mysqlOkPacket(2)); + return; + } + if (payload[0] === 0x01 /* COM_QUIT */) return socket.end(); + if (payload[0] !== 0x03 /* COM_QUERY */) return socket.write(mysqlOkPacket(1)); + seenQueries += 1; + // Answer A, and in the same segment append the unsolicited ghost. + // B's COM_QUERY (seenQueries === 2) is never answered. + if (seenQueries === 1) socket.write(Buffer.concat([a.bytes, ghost.bytes])); + }); + }); + }); + + const sql = new SQL({ + adapter: "mysql", + hostname: "127.0.0.1", + port, + username: "u", + password: "", + database: "d", + tls: false, + max: 1, + }); + try { + const r: Record = {}; + const qa = track(sql.unsafe("select a").simple(), r, "A"); + const qb = track(sql.unsafe("select b").simple(), r, "B"); + await Promise.allSettled([qa, qb]); + + // A settles with its real row regardless of the fix. + expect(r.A).toEqual({ state: "ok", value: [{ a: "Arow" }] }); + // B MUST NOT resolve with the ghost rows. The connection must fail on the + // out-of-order packet, rejecting B and every later query. + expect(r.B).toEqual({ state: "rej", code: "ERR_MYSQL_PACKETS_OUT_OF_ORDER" }); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + for (const s of sockets) s.destroy(); + await new Promise(resolve => server.close(() => resolve())); + } +}); + +// Same shape with A receiving an ERR packet instead of a result set: the ERR +// settles A (rejected), advances the queue, and trailing ghost bytes carrying +// continuation sequence ids must not be routed to B. +test("MySQL residual bytes after an ERR packet are not delivered to the next queued query", async () => { + // ERR_Packet: Int<1>(0xff) Int<2>(error_code) '#' String<5>(sql_state) String(message) + const errForA = mysqlRawPacket( + 1, + Buffer.concat([Buffer.from([0xff, 0x28, 0x04]), Buffer.from("#42000"), Buffer.from("syntax error")]), + ); + const ghost = textResultSet(2, "g", "GHOST"); + let seenQueries = 0; + const sockets: import("node:net").Socket[] = []; + + const { port, server } = await listeningServer(socket => { + sockets.push(socket); + let buffered = Buffer.alloc(0); + let authed = false; + socket.on("error", () => {}); + socket.write(mysqlHandshakeV10()); + socket.on("data", chunk => { + buffered = Buffer.concat([buffered, chunk]); + buffered = mysqlReadPackets(buffered, (_seq, payload) => { + if (!authed) { + authed = true; + socket.write(mysqlOkPacket(2)); + return; + } + if (payload[0] === 0x01 /* COM_QUIT */) return socket.end(); + if (payload[0] !== 0x03 /* COM_QUERY */) return socket.write(mysqlOkPacket(1)); + seenQueries += 1; + if (seenQueries === 1) socket.write(Buffer.concat([errForA, ghost.bytes])); + }); + }); + }); + + const sql = new SQL({ + adapter: "mysql", + hostname: "127.0.0.1", + port, + username: "u", + password: "", + database: "d", + tls: false, + max: 1, + }); + try { + const r: Record = {}; + const qa = track(sql.unsafe("select a").simple(), r, "A"); + const qb = track(sql.unsafe("select b").simple(), r, "B"); + await Promise.allSettled([qa, qb]); + + // A was rejected by the server's ERR packet (errno 1064). + expect(r.A).toMatchObject({ state: "rej" }); + expect((r.A as Settled & { code?: string }).code).not.toBe("ERR_MYSQL_PACKETS_OUT_OF_ORDER"); + // B MUST NOT resolve with the ghost rows. + expect(r.B).toEqual({ state: "rej", code: "ERR_MYSQL_PACKETS_OUT_OF_ORDER" }); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + for (const s of sockets) s.destroy(); + await new Promise(resolve => server.close(() => resolve())); + } +}); + +// Baseline: when the server answers each query with a well-formed result set +// whose sequence ids restart at 1 (as the protocol requires), both queries +// resolve with their own rows. This guards against the validation rejecting +// the legitimate case. +test("MySQL sequential queries on one connection each receive their own rows", async () => { + const answers = [textResultSet(1, "a", "one").bytes, textResultSet(1, "b", "two").bytes]; + let seenQueries = 0; + const sockets: import("node:net").Socket[] = []; + + const { port, server } = await listeningServer(socket => { + sockets.push(socket); + let buffered = Buffer.alloc(0); + let authed = false; + socket.on("error", () => {}); + socket.write(mysqlHandshakeV10()); + socket.on("data", chunk => { + buffered = Buffer.concat([buffered, chunk]); + buffered = mysqlReadPackets(buffered, (_seq, payload) => { + if (!authed) { + authed = true; + socket.write(mysqlOkPacket(2)); + return; + } + if (payload[0] === 0x01 /* COM_QUIT */) return socket.end(); + if (payload[0] !== 0x03 /* COM_QUERY */) return socket.write(mysqlOkPacket(1)); + const answer = answers[seenQueries++]; + if (answer) socket.write(answer); + }); + }); + }); + + const sql = new SQL({ + adapter: "mysql", + hostname: "127.0.0.1", + port, + username: "u", + password: "", + database: "d", + tls: false, + max: 1, + }); + try { + const [ra, rb] = await Promise.all([sql.unsafe("select a").simple(), sql.unsafe("select b").simple()]); + expect({ ra, rb }).toEqual({ ra: [{ a: "one" }], rb: [{ b: "two" }] }); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + for (const s of sockets) s.destroy(); + await new Promise(resolve => server.close(() => resolve())); + } +}); From 647a62cf9a78cebef8712b74f4d1a1c5e070f354 Mon Sep 17 00:00:00 2001 From: robobun Date: Sun, 12 Jul 2026 17:01:34 +0000 Subject: [PATCH 2/4] test: trim narrative comments, add 255->0 seq-id wrap coverage --- test/js/sql/sql-mysql-sequence-desync.test.ts | 85 ++++++++++++++++--- 1 file changed, 71 insertions(+), 14 deletions(-) diff --git a/test/js/sql/sql-mysql-sequence-desync.test.ts b/test/js/sql/sql-mysql-sequence-desync.test.ts index 66c3fa69292d..346e4827dd73 100644 --- a/test/js/sql/sql-mysql-sequence-desync.test.ts +++ b/test/js/sql/sql-mysql-sequence-desync.test.ts @@ -40,13 +40,9 @@ function track(q: Promise, into: Record, return q; } -// Two concurrent simple queries A and B on a max:1 pool (B queues client-side). -// The server answers A with its result set PLUS an unsolicited "GHOST" result -// set in the SAME write. The ghost packets carry the continuation sequence ids -// (5..8); a real answer to B would restart at seq 1 after B's COM_QUERY (seq 0). -// Without sequence-id validation, A's terminator advances the queue, B's -// COM_QUERY is sent, and the trailing ghost bytes in the buffer are routed to -// B, which resolves with rows the server never produced for it. +// A and B share max:1. The server appends an unsolicited GHOST result set +// (seq 5..8) after A's terminator in the same write; B's real response would +// restart at seq 1, so the ghost must fail the connection instead of reaching B. test("MySQL residual bytes after a completed result set are not delivered to the next queued query", async () => { const a = textResultSet(1, "a", "Arow"); const ghost = textResultSet(a.nextSeq, "g", "GHOST"); @@ -105,9 +101,8 @@ test("MySQL residual bytes after a completed result set are not delivered to the } }); -// Same shape with A receiving an ERR packet instead of a result set: the ERR -// settles A (rejected), advances the queue, and trailing ghost bytes carrying -// continuation sequence ids must not be routed to B. +// Same shape with A receiving an ERR packet: trailing ghost bytes carrying +// continuation sequence ids must not be routed to B after the queue advances. test("MySQL residual bytes after an ERR packet are not delivered to the next queued query", async () => { // ERR_Packet: Int<1>(0xff) Int<2>(error_code) '#' String<5>(sql_state) String(message) const errForA = mysqlRawPacket( @@ -168,10 +163,8 @@ test("MySQL residual bytes after an ERR packet are not delivered to the next que } }); -// Baseline: when the server answers each query with a well-formed result set -// whose sequence ids restart at 1 (as the protocol requires), both queries -// resolve with their own rows. This guards against the validation rejecting -// the legitimate case. +// Baseline: well-formed responses restarting at seq 1 are accepted and each +// query receives its own rows. test("MySQL sequential queries on one connection each receive their own rows", async () => { const answers = [textResultSet(1, "a", "one").bytes, textResultSet(1, "b", "two").bytes]; let seenQueries = 0; @@ -218,3 +211,67 @@ test("MySQL sequential queries on one connection each receive their own rows", a await new Promise(resolve => server.close(() => resolve())); } }); + +// Baseline: a result set with >256 packets wraps the u8 sequence id through +// 255 -> 0; the validation must accept the wrapped sequence and the follow-up +// query must still be accepted after the reset. +test("MySQL sequence-id validation accepts the 255->0 wrap within a result set", async () => { + const rowCount = 300; + function bigResultSet(): Buffer { + let seq = 1; + const parts: Buffer[] = [ + mysqlRawPacket(seq++, mysqlLenencInt(1)), + mysqlColumnDefinition(seq++, { name: "n", type: 0xfd }), + ]; + for (let i = 0; i < rowCount; i++) parts.push(mysqlTextResultSetRow(seq++ & 0xff, [String(i)])); + parts.push(mysqlOkPacket(seq & 0xff, 0xfe)); + return Buffer.concat(parts); + } + const answers = [bigResultSet(), textResultSet(1, "after", "ok").bytes]; + let seenQueries = 0; + const sockets: import("node:net").Socket[] = []; + + const { port, server } = await listeningServer(socket => { + sockets.push(socket); + let buffered = Buffer.alloc(0); + let authed = false; + socket.on("error", () => {}); + socket.write(mysqlHandshakeV10()); + socket.on("data", chunk => { + buffered = Buffer.concat([buffered, chunk]); + buffered = mysqlReadPackets(buffered, (_seq, payload) => { + if (!authed) { + authed = true; + socket.write(mysqlOkPacket(2)); + return; + } + if (payload[0] === 0x01 /* COM_QUIT */) return socket.end(); + if (payload[0] !== 0x03 /* COM_QUERY */) return socket.write(mysqlOkPacket(1)); + const answer = answers[seenQueries++]; + if (answer) socket.write(answer); + }); + }); + }); + + const sql = new SQL({ + adapter: "mysql", + hostname: "127.0.0.1", + port, + username: "u", + password: "", + database: "d", + tls: false, + max: 1, + }); + try { + const big = (await sql.unsafe("select n").simple()) as Array<{ n: string }>; + expect(big.length).toBe(rowCount); + expect({ first: big[0], last: big[rowCount - 1] }).toEqual({ first: { n: "0" }, last: { n: String(rowCount - 1) } }); + const after = await sql.unsafe("select after").simple(); + expect(after).toEqual([{ after: "ok" }]); + } finally { + await sql.close({ timeout: 0 }).catch(() => {}); + for (const s of sockets) s.destroy(); + await new Promise(resolve => server.close(() => resolve())); + } +}); From ec2f4f614167d101551eaaaeb0e3a5843c2f60d3 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Sun, 12 Jul 2026 17:03:36 +0000 Subject: [PATCH 3/4] [autofix.ci] apply automated fixes --- test/js/sql/sql-mysql-sequence-desync.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/js/sql/sql-mysql-sequence-desync.test.ts b/test/js/sql/sql-mysql-sequence-desync.test.ts index 346e4827dd73..0daf8c9f189e 100644 --- a/test/js/sql/sql-mysql-sequence-desync.test.ts +++ b/test/js/sql/sql-mysql-sequence-desync.test.ts @@ -266,7 +266,10 @@ test("MySQL sequence-id validation accepts the 255->0 wrap within a result set", try { const big = (await sql.unsafe("select n").simple()) as Array<{ n: string }>; expect(big.length).toBe(rowCount); - expect({ first: big[0], last: big[rowCount - 1] }).toEqual({ first: { n: "0" }, last: { n: String(rowCount - 1) } }); + expect({ first: big[0], last: big[rowCount - 1] }).toEqual({ + first: { n: "0" }, + last: { n: String(rowCount - 1) }, + }); const after = await sql.unsafe("select after").simple(); expect(after).toEqual([{ after: "ok" }]); } finally { From b0d8643cc20601c1dfbf24388c798e3326f449c5 Mon Sep 17 00:00:00 2001 From: robobun Date: Sun, 12 Jul 2026 17:25:02 +0000 Subject: [PATCH 4/4] ci: retrigger