diff --git a/src/sql/postgres/protocol/ArrayList.rs b/src/sql/postgres/protocol/ArrayList.rs index d0c2da8f0784..dc9b8dbc635d 100644 --- a/src/sql/postgres/protocol/ArrayList.rs +++ b/src/sql/postgres/protocol/ArrayList.rs @@ -63,6 +63,9 @@ impl<'a> WriterContext for ArrayListCtx<'a> { arr[i..i + bytes.len()].copy_from_slice(bytes); Ok(()) } + fn truncate(mut self, offset: usize) { + self.array_mut().truncate(offset); + } } pub type Writer<'a> = NewWriter>; diff --git a/src/sql/postgres/protocol/NewWriter.rs b/src/sql/postgres/protocol/NewWriter.rs index 811968b3d14b..86e9ed9b7ef3 100644 --- a/src/sql/postgres/protocol/NewWriter.rs +++ b/src/sql/postgres/protocol/NewWriter.rs @@ -7,6 +7,10 @@ pub trait WriterContext: Copy { fn offset(self) -> usize; fn write(self, bytes: &[u8]) -> Result<(), AnyPostgresError>; fn pwrite(self, bytes: &[u8], offset: usize) -> Result<(), AnyPostgresError>; + /// Discard everything written at or after `offset` (a value previously + /// returned by [`offset`]). Used to roll back a partially-written message + /// when encoding fails partway through. + fn truncate(self, offset: usize); } #[derive(Copy, Clone)] @@ -60,6 +64,22 @@ impl NewWriter { C::pwrite(self.wrapped, data, i) } + /// Run `f`; on error, discard every byte it wrote so a half-serialised + /// message (e.g. a Bind whose parameter coercion threw in JS) is never + /// left in the buffer to desync the next query's flush. + #[inline] + pub fn atomically( + self, + f: impl FnOnce(Self) -> Result, + ) -> Result { + let start = self.offset(); + let result = f(self); + if result.is_err() { + C::truncate(self.wrapped, start); + } + result + } + pub fn int4(self, value: PostgresInt32) -> Result<(), AnyPostgresError> { self.write(&value.to_be_bytes()) } diff --git a/src/sql_jsc/postgres/PostgresRequest.rs b/src/sql_jsc/postgres/PostgresRequest.rs index 72ecc08dbbd4..b640f35df5d5 100644 --- a/src/sql_jsc/postgres/PostgresRequest.rs +++ b/src/sql_jsc/postgres/PostgresRequest.rs @@ -289,36 +289,38 @@ pub(crate) fn prepare_and_query_with_signature( global: &JSGlobalObject, query: &[u8], array_value: JSValue, - mut writer: protocol::NewWriter, + writer: protocol::NewWriter, signature: &mut Signature, ) -> Result<(), AnyPostgresError> { - write_query( - query, - &signature.prepared_statement_name, - &signature.fields, - writer, - )?; - write_bind( - &signature.prepared_statement_name, - BunString::empty(), - global, - array_value, - JSValue::ZERO, - &[], - &[], - writer, - )?; - let exec = protocol::Execute { - p: protocol::PortalOrPreparedStatement::PreparedStatement( + writer.atomically(|mut writer| { + write_query( + query, &signature.prepared_statement_name, - ), - ..Default::default() - }; - exec.write_internal(&mut writer)?; + &signature.fields, + writer, + )?; + write_bind( + &signature.prepared_statement_name, + BunString::empty(), + global, + array_value, + JSValue::ZERO, + &[], + &[], + writer, + )?; + let exec = protocol::Execute { + p: protocol::PortalOrPreparedStatement::PreparedStatement( + &signature.prepared_statement_name, + ), + ..Default::default() + }; + exec.write_internal(&mut writer)?; - writer.write(&protocol::FLUSH)?; - writer.write(&protocol::SYNC)?; - Ok(()) + writer.write(&protocol::FLUSH)?; + writer.write(&protocol::SYNC)?; + Ok(()) + }) } pub(crate) fn bind_and_execute( @@ -326,29 +328,31 @@ pub(crate) fn bind_and_execute( statement: &PostgresSQLStatement, array_value: JSValue, columns_value: JSValue, - mut writer: protocol::NewWriter, + writer: protocol::NewWriter, ) -> Result<(), AnyPostgresError> { - write_bind( - &statement.signature.prepared_statement_name, - BunString::empty(), - global, - array_value, - columns_value, - &statement.parameters, - &statement.fields, - writer, - )?; - let exec = protocol::Execute { - p: protocol::PortalOrPreparedStatement::PreparedStatement( + writer.atomically(|mut writer| { + write_bind( &statement.signature.prepared_statement_name, - ), - ..Default::default() - }; - exec.write_internal(&mut writer)?; + BunString::empty(), + global, + array_value, + columns_value, + &statement.parameters, + &statement.fields, + writer, + )?; + let exec = protocol::Execute { + p: protocol::PortalOrPreparedStatement::PreparedStatement( + &statement.signature.prepared_statement_name, + ), + ..Default::default() + }; + exec.write_internal(&mut writer)?; - writer.write(&protocol::FLUSH)?; - writer.write(&protocol::SYNC)?; - Ok(()) + writer.write(&protocol::FLUSH)?; + writer.write(&protocol::SYNC)?; + Ok(()) + }) } /// Atomically sends Parse + [Describe] + Bind + Execute + Flush + Sync as a single message batch. @@ -363,61 +367,63 @@ pub fn parse_and_bind_and_execute( array_value: JSValue, columns_value: JSValue, include_describe: bool, - mut writer: protocol::NewWriter, + writer: protocol::NewWriter, ) -> Result<(), AnyPostgresError> { - let name = &statement.signature.prepared_statement_name; + writer.atomically(|mut writer| { + let name = &statement.signature.prepared_statement_name; - // Parse - { - let q = protocol::Parse { - name, - params: &statement.signature.fields, - query, + // Parse + { + let q = protocol::Parse { + name, + params: &statement.signature.fields, + query, + }; + q.write_internal(&mut writer)?; + bun_core::scoped_log!(Postgres, "Parse: {}", bun_fmt::quote(query)); + } + + // Describe (needed on first execution to learn parameter/result types for caching) + if include_describe { + let d = protocol::Describe { + p: protocol::PortalOrPreparedStatement::PreparedStatement(name), + }; + d.write_internal(writer)?; + bun_core::scoped_log!(Postgres, "Describe: {}", bun_fmt::quote(name)); + } + + // Bind — use server-provided types if available (binary format), otherwise + // fall back to signature types (text format for unknowns). The server will + // handle text-to-type conversion based on the parameter types from Parse. + let param_fields = if !statement.parameters.is_empty() { + &statement.parameters[..] + } else { + &statement.signature.fields[..] }; - q.write_internal(&mut writer)?; - bun_core::scoped_log!(Postgres, "Parse: {}", bun_fmt::quote(query)); - } + let result_fields = &statement.fields; - // Describe (needed on first execution to learn parameter/result types for caching) - if include_describe { - let d = protocol::Describe { + write_bind( + name, + BunString::empty(), + global, + array_value, + columns_value, + param_fields, + result_fields, + writer, + )?; + + // Execute + let exec = protocol::Execute { p: protocol::PortalOrPreparedStatement::PreparedStatement(name), + ..Default::default() }; - d.write_internal(writer)?; - bun_core::scoped_log!(Postgres, "Describe: {}", bun_fmt::quote(name)); - } + exec.write_internal(&mut writer)?; - // Bind — use server-provided types if available (binary format), otherwise - // fall back to signature types (text format for unknowns). The server will - // handle text-to-type conversion based on the parameter types from Parse. - let param_fields = if !statement.parameters.is_empty() { - &statement.parameters[..] - } else { - &statement.signature.fields[..] - }; - let result_fields = &statement.fields; - - write_bind( - name, - BunString::empty(), - global, - array_value, - columns_value, - param_fields, - result_fields, - writer, - )?; - - // Execute - let exec = protocol::Execute { - p: protocol::PortalOrPreparedStatement::PreparedStatement(name), - ..Default::default() - }; - exec.write_internal(&mut writer)?; - - writer.write(&protocol::FLUSH)?; - writer.write(&protocol::SYNC)?; - Ok(()) + writer.write(&protocol::FLUSH)?; + writer.write(&protocol::SYNC)?; + Ok(()) + }) } pub(crate) fn execute_query( diff --git a/src/sql_jsc/postgres/PostgresSQLConnection.rs b/src/sql_jsc/postgres/PostgresSQLConnection.rs index f42d7c179971..d22b478426b9 100644 --- a/src/sql_jsc/postgres/PostgresSQLConnection.rs +++ b/src/sql_jsc/postgres/PostgresSQLConnection.rs @@ -1685,6 +1685,13 @@ impl protocol::WriterContext for Writer { fn pwrite(mut self, bytes: &[u8], i: usize) -> Result<(), AnyPostgresError> { Writer::pwrite(&mut self, bytes, i) } + #[inline] + fn truncate(self, offset: usize) { + self.connection.write_buffer.with_mut(|b| { + debug_assert!(b.head as usize + offset <= b.byte_list.len()); + b.byte_list.truncate(b.head as usize + offset); + }); + } } impl PostgresSQLConnection { diff --git a/test/js/sql/postgres-bind-throw-torn-frame.test.ts b/test/js/sql/postgres-bind-throw-torn-frame.test.ts new file mode 100644 index 000000000000..5eec681d6f2e --- /dev/null +++ b/test/js/sql/postgres-bind-throw-torn-frame.test.ts @@ -0,0 +1,156 @@ +// 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. +// +// A parameter whose valueOf()/toString() throws mid-Bind must not leave the +// half-written `B\0\0\0\0…` prefix in the write buffer; flushed ahead of the +// next query it reads as `invalid message length` and drops the connection. +import { SQL } from "bun"; +import { afterAll, expect, test } from "bun:test"; +import { + listeningServer, + pgAuthenticationOk, + pgBindComplete, + pgCommandComplete, + pgDataRow, + pgParameterDescription, + pgParseComplete, + pgReadFrontendMessages, + pgReadyForQuery, + pgRowDescription, +} from "./wire-frames"; + +// Mock backend: replies to startup / Parse / Bind with the minimal happy-path +// sequence and records every raw byte received after startup so the test can +// verify the client's wire framing directly. +let received!: Buffer; +const { port, server } = await listeningServer(socket => { + let pending = Buffer.alloc(0); + let sawStartup = false; + socket.on("data", chunk => { + if (sawStartup) received = Buffer.concat([received, chunk]); + pending = Buffer.concat([pending, chunk]); + if (!sawStartup) { + if (pending.length < 4) return; + const len = pending.readInt32BE(0); + if (pending.length < len) return; + pending = pending.subarray(len); + sawStartup = true; + received = Buffer.concat([received, pending]); + socket.write(Buffer.concat([pgAuthenticationOk(), pgReadyForQuery()])); + } + pending = pgReadFrontendMessages(pending, type => { + if (type === 0x50 /* Parse 'P' */) { + socket.write( + Buffer.concat([ + pgParseComplete(), + pgParameterDescription([23 /* int4 */]), + pgRowDescription([{ name: "v", typeOid: 23, format: 1 }]), + pgReadyForQuery(), + ]), + ); + } else if (type === 0x42 /* Bind 'B' */) { + socket.write( + Buffer.concat([ + pgBindComplete(), + pgDataRow([Buffer.from([0, 0, 0, 2])]), // int4 value 2, binary + pgCommandComplete("SELECT 1"), + pgReadyForQuery(), + ]), + ); + } + }); + }); + socket.on("error", () => {}); +}); +afterAll(() => new Promise(r => server.close(() => r()))); + +function newClient() { + return new SQL({ + adapter: "postgres", + hostname: "127.0.0.1", + port, + username: "u", + database: "db", + tls: false, + max: 1, + prepare: true, + connectionTimeout: 2, + }); +} + +/** Walk `buf` as Byte1-type + Int32-length frontend messages: returns the type + * list iff every declared length is ≥ 4 and messages tile the buffer exactly, + * otherwise the torn offset and a hex dump of the bytes there. */ +function frameTypes(buf: Buffer): { types: string[] } | { tornAt: number; head: string } { + const types: string[] = []; + let o = 0; + while (o + 5 <= buf.length) { + const len = buf.readInt32BE(o + 1); + if (len < 4 || o + 1 + len > buf.length) { + return { tornAt: o, head: [...buf.subarray(o, o + 16)].map(b => b.toString(16).padStart(2, "0")).join(" ") }; + } + types.push(String.fromCharCode(buf[o])); + o += 1 + len; + } + if (o !== buf.length) { + return { tornAt: o, head: [...buf.subarray(o)].map(b => b.toString(16).padStart(2, "0")).join(" ") }; + } + return { types }; +} + +async function run(evil: unknown, errorType: new (...a: any[]) => Error, message: string) { + received = Buffer.alloc(0); + const db = newClient(); + try { + // First query: the parameter's JS coercion throws mid-Bind, after the + // 'B' tag + zero length placeholder + names + format codes are buffered. + const first = await db`select ${evil as any}::int4 as v`.catch(e => e); + expect(first).toBeInstanceOf(errorType); + expect((first as Error).message).toBe(message); + + // Second, innocent query on the same connection. + const rows: any = await db`select ${1}::int4 as v`; + + // Every byte sent after startup must be a well-formed frontend message; + // a torn Bind fails frameTypes() with head `42 00 00 00 00 …`. + const framed = frameTypes(received); + expect(framed).toEqual({ + types: expect.arrayContaining(["B", "E", "S"]), + }); + // Exactly one Bind reached the wire (the second query's), and it decoded + // to the row the server sent for it. + expect({ + binds: ((framed as { types: string[] }).types ?? []).filter(t => t === "B"), + row: rows[0], + }).toEqual({ binds: ["B"], row: { v: 2 } }); + } finally { + await db.close({ timeout: 0 }).catch(() => {}); + } +} + +test("postgres: a throwing valueOf() during Bind does not leave a torn frame on the wire", async () => { + await run( + { + valueOf() { + throw new RangeError("evil valueOf"); + }, + }, + RangeError, + "evil valueOf", + ); +}); + +test("postgres: a throwing toString() during Bind does not leave a torn frame on the wire", async () => { + await run( + { + toString() { + throw new TypeError("evil toString"); + }, + }, + TypeError, + "evil toString", + ); +});