pkg/sql: fix child role does not inherit parent role table access

This change ensures that policies correctly respect role inheritance.
Inside the `policy.go` `AppliesToRole` method a call to GetRolesForMember
was added to check all inherited roles for the member also.

Fixes: #144780
Epic: CRDB-11724

Release note: None.

Author: Dedej-Bergin

pkg/sql/logictest/testdata/logic_test/row_level_security

@@ -4288,4 +4288,114 @@ alter_policy_table_locked  CREATE TABLE public.alter_policy_table_locked (
                            ) WITH (schema_locked = true);
                            CREATE POLICY p_sel ON public.alter_policy_table_locked AS PERMISSIVE FOR ALL TO public WITH CHECK (false)
 
+subtest bug_fix_policy_respects_inheritance_#144780
+
+statement ok
+CREATE ROLE parent_role
+
+statement ok
+CREATE ROLE child_role
+
+statement ok
+GRANT parent_role TO child_role
+
+statement ok
+CREATE TABLE employees (id SERIAL PRIMARY KEY, name TEXT, department TEXT)
+
+statement ok
+INSERT INTO employees VALUES (1, 'Alice', 'Engineering')
+
+statement ok
+ALTER TABLE employees ENABLE ROW LEVEL SECURITY
+
+statement ok
+CREATE POLICY parent_policy_select ON employees FOR SELECT TO parent_role USING (true)
+
+statement ok
+CREATE POLICY parent_policy_insert ON employees FOR INSERT TO parent_role WITH CHECK (department IN ('Engineering', 'Sales'))
+
+statement ok
+CREATE POLICY parent_policy_update ON employees FOR UPDATE TO parent_role USING (false)
+
+statement ok
+GRANT SELECT, INSERT, UPDATE ON employees TO parent_role
+
+# Test with role inheritance - child_role should inherit parent_role's permissions.
+statement ok
+SET ROLE child_role
+
+query ITT
+SELECT * FROM employees
+----
+1  Alice  Engineering
+
+# Test INSERT with role inheritance - should succeed for allowed department.
+statement ok
+INSERT INTO employees (id, name, department) VALUES (2, 'Bob', 'Engineering')
+
+# Test INSERT with role inheritance - should succeed for another allowed department.
+statement ok
+INSERT INTO employees (id, name, department) VALUES (3, 'Carol', 'Sales')
+
+# Test INSERT with role inheritance - should fail for disallowed department.
+statement error pq: new row violates row-level security policy for table "employees"
+INSERT INTO employees (id, name, department) VALUES (4, 'Dave', 'Finance')
+
+statement ok
+UPDATE employees SET name = 'Robert' WHERE name = 'Bob'
+
+query ITT
+SELECT * FROM employees ORDER BY id
+----
+1  Alice  Engineering
+2  Bob    Engineering
+3  Carol  Sales
+
+statement ok
+RESET ROLE
+
+# Now revoke the parent role and ensure permissions don't apply anymore.
+statement ok
+REVOKE parent_role FROM child_role
+
+statement ok
+GRANT ALL ON employees TO child_role
+
+statement ok
+SET ROLE child_role
+
+# Should see no rows because the policy doesn't apply anymore.
+query ITT
+SELECT * FROM employees
+----
+
+# All write operations should fail now.
+statement error pq: new row violates row-level security policy for table "employees"
+INSERT INTO employees (id, name, department) VALUES (3, 'Eve', 'Engineering')
+
+# Test UPDATE with role inheritance - should succeed, 
+# but the update will not go as the default with ENABLED ROW LEVEL SECURITY.
+statement ok
+UPDATE employees SET name = 'Alice 2.0' WHERE name = 'Alice'
+
+statement ok
+RESET ROLE
+
+# Can see that no change was made by the update.
+query ITT
+SELECT * FROM employees ORDER BY id
+----
+1  Alice  Engineering
+2  Bob    Engineering
+3  Carol  Sales
+
+statement ok
+DROP TABLE employees CASCADE
+
+statement ok
+DROP ROLE child_role
+
+statement ok
+DROP ROLE parent_role
+
 subtest end

pkg/sql/opt/cat/catalog.go

@@ -236,6 +236,10 @@ type Catalog interface {
 	// UserHasAdminRole checks if the specified user has admin privileges.
 	UserHasAdminRole(ctx context.Context, user username.SQLUsername) (bool, error)
 
+	// UserIsMemberOfAnyRole checks if the specified user is a member of any of the given roles,
+	// either directly or indirectly through role inheritance.
+	UserIsMemberOfAnyRole(ctx context.Context, user username.SQLUsername, roles map[username.SQLUsername]struct{}) (bool, error)
+
 	// HasRoleOption converts the roleoption to its SQL column name and checks if
 	// the user belongs to a role where the option has value true. Requires a
 	// valid transaction to be open.

pkg/sql/opt/cat/policy.go

@@ -6,6 +6,8 @@
 package cat
 
 import (
+	"context"
+
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
@@ -61,7 +63,7 @@ type Policy struct {
 	Command catpb.PolicyCommand
 	// roles are the roles the applies to. If the policy applies to all roles (aka
 	// public), this will be nil.
-	roles map[string]struct{}
+	roles map[username.SQLUsername]struct{}
 }
 
 // Policies contains the policies for a single table.
@@ -76,7 +78,7 @@ func (p *Policy) InitRoles(roleNames []string) {
 		p.roles = nil
 		return
 	}
-	roles := make(map[string]struct{})
+	roles := make(map[username.SQLUsername]struct{})
 	for _, r := range roleNames {
 		if r == username.PublicRole {
 			// If the public role is defined, there is no need to check the
@@ -85,17 +87,23 @@ func (p *Policy) InitRoles(roleNames []string) {
 			roles = nil
 			break
 		}
-		roles[r] = struct{}{}
+		roleUsername := username.MakeSQLUsernameFromPreNormalizedString(r)
+		roles[roleUsername] = struct{}{}
 	}
 	p.roles = roles
 }
 
-// AppliesToRole checks whether the policy applies to the give role.
-func (p *Policy) AppliesToRole(user username.SQLUsername) bool {
+// AppliesToRole checks whether the policy applies to the given role.
+func (p *Policy) AppliesToRole(ctx context.Context, cat Catalog, user username.SQLUsername) bool {
 	// If no roles are specified, assume the policy applies to all users (public role).
 	if p.roles == nil {
 		return true
 	}
-	_, found := p.roles[user.Normalized()]
-	return found
+
+	// Check if the user belongs to any of the roles in the policy
+	belongs, err := cat.UserIsMemberOfAnyRole(ctx, user, p.roles)
+	if err != nil {
+		panic(err)
+	}
+	return belongs
 }

pkg/sql/opt/optbuilder/mutation_builder.go

@@ -1150,7 +1150,7 @@ func (mb *mutationBuilder) genPolicyExpr(
 
 	// Create a closure to handle building the expression for one policy.
 	buildForPolicy := func(p cat.Policy, combineScalars func(opt.ScalarExpr, opt.ScalarExpr) opt.ScalarExpr) {
-		if !p.AppliesToRole(mb.b.checkPrivilegeUser) || !policyAppliesToCommandScope(p, cmdScope) {
+		if !p.AppliesToRole(mb.b.ctx, mb.b.catalog, mb.b.checkPrivilegeUser) || !policyAppliesToCommandScope(p, cmdScope) {
 			return
 		}
 		policiesUsed.Add(p.ID)

pkg/sql/opt/optbuilder/row_level_security.go

@@ -117,7 +117,7 @@ func (b *Builder) buildRowLevelSecurityUsingExpressionForCommand(
 
 	// Create a closure to handle building the expression for one policy.
 	buildForPolicy := func(policy cat.Policy, restrictive bool) {
-		if !policy.AppliesToRole(b.checkPrivilegeUser) || !policyAppliesToCommandScope(policy, cmdScope) {
+		if !policy.AppliesToRole(b.ctx, b.catalog, b.checkPrivilegeUser) || !policyAppliesToCommandScope(policy, cmdScope) {
 			return
 		}
 		strExpr := policy.UsingExpr

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

@@ -343,6 +343,16 @@ func (tc *Catalog) UserHasAdminRole(ctx context.Context, user username.SQLUserna
 	return roleMembership.isMemberOfAdminRole, nil
 }
 
+// UserIsMemberOfAnyRole is part of the cat.Catalog interface.
+func (tc *Catalog) UserIsMemberOfAnyRole(
+	ctx context.Context, user username.SQLUsername, roles map[username.SQLUsername]struct{},
+) (bool, error) {
+	if _, found := roles[user]; found {
+		return true, nil
+	}
+	return false, nil
+}
+
 // HasRoleOption is part of the cat.Catalog interface.
 func (tc *Catalog) HasRoleOption(ctx context.Context, roleOption roleoption.Option) (bool, error) {
 	return true, nil

pkg/sql/opt_catalog.go

@@ -489,6 +489,31 @@ func (oc *optCatalog) UserHasAdminRole(
 	return oc.planner.UserHasAdminRole(ctx, user)
 }
 
+// UserIsMemberOfAnyRole is part of the cat.Catalog interface.
+func (oc *optCatalog) UserIsMemberOfAnyRole(
+	ctx context.Context, user username.SQLUsername, roles map[username.SQLUsername]struct{},
+) (bool, error) {
+	// First check if the user directly matches any of the roles
+	if _, found := roles[user]; found {
+		return true, nil
+	}
+
+	// Get all roles the user belongs to
+	memberRoles, err := oc.planner.MemberOfWithAdminOption(ctx, user)
+	if err != nil {
+		return false, err
+	}
+
+	// Check if any of the target roles are in the user's roles
+	for role := range roles {
+		if _, isMember := memberRoles[role]; isMember {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
 // HasRoleOption is part of the cat.Catalog interface.
 func (oc *optCatalog) HasRoleOption(
 	ctx context.Context, roleOption roleoption.Option,