Support column-level privileges, as required by SQL standard.

Stephen Frost, with help from KaiGai Kohei and others

1 files changed, 157 insertions, 16 deletions

diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index 49ad0240397..eb98f470ee3 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/parser/parse_relation.c,v 1.140 2009/01/01 17:23:46 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/parser/parse_relation.c,v 1.141 2009/01/22 20:16:05 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -39,6 +39,8 @@ static RangeTblEntry *scanNameSpaceForRefname(ParseState *pstate,
 						const char *refname, int location);
 static RangeTblEntry *scanNameSpaceForRelid(ParseState *pstate, Oid relid,
 											int location);
+static void markRTEForSelectPriv(ParseState *pstate, RangeTblEntry *rte,
+								 int rtindex, AttrNumber col);
 static bool isLockedRel(ParseState *pstate, char *refname);
 static void expandRelation(Oid relid, Alias *eref,
 			   int rtindex, int sublevels_up,
@@ -435,14 +437,8 @@ GetCTEForRTE(ParseState *pstate, RangeTblEntry *rte, int rtelevelsup)
  *	  If found, return an appropriate Var node, else return NULL.
  *	  If the name proves ambiguous within this RTE, raise error.
  *
- * Side effect: if we find a match, mark the RTE as requiring read access.
- * See comments in setTargetTable().
- *
- * NOTE: if the RTE is for a join, marking it as requiring read access does
- * nothing.  It might seem that we need to propagate the mark to all the
- * contained RTEs, but that is not necessary.  This is so because a join
- * expression can only appear in a FROM clause, and any table named in
- * FROM will be marked as requiring read access from the beginning.
+ * Side effect: if we find a match, mark the RTE as requiring read access
+ * for the column.
  */
 Node *
 scanRTEForColumn(ParseState *pstate, RangeTblEntry *rte, char *colname,
@@ -450,6 +446,7 @@ scanRTEForColumn(ParseState *pstate, RangeTblEntry *rte, char *colname,
 {
 	Node	   *result = NULL;
 	int			attnum = 0;
+	Var		   *var;
 	ListCell   *c;
 
 	/*
@@ -476,9 +473,10 @@ scanRTEForColumn(ParseState *pstate, RangeTblEntry *rte, char *colname,
 						 errmsg("column reference \"%s\" is ambiguous",
 								colname),
 						 parser_errposition(pstate, location)));
-			result = (Node *) make_var(pstate, rte, attnum, location);
-			/* Require read access */
-			rte->requiredPerms |= ACL_SELECT;
+			var = make_var(pstate, rte, attnum, location);
+			/* Require read access to the column */
+			markVarForSelectPriv(pstate, var, rte);
+			result = (Node *) var;
 		}
 	}
 
@@ -504,9 +502,10 @@ scanRTEForColumn(ParseState *pstate, RangeTblEntry *rte, char *colname,
 									 Int16GetDatum(attnum),
 									 0, 0))
 			{
-				result = (Node *) make_var(pstate, rte, attnum, location);
-				/* Require read access */
-				rte->requiredPerms |= ACL_SELECT;
+				var = make_var(pstate, rte, attnum, location);
+				/* Require read access to the column */
+				markVarForSelectPriv(pstate, var, rte);
+				result = (Node *) var;
 			}
 		}
 	}
@@ -595,6 +594,122 @@ qualifiedNameToVar(ParseState *pstate,
 }
 
 /*
+ * markRTEForSelectPriv
+ *	   Mark the specified column of an RTE as requiring SELECT privilege
+ *
+ * col == InvalidAttrNumber means a "whole row" reference
+ *
+ * The caller should pass the actual RTE if it has it handy; otherwise pass
+ * NULL, and we'll look it up here.  (This uglification of the API is
+ * worthwhile because nearly all external callers have the RTE at hand.)
+ */
+static void
+markRTEForSelectPriv(ParseState *pstate, RangeTblEntry *rte,
+					 int rtindex, AttrNumber col)
+{
+	if (rte == NULL)
+		rte = rt_fetch(rtindex, pstate->p_rtable);
+
+	if (rte->rtekind == RTE_RELATION)
+	{
+		/* Make sure the rel as a whole is marked for SELECT access */
+		rte->requiredPerms |= ACL_SELECT;
+		/* Must offset the attnum to fit in a bitmapset */
+		rte->selectedCols = bms_add_member(rte->selectedCols,
+									col - FirstLowInvalidHeapAttributeNumber);
+	}
+	else if (rte->rtekind == RTE_JOIN)
+	{
+		if (col == InvalidAttrNumber)
+		{
+			/*
+			 * A whole-row reference to a join has to be treated as
+			 * whole-row references to the two inputs.
+			 */
+			JoinExpr   *j;
+
+			if (rtindex > 0 && rtindex <= list_length(pstate->p_joinexprs))
+				j = (JoinExpr *) list_nth(pstate->p_joinexprs, rtindex - 1);
+			else
+				j = NULL;
+			if (j == NULL)
+				elog(ERROR, "could not find JoinExpr for whole-row reference");
+			Assert(IsA(j, JoinExpr));
+
+			/* Note: we can't see FromExpr here */
+			if (IsA(j->larg, RangeTblRef))
+			{
+				int		varno = ((RangeTblRef *) j->larg)->rtindex;
+
+				markRTEForSelectPriv(pstate, NULL, varno, InvalidAttrNumber);
+			}
+			else if (IsA(j->larg, JoinExpr))
+			{
+				int		varno = ((JoinExpr *) j->larg)->rtindex;
+
+				markRTEForSelectPriv(pstate, NULL, varno, InvalidAttrNumber);
+			}
+			else
+				elog(ERROR, "unrecognized node type: %d",
+					 (int) nodeTag(j->larg));
+			if (IsA(j->rarg, RangeTblRef))
+			{
+				int		varno = ((RangeTblRef *) j->rarg)->rtindex;
+
+				markRTEForSelectPriv(pstate, NULL, varno, InvalidAttrNumber);
+			}
+			else if (IsA(j->rarg, JoinExpr))
+			{
+				int		varno = ((JoinExpr *) j->rarg)->rtindex;
+
+				markRTEForSelectPriv(pstate, NULL, varno, InvalidAttrNumber);
+			}
+			else
+				elog(ERROR, "unrecognized node type: %d",
+					 (int) nodeTag(j->rarg));
+		}
+		else
+		{
+			/*
+			 * Regular join attribute, look at the alias-variable list.
+			 *
+			 * The aliasvar could be either a Var or a COALESCE expression,
+			 * but in the latter case we should already have marked the two
+			 * referent variables as being selected, due to their use in the
+			 * JOIN clause.  So we need only be concerned with the simple
+			 * Var case.
+			 */
+			Var	   *aliasvar;
+
+			Assert(col > 0 && col <= list_length(rte->joinaliasvars));
+			aliasvar = (Var *) list_nth(rte->joinaliasvars, col - 1);
+			if (IsA(aliasvar, Var))
+				markVarForSelectPriv(pstate, aliasvar, NULL);
+		}
+	}
+	/* other RTE types don't require privilege marking */
+}
+
+/*
+ * markVarForSelectPriv
+ *	   Mark the RTE referenced by a Var as requiring SELECT privilege
+ *
+ * The caller should pass the Var's referenced RTE if it has it handy
+ * (nearly all do); otherwise pass NULL.
+ */
+void
+markVarForSelectPriv(ParseState *pstate, Var *var, RangeTblEntry *rte)
+{
+	Index	lv;
+
+	Assert(IsA(var, Var));
+	/* Find the appropriate pstate if it's an uplevel Var */
+	for (lv = 0; lv < var->varlevelsup; lv++)
+		pstate = pstate->parentParseState;
+	markRTEForSelectPriv(pstate, rte, var->varno, var->varattno);
+}
+
+/*
  * buildRelationAliases
  *		Construct the eref column name list for a relation RTE.
  *		This code is also used for the case of a function RTE returning
@@ -838,6 +953,8 @@ addRangeTableEntry(ParseState *pstate,
 
 	rte->requiredPerms = ACL_SELECT;
 	rte->checkAsUser = InvalidOid;		/* not set-uid by default, either */
+	rte->selectedCols = NULL;
+	rte->modifiedCols = NULL;
 
 	/*
 	 * Add completed RTE to pstate's range table list, but not to join list
@@ -891,6 +1008,8 @@ addRangeTableEntryForRelation(ParseState *pstate,
 
 	rte->requiredPerms = ACL_SELECT;
 	rte->checkAsUser = InvalidOid;		/* not set-uid by default, either */
+	rte->selectedCols = NULL;
+	rte->modifiedCols = NULL;
 
 	/*
 	 * Add completed RTE to pstate's range table list, but not to join list
@@ -969,6 +1088,8 @@ addRangeTableEntryForSubquery(ParseState *pstate,
 
 	rte->requiredPerms = 0;
 	rte->checkAsUser = InvalidOid;
+	rte->selectedCols = NULL;
+	rte->modifiedCols = NULL;
 
 	/*
 	 * Add completed RTE to pstate's range table list, but not to join list
@@ -1101,6 +1222,8 @@ addRangeTableEntryForFunction(ParseState *pstate,
 
 	rte->requiredPerms = 0;
 	rte->checkAsUser = InvalidOid;
+	rte->selectedCols = NULL;
+	rte->modifiedCols = NULL;
 
 	/*
 	 * Add completed RTE to pstate's range table list, but not to join list
@@ -1168,8 +1291,11 @@ addRangeTableEntryForValues(ParseState *pstate,
 	 */
 	rte->inh = false;			/* never true for values RTEs */
 	rte->inFromCl = inFromCl;
+
 	rte->requiredPerms = 0;
 	rte->checkAsUser = InvalidOid;
+	rte->selectedCols = NULL;
+	rte->modifiedCols = NULL;
 
 	/*
 	 * Add completed RTE to pstate's range table list, but not to join list
@@ -1239,6 +1365,8 @@ addRangeTableEntryForJoin(ParseState *pstate,
 
 	rte->requiredPerms = 0;
 	rte->checkAsUser = InvalidOid;
+	rte->selectedCols = NULL;
+	rte->modifiedCols = NULL;
 
 	/*
 	 * Add completed RTE to pstate's range table list, but not to join list
@@ -1320,6 +1448,8 @@ addRangeTableEntryForCTE(ParseState *pstate,
 
 	rte->requiredPerms = 0;
 	rte->checkAsUser = InvalidOid;
+	rte->selectedCols = NULL;
+	rte->modifiedCols = NULL;
 
 	/*
 	 * Add completed RTE to pstate's range table list, but not to join list
@@ -1803,6 +1933,7 @@ expandTupleDesc(TupleDesc tupdesc, Alias *eref,
  * As with expandRTE, rtindex/sublevels_up determine the varno/varlevelsup
  * fields of the Vars produced, and location sets their location.
  * pstate->p_next_resno determines the resnos assigned to the TLEs.
+ * The referenced columns are marked as requiring SELECT access.
  */
 List *
 expandRelAttrs(ParseState *pstate, RangeTblEntry *rte,
@@ -1817,10 +1948,17 @@ expandRelAttrs(ParseState *pstate, RangeTblEntry *rte,
 	expandRTE(rte, rtindex, sublevels_up, location, false,
 			  &names, &vars);
 
+	/*
+	 * Require read access to the table.  This is normally redundant with the
+	 * markVarForSelectPriv calls below, but not if the table has zero
+	 * columns.
+	 */
+	rte->requiredPerms |= ACL_SELECT;
+
 	forboth(name, names, var, vars)
 	{
 		char	   *label = strVal(lfirst(name));
-		Node	   *varnode = (Node *) lfirst(var);
+		Var		   *varnode = (Var *) lfirst(var);
 		TargetEntry *te;
 
 		te = makeTargetEntry((Expr *) varnode,
@@ -1828,6 +1966,9 @@ expandRelAttrs(ParseState *pstate, RangeTblEntry *rte,
 							 label,
 							 false);
 		te_list = lappend(te_list, te);
+
+		/* Require read access to each column */
+		markVarForSelectPriv(pstate, varnode, rte);
 	}
 
 	Assert(name == NULL && var == NULL);	/* lists not the same length? */