當前位置：首頁 > 运维知识 > 数据库 >内容正文

数据库

基于 mycat，我实现了一个数据库透明加密中间件

發布時間：2023/12/16 数据库 54 豆豆

生活随笔收集整理的這篇文章主要介紹了基于 mycat，我实现了一个数据库透明加密中间件小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

前言

數據是信息系統的核心資產，數據安全現在越來越受到企業的重視，雖然在數據庫外圍能夠很大程度上防止數據泄露，但核心數據的安全容不得半點差錯，數據庫加密因此應運而生，它作為數據安全的最后一道防線，確保即使數據庫被攻陷，存儲在數據庫中的數據仍然可以得到有效保護。數據庫加密主要用來解決以下兩個方面的問題：1.數據庫被拖庫后，避免因為明文存儲導致數據泄露；2.防范內部高權限用戶，竊取數據導致數據泄露。數據加密通常也被叫做數據脫敏，但大多數系統的數據脫敏僅限于應用層脫敏，數據庫層面存儲的依舊是明文，這始終都是一個安全隱患。

年前的時候，公司客戶提出了強制性要求，需要對存儲在數據庫的人員信息，比如身份證號，姓名，住址等敏感信息進行數據脫敏，之前其實一直有人在研究，但始終沒有找到好的辦法，所以一度處于擱置狀態，現在客戶催的緊，火燒眉毛，于是轉為由我負責推動這件事。

最后，經過多方對比，研究，討論，我選擇對 mycat 的源碼進行改造，終于實現了數據庫透明加解密的需求，可以進行列級別的加密存儲和解密，自研加密算法，可以支持加密字段的 like 查詢，也可以自定義加解密算法，對應用無感知，無需修改業務代碼，目前已在公司投入使用近一個月，一直處于穩定運行的狀態，下面詳細介紹一下。（源碼等資料可于文末獲取）

方案選擇

我查閱了一些資料，數據庫加密主要有以下幾種主流方案：

1.應用系統加密

此種加密方式是最容易想到的，即在編碼層面進行數據加密，對需要加密的字段在存儲或更新的時候加密，查詢的時候解密，但是改造成本實在過于高昂，而且對業務侵入性過強，日后如果有新的字段加密需求又是一波新的改造。

2.前置代理網關加密

前置代理網關本質上是數據庫代理中間件，通過對 sql 攔截解析，對需要加密的字段加密，對響應結果解密，返回給客戶端，對應用系統基本透明，只需要修改數據庫連接配置即可，但由于使用了中間層代理，對代理的穩定性要求較高，并且不可避免的會造成一定程度的性能損耗。

3.數據庫后置加密

某些數據庫廠商在數據庫引擎層增加了一些擴展接口和擴展機制，通過這些，數據庫系統用戶可以通過外部接口調用的方式對數據進行加解密處理，對應用系統透明，但支持的數據庫類型有限，不支持 mysql。

經過公司內部討論，決定采用第二種方案，當然，現在市面上有專門提供數據庫加密服務的廠商，只要花錢就可以搞定，但不到萬不得已，不會花錢的，懂的都懂，不再多說。

自己從頭寫一個數據庫代理并不現實，于是決定站在巨人的肩膀上實現，現在 Java 領域比較出名的數據庫中間件主要是 mycat 和 shardingSphere，雖然二者主打的功能是分庫分表，但同樣也是優秀的數據庫代理中間件，其實依本人所見，分庫分表能不用則不用，威力可能很大，但最后數據庫也給整殘廢了，維護起來成本過高，而且這種配置方式的分庫分表其實算偽分布式數據庫的實現，日后如果擴充節點，改變分片規則等會異常麻煩，還是多從業務角度考慮較好一些。

最后，通過拜讀二者源碼，發現 mycat 修改起來相對容易一些，shardingSphere 雖然有數據脫敏的功能，但經過測試，發現 like 查詢并不支持，復雜的子查詢 sql 解析不了，仍需完善，并且源碼體積異常龐大，修改難度過高，最后選擇修改 mycat 源碼，增加加解密模塊，實現數據庫加解密功能。

改造 mycat 實現加解密

最后經過各種測試，決定在 1.6.7.6-release 版本的基礎上進行改造，改造的部分主要包括：攔截解析 sql 加密，返回結果解密，此外，為了實現可配置的加解密功能，我添加了 encrypt.xml 配置文件，在 server.xml 中增加了加解密相關的屬性，在下一小節詳細介紹，在這里我們只關注核心部分的代碼，如果大家有興趣，可以在文末獲取源碼研究。

1.攔截解析 sql 加密

我們需要在 mycat 對 sql 路由之前改寫 sql，實現加密字段的加密，代碼位于 ServerConnection 的 routeEndExecuteSQL 方法中，我們在這里添加相關的加密邏輯即可，核心代碼如下：

public void routeEndExecuteSQL(String sql, final int type, final SchemaConfig schema) {// 路由計算RouteResultset rrs = null;try {// ==== sql 攔截解析加密 start =======boolean sqlPass = !sql.contains("information_schema") && (type == ServerParse.DELETE || type == ServerParse.INSERT || type == ServerParse.UPDATE || type == ServerParse.SELECT);//zrx 如果是增刪查改if (sqlPass) {// zrx 路由之前修改sql//zrx 獲取是否開啟加密boolean encrypt = YesOrNo.YES.getCode().equals(MycatServer.getInstance().getConfig().getSystem().getEncrypt());String schemaName = schema.getName();String[] nodes = schema.getAllDataNodes().toArray(new String[]{});//獲取當前的數據庫MycatConfig conf = MycatServer.getInstance().getConfig();Map<String, EncryptServer> encryptConfigMap = conf.getEncryptConfigMap();boolean passSchema = encryptConfigMap.containsKey(schemaName);boolean passDataSource = false;String dataSource = null;if (passSchema) {//如果是同一個 schema 下的，數據庫結構都是一樣的for (String node : nodes) {dataSource = conf.getDataNodes().get(node).getDatabase();if (encryptConfigMap.get(schemaName).getEncryptDataSourceMap().containsKey(dataSource)) {passDataSource = true;}}}boolean shouldEncrypt = encrypt && passSchema && passDataSource;//zrx 處理加密if (shouldEncrypt) {EncryptDataSource encryptDataSource = encryptConfigMap.get(schemaName).getEncryptDataSourceMap().get(dataSource);//獲取需要被加密的表和字段Map<String, Set<String>> encryptTableColMap = encryptDataSource.getEncryptTableColMap();//根據sql類型解析sqltry {//解析sqlStatement statement = CCJSqlParserUtil.parse(sql);if (ServerParse.UPDATE == type) {Update updateStatement = (Update) statement;Table table = updateStatement.getTable();String updateTable = table.getName().toLowerCase().replaceAll("`", "");//別名Alias alias = table.getAlias();if (encryptTableColMap.containsKey(updateTable)) {//獲取需要加密的列Set<String> columns = encryptTableColMap.get(updateTable);//獲取sql中的列List<Column> sqlColumns = updateStatement.getColumns();List<Expression> expressions = updateStatement.getExpressions();if (sqlColumns != null && expressions != null) {//遍歷更新的列，查看是否需要加密replaceSqlValue(columns, sqlColumns, expressions);//替換sqlsql = updateStatement.toString();}//獲取 where 條件Expression where = updateStatement.getWhere();if (where != null) {encryptParser(encryptTableColMap, where, columns, alias == null ? null : alias.getName().toLowerCase().replaceAll("`", ""), updateTable);sql = updateStatement.toString();}}} else if (ServerParse.INSERT == type) {Insert insertStatement = (Insert) statement;String inerstTable = insertStatement.getTable().getName().toLowerCase().replaceAll("`", "");if (encryptTableColMap.containsKey(inerstTable)) {//獲取需要加密的列Set<String> columns = encryptTableColMap.get(inerstTable);//如果insert語句中包含需要加密的表，獲取插入的列和值List<Column> sqlColumns = insertStatement.getColumns();ExpressionList itemsList = (ExpressionList) insertStatement.getItemsList();if (sqlColumns != null && itemsList != null) {//遍歷插入的列，查看是否有需要加密的List<Expression> sqlInsertValues = itemsList.getExpressions();replaceSqlValue(columns, sqlColumns, sqlInsertValues);//替換sqlsql = insertStatement.toString();}}} else if (ServerParse.DELETE == type) {Delete deleteStatement = (Delete) statement;Table table = deleteStatement.getTable();String deleteTable = table.getName().toLowerCase().replaceAll("`", "");//別名Alias alias = table.getAlias();if (encryptTableColMap.containsKey(deleteTable)) {//獲取需要加密的列Set<String> columns = encryptTableColMap.get(deleteTable);Expression where = deleteStatement.getWhere();if (where != null) {encryptParser(encryptTableColMap, where, columns, alias == null ? null : alias.getName().toLowerCase().replaceAll("`", ""), deleteTable);sql = deleteStatement.toString();}}} else {Select selectStatement = (Select) statement;SelectBody body = selectStatement.getSelectBody();encryptParserSelect(body, encryptTableColMap);sql = selectStatement.toString();}} catch (Exception e) {LOGGER.error("encrypt sql parser error:", e);if (!sql.contains("convert(no,SIGNED)")) {writeErrMessage(ErrorCode.ERR_HANDLE_DATA, "encrypt sql parser error:" + e.toString());return;}}}}// ==== sql 攔截解析加密 end=======//生成 rrs 路由對象rrs = MycatServer.getInstance().getRouterservice().route(MycatServer.getInstance().getConfig().getSystem(),schema, type, sql, this.charset, this);} catch (Exception e) {StringBuilder s = new StringBuilder();LOGGER.warn(s.append(this).append(sql).toString() + " err:" + e.toString(), e);String msg = e.getMessage();writeErrMessage(ErrorCode.ER_PARSE_ERROR, msg == null ? e.getClass().getSimpleName() : msg);return;}if (rrs != null) {// #支持mariadb驅動useBatchMultiSend=true,連續接收到的sql先放入隊列，等待前面處理完成后再繼續處理。// 參考https://mariadb.com/kb/en/option-batchmultisend-description/boolean executeNow = false;synchronized (this.executeSqlQueue) {executeNow = this.executeSqlQueue.isEmpty();this.executeSqlQueue.add(new SqlEntry(sql, type, rrs));if (LOGGER.isDebugEnabled()) {LOGGER.debug("add queue,executeSqlQueue size {}", executeSqlQueue.size());}}if (executeNow) {this.executeSqlId++;session.execute(rrs, rrs.isSelectForUpdate() ? ServerParse.UPDATE : type);}}}

2.返回結果解密

改寫返回結果的代碼位于 SingleNodeHandler 和 MultiNodeQueryHandler 的 rowResponse 方法中，在這里我們對返回結果攔截，獲取需要解密的數據進行解密，按照 mysql 協議打包重新發送給客戶端，核心代碼如下：

public void rowResponse(byte[] row, BackendConnection conn) {....//zrx處理加密數據row = ResponseEncryptHandler.getBytes(rrs, session, fieldCount, row);....}public static byte[] getBytes(RouteResultset rrs, NonBlockingSession session, int fieldCount, byte[] row) {boolean encrypt = YesOrNo.YES.getCode().equals(MycatServer.getInstance().getConfig().getSystem().getEncrypt());Map<String, EncryptServer> encryptConfigMap = MycatServer.getInstance().getConfig().getEncryptConfigMap();//zrx 處理返回結果if (encrypt && rrs.getSqlType() == ServerParse.SELECT && encryptConfigMap.containsKey(session.getSource().getSchema())) {//讀取RowDataPacket resultsetRow = new RowDataPacket(fieldCount, 10);//解密讀取resultsetRow.readDecrypt(row);if (!resultsetRow.decryptIndexs.isEmpty()) {//如果有需要加密的列List<byte[]> fieldValues = resultsetRow.fieldValues;for (Integer index : resultsetRow.decryptIndexs) {//讀取加密后的字符串String encryptHex = new String(fieldValues.get(index), StandardCharsets.UTF_8);//解密并設置值fieldValues.set(index, EncryptHelper.decode(EncryptHelper.hexStringToBytes(encryptHex)));}ByteBuffer buffer = ByteBuffer.allocate(row.length);resultsetRow.write(buffer);buffer.flip();byte[] bytes = new byte[buffer.remaining()];buffer.get(bytes, 0, bytes.length);//改變結果集row = bytes;}}return row;}

以上便是核心部分的代碼，由于改造的代碼很多，沒有辦法在這里一一介紹，最后改造完畢的代碼主要包含如下功能：

encrypt.xml 配置文件，配置加密相關信息；

支持自動對需要加密的字段加密；

內置加密算法，支持對加密字段的 like 查詢；

支持自定義加密算法；

支持復雜的 sql 查詢，自動對 where 條件中的加密字段加密；

增，改數據時自動對需要加密的字段加密；

支持 mysql8

使用方法

首先下載安裝包（文末獲取），解壓，進入 conf 文件夾進行 server.xml，schema.xml，encrypt.xml 的配置。

改造后的程序跟 mycat 的使用方法完全一致，在這里只介紹加解密相關的使用，加解密的使用方法也很簡單，首先配置 server.xml 的加密相關的屬性，如下：

<property name="encrypt">1</property>   <property name="useInternalEncryptAlgorithm">1</property>  <property name="secretKey">12345678ABCDEFGH</property> <user name="root"><property name="password">123456</property><property name="schemas">test</property></user>

然后配置 schema.xml，這里的配置跟 mycat 完全一致，如：

<schema name="test" checkSQLschema="true" sqlMaxLimit="100" dataNode="test"></schema><dataNode name="test" dataHost="test" database="test" /><dataHost name="test" maxCon="1000" minCon="10" balance="0"writeType="0" dbType="mysql" dbDriver="native" switchType="1" slaveThreshold="100"><heartbeat>select user()</heartbeat><writeHost host="hostM1" url="localhost:3307" user="root"password="root"></writeHost></dataHost>

最后配置 encrypt.xml，我們配置需要加密的字段為 card_code 和 name，表都為 real_people_copy1，設auto 為 1，啟動自動加密：

<server schema="test"><datasource name="test" ip="localhost" port="3307" username="root" password="root" auto="1" dbType="mysql"><column name="card_code"><table name="real_people_copy1" pk="people_id"/></column><column name="name"><table name="real_people_copy1" pk="people_id"/></column></datasource></server>

配置完畢后，進入 bin 目錄，啟動：

./mycat start # 出現以下提示代表啟動成功 MyCAT Server startup successfully. see logs in logs/mycat.log

使用 navicat 或其他工具連接 mycat 代理服務，端口默認是 8066，連接方式選擇 mysql，如果代理的不是 mysql 數據庫，使用控制臺方式連接：mysql -u root -P8066 -p 123456，因為 mycat 模擬的是 mysql ，如果使用客戶端工具連接發送的 sql 請求是基于 mysql 協議發送的，mycat 轉發到數據庫中執行會出錯。

這里以 navicat 為例，配置好連接信息，測試連接，連接成功：

然后我們查看真正的數據庫中 real_people_copy1 中的數據，查看 card_code 和 name 字段是否已被加密：

如上所示，card_code 和 name 字段已被成功加密，然后我們通過 mycat 查看該表，發現展示的是解密后的數據（敏感數據部分打碼）：

然后我們在 mycat 執行 sql 測試，發現成功查詢出了結果：

程序中打斷點，發現對查詢的 card_code 的值都成功做了替換：

增，刪，改同樣沒有問題，在這里就不一一測試了，至此，數據庫透明加解密已經成功實現。

監控管理

mycat 的管理端口為 9066，主要用于 io流量，數據庫連接，內存，tps，sql 統計等的監控管理，連接方式跟 mysql 一致：mysql -u 用戶名 -P 9066 -p 密碼，常用指令如 show @@sql，show @@sql.slow，show @@sql.high 等，我在 mycat-web 的基礎上改造了一版，精簡了功能，修復了已經存在的 bug，兼容 mysql8，效果如下（源碼文末獲取）：

結語

在 mycat 添加透明加解密的功能其實并不容易，首先要徹底讀懂源碼，然后才能在此基礎上進行修改，其次還要對 mysql 的相關協議深入了解，才能對返回結果進行封裝，sql 解析同樣是個困難的問題，主要是查詢語句的解析，說實話是非常復雜的，雖然過程比較坎坷，但中途也學到了不少東西，總歸是有所收獲。

使用數據庫代理中間件必然會對性能造成一定的損失，經測試，通常情況下，在中間件中額外花費的時間大約在 10-30ms 之間，在可以接受的范圍之內。

相關源碼資料可以通過關注公眾號螺旋編程極客獲取，readme 中有更加詳細的使用方法，包括自定義加密算法，手動解密數據等，需要的朋友可以自行查閱。

本文到這里就結束了，我們下次再見！

關注公眾號螺旋編程極客可進群一起探討，大家一起學習，共同進步，同時有海量學習資源領取。
關注公眾號螺旋編程極客發送透明加密獲取相關源碼資料！

總結

以上是生活随笔為你收集整理的基于 mycat，我实现了一个数据库透明加密中间件的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：家庭局域网_第二篇如何让电信IPTV与路
下一篇：通过谷歌Google轻松拥有自己的站内搜