ALTER TABLE .. FORCE ROW LEVEL SECURITY

To allow users to force RLS to always be applied, even for table owners,
add ALTER TABLE .. FORCE ROW LEVEL SECURITY.

row_security=off overrides FORCE ROW LEVEL SECURITY, to ensure pg_dump
output is complete (by default).

Also add SECURITY_NOFORCE_RLS context to avoid data corruption when
ALTER TABLE .. FORCE ROW SECURITY is being used. The
SECURITY_NOFORCE_RLS security context is used only during referential
integrity checks and is only considered in check_enable_rls() after we
have already checked that the current user is the owner of the relation
(which should always be the case during referential integrity checks).

Back-patch to 9.5 where RLS was added.

src/backend/catalog/heap.c

@@ -802,6 +802,7 @@ InsertPgClassTuple(Relation pg_class_desc,
 	values[Anum_pg_class_relhasrules - 1] = BoolGetDatum(rd_rel->relhasrules);
 	values[Anum_pg_class_relhastriggers - 1] = BoolGetDatum(rd_rel->relhastriggers);
 	values[Anum_pg_class_relrowsecurity - 1] = BoolGetDatum(rd_rel->relrowsecurity);
+	values[Anum_pg_class_relforcerowsecurity - 1] = BoolGetDatum(rd_rel->relforcerowsecurity);
 	values[Anum_pg_class_relhassubclass - 1] = BoolGetDatum(rd_rel->relhassubclass);
 	values[Anum_pg_class_relispopulated - 1] = BoolGetDatum(rd_rel->relispopulated);
 	values[Anum_pg_class_relreplident - 1] = CharGetDatum(rd_rel->relreplident);

src/backend/commands/tablecmds.c

@@ -418,6 +418,7 @@ static void ATExecReplicaIdentity(Relation rel, ReplicaIdentityStmt *stmt, LOCKM
 static void ATExecGenericOptions(Relation rel, List *options);
 static void ATExecEnableRowSecurity(Relation rel);
 static void ATExecDisableRowSecurity(Relation rel);
+static void ATExecForceNoForceRowSecurity(Relation rel, bool force_rls);
 
 static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
 				   ForkNumber forkNum, char relpersistence);
@@ -2929,6 +2930,8 @@ AlterTableGetLockLevel(List *cmds)
 			case AT_SetNotNull:
 			case AT_EnableRowSecurity:
 			case AT_DisableRowSecurity:
+			case AT_ForceRowSecurity:
+			case AT_NoForceRowSecurity:
 				cmd_lockmode = AccessExclusiveLock;
 				break;
 
@@ -3354,6 +3357,8 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
 		case AT_DropOf: /* NOT OF */
 		case AT_EnableRowSecurity:
 		case AT_DisableRowSecurity:
+		case AT_ForceRowSecurity:
+		case AT_NoForceRowSecurity:
 			ATSimplePermissions(rel, ATT_TABLE);
 			/* These commands never recurse */
 			/* No command-specific prep needed */
@@ -3670,6 +3675,12 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
 		case AT_DisableRowSecurity:
 			ATExecDisableRowSecurity(rel);
 			break;
+		case AT_ForceRowSecurity:
+			ATExecForceNoForceRowSecurity(rel, true);
+			break;
+		case AT_NoForceRowSecurity:
+			ATExecForceNoForceRowSecurity(rel, false);
+			break;
 		case AT_GenericOptions:
 			ATExecGenericOptions(rel, (List *) cmd->def);
 			break;
@@ -11048,6 +11059,35 @@ ATExecDisableRowSecurity(Relation rel)
 	heap_freetuple(tuple);
 }
 
+/*
+ * ALTER TABLE FORCE/NO FORCE ROW LEVEL SECURITY
+ */
+static void
+ATExecForceNoForceRowSecurity(Relation rel, bool force_rls)
+{
+	Relation	pg_class;
+	Oid			relid;
+	HeapTuple	tuple;
+
+	relid = RelationGetRelid(rel);
+
+	pg_class = heap_open(RelationRelationId, RowExclusiveLock);
+
+	tuple = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(relid));
+
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for relation %u", relid);
+
+	((Form_pg_class) GETSTRUCT(tuple))->relforcerowsecurity = force_rls;
+	simple_heap_update(pg_class, &tuple->t_self, tuple);
+
+	/* keep catalog indexes current */
+	CatalogUpdateIndexes(pg_class, tuple);
+
+	heap_close(pg_class, RowExclusiveLock);
+	heap_freetuple(tuple);
+}
+
 /*
  * ALTER FOREIGN TABLE <name> OPTIONS (...)
  */

src/backend/parser/gram.y

@@ -2332,6 +2332,20 @@ alter_table_cmd:
 					n->subtype = AT_DisableRowSecurity;
 					$$ = (Node *)n;
 				}
+			/* ALTER TABLE <name> FORCE ROW LEVEL SECURITY */
+			| FORCE ROW LEVEL SECURITY
+				{
+					AlterTableCmd *n = makeNode(AlterTableCmd);
+					n->subtype = AT_ForceRowSecurity;
+					$$ = (Node *)n;
+				}
+			/* ALTER TABLE <name> NO FORCE ROW LEVEL SECURITY */
+			| NO FORCE ROW LEVEL SECURITY
+				{
+					AlterTableCmd *n = makeNode(AlterTableCmd);
+					n->subtype = AT_NoForceRowSecurity;
+					$$ = (Node *)n;
+				}
 			| alter_generic_options
 				{
 					AlterTableCmd *n = makeNode(AlterTableCmd);

src/backend/utils/adt/ri_triggers.c

@@ -3014,7 +3014,8 @@ ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
 	/* Switch to proper UID to perform check as */
 	GetUserIdAndSecContext(&save_userid, &save_sec_context);
 	SetUserIdAndSecContext(RelationGetForm(query_rel)->relowner,
-						   save_sec_context | SECURITY_LOCAL_USERID_CHANGE);
+						   save_sec_context | SECURITY_LOCAL_USERID_CHANGE |
+						   SECURITY_NOFORCE_RLS);
 
 	/* Create the plan */
 	qplan = SPI_prepare(querystr, nargs, argtypes);
@@ -3134,7 +3135,8 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo,
 	/* Switch to proper UID to perform check as */
 	GetUserIdAndSecContext(&save_userid, &save_sec_context);
 	SetUserIdAndSecContext(RelationGetForm(query_rel)->relowner,
-						   save_sec_context | SECURITY_LOCAL_USERID_CHANGE);
+						   save_sec_context | SECURITY_LOCAL_USERID_CHANGE |
+						   SECURITY_NOFORCE_RLS);
 
 	/* Finally we can run the query. */
 	spi_result = SPI_execute_snapshot(qplan,

src/backend/utils/init/miscinit.c

@@ -341,7 +341,7 @@ GetAuthenticatedUserId(void)
  * GetUserIdAndSecContext/SetUserIdAndSecContext - get/set the current user ID
  * and the SecurityRestrictionContext flags.
  *
- * Currently there are two valid bits in SecurityRestrictionContext:
+ * Currently there are three valid bits in SecurityRestrictionContext:
  *
  * SECURITY_LOCAL_USERID_CHANGE indicates that we are inside an operation
  * that is temporarily changing CurrentUserId via these functions.  This is
@@ -359,6 +359,13 @@ GetAuthenticatedUserId(void)
  * where the called functions are really supposed to be side-effect-free
  * anyway, such as VACUUM/ANALYZE/REINDEX.
  *
+ * SECURITY_NOFORCE_RLS indicates that we are inside an operation which should
+ * ignore the FORCE ROW LEVEL SECURITY per-table indication.  This is used to
+ * ensure that FORCE RLS does not mistakenly break referential integrity
+ * checks.  Note that this is intentionally only checked when running as the
+ * owner of the table (which should always be the case for referential
+ * integrity checks).
+ *
  * Unlike GetUserId, GetUserIdAndSecContext does *not* Assert that the current
  * value of CurrentUserId is valid; nor does SetUserIdAndSecContext require
  * the new value to be valid.  In fact, these routines had better not
@@ -401,6 +408,15 @@ InSecurityRestrictedOperation(void)
 	return (SecurityRestrictionContext & SECURITY_RESTRICTED_OPERATION) != 0;
 }
 
+/*
+ * InNoForceRLSOperation - are we ignoring FORCE ROW LEVEL SECURITY ?
+ */
+bool
+InNoForceRLSOperation(void)
+{
+	return (SecurityRestrictionContext & SECURITY_NOFORCE_RLS) != 0;
+}
+
 
 /*
  * These are obsolete versions of Get/SetUserIdAndSecContext that are

src/backend/utils/misc/rls.c

@@ -55,6 +55,7 @@ check_enable_rls(Oid relid, Oid checkAsUser, bool noError)
 	HeapTuple	tuple;
 	Form_pg_class classform;
 	bool		relrowsecurity;
+	bool		relforcerowsecurity;
 	Oid			user_id = checkAsUser ? checkAsUser : GetUserId();
 
 	/* Nothing to do for built-in relations */
@@ -68,6 +69,7 @@ check_enable_rls(Oid relid, Oid checkAsUser, bool noError)
 	classform = (Form_pg_class) GETSTRUCT(tuple);
 
 	relrowsecurity = classform->relrowsecurity;
+	relforcerowsecurity = classform->relforcerowsecurity;
 
 	ReleaseSysCache(tuple);
 
@@ -76,14 +78,46 @@ check_enable_rls(Oid relid, Oid checkAsUser, bool noError)
 		return RLS_NONE;
 
 	/*
-	 * Table owners and BYPASSRLS users bypass RLS.  Note that a superuser
-	 * qualifies as both.  Return RLS_NONE_ENV to indicate that this decision
-	 * depends on the environment (in this case, the user_id).
+	 * BYPASSRLS users always bypass RLS.  Note that superusers are always
+	 * considered to have BYPASSRLS.
+	 *
+	 * Return RLS_NONE_ENV to indicate that this decision depends on the
+	 * environment (in this case, the user_id).
 	 */
-	if (pg_class_ownercheck(relid, user_id) ||
-		has_bypassrls_privilege(user_id))
+	if (has_bypassrls_privilege(user_id))
 		return RLS_NONE_ENV;
 
+	/*
+	 * Table owners generally bypass RLS, except if row_security=true and the
+	 * table has been set (by an owner) to FORCE ROW SECURITY, and this is not
+	 * a referential integrity check.
+	 *
+	 * Return RLS_NONE_ENV to indicate that this decision depends on the
+	 * environment (in this case, the user_id).
+	 */
+	if (pg_class_ownercheck(relid, user_id))
+	{
+		/*
+		 * If row_security=true and FORCE ROW LEVEL SECURITY has been set on
+		 * the relation then we return RLS_ENABLED to indicate that RLS should
+		 * still be applied.  If we are in a SECURITY_NOFORCE_RLS context or if
+		 * row_security=false then we return RLS_NONE_ENV.
+		 *
+		 * The SECURITY_NOFORCE_RLS indicates that we should not apply RLS even
+		 * if the table has FORCE RLS set- IF the current user is the owner.
+		 * This is specifically to ensure that referential integrity checks are
+		 * able to still run correctly.
+		 *
+		 * This is intentionally only done after we have checked that the user
+		 * is the table owner, which should always be the case for referential
+		 * integrity checks.
+		 */
+		if (row_security && relforcerowsecurity && !InNoForceRLSOperation())
+			return RLS_ENABLED;
+		else
+			return RLS_NONE_ENV;
+	}
+
 	/* row_security GUC says to bypass RLS, but user lacks permission */
 	if (!row_security && !noError)
 		ereport(ERROR,