From ffd1492ea2ebfe7610282bb58aa04b24a29b468b Mon Sep 17 00:00:00 2001 From: heavyrain2012 Date: Tue, 26 Oct 2021 10:11:05 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E9=99=90=E9=A2=91=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- broker/config/wildfirechat.conf | 14 ++++++++ .../com/xiaoleilu/loServer/action/Action.java | 33 +++++++++++++++++++ .../loServer/action/admin/AdminAction.java | 19 ++++++----- .../action/admin/BlockUserAction.java | 2 +- .../action/channel/ChannelAction.java | 3 +- .../loServer/action/robot/RobotAction.java | 3 +- .../java/io/moquette/BrokerConstants.java | 4 +++ .../java/io/moquette/imhandler/IMHandler.java | 23 ++++++++++--- .../main/java/io/moquette/server/Server.java | 2 ++ .../moquette/spi/impl/Qos1PublishHandler.java | 16 +-------- .../src/main/resources/wildfirechat.conf | 14 ++++++++ .../main/java/cn/wildfirechat/sdk/Main.java | 4 +-- 12 files changed, 101 insertions(+), 36 deletions(-) diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index 2d47d456..7287ba96 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -154,6 +154,20 @@ group.visible_quit_or_kickoff_notification 0 # 因此需要分段多次同步，此开关定义了每次同步的数据的条数。此需要客户端支持，客户端SDK在2021.5.25日后支持此功能。如果确认所有客户端都在这个日志之后才可以打开此设置。 #sync.data_part_size 1000 +#********************************************************************* +# 限频配置 +#********************************************************************* +##API接口限频设置，时间是10秒内允许的请求次数。 +##管理api频率限制，默认是10秒10000次。 +http.admin.rate_limit 10000 +##机器人api频率限制，默认是10秒1000次，按照机器人ID区分。 +http.robot.rate_limit 1000 +##频道api频率限制，默认是10秒1000次，按照频道ID区分。 +http.channel.rate_limit 1000 + +##客户端请求频率限制，默认是5秒100次。通常情况下不要修改这个值。 +client.request_rate_limit 100 + #********************************************************************* # hazelcase configuration #********************************************************************* diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java b/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java index bb36da53..44a376b4 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java @@ -13,6 +13,7 @@ import com.xiaoleilu.loServer.RestResult; import com.xiaoleilu.loServer.annotation.RequireAuthentication; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; +import io.moquette.server.config.IConfig; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.Utils; @@ -21,10 +22,12 @@ import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import cn.wildfirechat.common.ErrorCode; +import win.liyufan.im.RateLimiter; import java.nio.charset.StandardCharsets; import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_SUCCESS; +import static io.moquette.BrokerConstants.*; /** * 请求处理接口
@@ -38,6 +41,36 @@ abstract public class Action { public static ISessionsStore sessionsStore = null; public ChannelHandlerContext ctx; + + protected static RateLimiter adminLimiter = null; + protected static RateLimiter robotLimiter = null; + protected static RateLimiter channelLimiter = null; + + public static void init(IConfig config) { + int adminRate = 10000; + int robotRate = 1000; + int channelRate = 1000; + + try { + adminRate = Integer.parseInt(config.getProperty(HTTP_ADMIN_RATE_LIMIT, "10000")); + } catch (NumberFormatException e) { + + } + try { + robotRate = Integer.parseInt(config.getProperty(HTTP_ROBOT_RATE_LIMIT, "1000")); + } catch (NumberFormatException e) { + + } + try { + channelRate = Integer.parseInt(config.getProperty(HTTP_CHANNEL_RATE_LIMIT, "1000")); + } catch (NumberFormatException e) { + + } + adminLimiter = new RateLimiter(10, adminRate); + robotLimiter = new RateLimiter(10, robotRate); + channelLimiter = new RateLimiter(10, channelRate); + } + protected class Result { Object data; ErrorCode errorCode; diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java index 7d677de3..624beba1 100644 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java @@ -29,7 +29,6 @@ abstract public class AdminAction extends Action { private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(AdminAction.class); private static String SECRET_KEY = "123456"; private static boolean NO_CHECK_TIME = false; - private final RateLimiter mLimitCounter = new RateLimiter(10, 500); public static void setSecretKey(String secretKey) { SECRET_KEY = secretKey; } @@ -44,7 +43,7 @@ abstract public class AdminAction extends Action { @Override public ErrorCode preAction(Request request, Response response) { - if (!mLimitCounter.isGranted("admin")) { + if (!adminLimiter.isGranted("admin")) { return ErrorCode.ERROR_CODE_OVER_FREQUENCY; } @@ -88,14 +87,16 @@ abstract public class AdminAction extends Action { } protected void sendResponse(Response response, ErrorCode errorCode, Object data) { - response.setStatus(HttpResponseStatus.OK); - if (errorCode == null) { - errorCode = ErrorCode.ERROR_CODE_SUCCESS; - } + if(response != null) { + response.setStatus(HttpResponseStatus.OK); + if (errorCode == null) { + errorCode = ErrorCode.ERROR_CODE_SUCCESS; + } - RestResult result = RestResult.resultOf(errorCode, errorCode.getMsg(), data); - response.setContent(new Gson().toJson(result)); - response.send(); + RestResult result = RestResult.resultOf(errorCode, errorCode.getMsg(), data); + response.setContent(new Gson().toJson(result)); + response.send(); + } } protected void sendApiMessage(Response response, String fromUser, String topic, byte[] message, ApiCallback callback) { sendApiMessage(response, fromUser, null, topic, message, callback, false); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java index c8ff5ee0..0d4d74cf 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java @@ -45,7 +45,7 @@ public class BlockUserAction extends AdminAction { response.setContent(new Gson().toJson(result)); if (inputUserBlock.getStatus() == 2) { - sendApiMessage(response, null, ServerAPIHelper.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), null); + sendApiMessage(null, null, ServerAPIHelper.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), null); } sendResponse(response, ErrorCode.ERROR_CODE_SUCCESS, null); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java index 4a68a83e..ccf9055f 100644 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java @@ -31,7 +31,6 @@ import java.util.concurrent.Executor; abstract public class ChannelAction extends Action { private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(ChannelAction.class); - private final RateLimiter mLimitCounter = new RateLimiter(10, 200); protected WFCMessage.ChannelInfo channelInfo; protected interface Callback { @@ -61,7 +60,7 @@ abstract public class ChannelAction extends Action { return ErrorCode.ERROR_CODE_API_NOT_SIGNED; } - if (!mLimitCounter.isGranted(cid)) { + if (!channelLimiter.isGranted(cid)) { return ErrorCode.ERROR_CODE_OVER_FREQUENCY; } diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java index 22b1968b..9eeff045 100644 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java @@ -33,7 +33,6 @@ import java.util.concurrent.Executor; abstract public class RobotAction extends Action { protected static final Logger LOG = LoggerFactory.getLogger(RobotAction.class); - private final RateLimiter mLimitCounter = new RateLimiter(10, 1000); protected WFCMessage.Robot robot; @@ -64,7 +63,7 @@ abstract public class RobotAction extends Action { return ErrorCode.ERROR_CODE_API_NOT_SIGNED; } - if (!mLimitCounter.isGranted(rid)) { + if (!robotLimiter.isGranted(rid)) { return ErrorCode.ERROR_CODE_OVER_FREQUENCY; } diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 4255160c..6e1a962d 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -178,6 +178,10 @@ public final class BrokerConstants { 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"; + public static final String HTTP_ADMIN_RATE_LIMIT = "http.admin.rate_limit"; + public static final String HTTP_ROBOT_RATE_LIMIT = "http.robot.rate_limit"; + public static final String HTTP_CHANNEL_RATE_LIMIT = "http.channel.rate_limit"; + public static final String CLIENT_REQUEST_RATE_LIMIT = "client.request_rate_limit"; private BrokerConstants() { } diff --git a/broker/src/main/java/io/moquette/imhandler/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java index 8a0734d0..4e4385b6 100755 --- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java @@ -38,6 +38,7 @@ import java.util.*; import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_OVER_FREQUENCY; import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_SUCCESS; +import static io.moquette.BrokerConstants.CLIENT_REQUEST_RATE_LIMIT; /** * 请求处理接口
@@ -52,7 +53,7 @@ abstract public class IMHandler { protected static Server mServer = null; protected static MessagesPublisher publisher; private static ThreadPoolExecutorWrapper m_imBusinessExecutor; - private static final RateLimiter mLimitCounter = new RateLimiter(5, 100); + private static RateLimiter mLimitCounter; private Method parseDataMethod; private Class dataCls; @@ -147,12 +148,24 @@ abstract public class IMHandler { publisher = p; m_imBusinessExecutor = businessExecutor; mServer = server; + int clientRateLimit = 100; + try { + clientRateLimit = Integer.parseInt(server.getConfig().getProperty(CLIENT_REQUEST_RATE_LIMIT, "100")); + } catch (Exception e) { + + } + + if(clientRateLimit == 0) { + clientRateLimit = 100; + } + + mLimitCounter = new RateLimiter(5, clientRateLimit); } - public ErrorCode preAction(String clientID, String fromUser, String topic, Qos1PublishHandler.IMCallback callback) { + public ErrorCode preAction(String clientID, String fromUser, String topic, Qos1PublishHandler.IMCallback callback, boolean isAdmin, boolean isRobotOrChannel) { LOG.info("imHandler fromUser={}, clientId={}, topic={}", fromUser, clientID, topic); - if(!mLimitCounter.isGranted(clientID + fromUser + topic)) { + if(!isAdmin && !isRobotOrChannel && !mLimitCounter.isGranted(clientID + fromUser + topic)) { ByteBuf ackPayload = Unpooled.buffer(); ackPayload.ensureWritable(1).writeByte(ERROR_CODE_OVER_FREQUENCY.getCode()); try { @@ -166,7 +179,7 @@ abstract public class IMHandler { return ErrorCode.ERROR_CODE_SUCCESS; } - public void doHandler(String clientID, String fromUser, String topic, byte[] payloadContent, Qos1PublishHandler.IMCallback callback, boolean isAdmin) { + public void doHandler(String clientID, String fromUser, String topic, byte[] payloadContent, Qos1PublishHandler.IMCallback callback, boolean isAdmin, boolean isRobotOrChannel) { m_imBusinessExecutor.execute(() -> { Qos1PublishHandler.IMCallback callbackWrapper = new Qos1PublishHandler.IMCallback() { @Override @@ -177,7 +190,7 @@ abstract public class IMHandler { } }; - ErrorCode preActionCode = preAction(clientID, fromUser, topic, callbackWrapper); + ErrorCode preActionCode = preAction(clientID, fromUser, topic, callbackWrapper, isAdmin, isRobotOrChannel); if (preActionCode == ErrorCode.ERROR_CODE_SUCCESS) { ByteBuf ackPayload = Unpooled.buffer(1); diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index e0a76deb..f2462612 100755 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -24,6 +24,7 @@ import com.hazelcast.config.FileSystemXmlConfig; import com.hazelcast.core.*; import com.xiaoleilu.loServer.LoServer; import com.xiaoleilu.loServer.ServerSetting; +import com.xiaoleilu.loServer.action.Action; import com.xiaoleilu.loServer.action.admin.AdminAction; import io.moquette.BrokerConstants; import io.moquette.persistence.*; @@ -257,6 +258,7 @@ public class Server { initMediaServerConfig(config); + Action.init(config); String monitorEventAddress = config.getProperty(MONITOR_Exception_Event_Address); Utility.setMonitorEventAddress(monitorEventAddress); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 2ee2def6..c0fae6ba 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -69,7 +69,6 @@ public class Qos1PublishHandler extends QosPublishHandler { private final MessagesPublisher publisher; private final ISessionsStore m_sessionStore; private final ThreadPoolExecutorWrapper m_imBusinessExecutor; - private final RateLimiter mLimitCounter = new RateLimiter(5, 100); private HashMap m_imHandlers = new HashMap<>(); @@ -148,19 +147,6 @@ public class Qos1PublishHandler extends QosPublishHandler { void imHandler(String clientID, String fromUser, String topic, byte[] payloadContent, IMCallback callback, boolean isAdmin, boolean isRobotOrChannel) { LOG.info("imHandler fromUser={}, topic={}", fromUser, topic); - if(!isAdmin && !isRobotOrChannel && !mLimitCounter.isGranted(clientID + fromUser + topic)) { - ByteBuf ackPayload = Unpooled.buffer(); - ackPayload.ensureWritable(1).writeByte(ERROR_CODE_OVER_FREQUENCY.getCode()); - try { - callback.onIMHandled(ERROR_CODE_OVER_FREQUENCY, ackPayload); - } catch (Exception e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); - } - LOG.warn("user {} request over frequency", fromUser); - return; - } - IMCallback wrapper = (errorcode, ackPayload) -> { ackPayload.resetReaderIndex(); byte code = ackPayload.readByte(); @@ -215,7 +201,7 @@ public class Qos1PublishHandler extends QosPublishHandler { IMHandler handler = m_imHandlers.get(topic); if (handler != null) { - handler.doHandler(clientID, fromUser, topic, payloadContent, wrapper, isAdmin); + handler.doHandler(clientID, fromUser, topic, payloadContent, wrapper, isAdmin, isRobotOrChannel); } else { LOG.error("imHandler unknown topic={}", topic); ByteBuf ackPayload = Unpooled.buffer(); diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf index 2079ed15..4275356a 100755 --- a/distribution/src/main/resources/wildfirechat.conf +++ b/distribution/src/main/resources/wildfirechat.conf @@ -150,6 +150,20 @@ group.visible_quit_or_kickoff_notification 0 # 因此需要分段多次同步，此开关定义了每次同步的数据的条数。此需要客户端支持，客户端SDK在2021.5.25日后支持此功能。如果确认所有客户端都在这个日志之后才可以打开此设置。 #sync.data_part_size 1000 +#********************************************************************* +# 限频配置 +#********************************************************************* +##API接口限频设置，时间是10秒内允许的请求次数。 +##管理api频率限制，默认是10秒10000次。 +http.admin.rate_limit 10000 +##机器人api频率限制，默认是10秒1000次，按照机器人ID区分。 +http.robot.rate_limit 1000 +##频道api频率限制，默认是10秒1000次，按照频道ID区分。 +http.channel.rate_limit 1000 + +##客户端请求频率限制，默认是5秒100次。通常情况下不要修改这个值。 +client.request_rate_limit 100 + #********************************************************************* # hazelcase configuration #********************************************************************* diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java index a69dd74a..de93efee 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java @@ -292,7 +292,7 @@ public class Main { } IMResult result = RelationAdmin.sendFriendRequest("ff1", "ff2", "hello", true); - if (result != null && result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + if (result != null && (result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS || result.getErrorCode() == ErrorCode.ERROR_CODE_ALREADY_FRIENDS)) { System.out.println("send friend request success"); } else { System.out.println("failure"); @@ -308,7 +308,7 @@ public class Main { } result = RelationAdmin.sendFriendRequest("ff1", "ff2", "hello3", true); - if (result != null && result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + if (result != null && (result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS || result.getErrorCode() == ErrorCode.ERROR_CODE_ALREADY_FRIENDS)) { System.out.println("success"); } else { System.out.println("send friend request success");