tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

IP rate limits for MQTT

pull/5875/head
Andrii Shvaika 5 years ago
parent
d8097d2b76
commit
467085a827
8 changed files with 138 additions and 20 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 6
      application/src/main/resources/thingsboard.yml
  2. 13
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
  3. 6
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
  4. 79
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java
  5. 33
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/InetAddressRateLimitStats.java
  6. 7
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitService.java
  7. 5
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
  8. 9
      transport/mqtt/src/main/resources/tb-mqtt-transport.yml

6
application/src/main/resources/thingsboard.yml
View File

@ -620,12 +620,12 @@ transport:
enabled: "${TB_TRANSPORT_LOG_ENABLED:true}"
max_length: "${TB_TRANSPORT_LOG_MAX_LENGTH:1024}"
rate_limits:
# Maximum number of simultaneous connections from a single ip address
max_connections_per_ip: "${TB_TRANSPORT_MAX_CONNECTIONS_PER_IP:50}"
# Enable or disable generic rate limits. Device and Tenant specific rate limits are controlled in Tenant Profile.
ip_limits_enabled: "${TB_TRANSPORT_IP_RATE_LIMITS_ENABLED:false}"
# Maximum number of connect attempts with invalid credentials
max_wrong_credentials_per_ip: "${TB_TRANSPORT_MAX_WRONG_CREDENTIALS_PER_IP:10}"
# Timeout to expire block IP addresses
ip_block_timeout: "${TB_TRANSPORT_IP_BLOCK_TIMEOUT:10000}"
ip_block_timeout: "${TB_TRANSPORT_IP_BLOCK_TIMEOUT:60000}"
# Local HTTP transport parameters
http:
enabled: "${HTTP_ENABLED:true}"

13
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
View File

@ -23,15 +23,12 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.common.transport.TransportContext;
import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor;
import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
/**
@ -94,12 +91,16 @@ public class MqttTransportContext extends TransportContext {
connectionsCounter.decrementAndGet();
}
public boolean checkAddress(InetSocketAddress address){
public boolean checkAddress(InetSocketAddress address) {
return rateLimitService.checkAddress(address);
}
public void onAuthFailed(InetSocketAddress address){
rateLimitService.onAuthFailed(address);
public void onAuthSuccess(InetSocketAddress address) {
rateLimitService.onAuthSuccess(address);
}
public void onAuthFailure(InetSocketAddress address) {
rateLimitService.onAuthFailure(address);
}
}

6
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View File

@ -20,7 +20,6 @@ import com.google.gson.JsonParseException;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.haproxy.HAProxyMessage;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
@ -823,7 +822,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
}
});
} catch (Exception e) {
context.onAuthFailed(address);
context.onAuthFailure(address);
ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_NOT_AUTHORIZED, connectMessage));
log.trace("[{}] X509 auth failure: {}", sessionId, address, e);
ctx.close();
@ -935,10 +934,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
private void onValidateDeviceResponse(ValidateDeviceCredentialsResponse msg, ChannelHandlerContext ctx, MqttConnectMessage connectMessage) {
if (!msg.hasDeviceInfo()) {
context.onAuthFailed(address);
context.onAuthFailure(address);
ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_REFUSED_NOT_AUTHORIZED, connectMessage));
ctx.close();
} else {
context.onAuthSuccess(address);
deviceSessionCtx.setDeviceInfo(msg.getDeviceInfo());
deviceSessionCtx.setDeviceProfile(msg.getDeviceProfile());
deviceSessionCtx.setSessionInfo(SessionInfoCreator.create(msg, context, sessionId));

79
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java
View File

@ -16,6 +16,7 @@
package org.thingsboard.server.common.transport.limits;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.thingsboard.server.common.data.EntityType;
@ -25,14 +26,13 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
import org.thingsboard.server.common.data.tenant.profile.TenantProfileData;
import org.thingsboard.server.common.msg.tools.TbRateLimits;
import org.thingsboard.server.common.transport.TransportTenantProfileCache;
import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult;
import org.thingsboard.server.queue.util.TbTransportComponent;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -49,9 +49,17 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi
private final ConcurrentMap<TenantId, Set<DeviceId>> tenantDevices = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, EntityTransportRateLimits> perTenantLimits = new ConcurrentHashMap<>();
private final ConcurrentMap<DeviceId, EntityTransportRateLimits> perDeviceLimits = new ConcurrentHashMap<>();
private final Map<InetAddress, InetAddressRateLimitStats> ipMap = new ConcurrentHashMap<>();
private final TransportTenantProfileCache tenantProfileCache;
@Value("${transport.rate_limits.ip_limits_enabled:false}")
private boolean ipRateLimitsEnabled;
@Value("${transport.rate_limits.max_wrong_credentials_per_ip:10}")
private int maxWrongCredentialsPerIp;
@Value("${transport.rate_limits.ip_block_timeout:60000}")
private long ipBlockTimeout;
public DefaultTransportRateLimitService(TransportTenantProfileCache tenantProfileCache) {
this.tenantProfileCache = tenantProfileCache;
}
@ -118,16 +126,73 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi
tenantAllowed.put(tenantId, allowed);
}
private Set<InetAddress> blockedAddresses = new HashSet<>();
@Override
public boolean checkAddress(InetSocketAddress address) {
return !blockedAddresses.contains(address.getAddress());
if (!ipRateLimitsEnabled) {
return true;
}
var stats = ipMap.computeIfAbsent(address.getAddress(), a -> new InetAddressRateLimitStats());
return !stats.isBlocked() || (stats.getLastActivityTs() + ipBlockTimeout < System.currentTimeMillis());
}
@Override
public void onAuthSuccess(InetSocketAddress address) {
if (!ipRateLimitsEnabled) {
return;
}
var stats = ipMap.computeIfAbsent(address.getAddress(), a -> new InetAddressRateLimitStats());
stats.getLock().lock();
try {
stats.setLastActivityTs(System.currentTimeMillis());
stats.setFailureCount(0);
if (stats.isBlocked()) {
stats.setBlocked(false);
log.info("[{}] IP address un-blocked due to correct credentials.", address.getAddress());
}
} finally {
stats.getLock().unlock();
}
}
@Override
public void onAuthFailed(InetSocketAddress address) {
blockedAddresses.add(address.getAddress());
public void onAuthFailure(InetSocketAddress address) {
if (!ipRateLimitsEnabled) {
return;
}
var stats = ipMap.computeIfAbsent(address.getAddress(), a -> new InetAddressRateLimitStats());
stats.getLock().lock();
try {
stats.setLastActivityTs(System.currentTimeMillis());
int failureCount = stats.getFailureCount() + 1;
stats.setFailureCount(failureCount);
if (failureCount >= maxWrongCredentialsPerIp) {
log.info("[{}] IP address blocked due to constantly wrong credentials.", address.getAddress());
stats.setBlocked(true);
}
} finally {
stats.getLock().unlock();
}
}
@Override
public void invalidateRateLimitsIpTable(long sessionInactivityTimeout) {
if (!ipRateLimitsEnabled) {
return;
}
long currentTime = System.currentTimeMillis();
long expTime = currentTime - Math.max(sessionInactivityTimeout, ipBlockTimeout);
for (var entry : ipMap.entrySet()) {
var stats = entry.getValue();
if (stats.getLastActivityTs() < expTime) {
log.debug("[{}] IP address removed due to session inactivity timeout.", entry.getKey());
ipMap.remove(entry.getKey());
} else if (stats.isBlocked() && (stats.getLastActivityTs() + ipBlockTimeout < currentTime)) {
log.info("[{}] IP address unblocked due ip block timeout.", entry.getKey());
stats.setBlocked(false);
}
}
}
private <T extends EntityId> void mergeLimits(T entityId, EntityTransportRateLimits newRateLimits,

33
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/InetAddressRateLimitStats.java
View File

@ -0,0 +1,33 @@
/**
* Copyright © 2016-2021 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.common.transport.limits;
import lombok.Data;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@Data
public class InetAddressRateLimitStats {
private final Lock lock = new ReentrantLock();
private boolean blocked;
private long lastActivityTs;
private int failureCount;
private int connectionsCount;
}

7
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitService.java
View File

@ -38,5 +38,10 @@ public interface TransportRateLimitService {
boolean checkAddress(InetSocketAddress address);
void onAuthFailed(InetSocketAddress address);
void onAuthSuccess(InetSocketAddress address);
void onAuthFailure(InetSocketAddress address);
void invalidateRateLimitsIpTable(long sessionInactivityTimeout);
}

5
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View File

@ -209,6 +209,7 @@ public class DefaultTransportService implements TransportService {
this.transportApiStats = statsFactory.createMessagesStats(StatsType.TRANSPORT.getName() + ".producer");
this.transportCallbackExecutor = ThingsBoardExecutors.newWorkStealingPool(20, getClass());
this.scheduler.scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS);
this.scheduler.scheduleAtFixedRate(this::invalidateRateLimits, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS);
transportApiRequestTemplate = queueProvider.createTransportApiRequestTemplate();
transportApiRequestTemplate.setMessagesStats(transportApiStats);
ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer();
@ -247,6 +248,10 @@ public class DefaultTransportService implements TransportService {
});
}
private void invalidateRateLimits() {
rateLimitService.invalidateRateLimitsIpTable(sessionInactivityTimeout);
}
@PreDestroy
public void destroy() {
stopped = true;

9
transport/mqtt/src/main/resources/tb-mqtt-transport.yml
View File

@ -149,6 +149,15 @@ transport:
stats:
enabled: "${TB_TRANSPORT_STATS_ENABLED:true}"
print-interval-ms: "${TB_TRANSPORT_STATS_PRINT_INTERVAL_MS:60000}"
client_side_rpc:
timeout: "${CLIENT_SIDE_RPC_TIMEOUT:60000}"
rate_limits:
# Enable or disable generic rate limits. Device and Tenant specific rate limits are controlled in Tenant Profile.
ip_limits_enabled: "${TB_TRANSPORT_IP_RATE_LIMITS_ENABLED:false}"
# Maximum number of connect attempts with invalid credentials
max_wrong_credentials_per_ip: "${TB_TRANSPORT_MAX_WRONG_CREDENTIALS_PER_IP:10}"
# Timeout to expire block IP addresses
ip_block_timeout: "${TB_TRANSPORT_IP_BLOCK_TIMEOUT:60000}"
queue:

Write Preview
Loading…
Cancel
Save