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

doc/src/sgml/ref/create_view.sgml

@@ -323,12 +323,6 @@ CREATE VIEW vista AS SELECT text 'Hello World' AS hello;
        or set-returning functions.
       </para>
      </listitem>
-
-     <listitem>
-      <para>
-       The view must not have the <literal>security_barrier</> property.
-      </para>
-     </listitem>
     </itemizedlist>
    </para>
 
@@ -361,6 +355,19 @@ CREATE VIEW vista AS SELECT text 'Hello World' AS hello;
     such rows that are not visible through the view.
    </para>
 
+   <para>
+    If an automatically updatable view is marked with the
+    <literal>security_barrier</> property then all the view's <literal>WHERE</>
+    conditions (and any conditions using operators which are marked as LEAKPROOF)
+    will always be evaluated before any conditions that a user of the view has
+    added.   See <xref linkend="rules-privileges"> for full details.  Note that,
+    due to this, rows which are not ultimately returned (because they do not
+    pass the user's <literal>WHERE</> conditions) may still end up being locked.
+    <command>EXPLAIN</command> can be used to see which conditions are
+    applied at the relation level (and therefore do not lock rows) and which are
+    not.
+   </para>
+
    <para>
     A more complex view that does not satisfy all these conditions is
     read-only by default: the system will not allow an insert, update, or

src/backend/commands/tablecmds.c

@@ -8910,16 +8910,13 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 		List	   *view_options = untransformRelOptions(newOptions);
 		ListCell   *cell;
 		bool		check_option = false;
-		bool		security_barrier = false;
 
 		foreach(cell, view_options)
 		{
 			DefElem    *defel = (DefElem *) lfirst(cell);
 
 			if (pg_strcasecmp(defel->defname, "check_option") == 0)
 				check_option = true;
-			if (pg_strcasecmp(defel->defname, "security_barrier") == 0)
-				security_barrier = defGetBoolean(defel);
 		}
 
 		/*
@@ -8929,8 +8926,7 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 		if (check_option)
 		{
 			const char *view_updatable_error =
-				view_query_is_auto_updatable(view_query,
-											 security_barrier, true);
+				view_query_is_auto_updatable(view_query, true);
 
 			if (view_updatable_error)
 				ereport(ERROR,

src/backend/commands/view.c

@@ -396,7 +396,6 @@ DefineView(ViewStmt *stmt, const char *queryString)
 	RangeVar   *view;
 	ListCell   *cell;
 	bool		check_option;
-	bool		security_barrier;
 
 	/*
 	 * Run parse analysis to convert the raw parse tree to a Query.  Note this
@@ -451,16 +450,13 @@ DefineView(ViewStmt *stmt, const char *queryString)
 	 * specified.
 	 */
 	check_option = false;
-	security_barrier = false;
 
 	foreach(cell, stmt->options)
 	{
 		DefElem    *defel = (DefElem *) lfirst(cell);
 
 		if (pg_strcasecmp(defel->defname, "check_option") == 0)
 			check_option = true;
-		if (pg_strcasecmp(defel->defname, "security_barrier") == 0)
-			security_barrier = defGetBoolean(defel);
 	}
 
 	/*
@@ -470,7 +466,7 @@ DefineView(ViewStmt *stmt, const char *queryString)
 	if (check_option)
 	{
 		const char *view_updatable_error =
-			view_query_is_auto_updatable(viewParse, security_barrier, true);
+			view_query_is_auto_updatable(viewParse, true);
 
 		if (view_updatable_error)
 			ereport(ERROR,

src/backend/nodes/copyfuncs.c

@@ -1998,6 +1998,7 @@ _copyRangeTblEntry(const RangeTblEntry *from)
 	COPY_SCALAR_FIELD(checkAsUser);
 	COPY_BITMAPSET_FIELD(selectedCols);
 	COPY_BITMAPSET_FIELD(modifiedCols);
+	COPY_NODE_FIELD(securityQuals);
 
 	return newnode;
 }

src/backend/nodes/equalfuncs.c

@@ -2296,6 +2296,7 @@ _equalRangeTblEntry(const RangeTblEntry *a, const RangeTblEntry *b)
 	COMPARE_SCALAR_FIELD(checkAsUser);
 	COMPARE_BITMAPSET_FIELD(selectedCols);
 	COMPARE_BITMAPSET_FIELD(modifiedCols);
+	COMPARE_NODE_FIELD(securityQuals);
 
 	return true;
 }

src/backend/nodes/nodeFuncs.c

@@ -2020,6 +2020,9 @@ range_table_walker(List *rtable,
 					return true;
 				break;
 		}
+
+		if (walker(rte->securityQuals, context))
+			return true;
 	}
 	return false;
 }
@@ -2755,6 +2758,7 @@ range_table_mutator(List *rtable,
 				MUTATE(newrte->values_lists, rte->values_lists, List *);
 				break;
 		}
+		MUTATE(newrte->securityQuals, rte->securityQuals, List *);
 		newrt = lappend(newrt, newrte);
 	}
 	return newrt;

src/backend/nodes/outfuncs.c

@@ -2409,6 +2409,7 @@ _outRangeTblEntry(StringInfo str, const RangeTblEntry *node)
 	WRITE_OID_FIELD(checkAsUser);
 	WRITE_BITMAPSET_FIELD(selectedCols);
 	WRITE_BITMAPSET_FIELD(modifiedCols);
+	WRITE_NODE_FIELD(securityQuals);
 }
 
 static void

src/backend/nodes/readfuncs.c

@@ -1252,6 +1252,7 @@ _readRangeTblEntry(void)
 	READ_OID_FIELD(checkAsUser);
 	READ_BITMAPSET_FIELD(selectedCols);
 	READ_BITMAPSET_FIELD(modifiedCols);
+	READ_NODE_FIELD(securityQuals);
 
 	READ_DONE();
 }

src/backend/optimizer/plan/planner.c

@@ -915,6 +915,12 @@ inheritance_planner(PlannerInfo *root)
 		/* Generate plan */
 		subplan = grouping_planner(&subroot, 0.0 /* retrieve all tuples */ );
 
+		/*
+		 * Planning may have modified the query result relation (if there
+		 * were security barrier quals on the result RTE).
+		 */
+		appinfo->child_relid = subroot.parse->resultRelation;
+
 		/*
 		 * If this child rel was excluded by constraint exclusion, exclude it
 		 * from the result plan.
@@ -932,9 +938,40 @@ inheritance_planner(PlannerInfo *root)
 		if (final_rtable == NIL)
 			final_rtable = subroot.parse->rtable;
 		else
-			final_rtable = list_concat(final_rtable,
+		{
+			List	   *tmp_rtable = NIL;
+			ListCell   *cell1, *cell2;
+
+			/*
+			 * Check to see if any of the original RTEs were turned into
+			 * subqueries during planning.  Currently, this should only ever
+			 * happen due to securityQuals being involved which push a
+			 * relation down under a subquery, to ensure that the security
+			 * barrier quals are evaluated first.
+			 *
+			 * When this happens, we want to use the new subqueries in the
+			 * final rtable.
+			 */
+			forboth(cell1, final_rtable, cell2, subroot.parse->rtable)
+			{
+				RangeTblEntry *rte1 = (RangeTblEntry *) lfirst(cell1);
+				RangeTblEntry *rte2 = (RangeTblEntry *) lfirst(cell2);
+
+				if (rte1->rtekind == RTE_RELATION &&
+					rte2->rtekind == RTE_SUBQUERY)
+				{
+					/* Should only be when there are securityQuals today */
+					Assert(rte1->securityQuals != NIL);
+					tmp_rtable = lappend(tmp_rtable, rte2);
+				}
+				else
+					tmp_rtable = lappend(tmp_rtable, rte1);
+			}
+
+			final_rtable = list_concat(tmp_rtable,
 									   list_copy_tail(subroot.parse->rtable,
 												 list_length(final_rtable)));
+		}
 
 		/*
 		 * We need to collect all the RelOptInfos from all child plans into
@@ -1162,6 +1199,12 @@ grouping_planner(PlannerInfo *root, double tuple_fraction)
 		/* Preprocess targetlist */
 		tlist = preprocess_targetlist(root, tlist);
 
+		/*
+		 * Expand any rangetable entries that have security barrier quals.
+		 * This may add new security barrier subquery RTEs to the rangetable.
+		 */
+		expand_security_quals(root, tlist);
+
 		/*
 		 * Locate any window functions in the tlist.  (We don't need to look
 		 * anywhere else, since expressions used in ORDER BY will be in there

src/backend/optimizer/prep/Makefile

@@ -12,6 +12,6 @@ subdir = src/backend/optimizer/prep
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = prepjointree.o prepqual.o preptlist.o prepunion.o
+OBJS = prepjointree.o prepqual.o prepsecurity.o preptlist.o prepunion.o
 
 include $(top_srcdir)/src/backend/common.mk