sql: support BYPASSRLS role option for RLS policies

Previously, attempts to use the BYPASSRLS role option with row-level
security (RLS) policies resulted in an unimplemented error. This change
lifts those restrictions and adds proper support for roles or system
privileges with the BYPASSRLS flag, allowing them to bypass RLS policies
on tables where it is enabled.

Closes #136910

Epic: CRDB-11724
Release note: none

Author: spilchen

pkg/sql/authorization.go

@@ -89,6 +89,10 @@ type AuthorizationAccessor interface {
 	// has a global privilege or the corresponding legacy role option.
 	HasGlobalPrivilegeOrRoleOption(ctx context.Context, privilege privilege.Kind) (bool, error)
 
+	// UserHasGlobalPrivilegeOrRoleOption is like HasGlobalPrivilegeOrRoleOption,
+	// except that it is for a specific user.
+	UserHasGlobalPrivilegeOrRoleOption(ctx context.Context, privilege privilege.Kind, user username.SQLUsername) (bool, error)
+
 	// CheckGlobalPrivilegeOrRoleOption checks if the current user has a global privilege
 	// or the corresponding legacy role option, and returns an error if the user does not.
 	CheckGlobalPrivilegeOrRoleOption(ctx context.Context, privilege privilege.Kind) error
@@ -692,7 +696,15 @@ func (p *planner) CheckRoleOption(ctx context.Context, roleOption roleoption.Opt
 func (p *planner) HasGlobalPrivilegeOrRoleOption(
 	ctx context.Context, privilege privilege.Kind,
 ) (bool, error) {
-	ok, err := p.HasPrivilege(ctx, syntheticprivilege.GlobalPrivilegeObject, privilege, p.User())
+	return p.UserHasGlobalPrivilegeOrRoleOption(ctx, privilege, p.User())
+}
+
+// UserHasGlobalPrivilegeOrRoleOption is like HasGlobalPrivilegeOrRoleOption, but
+// is for a specific user.
+func (p *planner) UserHasGlobalPrivilegeOrRoleOption(
+	ctx context.Context, privilege privilege.Kind, user username.SQLUsername,
+) (bool, error) {
+	ok, err := p.HasPrivilege(ctx, syntheticprivilege.GlobalPrivilegeObject, privilege, user)
 	if err != nil {
 		return false, err
 	}
@@ -701,7 +713,7 @@ func (p *planner) HasGlobalPrivilegeOrRoleOption(
 	}
 	maybeRoleOptionName := string(privilege.DisplayName())
 	if roleOption, ok := roleoption.ByName[maybeRoleOptionName]; ok {
-		return p.HasRoleOption(ctx, roleOption)
+		return p.UserHasRoleOption(ctx, user, roleOption)
 	}
 	return false, nil
 }

pkg/sql/logictest/testdata/logic_test/row_level_security

@@ -2919,4 +2919,175 @@ DROP ROLE test_role1;
 statement ok
 DROP ROLE test_role2;
 
+# Ensure that if a role has the BYPASSRLS option or privilege set, that it's
+# exempt from policies.
+subtest bypassrls
+
+statement ok
+CREATE TABLE bypassrls (id INT PRIMARY KEY, val TEXT);
+
+statement ok
+CREATE USER bypassrls_user;
+
+statement ok
+GRANT ALL ON bypassrls TO bypassrls_user;
+
+statement ok
+ALTER TABLE bypassrls ENABLE ROW LEVEL SECURITY;
+
+statement ok
+CREATE POLICY pol1 ON bypassrls TO bypassrls_user USING (val like 'visible: %');
+
+statement ok
+CREATE FUNCTION insert_policy_violation_as_session_user(id INT) RETURNS TABLE(id INT, description TEXT)
+LANGUAGE SQL AS
+$$
+ INSERT INTO bypassrls VALUES (id, 'hidden: value') RETURNING id, val
+$$;
+
+# This function is like the above, except it will always be run as admin since
+# it uses SECURITY DEFINER.
+statement ok
+CREATE FUNCTION insert_policy_violation_as_admin(id INT) RETURNS TABLE(id INT, description TEXT)
+LANGUAGE SQL AS
+$$
+ INSERT INTO bypassrls VALUES (id, 'hidden: value') RETURNING id, val
+$$ SECURITY DEFINER;
+
+statement ok
+INSERT INTO bypassrls VALUES (0, 'visible: 0'), (1, 'hidden: 1'), (2, 'visible: 2');
+
+query IT
+SELECT * FROM insert_policy_violation_as_admin(10);
+----
+10 hidden: value
+
+query IT
+SELECT * FROM insert_policy_violation_as_session_user(11);
+----
+11 hidden: value
+
+query IT
+SELECT * FROM bypassrls ORDER BY id;
+----
+0 visible: 0
+1 hidden: 1
+2 visible: 2
+10 hidden: value
+11 hidden: value
+
+statement ok
+SET ROLE bypassrls_user;
+
+query IT
+SELECT * FROM bypassrls ORDER BY id;
+----
+0 visible: 0
+2 visible: 2
+
+query IT
+SELECT * FROM insert_policy_violation_as_admin(20);
+----
+20 hidden: value
+
+statement error pq: new row violates row-level security policy for table "bypassrls"
+SELECT * FROM insert_policy_violation_as_session_user(21);
+
+statement ok
+SET ROLE root;
+
+statement error pq: conflicting role options
+ALTER ROLE bypassrls_user BYPASSRLS NOBYPASSRLS;
+
+statement ok
+ALTER ROLE bypassrls_user BYPASSRLS;
+
+statement ok
+SET ROLE bypassrls_user;
+
+query IT
+SELECT * FROM bypassrls ORDER BY id;
+----
+0 visible: 0
+1 hidden: 1
+2 visible: 2
+10 hidden: value
+11 hidden: value
+20 hidden: value
+
+query IT
+SELECT * FROM insert_policy_violation_as_admin(30);
+----
+30 hidden: value
+
+query IT
+SELECT * FROM insert_policy_violation_as_session_user(31);
+----
+31 hidden: value
+
+statement ok
+SET ROLE root
+
+statement ok
+ALTER ROLE bypassrls_user NOBYPASSRLS;
+
+statement ok
+GRANT SYSTEM BYPASSRLS TO bypassrls_user;
+
+query TTB colnames
+SELECT * FROM [SHOW SYSTEM GRANTS] WHERE privilege_type = 'BYPASSRLS';
+----
+grantee         privilege_type  is_grantable
+bypassrls_user  BYPASSRLS             false
+
+statement ok
+SET ROLE bypassrls_user;
+
+query IT
+SELECT * FROM bypassrls ORDER BY id;
+----
+0   visible: 0
+1   hidden: 1
+2   visible: 2
+10  hidden: value
+11  hidden: value
+20  hidden: value
+30  hidden: value
+31  hidden: value
+
+query IT
+SELECT * FROM insert_policy_violation_as_session_user(40);
+----
+40 hidden: value
+
+statement ok
+SET ROLE root;
+
+statement ok
+REVOKE SYSTEM BYPASSRLS FROM bypassrls_user;
+
+statement ok
+SET ROLE bypassrls_user;
+
+query IT
+SELECT * FROM bypassrls ORDER BY id;
+----
+0 visible: 0
+2 visible: 2
+
+statement error pq: new row violates row-level security policy for table "bypassrls"
+SELECT * FROM insert_policy_violation_as_session_user(50);
+
+statement ok
+SET ROLE root;
+
+statement ok
+DROP FUNCTION insert_policy_violation_as_admin, insert_policy_violation_as_session_user
+
+statement ok
+DROP TABLE bypassrls;
+
+statement ok
+DROP USER bypassrls_user;
+
 subtest end

pkg/sql/opt/cat/catalog.go

@@ -241,6 +241,10 @@ type Catalog interface {
 	// NOLOGIN instead of LOGIN.
 	HasRoleOption(ctx context.Context, roleOption roleoption.Option) (bool, error)
 
+	// UserHasGlobalPrivilegeOrRoleOption returns a bool representing whether the given user
+	// has a global privilege or the corresponding legacy role option.
+	UserHasGlobalPrivilegeOrRoleOption(ctx context.Context, privilege privilege.Kind, user username.SQLUsername) (bool, error)
+
 	// FullyQualifiedName retrieves the fully qualified name of a data source.
 	// Note that:
 	//  - this call may involve a database operation so it shouldn't be used in

pkg/sql/opt/exec/execbuilder/relational.go

@@ -463,7 +463,7 @@ func (b *Builder) maybeAnnotatePolicyInfo(node exec.Node, e memo.RelExpr) {
 			policiesApplied, found := rlsMeta.PoliciesApplied[tabID]
 			if found {
 				val := exec.RLSPoliciesApplied{
-					PoliciesSkippedForRole: rlsMeta.HasAdminRole || policiesApplied.NoForceExempt,
+					PoliciesSkippedForRole: rlsMeta.HasAdminRole || policiesApplied.NoForceExempt || policiesApplied.BypassRLS,
 				}
 				if applyFilterExpr {
 					val.Policies = policiesApplied.Filter

pkg/sql/opt/memo/memo_test.go

@@ -634,7 +634,8 @@ func TestMemoIsStale(t *testing.T) {
 
 	// User changes (with RLS)
 	o.Memo().Metadata().SetRLSEnabled(evalCtx.SessionData().User(), true, /* admin */
-		1 /* tableID */, false /* isTableOwnerAndNotForced */)
+		1 /* tableID */, false, /* isTableOwnerAndNotForced */
+		false /* bypassRLS */)
 	notStale()
 	evalCtx.SessionData().UserProto = newUser
 	stale()

pkg/sql/opt/metadata.go

@@ -1187,10 +1187,13 @@ func (md *Metadata) TestingPrivileges() map[cat.StableID]privilegeBitmap {
 // SetRLSEnabled will update the metadata to indicate we came across a table
 // that had row-level security enabled.
 func (md *Metadata) SetRLSEnabled(
-	user username.SQLUsername, isAdmin bool, tableID TableID, isTableOwnerAndNotForced bool,
+	user username.SQLUsername,
+	isAdmin bool,
+	tableID TableID,
+	isTableOwnerAndNotForced, bypassRLS bool,
 ) {
 	md.rlsMeta.MaybeInit(user, isAdmin)
-	md.rlsMeta.AddTableUse(tableID, isTableOwnerAndNotForced)
+	md.rlsMeta.AddTableUse(tableID, isTableOwnerAndNotForced, bypassRLS)
 }
 
 // ClearRLSEnabled will clear out the initialized state for the rls meta. This
@@ -1231,8 +1234,26 @@ func (md *Metadata) checkRLSDependencies(
 		return false, nil
 	}
 
-	// We do not check for specific policy changes. Any time a policy is modified
-	// on a table, a new version of the table descriptor is created. The metadata
-	// dependency check already accounts for changes in the table descriptor version.
+	// Check if the current user has a role option/privilege that changed
+	// affecting the exemption of policies.
+	for i := range md.tables {
+		table := &md.tables[i]
+		policiesApplied, ok := md.rlsMeta.PoliciesApplied[table.MetaID]
+		if !ok {
+			continue
+		}
+		bypassRLS, err := optCatalog.UserHasGlobalPrivilegeOrRoleOption(ctx, privilege.BYPASSRLS, md.rlsMeta.User)
+		if err != nil {
+			return false, err
+		}
+		if bypassRLS != policiesApplied.BypassRLS {
+			return false, nil
+		}
+	}
+
+	// We do not check for specific policy changes or exemption due to FORCE RLS.
+	// Any time a policy or table attribute such as forced is modified on a table,
+	// a new version of the table descriptor is created. The metadata dependency
+	// check already accounts for changes in the table descriptor version.
 	return true, nil
 }

pkg/sql/opt/optbuilder/row_level_security.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/cat"
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/memo"
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
+	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/intsets"
@@ -42,9 +43,14 @@ func (b *Builder) addRowLevelSecurityFilter(
 	if err != nil {
 		panic(err)
 	}
-	b.factory.Metadata().SetRLSEnabled(b.checkPrivilegeUser, isAdmin, tabMeta.MetaID, isOwnerAndNotForced)
+	bypassRLS, err := b.catalog.UserHasGlobalPrivilegeOrRoleOption(b.ctx, privilege.BYPASSRLS, b.checkPrivilegeUser)
+	if err != nil {
+		panic(err)
+	}
+	b.factory.Metadata().SetRLSEnabled(b.checkPrivilegeUser, isAdmin, tabMeta.MetaID,
+		isOwnerAndNotForced, bypassRLS)
 	// Check if RLS filtering is exempt.
-	if isAdmin || isOwnerAndNotForced {
+	if isAdmin || isOwnerAndNotForced || bypassRLS {
 		return
 	}
 
@@ -225,8 +231,12 @@ func (r *optRLSConstraintBuilder) genExpression(ctx context.Context) (string, []
 	if err != nil {
 		panic(err)
 	}
-	r.md.SetRLSEnabled(r.user, isAdmin, r.tabMeta.MetaID, isOwnerAndNotForced)
-	if isAdmin || isOwnerAndNotForced {
+	bypassRLS, err := r.oc.UserHasGlobalPrivilegeOrRoleOption(ctx, privilege.BYPASSRLS, r.user)
+	if err != nil {
+		panic(err)
+	}
+	r.md.SetRLSEnabled(r.user, isAdmin, r.tabMeta.MetaID, isOwnerAndNotForced, bypassRLS)
+	if isAdmin || isOwnerAndNotForced || bypassRLS {
 		// Return a constraint check that always passes.
 		return "true", nil
 	}

pkg/sql/opt/row_level_security.go

@@ -54,10 +54,13 @@ func (r *RowLevelSecurityMeta) Clear() {
 // AddTableUse indicates that an RLS-enabled table was encountered while
 // building the query plan. If any policies are in use, they will be added
 // via the AddPolicyUse call.
-func (r *RowLevelSecurityMeta) AddTableUse(tableID TableID, isTableOwnerAndNotForced bool) {
+func (r *RowLevelSecurityMeta) AddTableUse(
+	tableID TableID, isTableOwnerAndNotForced, bypassRLS bool,
+) {
 	if _, found := r.PoliciesApplied[tableID]; !found {
 		r.PoliciesApplied[tableID] = PoliciesApplied{
 			NoForceExempt: isTableOwnerAndNotForced,
+			BypassRLS:     bypassRLS,
 			Filter:        PolicyIDSet{},
 			Check:         PolicyIDSet{},
 		}
@@ -87,6 +90,11 @@ type PoliciesApplied struct {
 	// NoForceExempt is true if the policies were exempt because they were the
 	// table owner and force RLS wasn't set.
 	NoForceExempt bool
+	// BypassRLS is true if the policies were bypassed because the user had the
+	// BYPASSRLS option/privilege. There is an argument to combine the two bools
+	// for exemption. However, the memo staleness checks have different handling
+	// for when these exemptions change, so they need to be stored separately.
+	BypassRLS bool
 	// Filter is the set of policy IDs that were applied to filter out existing
 	// rows. The USING expression in each policy is used to derive the filter.
 	Filter PolicyIDSet
@@ -99,6 +107,7 @@ type PoliciesApplied struct {
 func (p *PoliciesApplied) Copy() PoliciesApplied {
 	return PoliciesApplied{
 		NoForceExempt: p.NoForceExempt,
+		BypassRLS:     p.BypassRLS,
 		Filter:        p.Filter.Copy(),
 		Check:         p.Check.Copy(),
 	}

pkg/sql/opt/testutils/testcat/test_catalog.go

@@ -336,6 +336,13 @@ func (tc *Catalog) HasRoleOption(ctx context.Context, roleOption roleoption.Opti
 	return true, nil
 }
 
+// UserHasGlobalPrivilegeOrRoleOption is part of the cat.Catalog interface.
+func (tc *Catalog) UserHasGlobalPrivilegeOrRoleOption(
+	ctx context.Context, privilege privilege.Kind, user username.SQLUsername,
+) (bool, error) {
+	return false, nil
+}
+
 // FullyQualifiedName is part of the cat.Catalog interface.
 func (tc *Catalog) FullyQualifiedName(
 	ctx context.Context, ds cat.DataSource,

pkg/sql/opt_catalog.go

@@ -470,6 +470,13 @@ func (oc *optCatalog) HasRoleOption(
 	return oc.planner.HasRoleOption(ctx, roleOption)
 }
 
+// UserHasGlobalPrivilegeOrRoleOption is part of the cat.Catalog interface.
+func (oc *optCatalog) UserHasGlobalPrivilegeOrRoleOption(
+	ctx context.Context, privilege privilege.Kind, user username.SQLUsername,
+) (bool, error) {
+	return oc.planner.UserHasGlobalPrivilegeOrRoleOption(ctx, privilege, user)
+}
+
 // FullyQualifiedName is part of the cat.Catalog interface.
 func (oc *optCatalog) FullyQualifiedName(
 	ctx context.Context, ds cat.DataSource,