使用 Apache ShardingSphere 进行模糊查询的指南

Apache ShardingSphere 是一个开源分布式数据库，也是用户和开发者为其数据库提供定制和云原生体验所需的生态系统。 其 最新版本 包含许多新功能，包括与现有 SQL 工作流集成的数据加密。 最重要的是，它允许对加密数据进行模糊查询。

问题

通过解析用户的 SQL 输入并根据用户的加密规则重写 SQL，原始数据被加密并与密文数据同时存储在底层数据库中。

当用户查询数据时，它从数据库中获取密文数据，对其进行解密，并将解密的原始数据返回给用户。但是，由于加密算法加密的是整个字符串，因此用户无法运行模糊查询。

然而，许多企业需要在数据加密后进行模糊查询。在 5.3.0 版本中，Apache ShardingSphere 为用户提供了一种默认的模糊查询算法，该算法支持加密字段。该算法还支持热插拔，用户可以对其进行自定义。可以通过配置实现模糊查询。

如何在加密场景中实现模糊查询

将数据加载到内存数据库 (IMDB)

首先，将所有数据加载到 IMDB 中进行解密。然后，它就像查询原始数据一样。这种方法可以实现模糊查询。如果数据量很小，这种方法将被证明是简单且具有成本效益的。但是，如果数据量很大，那将是一场灾难。

实现与数据库程序一致的加密和解密功能

第二种方法是修改模糊查询条件，并使用数据库解密功能首先解密数据，然后实现模糊查询。此方法的优点是实现、开发和使用成本低。

用户只需要稍微修改之前的模糊查询条件即可。但是，密文和加密功能一起存储在数据库中，无法解决帐户数据泄漏的问题。

原生 SQL

select * from user where name like "%xxx%"

实现解密功能后

ѕеlесt * frоm uѕеr whеrе dесоdе(namе) lіkе "%ххх%"

数据掩码后存储

对密文执行数据掩码，然后将其存储在模糊查询列中。这种方法可能缺乏精确性。

例如，手机号码 13012345678 在执行掩码算法后变为 130****5678。

在分词和组合后执行加密存储

此方法对密文数据执行分词和组合，然后通过对固定长度的字符进行分组并将一个字段拆分为多个字段来加密结果集。例如，我们使用四个英文字符和两个中文字符作为查询条件：ningyu1 使用四个字符作为一个组进行加密，因此第一组是 ning，第二组是 ingy，第三组是 ngyu，第四组是 gyu1，依此类推。所有字符都被加密并存储在模糊查询列中。如果要检索包含四个字符的所有数据，例如 ingy，请加密字符并使用键 like"%partial%" 进行查询。

缺点

1. 增加的存储成本：自由分组将增加数据量，并且加密后数据长度会增加。
2. 模糊查询中的有限长度：由于安全问题，自由分组的长度不能太短，否则 彩虹表 很容易破解它。 就像我上面提到的示例一样，模糊查询字符的长度必须大于或等于四个字母/数字或两个中文字符。

单字符摘要算法（ShardingSphere 5.3.0 版本中提供的默认模糊查询算法）

尽管上述方法都是可行的，但人们自然会想知道是否有更好的替代方案。在我们的社区中，我们发现单字符加密和存储可以平衡性能和查询，但未能满足安全要求。

那么理想的解决方案是什么？受到掩码算法和密码散列函数的启发，我们发现可以使用数据丢失和单向函数。

密码散列函数应具有以下四个特征

1. 应该很容易计算任何给定消息的哈希值。
2. 从已知的哈希值推断原始消息应该很困难。
3. 修改消息而不更改哈希值应该是不可行的。
4. 两个不同的消息产生相同哈希值的可能性应该非常低。

安全性：由于单向函数，不可能推断出原始消息。为了提高模糊查询的准确性，我们希望加密单个字符，但彩虹表会破解它。

因此，我们采用单向函数（以确保每个字符在加密后都相同），并增加冲突的频率（以确保每个字符串都是 1: N 反向），这大大提高了安全性。

模糊查询算法

Apache ShardingSphere 使用以下单字符摘要算法 org.apache.shardingsphere.encrypt.algorithm.like.CharDigestLikeEncryptAlgorithm 实现通用模糊查询算法。

public final class CharDigestLikeEncryptAlgorithm implements LikeEncryptAlgorithm<Object, String> {
  
    private static final String DELTA = "delta";
  
    private static final String MASK = "mask";
  
    private static final String START = "start";
  
    private static final String DICT = "dict";
  
    private static final int DEFAULT_DELTA = 1;
  
    private static final int DEFAULT_MASK = 0b1111_0111_1101;
  
    private static final int DEFAULT_START = 0x4e00;
  
    private static final int MAX_NUMERIC_LETTER_CHAR = 255;
  
    @Getter
    private Properties props;
  
    private int delta;
  
    private int mask;
  
    private int start;
  
    private Map<Character, Integer> charIndexes;
  
    @Override
    public void init(final Properties props) {
        this.props = props;
        delta = createDelta(props);
        mask = createMask(props);
        start = createStart(props);
        charIndexes = createCharIndexes(props);
    }
  
    private int createDelta(final Properties props) {
        if (props.containsKey(DELTA)) {
            String delta = props.getProperty(DELTA);
            try {
                return Integer.parseInt(delta);
            } catch (NumberFormatException ex) {
                throw new EncryptAlgorithmInitializationException("CHAR_DIGEST_LIKE", "delta can only be a decimal number");
            }
        }
        return DEFAULT_DELTA;
    }
  
    private int createMask(final Properties props) {
        if (props.containsKey(MASK)) {
            String mask = props.getProperty(MASK);
            try {
                return Integer.parseInt(mask);
            } catch (NumberFormatException ex) {
                throw new EncryptAlgorithmInitializationException("CHAR_DIGEST_LIKE", "mask can only be a decimal number");
            }
        }
        return DEFAULT_MASK;
    }
  
    private int createStart(final Properties props) {
        if (props.containsKey(START)) {
            String start = props.getProperty(START);
            try {
                return Integer.parseInt(start);
            } catch (NumberFormatException ex) {
                throw new EncryptAlgorithmInitializationException("CHAR_DIGEST_LIKE", "start can only be a decimal number");
            }
        }
        return DEFAULT_START;
    }
  
    private Map<Character, Integer> createCharIndexes(final Properties props) {
        String dictContent = props.containsKey(DICT) && !Strings.isNullOrEmpty(props.getProperty(DICT)) ? props.getProperty(DICT) : initDefaultDict();
        Map<Character, Integer> result = new HashMap<>(dictContent.length(), 1);
        for (int index = 0; index < dictContent.length(); index++) {
            result.put(dictContent.charAt(index), index);
        }
        return result;
    }
  
    @SneakyThrows
    private String initDefaultDict() {
        InputStream inputStream = CharDigestLikeEncryptAlgorithm.class.getClassLoader().getResourceAsStream("algorithm/like/common_chinese_character.dict");
        LineProcessor<String> lineProcessor = new LineProcessor<String>() {
  
            private final StringBuilder builder = new StringBuilder();
  
            @Override
            public boolean processLine(final String line) {
                if (line.startsWith("#") || 0 == line.length()) {
                    return true;
                } else {
                    builder.append(line);
                    return false;
                }
            }
  
            @Override
            public String getResult() {
                return builder.toString();
            }
        };
        return CharStreams.readLines(new InputStreamReader(inputStream, Charsets.UTF_8), lineProcessor);
    }
  
    @Override
    public String encrypt(final Object plainValue, final EncryptContext encryptContext) {
        return null == plainValue ? null : digest(String.valueOf(plainValue));
    }
  
    private String digest(final String plainValue) {
        StringBuilder result = new StringBuilder(plainValue.length());
        for (char each : plainValue.toCharArray()) {
            char maskedChar = getMaskedChar(each);
            if ('%' == maskedChar) {
                result.append(each);
            } else {
                result.append(maskedChar);
            }
        }
        return result.toString();
    }
  
    private char getMaskedChar(final char originalChar) {
        if ('%' == originalChar) {
            return originalChar;
        }
        if (originalChar <= MAX_NUMERIC_LETTER_CHAR) {
            return (char) ((originalChar + delta) & mask);
        }
        if (charIndexes.containsKey(originalChar)) {
            return (char) (((charIndexes.get(originalChar) + delta) & mask) + start);
        }
        return (char) (((originalChar + delta) & mask) + start);
    }
  
    @Override
    public String getType() {
        return "CHAR_DIGEST_LIKE";
    }
}

• 定义二进制 mask 代码以丢失精度 0b1111_0111_1101 (mask)。
• 以类似 map 字典的方式保存具有被打乱顺序的常用汉字。
• 获取数字、英语和拉丁文的单个字符串的 Unicode。
• 获取属于字典的汉字的 index。
• 其他字符获取单个字符串的 Unicode。
• 将 1 (delta) 添加到上述不同类型获得的数字，以防止任何原始文本出现在数据库中。
• 然后将偏移量 Unicode 转换为二进制，使用 mask 执行 AND 运算，并执行两位数字丢失。
• 在丢失精度后直接输出数字、英语和拉丁文。
• 在丢失精度后，剩余字符转换为十进制并使用公共字符 start 代码输出。

模糊算法的开发进展

第一版

简单地使用 Unicode 和常用字符的 mask 代码来执行 AND 运算。

Mask: 0b11111111111001111101
The original character: 0b1000101110101111讯
After encryption: 0b1000101000101101設

假设我们知道密钥和加密算法，则反向传递后的原始字符串为

1.0b1000101100101101 謭
2.0b1000101100101111 謯
3.0b1000101110101101 训
4.0b1000101110101111 讯
5.0b1000101010101101 読
6.0b1000101010101111 誯
7.0b1000101000101111 訯
8.0b1000101000101101 設

根据缺失的位，我们发现每个字符串都可以向后推导出 2^n 个汉字。当常用汉字的 Unicode 为十进制时，它们的间隔非常大。请注意，向后推断的汉字不是常用字符，并且更容易推断出原始字符。

图片来源：

（熊高翔，CC BY-SA 4.0）

第二版

Unicode 中常用汉字的间隔是不规则的。我们计划保留 Unicode 中汉字的最后几位，并将它们转换为十进制作为 index 来获取一些常用汉字。这样，当算法已知时，反向传递后不会出现非常用字符，并且干扰项不再容易消除。

如果我们保留 Unicode 中汉字的最后几位，它与模糊查询的准确性和反解密复杂性有关。 准确性越高，解密难度越低。

让我们看一下在我们的算法下常用汉字的冲突程度

1. 当 mask=0b0011_1111_1111 时

图片来源：

（熊高翔，CC BY-SA 4.0）

2. 当 mask=0b0001_1111_1111 时

图片来源：

（熊高翔，CC BY-SA 4.0）

对于汉字的尾数，保留 10 位和 9 位。10 位查询更准确，因为它的冲突要弱得多。然而，如果算法和密钥已知，则可以向后推导出 1:1 字符的原始文本。

九位查询的准确性较低，因为九位冲突相对较强，但 1:1 字符较少。 尽管我们改变了冲突，无论我们保留十位还是九位，但由于汉字的 Unicode 不规则，分布是不平衡的。 无法控制总体冲突概率。

第三版

为了解决第二版中发现的分布不均匀问题，我们采用具有被打乱顺序的常用字符作为字典表。

1. 加密文本首先在无序字典表中查找 index。 我们使用 index 和下标来替换没有规则的 Unicode。 对于非常用字符，使用 Unicode。（注意：尽可能均匀地分配要计算的代码。）

2. 下一步是使用 mask 执行 AND 运算并丢失两位精度以增加冲突频率。

让我们看一下在我们的算法下常用汉字的冲突程度

1. 当 mask=0b1111_1011_1101 时

图片来源：

（熊高翔，CC BY-SA 4.0）

2. 当 mask=0b0111_1011_1101 时

图片来源：

（熊高翔，CC BY-SA 4.0）

当 mask 保留 11 位时，可以看到碰撞分布集中在 1:4。当 mask 保留 10 位时，比例变为 1:8。此时，我们只需要调整精度损失的数量，就可以控制碰撞是 1:2、1:4 还是 1:8。

如果 mask 选择为 1，并且算法和密钥已知，则会出现 1:1 的汉字，因为此时我们计算的是常用字符的碰撞程度。如果我们在汉字的 16 位二进制之前添加缺失的四位，情况就变成了 2^5=32 种情况。

由于我们加密的是整个文本，即使单个字符被反向推断出来，对整体安全性影响也很小，不会造成大规模的数据泄露。同时，反向推导的前提是要知道算法、密钥、delta 和字典，所以从数据库中的数据反推是不可能实现的。

如何使用模糊查询

模糊查询需要在加密配置中配置 encryptors (加密算法配置), likeQueryColumn (模糊查询列名) 和 likeQueryEncryptorName (模糊查询列的加密算法名称)。

请参考以下配置。添加您自己的分片算法和数据源。

dataSources:
  ds_0:
    dataSourceClassName: com.zaxxer.hikari.HikariDataSource
    driverClassName: com.mysql.jdbc.Driver
    jdbcUrl: jdbc:mysql://127.0.0.1:3306/test?allowPublicKeyRetrieval=true
    username: root
    password: root
    
rules:
- !ENCRYPT
  encryptors:
    like_encryptor:
      type: CHAR_DIGEST_LIKE
    aes_encryptor:
      type: AES
      props:
        aes-key-value: 123456abc
  tables:
    user:
      columns:
        name:
          cipherColumn: name
          encryptorName: aes_encryptor
          assistedQueryColumn: name_ext
          assistedQueryEncryptorName: aes_encryptor
          likeQueryColumn: name_like
          likeQueryEncryptorName: like_encryptor
        phone:
          cipherColumn: phone
          encryptorName: aes_encryptor
          likeQueryColumn: phone_like
          likeQueryEncryptorName: like_encryptor
  queryWithCipherColumn: true


props:
  sql-show: true

插入

Logic SQL: insert into user ( id, name, phone, sex) values ( 1, '熊高祥', '13012345678', '男')
Actual SQL: ds_0 ::: insert into user ( id, name, name_ext, name_like, phone, phone_like, sex) values (1, 'gyVPLyhIzDIZaWDwTl3n4g==', 'gyVPLyhIzDIZaWDwTl3n4g==', '佹堝偀', 'qEmE7xRzW0d7EotlOAt6ww==', '04101454589', '男')

更新

Logic SQL: update user set name = '熊高祥123', sex = '男1' where sex ='男' and phone like '130%'
Actual SQL: ds_0 ::: update user set name = 'K22HjufsPPy4rrf4PD046A==', name_ext = 'K22HjufsPPy4rrf4PD046A==', name_like = '佹堝偀014', sex = '男1' where sex ='男' and phone_like like '041%'

选择

Logic SQL: select * from user where (id = 1 or phone = '13012345678') and name like '熊%'
Actual SQL: ds_0 ::: select `user`.`id`, `user`.`name` AS `name`, `user`.`sex`, `user`.`phone` AS `phone`, `user`.`create_time` from user where (id = 1 or phone = 'qEmE7xRzW0d7EotlOAt6ww==') and name_like like '佹%'

选择: 联邦表子查询

Logic SQL: select * from user LEFT JOIN user_ext on user.id=user_ext.id where user.id in (select id from user where sex = '男' and name like '熊%')
Actual SQL: ds_0 ::: select `user`.`id`, `user`.`name` AS `name`, `user`.`sex`, `user`.`phone` AS `phone`, `user`.`create_time`, `user_ext`.`id`, `user_ext`.`address` from user LEFT JOIN user_ext on user.id=user_ext.id where user.id in (select id from user where sex = '男' and name_like like '佹%')

删除

Logic SQL: delete from user where sex = '男' and name like '熊%'
Actual SQL: ds_0 ::: delete from user where sex = '男' and name_like like '佹%'

上面的例子演示了模糊查询列如何在不同的 SQL 语法中重写 SQL 以支持模糊查询。

总结

本文向您介绍了模糊查询的工作原理，并使用具体示例演示了如何使用它。我希望通过本文，您将对模糊查询有一个基本的了解。

---

本文最初发表在 Medium 上，并已获得作者的许可转载。