Make security barrier views automatically updatable

Views which are marked as security_barrier must have their quals
applied before any user-defined quals are called, to prevent
user-defined functions from being able to see rows which the
security barrier view is intended to prevent them from seeing.

Remove the restriction on security barrier views being automatically
updatable by adding a new securityQuals list to the RTE structure
which keeps track of the quals from security barrier views at each
level, independently of the user-supplied quals.  When RTEs are
later discovered which have securityQuals populated, they are turned
into subquery RTEs which are marked as security_barrier to prevent
any user-supplied quals being pushed down (modulo LEAKPROOF quals).

Dean Rasheed, reviewed by Craig Ringer, Simon Riggs, KaiGai Kohei

1 files changed, 46 insertions, 14 deletions

diff --git a/src/backend/optimizer/prep/prepunion.c b/src/backend/optimizer/prep/prepunion.c
index 52dcc720de7..cdf541d34d5 100644
--- a/src/backend/optimizer/prep/prepunion.c
+++ b/src/backend/optimizer/prep/prepunion.c
@@ -55,6 +55,7 @@ typedef struct
 {
 	PlannerInfo *root;
 	AppendRelInfo *appinfo;
+	int			sublevels_up;
 } adjust_appendrel_attrs_context;
 
 static Plan *recurse_set_operations(Node *setOp, PlannerInfo *root,
@@ -1580,8 +1581,9 @@ translate_col_privs(const Bitmapset *parent_privs,
  *	  child rel instead.  We also update rtindexes appearing outside Vars,
  *	  such as resultRelation and jointree relids.
  *
- * Note: this is only applied after conversion of sublinks to subplans,
- * so we don't need to cope with recursion into sub-queries.
+ * Note: this is applied after conversion of sublinks to subplans in the
+ * query jointree, but there may still be sublinks in the security barrier
+ * quals of RTEs, so we do need to cope with recursion into sub-queries.
  *
  * Note: this is not hugely different from what pullup_replace_vars() does;
  * maybe we should try to fold the two routines together.
@@ -1594,9 +1596,12 @@ adjust_appendrel_attrs(PlannerInfo *root, Node *node, AppendRelInfo *appinfo)
 
 	context.root = root;
 	context.appinfo = appinfo;
+	context.sublevels_up = 0;
 
 	/*
-	 * Must be prepared to start with a Query or a bare expression tree.
+	 * Must be prepared to start with a Query or a bare expression tree; if
+	 * it's a Query, go straight to query_tree_walker to make sure that
+	 * sublevels_up doesn't get incremented prematurely.
 	 */
 	if (node && IsA(node, Query))
 	{
@@ -1635,7 +1640,7 @@ adjust_appendrel_attrs_mutator(Node *node,
 	{
 		Var		   *var = (Var *) copyObject(node);
 
-		if (var->varlevelsup == 0 &&
+		if (var->varlevelsup == context->sublevels_up &&
 			var->varno == appinfo->parent_relid)
 		{
 			var->varno = appinfo->child_relid;
@@ -1652,6 +1657,7 @@ adjust_appendrel_attrs_mutator(Node *node,
 				if (newnode == NULL)
 					elog(ERROR, "attribute %d of relation \"%s\" does not exist",
 						 var->varattno, get_rel_name(appinfo->parent_reloid));
+				((Var *) newnode)->varlevelsup += context->sublevels_up;
 				return newnode;
 			}
 			else if (var->varattno == 0)
@@ -1694,10 +1700,16 @@ adjust_appendrel_attrs_mutator(Node *node,
 					RowExpr    *rowexpr;
 					List	   *fields;
 					RangeTblEntry *rte;
+					ListCell   *lc;
 
 					rte = rt_fetch(appinfo->parent_relid,
 								   context->root->parse->rtable);
 					fields = (List *) copyObject(appinfo->translated_vars);
+					foreach(lc, fields)
+					{
+						Var		   *field = (Var *) lfirst(lc);
+						field->varlevelsup += context->sublevels_up;
+					}
 					rowexpr = makeNode(RowExpr);
 					rowexpr->args = fields;
 					rowexpr->row_typeid = var->vartype;
@@ -1716,7 +1728,8 @@ adjust_appendrel_attrs_mutator(Node *node,
 	{
 		CurrentOfExpr *cexpr = (CurrentOfExpr *) copyObject(node);
 
-		if (cexpr->cvarno == appinfo->parent_relid)
+		if (context->sublevels_up == 0 &&
+			cexpr->cvarno == appinfo->parent_relid)
 			cexpr->cvarno = appinfo->child_relid;
 		return (Node *) cexpr;
 	}
@@ -1724,7 +1737,8 @@ adjust_appendrel_attrs_mutator(Node *node,
 	{
 		RangeTblRef *rtr = (RangeTblRef *) copyObject(node);
 
-		if (rtr->rtindex == appinfo->parent_relid)
+		if (context->sublevels_up == 0 &&
+			rtr->rtindex == appinfo->parent_relid)
 			rtr->rtindex = appinfo->child_relid;
 		return (Node *) rtr;
 	}
@@ -1737,7 +1751,8 @@ adjust_appendrel_attrs_mutator(Node *node,
 											  adjust_appendrel_attrs_mutator,
 												 (void *) context);
 		/* now fix JoinExpr's rtindex (probably never happens) */
-		if (j->rtindex == appinfo->parent_relid)
+		if (context->sublevels_up == 0 &&
+			j->rtindex == appinfo->parent_relid)
 			j->rtindex = appinfo->child_relid;
 		return (Node *) j;
 	}
@@ -1750,7 +1765,7 @@ adjust_appendrel_attrs_mutator(Node *node,
 											  adjust_appendrel_attrs_mutator,
 														 (void *) context);
 		/* now fix PlaceHolderVar's relid sets */
-		if (phv->phlevelsup == 0)
+		if (phv->phlevelsup == context->sublevels_up)
 			phv->phrels = adjust_relid_set(phv->phrels,
 										   appinfo->parent_relid,
 										   appinfo->child_relid);
@@ -1822,12 +1837,29 @@ adjust_appendrel_attrs_mutator(Node *node,
 		return (Node *) newinfo;
 	}
 
-	/*
-	 * NOTE: we do not need to recurse into sublinks, because they should
-	 * already have been converted to subplans before we see them.
-	 */
-	Assert(!IsA(node, SubLink));
-	Assert(!IsA(node, Query));
+	if (IsA(node, Query))
+	{
+		/*
+		 * Recurse into sublink subqueries. This should only be possible in
+		 * security barrier quals of top-level RTEs. All other sublinks should
+		 * have already been converted to subplans during expression
+		 * preprocessing, but this doesn't happen for security barrier quals,
+		 * since they are destined to become quals of a subquery RTE, which
+		 * will be recursively planned, and so should not be preprocessed at
+		 * this stage.
+		 *
+		 * We don't explicitly Assert() for securityQuals here simply because
+		 * it's not trivial to do so.
+		 */
+		Query	   *newnode;
+
+		context->sublevels_up++;
+		newnode = query_tree_mutator((Query *) node,
+									 adjust_appendrel_attrs_mutator,
+									 (void *) context, 0);
+		context->sublevels_up--;
+		return (Node *) newnode;
+	}
 
 	return expression_tree_mutator(node, adjust_appendrel_attrs_mutator,
 								   (void *) context);