Respect permissions within logical replication.

jeff-davis · jeff-davis · commit a2ab9c06ea15 · 2022-01-07T17:40:56.000-08:00
Prevent logical replication workers from performing insert, update, delete, truncate, or copy commands on tables unless the subscription owner has permission to do so. Prevent subscription owners from circumventing row-level security by forbidding replication into tables with row-level security policies which the subscription owner is subject to, without regard to whether the policy would ordinarily allow the INSERT, UPDATE, DELETE or TRUNCATE which is being replicated. This seems sufficient for now, as superusers, roles with bypassrls, and target table owners should still be able to replicate despite RLS policies. We can revisit the question of applying row-level security policies on a per-row basis if this restriction proves too severe in practice. Author: Mark Dilger Reviewed-by: Jeff Davis, Andrew Dunstan, Ronan Dunklau Discussion: https://postgr.es/m/9DFC88D3-1300-4DE8-ACBC-4CEF84399A53%40enterprisedb.com
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
@@ -330,14 +330,27 @@
    will simply be skipped.
   </para>
 
+  <para>
+   Logical replication operations are performed with the privileges of the role
+   which owns the subscription.  Permissions failures on target tables will
+   cause replication conflicts, as will enabled
+   <link linkend="ddl-rowsecurity">row-level security</link> on target tables
+   that the subscription owner is subject to, without regard to whether any
+   policy would ordinary reject the <command>INSERT</command>,
+   <command>UPDATE</command>, <command>DELETE</command> or
+   <command>TRUNCATE</command> which is being replicated.  This restriction on
+   row-level security may be lifted in a future version of
+   <productname>PostgreSQL</productname>.
+  </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
    the subscriber's server log.
   </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,9 +543,9 @@
 
   <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
+   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.
+   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> or
    <literal>FOR ALL TABLES IN SCHEMA</literal> only when superusers trust
@@ -576,13 +589,20 @@
 
   <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 transaction when applied. If a worker is in the process of applying a
+   transaction when the ownership of the subscription is changed by a
+   concurrent transaction, the application of the current transaction will
+   continue under the old owner's privileges.
   </para>
  </sect1>
 
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
@@ -1481,6 +1481,8 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
 
 	InvokeObjectPostAlterHook(SubscriptionRelationId,
 							  form->oid, 0);
+
+	ApplyLauncherWakeupAtCommit();
 }
 
 /*
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
@@ -111,9 +111,11 @@
 #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"
+#include "utils/rls.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
 
@@ -924,6 +926,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	char		relstate;
 	XLogRecPtr	relstate_lsn;
 	Relation	rel;
+	AclResult	aclresult;
 	WalRcvExecResult *res;
 	char		originname[NAMEDATALEN];
 	RepOriginId originid;
@@ -1042,6 +1045,31 @@ 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("\"%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
diff --git 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"
@@ -189,6 +190,7 @@
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
+#include "utils/rls.h"
 #include "utils/syscache.h"
 #include "utils/timeout.h"
 
@@ -1530,6 +1532,38 @@ GetRelationIdentityOrPK(Relation rel)
 	return idxoid;
 }
 
+/*
+ * Check that we (the subscription owner) have sufficient privileges on the
+ * target relation to perform the given operation.
+ */
+static void
+TargetPrivilegesCheck(Relation rel, AclMode mode)
+{
+	Oid				relid;
+	AclResult		aclresult;
+
+	relid = RelationGetRelid(rel);
+	aclresult = pg_class_aclcheck(relid, GetUserId(), mode);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult,
+					   get_relkind_objtype(rel->rd_rel->relkind),
+					   get_rel_name(relid));
+
+	/*
+	 * We lack the infrastructure to honor RLS policies.  It might be possible
+	 * to add such infrastructure here, but tablesync workers lack it, too, so
+	 * we don't bother.  RLS does not ordinarily apply to TRUNCATE commands,
+	 * but it seems dangerous to replicate a TRUNCATE and then refuse to
+	 * replicate subsequent INSERTs, so we forbid all commands the same.
+	 */
+	if (check_enable_rls(relid, InvalidOid, false) == RLS_ENABLED)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("\"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
+						GetUserNameFromId(GetUserId(), true),
+						RelationGetRelationName(rel))));
+}
+
 /*
  * Handle INSERT message.
  */
@@ -1613,6 +1647,7 @@ apply_handle_insert_internal(ApplyExecutionData *edata,
 	ExecOpenIndices(relinfo, false);
 
 	/* Do the insert. */
+	TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_INSERT);
 	ExecSimpleRelationInsert(relinfo, estate, remoteslot);
 
 	/* Cleanup. */
@@ -1796,6 +1831,7 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		EvalPlanQualSetSlot(&epqstate, remoteslot);
 
 		/* Do the actual update. */
+		TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE);
 		ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot,
 								 remoteslot);
 	}
@@ -1917,6 +1953,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 		EvalPlanQualSetSlot(&epqstate, localslot);
 
 		/* Do the actual delete. */
+		TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_DELETE);
 		ExecSimpleRelationDelete(relinfo, estate, &epqstate, localslot);
 	}
 	else
@@ -2110,6 +2147,8 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 					ExecOpenIndices(partrelinfo, false);
 
 					EvalPlanQualSetSlot(&epqstate, remoteslot_part);
+					TargetPrivilegesCheck(partrelinfo->ri_RelationDesc,
+										  ACL_UPDATE);
 					ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate,
 											 localslot, remoteslot_part);
 					ExecCloseIndices(partrelinfo);
@@ -2236,6 +2275,7 @@ apply_handle_truncate(StringInfo s)
 		}
 
 		remote_rels = lappend(remote_rels, rel);
+		TargetPrivilegesCheck(rel->localrel, ACL_TRUNCATE);
 		rels = lappend(rels, rel->localrel);
 		relids = lappend_oid(relids, rel->localreloid);
 		if (RelationIsLogicallyLogged(rel->localrel))
@@ -2273,6 +2313,7 @@ apply_handle_truncate(StringInfo s)
 					continue;
 				}
 
+				TargetPrivilegesCheck(childrel, ACL_TRUNCATE);
 				rels = lappend(rels, childrel);
 				part_rels = lappend(part_rels, childrel);
 				relids = lappend_oid(relids, childrelid);
@@ -2915,6 +2956,7 @@ maybe_reread_subscription(void)
 		strcmp(newsub->slotname, MySubscription->slotname) != 0 ||
 		newsub->binary != MySubscription->binary ||
 		newsub->stream != MySubscription->stream ||
+		newsub->owner != MySubscription->owner ||
 		!equal(newsub->publications, MySubscription->publications))
 	{
 		ereport(LOG,
diff --git a/src/test/perl/PostgreSQL/Test/Cluster.pm b/src/test/perl/PostgreSQL/Test/Cluster.pm
@@ -2599,6 +2599,42 @@ sub wait_for_slot_catchup
 
 =pod
 
+=item $node->wait_for_log(regexp, offset)
+
+Waits for the contents of the server log file, starting at the given offset, to
+match the supplied regular expression.  Checks the entire log if no offset is
+given.  Times out after 180 seconds.
+
+If successful, returns the length of the entire log file, in bytes.
+
+=cut
+
+sub wait_for_log
+{
+	my ($self, $regexp, $offset) = @_;
+	$offset = 0 unless defined $offset;
+
+	my $max_attempts = 180 * 10;
+	my $attempts     = 0;
+
+	while ($attempts < $max_attempts)
+	{
+		my $log = PostgreSQL::Test::Utils::slurp_file($self->logfile, $offset);
+
+		return $offset+length($log) if ($log =~ m/$regexp/);
+
+		# Wait 0.1 second before retrying.
+		usleep(100_000);
+
+		$attempts++;
+	}
+
+	# The logs didn't match within 180 seconds. Give up.
+	croak "timed out waiting for match: $regexp";
+}
+
+=pod
+
 =item $node->query_hash($dbname, $query, @columns)
 
 Execute $query on $dbname, replacing any appearance of the string __COLUMNS__
diff --git a/src/test/subscription/t/027_nosuperuser.pl b/src/test/subscription/t/027_nosuperuser.pl

Original file line number	Diff line number	Diff line change
`@@ -1481,6 +1481,8 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)`
`1481`	`1481`
`1482`	`1482`	`InvokeObjectPostAlterHook(SubscriptionRelationId,`
`1483`	`1483`	`form->oid, 0);`
	`1484`	`+`
	`1485`	`+ ApplyLauncherWakeupAtCommit();`
`1484`	`1486`	`}`
`1485`	`1487`
`1486`	`1488`	`/*`