From 045e704b858776203b7ba31b473a426786d330df Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Mon, 30 Aug 2021 14:01:40 -0700
Subject: [PATCH v6 17/19] 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 ++++++++++++
 3 files changed, 78 insertions(+), 13 deletions(-)

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 bfb7d1a261..86ca118b01 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"
@@ -1572,6 +1573,7 @@ apply_handle_insert(StringInfo s)
 	LogicalRepRelMapEntry *rel;
 	LogicalRepTupleData newtup;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	TupleTableSlot *remoteslot;
@@ -1594,6 +1596,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));
 
 	/* Set relation for error callback */
 	apply_error_callback_arg.rel = rel;
@@ -1694,6 +1702,7 @@ apply_handle_update(StringInfo s)
 {
 	LogicalRepRelMapEntry *rel;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	LogicalRepTupleData oldtup;
@@ -1721,6 +1730,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));
 
 	/* Set relation for error callback */
 	apply_error_callback_arg.rel = rel;
@@ -1861,6 +1876,7 @@ apply_handle_delete(StringInfo s)
 	LogicalRepRelMapEntry *rel;
 	LogicalRepTupleData oldtup;
 	LogicalRepRelId relid;
+	AclResult	aclresult;
 	ApplyExecutionData *edata;
 	EState	   *estate;
 	TupleTableSlot *remoteslot;
@@ -1883,6 +1899,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));
 
 	/* Set relation for error callback */
 	apply_error_callback_arg.rel = rel;
@@ -2255,6 +2277,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))
@@ -2266,6 +2289,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);
-- 
2.21.1 (Apple Git-122.3)

