Dynamic Data Masking 动态数据脱敏

实现思路

  1. 解析查询语句
  2. 脱敏规则验证
  3. 改写SQL/对结果脱敏

具体实现

解析查询语句

主要有两个选择
阿里的druid,作为数据库连接池,它自身也有一个SQL解析模块,开源的分布式数据库中间件mycat就是基于该模块实现的语句解析。
去哪儿的inception,自身有一个语法树打印的功能,能够解析查询语句,最终输出为可视化的query_tree。

druid demo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
package com.tops001.sqlparser.durid.sqlparser;


import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.statement.SQLSelect;
import com.alibaba.druid.sql.ast.statement.SQLSelectQueryBlock;
import com.alibaba.druid.sql.ast.statement.SQLSelectStatement;
import com.alibaba.druid.sql.dialect.mysql.parser.MySqlStatementParser;
import com.alibaba.druid.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor;


public class ParseTest {

public static void main(String[] args) {
String sql = "SELECT\n" +
" a.id,\n" +
" username,\n" +
" b.workflow_name,\n" +
" reviewok_time\n" +
"FROM archer.sql_users a\n" +
" JOIN sql_workflow b ON a.username = b.engineer;\n";

// 新建 MySQL Parser
MySqlStatementParser parser = new MySqlStatementParser(sql);

// 使用Parser解析生成AST,这里SQLStatement就是AST
SQLStatement statement = parser.parseStatement();

// 使用visitor来访问AST
MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor();
statement.accept(visitor);

// 从visitor中拿出你所关注的信息
System.out.println(visitor.getTables());
System.out.println(visitor.getColumns());

//使用mysql解析
MySqlStatementParser mySqlStatementParser = new MySqlStatementParser(sql) ;

//解析select查询
SQLSelectStatement sqlSelectStatement = (SQLSelectStatement) mySqlStatementParser.parseSelect();
SQLSelect sqlSelect = sqlSelectStatement.getSelect() ;

//获取sql查询块
SQLSelectQueryBlock sqlSelectQueryBlock = (SQLSelectQueryBlock)sqlSelect.getQuery() ;

System.out.println(sqlSelectQueryBlock.getSelectList());
}



}

解析结果

1
2
3
{archer.sql_users=Select, sql_workflow=Select}
[archer.sql_users.username, sql_workflow.engineer, archer.sql_users.id, UNKNOWN.username, sql_workflow.workflow_name, UNKNOWN.reviewok_time]
[a.id, username, b.workflow_name, reviewok_time]

inception demo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
#!/usr/bin/python
# coding: utf-8
import MySQLdb
import json

sqlcontent = '''
use archer;
SELECT
a.id,
username,
b.workflow_name,
reviewok_time
FROM archer.sql_users a
JOIN sql_workflow b ON a.username = b.engineer;
'''
sql = "/*--user=;--password=;--host=;--port=;--enable-query-print;*/ \
inception_magic_start;\
%s\
inception_magic_commit;" % sqlcontent
try:
conn = MySQLdb.connect(host='', user='', passwd='', db='', port=6999)
cur = conn.cursor()
ret = cur.execute(sql)
result = cur.fetchall()
num_fields = len(cur.description)
field_names = [i[0] for i in cur.description]
print (field_names)
for row in result:
print (
row[0], "|", row[1], "|", row[2], "|", row[3], "|", row[4])
# print (
# row[0], "|", row[1], "|", row[2], "|", row[3], "|", row[4], "|", row[5], "|", row[6], "|", row[7], "|",
# row[8], "|", row[9], "|", row[10])
cur.close()
conn.close()
except MySQLdb.Error as e:
print ("Mysql Error %d: %s" % (e.args[0], e.args[1]))

解析结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
{
"command": "select",
"select_list": [{
"type": "FIELD_ITEM",
"db": "archer",
"table": "sql_users",
"field": "id"
}, {
"type": "FIELD_ITEM",
"db": "archer",
"table": "sql_users",
"field": "username"
}, {
"type": "FIELD_ITEM",
"db": "archer",
"table": "sql_workflow",
"field": "workflow_name"
}, {
"type": "FIELD_ITEM",
"db": "archer",
"table": "sql_workflow",
"field": "reviewok_time"
}],
"table_ref": [{
"db": "archer",
"table": "sql_users"
}, {
"db": "archer",
"table": "sql_workflow"
}],
"join_on": [{
"type": "FUNC_ITEM",
"func": "=",
"args": [{
"type": "FIELD_ITEM",
"db": "archer",
"table": "sql_users",
"field": "username"
}, {
"type": "FIELD_ITEM",
"db": "archer",
"table": "sql_workflow",
"field": "engineer"
}]
}]
}

由于archer本身是基于inception做的SQL上线审核,最后SQL查询解析模块也顺势选择了inception,但是inception有一些已知的无法解析的语法,如子查询嵌套,但用于简单查询已足够。

脱敏规则

脱敏规则定义表

1
2
3
4
5
6
7
8
9
10
CREATE TABLE `data_masking_rules` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`rule_type` TINYINT NOT NULL DEFAULT 0 COMMENT '规则类型,1手机号,2姓名,3证件号码,4银行卡,5邮箱',
`rule_regex` varchar(255) NOT NULL DEFAULT '' COMMENT'规则脱敏所用的正则表达式,表达式必须分组,隐藏的组会使用****代替',
`hide_group` TINYINT NOT NULL DEFAULT 0 COMMENT'需要隐藏的组',
`rule_desc` varchar(100) NOT NULL DEFAULT '规则描述',
`sys_time` TIMESTAMP NOT NULL DEFAULT current_timestamp COMMENT '修改时间',
PRIMARY KEY (`id`),
UNIQUE KEY `rule_type` (`rule_type`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

脱敏字段定义表

1
2
3
4
5
6
7
8
9
10
11
12
13
CREATE TABLE `data_masking_columns` (
`column_id` int(11) NOT NULL AUTO_INCREMENT,
`rule_type` TINYINT NOT NULL DEFAULT 0 COMMENT '规则类型,1手机号,2姓名,3证件号码,4银行卡,5邮箱',
`active` TINYINT DEFAULT 0 COMMENT '激活状态,0未激活,1激活',
`table_schema` varchar(64) NOT NULL DEFAULT '' COMMENT '字段所在库名',
`table_name` varchar(64) NOT NULL DEFAULT '' COMMENT '字段所在表名',
`column_name` varchar(64) NOT NULL DEFAULT '' COMMENT '字段名',
`column_comment` varchar(1024) NOT NULL DEFAULT '' COMMENT '字段描述',
`create_time` TIMESTAMP NOT NULL DEFAULT current_timestamp COMMENT '创建时间',
`sys_time` TIMESTAMP NOT NULL DEFAULT current_timestamp COMMENT '修改时间',
`cluster_name` varchar(50) NOT NULL DEFAULT '' COMMENT '字段所在集群',
PRIMARY KEY (`column_id`),
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

数据脱敏

改写SQL

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
手机号:保留前三后四,中间用*代替  
SELECT
insert(phone, 4, 4, '****'),
phone
FROM user
WHERE phone<>''
LIMIT 100;

邮箱:去除后缀
SELECT
substring_index(email, '@', 1),
email
FROM user
WHERE email<>''
LIMIT 100;

正则处理结果集

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
# -*- coding:utf-8 -*-
from .inception import InceptionDao
from .models import DataMaskingRules, DataMaskingColumns
import json
import re

inceptionDao = InceptionDao()


class Masking(object):
# 脱敏数据
def data_masking(self, cluster_name, db_name, sql, sql_result):
result = {'status': 0, 'msg': 'ok', 'data': []}
# 通过inception获取语法树,并进行解析
print_info = self.query_tree(sql, cluster_name, db_name)
if print_info is None:
result['status'] = 1
result['msg'] = 'inception返回的结果集为空!可能是SQL语句有语法错误'
elif print_info['errlevel'] != 0:
result['status'] = 2
result['msg'] = 'inception返回异常:\n' + print_info['errmsg']
else:
query_tree = print_info['query_tree']
# 获取集群所属环境,获取命中脱敏规则的列数据
table_hit_columns, hit_columns = self.analy_query_tree(query_tree, cluster_name)

# 存在select * 的查询,遍历column_list,获取命中列的index,添加到hit_columns
if table_hit_columns and sql_result.get('rows'):
column_list = sql_result['column_list']
table_hit_column = {}
for column_info in table_hit_columns:
table_hit_column_info = {}
rule_type = column_info['rule_type']
table_hit_column_info[column_info['column_name']] = rule_type
table_hit_column.update(table_hit_column_info)

for index, item in enumerate(column_list):
if item in table_hit_column.keys():
column = {}
column['column_name'] = item
column['index'] = index
column['rule_type'] = table_hit_column.get(item)
hit_columns.append(column)

# 对命中规则列hit_columns的数据进行脱敏
# 获取全部脱敏规则信息,减少循环查询,提升效率
DataMaskingRulesOb = DataMaskingRules.objects.all()
if hit_columns and sql_result.get('rows'):
rows = list(sql_result['rows'])
for column in hit_columns:
index = column['index']
for idx, item in enumerate(rows):
rows[idx] = list(item)
rows[idx][index] = self.regex(DataMaskingRulesOb, column['rule_type'], rows[idx][index])
sql_result['rows'] = rows
return result

# 通过inception获取语法树
def query_tree(self, sqlContent, cluster_name, dbName):
print_info = inceptionDao.query_print(sqlContent, cluster_name, dbName)
if print_info:
id = print_info[0][0]
statement = print_info[0][1]
# 返回值为非0的情况下,说明是有错的,1表示警告,不影响执行,2表示严重错误,必须修改
errlevel = print_info[0][2]
query_tree = print_info[0][3]
errmsg = print_info[0][4]
# 提交给inception语法错误的情况
if errmsg == 'Global environment':
errlevel = 2
errmsg = 'Global environment: ' + query_tree
if errlevel == 0:
print(json.dumps(json.loads(query_tree), indent=4, sort_keys=False, ensure_ascii=False))
return {'id': id, 'statement': statement, 'errlevel': errlevel, 'query_tree': query_tree,
'errmsg': errmsg}
else:
return None

# 解析语法树,获取语句涉及的表,用于查询权限限制
def query_table_ref(self, sqlContent, cluster_name, dbName):
result = {'status': 0, 'msg': 'ok', 'data': []}
print_info = self.query_tree(sqlContent, cluster_name, dbName)
if print_info is None:
result['status'] = 1
result['msg'] = 'inception返回的结果集为空!可能是SQL语句有语法错误'
elif print_info['errlevel'] != 0:
result['status'] = 2
result['msg'] = 'inception返回异常:\n' + print_info['errmsg']
else:
table_ref = json.loads(print_info['query_tree'])['table_ref']
result['data'] = table_ref
return result

# 解析query_tree,获取语句信息,并返回命中脱敏规则的列信息
def analy_query_tree(self, query_tree, cluster_name):
query_tree_dict = json.loads(query_tree)
select_list = query_tree_dict.get('select_list')
table_ref = query_tree_dict.get('table_ref')

# 获取全部脱敏字段信息,减少循环查询,提升效率
DataMaskingColumnsOb = DataMaskingColumns.objects.all()

# 遍历select_list
columns = []
hit_columns = [] # 命中列
table_hit_columns = [] # 涉及表命中的列

# 获取select信息的规则,仅处理type为FIELD_ITEM的select信息,如[*],[*,column_a],[column_a,*],[column_a,a.*,column_b],[a.*,column_a,b.*],
select_index = [select_item['field'] for select_item in select_list if
select_item['type'] == 'FIELD_ITEM']

if select_index:
# 如果发现存在field='*',则遍历所有表,找出所有的命中字段
if '*' in select_index:
for table in table_ref:
hit_columns_info = self.hit_table(DataMaskingColumnsOb, cluster_name, table['db'],
table['table'])
table_hit_columns.extend(hit_columns_info)
# [*]
if re.match(r"^(\*,?)+$", ','.join(select_index)):
hit_columns = []
# [*,column_a]
elif re.match(r"^(\*,)+(\w,?)+$", ','.join(select_index)):
# 找出field不为* 的列信息, 循环判断列是否命中脱敏规则,并增加规则类型和index,index采取后切片
for index, item in enumerate(select_list):
if item['type'] == 'FIELD_ITEM':
item['index'] = index - len(select_list)
if item['field'] != '*':
columns.append(item)

for column in columns:
hit_info = self.hit_column(DataMaskingColumnsOb, cluster_name, column['db'],
column['table'], column['field'])
if hit_info['is_hit']:
hit_info['index'] = column['index']
hit_columns.append(hit_info)
# [column_a, *]
elif re.match(r"^(\w,?)+(\*,?)+$", ','.join(select_index)):
# 找出field不为* 的列信息, 循环判断列是否命中脱敏规则,并增加规则类型和index,index采取前切片
for index, item in enumerate(select_list):
if item['type'] == 'FIELD_ITEM':
item['index'] = index
if item['field'] != '*':
columns.append(item)

for column in columns:
hit_info = self.hit_column(DataMaskingColumnsOb, cluster_name, column['db'],
column['table'], column['field'])
if hit_info['is_hit']:
hit_info['index'] = column['index']
hit_columns.append(hit_info)
# [column_a,a.*,column_b]
elif re.match(r"^(\w,?)+(\*,?)+(\w,?)+$", ','.join(select_index)):
# 找出field不为* 的列信息, 循环判断列是否命中脱敏规则,并增加规则类型和index,*前面的字段index采取前切片,*后面的字段采取后切片
for index, item in enumerate(select_list):
if item['type'] == 'FIELD_ITEM':
item['index'] = index
if item['field'] == '*':
first_idx = index
break

select_list.reverse()
for index, item in enumerate(select_list):
if item['type'] == 'FIELD_ITEM':
item['index'] = index
if item['field'] == '*':
last_idx = len(select_list) - index - 1
break

select_list.reverse()
for index, item in enumerate(select_list):
if item['type'] == 'FIELD_ITEM':
if item['field'] != '*' and index < first_idx:
item['index'] = index
columns.append(item)

if item['field'] != '*' and index > last_idx:
item['index'] = index - len(select_list)
columns.append(item)

for column in columns:
hit_info = self.hit_column(DataMaskingColumnsOb, cluster_name, column['db'],
column['table'], column['field'])
if hit_info['is_hit']:
hit_info['index'] = column['index']
hit_columns.append(hit_info)

# [a.*, column_a, b.*]
else:
hit_columns = []
return table_hit_columns, hit_columns
# 没有*的查询,直接遍历查询命中字段,query_tree的列index就是查询语句列的index
else:
for index, item in enumerate(select_list):
if item['type'] == 'FIELD_ITEM':
item['index'] = index
if item['field'] != '*':
columns.append(item)

for column in columns:
hit_info = self.hit_column(DataMaskingColumnsOb, cluster_name, column['db'], column['table'],
column['field'])
if hit_info['is_hit']:
hit_info['index'] = column['index']
hit_columns.append(hit_info)
return table_hit_columns, hit_columns

# 判断字段是否命中脱敏规则,如果命中则返回脱敏的规则id和规则类型
def hit_column(self, DataMaskingColumnsOb, cluster_name, table_schema, table_name, column_name):
column_info = DataMaskingColumnsOb.filter(cluster_name=cluster_name, table_schema=table_schema,
table_name=table_name, column_name=column_name, active=1)

hit_column_info = {}
hit_column_info['cluster_name'] = cluster_name
hit_column_info['table_schema'] = table_schema
hit_column_info['table_name'] = table_name
hit_column_info['column_name'] = column_name
hit_column_info['rule_type'] = 0
hit_column_info['is_hit'] = False

# 命中规则
if column_info:
hit_column_info['rule_type'] = column_info[0].rule_type
hit_column_info['is_hit'] = True

return hit_column_info

# 获取表中所有命中脱敏规则的字段信息
def hit_table(self, DataMaskingColumnsOb, cluster_name, table_schema, table_name):
columns_info = DataMaskingColumnsOb.filter(cluster_name=cluster_name, table_schema=table_schema,
table_name=table_name, active=1)

# 命中规则
hit_columns_info = []
for column in columns_info:
hit_column_info = {}
hit_column_info['cluster_name'] = cluster_name
hit_column_info['table_schema'] = table_schema
hit_column_info['table_name'] = table_name
hit_column_info['is_hit'] = True
hit_column_info['column_name'] = column.column_name
hit_column_info['rule_type'] = column.rule_type
hit_columns_info.append(hit_column_info)
return hit_columns_info

# 利用正则表达式脱敏数据
def regex(self, DataMaskingRulesOb, rule_type, str):
rules_info = DataMaskingRulesOb.get(rule_type=rule_type)
if rules_info:
rule_regex = rules_info.rule_regex
hide_group = rules_info.hide_group
# 正则匹配必须分组,隐藏的组会使用****代替
try:
p = re.compile(rule_regex)
m = p.search(str)
masking_str = ''
for i in range(m.lastindex):
if i == hide_group-1:
group = '****'
else:
group = m.group(i+1)
masking_str = masking_str + group
return masking_str
except Exception:
return str
else:
return str