From a909e79c509893f4846760c24b9da6f0fc2e4924 Mon Sep 17 00:00:00 2001 From: heavyrain2012 Date: Mon, 12 Jul 2021 11:20:26 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E9=85=8D=E7=BD=AE=E6=98=AF?= =?UTF-8?q?=E5=90=A6=E6=98=8E=E6=96=87=E5=AD=98=E5=82=A8=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E5=88=B0=E6=95=B0=E6=8D=AE=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- broker/config/wildfirechat.conf | 4 ++ .../java/io/moquette/BrokerConstants.java | 2 + .../moquette/persistence/DatabaseStore.java | 56 +++++++++++++++++-- .../persistence/MemoryMessagesStore.java | 6 ++ .../src/main/java/win/liyufan/im/DBUtil.java | 2 +- .../src/main/resources/wildfirechat.conf | 4 ++ 6 files changed, 67 insertions(+), 7 deletions(-) diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index 23fa122b..39fc4efa 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -74,6 +74,10 @@ message.blacklist.strategy 0 ##是否禁止服务器端消息搜索，该功能暂未实现。目前的影响是如果打开，则存储消息时不单独保存_searchable_content字段 message.disable_remote_search false +##是否数据库中加密消息内容。注意这个开关在服务运行起来后不能改变了，避免读取数据库时无法恢复历史消息。 +##如果要避免数据库中明文存储消息内容，请打开这个开关和message.disable_remote_search开关 +message.encrypt_message_content false + ##允许客户端撤回用户自己发送的消息时限，单位是秒。群组管理员或群主撤回群成员消息、server api不受此限制，可以撤回任意时限内的消息 message.recall_time_limit 120 diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index d02a7390..4a762187 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -167,6 +167,8 @@ public final class BrokerConstants { public static final String MESSAGES_GLOBAL_MUTE_EXCEPTION_TYPES = "message.global_mute_exception_types"; public static final String MESSAGES_DISABLE_REMOTE_SEARCH = "message.disable_remote_search"; + public static final String MESSAGES_ENCRYPT_MESSAGE_CONTENT = "message.encrypt_message_content"; + public static final String MESSAGES_RECALL_TIME_LIMIT = "message.recall_time_limit"; public static final String MESSAGES_DISABLE_GROUP_MANAGER_RECALL = "message.disable_group_manager_recall"; diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 79e6708c..614794dc 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -8,11 +8,11 @@ package io.moquette.persistence; -import cn.wildfirechat.common.IMExceptionEvent; import cn.wildfirechat.pojos.SystemSettingPojo; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import cn.wildfirechat.server.ThreadPoolExecutorWrapper; +import com.google.protobuf.InvalidProtocolBufferException; import com.hazelcast.core.HazelcastInstance; import com.hazelcast.core.MultiMap; import com.hazelcast.util.StringUtil; @@ -26,7 +26,9 @@ import win.liyufan.im.MessageBundle; import win.liyufan.im.MessageShardingUtil; import win.liyufan.im.Utility; +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; import java.security.MessageDigest; import java.sql.*; import java.util.*; @@ -42,10 +44,15 @@ public class DatabaseStore { private static final Logger LOG = LoggerFactory.getLogger(DatabaseStore.class); private final ThreadPoolExecutorWrapper mScheduler; private boolean disableRemoteMessageSearch = false; + private boolean encryptMessage = false; public void setDisableRemoteMessageSearch(boolean disableRemoteMessageSearch) { this.disableRemoteMessageSearch = disableRemoteMessageSearch; } + public void setEncryptMessage(boolean encryptMessage) { + this.encryptMessage = encryptMessage; + } + public DatabaseStore(ThreadPoolExecutorWrapper scheduler) { this.mScheduler = scheduler; } @@ -537,7 +544,7 @@ public class DatabaseStore { statement.setString(index++, message.getConversation().getTarget()); statement.setInt(index++, message.getConversation().getLine()); Blob blob = connection.createBlob(); - blob.setBytes(1, message.getContent().toByteArray()); + blob.setBytes(1, encryptMessageContent(message.getContent().toByteArray(), false)); statement.setBlob(index++, blob); if (!disableRemoteMessageSearch) { statement.setString(index++, searchableContent); @@ -571,6 +578,15 @@ public class DatabaseStore { }); } + byte[] encryptMessageContent(byte[] in, boolean force) { + if(in != null && (encryptMessage || force)) { + for (int i = 0; i < in.length; i++) { + in[i] ^= 0xBD; + } + } + return in; + } + void persistSensitiveMessage(final WFCMessage.Message message) { if(message.getContent().getPersistFlag() == Transparent) { return; @@ -599,7 +615,7 @@ public class DatabaseStore { statement.setString(index++, message.getConversation().getTarget()); statement.setInt(index++, message.getConversation().getLine()); Blob blob = connection.createBlob(); - blob.setBytes(1, message.getContent().toByteArray()); + blob.setBytes(1, encryptMessageContent(message.getContent().toByteArray(), false)); statement.setBlob(index++, blob); statement.setString(index++, searchableContent); statement.setTimestamp(index++, new Timestamp(message.getServerTimestamp())); @@ -668,7 +684,16 @@ public class DatabaseStore { builder.setConversation(cb.build()); Blob blob = resultSet.getBlob(index++); - WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.parseFrom(blob.getBinaryStream()); + WFCMessage.MessageContent messageContent; + try { + messageContent = WFCMessage.MessageContent.parseFrom(encryptMessageContent(toByteArray(blob.getBinaryStream()), false)); + } catch (InvalidProtocolBufferException e) { + if(encryptMessage) { + messageContent = WFCMessage.MessageContent.parseFrom(blob.getBinaryStream()); + } else { + messageContent = WFCMessage.MessageContent.parseFrom(encryptMessageContent(toByteArray(blob.getBinaryStream()), true)); + } + } builder.setContent(messageContent); builder.setServerTimestamp(resultSet.getTimestamp(index++).getTime()); WFCMessage.Message message = builder.build(); @@ -702,6 +727,16 @@ public class DatabaseStore { return out; } + public byte[] toByteArray(InputStream input) throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + byte[] buffer = new byte[4096]; + int n = 0; + while (-1 != (n = input.read(buffer))) { + output.write(buffer, 0, n); + } + return output.toByteArray(); + } + MessageBundle getMessage(long messageId) { String sql = "select `_from`, `_type`, `_target`, `_line`, `_data`, `_dt` from " + MessageShardingUtil.getMessageTable(messageId) +" where _mid = ? order by `_dt` DESC limit 1"; Connection connection = null; @@ -724,7 +759,16 @@ public class DatabaseStore { builder.setConversation(cb.build()); Blob blob = resultSet.getBlob(index++); - WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.parseFrom(blob.getBinaryStream()); + WFCMessage.MessageContent messageContent; + try { + messageContent = WFCMessage.MessageContent.parseFrom(encryptMessageContent(toByteArray(blob.getBinaryStream()), false)); + } catch (InvalidProtocolBufferException e) { + if(encryptMessage) { + messageContent = WFCMessage.MessageContent.parseFrom(blob.getBinaryStream()); + } else { + messageContent = WFCMessage.MessageContent.parseFrom(encryptMessageContent(toByteArray(blob.getBinaryStream()), true)); + } + } builder.setContent(messageContent); builder.setServerTimestamp(resultSet.getTimestamp(index++).getTime()); WFCMessage.Message message = builder.build(); @@ -829,7 +873,7 @@ public class DatabaseStore { builder.setConversation(cb.build()); Blob blob = resultSet.getBlob(index++); - WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.parseFrom(blob.getBinaryStream()); + WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.parseFrom(encryptMessageContent(toByteArray(blob.getBinaryStream()), false)); builder.setContent(messageContent); builder.setServerTimestamp(resultSet.getTimestamp(index++).getTime()); diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 86609e53..776c161e 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -268,6 +268,12 @@ public class MemoryMessagesStore implements IMessagesStore { } catch (Exception e) { } + try { + boolean encryptMessage = Boolean.parseBoolean(m_Server.getConfig().getProperty(BrokerConstants.MESSAGES_ENCRYPT_MESSAGE_CONTENT, "false")); + databaseStore.setEncryptMessage(encryptMessage); + } catch (Exception e) { + } + try { mRecallTimeLimit = Long.parseLong(m_Server.getConfig().getProperty(BrokerConstants.MESSAGES_RECALL_TIME_LIMIT)); } catch (Exception e) { diff --git a/broker/src/main/java/win/liyufan/im/DBUtil.java b/broker/src/main/java/win/liyufan/im/DBUtil.java index 81b2618b..8d6a92aa 100755 --- a/broker/src/main/java/win/liyufan/im/DBUtil.java +++ b/broker/src/main/java/win/liyufan/im/DBUtil.java @@ -108,7 +108,7 @@ public class DBUtil { Flyway flyway = Flyway.configure().dataSource(comboPooledDataSource).locations(migrateLocation).baselineOnMigrate(true).load(); if(!IsEmbedDB && (flyway.info().current() == null || flyway.info().current().getVersion().getMajor().intValue() < 43)) { - System.out.println("数据库升级需要较长时间，可能长达数分钟或更长，请耐心等待，不要中断。。。"); + System.out.println("数据库执行初始化或升级需要较长时间，可能长达数分钟或更长，请耐心等待，不要中断。。。"); } flyway.migrate(); diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf index d1d7a94d..59831bf5 100755 --- a/distribution/src/main/resources/wildfirechat.conf +++ b/distribution/src/main/resources/wildfirechat.conf @@ -69,6 +69,10 @@ message.blacklist.strategy 0 ##是否禁止服务器端消息搜索，该功能暂未实现。目前的影响是如果打开，则存储消息时不单独保存_searchable_content字段 message.disable_remote_search false +##是否数据库中加密消息内容。注意这个开关在服务运行起来后不能改变了，避免读取数据库时无法恢复历史消息。 +##如果要避免数据库中明文存储消息内容，请打开这个开关和message.disable_remote_search开关 +message.encrypt_message_content false + ##允许客户端撤回用户自己发送的消息时限，单位是秒。群组管理员或群主撤回群成员消息、server api不受此限制，可以撤回任意时限内的消息 message.recall_time_limit 120