From b8dc90f386fe3e7afb4d350b45a22cbb03e66a16 Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Wed, 23 Jun 2021 16:03:49 -0700
Subject: [PATCH v3 1/4] Add default role for managing logical replication

Reducing the number of administrative tasks that require superuser
privileges by adding a new pg_logical_replication role and
delegating to it the authority to add, alter, or drop publications
and subscriptions on any table regardless of table ownership, while
respecting table privileges in table sync and apply workers.

With this change, by creating a user in role pg_logical_replication,
only giving that user INSERT, UPDATE, DELETE, or TRUNCATE privileges
as appropriate on the intended tables, and having that user rather
than a superuser create a subscription, one may prevent the
replication of unwanted DML on these tables as well as the
replication of any DML to any other tables.  The prior situation
prevented new tables added to a publication from being replicated so
long as a ALTER SUBSCRIPTION ... REFRESH PUBLICATION was avoided,
but that has two deficiencies.  First, there is a race condition
between when the administrator on the subscriber side checks the
list of tables being published on the publication side and when the
subscription is created.  Second, all DML on the subscribed tables
was replicated, whereas now the administrator on the subscriber side
can restrict, for example, to only accept INSERTs.

The use of these new restrictions on logical replication will likely
result in an increase of field reports of subscriptions going into
an infinite loop of failing to apply, aborting, restarting, and
retrying.  Loops of that kind are already possible for other kinds
of errors, such as unique violations on the subscriber side.  As
such, this commit leaves the problem of resolving stuck
subscriptions for another day.
---
 doc/src/sgml/ref/create_role.sgml             |   7 +-
 doc/src/sgml/user-manag.sgml                  |  13 +
 src/backend/commands/publicationcmds.c        |  27 +-
 src/backend/commands/subscriptioncmds.c       |  26 +-
 src/backend/replication/logical/tablesync.c   |  13 +
 src/backend/replication/logical/worker.c      |  25 ++
 src/include/catalog/pg_authid.dat             |   5 +
 src/test/regress/expected/subscription.out    |   4 +-
 .../subscription/t/021_priv_escalation.pl     | 330 ++++++++++++++++++
 9 files changed, 429 insertions(+), 21 deletions(-)
 create mode 100644 src/test/subscription/t/021_priv_escalation.pl

diff --git a/doc/src/sgml/ref/create_role.sgml b/doc/src/sgml/ref/create_role.sgml
index b6a4ea1f72..9468ff162c 100644
--- a/doc/src/sgml/ref/create_role.sgml
+++ b/doc/src/sgml/ref/create_role.sgml
@@ -174,9 +174,10 @@ in sync when changing the above synopsis!
        <para>
         These clauses determine whether a role is a replication role.  A role
         must have this attribute (or be a superuser) in order to be able to
-        connect to the server in replication mode (physical or logical
-        replication) and in order to be able to create or drop replication
-        slots.
+        connect to the server in physical replication mode.  Roles with this
+        attribute, along with the <literal>pg_logical_replication</literal>
+        role, can connect to the server in logical replication mode and can
+        create or drop replication slots. 
         A role having the <literal>REPLICATION</literal> attribute is a very
         highly privileged role, and should only be used on roles actually
         used for replication. If not specified,
diff --git a/doc/src/sgml/user-manag.sgml b/doc/src/sgml/user-manag.sgml
index fe0bdb7599..be74599b54 100644
--- a/doc/src/sgml/user-manag.sgml
+++ b/doc/src/sgml/user-manag.sgml
@@ -581,6 +581,12 @@ DROP ROLE doomed_role;
        <entry>Allow executing programs on the database server as the user the database runs as with
        COPY and other functions which allow executing a server-side program.</entry>
       </row>
+      <row>
+       <entry>pg_logical_replication</entry>
+       <entry>Create or drop replication slots and logical replication
+       subscriptions and publications, and connect to the server in logical
+       replication mode.</entry>
+      </row>
      </tbody>
     </tgroup>
    </table>
@@ -624,6 +630,13 @@ DROP ROLE doomed_role;
   great care should be taken when granting these roles to users.
   </para>
 
+  <para> The <literal>pg_logical_replication</literal> role is intended to
+  allow administrators to enable trusted, but non-superuser, roles which are
+  able to configure logical replication and to replicate data from a
+  publication while still being subject to privilege checks on the target
+  tables into which data is replicated.
+  </para>
+
   <para>
   Care should be taken when granting these roles to ensure they are only used where
   needed and with the understanding that these roles grant access to privileged
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 95c253c8e0..681fe8adf7 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -24,6 +24,7 @@
 #include "catalog/objectaccess.h"
 #include "catalog/objectaddress.h"
 #include "catalog/partition.h"
+#include "catalog/pg_authid.h"
 #include "catalog/pg_inherits.h"
 #include "catalog/pg_publication.h"
 #include "catalog/pg_publication_rel.h"
@@ -165,11 +166,15 @@ CreatePublication(CreatePublicationStmt *stmt)
 		aclcheck_error(aclresult, OBJECT_DATABASE,
 					   get_database_name(MyDatabaseId));
 
-	/* FOR ALL TABLES requires superuser */
-	if (stmt->for_all_tables && !superuser())
+	/*
+	 * FOR ALL TABLES requires superuser or membership in the
+	 * pg_logical_replication role
+	 */
+	if (stmt->for_all_tables && !superuser() &&
+		!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("must be superuser to create FOR ALL TABLES publication")));
+				 errmsg("must be superuser or a member of the pg_logical_replication role to create FOR ALL TABLES publication")));
 
 	rel = table_open(PublicationRelationId, RowExclusiveLock);
 
@@ -454,7 +459,8 @@ AlterPublication(AlterPublicationStmt *stmt)
 	pubform = (Form_pg_publication) GETSTRUCT(tup);
 
 	/* must be owner */
-	if (!pg_publication_ownercheck(pubform->oid, GetUserId()))
+	if (!pg_publication_ownercheck(pubform->oid, GetUserId()) &&
+		!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
 					   stmt->pubname);
 
@@ -615,8 +621,12 @@ PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
 		Relation	rel = (Relation) lfirst(lc);
 		ObjectAddress obj;
 
-		/* Must be owner of the table or superuser. */
-		if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId()))
+		/*
+		 * Must be owner of the table or superuser or a member of the
+		 * pg_logical_replication role.
+		 */
+		if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId()) &&
+			!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 			aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
 						   RelationGetRelationName(rel));
 
@@ -679,7 +689,8 @@ AlterPublicationOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
 	if (form->pubowner == newOwnerId)
 		return;
 
-	if (!superuser())
+	if (!superuser() &&
+		!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 	{
 		AclResult	aclresult;
 
@@ -702,7 +713,7 @@ AlterPublicationOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
 					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 					 errmsg("permission denied to change owner of publication \"%s\"",
 							NameStr(form->pubname)),
-					 errhint("The owner of a FOR ALL TABLES publication must be a superuser.")));
+					 errhint("The owner of a FOR ALL TABLES publication must be a superuser or a member of the pg_logical_replication role.")));
 	}
 
 	form->pubowner = newOwnerId;
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index b862e59f1d..5a1f193671 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -23,6 +23,7 @@
 #include "catalog/namespace.h"
 #include "catalog/objectaccess.h"
 #include "catalog/objectaddress.h"
+#include "catalog/pg_authid.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_subscription_rel.h"
 #include "catalog/pg_type.h"
@@ -372,10 +373,11 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
 	if (create_slot)
 		PreventInTransactionBlock(isTopLevel, "CREATE SUBSCRIPTION ... WITH (create_slot = true)");
 
-	if (!superuser())
+	if (!superuser() &&
+		!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("must be superuser to create subscriptions")));
+				 errmsg("must be superuser or a member of the pg_logical_replication role to create subscriptions")));
 
 	/*
 	 * If built with appropriate switch, whine when regression-testing
@@ -781,7 +783,8 @@ AlterSubscription(AlterSubscriptionStmt *stmt, bool isTopLevel)
 	subid = form->oid;
 
 	/* must be owner */
-	if (!pg_subscription_ownercheck(subid, GetUserId()))
+	if (!pg_subscription_ownercheck(subid, GetUserId()) &&
+		!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
 					   stmt->subname);
 
@@ -1098,7 +1101,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	subid = form->oid;
 
 	/* must be owner */
-	if (!pg_subscription_ownercheck(subid, GetUserId()))
+	if (!pg_subscription_ownercheck(subid, GetUserId()) &&
+		!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
 					   stmt->subname);
 
@@ -1383,17 +1387,23 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
 	if (form->subowner == newOwnerId)
 		return;
 
-	if (!pg_subscription_ownercheck(form->oid, GetUserId()))
+	if (!pg_subscription_ownercheck(form->oid, GetUserId()) &&
+		!has_privs_of_role(GetUserId(), ROLE_PG_LOGICAL_REPLICATION))
 		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
 					   NameStr(form->subname));
 
-	/* New owner must be a superuser */
-	if (!superuser_arg(newOwnerId))
+	/*
+	 * New owner must be a superuser or a member of the pg_logical_replication
+	 * role
+	 */
+	if (!superuser_arg(newOwnerId) &&
+		!has_privs_of_role(newOwnerId, ROLE_PG_LOGICAL_REPLICATION))
+
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 				 errmsg("permission denied to change owner of subscription \"%s\"",
 						NameStr(form->subname)),
-				 errhint("The owner of a subscription must be a superuser.")));
+				 errhint("The owner of a subscription must be a superuser or a member of the pg_logical_replication role.")));
 
 	form->subowner = newOwnerId;
 	CatalogTupleUpdate(rel, &tup->t_self, tup);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 682c107e74..9beaf59cbd 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -110,6 +110,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"
@@ -927,6 +928,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	char		relstate;
 	XLogRecPtr	relstate_lsn;
 	Relation	rel;
+	AclResult	aclresult;
 	WalRcvExecResult *res;
 	char		originname[NAMEDATALEN];
 	RepOriginId originid;
@@ -1045,6 +1047,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(MyLogicalRepWorker->relid, 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 bbb659dad0..872d3b1663 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -105,6 +105,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"
@@ -1260,6 +1261,7 @@ apply_handle_insert(StringInfo s)
 	LogicalRepRelMapEntry *rel;
 	LogicalRepTupleData newtup;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	TupleTableSlot *remoteslot;
@@ -1282,6 +1284,11 @@ apply_handle_insert(StringInfo s)
 		end_replication_step();
 		return;
 	}
+	aclresult = pg_class_aclcheck(rel->localreloid, 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);
@@ -1376,6 +1383,7 @@ apply_handle_update(StringInfo s)
 {
 	LogicalRepRelMapEntry *rel;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	LogicalRepTupleData oldtup;
@@ -1403,6 +1411,11 @@ apply_handle_update(StringInfo s)
 		end_replication_step();
 		return;
 	}
+	aclresult = pg_class_aclcheck(rel->localreloid, 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);
@@ -1537,6 +1550,7 @@ apply_handle_delete(StringInfo s)
 	LogicalRepRelMapEntry *rel;
 	LogicalRepTupleData oldtup;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	TupleTableSlot *remoteslot;
@@ -1559,6 +1573,11 @@ apply_handle_delete(StringInfo s)
 		end_replication_step();
 		return;
 	}
+	aclresult = pg_class_aclcheck(rel->localreloid, 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);
@@ -1925,6 +1944,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))
@@ -1936,6 +1956,11 @@ apply_handle_truncate(StringInfo s)
 			logicalrep_rel_close(rel, lockmode);
 			continue;
 		}
+		aclresult = pg_class_aclcheck(rel->localreloid, 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/include/catalog/pg_authid.dat b/src/include/catalog/pg_authid.dat
index 3da68016b6..0048a5fab7 100644
--- a/src/include/catalog/pg_authid.dat
+++ b/src/include/catalog/pg_authid.dat
@@ -79,5 +79,10 @@
   rolcreaterole => 'f', rolcreatedb => 'f', rolcanlogin => 'f',
   rolreplication => 'f', rolbypassrls => 'f', rolconnlimit => '-1',
   rolpassword => '_null_', rolvaliduntil => '_null_' },
+{ oid => '9276', oid_symbol => 'ROLE_PG_LOGICAL_REPLICATION',
+  rolname => 'pg_logical_replication', rolsuper => 'f', rolinherit => 't',
+  rolcreaterole => 'f', rolcreatedb => 'f', rolcanlogin => 'f',
+  rolreplication => 'f', rolbypassrls => 'f', rolconnlimit => '-1',
+  rolpassword => '_null_', rolvaliduntil => '_null_' },
 
 ]
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 57f7dd9b0a..07fb04a008 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -43,7 +43,7 @@ ERROR:  subscription "regress_testsub" already exists
 -- fail - must be superuser
 SET SESSION AUTHORIZATION 'regress_subscription_user2';
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION foo WITH (connect = false);
-ERROR:  must be superuser to create subscriptions
+ERROR:  must be superuser or a member of the pg_logical_replication role to create subscriptions
 SET SESSION AUTHORIZATION 'regress_subscription_user';
 -- fail - invalid option combinations
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = true);
@@ -137,7 +137,7 @@ ALTER SUBSCRIPTION regress_testsub_foo RENAME TO regress_testsub;
 -- fail - new owner must be superuser
 ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user2;
 ERROR:  permission denied to change owner of subscription "regress_testsub"
-HINT:  The owner of a subscription must be a superuser.
+HINT:  The owner of a subscription must be a superuser or a member of the pg_logical_replication role.
 ALTER ROLE regress_subscription_user2 SUPERUSER;
 -- now it works
 ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user2;
diff --git a/src/test/subscription/t/021_priv_escalation.pl b/src/test/subscription/t/021_priv_escalation.pl
new file mode 100644
index 0000000000..b765499374
--- /dev/null
+++ b/src/test/subscription/t/021_priv_escalation.pl
@@ -0,0 +1,330 @@
+# 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' in role 'pg_logical_replication' which is intentionally
+# not a superuser.  Logical replication subscriptions are created by this user,
+# and consequently the table sync and apply workers run as this user.  If the
+# database security surrounding logical replication is working correctly,
+# logical replication of data to the subscriber node which causes the workers
+# to modify tables will fail unless the replication_manager role has sufficient
+# privileges.
+#
+# Failures in the logical replication workers result in infinite retry loops on
+# the subscriber node.  To prevent the tests from getting stuck, we need to
+# stop, clean, recreate and restart the nodes after each test where we expect
+# such failures, thereby breaking the failure loops.
+#
+
+use strict;
+use warnings;
+use Time::HiRes qw(usleep);
+use PostgresNode;
+use TestLib;
+use Scalar::Util qw(blessed);
+use Test::More tests => 11;
+
+my ($cmd, $publisher, $subscriber);
+
+my $testno = 0;
+sub create_fresh_nodes
+{
+	$testno++;
+
+	$publisher = get_new_node("publisher_$testno");
+	$publisher->init(allows_streaming => 'logical');
+	$publisher->append_conf('postgresql.conf',
+							'max_worker_processes = 12');
+	$publisher->append_conf('postgresql.conf',
+							'max_logical_replication_workers = 6');
+	$publisher->start;
+
+	$subscriber = get_new_node("subscriber_$testno");
+	$subscriber->init;
+	$subscriber->append_conf('postgresql.conf',
+							 'max_worker_processes = 12');
+	$subscriber->append_conf('postgresql.conf',
+							 'max_logical_replication_workers = 6');
+	$subscriber->start;
+
+	# Create identical schema structure on both nodes
+	$cmd = q(
+		CREATE USER db_superuser WITH SUPERUSER;
+		CREATE USER replication_manager IN ROLE pg_logical_replication;
+
+		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;
+	);
+	$publisher->safe_psql('postgres', $cmd);
+	$subscriber->safe_psql('postgres', $cmd);
+}
+
+sub configure_logical_replication
+{
+	my %options = @_;
+
+	# On the subscriber side only, create a logging mechanism to track what
+	# operations are performed on which tables and by which users.
+	$subscriber->safe_psql('postgres', q(
+		CREATE TABLE log (
+			i INTEGER,
+			table_name TEXT,
+			op TEXT,
+			current_username TEXT,
+			session_username TEXT
+		);
+		GRANT ALL PRIVILEGES ON log TO public;
+		CREATE FUNCTION log_row_trigger_func() RETURNS TRIGGER AS $$
+		BEGIN
+			INSERT INTO public.log
+				VALUES (NEW.i, TG_TABLE_NAME, TG_OP, CURRENT_USER, SESSION_USER);
+			RETURN NEW;
+		END;
+		$$ LANGUAGE plpgsql;
+		GRANT ALL PRIVILEGES ON FUNCTION log_row_trigger_func() TO public;
+
+		SET SESSION AUTHORIZATION db_superuser;
+		CREATE TRIGGER shared_log_row_trigger
+			AFTER INSERT OR UPDATE ON shared_tbl
+			FOR EACH ROW EXECUTE FUNCTION log_row_trigger_func();
+		ALTER TABLE shared_tbl
+			ENABLE ALWAYS TRIGGER shared_log_row_trigger;
+		CREATE TRIGGER private_log_row_trigger
+			AFTER INSERT OR UPDATE ON private_tbl
+			FOR EACH ROW EXECUTE FUNCTION log_row_trigger_func();
+		ALTER TABLE private_tbl
+			ENABLE ALWAYS TRIGGER private_log_row_trigger;
+		RESET SESSION AUTHORIZATION;
+		));
+
+	my $publisher_connstr = $publisher->connstr . ' dbname=postgres';
+	$publisher->safe_psql('postgres', qq(
+		SET SESSION AUTHORIZATION replication_manager;
+		CREATE PUBLICATION private_tbl_pub
+			FOR TABLE private_tbl;
+		CREATE PUBLICATION shared_tbl_pub
+			FOR TABLE shared_tbl;
+	));
+
+	$subscriber->safe_psql('postgres', qq(
+		SET SESSION AUTHORIZATION replication_manager;
+		CREATE SUBSCRIPTION shared_tbl_sub
+			CONNECTION '$publisher_connstr'
+			PUBLICATION shared_tbl_pub;
+	));
+
+	if ($options{'attempt_private_replication'})
+	{
+		$subscriber->safe_psql('postgres', qq(
+			SET SESSION AUTHORIZATION replication_manager;
+			CREATE SUBSCRIPTION private_tbl_sub
+				CONNECTION '$publisher_connstr'
+				PUBLICATION private_tbl_pub;
+		));
+	}
+}
+
+sub destroy_nodes
+{
+	$subscriber->teardown_node();
+	$publisher->teardown_node();
+	$subscriber->clean_node();
+	$publisher->clean_node();
+}
+
+sub truncate_log
+{
+	$subscriber->safe_psql('postgres', qq(TRUNCATE log));
+}
+
+sub truncate_test_data
+{
+	$subscriber->safe_psql('postgres', qq(
+		TRUNCATE shared_tbl;
+		TRUNCATE private_tbl;
+		TRUNCATE log));
+}
+
+# Wait for a subscription to finish replicating or to fail in the attempt.
+#
+# failure_re: a regular expression that should match the subscriber node's log
+#             if and only if the replication failed.
+# success_sql: a sql query which will return true on the subscriber node if and
+#              only if the data being awaited replicated successfully.
+# expected: "fail" or "success", depending on whether we expect the replication
+#           attempt to fail or to succeed.
+#
+sub expect_replication
+{
+	my ($failure_re, $success_sql, $expected, $testname) = @_;
+
+	for (1..1800)	# 180 seconds max
+	{
+		# Has the subscription failed?
+		my $contents = TestLib::slurp_file($subscriber->logfile);
+		if ($contents =~ m/$failure_re/ms)
+		{
+			is ("fail", $expected, $testname);
+			return;
+		}
+
+		# Has the subscription caught up?
+		if ($subscriber->safe_psql('postgres', $success_sql) eq 't')
+		{
+			is ("success", $expected, $testname);
+			return;
+		}
+
+		# Wait 0.1 second before retrying.
+		usleep(100_000);
+	}
+
+	diag("replication_fails timed out waiting for failure matching /$failure_re/" .
+		 "or for true value returned from '$success_sql'");
+}
+
+sub expect_log
+{
+	my ($expected, $testname) = @_;
+
+	is ($subscriber->safe_psql('postgres', qq(SELECT * FROM log)),
+		$expected, $testname);
+}
+
+sub expect_value
+{
+	my ($table, $expected, $testname) = @_;
+
+	is ($subscriber->safe_psql('postgres', qq(SELECT i FROM $table)),
+		$expected, $testname);
+}
+
+#
+# Check that 'replication_manager' can replicate INSERT to the shared table
+#
+create_fresh_nodes();
+configure_logical_replication();
+$publisher->wait_for_catchup('shared_tbl_sub');
+$publisher->safe_psql('postgres', qq(
+	SET SESSION AUTHORIZATION db_superuser;
+	INSERT INTO shared_tbl (i) VALUES (1);
+));
+expect_replication(
+	qr/ERROR/,		# We should not see any ERROR message
+	qq(SELECT COUNT(1) > 0 FROM shared_tbl),
+	"success",
+	"replication can propogate inserts into shared_tbl");
+expect_log(
+	"1|shared_tbl|INSERT|replication_manager|replication_manager",
+	"successful replication of INSERT to shared_tbl is logged");
+expect_value(
+	"shared_tbl", "1",
+	"shared_tbl contains expected row after replication of INSERT");
+
+#
+# Check that 'replication_manager' can replicate UPDATE to the shared table
+#
+truncate_log();
+$publisher->safe_psql('postgres', qq(
+	SET SESSION AUTHORIZATION db_superuser;
+	UPDATE shared_tbl SET i = i + 1;
+));
+expect_replication(
+	qr/ERROR/,		# We should not see any ERROR message
+	qq(SELECT COUNT(1) > 0 FROM shared_tbl WHERE i = 2),
+	"success",
+	"replication can propogate updates into shared_tbl");
+expect_log(
+	"2|shared_tbl|UPDATE|replication_manager|replication_manager",
+	"successful replication of UPDATE to shared_tbl is logged");
+expect_value(
+	"shared_tbl", "2",
+	"shared_tbl contains expected row after replication of UPDATE");
+
+#
+# Check that having privileges to replicate into a table allows triggering
+# functions which replication_manager would otherwise not have permission to
+# execute.  This may seem wrong, but it is how function execution privileges
+# are handled vis-a-vis triggers under non-logical-replication circumstances,
+# and it would be more surprising for logical replication to diverge from this
+# behavior than to adhere to it.
+#
+truncate_test_data();
+$subscriber->safe_psql('postgres', q(
+	SET SESSION AUTHORIZATION db_superuser;
+	CREATE FUNCTION db_superuser_func() returns trigger as $$
+	BEGIN
+		NEW.i = NEW.i + 1;
+		RETURN NEW;
+	END;
+	$$ LANGUAGE plpgsql;
+	REVOKE ALL PRIVILEGES ON FUNCTION db_superuser_func() FROM PUBLIC;
+	CREATE TRIGGER escalation_trig
+		BEFORE INSERT OR UPDATE
+		ON shared_tbl
+		FOR EACH ROW EXECUTE FUNCTION db_superuser_func();
+	ALTER TABLE shared_tbl
+		ENABLE ALWAYS TRIGGER escalation_trig;
+	GRANT INSERT ON shared_tbl TO replication_manager;
+));
+$publisher->safe_psql('postgres', qq(
+	SET SESSION AUTHORIZATION db_superuser;
+	INSERT INTO shared_tbl (i) VALUES (3);
+));
+expect_replication(
+	qr/ERROR/,		# We should not see any ERROR message
+	qq(SELECT COUNT(1) > 0 FROM shared_tbl WHERE i IN (3, 4)),
+	"success",
+	"replication performs inserts despite lacking trigger function privileges");
+expect_log(
+	"4|shared_tbl|INSERT|replication_manager|replication_manager",
+	"successful replication and triggered modification of INSERT into shared_tbl is logged");
+expect_value(
+	"shared_tbl", "4",
+	"shared_tbl contains expected, trigger-altered value");
+truncate_test_data();
+
+#
+# Check that table sync workers cannot INSERT into the private table
+#
+destroy_nodes();
+create_fresh_nodes();
+$publisher->safe_psql('postgres', qq(
+SET SESSION AUTHORIZATION db_superuser;
+INSERT INTO private_tbl (i) VALUES (6);
+));
+configure_logical_replication(
+	attempt_private_replication => 1);
+expect_replication(
+	qr/ERROR/,
+	qq(SELECT COUNT(1) > 0 FROM private_tbl WHERE i = 6),
+	"fail",
+	"table sync workers cannot insert into private table");
+expect_value(
+	"private_tbl", "",
+	"private_tbl empty after failed table sync");
-- 
2.21.1 (Apple Git-122.3)

