Perform RLS WITH CHECK before constraints, etc

The RLS capability is built on top of the WITH CHECK OPTION
system which was added for auto-updatable views, however, unlike
WCOs on views (which are mandated by the SQL spec to not fire until
after all other constraints and checks are done), it makes much more
sense for RLS checks to happen earlier than constraint and uniqueness
checks.

This patch reworks the structure which holds the WCOs a bit to be
explicitly either VIEW or RLS checks and the RLS-related checks are
done prior to the constraint and uniqueness checks.  This also allows
better error reporting as we are now reporting when a violation is due
to a WITH CHECK OPTION and when it's due to an RLS policy violation,
which was independently noted by Craig Ringer as being confusing.

The documentation is also updated to include a paragraph about when RLS
WITH CHECK handling is performed, as there have been a number of
questions regarding that and the documentation was previously silent on
the matter.

Author: Dean Rasheed, with some kabitzing and comment changes by me.

Author: sfrost

doc/src/sgml/ref/create_policy.sgml

@@ -60,6 +60,14 @@ CREATE POLICY <replaceable class="parameter">name</replaceable> ON <replaceable
    expressions, as they are assumed to be trustworthy.
   </para>
 
+  <para>
+    For INSERT and UPDATE queries, WITH CHECK expressions are enforced after
+    BEFORE triggers are fired, and before any data modifications are made.
+    Thus a BEFORE ROW trigger may modify the data to be inserted, affecting
+    the result of the security policy check.  WITH CHECK expressions are
+    enforced before any other constraints.
+  </para>
+
   <para>
    Policy names are per-table, therefore one policy name can be used for many
    different tables and have a definition for each table which is appropriate to

src/backend/executor/execMain.c

@@ -1673,9 +1673,15 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 
 /*
  * ExecWithCheckOptions -- check that tuple satisfies any WITH CHECK OPTIONs
+ * of the specified kind.
+ *
+ * Note that this needs to be called multiple times to ensure that all kinds of
+ * WITH CHECK OPTIONs are handled (both those from views which have the WITH
+ * CHECK OPTION set and from row level security policies).  See ExecInsert()
+ * and ExecUpdate().
  */
 void
-ExecWithCheckOptions(ResultRelInfo *resultRelInfo,
+ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 					 TupleTableSlot *slot, EState *estate)
 {
 	Relation	rel = resultRelInfo->ri_RelationDesc;
@@ -1700,6 +1706,13 @@ ExecWithCheckOptions(ResultRelInfo *resultRelInfo,
 		WithCheckOption *wco = (WithCheckOption *) lfirst(l1);
 		ExprState  *wcoExpr = (ExprState *) lfirst(l2);
 
+		/*
+		 * Skip any WCOs which are not the kind we are looking for at this
+		 * time.
+		 */
+		if (wco->kind != kind)
+			continue;
+
 		/*
 		 * WITH CHECK OPTION checks are intended to ensure that the new tuple
 		 * is visible (in the case of a view) or that it passes the
@@ -1714,19 +1727,42 @@ ExecWithCheckOptions(ResultRelInfo *resultRelInfo,
 			char	   *val_desc;
 			Bitmapset  *modifiedCols;
 
-			modifiedCols = GetModifiedColumns(resultRelInfo, estate);
-			val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
-													 slot,
-													 tupdesc,
-													 modifiedCols,
-													 64);
-
-			ereport(ERROR,
-					(errcode(ERRCODE_WITH_CHECK_OPTION_VIOLATION),
-				 errmsg("new row violates WITH CHECK OPTION for \"%s\"",
-						wco->viewname),
-					val_desc ? errdetail("Failing row contains %s.", val_desc) :
-							   0));
+			switch (wco->kind)
+			{
+				/*
+				 * For WITH CHECK OPTIONs coming from views, we might be able to
+				 * provide the details on the row, depending on the permissions
+				 * on the relation (that is, if the user could view it directly
+				 * anyway).  For RLS violations, we don't include the data since
+				 * we don't know if the user should be able to view the tuple as
+				 * as that depends on the USING policy.
+				 */
+				case WCO_VIEW_CHECK:
+					modifiedCols = GetModifiedColumns(resultRelInfo, estate);
+					val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
+															 slot,
+															 tupdesc,
+															 modifiedCols,
+															 64);
+
+					ereport(ERROR,
+							(errcode(ERRCODE_WITH_CHECK_OPTION_VIOLATION),
+						errmsg("new row violates WITH CHECK OPTION for \"%s\"",
+							   wco->relname),
+							 val_desc ? errdetail("Failing row contains %s.",
+												  val_desc) : 0));
+					break;
+				case WCO_RLS_INSERT_CHECK:
+				case WCO_RLS_UPDATE_CHECK:
+					ereport(ERROR,
+							(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+						 errmsg("new row violates row level security policy for \"%s\"",
+								wco->relname)));
+					break;
+				default:
+					elog(ERROR, "unrecognized WCO kind: %u", wco->kind);
+					break;
+			}
 		}
 	}
 }

src/backend/executor/nodeModifyTable.c

@@ -252,6 +252,16 @@ ExecInsert(TupleTableSlot *slot,
 		 */
 		tuple->t_tableOid = RelationGetRelid(resultRelationDesc);
 
+		/*
+		 * Check any RLS INSERT WITH CHECK policies
+		 *
+		 * ExecWithCheckOptions() will skip any WCOs which are not of
+		 * the kind we are looking for at this point.
+		 */
+		if (resultRelInfo->ri_WithCheckOptions != NIL)
+			ExecWithCheckOptions(WCO_RLS_INSERT_CHECK,
+								 resultRelInfo, slot, estate);
+
 		/*
 		 * Check the constraints of the tuple
 		 */
@@ -287,9 +297,21 @@ ExecInsert(TupleTableSlot *slot,
 
 	list_free(recheckIndexes);
 
-	/* Check any WITH CHECK OPTION constraints */
+	/*
+	 * Check any WITH CHECK OPTION constraints from parent views.  We
+	 * are required to do this after testing all constraints and
+	 * uniqueness violations per the SQL spec, so we do it after actually
+	 * inserting the record into the heap and all indexes.
+	 *
+	 * ExecWithCheckOptions will elog(ERROR) if a violation is found, so
+	 * the tuple will never be seen, if it violates the the WITH CHECK
+	 * OPTION.
+	 *
+	 * ExecWithCheckOptions() will skip any WCOs which are not of
+	 * the kind we are looking for at this point.
+	 */
 	if (resultRelInfo->ri_WithCheckOptions != NIL)
-		ExecWithCheckOptions(resultRelInfo, slot, estate);
+		ExecWithCheckOptions(WCO_VIEW_CHECK, resultRelInfo, slot, estate);
 
 	/* Process RETURNING if present */
 	if (resultRelInfo->ri_projectReturning)
@@ -653,15 +675,25 @@ ExecUpdate(ItemPointer tupleid,
 		tuple->t_tableOid = RelationGetRelid(resultRelationDesc);
 
 		/*
-		 * Check the constraints of the tuple
+		 * Check any RLS UPDATE WITH CHECK policies
 		 *
 		 * If we generate a new candidate tuple after EvalPlanQual testing, we
-		 * must loop back here and recheck constraints.  (We don't need to
-		 * redo triggers, however.  If there are any BEFORE triggers then
-		 * trigger.c will have done heap_lock_tuple to lock the correct tuple,
-		 * so there's no need to do them again.)
+		 * must loop back here and recheck any RLS policies and constraints.
+		 * (We don't need to redo triggers, however.  If there are any BEFORE
+		 * triggers then trigger.c will have done heap_lock_tuple to lock the
+		 * correct tuple, so there's no need to do them again.)
+		 *
+		 * ExecWithCheckOptions() will skip any WCOs which are not of
+		 * the kind we are looking for at this point.
 		 */
 lreplace:;
+		if (resultRelInfo->ri_WithCheckOptions != NIL)
+			ExecWithCheckOptions(WCO_RLS_UPDATE_CHECK,
+								 resultRelInfo, slot, estate);
+
+		/*
+		 * Check the constraints of the tuple
+		 */
 		if (resultRelationDesc->rd_att->constr)
 			ExecConstraints(resultRelInfo, slot, estate);
 
@@ -780,9 +812,17 @@ lreplace:;
 
 	list_free(recheckIndexes);
 
-	/* Check any WITH CHECK OPTION constraints */
+	/*
+	 * Check any WITH CHECK OPTION constraints from parent views.  We
+	 * are required to do this after testing all constraints and
+	 * uniqueness violations per the SQL spec, so we do it after actually
+	 * updating the record in the heap and all indexes.
+	 *
+	 * ExecWithCheckOptions() will skip any WCOs which are not of
+	 * the kind we are looking for at this point.
+	 */
 	if (resultRelInfo->ri_WithCheckOptions != NIL)
-		ExecWithCheckOptions(resultRelInfo, slot, estate);
+		ExecWithCheckOptions(WCO_VIEW_CHECK, resultRelInfo, slot, estate);
 
 	/* Process RETURNING if present */
 	if (resultRelInfo->ri_projectReturning)

src/backend/nodes/copyfuncs.c

@@ -2064,7 +2064,8 @@ _copyWithCheckOption(const WithCheckOption *from)
 {
 	WithCheckOption *newnode = makeNode(WithCheckOption);
 
-	COPY_STRING_FIELD(viewname);
+	COPY_SCALAR_FIELD(kind);
+	COPY_STRING_FIELD(relname);
 	COPY_NODE_FIELD(qual);
 	COPY_SCALAR_FIELD(cascaded);
 

src/backend/nodes/equalfuncs.c

@@ -2363,7 +2363,8 @@ _equalRangeTblFunction(const RangeTblFunction *a, const RangeTblFunction *b)
 static bool
 _equalWithCheckOption(const WithCheckOption *a, const WithCheckOption *b)
 {
-	COMPARE_STRING_FIELD(viewname);
+	COMPARE_SCALAR_FIELD(kind);
+	COMPARE_STRING_FIELD(relname);
 	COMPARE_NODE_FIELD(qual);
 	COMPARE_SCALAR_FIELD(cascaded);
 

src/backend/nodes/outfuncs.c

@@ -2332,7 +2332,8 @@ _outWithCheckOption(StringInfo str, const WithCheckOption *node)
 {
 	WRITE_NODE_TYPE("WITHCHECKOPTION");
 
-	WRITE_STRING_FIELD(viewname);
+	WRITE_ENUM_FIELD(kind, WCOKind);
+	WRITE_STRING_FIELD(relname);
 	WRITE_NODE_FIELD(qual);
 	WRITE_BOOL_FIELD(cascaded);
 }

src/backend/nodes/readfuncs.c

@@ -266,7 +266,8 @@ _readWithCheckOption(void)
 {
 	READ_LOCALS(WithCheckOption);
 
-	READ_STRING_FIELD(viewname);
+	READ_ENUM_FIELD(kind, WCOKind);
+	READ_STRING_FIELD(relname);
 	READ_NODE_FIELD(qual);
 	READ_BOOL_FIELD(cascaded);
 

src/backend/rewrite/rewriteHandler.c

@@ -2947,7 +2947,8 @@ rewriteTargetView(Query *parsetree, Relation view)
 			WithCheckOption *wco;
 
 			wco = makeNode(WithCheckOption);
-			wco->viewname = pstrdup(RelationGetRelationName(view));
+			wco->kind = WCO_VIEW_CHECK;
+			wco->relname = pstrdup(RelationGetRelationName(view));
 			wco->qual = NULL;
 			wco->cascaded = cascaded;
 

src/backend/rewrite/rowsecurity.c

@@ -259,7 +259,9 @@ get_row_security_policies(Query* root, RangeTblEntry* rte, int rt_index,
 			WithCheckOption	   *wco;
 
 			wco = (WithCheckOption *) makeNode(WithCheckOption);
-			wco->viewname = pstrdup(RelationGetRelationName(rel));
+			wco->kind = root->commandType == CMD_INSERT ? WCO_RLS_INSERT_CHECK :
+														  WCO_RLS_UPDATE_CHECK;
+			wco->relname = pstrdup(RelationGetRelationName(rel));
 			wco->qual = (Node *) hook_with_check_expr_restrictive;
 			wco->cascaded = false;
 			*withCheckOptions = lappend(*withCheckOptions, wco);
@@ -274,7 +276,9 @@ get_row_security_policies(Query* root, RangeTblEntry* rte, int rt_index,
 			WithCheckOption	   *wco;
 
 			wco = (WithCheckOption *) makeNode(WithCheckOption);
-			wco->viewname = pstrdup(RelationGetRelationName(rel));
+			wco->kind = root->commandType == CMD_INSERT ? WCO_RLS_INSERT_CHECK :
+														  WCO_RLS_UPDATE_CHECK;
+			wco->relname = pstrdup(RelationGetRelationName(rel));
 			wco->qual = (Node *) rowsec_with_check_expr;
 			wco->cascaded = false;
 			*withCheckOptions = lappend(*withCheckOptions, wco);
@@ -285,7 +289,9 @@ get_row_security_policies(Query* root, RangeTblEntry* rte, int rt_index,
 			WithCheckOption	   *wco;
 
 			wco = (WithCheckOption *) makeNode(WithCheckOption);
-			wco->viewname = pstrdup(RelationGetRelationName(rel));
+			wco->kind = root->commandType == CMD_INSERT ? WCO_RLS_INSERT_CHECK :
+														  WCO_RLS_UPDATE_CHECK;
+			wco->relname = pstrdup(RelationGetRelationName(rel));
 			wco->qual = (Node *) hook_with_check_expr_permissive;
 			wco->cascaded = false;
 			*withCheckOptions = lappend(*withCheckOptions, wco);
@@ -297,13 +303,18 @@ get_row_security_policies(Query* root, RangeTblEntry* rte, int rt_index,
 			List			   *combined_quals = NIL;
 			Expr			   *combined_qual_eval;
 
-			combined_quals = lcons(copyObject(rowsec_with_check_expr), combined_quals);
-			combined_quals = lcons(copyObject(hook_with_check_expr_permissive), combined_quals);
+			combined_quals = lcons(copyObject(rowsec_with_check_expr),
+								   combined_quals);
+
+			combined_quals = lcons(copyObject(hook_with_check_expr_permissive),
+								   combined_quals);
 
 			combined_qual_eval = makeBoolExpr(OR_EXPR, combined_quals, -1);
 
 			wco = (WithCheckOption *) makeNode(WithCheckOption);
-			wco->viewname = pstrdup(RelationGetRelationName(rel));
+			wco->kind = root->commandType == CMD_INSERT ? WCO_RLS_INSERT_CHECK :
+														  WCO_RLS_UPDATE_CHECK;
+			wco->relname = pstrdup(RelationGetRelationName(rel));
 			wco->qual = (Node *) combined_qual_eval;
 			wco->cascaded = false;
 			*withCheckOptions = lappend(*withCheckOptions, wco);
@@ -332,7 +343,8 @@ get_row_security_policies(Query* root, RangeTblEntry* rte, int rt_index,
 			Expr   *combined_qual_eval;
 
 			combined_quals = lcons(copyObject(rowsec_expr), combined_quals);
-			combined_quals = lcons(copyObject(hook_expr_permissive), combined_quals);
+			combined_quals = lcons(copyObject(hook_expr_permissive),
+								   combined_quals);
 
 			combined_qual_eval = makeBoolExpr(OR_EXPR, combined_quals, -1);
 

src/include/executor/executor.h

@@ -193,7 +193,7 @@ extern ResultRelInfo *ExecGetTriggerResultRel(EState *estate, Oid relid);
 extern bool ExecContextForcesOids(PlanState *planstate, bool *hasoids);
 extern void ExecConstraints(ResultRelInfo *resultRelInfo,
 				TupleTableSlot *slot, EState *estate);
-extern void ExecWithCheckOptions(ResultRelInfo *resultRelInfo,
+extern void ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 					 TupleTableSlot *slot, EState *estate);
 extern ExecRowMark *ExecFindRowMark(EState *estate, Index rti);
 extern ExecAuxRowMark *ExecBuildAuxRowMark(ExecRowMark *erm, List *targetlist);

src/include/nodes/execnodes.h

@@ -303,7 +303,7 @@ typedef struct JunkFilter
  *		TrigInstrument			optional runtime measurements for triggers
  *		FdwRoutine				FDW callback functions, if foreign table
  *		FdwState				available to save private state of FDW
- *		WithCheckOptions		list of WithCheckOption's for views
+ *		WithCheckOptions		list of WithCheckOption's to be checked
  *		WithCheckOptionExprs	list of WithCheckOption expr states
  *		ConstraintExprs			array of constraint-checking expr states
  *		junkFilter				for removing junk attributes from tuples

src/include/nodes/parsenodes.h

@@ -872,14 +872,23 @@ typedef struct RangeTblFunction
 /*
  * WithCheckOption -
  *		representation of WITH CHECK OPTION checks to be applied to new tuples
- *		when inserting/updating an auto-updatable view.
+ *		when inserting/updating an auto-updatable view, or RLS WITH CHECK
+ *		policies to be applied when inserting/updating a relation with RLS.
  */
+typedef enum WCOKind
+{
+	WCO_VIEW_CHECK,				/* WCO on an auto-updatable view */
+	WCO_RLS_INSERT_CHECK,		/* RLS INSERT WITH CHECK policy */
+	WCO_RLS_UPDATE_CHECK		/* RLS UPDATE WITH CHECK policy */
+} WCOKind;
+
 typedef struct WithCheckOption
 {
 	NodeTag		type;
-	char	   *viewname;		/* name of view that specified the WCO */
+	WCOKind		kind;			/* kind of WCO */
+	char	   *relname;		/* name of relation that specified the WCO */
 	Node	   *qual;			/* constraint qual to check */
-	bool		cascaded;		/* true = WITH CASCADED CHECK OPTION */
+	bool		cascaded;		/* true for a cascaded WCO on a view */
 } WithCheckOption;
 
 /*