数据权限筛选（RLS）的两种实现介绍

Z时代
2024-01-10
分类：综合

database

在应用程序中，尤其是在统计的时候，需要使用数据权限来筛选数据行。简单的说，张三看张三部门的数据，李四看李四部门的数据；或者员工只能看自己的数据，经理可以看部门的数据。这个在微软的文档中叫Row Level Security，字面翻译叫行级数据安全，简称RLS。

要实现RLS，简单的思路就是加Where条件语句来做数据筛选。但是必须是先Where, 也就是在其他Where条件和OrderBy、Fetch Rows 之前执行，否则会对排序、分页查询造成影响。这是一个难点。
另一个难点是如何对现有的业务代码侵入性降到最低——不影响现有查询逻辑的写法，甚至当需要的时候，可以关闭RLS。为了校验数据，必须保持RLS开关的灵活性，尤其是在开发阶段。

下面介绍我在项目中使用过的两种实现方式。

数据权限筛选（RLS）的实现（一） -- Security Policy方式实现
这个主要参考微软的官文介绍实现，分三个步骤， a. 定义Predicate函数, 根据user参数来筛选数据， b. 定义Security Policy, 使用前面指定的Predicate函数, c.在指定表上应用Security Policy。
其中的user, 一种是通过当前连接数据库的登录用户来获取，一种是通过exec sp_set_session_context @key=N"userId", @value=@userId 来传入用户。后者更适合我们在应用查询中使用统一的连接字符串。由于我们数据访问层是通过EF来实现的，所以我们统一在自定义的DbContext类型中做了改造：

 1publicabstractclass RlsDbContext : DbContext
 2{
 3
 4protectedreadonly IUserProvider userProvider;
 5protected RlsDbContext(
 6string connectionString,
 7    IUserProvider userProvider)
 8     : base(options)
 9    {
10this.connectionString = connectionString;
11this.userProvider = userProvider;
12    }
13
14protectedoverridevoid OnConfiguring(DbContextOptionsBuilder optionsBuilder)
15    {
16         connection = new SqlConnection(connectionString);
17if (enableRLS)
18        {
19             connection.StateChange += Connection_StateChange;
20        }
21
22if (!enableMemoryDb)
23        {
24            optionsBuilder.UseSqlServer(connection);
25        }
26
27base.OnConfiguring(optionsBuilder);
28    }
29
30privatevoid Connection_StateChange(object sender, System.Data.StateChangeEventArgs e)
31    {
32if (e.CurrentState == ConnectionState.Open)
33        {
34string userId = userProvider.CurrentUserId;
35//此处判断条件用于流程Hook接口未配置认证而获取不到用户的情况
36if (!string.IsNullOrEmpty(userId))
37            {
38                 SqlCommand cmd = connection.CreateCommand();
39                 cmd.CommandText = @"exec sp_set_session_context @key=N"userId", @value=@userId";
40                 cmd.Parameters.AddWithValue("@userId", userId);
41                cmd.ExecuteNonQuery();
42            }
43        }
44elseif (e.CurrentState == ConnectionState.Closed)
45        {
46//暂时注释：在分页查询场景下存在RLS获取总数之前SQL连接关闭的情况
47//connection.StateChange -= Connection_StateChange;
48        }
49    }
50
51 }

这样，我们就能确保在访问数据库的适合，传入了当前用户信息

具体的示例，可以参考《Row-Level Security》
但是这个方式有个很大的问题，就是性能不理想，尤其是在判断条件中有or逻辑的时候。比如这个场景：每个部门只能看自己的数据，如果是数据管理员，不论在哪个部门，可以看所有部门的数据。加了or逻辑后，大概1w行数据查询需要10s钟，这超出了应用能接收的范围。示例Predicate Function如下

 1CREATEFUNCTION[dbo].[Predicate_MyFilter_RLS]
 2(
 3@orgIdnvarchar(200)
 4)
 5RETURNSTABLE
 6WITH SCHEMABINDING
 7AS
 8RETURN
 9SELECTTOP11AS AccessPredicateResult
10FROM dbo.[User] a
11WHERE
12        a.UserId = SESSION_CONTEXT(N"UserId")
13AND
14      (
15         a.OrgId =@orgIdOR a.OrgId ="0000000000000000000000"
16      )
17GO

关于性能问题的佐证，可以参考《Row-Level Security for Middle-Tier Apps – Using Disjunctions in the Predicate》

由于性能问题的障碍，所以我们放弃了这种实现方式。但是这种方式比较优雅的满足了上述的两个条件，即实现了底层数据先筛选的逻辑，也对业务查询方法无侵入。在简单的场景中，应该是一款适合的方案。

数据权限筛选（RLS）的实现（二） -- 后台RlsStrategy方式实现
另一种做法，是我们自行研究的RlsStrategy的实现方式。首先我们了解下接口IRlsStragety

 1publicinterface IRlsStragety<TEntity, TUserConstraintEntity>
 2{
 3     Expression<Func<TUserConstraintEntity, bool>> UserPredicate
 4    {
 5get;
 6    }
 7
 8     Expression<Func<TEntity, object>> OuterKeySelector
 9    {
10get;
11    }
12
13     Expression<Func<TUserConstraintEntity, object>> InnerKeySelector
14    {
15get;
16    }
17
18bool Skip();
19 }

这里面提供了三个表达式和一个bool 方法判断是否要略过RLS筛选。
下面是一个基本的实现：

 1publicclass GenericUserOrgRlsStragety<TEntity, TOrgUser> : IRlsStragety<TEntity, TOrgUser>
 2where TEntity : class, IUserId
 3where TOrgUser : class, IOrgUser
 4{
 5privatereadonly IOrgProvider userOrgProvider;
 6public GenericUserOrgRlsStragety(IOrgProvider userOrgProvider)
 7    {
 8this.userOrgProvider = userOrgProvider;
 9    }
10
11publicvirtual Expression<Func<TOrgUser, bool>> UserPredicate
12     => user => user.OrgId == userOrgProvider.CurrentUserOrgId;
13
14publicvirtual Expression<Func<TEntity, object>> OuterKeySelector
15     => entry => entry.UserId;
16
17publicvirtual Expression<Func<TOrgUser, object>> InnerKeySelector
18     => user => user.UserId;
19
20publicvirtualbool Skip()
21    {
22returnfalse;
23    }
24 }

下面我来解释下这个逻辑。假设应用中有这样两张表
T_BizData(Id, BizAmount, Org) 和T_OrgUser(Org, User), 前者是业务表，记录了业务数据和所属业务组织的机构，后者是机构人员表，记录了人员和机构之间的关系。根据这两个表，我们可以实现OrgA的用户可以查看OrgA的数据， OrgB的用户可以查看OrgB的数据

如果不考虑RLS，则查询语句是

Select*from T_BizData

如果考虑RLS，则查询语句是

Select a.*from T_BizData a
innerjoin T_OrgUser b on a.Org=b.org
where b.User=@user

两者比较，我们发现多了一个限制表和三处灵活点：
1 限制表就是 inner join T_OrgUser b,
2 灵活点 a）取左表属性； b）取右表属性； c）取右表条件判断

这三个灵活点就是我们接口定义的三个表达式，限制表是作为泛型类型传入进来的。

理解了这一点，我们就可以看看下面这个代码

 1publicstatic IQueryable<TEntity> FilterByUser<TDbContext, TEntity, TUserConstraintEntity>(
 2this IQueryable<TEntity> queryable,
 3                TDbContext dbContext,
 4                 IRlsStragety<TEntity, TUserConstraintEntity> rlsStragety
 5                )
 6where TDbContext : DbContext
 7where TEntity : class
 8where TUserConstraintEntity : class, IUserId
 9        {
10if (dbContext isnull)
11            {
12thrownew System.ArgumentNullException(nameof(dbContext));
13            }
14
15if (rlsStragety == null
16                 || rlsStragety.UserPredicate == null
17                 || rlsStragety.OuterKeySelector == null
18                 || rlsStragety.InnerKeySelector == null
19                 || rlsStragety.Skip()
20                )
21            {
22return queryable;
23            }
24
25
26             IQueryable<TEntity> result = queryable.Join(
27                        dbContext.Set<TUserConstraintEntity>()
28                                .Where(rlsStragety.UserPredicate)
29                     , rlsStragety.OuterKeySelector
30                     , rlsStragety.InnerKeySelector
31                      , (p, q) => p
32                   );
33return result;
34         }

我们都知道queryable 是EF实现查询的对象，它描述了查询的过程，所以我们在原queryable对象的基础上扩充了join逻辑，从而实现了类似sql 语句的两表inner join查询。该过程是在分页之前加入的，这样才能保证查询的结果。

 1publicvirtualasync Task<IPaged<TEntity>> GetPagedListAsync<TEntity>(object filter, CancellationToken cancellationToken = default) where TEntity : class
 2        {
 3if (filter == null)
 4            {
 5                 filter = newobject();
 6            }
 7             IPaged<TEntity> result = new Paged<TEntity>();
 8
 9             IQueryable<TEntity> queryable = GetPagedQueryable<TEntity>(filter);
10             result.Rows = await queryable.ToListAsync(cancellationToken).ConfigureAwait(false);
11
12             IQueryable<TEntity> queryableForCount = GetCountQueryable<TEntity>(filter);
13             result.Total = await queryableForCount.CountAsync(cancellationToken).ConfigureAwait(false);
14
15return result;
16         }

以上准备工作做好了，在查询的时候，就可以这样写了：

stragety =
serviceProvider.GetService<MyRlsStragety>();
var pageList = await rlsDataInquirer.GetPagedListAsync(filter, stragety);

最后，补充下skip()方法的逻辑。

publicoverridebool Skip()
        {
string orgId = userOrgProvider.CurrentUserOrgId;
// 如果是信息管理部则跳过关联判断
return orgId.Equals(InfoSupervisorDepartmentOrgId, StringComparison.CurrentCultureIgnoreCase);
        }