From 3c26436948edf93c7f85035219270b239498f3da Mon Sep 17 00:00:00 2001 From: Nikolay Samokhvalov Date: Sat, 18 Jul 2026 02:10:59 +0000 Subject: [PATCH] Validate remote sequence data in sequencesync worker get_and_validate_seq_info() extracted most columns of the result set returned by the publisher's sequence synchronization query with a bare Assert(!isnull). Remote responses must be validated at runtime rather than assumed well-formed: an incompatible, broken, or malicious publisher could return NULL values that the sync query normally never produces. Such a response would crash assert-enabled builds; production builds would interpret the NULL datum as zero, feeding garbage such as a zero page LSN or bogus sequence parameters into the local sequence state, or dereference a null pointer on platforms where int64 is passed by reference. In addition, the sequence index echoed back by the publisher was used with list_nth() without verifying that it identifies a unique member of the current batch. An out-of-range value would perform an out-of-bounds access, resulting in undefined behavior, while an out-of-batch or duplicate index would associate the returned state with the wrong local sequence and corrupt the accounting used to detect rows missing from the response. Fix this by replacing the assertions with runtime checks that report a protocol violation naming the affected sequence and column, and by verifying that each sequence index received falls within the current batch and appears at most once. Also declare the index column of the result set as int4, matching what the query returns, instead of narrowing an int8 with DatumGetInt32(). This follows the precedent of commit 3cf5264557be, which added a response-shape check for CREATE_REPLICATION_SLOT one layer down in libpqwalreceiver. The column count of the result set needs no additional check here, as libpqrcv_processTuples() already verifies it against the expected number of fields. The has_sequence_privilege and last_value columns are also left alone, since NULL is a meaningful result for those and is already handled. Sequence synchronization is new in the current development cycle, so no released branch is affected. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PtxA6B4d47ietk5dGsseuP --- .../replication/logical/sequencesync.c | 119 ++++++++++++++---- 1 file changed, 98 insertions(+), 21 deletions(-) diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c index 63ad46d7fd7b2..64b1ac54cecd7 100644 --- a/src/backend/replication/logical/sequencesync.c +++ b/src/backend/replication/logical/sequencesync.c @@ -256,15 +256,50 @@ report_sequence_errors(List *mismatched_seqs_idx, MySubscription->name)); } +/* + * slot_getattr_notnull + * + * Like slot_getattr(), but report an error if the publisher unexpectedly + * returned a NULL value for the given column. + * + * The data returned by the publisher cannot be trusted blindly: an + * incompatible, broken, or malicious publisher could return NULL values that + * the sync query normally never produces. + */ +static Datum +slot_getattr_notnull(TupleTableSlot *slot, int col, const char *colname, + LogicalRepSequenceInfo *seqinfo) +{ + bool isnull; + Datum datum; + + datum = slot_getattr(slot, col, &isnull); + if (isnull) + ereport(ERROR, + errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid response from publisher"), + errdetail("Value of column \"%s\" for sequence \"%s.%s\" is null.", + colname, seqinfo->nspname, seqinfo->seqname)); + + return datum; +} + /* * get_and_validate_seq_info * * Extracts remote sequence information from the tuple slot received from the * publisher, and validates it against the corresponding local sequence * definition. + * + * The sequence index echoed back by the publisher is validated before use: + * it must identify a unique member of the current batch, which starts at + * index 'batch_base' and contains 'batch_size' entries of the seqinfos list. + * 'batch_seen' is a caller-provided array of 'batch_size' entries, false on + * first call for a batch, used to detect duplicates. */ static CopySeqResult -get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel, +get_and_validate_seq_info(TupleTableSlot *slot, int batch_base, int batch_size, + bool *batch_seen, Relation *sequence_rel, LogicalRepSequenceInfo **seqinfo, int *seqidx) { bool isnull; @@ -282,8 +317,38 @@ get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel, Form_pg_sequence local_seq; LogicalRepSequenceInfo *seqinfo_local; - *seqidx = DatumGetInt32(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + datum = slot_getattr(slot, ++col, &isnull); + if (isnull) + ereport(ERROR, + errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid response from publisher"), + errdetail("Sequence index is null.")); + + *seqidx = DatumGetInt32(datum); + + /* + * The index is sent to the publisher and echoed back in the result set. + * Before using it, make sure that it identifies a unique member of the + * current batch: an out-of-range index would perform an out-of-bounds + * access on the seqinfos list, and an out-of-batch or duplicate index + * would associate the returned state with the wrong local sequence and + * corrupt the accounting used to detect rows missing from the response. + */ + if (*seqidx < batch_base || *seqidx >= batch_base + batch_size) + ereport(ERROR, + errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid response from publisher"), + errdetail("Sequence index %d is outside the valid range %d..%d.", + *seqidx, batch_base, batch_base + batch_size - 1)); + + if (batch_seen[*seqidx - batch_base]) + ereport(ERROR, + errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid response from publisher"), + errdetail("Sequence index %d appears more than once.", + *seqidx)); + + batch_seen[*seqidx - batch_base] = true; /* Identify the corresponding local sequence for the given index. */ *seqinfo = seqinfo_local = @@ -313,29 +378,33 @@ get_and_validate_seq_info(TupleTableSlot *slot, Relation *sequence_rel, seqinfo_local->last_value = DatumGetInt64(datum); - seqinfo_local->is_called = DatumGetBool(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + seqinfo_local->is_called = + DatumGetBool(slot_getattr_notnull(slot, ++col, "is_called", + seqinfo_local)); - seqinfo_local->page_lsn = DatumGetLSN(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + seqinfo_local->page_lsn = + DatumGetLSN(slot_getattr_notnull(slot, ++col, "page_lsn", + seqinfo_local)); - remote_typid = DatumGetObjectId(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + remote_typid = DatumGetObjectId(slot_getattr_notnull(slot, ++col, + "seqtypid", + seqinfo_local)); - remote_start = DatumGetInt64(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + remote_start = DatumGetInt64(slot_getattr_notnull(slot, ++col, "seqstart", + seqinfo_local)); - remote_increment = DatumGetInt64(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + remote_increment = DatumGetInt64(slot_getattr_notnull(slot, ++col, + "seqincrement", + seqinfo_local)); - remote_min = DatumGetInt64(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + remote_min = DatumGetInt64(slot_getattr_notnull(slot, ++col, "seqmin", + seqinfo_local)); - remote_max = DatumGetInt64(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + remote_max = DatumGetInt64(slot_getattr_notnull(slot, ++col, "seqmax", + seqinfo_local)); - remote_cycle = DatumGetBool(slot_getattr(slot, ++col, &isnull)); - Assert(!isnull); + remote_cycle = DatumGetBool(slot_getattr_notnull(slot, ++col, "seqcycle", + seqinfo_local)); /* Sanity check */ Assert(col == REMOTE_SEQ_COL_COUNT); @@ -455,7 +524,7 @@ copy_sequences(WalReceiverConn *conn) while (cur_batch_base_index < n_seqinfos) { - Oid seqRow[REMOTE_SEQ_COL_COUNT] = {INT8OID, BOOLOID, INT8OID, + Oid seqRow[REMOTE_SEQ_COL_COUNT] = {INT4OID, BOOLOID, INT8OID, BOOLOID, LSNOID, OIDOID, INT8OID, INT8OID, INT8OID, INT8OID, BOOLOID}; int batch_size = 0; int batch_succeeded_count = 0; @@ -464,6 +533,7 @@ copy_sequences(WalReceiverConn *conn) int batch_sub_insuffperm_count = 0; int batch_pub_insuffperm_count = 0; int batch_missing_count; + bool *batch_seen; WalRcvExecResult *res; TupleTableSlot *slot; @@ -535,6 +605,9 @@ copy_sequences(WalReceiverConn *conn) "JOIN LATERAL pg_get_sequence_data(seq.seqrelid) AS ps ON true\n", seqstr.data); + /* Track which batch members have been seen in the response so far. */ + batch_seen = palloc0_array(bool, batch_size); + res = walrcv_exec(conn, cmd.data, lengthof(seqRow), seqRow); if (res->status != WALRCV_OK_TUPLES) ereport(ERROR, @@ -558,7 +631,10 @@ copy_sequences(WalReceiverConn *conn) ProcessConfigFile(PGC_SIGHUP); } - sync_status = get_and_validate_seq_info(slot, &sequence_rel, + sync_status = get_and_validate_seq_info(slot, + cur_batch_base_index, + batch_size, batch_seen, + &sequence_rel, &seqinfo, &seqidx); if (sync_status == COPYSEQ_SUCCESS) sync_status = copy_sequence(seqinfo, @@ -636,6 +712,7 @@ copy_sequences(WalReceiverConn *conn) ExecDropSingleTupleTableSlot(slot); walrcv_clear_result(res); + pfree(batch_seen); resetStringInfo(&seqstr); resetStringInfo(&cmd);