From a0576815d336dbca80e4b750a37edba6113e796c Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Thu, 29 Jul 2021 14:57:29 -0700
Subject: [PATCH v5 10/12] Respect permissions within logical replication

Prevent logical replication workers from performing insert, update,
delete, truncate, or copy commands on tables unless the subscription
owner has permission to do so.
---
 doc/src/sgml/logical-replication.sgml       |  49 +++-
 src/backend/replication/logical/tablesync.c |  13 +
 src/backend/replication/logical/worker.c    |  29 ++
 src/test/subscription/t/023_aclcheck.pl     | 297 ++++++++++++++++++++
 4 files changed, 375 insertions(+), 13 deletions(-)
 create mode 100644 src/test/subscription/t/023_aclcheck.pl

diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 88646bc859..d6c7698847 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -330,6 +330,15 @@
    will simply be skipped.
   </para>
 
+  <para>
+   Logical replication operations are performed with the privileges of the role
+   which owns the subscription.  Only superusers may create subscriptions, but
+   if the ownership of the subscription is transferred, or if the owner
+   subsequently has superuser privileges revoked, permissions failures can
+   cause conflicts.  (Note that <productname>PostgreSQL</productname> prior to
+   version 15.0 did no permissions checking when applying changes.)
+  </para>
+
   <para>
    A conflict will produce an error and will stop the replication; it must be
    resolved manually by the user.  Details about the conflict can be found in
@@ -337,7 +346,7 @@
   </para>
 
   <para>
-   The resolution can be done either by changing data on the subscriber so
+   The resolution can be done either by changing data or permissions on the subscriber so
    that it does not conflict with the incoming change or by skipping the
    transaction that conflicts with the existing data.  The transaction can be
    skipped by calling the <link linkend="pg-replication-origin-advance">
@@ -530,13 +539,15 @@
 
   <para>
    A user able to modify the schema of subscriber-side tables can execute
-   arbitrary code as a superuser.  Limit ownership
-   and <literal>TRIGGER</literal> privilege on such tables to roles that
-   superusers trust.  Moreover, if untrusted users can create tables, use only
-   publications that list tables explicitly.  That is to say, create a
-   subscription <literal>FOR ALL TABLES</literal> only when superusers trust
-   every user permitted to create a non-temp table on the publisher or the
-   subscriber.
+   arbitrary code as the role which owns any subscription which modifies those
+   tables.  Limit ownership and <literal>TRIGGER</literal> privilege on such
+   tables to trusted roles.  Likewise, limit privileges of the subscription
+   owner to only the privileges needed to modify the tables that the
+   subscription should be able to change.  Moreover, if untrusted users can
+   create tables, use only publications that list tables explicitly.  That is
+   to say, create a subscription <literal>FOR ALL TABLES</literal> only when
+   superusers trust every user permitted to create a non-temp table on the
+   publisher or the subscriber.
   </para>
 
   <para>
@@ -569,18 +580,30 @@
   </para>
 
   <para>
-   To create a subscription, the user must be a superuser.
+   To create a subscription, the user must be a superuser.  The ownership can
+   subsequently be transferred or the role have superuser privilege removed.
   </para>
 
   <para>
    The subscription apply process will run in the local database with the
-   privileges of a superuser.
+   privileges of the subscription owner.
+  </para>
+
+  <para>
+   On the publisher, privileges are only checked once at the start of a
+   replication connection and are not re-checked as each change record is read.
   </para>
 
   <para>
-   Privileges are only checked once at the start of a replication connection.
-   They are not re-checked as each change record is read from the publisher,
-   nor are they re-checked for each change when applied.
+   On the subscriber, the subscription owner's privileges are re-checked for
+   each change record when applied, but beware that a change of ownership for a
+   subscription may not be noticed immediately by the replication workers.
+   Changes made on the publisher may be applied on the subscriber as
+   the old owner.  In such cases, the old owner's privileges will be the ones
+   that matter.  Worse still, it may be hard to predict when replication
+   workers will notice the new ownership.  Subscriptions created disabled and
+   only enabled after ownership has been changed will not be subject to this
+   race condition.
   </para>
  </sect1>
 
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a43c..2400ef8c45 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -111,6 +111,7 @@
 #include "replication/origin.h"
 #include "storage/ipc.h"
 #include "storage/lmgr.h"
+#include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
@@ -924,6 +925,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	char		relstate;
 	XLogRecPtr	relstate_lsn;
 	Relation	rel;
+	AclResult	aclresult;
 	WalRcvExecResult *res;
 	char		originname[NAMEDATALEN];
 	RepOriginId originid;
@@ -1042,6 +1044,17 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	 */
 	rel = table_open(MyLogicalRepWorker->relid, RowExclusiveLock);
 
+	/*
+	 * Check that our table sync worker has permission to insert into the
+	 * target table.
+	 */
+	aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
+								  ACL_INSERT);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult,
+					   get_relkind_objtype(rel->rd_rel->relkind),
+					   RelationGetRelationName(rel));
+
 	/*
 	 * Start a transaction in the remote node in REPEATABLE READ mode.  This
 	 * ensures that both the replication slot we create (see below) and the
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 38b493e4f5..6ee37ceb90 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -179,6 +179,7 @@
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "tcop/tcopprot.h"
+#include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/catcache.h"
 #include "utils/dynahash.h"
@@ -1569,6 +1570,7 @@ apply_handle_insert(StringInfo s)
 	LogicalRepRelMapEntry *rel;
 	LogicalRepTupleData newtup;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	TupleTableSlot *remoteslot;
@@ -1591,6 +1593,12 @@ apply_handle_insert(StringInfo s)
 		end_replication_step();
 		return;
 	}
+	aclresult = pg_class_aclcheck(RelationGetRelid(rel->localrel), GetUserId(),
+								  ACL_INSERT);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult,
+					   get_relkind_objtype(rel->localrel->rd_rel->relkind),
+					   get_rel_name(rel->localreloid));
 
 	/* Initialize the executor state. */
 	edata = create_edata_for_relation(rel);
@@ -1685,6 +1693,7 @@ apply_handle_update(StringInfo s)
 {
 	LogicalRepRelMapEntry *rel;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	LogicalRepTupleData oldtup;
@@ -1712,6 +1721,12 @@ apply_handle_update(StringInfo s)
 		end_replication_step();
 		return;
 	}
+	aclresult = pg_class_aclcheck(RelationGetRelid(rel->localrel), GetUserId(),
+								  ACL_UPDATE);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult,
+					   get_relkind_objtype(rel->localrel->rd_rel->relkind),
+					   get_rel_name(rel->localreloid));
 
 	/* Check if we can do the update. */
 	check_relation_updatable(rel);
@@ -1846,6 +1861,7 @@ apply_handle_delete(StringInfo s)
 	LogicalRepRelMapEntry *rel;
 	LogicalRepTupleData oldtup;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	TupleTableSlot *remoteslot;
@@ -1868,6 +1884,12 @@ apply_handle_delete(StringInfo s)
 		end_replication_step();
 		return;
 	}
+	aclresult = pg_class_aclcheck(RelationGetRelid(rel->localrel), GetUserId(),
+								  ACL_DELETE);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult,
+					   get_relkind_objtype(rel->localrel->rd_rel->relkind),
+					   get_rel_name(rel->localreloid));
 
 	/* Check if we can do the delete. */
 	check_relation_updatable(rel);
@@ -2234,6 +2256,7 @@ apply_handle_truncate(StringInfo s)
 	{
 		LogicalRepRelId relid = lfirst_oid(lc);
 		LogicalRepRelMapEntry *rel;
+		AclResult	aclresult;
 
 		rel = logicalrep_rel_open(relid, lockmode);
 		if (!should_apply_changes_for_rel(rel))
@@ -2245,6 +2268,12 @@ apply_handle_truncate(StringInfo s)
 			logicalrep_rel_close(rel, lockmode);
 			continue;
 		}
+		aclresult = pg_class_aclcheck(RelationGetRelid(rel->localrel),
+									  GetUserId(), ACL_TRUNCATE);
+		if (aclresult != ACLCHECK_OK)
+			aclcheck_error(aclresult,
+						   get_relkind_objtype(rel->localrel->rd_rel->relkind),
+						   get_rel_name(rel->localreloid));
 
 		remote_rels = lappend(remote_rels, rel);
 		rels = lappend(rels, rel->localrel);
diff --git a/src/test/subscription/t/023_aclcheck.pl b/src/test/subscription/t/023_aclcheck.pl
new file mode 100644
index 0000000000..c5450634e4
--- /dev/null
+++ b/src/test/subscription/t/023_aclcheck.pl
@@ -0,0 +1,297 @@
+# Copyright (c) 2021, PostgreSQL Global Development Group
+
+# This test checks that logical replication table sync workers and apply workers
+# respect privileges on database objects.
+#
+# We create a publisher node, a subscriber node, and a user named
+# 'replication_manager' which owns the publications and subscriptions but is
+# not a superuser.  If the database security surrounding logical replication is
+# working correctly, logical replication of data to the subscriber node will
+# fail or succeed depending on whether replication_manager has sufficient
+# privileges on the table involved.
+#
+
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 15;
+use Time::HiRes qw(usleep);
+
+my ($cmd, $publisher, $subscriber);
+
+$publisher = PostgresNode->new('publisher');
+$publisher->init(allows_streaming => 'logical');
+
+$subscriber = PostgresNode->new('subscriber');
+$subscriber->init;
+
+$publisher->start;
+$subscriber->start;
+
+# Create identical schema structure on both nodes
+both(qq(
+	CREATE USER db_superuser WITH SUPERUSER;
+	CREATE USER replication_manager WITH NOSUPERUSER;
+
+	GRANT CREATE ON DATABASE postgres TO db_superuser;
+	GRANT CREATE ON DATABASE postgres TO replication_manager;
+
+	SET SESSION AUTHORIZATION db_superuser;
+
+	-- Create a table owned by db_superuser with privileges granted to
+	-- replication_manager
+	CREATE TABLE shared_tbl (i INTEGER);
+	CREATE UNIQUE INDEX shared_tbl_idx ON shared_tbl(i);
+	ALTER TABLE shared_tbl REPLICA IDENTITY FULL;
+	REVOKE ALL PRIVILEGES ON TABLE shared_tbl FROM PUBLIC;
+	GRANT INSERT, UPDATE, DELETE, TRUNCATE
+		ON shared_tbl
+		TO replication_manager;
+
+	-- Create a table owned by db_superuser without privileges granted to
+	-- anybody else
+	CREATE TABLE private_tbl (i INTEGER);
+	CREATE UNIQUE INDEX private_tbl_idx ON private_tbl(i);
+	ALTER TABLE private_tbl REPLICA IDENTITY FULL;
+	REVOKE ALL PRIVILEGES ON TABLE private_tbl FROM PUBLIC;
+
+	RESET SESSION AUTHORIZATION;
+));
+
+# Check that replication_manager can replicate data for the shared table
+# for which replication_manager has privileges.
+create_publication_and_subscription("shared_tbl");
+publish_and_wait(
+	command_to_execute_on_publisher =>
+		"INSERT INTO shared_tbl values (generate_series(1,10))",
+	subscription_to_wait_for_catchup =>
+		"shared_tbl_sub");
+expect_subscriber_table_contents(
+	table => "shared_tbl",
+	new_minimum_column_value => 1,
+	new_maximum_column_value => 10,
+	new_row_count => 10,
+	testname => "replication of initial data with permissions");
+publish_and_wait(
+	command_to_execute_on_publisher =>
+		"DELETE FROM shared_tbl",
+	subscription_to_wait_for_catchup =>
+		"shared_tbl_sub");
+expect_subscriber_table_contents(
+	table => "shared_tbl",
+	new_minimum_column_value => '',
+	new_maximum_column_value => '',
+	new_row_count => 0,
+	testname => "replication of DELETEs with permissions");
+publish_and_wait(
+	command_to_execute_on_publisher =>
+		"INSERT INTO shared_tbl values (generate_series(11,20))",
+	subscription_to_wait_for_catchup =>
+		"shared_tbl_sub");
+expect_subscriber_table_contents(
+	table => "shared_tbl",
+	new_minimum_column_value => 11,
+	new_maximum_column_value => 20,
+	new_row_count => 10,
+	testname => "replication of INSERTs with permissions");
+publish_and_wait(
+	command_to_execute_on_publisher =>
+		"UPDATE shared_tbl SET i = i + 10",
+	subscription_to_wait_for_catchup =>
+		"shared_tbl_sub");
+expect_subscriber_table_contents(
+	table => "shared_tbl",
+	new_minimum_column_value => 21,
+	new_maximum_column_value => 30,
+	new_row_count => 10,
+	testname => "replication of UPDATEs with permissions");
+publish_and_wait(
+	command_to_execute_on_publisher =>
+		"TRUNCATE shared_tbl",
+	subscription_to_wait_for_catchup =>
+		"shared_tbl_sub");
+expect_subscriber_table_contents(
+	table => "shared_tbl",
+	new_minimum_column_value => '',
+	new_maximum_column_value => '',
+	new_row_count => 0,
+	testname => "replication of TRUNCATEs with permissions");
+
+# Check that replication_manager cannot replicate data for the private table
+# for which replication_manager has no privileges, but that granting privileges
+# on the table will clear the error and allow the data to replicate.
+create_publication_and_subscription("private_tbl");
+$publisher->safe_psql(
+	'postgres',
+	"INSERT INTO private_tbl values (generate_series(1,10))");
+expect_and_clear_subscriber_side_error(
+	table => "private_tbl",
+	new_minimum_column_value => 1,
+	new_maximum_column_value => 10,
+	new_row_count => 10,
+	testname => "replication of initial data for private_tbl");
+
+$publisher->safe_psql(
+	'postgres',
+	"DELETE FROM private_tbl");
+expect_and_clear_subscriber_side_error(
+	table => "private_tbl",
+	new_minimum_column_value => '',
+	new_maximum_column_value => '',
+	new_row_count => 0,
+	testname => "replication of DELETEs for private_tbl");
+
+$publisher->safe_psql(
+	'postgres',
+	"INSERT INTO private_tbl values (generate_series(11,20))");
+expect_and_clear_subscriber_side_error(
+	table => "private_tbl",
+	new_minimum_column_value => 11,
+	new_maximum_column_value => 20,
+	new_row_count => 10,
+	testname => "replication of INSERTs for private_tbl");
+
+$publisher->safe_psql(
+	'postgres',
+	"UPDATE private_tbl SET i = i + 10");
+expect_and_clear_subscriber_side_error(
+	table => "private_tbl",
+	new_minimum_column_value => 21,
+	new_maximum_column_value => 30,
+	new_row_count => 10,
+	testname => "replication of UPDATEs for private_tbl");
+
+$publisher->safe_psql(
+	'postgres',
+	"TRUNCATE private_tbl");
+expect_and_clear_subscriber_side_error(
+	table => "private_tbl",
+	new_minimum_column_value => '',
+	new_maximum_column_value => '',
+	new_row_count => 0,
+	testname => "replication of TRUNCATEs for private_tbl");
+
+sub both
+{
+	my ($cmd) = @_;
+	$publisher->safe_psql('postgres', $cmd);
+	$subscriber->safe_psql('postgres', $cmd);
+}
+
+sub create_publication_and_subscription
+{
+	my ($table) = @_;
+
+	my $publication = "${table}_pub";
+	my $subscription = "${table}_sub";
+
+	$publisher->safe_psql('postgres', qq(
+		CREATE PUBLICATION $publication FOR TABLE $table;
+		ALTER PUBLICATION $publication OWNER TO replication_manager;));
+	my $publisher_connstr = $publisher->connstr . ' dbname=postgres';
+
+	# If the subscription were created enabled (the default) rather than
+	# disabled, a race condition with the logical replication launcher would
+	# exist.  If the launcher were to read the subscription entry before the
+	# ownership were reassigned, it could launch the tablesync worker as the
+	# database owner rather than as the replication_manager.  This is not
+	# theoretical, but observed to occur.
+	$subscriber->safe_psql('postgres', qq(
+		CREATE SUBSCRIPTION $subscription
+			CONNECTION '$publisher_connstr'
+			PUBLICATION $publication
+			WITH (enabled = false);
+		ALTER SUBSCRIPTION $subscription OWNER TO replication_manager;
+		ALTER SUBSCRIPTION $subscription ENABLE;
+		));
+}
+
+sub publish_and_wait
+{
+	my %params = @_;
+
+	$publisher->safe_psql(
+		'postgres',
+		$params{command_to_execute_on_publisher});
+	$publisher->wait_for_catchup(
+		$params{subscription_to_wait_for_catchup});
+}
+
+sub query_subscriber_values
+{
+	my %params = @_;
+	return $subscriber->safe_psql('postgres',
+		qq(SELECT min(i), max(i), count(i) FROM $params{table}));
+}
+
+sub expected_subscriber_values
+{
+	my %params = @_;
+
+	return join('|',
+				"$params{new_minimum_column_value}",
+				"$params{new_maximum_column_value}",
+				"$params{new_row_count}");
+}
+
+sub expect_subscriber_table_contents
+{
+	my %params = @_;
+
+	is (query_subscriber_values(%params),
+		expected_subscriber_values(%params),
+		$params{testname});
+}
+
+sub expect_and_clear_subscriber_side_error
+{
+	my %params = @_;
+
+	# We cannot merely call wait_for_catchup(), as the subscriber side error we
+	# expect will prevent it from ever happening.  Instead, we must poll the
+	# subscriber log file for the error message.  On the other hand, if the
+	# error does not happen, we'll hang unless we also periodically check for
+	# the new data to arrive.
+	my $retries_remaining = 30;
+	my $expected = expected_subscriber_values(%params);
+	my $failure_re = qr/ERROR:  permission denied for table $params{table}/;
+	my ($actual, $contents);
+	do
+	{
+		$actual = query_subscriber_values(%params);
+		$contents = TestLib::slurp_file($subscriber->logfile);
+	}
+	while ($actual ne $expected && $contents !~ /$failure_re/ms &&
+		   defined(usleep(10000)) && $retries_remaining--);
+	BAIL_OUT("Subscription for $params{table} neither failed nor caught up!")
+		unless $retries_remaining;
+	like($contents, $failure_re, $params{testname});
+
+	# To clear the data, we must give the replication manager permission to
+	# write the new data to the table.  But since the lsn for the new data
+	# has already been attempted and failed, we perform a meaningless update
+	# of the table on the publisher side to move the lsn.
+	both("GRANT ALL PRIVILEGES ON $params{table} TO replication_manager");
+	$retries_remaining = 30;
+	do
+	{
+		$publisher->safe_psql('postgres',
+			qq(INSERT INTO $params{table} (i) VALUES (-1)));
+		$publisher->safe_psql('postgres',
+			qq(DELETE FROM $params{table} WHERE i = -1));
+		$actual = query_subscriber_values(%params);
+	}
+	while ($actual ne $expected && defined(usleep(10000)) &&
+		   $retries_remaining--);
+
+	$params{testname} .= " cleared error";
+	expect_subscriber_table_contents(%params);
+
+	# Restore replication_manager's permissions and clear out the logfile failure
+	# so that subsequent tests can run properly.
+	both("REVOKE ALL PRIVILEGES ON $params{table} FROM replication_manager");
+	$subscriber->stop;
+	unlink $subscriber->logfile() or die "Cannot remove subscriber logfile: $!";
+	$subscriber->start;
+}
-- 
2.21.1 (Apple Git-122.3)

