支持面向用户级别的数据脱敏访问控制，即特定用户只能访问到脱敏后的数据。此方案是实时领域Flink的解决方案，类似于离线数仓Hive中Ranger Column Masking方案。
源码地址: https://github.com/HamaWhiteGG/flink-sql-security
注: 如果用IntelliJ IDEA打开源码，请提前安装 Manifold 插件。
数据脱敏(Data Masking)是一种数据安全技术，用于保护敏感数据，以防止未经授权的访问。该技术通过将敏感数据替换为虚假数据或不可识别的数据来实现。 例如可以使用数据脱敏技术将信用卡号码、社会安全号码等敏感信息替换为随机生成的数字或字母，以保护这些信息的隐私和安全。
下面用订单表orders的两行数据来举例，示例数据如下:
管理员配置用户、表、字段、脱敏条件，例如下面的配置。
当用户在Flink上查询orders表的数据时，会在底层结合该用户的脱敏条件重新生成SQL，即让数据脱敏生效。
当用户A和用户B在执行下面相同的SQL时，会看到不同的结果数据。
SELECT * FROM orders用户A查看到的结果数据如下，customer_name字段的数据被全部掩盖掉。
用户B查看到的结果数据如下，customer_name字段的数据只会显示前4位，剩下的用x代替。
在离线数仓工具Hive领域，由于发展多年已有Ranger来支持字段数据的脱敏控制，详见参考文献[1]。
下图是在Ranger里配置Hive表数据脱敏条件的页面，供参考。
但由于Flink实时数仓领域发展相对较短，Ranger还不支持FlinkSQL，以及依赖Ranger的话会导致系统部署和运维过重，因此开始自研实时数仓的数据脱敏解决工具。 当然本文中的核心思想也适用于Ranger中，可以基于此较快开发出ranger-flink插件。
可以参考作者文章[FlinkSQL字段血缘解决方案及源码]，本文根据Flink1.16修正和简化后的执行流程如下图所示。
在CalciteParser进行parse()和validate()处理后会得到一个SqlNode类型的抽象语法树(Abstract Syntax Tree，简称AST)，本文会针对此抽象语法树来组装行级过滤条件后生成新的AST，以实现数据脱敏控制。
下面章节要用到Calcite中的SqlNode、SqlCall、SqlIdentifier、SqlJoin、SqlBasicCall和SqlSelect等类，此处进行简单介绍以及展示它们间继承关系，以便读者阅读本文源码。
| 序号 | 类 | 介绍 |
|---|---|---|
| 1 | SqlNode | A SqlNode is a SQL parse tree. |
| 2 | SqlCall | A SqlCall is a call to an SqlOperator operator. |
| 3 | SqlIdentifier | A SqlIdentifier is an identifier, possibly compound. |
| 4 | SqlJoin | Parse tree node representing a JOIN clause. |
| 5 | SqlBasicCall | Implementation of SqlCall that keeps its operands in an array. |
| 6 | SqlSelect | A SqlSelect is a node of a parse tree which represents a select statement, the parent class is SqlCall |
针对输入的Flink SQL，在CalciteParser进行语法解析(parse)和语法校验(validate)后生成抽象语法树(Abstract Syntax Tree，简称AST)后，采用自定义
Calcite SqlBasicVisitor的方法遍历AST中的所有SqlSelect，获取到里面的每个输入表。如果输入表中字段有配置脱敏条件，则针对输入表生成子查询语句，
并把脱敏字段改写成CAST(脱敏函数(字段名) AS 字段类型) AS 字段名,再通过CalciteParser.parseExpression()把子查询转换成SqlSelect，
并用此SqlSelect替换原AST中的输入表来生成新的AST，最后得到新的SQL来继续执行。
通过对Flink SQL 语法的分析和研究，最终出现输入表的只包含以下两种情况:
- SELECT 语句的FROM子句，如果是子查询，则递归继续遍历。
- SELECT ... JOIN 语句的Left和Right子句，如果是多表JOIN，则递归查询遍历。
因此，下面的主要步骤会根据FROM子句的类型来寻找输入表。
主要通过Calcite提供的访问者模式自定义DataMaskVisitor来实现，遍历AST中所有的SqlSelect对象用子查询替换里面的输入表。 下面详细描述替换输入表的步骤，整体流程如下图所示。
- 遍历AST中的SELECT语句。
- 判断是否自定义的SELECT语句(由下面步骤9生成)，是则跳转到步骤10，否则继续步骤3。
- 判断SELECT语句中的FROM类型，按照不同类型对应执行下面的步骤4、5和10。
- 如果FROM是SqlJoin类型，则分别遍历其左Left和Right右节点，即执行当前步骤4和步骤6。由于可能是三张表及以上的Join，因此进行递归处理，即针对其左节点跳回到步骤3。
- 如果FROM是SqlBasicCall类型，还需要判断是否来自子查询，是则跳转到步骤10继续遍历AST，后续步骤1会对子查询中的SELECT语句进行处理。否则跳转到步骤7。
- 递归处理Join的右节点，即跳回到步骤3。
- 遍历表中的每个字段，如果某个字段有定义脱敏条件，则把改字段改写成格式
CAST(脱敏函数(字段名) AS 字段类型) AS 字段名，否则用原字段名。 - 针对步骤7处理后的字段，构建子查询语句，形如
(SELECT 字段名1, 字段名2, CAST(脱敏函数(字段名3) AS 字段类型) AS 字段名3、字段名4 FROM 表名) AS 表别名。 - 对步骤8的子查询调用
CalciteParser.parseExpression()进行解析，生成自定义的SELECT语句，并替换掉原FROM。 - 继续遍历AST，找到里面的SELECT语句进行处理，跳回到步骤1。
在Ranger中，默认的脱敏策略的如下所示。通过调研发现Ranger的大部分脱敏策略是通过调用Hive自带或自定义的系统函数实现的。
| 序号 | 策略名 | 策略说明 | Hive系统函数 |
|---|---|---|---|
| 1 | Redact | 用x屏蔽字母字符，用n屏蔽数字字符 | mask |
| 2 | Partial mask: show last 4 | 仅显示最后四个字符,其他用x代替 | mask_show_last_n |
| 3 | Partial mask: show first 4 | 仅显示前四个字符,其他用x代替 | mask_show_first_n |
| 4 | Hash | 用值的哈希值替换原值 | mask_hash |
| 5 | Nullify | 用NULL值替换原值 | Ranger自身实现 |
| 6 | Unmasked | 原样显示 | Ranger自身实现 |
| 7 | Date: show only year | 仅显示日期字符串的年份 | mask |
| 8 | Custom | Hive UDF来自定义策略 |
由于Flink支持Hive Catalog，在Flink能调用Hive系统函数。 因此，本方案也支持在Flink SQL配置Ranger的脱敏策略。
用例测试数据来自于CDC Connectors for Apache Flink
[4]官网，
本文给orders表增加一个region字段，再增加'connector'='print'类型的print_sink表，其字段和orders表的一样。数据库建表及初始化SQL位于data/database目录下。
测试用例中的catalog名称是hive，database名称是default。
下载本文源码后，可通过Maven运行单元测试。
$ cd flink-sql-security
$ mvn test详细测试用例可查看源码中的单测RewriteDataMaskTest和ExecuteDataMaskTest，下面只描述两个案例。
用户A执行下述SQL:
SELECT
order_id,
customer_name,
product_id,
region
FROM
orders- 输入SQL是一个简单SELECT语句，经过语法分析和语法校验后FROM类型是
SqlBasicCall，SQL中的表名orders会被替换为完整的hive.default.orders，别名是orders。 - 由于用户A针对字段
customer_name定义脱敏条件MASK(对应函数是脱敏函数是mask)，该字段在流程图中的步骤8中被改写为CAST(mask(customer_name) AS STRING) AS customer_name，其余字段未定义脱敏条件则保持不变。 - 然后在步骤8的操作中，表名
hive.default.orders被改写成如下子查询，子查询两侧用括号()进行包裹，并且用AS 别名来增加表别名。
(SELECT
order_id,
order_date,
CAST(mask(customer_name) AS STRING) AS customer_name,
product_id,
price,
order_status,
region
FROM
hive.default.orders
) AS orders最终执行的改写后SQL如下所示，这样用户A查询到的顾客姓名customer_name字段都是掩盖后的数据。
SELECT
orders.order_id,
orders.customer_name,
orders.product_id,
orders.region
FROM (
SELECT
order_id,
order_date,
CAST(mask(customer_name) AS STRING) AS customer_name,
product_id,
price,
order_status,
region
FROM
hive.default.orders
) AS orders用户A执行下述SQL:
INSERT INTO print_sink SELECT * FROM orders通过自定义Calcite DataMaskVisitor访问生成的AST，能找到对应的SELECT语句如下，注意在语法校验阶段 * 会被改写成表中所有字段。
SELECT
orders.order_id,
orders.order_date,
orders.customer_name,
orders.product_id,
orders.price,
orders.order_status,
orders.region
FROM
hive.default.orders AS orders针对此SELECT语句的改写逻辑同上，不再阐述。
最终执行的改写后SQL如下所示，注意插入到print_sink表的customer_name字段是掩盖后的数据。
INSERT INTO print_sink (
SELECT
orders.order_id,
orders.order_date,
orders.customer_name,
orders.product_id,
orders.price,
orders.order_status,
orders.region
FROM (
SELECT
order_id,
order_date,
CAST(mask(customer_name) AS STRING) AS customer_name,
product_id,
price,
order_status,
region
FROM
hive.default.orders
) AS orders
)