4 files changed, 84 insertions, 37 deletions

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index d592655258a..599e160ca64 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -142,6 +142,15 @@ static relopt_bool boolRelOpts[] =
 	},
 	{
 		{
+			"security_invoker",
+			"Privileges on underlying relations are checked as the invoking user, not the view owner",
+			RELOPT_KIND_VIEW,
+			AccessExclusiveLock
+		},
+		false
+	},
+	{
+		{
 			"vacuum_truncate",
 			"Enables vacuum to truncate empty pages at the end of this table",
 			RELOPT_KIND_HEAP | RELOPT_KIND_TOAST,
@@ -1996,6 +2005,8 @@ view_reloptions(Datum reloptions, bool validate)
 	static const relopt_parse_elt tab[] = {
 		{"security_barrier", RELOPT_TYPE_BOOL,
 		offsetof(ViewOptions, security_barrier)},
+		{"security_invoker", RELOPT_TYPE_BOOL,
+		offsetof(ViewOptions, security_invoker)},
 		{"check_option", RELOPT_TYPE_ENUM,
 		offsetof(ViewOptions, check_option)}
 	};
diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c
index 4b3f79704f8..b97b8b0435c 100644
--- a/src/backend/commands/lockcmds.c
+++ b/src/backend/commands/lockcmds.c
@@ -169,7 +169,7 @@ typedef struct
 {
 	LOCKMODE	lockmode;		/* lock mode to use */
 	bool		nowait;			/* no wait mode */
-	Oid			viewowner;		/* view owner for checking the privilege */
+	Oid			check_as_user;	/* user for checking the privilege */
 	Oid			viewoid;		/* OID of the view to be locked */
 	List	   *ancestor_views; /* OIDs of ancestor views */
 } LockViewRecurse_context;
@@ -215,8 +215,12 @@ LockViewRecurse_walker(Node *node, LockViewRecurse_context *context)
 			if (list_member_oid(context->ancestor_views, relid))
 				continue;
 
-			/* Check permissions with the view owner's privilege. */
-			aclresult = LockTableAclCheck(relid, context->lockmode, context->viewowner);
+			/*
+			 * Check permissions as the specified user.  This will either be
+			 * the view owner or the current user.
+			 */
+			aclresult = LockTableAclCheck(relid, context->lockmode,
+										  context->check_as_user);
 			if (aclresult != ACLCHECK_OK)
 				aclcheck_error(aclresult, get_relkind_objtype(relkind), relname);
 
@@ -259,9 +263,16 @@ LockViewRecurse(Oid reloid, LOCKMODE lockmode, bool nowait,
 	view = table_open(reloid, NoLock);
 	viewquery = get_view_query(view);
 
+	/*
+	 * If the view has the security_invoker property set, check permissions as
+	 * the current user.  Otherwise, check permissions as the view owner.
+	 */
 	context.lockmode = lockmode;
 	context.nowait = nowait;
-	context.viewowner = view->rd_rel->relowner;
+	if (RelationHasSecurityInvoker(view))
+		context.check_as_user = GetUserId();
+	else
+		context.check_as_user = view->rd_rel->relowner;
 	context.viewoid = reloid;
 	context.ancestor_views = lappend_oid(ancestor_views, reloid);
 
diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c
index 3d82138cb39..4eeed580b16 100644
--- a/src/backend/rewrite/rewriteHandler.c
+++ b/src/backend/rewrite/rewriteHandler.c
@@ -3242,18 +3242,24 @@ rewriteTargetView(Query *parsetree, Relation view)
 				   0);
 
 	/*
-	 * Mark the new target RTE for the permissions checks that we want to
-	 * enforce against the view owner, as distinct from the query caller.  At
-	 * the relation level, require the same INSERT/UPDATE/DELETE permissions
-	 * that the query caller needs against the view.  We drop the ACL_SELECT
-	 * bit that is presumably in new_rte->requiredPerms initially.
+	 * If the view has "security_invoker" set, mark the new target RTE for the
+	 * permissions checks that we want to enforce against the query caller.
+	 * Otherwise we want to enforce them against the view owner.
+	 *
+	 * At the relation level, require the same INSERT/UPDATE/DELETE
+	 * permissions that the query caller needs against the view.  We drop the
+	 * ACL_SELECT bit that is presumably in new_rte->requiredPerms initially.
 	 *
 	 * Note: the original view RTE remains in the query's rangetable list.
 	 * Although it will be unused in the query plan, we need it there so that
 	 * the executor still performs appropriate permissions checks for the
 	 * query caller's use of the view.
 	 */
-	new_rte->checkAsUser = view->rd_rel->relowner;
+	if (RelationHasSecurityInvoker(view))
+		new_rte->checkAsUser = InvalidOid;
+	else
+		new_rte->checkAsUser = view->rd_rel->relowner;
+
 	new_rte->requiredPerms = view_rte->requiredPerms;
 
 	/*
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index fccffce5729..fbd11883e17 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -722,6 +722,8 @@ RelationBuildTupleDesc(Relation relation)
  * entry, because that keeps the update logic in RelationClearRelation()
  * manageable.  The other subsidiary data structures are simple enough
  * to be easy to free explicitly, anyway.
+ *
+ * Note: The relation's reloptions must have been extracted first.
  */
 static void
 RelationBuildRuleLock(Relation relation)
@@ -787,6 +789,7 @@ RelationBuildRuleLock(Relation relation)
 		Datum		rule_datum;
 		char	   *rule_str;
 		RewriteRule *rule;
+		Oid			check_as_user;
 
 		rule = (RewriteRule *) MemoryContextAlloc(rulescxt,
 												  sizeof(RewriteRule));
@@ -826,10 +829,23 @@ RelationBuildRuleLock(Relation relation)
 		pfree(rule_str);
 
 		/*
-		 * We want the rule's table references to be checked as though by the
-		 * table owner, not the user referencing the rule.  Therefore, scan
-		 * through the rule's actions and set the checkAsUser field on all
-		 * rtable entries.  We have to look at the qual as well, in case it
+		 * If this is a SELECT rule defining a view, and the view has
+		 * "security_invoker" set, we must perform all permissions checks on
+		 * relations referred to by the rule as the invoking user.
+		 *
+		 * In all other cases (including non-SELECT rules on security invoker
+		 * views), perform the permissions checks as the relation owner.
+		 */
+		if (rule->event == CMD_SELECT &&
+			relation->rd_rel->relkind == RELKIND_VIEW &&
+			RelationHasSecurityInvoker(relation))
+			check_as_user = InvalidOid;
+		else
+			check_as_user = relation->rd_rel->relowner;
+
+		/*
+		 * Scan through the rule's actions and set the checkAsUser field on
+		 * all rtable entries. We have to look at the qual as well, in case it
 		 * contains sublinks.
 		 *
 		 * The reason for doing this when the rule is loaded, rather than when
@@ -838,8 +854,8 @@ RelationBuildRuleLock(Relation relation)
 		 * the rule tree during load is relatively cheap (compared to
 		 * constructing it in the first place), so we do it here.
 		 */
-		setRuleCheckAsUser((Node *) rule->actions, relation->rd_rel->relowner);
-		setRuleCheckAsUser(rule->qual, relation->rd_rel->relowner);
+		setRuleCheckAsUser((Node *) rule->actions, check_as_user);
+		setRuleCheckAsUser(rule->qual, check_as_user);
 
 		if (numlocks >= maxlocks)
 		{
@@ -1164,27 +1180,6 @@ retry:
 	 */
 	RelationBuildTupleDesc(relation);
 
-	/*
-	 * Fetch rules and triggers that affect this relation
-	 */
-	if (relation->rd_rel->relhasrules)
-		RelationBuildRuleLock(relation);
-	else
-	{
-		relation->rd_rules = NULL;
-		relation->rd_rulescxt = NULL;
-	}
-
-	if (relation->rd_rel->relhastriggers)
-		RelationBuildTriggers(relation);
-	else
-		relation->trigdesc = NULL;
-
-	if (relation->rd_rel->relrowsecurity)
-		RelationBuildRowSecurity(relation);
-	else
-		relation->rd_rsdesc = NULL;
-
 	/* foreign key data is not loaded till asked for */
 	relation->rd_fkeylist = NIL;
 	relation->rd_fkeyvalid = false;
@@ -1217,6 +1212,30 @@ retry:
 	RelationParseRelOptions(relation, pg_class_tuple);
 
 	/*
+	 * Fetch rules and triggers that affect this relation.
+	 *
+	 * Note that RelationBuildRuleLock() relies on this being done after
+	 * extracting the relation's reloptions.
+	 */
+	if (relation->rd_rel->relhasrules)
+		RelationBuildRuleLock(relation);
+	else
+	{
+		relation->rd_rules = NULL;
+		relation->rd_rulescxt = NULL;
+	}
+
+	if (relation->rd_rel->relhastriggers)
+		RelationBuildTriggers(relation);
+	else
+		relation->trigdesc = NULL;
+
+	if (relation->rd_rel->relrowsecurity)
+		RelationBuildRowSecurity(relation);
+	else
+		relation->rd_rsdesc = NULL;
+
+	/*
 	 * initialize the relation lock manager information
 	 */
 	RelationInitLockInfo(relation); /* see lmgr.c */