From a149139143dd88041b96ff4c7eb4e24704153097 Mon Sep 17 00:00:00 2001 From: "heavyrain.lee" Date: Wed, 12 Jun 2019 22:57:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=95=8F=E6=84=9F=E8=AF=8D?= =?UTF-8?q?=E5=91=BD=E4=B8=AD=E6=B6=88=E6=81=AF=E8=A1=A8t=5Fsensitive=5Fme?= =?UTF-8?q?ssages?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../migrate/h2/V9__add_sensitive_messages.sql | 15 ++++++ .../mysql/V9__add_sensitive_messages.sql | 18 +++++++ .../imhandler/SendMessageHandler.java | 1 + .../moquette/persistence/DatabaseStore.java | 51 +++++++++++++++++++ .../persistence/MemoryMessagesStore.java | 6 +++ .../java/io/moquette/spi/IMessagesStore.java | 1 + 6 files changed, 92 insertions(+) create mode 100755 broker/migrate/h2/V9__add_sensitive_messages.sql create mode 100755 broker/migrate/mysql/V9__add_sensitive_messages.sql diff --git a/broker/migrate/h2/V9__add_sensitive_messages.sql b/broker/migrate/h2/V9__add_sensitive_messages.sql new file mode 100755 index 00000000..da15c52a --- /dev/null +++ b/broker/migrate/h2/V9__add_sensitive_messages.sql @@ -0,0 +1,15 @@ + +DROP TABLE IF EXISTS `t_sensitive_messages`; +CREATE TABLE `t_sensitive_messages` ( + `id` int(11) NOT NULL PRIMARY KEY AUTO_INCREMENT, + `_mid` bigint(20) NOT NULL, + `_from` varchar(64) NOT NULL, + `_type` tinyint NOT NULL DEFAULT 0, + `_target` varchar(64) NOT NULL, + `_line` int(11) NOT NULL DEFAULT 0, + `_data` BLOB NOT NULL, + `_searchable_key` TEXT DEFAULT NULL, + `_dt` DATETIME NOT NULL DEFAULT NOW(), + `_content_type` int(11) NOT NULL DEFAULT '0', + UNIQUE INDEX `message_uid_index` (`_mid` ASC) +); diff --git a/broker/migrate/mysql/V9__add_sensitive_messages.sql b/broker/migrate/mysql/V9__add_sensitive_messages.sql new file mode 100755 index 00000000..22a445c8 --- /dev/null +++ b/broker/migrate/mysql/V9__add_sensitive_messages.sql @@ -0,0 +1,18 @@ + +DROP TABLE IF EXISTS `t_sensitive_messages`; +CREATE TABLE `t_sensitive_messages` ( + `id` int(11) NOT NULL PRIMARY KEY AUTO_INCREMENT, + `_mid` bigint(20) NOT NULL, + `_from` varchar(64) NOT NULL, + `_type` tinyint NOT NULL DEFAULT 0, + `_target` varchar(64) NOT NULL, + `_line` int(11) NOT NULL DEFAULT 0, + `_data` BLOB NOT NULL, + `_searchable_key` TEXT DEFAULT NULL, + `_dt` DATETIME NOT NULL, + `_content_type` int(11) NOT NULL DEFAULT '0', + INDEX `message_uid_index` (`_mid` DESC) + ) +ENGINE = InnoDB +DEFAULT CHARACTER SET = utf8mb4 +COLLATE = utf8mb4_unicode_ci; diff --git a/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java b/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java index e1f8cdfe..380d37a5 100644 --- a/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java @@ -81,6 +81,7 @@ public class SendMessageHandler extends IMHandler { if (!isAdmin && message.getContent().getType() == Text) { Set matched = m_messagesStore.handleSensitiveWord(message.getContent().getSearchableContent()); if (matched != null && !matched.isEmpty()) { + m_messagesStore.storeSensitiveMessage(message); if (mSensitiveType == 0) { errorCode = ErrorCode.ERROR_CODE_SENSITIVE_MATCHED; } else if(mSensitiveType == 1) { diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 9a0157e6..4f841a2e 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -424,6 +424,57 @@ public class DatabaseStore { }); } + void persistSensitiveMessage(final WFCMessage.Message message) { + if(message.getContent().getPersistFlag() == Transparent) { + return; + } + + mScheduler.execute(()-> { + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + String sql = "insert into " + "t_sensitive_messages" + + " (`_mid`, `_from`, `_type`, `_target`, `_line`, `_data`, `_searchable_key`, `_dt`, `_content_type`) values(?, ?, ?, ?, ?, ?, ?, ?, ?)" + + " ON DUPLICATE KEY UPDATE " + + "`_data` = ?," + + "`_searchable_key` = ?," + + "`_dt` = ?," + + "`_content_type` = ?"; + + String searchableContent = message.getContent().getSearchableContent() == null ? "" : message.getContent().getSearchableContent(); + + statement = connection.prepareStatement(sql); + int index = 1; + statement.setLong(index++, message.getMessageId()); + statement.setString(index++, message.getFromUser()); + statement.setInt(index++, message.getConversation().getType()); + statement.setString(index++, message.getConversation().getTarget()); + statement.setInt(index++, message.getConversation().getLine()); + Blob blob = connection.createBlob(); + blob.setBytes(1, message.getContent().toByteArray()); + statement.setBlob(index++, blob); + statement.setString(index++, searchableContent); + statement.setTimestamp(index++, new Timestamp(message.getServerTimestamp())); + statement.setInt(index++, message.getContent().getType()); + + statement.setBlob(index++, blob); + statement.setString(index++, searchableContent); + statement.setTimestamp(index++, new Timestamp(message.getServerTimestamp())); + statement.setInt(index++, message.getContent().getType()); + + int count = statement.executeUpdate(); + LOG.info("Update rows {}", count); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e); + } finally { + DBUtil.closeDB(connection, statement); + } + }); + } + Map getMessages(Collection keys) { Map> messageTableMap = new HashMap<>(); for (Long key : keys) { diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 16d8e6b2..00cb835c 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -169,6 +169,12 @@ public class MemoryMessagesStore implements IMessagesStore { return message; } + @Override + public void storeSensitiveMessage(WFCMessage.Message message) { + databaseStore.persistSensitiveMessage(message); + } + + @Override public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set notifyReceivers) { WFCMessage.Message message = messageBuilder.build(); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index e69aaef8..c91b2cb4 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -96,6 +96,7 @@ public interface IMessagesStore { DatabaseStore getDatabaseStore(); WFCMessage.Message storeMessage(String fromUser, String fromClientId, WFCMessage.Message message); + void storeSensitiveMessage(WFCMessage.Message message); int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder message, Set notifyReceivers); Set getAllEnds(); WFCMessage.PullMessageResult fetchMessage(String user, String exceptClientId, long fromMessageId, int pullType);