Perform permission checks in functions manipulating distributed tables.

Previously several commands, amongst them commands like
master_create_distributed_table(), were allowed for everyone. That's not
good: Even though citus currently requires superuser permissions, we
shouldn't allow non-superusers to perform actions as sensitive as making
a table distributed.

There's no checks on the worker_* functions, as these usually just punt
the action to underlying postgres functionality, which then perform the
necessary checks.

src/backend/distributed/commands/create_distributed_table.c

@@ -98,6 +98,8 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
 	distributedRelation = relation_open(distributedRelationId, AccessExclusiveLock);
 	distributedRelationName = RelationGetRelationName(distributedRelation);
 
+	EnsureTableOwner(distributedRelationId);
+
 	/* open system catalog and insert new tuple */
 	pgDistPartition = heap_open(DistPartitionRelationId(), RowExclusiveLock);
 

src/backend/distributed/master/master_create_shards.c

@@ -88,6 +88,14 @@ master_create_worker_shards(PG_FUNCTION_ARGS)
 	/* make sure table is hash partitioned */
 	CheckHashPartitionedTable(distributedTableId);
 
+	/*
+	 * In contrast to append/range partitioned tables it makes more sense to
+	 * require ownership privileges - shards for hash-partitioned tables are
+	 * only created once, not continually during ingest as for the other
+	 * partitioning types.
+	 */
+	EnsureTableOwner(distributedTableId);
+
 	/* we plan to add shards: get an exclusive metadata lock */
 	LockRelationDistributionMetadata(distributedTableId, ExclusiveLock);
 

src/backend/distributed/master/master_delete_protocol.c

@@ -112,7 +112,9 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
 	schemaName = deleteStatement->relation->schemaname;
 	relationName = deleteStatement->relation->relname;
 	relationId = RangeVarGetRelid(deleteStatement->relation, NoLock, failOK);
+
 	CheckDistributedTable(relationId);
+	EnsureTablePermissions(relationId, ACL_DELETE);
 
 	queryTreeList = pg_analyze_and_rewrite(queryTreeNode, queryString, NULL, 0);
 	deleteQuery = (Query *) linitial(queryTreeList);
@@ -187,12 +189,29 @@ master_drop_all_shards(PG_FUNCTION_ARGS)
 		/* ensure proper values are used if the table exists */
 		Oid schemaId = get_rel_namespace(relationId);
 		schemaName = get_namespace_name(schemaId);
+
+		/*
+		 * Only allow the owner to drop all shards, this is more akin to DDL
+		 * than DELETE.
+		 */
+		EnsureTableOwner(relationId);
 	}
 	else
 	{
 		/* table has been dropped, rely on user-supplied values */
 		schemaName = text_to_cstring(schemaNameText);
 		relationName = text_to_cstring(relationNameText);
+
+		/*
+		 * Verify that this only is run as superuser - that's how it's used in
+		 * our drop event trigger, and we can't verify permissions for an
+		 * already dropped relation.
+		 */
+		if (!superuser())
+		{
+			ereport(ERROR, (errmsg("cannot drop all shards of a dropped table as "
+								   "non-superuser")));
+		}
 	}
 
 	shardIntervalList = LoadShardIntervalList(relationId);

src/backend/distributed/master/master_metadata_utility.c

@@ -12,6 +12,7 @@
 
 #include "postgres.h"
 #include "funcapi.h"
+#include "miscadmin.h"
 
 #include "access/htup_details.h"
 #include "access/xact.h"
@@ -28,6 +29,7 @@
 #include "distributed/worker_manager.h"
 #include "nodes/makefuncs.h"
 #include "parser/scansup.h"
+#include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/datum.h"
 #include "utils/fmgroids.h"
@@ -611,3 +613,37 @@ BuildDistributionKeyFromColumnName(Relation distributedRelation, char *columnNam
 
 	return (Node *) column;
 }
+
+
+/*
+ * Check that the current user has `mode` permissions on relationId, error out
+ * if not. Superusers always have such permissions.
+ */
+void
+EnsureTablePermissions(Oid relationId, AclMode mode)
+{
+	AclResult aclresult;
+
+	aclresult = pg_class_aclcheck(relationId, GetUserId(), mode);
+
+	if (aclresult != ACLCHECK_OK)
+	{
+		aclcheck_error(aclresult, ACL_KIND_CLASS,
+					   get_rel_name(relationId));
+	}
+}
+
+
+/*
+ * Check that the current user has owner rights to relationId, error out if
+ * not. Superusers are regarded as owners.
+ */
+void
+EnsureTableOwner(Oid relationId)
+{
+	if (!pg_class_ownercheck(relationId, GetUserId()))
+	{
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
+					   get_rel_name(relationId));
+	}
+}

src/backend/distributed/master/master_repair_shards.c

@@ -76,6 +76,8 @@ master_copy_shard_placement(PG_FUNCTION_ARGS)
 	bool dataCopied = false;
 	char relationKind = '\0';
 
+	EnsureTableOwner(distributedTableId);
+
 	/*
 	 * By taking an exclusive lock on the shard, we both stop all modifications
 	 * (INSERT, UPDATE, or DELETE) and prevent concurrent repair operations from

src/backend/distributed/master/master_stage_protocol.c

@@ -82,6 +82,8 @@ master_create_empty_shard(PG_FUNCTION_ARGS)
 	char storageType = SHARD_STORAGE_TABLE;
 
 	Oid relationId = ResolveRelationId(relationNameText);
+
+	EnsureTablePermissions(relationId, ACL_INSERT);
 	CheckDistributedTable(relationId);
 
 	if (CStoreTable(relationId))
@@ -170,6 +172,9 @@ master_append_table_to_shard(PG_FUNCTION_ARGS)
 	bool cstoreTable = CStoreTable(relationId);
 
 	char storageType = shardInterval->storageType;
+
+	EnsureTablePermissions(relationId, ACL_INSERT);
+
 	if (storageType != SHARD_STORAGE_TABLE && !cstoreTable)
 	{
 		ereport(ERROR, (errmsg("cannot append to shardId " UINT64_FORMAT, shardId),

src/backend/distributed/utils/metadata_cache.c

@@ -669,6 +669,9 @@ CurrentUserName(void)
  * master_dist_partition_cache_invalidate is a trigger function that performs
  * relcache invalidations when the contents of pg_dist_partition are changed
  * on the SQL level.
+ *
+ * NB: We decided there is little point in checking permissions here, there
+ * are much easier ways to waste CPU than causing cache invalidations.
  */
 Datum
 master_dist_partition_cache_invalidate(PG_FUNCTION_ARGS)
@@ -727,6 +730,9 @@ master_dist_partition_cache_invalidate(PG_FUNCTION_ARGS)
  * master_dist_shard_cache_invalidate is a trigger function that performs
  * relcache invalidations when the contents of pg_dist_shard are changed
  * on the SQL level.
+ *
+ * NB: We decided there is little point in checking permissions here, there
+ * are much easier ways to waste CPU than causing cache invalidations.
  */
 Datum
 master_dist_shard_cache_invalidate(PG_FUNCTION_ARGS)

src/include/distributed/master_metadata_utility.h

@@ -18,6 +18,7 @@
 #include "access/tupdesc.h"
 #include "distributed/citus_nodes.h"
 #include "distributed/relay_utility.h"
+#include "utils/acl.h"
 #include "utils/relcache.h"
 
 
@@ -77,5 +78,7 @@ extern void DeleteShardPlacementRow(uint64 shardId, char *workerName, uint32 wor
 /* Remaining metadata utility functions  */
 extern Node * BuildDistributionKeyFromColumnName(Relation distributedRelation,
 												 char *columnName);
+extern void EnsureTablePermissions(Oid relationId, AclMode mode);
+extern void EnsureTableOwner(Oid relationId);
 
 #endif   /* MASTER_METADATA_UTILITY_H */