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);