PostgreSQL行级安全策略探究

前言

最近和朋友讨论oracle行级安全策略(VPD)时，查看了下官方文档，看起来VPD的原理是针对应用了Oracle行级安全策略的表、视图或同义词发出的 SQL 语句动态添加where子句。通俗理解就是将行级安全策略动态添加为where 条件。那么PG中的行级安全策略是怎么处理的呢？

行级安全简介

行级安全策略(Row Level Security)是更细粒度的数据安全控制策略。行级策略可以根据每个用户限制哪些行可以通过常规查询返回，哪些行可以通过数据修改命令插入、更新或删除。默认情况下，表没有任何行级安全策略，因此如果用户根据 SQL 权限系统具有表的访问权限，则其中的所有行都可以平等地用于查询或更新。

在PG中我们可以创建行级策略，在SQL执行时行级策略表达式将作为查询的一部分运行。
https://www.postgresql.org/docs/16/ddl-rowsecurity.html

行级安全演示

创建3个用户

postgres=# create user admin;
CREATE ROLE
postgres=# create user peter;
CREATE ROLE
postgres=# create user bob;
CREATE ROLE

创建一个rlsdb数据库

postgres=# create database rlsdb owner admin;
CREATE DATABASE

在rlsdb中使用admin用户创建表employee，并插入3个用户对应的数据

postgres=# \c rlsdb admin
You are now connected to database "rlsdb" as user "admin".
rlsdb=> create table employee ( empno int, ename text, address text, salary int, account_number text );
CREATE TABLE
rlsdb=> insert into employee values (1, 'admin', '2 down str',  80000, 'no0001' );
INSERT 0 1
rlsdb=> insert into employee values (2, 'peter', '132 south avn',  60000, 'no0002' );
INSERT 0 1
rlsdb=> insert into employee values (3, 'bob', 'Down st 17th',  60000, 'no0003' );
INSERT 0 1
rlsdb=>

授权后，三个用户都能看到employee表的所有数据

rlsdb=> grant select on table employee to peter;
GRANT
rlsdb=> grant select on table employee to bob;
GRANT
rlsdb=> select * from employee;
 empno | ename |    address    | salary | account_number 
-------+-------+---------------+--------+----------------
     1 | admin | 2 down str    |  80000 | no0001
     2 | peter | 132 south avn |  60000 | no0002
     3 | bob   | Down st 17th  |  60000 | no0003
(3 rows)

rlsdb=> 
rlsdb=> \c rlsdb peter
You are now connected to database "rlsdb" as user "peter".
rlsdb=> select * from employee;
 empno | ename |    address    | salary | account_number 
-------+-------+---------------+--------+----------------
     1 | admin | 2 down str    |  80000 | no0001
     2 | peter | 132 south avn |  60000 | no0002
     3 | bob   | Down st 17th  |  60000 | no0003
(3 rows)

rlsdb=> \c rlsdb bob
You are now connected to database "rlsdb" as user "bob".
rlsdb=> select * from employee;
 empno | ename |    address    | salary | account_number 
-------+-------+---------------+--------+----------------
     1 | admin | 2 down str    |  80000 | no0001
     2 | peter | 132 south avn |  60000 | no0002
     3 | bob   | Down st 17th  |  60000 | no0003
(3 rows)

使用admin用户创建行级安全策略，对于peter和bob就只能看到自己的数据了。

rlsdb=> \c rlsdb admin
You are now connected to database "rlsdb" as user "admin".
rlsdb=> CREATE POLICY emp_rls_policy ON employee FOR ALL TO PUBLIC USING (ename=current_user);
CREATE POLICY
rlsdb=> ALTER TABLE employee ENABLE ROW LEVEL SECURITY;
ALTER TABLE
rlsdb=> \c rlsdb peter
You are now connected to database "rlsdb" as user "peter".
rlsdb=> select * from employee;
 empno | ename |    address    | salary | account_number 
-------+-------+---------------+--------+----------------
     2 | peter | 132 south avn |  60000 | no0002
(1 row)

rlsdb=> \c rlsdb bob
You are now connected to database "rlsdb" as user "bob".
rlsdb=> select * from employee;
 empno | ename |   address    | salary | account_number 
-------+-------+--------------+--------+----------------
     3 | bob   | Down st 17th |  60000 | no0003
(1 row)

rlsdb=>

行级安全原理

先看下行级安全策略在数据库中的呈现是什么样的。
查看pg_policy表，可以看到我们创建的emp_rls_policy这个策略，具体的策略polqual是一串字符，熟悉parsetree结构的朋友能关注到这是一个OPEXPR node。我们常见的where 条件也是类似的结构。

我们可以使用函数让polqual以更适合人阅读的方式来展示。

创建策略时，其实是将策略转换为where子句存到pg_policy表中。

ObjectAddress
CreatePolicy(CreatePolicyStmt *stmt)
{
/*省略部分代码行*/
/*将策略转化为where子句*/
    qual = transformWhereClause(qual_pstate,
                                stmt->qual,
                                EXPR_KIND_POLICY,
                                "POLICY");
    with_check_qual = transformWhereClause(with_check_pstate,
                                           stmt->with_check,
                                           EXPR_KIND_POLICY,
                                           "POLICY");
    /* Fix up collation information */
    assign_expr_collations(qual_pstate, qual);
    assign_expr_collations(with_check_pstate, with_check_qual);
/* 将转换后的子句写入pg_policy*/
    /* Open pg_policy catalog */
    pg_policy_rel = table_open(PolicyRelationId, RowExclusiveLock);
    /* Set key - policy's relation id. */
    ScanKeyInit(&skey[0],
                Anum_pg_policy_polrelid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(table_id));
    /* Set key - policy's name. */
    ScanKeyInit(&skey[1],
                Anum_pg_policy_polname,
                BTEqualStrategyNumber, F_NAMEEQ,
                CStringGetDatum(stmt->policy_name));
    sscan = systable_beginscan(pg_policy_rel,
                               PolicyPolrelidPolnameIndexId, true, NULL, 2,
                               skey);
    policy_tuple = systable_getnext(sscan);
    /* Complain if the policy name already exists for the table */
    if (HeapTupleIsValid(policy_tuple))
        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("policy \"%s\" for table \"%s\" already exists",
                        stmt->policy_name, RelationGetRelationName(target_table))));
    policy_id = GetNewOidWithIndex(pg_policy_rel, PolicyOidIndexId,
                                   Anum_pg_policy_oid);
    values[Anum_pg_policy_oid - 1] = ObjectIdGetDatum(policy_id);
    values[Anum_pg_policy_polrelid - 1] = ObjectIdGetDatum(table_id);
    values[Anum_pg_policy_polname - 1] = DirectFunctionCall1(namein,
                                                             CStringGetDatum(stmt->policy_name));
    values[Anum_pg_policy_polcmd - 1] = CharGetDatum(polcmd);
    values[Anum_pg_policy_polpermissive - 1] = BoolGetDatum(stmt->permissive);
    values[Anum_pg_policy_polroles - 1] = PointerGetDatum(role_ids);
    /* Add qual if present. */
    if (qual)
        values[Anum_pg_policy_polqual - 1] = CStringGetTextDatum(nodeToString(qual));
    else
        isnull[Anum_pg_policy_polqual - 1] = true;
    /* Add WITH CHECK qual if present */
    if (with_check_qual)
        values[Anum_pg_policy_polwithcheck - 1] = CStringGetTextDatum(nodeToString(with_check_qual));
    else
        isnull[Anum_pg_policy_polwithcheck - 1] = true;
    policy_tuple = heap_form_tuple(RelationGetDescr(pg_policy_rel), values,
                                   isnull);
    CatalogTupleInsert(pg_policy_rel, policy_tuple);
    /* Record Dependencies */
    target.classId = RelationRelationId;
    target.objectId = table_id;
    target.objectSubId = 0;
    myself.classId = PolicyRelationId;
    myself.objectId = policy_id;
    myself.objectSubId = 0;
    recordDependencyOn(&myself, &target, DEPENDENCY_AUTO);
    recordDependencyOnExpr(&myself, qual, qual_pstate->p_rtable,
                           DEPENDENCY_NORMAL);
    recordDependencyOnExpr(&myself, with_check_qual,
                           with_check_pstate->p_rtable, DEPENDENCY_NORMAL);
    /* Register role dependencies */
    target.classId = AuthIdRelationId;
    target.objectSubId = 0;
    for (i = 0; i < nitems; i++)
    {
        target.objectId = DatumGetObjectId(role_oids[i]);
        /* no dependency if public */
        if (target.objectId != ACL_ID_PUBLIC)
            recordSharedDependencyOn(&myself, &target,
                                     SHARED_DEPENDENCY_POLICY);
    }
    InvokeObjectPostCreateHook(PolicyRelationId, policy_id, 0);
    /* Invalidate Relation Cache */
    CacheInvalidateRelcache(target_table);
    /* Clean up. */
    heap_freetuple(policy_tuple);
    free_parsestate(qual_pstate);
    free_parsestate(with_check_pstate);
    systable_endscan(sscan);
    relation_close(target_table, NoLock);
    table_close(pg_policy_rel, RowExclusiveLock);
    return myself;
}

在SQL执行时，查询重写阶段会将对应的安全策略拼接到parsetree里，最后生成执行计划去执行。
从执行计划来看SQL没有where条件，但是执行计划中存在 Filter: (ename = CURRENT_USER)，证明了这个过程。

rlsdb=> explain analyze select * from employee ;
                                             QUERY PLAN                                              
-----------------------------------------------------------------------------------------------------
 Seq Scan on employee  (cost=0.00..19.15 rows=3 width=104) (actual time=0.010..0.012 rows=1 loops=1)
   Filter: (ename = CURRENT_USER)
   Rows Removed by Filter: 2
 Planning Time: 0.416 ms
 Execution Time: 0.036 ms
(5 rows)

rlsdb=>

再debug验证下这个过程。
给fireRIRrules函数设置断点，进入断点后从stack可以看到目前是在QueryRewrite阶段，结合一些规则进行查询重写。

观察这个时候的parsetree，可以看到还没有将安全策略对应的OPEXPR拼接进来。

等执行到get_row_security_policies函数已获取到表对应安全策略securityQuals。
打印securityQuals可以看到和我们查询pg_policy中的OPEXPR是一致的。

接着将securityQuals加入到rte的list中，这样我们再去打印parsetree就可以看到安全策略securityQuals对应的OPEXPR已经被拼接进来。

然后就是去生成执行计划并执行。

小结

PG的RLS也是将对应的策略动态转换为where子句，在查询重写阶段将安全策略拼接到parsetree，生成执行计划去执行。

行级安全策略，可以提供更精细粒度的表数据权限管理，在一定的场景下，比如只让用户看到自己对应的数据，能做到更安全的权限把控。