Compare commits

..
7 Commits
17 changed files with 220 additions and 4069 deletions
+3 -2
View File
@@ -7,7 +7,7 @@
| [android-chat](https://github.com/wildfirechat/android-chat) | 野火IM Android SDK源码和App源码 |可以很方便地进行二次开发,或集成到现有应用当中 |
| [ios-chat](https://github.com/wildfirechat/ios-chat) | 野火IM iOS SDK源码和App源码 |可以很方便地进行二次开发,或集成到现有应用当中 |
| [pc-chat](https://github.com/wildfirechat/pc-chat) | 基于[Electron](https://electronjs.org/)开发的PC平台应用 | |
| [web-chat](https://github.com/wildfirechat/web-chat) | Web平台的Demo, [体验地址](http://web.wildfirechat.cn) | |
| [web-chat](https://github.com/wildfirechat/web-chat) | Web平台的Demo, [体验地址](http://web.wildfirechat.net) | |
| [wx-chat](https://github.com/wildfirechat/wx-chat) | 微信小程序平台的Demo | |
| [server](https://github.com/wildfirechat/server) | IM server | |
| [app server](https://github.com/wildfirechat/app_server) | 应用服务端 | |
@@ -16,7 +16,7 @@
| [docs](https://github.com/wildfirechat/docs) | 野火IM相关文档,包含设计、概念、开发、使用说明 | | | |
# server
本工程为野火IM 社区版IM服务软件。野火IM作为一个通用的即时通讯解决方案,可以集成到各种应用中。请阅读[docs](http://docs.wildfirechat.cn)或下载服务器[发布版本](https://github.com/wildfirechat/server/releases)
本工程为野火IM 社区版IM服务软件。野火IM作为一个通用的即时通讯解决方案,可以集成到各种应用中。请阅读[docs](http://docs.wildfirechat.net)或下载服务器[发布版本](https://github.com/wildfirechat/server/releases)
开发一套IM系统真的很艰辛,请路过的朋友们给点个star,支持我们坚持下去🙏🙏🙏🙏🙏
@@ -73,6 +73,7 @@
## 升级说明
1. 从0.42 版本增加了群成员数限制,默认为2000。如果您从之前的版本升级到这个版本或以后,需要注意到群成员数的限制。升级之后超出限制的群不受影响,但不能继续加人,如果您想修改默认值,可以在升级版本之后,修改t_setting表,把默认的大小改为您期望的人数。另外修改t_group表,把已经存在的群组max_member_count改成您期望的,然后重启。
2. 0.46和0.47版本升级到0.48及以后版本时,可能会提示flyway migrate 38错误,请执行 [修复脚本](https://github.com/wildfirechat/server/blob/wildfirechat/flyway_repaire_migrate_38.sql) 进行修复。0.46和0.47版本之外的版本不会出现此问题。
3. 从0.54之前版本升级到0.54及以后版本时,会提示flyway migrate错误。因为0.54版本删除了sql脚本中默认敏感词的内容,flyway checksum失败。请执行```update flyway_schema_history set checksum = 0 where script = 'V17__add_default_sensitive_word.sql';```来修复。
## 特别感谢
1. [moquette](https://github.com/moquette-io/moquette) 本项目是基于此项目二次开发而来,处理MQTT相关业务。
+14 -1
View File
@@ -213,6 +213,8 @@ push.ios.server.address http://localhost:8085/ios/push
##消息转发地址
##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!!
#message.forward.url http://localhost:8888/im_event/message
##需要转发的消息类型,当有多个时以英文逗号分割。如果转发所有消息,请注释掉配置或者设置为空
#message.forward.types 1,2,3
##设备信息转发地址
##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!!
@@ -241,5 +243,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
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -5,7 +5,7 @@
<relativePath>../</relativePath>
<artifactId>wildfirechat-parent</artifactId>
<groupId>cn.wildfirechat</groupId>
<version>0.53</version>
<version>0.54</version>
</parent>
<artifactId>moquette-broker</artifactId>
@@ -121,7 +121,12 @@ 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 MESSAGE_Forward_Types = "message.forward.types";
public static final String SERVER_MULTI_ENDPOINT = "server.multi_endpoint";
public static final String SERVER_MULTI_PLATFROM_NOTIFICATION = "server.multi_platform_notification";
@@ -204,8 +204,9 @@ abstract public class IMHandler<T> {
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);
@@ -8,32 +8,51 @@
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<WFCMessage.Message> {
private int mSensitiveType = 0; //命中敏感词时,0 失败,1 吞掉, 2 敏感词替换成*。
private String mForwardUrl = null;
private Set<Integer> mForwardMessageTypes = new HashSet<>();
private int mBlacklistStrategy = 0; //黑名单中时,0失败,1吞掉。
private String mRemoteSensitiveServerUrl = null;
private Set<Integer> mRemoteSensitiveMessageTypes;
private boolean mWaitRemoteSensitiveServerResponse = false;
public SendMessageHandler() {
super();
String forwardUrl = mServer.getConfig().getProperty(BrokerConstants.MESSAGE_Forward_Url);
if (!StringUtil.isNullOrEmpty(forwardUrl)) {
mForwardUrl = forwardUrl;
String forwardTypes = mServer.getConfig().getProperty(BrokerConstants.MESSAGE_Forward_Types);
if (!StringUtil.isNullOrEmpty(forwardTypes)) {
String[] tss = forwardTypes.split(",");
for (String ts:tss) {
mForwardMessageTypes.add(Integer.parseInt(ts.trim()));
}
}
}
try {
@@ -49,6 +68,24 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
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 +142,7 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
}
long timestamp = System.currentTimeMillis();
long messageId = 0;
final long messageId;
try {
messageId = MessageShardingUtil.generateId();
} catch (Exception e) {
@@ -114,26 +151,93 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
}
message = message.toBuilder().setFromUser(fromUser).setMessageId(messageId).setServerTimestamp(timestamp).build();
if (mForwardUrl != null) {
if (mForwardUrl != null && (mForwardMessageTypes.isEmpty() || mForwardMessageTypes.contains(message.getContent().getType()))) {
publisher.forwardMessage(message, mForwardUrl);
}
if (!isAdmin) {
Set<String> 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<String> 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;
}
}
}
}
@@ -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;
@@ -188,8 +188,7 @@ public class ProtocolProcessor {
String clientId = payload.clientIdentifier();
LOG.info("Processing CONNECT message. CId={}, username={}", clientId, payload.userName());
if (msg.variableHeader().version() < MqttVersion.MQTT_3_1_1.protocolLevel() ||
msg.variableHeader().version() >= MqttVersion.Wildfire_Max.protocolLevel()) {
if (msg.variableHeader().version() < MqttVersion.MQTT_3_1_1.protocolLevel()) {
MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
LOG.error("MQTT protocol version is not valid. CId={}", clientId);
@@ -27,7 +27,7 @@ public enum MqttVersion {
MQTT_3_1_1("MQTT", (byte) 4),
Wildfire_1("MQTT", (byte) 5),
Wildfire_2("MQTT", (byte) 6),
Wildfire_Max("MQTT", (byte) 7);
Wildfire_Future("MQTT", (byte) 7);
private final String name;
private final byte level;
@@ -59,6 +59,9 @@ public enum MqttVersion {
}
}
}
if(protocolLevel > Wildfire_Future.level) {
return Wildfire_Future;
}
throw new MqttUnacceptableProtocolVersionException(protocolName + "is unknown protocol name");
}
@@ -68,6 +71,9 @@ public enum MqttVersion {
return mv;
}
}
if(protocolLevel > Wildfire_Future.level) {
return Wildfire_Future;
}
throw new MqttUnacceptableProtocolVersionException(protocolLevel + "is unknown protocol level");
}
}
@@ -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();
}
+1 -1
View File
@@ -5,7 +5,7 @@
<parent>
<artifactId>wildfirechat-parent</artifactId>
<groupId>cn.wildfirechat</groupId>
<version>0.53</version>
<version>0.54</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -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"),
+1 -1
View File
@@ -6,7 +6,7 @@
<!--<relativePath>../pom.xml</relativePath>-->
<artifactId>wildfirechat-parent</artifactId>
<groupId>cn.wildfirechat</groupId>
<version>0.53</version>
<version>0.54</version>
</parent>
<artifactId>distribution</artifactId>
@@ -212,6 +212,8 @@ push.ios.server.address http://localhost:8085/ios/push
##消息转发地址
##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!!
#message.forward.url http://localhost:8087/im_event/message
##需要转发的消息类型,当有多个时以英文逗号分割。如果转发所有消息,请注释掉配置或者设置为空
#message.forward.types 1,2,3
##设备信息转发地址
##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!!
@@ -250,5 +252,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
+1 -1
View File
@@ -12,7 +12,7 @@
<artifactId>wildfirechat-parent</artifactId>
<packaging>pom</packaging>
<version>0.53</version>
<version>0.54</version>
<name>Wildfire Chat</name>
<description>Wildfire Chat Server</description>
<inceptionYear>2019</inceptionYear>
+1 -1
View File
@@ -5,7 +5,7 @@
<parent>
<artifactId>wildfirechat-parent</artifactId>
<groupId>cn.wildfirechat</groupId>
<version>0.53</version>
<version>0.54</version>
</parent>
<modelVersion>4.0.0</modelVersion>