Include policies based on ACLs needed

When considering which policies should be included, rather than look at
individual bits of the query (eg: if a RETURNING clause exists, or if a
WHERE clause exists which is referencing the table, or if it's a
FOR SHARE/UPDATE query), consider any case where we've determined
the user needs SELECT rights on the relation while doing an UPDATE or
DELETE to be a case where we apply SELECT policies, and any case where
we've deteremind that the user needs UPDATE rights on the relation while
doing a SELECT to be a case where we apply UPDATE policies.

This simplifies the logic and addresses concerns that a user could use
UPDATE or DELETE with a WHERE clauses to determine if rows exist, or
they could use SELECT .. FOR UPDATE to lock rows which they are not
actually allowed to modify through UPDATE policies.

Use list_append_unique() to avoid adding the same quals multiple times,
as, on balance, the cost of checking when adding the quals will almost
always be cheaper than keeping them and doing busywork for each tuple
during execution.

Back-patch to 9.5 where RLS was added.

2 files changed, 101 insertions, 67 deletions

diff --git a/src/backend/rewrite/rowsecurity.c b/src/backend/rewrite/rowsecurity.c
index c20a17802b1..3d7f7a3a18b 100644
--- a/src/backend/rewrite/rowsecurity.c
+++ b/src/backend/rewrite/rowsecurity.c
@@ -165,18 +165,58 @@ get_row_security_policies(Query *root, RangeTblEntry *rte, int rt_index,
 	commandType = rt_index == root->resultRelation ?
 				  root->commandType : CMD_SELECT;
 
-	get_policies_for_relation(rel, commandType, user_id, &permissive_policies,
-							  &restrictive_policies);
+	/*
+	 * In some cases, we need to apply USING policies (which control the
+	 * visibility of records) associated with multiple command types (see
+	 * specific cases below).
+	 *
+	 * When considering the order in which to apply these USING policies,
+	 * we prefer to apply higher privileged policies, those which allow the
+	 * user to lock records (UPDATE and DELETE), first, followed by policies
+	 * which don't (SELECT).
+	 *
+	 * Note that the optimizer is free to push down and reorder quals which
+	 * use leakproof functions.
+	 *
+	 * In all cases, if there are no policy clauses allowing access to rows in
+	 * the table for the specific type of operation, then a single always-false
+	 * clause (a default-deny policy) will be added (see add_security_quals).
+	 */
+
+	/*
+	 * For a SELECT, if UPDATE privileges are required (eg: the user has
+	 * specified FOR [KEY] UPDATE/SHARE), then add the UPDATE USING quals first.
+	 *
+	 * This way, we filter out any records from the SELECT FOR SHARE/UPDATE
+	 * which the user does not have access to via the UPDATE USING policies,
+	 * similar to how we require normal UPDATE rights for these queries.
+	 */
+	if (commandType == CMD_SELECT && rte->requiredPerms & ACL_UPDATE)
+	{
+		List	   *update_permissive_policies;
+		List	   *update_restrictive_policies;
+
+		get_policies_for_relation(rel, CMD_UPDATE, user_id,
+								  &update_permissive_policies,
+								  &update_restrictive_policies);
+
+		add_security_quals(rt_index,
+						   update_permissive_policies,
+						   update_restrictive_policies,
+						   securityQuals,
+						   hasSubLinks);
+	}
 
 	/*
-	 * For SELECT, UPDATE and DELETE, add security quals to enforce these
+	 * For SELECT, UPDATE and DELETE, add security quals to enforce the USING
 	 * policies.  These security quals control access to existing table rows.
 	 * Restrictive policies are "AND"d together, and permissive policies are
 	 * "OR"d together.
-	 *
-	 * If there are no policy clauses controlling access to the table, this
-	 * will add a single always-false clause (a default-deny policy).
 	 */
+
+	get_policies_for_relation(rel, commandType, user_id, &permissive_policies,
+							  &restrictive_policies);
+
 	if (commandType == CMD_SELECT ||
 		commandType == CMD_UPDATE ||
 		commandType == CMD_DELETE)
@@ -187,28 +227,27 @@ get_row_security_policies(Query *root, RangeTblEntry *rte, int rt_index,
 						   hasSubLinks);
 
 	/*
-	 * For the target relation, when there is a returning list, we need to
-	 * collect up CMD_SELECT policies and add them via add_security_quals.
-	 * This is because, for the RETURNING case, we have to filter any records
-	 * which are not visible through an ALL or SELECT USING policy.
+	 * Similar to above, during an UPDATE or DELETE, if SELECT rights are also
+	 * required (eg: when a RETURNING clause exists, or the user has provided
+	 * a WHERE clause which involves columns from the relation), we collect up
+	 * CMD_SELECT policies and add them via add_security_quals first.
 	 *
-	 * We don't need to worry about the non-target relation case because we are
-	 * checking the ALL and SELECT policies for those relations anyway (see
-	 * above).
+	 * This way, we filter out any records which are not visible through an ALL
+	 * or SELECT USING policy.
 	 */
-	if (root->returningList != NIL &&
-		(commandType == CMD_UPDATE || commandType == CMD_DELETE))
+	if ((commandType == CMD_UPDATE || commandType == CMD_DELETE) &&
+		rte->requiredPerms & ACL_SELECT)
 	{
-		List	   *returning_permissive_policies;
-		List	   *returning_restrictive_policies;
+		List	   *select_permissive_policies;
+		List	   *select_restrictive_policies;
 
 		get_policies_for_relation(rel, CMD_SELECT, user_id,
-								  &returning_permissive_policies,
-								  &returning_restrictive_policies);
+								  &select_permissive_policies,
+								  &select_restrictive_policies);
 
 		add_security_quals(rt_index,
-						   returning_permissive_policies,
-						   returning_restrictive_policies,
+						   select_permissive_policies,
+						   select_restrictive_policies,
 						   securityQuals,
 						   hasSubLinks);
 	}
@@ -261,21 +300,22 @@ get_row_security_policies(Query *root, RangeTblEntry *rte, int rt_index,
 								   hasSubLinks);
 
 			/*
-			 * Get and add ALL/SELECT policies, if there is a RETURNING clause,
-			 * also as WCO policies, again, to avoid silently dropping data.
+			 * Get and add ALL/SELECT policies, if SELECT rights are required
+			 * for this relation, also as WCO policies, again, to avoid
+			 * silently dropping data.  See above.
 			 */
-			if (root->returningList != NIL)
+			if (rte->requiredPerms & ACL_SELECT)
 			{
-				List	   *conflict_returning_permissive_policies = NIL;
-				List	   *conflict_returning_restrictive_policies = NIL;
+				List	   *conflict_select_permissive_policies = NIL;
+				List	   *conflict_select_restrictive_policies = NIL;
 
 				get_policies_for_relation(rel, CMD_SELECT, user_id,
-									  &conflict_returning_permissive_policies,
-									  &conflict_returning_restrictive_policies);
+									  &conflict_select_permissive_policies,
+									  &conflict_select_restrictive_policies);
 				add_with_check_options(rel, rt_index,
 									   WCO_RLS_CONFLICT_CHECK,
-									   conflict_returning_permissive_policies,
-									   conflict_returning_restrictive_policies,
+									   conflict_select_permissive_policies,
+									   conflict_select_restrictive_policies,
 									   withCheckOptions,
 									   hasSubLinks);
 			}
@@ -524,7 +564,7 @@ add_security_quals(int rt_index,
 				qual = copyObject(policy->qual);
 				ChangeVarNodes((Node *) qual, 1, rt_index, 0);
 
-				*securityQuals = lappend(*securityQuals, qual);
+				*securityQuals = list_append_unique(*securityQuals, qual);
 				*hasSubLinks |= policy->hassublinks;
 			}
 		}
@@ -539,7 +579,7 @@ add_security_quals(int rt_index,
 			rowsec_expr = makeBoolExpr(OR_EXPR, permissive_quals, -1);
 
 		ChangeVarNodes((Node *) rowsec_expr, 1, rt_index, 0);
-		*securityQuals = lappend(*securityQuals, rowsec_expr);
+		*securityQuals = list_append_unique(*securityQuals, rowsec_expr);
 	}
 	else
 		/*
@@ -631,7 +671,7 @@ add_with_check_options(Relation rel,
 
 		ChangeVarNodes(wco->qual, 1, rt_index, 0);
 
-		*withCheckOptions = lappend(*withCheckOptions, wco);
+		*withCheckOptions = list_append_unique(*withCheckOptions, wco);
 
 		/*
 		 * Now add WithCheckOptions for each of the restrictive policy clauses
@@ -657,7 +697,7 @@ add_with_check_options(Relation rel,
 				wco->qual = (Node *) qual;
 				wco->cascaded = false;
 
-				*withCheckOptions = lappend(*withCheckOptions, wco);
+				*withCheckOptions = list_append_unique(*withCheckOptions, wco);
 				*hasSubLinks |= policy->hassublinks;
 			}
 		}
diff --git a/src/test/regress/expected/rowsecurity.out b/src/test/regress/expected/rowsecurity.out
index 54091e5dede..0cde8fd0e3f 100644
--- a/src/test/regress/expected/rowsecurity.out
+++ b/src/test/regress/expected/rowsecurity.out
@@ -1220,23 +1220,21 @@ NOTICE:  f_leak => cde
 EXPLAIN (COSTS OFF) UPDATE t2 t2_1 SET b = t2_2.b FROM t2 t2_2
 WHERE t2_1.a = 3 AND t2_2.a = t2_1.a AND t2_2.b = t2_1.b
 AND f_leak(t2_1.b) AND f_leak(t2_2.b) RETURNING *, t2_1, t2_2;
-                             QUERY PLAN                              
----------------------------------------------------------------------
+                          QUERY PLAN                           
+---------------------------------------------------------------
  Update on t2 t2_1_1
    ->  Nested Loop
          Join Filter: (t2_1.b = t2_2.b)
          ->  Subquery Scan on t2_1
                Filter: f_leak(t2_1.b)
-               ->  Subquery Scan on t2_1_2
-                     Filter: ((t2_1_2.a % 2) = 1)
-                     ->  LockRows
-                           ->  Seq Scan on t2 t2_1_3
-                                 Filter: ((a = 3) AND ((a % 2) = 1))
+               ->  LockRows
+                     ->  Seq Scan on t2 t2_1_2
+                           Filter: ((a = 3) AND ((a % 2) = 1))
          ->  Subquery Scan on t2_2
                Filter: f_leak(t2_2.b)
                ->  Seq Scan on t2 t2_2_1
                      Filter: ((a = 3) AND ((a % 2) = 1))
-(14 rows)
+(12 rows)
 
 UPDATE t2 t2_1 SET b = t2_2.b FROM t2 t2_2
 WHERE t2_1.a = 3 AND t2_2.a = t2_1.a AND t2_2.b = t2_1.b
@@ -1251,8 +1249,8 @@ NOTICE:  f_leak => cde
 EXPLAIN (COSTS OFF) UPDATE t1 t1_1 SET b = t1_2.b FROM t1 t1_2
 WHERE t1_1.a = 4 AND t1_2.a = t1_1.a AND t1_2.b = t1_1.b
 AND f_leak(t1_1.b) AND f_leak(t1_2.b) RETURNING *, t1_1, t1_2;
-                             QUERY PLAN                              
----------------------------------------------------------------------
+                          QUERY PLAN                           
+---------------------------------------------------------------
  Update on t1 t1_1_3
    Update on t1 t1_1_3
    Update on t2 t1_1
@@ -1261,11 +1259,9 @@ AND f_leak(t1_1.b) AND f_leak(t1_2.b) RETURNING *, t1_1, t1_2;
          Join Filter: (t1_1.b = t1_2.b)
          ->  Subquery Scan on t1_1
                Filter: f_leak(t1_1.b)
-               ->  Subquery Scan on t1_1_4
-                     Filter: ((t1_1_4.a % 2) = 0)
-                     ->  LockRows
-                           ->  Seq Scan on t1 t1_1_5
-                                 Filter: ((a = 4) AND ((a % 2) = 0))
+               ->  LockRows
+                     ->  Seq Scan on t1 t1_1_4
+                           Filter: ((a = 4) AND ((a % 2) = 0))
          ->  Subquery Scan on t1_2
                Filter: f_leak(t1_2.b)
                ->  Append
@@ -1279,11 +1275,9 @@ AND f_leak(t1_1.b) AND f_leak(t1_2.b) RETURNING *, t1_1, t1_2;
          Join Filter: (t1_1_1.b = t1_2_1.b)
          ->  Subquery Scan on t1_1_1
                Filter: f_leak(t1_1_1.b)
-               ->  Subquery Scan on t1_1_6
-                     Filter: ((t1_1_6.a % 2) = 0)
-                     ->  LockRows
-                           ->  Seq Scan on t2 t1_1_7
-                                 Filter: ((a = 4) AND ((a % 2) = 0))
+               ->  LockRows
+                     ->  Seq Scan on t2 t1_1_5
+                           Filter: ((a = 4) AND ((a % 2) = 0))
          ->  Subquery Scan on t1_2_1
                Filter: f_leak(t1_2_1.b)
                ->  Append
@@ -1297,11 +1291,9 @@ AND f_leak(t1_1.b) AND f_leak(t1_2.b) RETURNING *, t1_1, t1_2;
          Join Filter: (t1_1_2.b = t1_2_2.b)
          ->  Subquery Scan on t1_1_2
                Filter: f_leak(t1_1_2.b)
-               ->  Subquery Scan on t1_1_8
-                     Filter: ((t1_1_8.a % 2) = 0)
-                     ->  LockRows
-                           ->  Seq Scan on t3 t1_1_9
-                                 Filter: ((a = 4) AND ((a % 2) = 0))
+               ->  LockRows
+                     ->  Seq Scan on t3 t1_1_6
+                           Filter: ((a = 4) AND ((a % 2) = 0))
          ->  Subquery Scan on t1_2_2
                Filter: f_leak(t1_2_2.b)
                ->  Append
@@ -1311,7 +1303,7 @@ AND f_leak(t1_1.b) AND f_leak(t1_2.b) RETURNING *, t1_1, t1_2;
                            Filter: ((a = 4) AND ((a % 2) = 0))
                      ->  Seq Scan on t3 t1_2_11
                            Filter: ((a = 4) AND ((a % 2) = 0))
-(58 rows)
+(52 rows)
 
 UPDATE t1 t1_1 SET b = t1_2.b FROM t1 t1_2
 WHERE t1_1.a = 4 AND t1_2.a = t1_1.a AND t1_2.b = t1_1.b
@@ -2743,15 +2735,17 @@ SELECT * FROM current_check;
 
 -- Plan should be a subquery TID scan
 EXPLAIN (COSTS OFF) UPDATE current_check SET payload = payload WHERE CURRENT OF current_check_cursor;
-                          QUERY PLAN                           
----------------------------------------------------------------
+                             QUERY PLAN                              
+---------------------------------------------------------------------
  Update on current_check current_check_1
    ->  Subquery Scan on current_check
-         ->  LockRows
-               ->  Tid Scan on current_check current_check_2
-                     TID Cond: CURRENT OF current_check_cursor
-                     Filter: (currentid = 4)
-(6 rows)
+         ->  Subquery Scan on current_check_2
+               Filter: ((current_check_2.currentid % 2) = 0)
+               ->  LockRows
+                     ->  Tid Scan on current_check current_check_3
+                           TID Cond: CURRENT OF current_check_cursor
+                           Filter: (currentid = 4)
+(8 rows)
 
 -- Similarly can only delete row 4
 FETCH ABSOLUTE 1 FROM current_check_cursor;