diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index 44fc52f0..ebb2e0b9 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -241,5 +241,16 @@ push.ios.server.address http://localhost:8085/ios/push #********************************************************************* # Sensitive configuration #********************************************************************* -##文本敏感词过滤处理方法，0 发送失败；1 发送成功但消息被服务器直接丢弃；2 命中的敏感词被替换成***；3 记录并继续发送 +## 内置文本敏感词过滤处理方法，0 发送失败；1 发送成功但消息被服务器直接丢弃；2 命中的敏感词被替换成***；3 记录并继续发送 sensitive.filter.type 0 + +## 如果内置敏感词库无法满足您的需求，可以自己开发部署敏感词审核服务。配置如下地址和消息类型，IM服务会把指定消息类型的消息回调到指定地址 +## 审核服务返回状态码200表示继续发送，如果需要替换内容，返回状态码200并且内容为替换后的内容。返回403表示不允许发送。其它错误继续发送。 +## 独立审核服务地址，IM服务会post SendMessageData。如果需要修改消息内容，返回 MessagePayload。请参考应用服务中 "/message/censor" Mapping. +#sensitive.remote_server_url http://192.168.3.202:8888/message/censor +## 需要进行敏感词审核的消息类型，当有多个时以英文逗号分割 +#sensitive.remote_sensitive_message_type 1,2,3 +## 如果远程审核服务返回403，返回给发送端发送失败还是成功？实际上这个值影响发送响应是否等待审核结果，当为false时，消息发送给审核服务就立即返回 +## 成功给客户端。当为true时，消息发送给审核服务，等待审核结果，如果结果为继续发送就返回给客户端为成功，结果为禁止发送就返回失败给客户端。 +## 客户端操作的超时一般为10s左右，且考虑到复杂的网络情况，建议处理时间不要超过3s。建议这个值为false。 +#sensitive.remote_fail_when_matched false diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 3260167c..8ed694be 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -121,6 +121,10 @@ public final class BrokerConstants { public static final String SENSITIVE_Filter_Type = "sensitive.filter.type"; + public static final String SENSITIVE_Remote_Server_URL = "sensitive.remote_server_url"; + public static final String SENSITIVE_Remote_Message_Type = "sensitive.remote_sensitive_message_type"; + public static final String SENSITIVE_Remote_Fail_When_Matched = "sensitive.remote_fail_when_matched"; + public static final String MESSAGE_Forward_Url = "message.forward.url"; public static final String SERVER_MULTI_ENDPOINT = "server.multi_endpoint"; diff --git a/broker/src/main/java/io/moquette/imhandler/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java index c4d39df0..8a0734d0 100755 --- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java @@ -204,8 +204,9 @@ abstract public class IMHandler { errorCode = ErrorCode.ERROR_CODE_SERVER_ERROR; } } - - response(ackPayload, errorCode, callback); + if(errorCode != ErrorCode.INVALID_ASYNC_HANDLING) { + response(ackPayload, errorCode, callback); + } } else { LOG.error("Handler {} preAction failure", this.getClass().getName()); ByteBuf ackPayload = Unpooled.buffer(1); diff --git a/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java b/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java index fad182bc..3c05bce5 100644 --- a/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java @@ -8,26 +8,36 @@ package io.moquette.imhandler; +import cn.wildfirechat.pojos.MessagePayload; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; +import com.google.gson.Gson; import com.hazelcast.util.StringUtil; import io.moquette.BrokerConstants; import io.moquette.spi.impl.Qos1PublishHandler; import io.netty.buffer.ByteBuf; import cn.wildfirechat.common.ErrorCode; +import io.netty.buffer.Unpooled; +import win.liyufan.im.HttpUtils; import win.liyufan.im.IMTopic; import win.liyufan.im.MessageShardingUtil; import win.liyufan.im.Utility; +import java.util.HashSet; import java.util.Set; -import static cn.wildfirechat.proto.ProtoConstants.ContentType.Text; +import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_SUCCESS; @Handler(value = IMTopic.SendMessageTopic) public class SendMessageHandler extends IMHandler { private int mSensitiveType = 0; //命中敏感词时，0 失败，1 吞掉， 2 敏感词替换成*。 private String mForwardUrl = null; private int mBlacklistStrategy = 0; //黑名单中时，0失败，1吞掉。 + + private String mRemoteSensitiveServerUrl = null; + private Set mRemoteSensitiveMessageTypes; + private boolean mWaitRemoteSensitiveServerResponse = false; + public SendMessageHandler() { super(); @@ -49,6 +59,24 @@ public class SendMessageHandler extends IMHandler { e.printStackTrace(); Utility.printExecption(LOG, e); } + try { + mRemoteSensitiveServerUrl = mServer.getConfig().getProperty(BrokerConstants.SENSITIVE_Remote_Server_URL); + if(!StringUtil.isNullOrEmpty(mRemoteSensitiveServerUrl)) { + mWaitRemoteSensitiveServerResponse = Boolean.parseBoolean(mServer.getConfig().getProperty(BrokerConstants.SENSITIVE_Remote_Fail_When_Matched, "false")); + String types = mServer.getConfig().getProperty(BrokerConstants.SENSITIVE_Remote_Message_Type, ""); + mRemoteSensitiveMessageTypes = new HashSet<>(); + if(!StringUtil.isNullOrEmpty(types)) { + String[] ts = types.split(","); + for (String t:ts) { + mRemoteSensitiveMessageTypes.add(Integer.parseInt(t.trim())); + } + } + } + } catch (Exception e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } + } @Override @@ -105,7 +133,7 @@ public class SendMessageHandler extends IMHandler { } long timestamp = System.currentTimeMillis(); - long messageId = 0; + final long messageId; try { messageId = MessageShardingUtil.generateId(); } catch (Exception e) { @@ -119,21 +147,88 @@ public class SendMessageHandler extends IMHandler { } if (!isAdmin) { - 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) { - ignoreMsg = true; - } else if(mSensitiveType == 2) { - String text = message.getContent().getSearchableContent(); - for (String word : matched) { - text = text.replace(word, "***"); + if(StringUtil.isNullOrEmpty(mRemoteSensitiveServerUrl)) { + 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) { + ignoreMsg = true; + } else if (mSensitiveType == 2) { + String text = message.getContent().getSearchableContent(); + for (String word : matched) { + text = text.replace(word, "***"); + } + message = message.toBuilder().setContent(message.getContent().toBuilder().setSearchableContent(text).build()).build(); } - message = message.toBuilder().setContent(message.getContent().toBuilder().setSearchableContent(text).build()).build(); - } else if(mSensitiveType == 3) { + } + } else { + if(mRemoteSensitiveMessageTypes.contains(message.getContent().getType())) { + final WFCMessage.Message finalMsg = message; + publisher.forwardMessageWithCallback(message, mRemoteSensitiveServerUrl, new HttpUtils.HttpCallback() { + @Override + public void onSuccess(String content) { + if(StringUtil.isNullOrEmpty(content)) { + saveAndPublish(fromUser, clientID, finalMsg); + } else { + MessagePayload payload = new Gson().fromJson(content, MessagePayload.class); + if (payload != null && payload.getType() > 0) { + WFCMessage.Message newMsg = finalMsg.toBuilder().setContent(payload.toProtoMessageContent()).build(); + saveAndPublish(fromUser, clientID, newMsg); + } else { + LOG.error("Response content {} from censor is invalid payload, ignore response and send original message", content); + saveAndPublish(fromUser, clientID, finalMsg); + } + } + if(mWaitRemoteSensitiveServerResponse) { + ByteBuf ackPayload = Unpooled.buffer(21); + ErrorCode errorCode = ERROR_CODE_SUCCESS; + ackPayload.writeByte(errorCode.getCode()); + ackPayload.writeLong(messageId); + ackPayload.writeLong(timestamp); + callback.onIMHandled(ErrorCode.ERROR_CODE_SUCCESS, ackPayload); + } + } + @Override + public void onFailure(int statusCode, String errorMessage) { + ByteBuf ackPayload = null; + ErrorCode errorCode = ERROR_CODE_SUCCESS; + if(statusCode == 403) { + if(mWaitRemoteSensitiveServerResponse) { + errorCode = ErrorCode.ERROR_CODE_SENSITIVE_MATCHED; + ackPayload = Unpooled.buffer(1); + ackPayload.writeByte(errorCode.getCode()); + } + } else { + LOG.warn("Failed to censor message with status {}, errorMessage {}. Send the original message", statusCode, errorMessage); + saveAndPublish(fromUser, clientID, finalMsg); + + if(mWaitRemoteSensitiveServerResponse) { + errorCode = ERROR_CODE_SUCCESS; + ackPayload = Unpooled.buffer(21); + ackPayload.writeByte(errorCode.getCode()); + ackPayload.writeLong(messageId); + ackPayload.writeLong(timestamp); + } + } + + if(mWaitRemoteSensitiveServerResponse) { + callback.onIMHandled(errorCode, ackPayload); + } + } + }); + + if(mWaitRemoteSensitiveServerResponse) { + ackPayload.clear(); + return ErrorCode.INVALID_ASYNC_HANDLING; + } else { + ackPayload = ackPayload.capacity(20); + ackPayload.writeLong(messageId); + ackPayload.writeLong(timestamp); + return ErrorCode.ERROR_CODE_SUCCESS; + } } } } diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 9f72319c..ecf011f5 100755 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -624,6 +624,29 @@ public class MessagesPublisher { }); } + public void forwardMessageWithCallback(final WFCMessage.Message message, String forwardUrl, HttpUtils.HttpCallback callback) { + try { + HttpUtils.httpJsonPost(forwardUrl, new Gson().toJson(OutputMessageData.fromProtoMessage(message), OutputMessageData.class), new HttpUtils.HttpCallback() { + @Override + public void onSuccess(String content) { + Server.getServer().getImBusinessScheduler().execute(()->{ + callback.onSuccess(content); + }); + } + + @Override + public void onFailure(int statusCode, String errorMessage) { + Server.getServer().getImBusinessScheduler().execute(()->{ + callback.onFailure(statusCode, errorMessage); + }); + } + }); + } catch (Exception e) { + e.printStackTrace(); + Utility.printExecption(LOG, e, EVENT_CALLBACK_Exception); + } + } + public void notifyChannelListenStatusChanged(WFCMessage.ChannelInfo channelInfo, String user, boolean listen) { if (channelInfo == null || StringUtil.isNullOrEmpty(channelInfo.getCallback())) { return; diff --git a/broker/src/main/java/win/liyufan/im/HttpUtils.java b/broker/src/main/java/win/liyufan/im/HttpUtils.java index def9e5a6..b5741e20 100644 --- a/broker/src/main/java/win/liyufan/im/HttpUtils.java +++ b/broker/src/main/java/win/liyufan/im/HttpUtils.java @@ -30,8 +30,16 @@ public class HttpUtils { private static final Logger LOG = LoggerFactory.getLogger(HttpUtils.class); public static final MediaType JSON = MediaType.get("application/json; charset=utf-8"); static private OkHttpClient client = new OkHttpClient(); + public interface HttpCallback { + void onSuccess(String content); + void onFailure(int statusCode, String errorMessage); + } - public static void httpJsonPost(final String url, final String jsonStr){ + public static void httpJsonPost(final String url, final String jsonStr) { + httpJsonPost(url, jsonStr, null); + } + + public static void httpJsonPost(final String url, final String jsonStr, final HttpCallback httpCallback) { //消息推送内容为 {"sender":"uCGUxUaa","senderName":"杨","convType":0,"target":"usq7v7UU","targetName":"鞋子","userId":"usq7v7UU","line":0,"cntType":400,"serverTime":1610590766485,"pushMessageType":1,"pushType":2,"pushContent":"","pushData":"","unReceivedMsg":1,"mentionedType":0,"packageName":"cn.wildfirechat.chat","deviceToken":"AFoieP9P6u6CccIkRK23gRwUJWKqSkdiqnb-6gC1kL7Wv-9XNoEYBPU7VsINU_q8_WTKfafe35qWu7ya7Z-NmgOTX9XVW3A3zd6ilh--quj6ccINXRvVnh8QmI9QQ","isHiddenDetail":false,"language":"zh"} //推送信息只打印前100个字符，防止敏感信息打印到日志中去。 LOG.info("POST to {} with data {}...", url, jsonStr.substring(0, Math.min(jsonStr.length(), 100))); @@ -50,12 +58,27 @@ public class HttpUtils { public void onFailure(Call call, IOException e) { e.printStackTrace(); LOG.info("POST to {} with data {} failure", url, jsonStr); + if(httpCallback != null) { + httpCallback.onFailure(-1, e.getMessage()); + } } @Override public void onResponse(Call call, Response response) throws IOException { LOG.info("POST to {} success with response: {}", url, response.body()); try { + if(httpCallback != null) { + int code = response.code(); + if(code == 200) { + if(response.body() != null && response.body().contentLength() > 0) { + httpCallback.onSuccess(response.body().string()); + } else { + httpCallback.onSuccess(null); + } + } else { + httpCallback.onFailure(code, response.message()); + } + } if (response.body() != null) { response.body().close(); } diff --git a/common/src/main/java/cn/wildfirechat/common/ErrorCode.java b/common/src/main/java/cn/wildfirechat/common/ErrorCode.java index 03b7a741..7e697cde 100755 --- a/common/src/main/java/cn/wildfirechat/common/ErrorCode.java +++ b/common/src/main/java/cn/wildfirechat/common/ErrorCode.java @@ -54,6 +54,7 @@ public enum ErrorCode { ERROR_CODE_TIMEOUT(249, "timeout"), ERROR_CODE_OVER_FREQUENCY(250, "over frequency"), INVALID_PARAMETER(251, "Invalid parameter"), + INVALID_ASYNC_HANDLING(252, "Async handling"), ERROR_CODE_NOT_EXIST(253, "not exist"), ERROR_CODE_NOT_IMPLEMENT(254, "not implement"), diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf index 427d1393..8c9929ed 100755 --- a/distribution/src/main/resources/wildfirechat.conf +++ b/distribution/src/main/resources/wildfirechat.conf @@ -250,5 +250,16 @@ push.ios.server.address http://localhost:8085/ios/push #********************************************************************* # Sensitive configuration #********************************************************************* -##文本敏感词过滤处理方法，0 发送失败；1 发送成功但消息被服务器直接丢弃；2 命中的敏感词被替换成*** +## 内置文本敏感词过滤处理方法，0 发送失败；1 发送成功但消息被服务器直接丢弃；2 命中的敏感词被替换成***；3 记录并继续发送 sensitive.filter.type 0 + +## 如果内置敏感词库无法满足您的需求，可以自己开发部署敏感词审核服务。配置如下地址和消息类型，IM服务会把指定消息类型的消息回调到指定地址 +## 审核服务返回状态码200表示继续发送，如果需要替换内容，返回状态码200并且内容为替换后的内容。返回403表示不允许发送。其它错误继续发送。 +## 独立审核服务地址，IM服务会post SendMessageData。如果需要修改消息内容，返回 MessagePayload。请参考应用服务中 "/message/censor" Mapping. +#sensitive.remote_server_url http://192.168.3.202:8888/message/censor +## 需要进行敏感词审核的消息类型，当有多个时以英文逗号分割 +#sensitive.remote_sensitive_message_type 1,2,3 +## 如果远程审核服务返回403，返回给发送端发送失败还是成功？实际上这个值影响发送响应是否等待审核结果，当为false时，消息发送给审核服务就立即返回 +## 成功给客户端。当为true时，消息发送给审核服务，等待审核结果，如果结果为继续发送就返回给客户端为成功，结果为禁止发送就返回失败给客户端。 +## 客户端操作的超时一般为10s左右，且考虑到复杂的网络情况，建议处理时间不要超过3s。建议这个值为false。 +#sensitive.remote_fail_when_matched false