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

Stephen Frost, with help from KaiGai Kohei and others

src/backend/executor/execMain.c

@@ -26,7 +26,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.320 2009/01/01 17:23:41 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.321 2009/01/22 20:16:03 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -34,6 +34,7 @@
 
 #include "access/heapam.h"
 #include "access/reloptions.h"
+#include "access/sysattr.h"
 #include "access/transam.h"
 #include "access/xact.h"
 #include "catalog/heap.h"
@@ -453,8 +454,12 @@ static void
 ExecCheckRTEPerms(RangeTblEntry *rte)
 {
 	AclMode		requiredPerms;
+	AclMode		relPerms;
+	AclMode		remainingPerms;
 	Oid			relOid;
 	Oid			userid;
+	Bitmapset  *tmpset;
+	int			col;
 
 	/*
 	 * Only plain-relation RTEs need to be checked here.  Function RTEs are
@@ -484,12 +489,110 @@ ExecCheckRTEPerms(RangeTblEntry *rte)
 	userid = rte->checkAsUser ? rte->checkAsUser : GetUserId();
 
 	/*
-	 * We must have *all* the requiredPerms bits, so use aclmask not aclcheck.
+	 * We must have *all* the requiredPerms bits, but some of the bits can be
+	 * satisfied from column-level rather than relation-level permissions.
+	 * First, remove any bits that are satisfied by relation permissions.
 	 */
-	if (pg_class_aclmask(relOid, userid, requiredPerms, ACLMASK_ALL)
-		!= requiredPerms)
-		aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
-					   get_rel_name(relOid));
+	relPerms = pg_class_aclmask(relOid, userid, requiredPerms, ACLMASK_ALL);
+	remainingPerms = requiredPerms & ~relPerms;
+	if (remainingPerms != 0)
+	{
+		/*
+		 * If we lack any permissions that exist only as relation permissions,
+		 * 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));
+
+		/*
+		 * Check to see if we have the needed privileges at column level.
+		 *
+		 * Note: failures just report a table-level error; it would be nicer
+		 * to report a column-level error if we have some but not all of the
+		 * column privileges.
+		 */
+		if (remainingPerms & ACL_SELECT)
+		{
+			/*
+			 * When the query doesn't explicitly reference any columns (for
+			 * example, SELECT COUNT(*) FROM table), allow the query if we
+			 * have SELECT on any column of the rel, as per SQL spec.
+			 */
+			if (bms_is_empty(rte->selectedCols))
+			{
+				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));
+			}
+
+			tmpset = bms_copy(rte->selectedCols);
+			while ((col = bms_first_member(tmpset)) >= 0)
+			{
+				/* remove the column number offset */
+				col += FirstLowInvalidHeapAttributeNumber;
+				if (col == InvalidAttrNumber)
+				{
+					/* 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));
+				}
+				else
+				{
+					if (pg_attribute_aclcheck(relOid, col, userid, ACL_SELECT)
+						!= ACLCHECK_OK)
+						aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
+									   get_rel_name(relOid));
+				}
+			}
+			bms_free(tmpset);
+		}
+
+		/*
+		 * Basically the same for the mod columns, with either INSERT or UPDATE
+		 * privilege as specified by remainingPerms.
+		 */
+		remainingPerms &= ~ACL_SELECT;
+		if (remainingPerms != 0)
+		{
+			/*
+			 * When the query doesn't explicitly change any columns, allow
+			 * the query if we have permission on any column of the rel.  This
+			 * is to handle SELECT FOR UPDATE as well as possible corner cases
+			 * in INSERT and UPDATE.
+			 */
+			if (bms_is_empty(rte->modifiedCols))
+			{
+				if (pg_attribute_aclcheck_all(relOid, userid, remainingPerms,
+											  ACLMASK_ANY) != ACLCHECK_OK)
+					aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
+								   get_rel_name(relOid));
+			}
+
+			tmpset = bms_copy(rte->modifiedCols);
+			while ((col = bms_first_member(tmpset)) >= 0)
+			{
+				/* remove the column number offset */
+				col += FirstLowInvalidHeapAttributeNumber;
+				if (col == InvalidAttrNumber)
+				{
+					/* whole-row reference can't happen here */
+					elog(ERROR, "whole-row update is not implemented");
+				}
+				else
+				{
+					if (pg_attribute_aclcheck(relOid, col, userid, remainingPerms)
+						!= ACLCHECK_OK)
+						aclcheck_error(ACLCHECK_NO_PRIV, ACL_KIND_CLASS,
+									   get_rel_name(relOid));
+				}
+			}
+			bms_free(tmpset);
+		}
+	}
 }
 
 /*