From a83edeaf684a70da02fa2bce8b7e3aefa3f906f5 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <msawada@postgresql.org>
Date: Fri, 9 Jun 2023 10:43:03 +0900
Subject: Honor run_as_owner option in tablesync worker.

Commit 482675987 introduced "run_as_owner" subscription option so that
subscription runs with either the permissions of the subscription
owner or the permission of the table owner. However, tablesync workers
did not use this option for the initial data copy.

With this change, tablesync workers run with appropriate permissions
based on "run_as_owner" option.

Ajin Cherian, with changes and regression tests added by me.

Reported-By: Amit Kapila
Author: Ajin Cherian, Masahiko Sawada
Reviewed-by: Ajin Cherian, Amit Kapila
Discussion: https://postgr.es/m/CAA4eK1L=qzRHPEn+qeMoKQGFBzqGoLBzt_ov0A89iFFiut+ppA@mail.gmail.com
---
 src/backend/replication/logical/tablesync.c | 64 ++++++++++++++++++-----------
 1 file changed, 39 insertions(+), 25 deletions(-)

(limited to 'src/backend/replication/logical/tablesync.c')

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index c56d42dcd2c..abae8d44dfc 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -120,6 +120,7 @@
 #include "utils/rls.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
+#include "utils/usercontext.h"
 
 static bool table_states_valid = false;
 static List *table_states_not_ready = NIL;
@@ -1252,7 +1253,9 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	WalRcvExecResult *res;
 	char		originname[NAMEDATALEN];
 	RepOriginId originid;
+	UserContext	ucxt;
 	bool		must_use_password;
+	bool		run_as_owner;
 
 	/* Check the state of the table synchronization. */
 	StartTransactionCommand();
@@ -1374,31 +1377,6 @@ 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));
-
-	/*
-	 * COPY FROM does not honor RLS policies.  That is not a problem for
-	 * subscriptions owned by roles with BYPASSRLS privilege (or superuser,
-	 * who has it implicitly), but other roles should not be able to
-	 * circumvent RLS.  Disallow logical replication into RLS enabled
-	 * relations for such roles.
-	 */
-	if (check_enable_rls(RelationGetRelid(rel), InvalidOid, false) == RLS_ENABLED)
-		ereport(ERROR,
-				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-				 errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
-						GetUserNameFromId(GetUserId(), true),
-						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
@@ -1456,6 +1434,39 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 						originname)));
 	}
 
+	/*
+	 * Make sure that the copy command runs as the table owner, unless
+	 * the user has opted out of that behaviour.
+	 */
+	run_as_owner = MySubscription->runasowner;
+	if (!run_as_owner)
+		SwitchToUntrustedUser(rel->rd_rel->relowner, &ucxt);
+
+	/*
+	 * 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));
+
+	/*
+	 * COPY FROM does not honor RLS policies.  That is not a problem for
+	 * subscriptions owned by roles with BYPASSRLS privilege (or superuser,
+	 * who has it implicitly), but other roles should not be able to
+	 * circumvent RLS.  Disallow logical replication into RLS enabled
+	 * relations for such roles.
+	 */
+	if (check_enable_rls(RelationGetRelid(rel), InvalidOid, false) == RLS_ENABLED)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
+						GetUserNameFromId(GetUserId(), true),
+						RelationGetRelationName(rel))));
+
 	/* Now do the initial data copy */
 	PushActiveSnapshot(GetTransactionSnapshot());
 	copy_table(rel);
@@ -1469,6 +1480,9 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 						res->err)));
 	walrcv_clear_result(res);
 
+	if(!run_as_owner)
+		RestoreUserContext(&ucxt);
+
 	table_close(rel, NoLock);
 
 	/* Make the copy visible. */
-- 
cgit v1.2.3