Centralize DML permissions-checking logic.

Remove bespoke code in DoCopy and RI_Initial_Check, which now instead
fabricate call ExecCheckRTPerms with a manufactured RangeTblEntry.
This is intended to make it feasible for an enhanced security provider
to actually make use of ExecutorCheckPerms_hook, but also has the
advantage that RI_Initial_Check can allow use of the fast-path when
column-level but not table-level permissions are present.

KaiGai Kohei.  Reviewed (in an earlier version) by Stephen Frost, and by me.
Some further changes to the comments by me.

src/backend/commands/copy.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/copy.c,v 1.327 2010/04/28 16:10:41 heikki Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/copy.c,v 1.328 2010/07/22 00:47:52 rhaas Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -21,6 +21,7 @@
 #include <arpa/inet.h>
 
 #include "access/heapam.h"
+#include "access/sysattr.h"
 #include "access/xact.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_type.h"
@@ -726,8 +727,6 @@ DoCopy(const CopyStmt *stmt, const char *queryString)
 	bool		force_quote_all = false;
 	bool		format_specified = false;
 	AclMode		required_access = (is_from ? ACL_INSERT : ACL_SELECT);
-	AclMode		relPerms;
-	AclMode		remainingPerms;
 	ListCell   *option;
 	TupleDesc	tupDesc;
 	int			num_phys_attrs;
@@ -988,6 +987,10 @@ DoCopy(const CopyStmt *stmt, const char *queryString)
 
 	if (stmt->relation)
 	{
+		RangeTblEntry  *rte;
+		List		   *attnums;
+		ListCell	   *cur;
+
 		Assert(!stmt->query);
 		cstate->queryDesc = NULL;
 
@@ -998,28 +1001,22 @@ DoCopy(const CopyStmt *stmt, const char *queryString)
 		tupDesc = RelationGetDescr(cstate->rel);
 
 		/* Check relation permissions. */
-		relPerms = pg_class_aclmask(RelationGetRelid(cstate->rel), GetUserId(),
-									required_access, ACLMASK_ALL);
-		remainingPerms = required_access & ~relPerms;
-		if (remainingPerms != 0)
-		{
-			/* We don't have table permissions, check per-column permissions */
-			List	   *attnums;
-			ListCell   *cur;
+		rte = makeNode(RangeTblEntry);
+		rte->rtekind = RTE_RELATION;
+		rte->relid = RelationGetRelid(cstate->rel);
+		rte->requiredPerms = required_access;
 
-			attnums = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
-			foreach(cur, attnums)
-			{
-				int			attnum = lfirst_int(cur);
+		attnums = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
+		foreach (cur, attnums)
+		{
+			int		attno = lfirst_int(cur) - FirstLowInvalidHeapAttributeNumber;
 
-				if (pg_attribute_aclcheck(RelationGetRelid(cstate->rel),
-										  attnum,
-										  GetUserId(),
-										  remainingPerms) != ACLCHECK_OK)
-					aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-								   RelationGetRelationName(cstate->rel));
-			}
+			if (is_from)
+				rte->modifiedCols = bms_add_member(rte->modifiedCols, attno);
+			else
+				rte->selectedCols = bms_add_member(rte->selectedCols, attno);
 		}
+		ExecCheckRTPerms(list_make1(rte), true);
 
 		/* check read-only transaction */
 		if (XactReadOnly && is_from && !cstate->rel->rd_islocaltemp)

src/backend/executor/execMain.c

@@ -26,7 +26,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.351 2010/07/12 17:01:05 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.352 2010/07/22 00:47:52 rhaas Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -75,8 +75,7 @@ static void ExecutePlan(EState *estate, PlanState *planstate,
 			long numberTuples,
 			ScanDirection direction,
 			DestReceiver *dest);
-static void ExecCheckRTPerms(List *rangeTable);
-static void ExecCheckRTEPerms(RangeTblEntry *rte);
+static bool ExecCheckRTEPerms(RangeTblEntry *rte);
 static void ExecCheckXactReadOnly(PlannedStmt *plannedstmt);
 static void EvalPlanQualStart(EPQState *epqstate, EState *parentestate,
 				  Plan *planTree);
@@ -409,26 +408,42 @@ ExecutorRewind(QueryDesc *queryDesc)
 /*
  * ExecCheckRTPerms
  *		Check access permissions for all relations listed in a range table.
+ *
+ * Returns true if permissions are adequate.  Otherwise, throws an appropriate
+ * error if ereport_on_violation is true, or simply returns false otherwise.
  */
-static void
-ExecCheckRTPerms(List *rangeTable)
+bool
+ExecCheckRTPerms(List *rangeTable, bool ereport_on_violation)
 {
 	ListCell   *l;
+	bool		result = true;
 
 	foreach(l, rangeTable)
 	{
-		ExecCheckRTEPerms((RangeTblEntry *) lfirst(l));
+		RangeTblEntry  *rte = (RangeTblEntry *) lfirst(l);
+
+		result = ExecCheckRTEPerms(rte);
+		if (!result)
+		{
+			Assert(rte->rtekind == RTE_RELATION);
+			if (ereport_on_violation)
+				aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
+							   get_rel_name(rte->relid));
+			return false;
+		}
 	}
 
 	if (ExecutorCheckPerms_hook)
-		(*ExecutorCheckPerms_hook)(rangeTable);
+		result = (*ExecutorCheckPerms_hook)(rangeTable,
+											ereport_on_violation);
+	return result;
 }
 
 /*
  * ExecCheckRTEPerms
  *		Check access permissions for a single RTE.
  */
-static void
+static bool
 ExecCheckRTEPerms(RangeTblEntry *rte)
 {
 	AclMode		requiredPerms;
@@ -445,14 +460,14 @@ ExecCheckRTEPerms(RangeTblEntry *rte)
 	 * Join, subquery, and special RTEs need no checks.
 	 */
 	if (rte->rtekind != RTE_RELATION)
-		return;
+		return true;
 
 	/*
 	 * No work if requiredPerms is empty.
 	 */
 	requiredPerms = rte->requiredPerms;
 	if (requiredPerms == 0)
-		return;
+		return true;
 
 	relOid = rte->relid;
 
@@ -480,8 +495,7 @@ ExecCheckRTEPerms(RangeTblEntry *rte)
 		 * we can fail straight away.
 		 */
 		if (remainingPerms & ~(ACL_SELECT | ACL_INSERT | ACL_UPDATE))
-			aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-						   get_rel_name(relOid));
+			return false;
 
 		/*
 		 * Check to see if we have the needed privileges at column level.
@@ -501,8 +515,7 @@ ExecCheckRTEPerms(RangeTblEntry *rte)
 			{
 				if (pg_attribute_aclcheck_all(relOid, userid, ACL_SELECT,
 											  ACLMASK_ANY) != ACLCHECK_OK)
-					aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-								   get_rel_name(relOid));
+					return false;
 			}
 
 			tmpset = bms_copy(rte->selectedCols);
@@ -515,15 +528,13 @@ ExecCheckRTEPerms(RangeTblEntry *rte)
 					/* Whole-row reference, must have priv on all cols */
 					if (pg_attribute_aclcheck_all(relOid, userid, ACL_SELECT,
 												  ACLMASK_ALL) != ACLCHECK_OK)
-						aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-									   get_rel_name(relOid));
+						return false;
 				}
 				else
 				{
-					if (pg_attribute_aclcheck(relOid, col, userid, ACL_SELECT)
-						!= ACLCHECK_OK)
-						aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-									   get_rel_name(relOid));
+					if (pg_attribute_aclcheck(relOid, col, userid,
+											  ACL_SELECT) != ACLCHECK_OK)
+						return false;
 				}
 			}
 			bms_free(tmpset);
@@ -546,8 +557,7 @@ ExecCheckRTEPerms(RangeTblEntry *rte)
 			{
 				if (pg_attribute_aclcheck_all(relOid, userid, remainingPerms,
 											  ACLMASK_ANY) != ACLCHECK_OK)
-					aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-								   get_rel_name(relOid));
+					return false;
 			}
 
 			tmpset = bms_copy(rte->modifiedCols);
@@ -562,15 +572,15 @@ ExecCheckRTEPerms(RangeTblEntry *rte)
 				}
 				else
 				{
-					if (pg_attribute_aclcheck(relOid, col, userid, remainingPerms)
-						!= ACLCHECK_OK)
-						aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-									   get_rel_name(relOid));
+					if (pg_attribute_aclcheck(relOid, col, userid,
+											  remainingPerms) != ACLCHECK_OK)
+						return false;
 				}
 			}
 			bms_free(tmpset);
 		}
 	}
+	return true;
 }
 
 /*
@@ -636,7 +646,7 @@ InitPlan(QueryDesc *queryDesc, int eflags)
 	/*
 	 * Do permissions checks
 	 */
-	ExecCheckRTPerms(rangeTable);
+	ExecCheckRTPerms(rangeTable, true);
 
 	/*
 	 * initialize the node's execution state

src/backend/utils/adt/ri_triggers.c

@@ -15,7 +15,7 @@
  *
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/backend/utils/adt/ri_triggers.c,v 1.118 2010/02/14 18:42:16 rhaas Exp $
+ * $PostgreSQL: pgsql/src/backend/utils/adt/ri_triggers.c,v 1.119 2010/07/22 00:47:52 rhaas Exp $
  *
  * ----------
  */
@@ -31,10 +31,12 @@
 #include "postgres.h"
 
 #include "access/xact.h"
+#include "access/sysattr.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_operator.h"
 #include "catalog/pg_type.h"
 #include "commands/trigger.h"
+#include "executor/executor.h"
 #include "executor/spi.h"
 #include "parser/parse_coerce.h"
 #include "parser/parse_relation.h"
@@ -2624,6 +2626,8 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
 	char		fkrelname[MAX_QUOTED_REL_NAME_LEN];
 	char		pkattname[MAX_QUOTED_NAME_LEN + 3];
 	char		fkattname[MAX_QUOTED_NAME_LEN + 3];
+	RangeTblEntry  *pkrte;
+	RangeTblEntry  *fkrte;
 	const char *sep;
 	int			i;
 	int			old_work_mem;
@@ -2631,6 +2635,9 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
 	int			spi_result;
 	SPIPlanPtr	qplan;
 
+	/* Fetch constraint info. */
+	ri_FetchConstraintInfo(&riinfo, trigger, fk_rel, false);
+
 	/*
 	 * Check to make sure current user has enough permissions to do the test
 	 * query.  (If not, caller can fall back to the trigger method, which
@@ -2638,12 +2645,29 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
 	 *
 	 * XXX are there any other show-stopper conditions to check?
 	 */
-	if (pg_class_aclcheck(RelationGetRelid(fk_rel), GetUserId(), ACL_SELECT) != ACLCHECK_OK)
-		return false;
-	if (pg_class_aclcheck(RelationGetRelid(pk_rel), GetUserId(), ACL_SELECT) != ACLCHECK_OK)
-		return false;
+	pkrte = makeNode(RangeTblEntry);
+	pkrte->rtekind = RTE_RELATION;
+	pkrte->relid = RelationGetRelid(pk_rel);
+	pkrte->requiredPerms = ACL_SELECT;
 
-	ri_FetchConstraintInfo(&riinfo, trigger, fk_rel, false);
+	fkrte = makeNode(RangeTblEntry);
+	fkrte->rtekind = RTE_RELATION;
+	fkrte->relid = RelationGetRelid(fk_rel);
+	fkrte->requiredPerms = ACL_SELECT;
+
+	for (i = 0; i < riinfo.nkeys; i++)
+	{
+		int		attno;
+
+		attno = riinfo.pk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
+		pkrte->selectedCols = bms_add_member(pkrte->selectedCols, attno);
+
+		attno = riinfo.fk_attnums[i] - FirstLowInvalidHeapAttributeNumber;
+		fkrte->selectedCols = bms_add_member(fkrte->selectedCols, attno);
+	}
+
+	if (!ExecCheckRTPerms(list_make2(fkrte, pkrte), false))
+		return false;
 
 	/*----------
 	 * The query string built is:

src/include/executor/executor.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/executor/executor.h,v 1.170 2010/07/12 17:01:06 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/executor/executor.h,v 1.171 2010/07/22 00:47:59 rhaas Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -75,7 +75,7 @@ typedef void (*ExecutorEnd_hook_type) (QueryDesc *queryDesc);
 extern PGDLLIMPORT ExecutorEnd_hook_type ExecutorEnd_hook;
 
 /* Hook for plugins to get control in ExecCheckRTPerms() */
-typedef void (*ExecutorCheckPerms_hook_type) (List *);
+typedef bool (*ExecutorCheckPerms_hook_type) (List *, bool);
 extern PGDLLIMPORT ExecutorCheckPerms_hook_type ExecutorCheckPerms_hook;
 
 
@@ -161,6 +161,7 @@ extern void standard_ExecutorRun(QueryDesc *queryDesc,
 extern void ExecutorEnd(QueryDesc *queryDesc);
 extern void standard_ExecutorEnd(QueryDesc *queryDesc);
 extern void ExecutorRewind(QueryDesc *queryDesc);
+extern bool ExecCheckRTPerms(List *rangeTable, bool ereport_on_violation);
 extern void InitResultRelInfo(ResultRelInfo *resultRelInfo,
 				  Relation resultRelationDesc,
 				  Index resultRelationIndex,