diff --git a/conn.go b/conn.go index 3ed7ecfa..f346c358 100644 --- a/conn.go +++ b/conn.go @@ -1101,7 +1101,14 @@ func (cn *conn) saveMessage(typ proto.ResponseCode, buf *readBuf) error { // recvMessage receives any message from the backend, or returns an error if // a problem occurred while reading the message. -func (cn *conn) recvMessage(r *readBuf) (proto.ResponseCode, error) { +// +// startup must only be true for callers reading messages during the +// connection-startup sequence (see [conn.recv]). It is never appropriate for +// steady-state message parsing: a v3 ErrorResponse is a normal, properly +// length-prefixed message post-handshake, and treating a large one as plain +// text (see the comment below) desynchronizes the wire and leaves inProgress +// stuck, poisoning the connection. +func (cn *conn) recvMessage(r *readBuf, startup bool) (proto.ResponseCode, error) { // workaround for a QueryRow bug, see exec if cn.saveMessageType != 0 { t := cn.saveMessageType @@ -1127,11 +1134,16 @@ func (cn *conn) recvMessage(r *readBuf) (proto.ResponseCode, error) { // When PostgreSQL cannot start a backend (e.g., an external process limit), // it sends plain text like "Ecould not fork new process [..]", which - // doesn't use the standard encoding for the Error message. + // doesn't use the standard encoding for the Error message. This can only + // happen before the protocol handshake completes: libpq only applies this + // heuristic while awaiting the startup response (fe-connect.c, + // PQconnectPoll), and once the protocol is established, an ErrorResponse + // is explicitly allowed to exceed MaxErrlen (fe-protocol3.c, + // VALID_LONG_MESSAGE_TYPE) since it's a normal, properly framed message. // // libpq checks "if ErrorResponse && (msgLength < 8 || msgLength > MAX_ERRLEN)", // but check < 4 since n represents bytes remaining to be read after length. - if t == proto.ErrorResponse && (n < 4 || n > proto.MaxErrlen) { + if startup && t == proto.ErrorResponse && (n < 4 || n > proto.MaxErrlen) { msg, _ := cn.buf.ReadString('\x00') return 0, fmt.Errorf("pq: server error: %s%s", string(x[1:]), strings.TrimSuffix(msg, "\x00")) } @@ -1160,7 +1172,7 @@ func (cn *conn) recvMessage(r *readBuf) (proto.ResponseCode, error) { func (cn *conn) recv() (proto.ResponseCode, *readBuf, error) { for { r := new(readBuf) - t, err := cn.recvMessage(r) + t, err := cn.recvMessage(r, true) if err != nil { return 0, nil, err } @@ -1185,7 +1197,7 @@ func (cn *conn) recv() (proto.ResponseCode, *readBuf, error) { // the caller to avoid an allocation. func (cn *conn) recv1Buf(r *readBuf) (proto.ResponseCode, error) { for { - t, err := cn.recvMessage(r) + t, err := cn.recvMessage(r, false) if err != nil { return 0, err } diff --git a/copy.go b/copy.go index 6a153b2e..c20ebe25 100644 --- a/copy.go +++ b/copy.go @@ -133,7 +133,7 @@ func (ci *copyin) flush(buf []byte) error { func (ci *copyin) resploop() { for { var r readBuf - t, err := ci.cn.recvMessage(&r) + t, err := ci.cn.recvMessage(&r, false) if err != nil { ci.setBad(driver.ErrBadConn) ci.setError(err) diff --git a/issues_test.go b/issues_test.go index 52c0ac6b..4c05f596 100644 --- a/issues_test.go +++ b/issues_test.go @@ -3,10 +3,14 @@ package pq import ( "context" "database/sql" + "errors" + "net" + "strings" "testing" "time" "github.com/lib/pq/internal/pqtest" + "github.com/lib/pq/internal/proto" "github.com/lib/pq/pqerror" ) @@ -133,3 +137,69 @@ func TestQueryCancelledReused(t *testing.T) { // get a connection: it must be valid connIsValid(t, db) } + +// #1324: an ErrorResponse larger than proto.MaxErrlen sent after the startup +// handshake was misparsed as a pre-protocol plain-text error (garbling the +// message) and left the connection desynchronized: the ErrorResponse body and +// the trailing ReadyForQuery were never drained, so inProgress stayed true +// and the poisoned connection was handed back out by the pool. +func TestOversizedErrorResponse(t *testing.T) { + t.Parallel() + + wantMsg := strings.Repeat("x", proto.MaxErrlen+10000) + + f := pqtest.NewFake(t, func(f pqtest.Fake, cn net.Conn) { + f.Startup(cn, nil) + for { + code, msg, ok := f.ReadMsg(cn) + if !ok { + return + } + switch code { + case proto.Terminate: + cn.Close() + return + case proto.Query: + switch strings.TrimRight(string(msg), "\x00") { + case ";": + // MustDB's Ping. + f.WriteMsg(cn, proto.EmptyQueryResponse, "") + f.WriteMsg(cn, proto.ReadyForQuery, "I") + case "select 1": + f.WriteMsg(cn, proto.CommandComplete, "SELECT 1\x00") + f.WriteMsg(cn, proto.ReadyForQuery, "I") + default: + f.WriteMsg(cn, proto.ErrorResponse, "SERROR\x00C58030\x00M"+wantMsg+"\x00\x00") + f.WriteMsg(cn, proto.ReadyForQuery, "I") + } + } + } + }) + defer f.Close() + + db := pqtest.MustDB(t, f.DSN()) + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + + _, err := db.Exec(`DO $$ BEGIN RAISE EXCEPTION 'x'; END $$;`) + if err == nil { + t.Fatal("first Exec: want non-nil error, got nil") + } + var pqErr *Error + if !errors.As(err, &pqErr) { + t.Fatalf("first Exec: want *pq.Error, got %T: %v", err, err) + } + if pqErr.Message != wantMsg { + t.Errorf("first Exec: Message mangled: got %d bytes, want %d bytes", len(pqErr.Message), len(wantMsg)) + } + if pqErr.Code != "58030" { + t.Errorf("first Exec: Code = %q, want 58030", pqErr.Code) + } + + // The connection must not be poisoned: a second query on the same pooled + // connection must not fail with errQueryInProgress. + _, err = db.Exec("select 1") + if err != nil { + t.Fatalf("second Exec: connection left poisoned: %v", err) + } +} diff --git a/notify.go b/notify.go index 4f4c4227..97a24b39 100644 --- a/notify.go +++ b/notify.go @@ -180,7 +180,7 @@ func (l *ListenerConn) setState(newState int32) bool { func (l *ListenerConn) listenerConnLoop() (err error) { r := &readBuf{} for { - t, err := l.cn.recvMessage(r) + t, err := l.cn.recvMessage(r, false) if err != nil { return err }