Java分佈式IP限流和防止惡意IP攻擊方案
- 2020 年 3 月 29 日
- 筆記
前言
限流是分佈式系統設計中經常提到的概念，在某些要求不嚴格的場景下，使用Guava RateLimiter就可以滿足。但是Guava RateLimiter只能應用於單進程，多進程間協同控制便無能為力。本文介紹一種簡單的處理方式，用於分佈式環境下接口調用頻次管控。
如何防止惡意IP攻擊某些暴露的接口呢（比如某些場景下短訊驗證碼服務）？本文介紹一種本地緩存和分佈式緩存集成方式判斷遠程IP是否為惡意調用接口的IP。
分佈式IP限流
思路是使用redis incr命令，完成一段時間內接口請求次數的統計，以此來完成限流相關邏輯。
private static final String LIMIT_LUA = "local my_limit = redis.call('incr', KEYS[1])n" + " if tonumber(my_limit) == 1 thenn" + " redis.call('expire', KEYS[1], ARGV[1])n" + " return 1n" + " elseif tonumber(my_limit) > tonumber(ARGV[2]) thenn" + " return 0n" + " elsen" + " return 1n" + " endn";
這裡為啥時候用lua腳本來實現呢？因為要保證incr命令和expire命令的原子性操作。KEYS[1]代表自增key值， ARGV[1]代表過期時間，ARGV[2]代表最大頻次，明白了這些參數的含義，整個lua腳本邏輯也就不言而喻了。
/** * @param limitKey 限制Key值 * @param maxRate 最大速率 * @param expire Key過期時間 */ public boolean access(String limitKey, int maxRate, int expire) { if (StringUtils.isBlank(limitKey)) { return true; } String cacheKey = LIMIT_KEY_PREFIX + limitKey; return REDIS_SUCCESS_STATUS.equals( this.cacheService.eval( LIMIT_LUA , Arrays.asList(cacheKey) , Arrays.asList(String.valueOf(expire), String.valueOf(maxRate)) ).toString() ); } public void unlimit(String limitKey) { if (StringUtils.isBlank(limitKey)) { return; } String cacheKey = LIMIT_KEY_PREFIX + limitKey; this.cacheService.decr(cacheKey); }
access方法用來判斷 limitKey 是否超過了最大訪問頻次。緩存服務對象（cacheService）的eval方法參數分別是lua腳本、key list、value list。
unlimit方法其實就是執行redis decr操作，在某些業務場景可以回退訪問頻次統計。
防止惡意IP攻擊
由於某些對外暴露的接口很容易被惡意用戶攻擊，必須做好防範措施。最近我就遇到了這麼一種情況，我們一個快應用產品，短訊驗證碼服務被惡意調用了。通過後台的日誌發現，IP固定，接口調用時間間隔固定，明顯是被人利用了。雖然我們針對每個手機號每天發送短訊驗證碼的次數限制在5次以內。但是短訊驗證碼服務每天這樣被重複調用，會打擾用戶併產生投訴。針對這種現象，簡單的做了一個方案，可以自動識別惡意攻擊的IP並加入黑名單。
思路是這樣的，針對某些業務場景，約定在一段時間內同一個IP訪問最大頻次，如果超過了這個最大頻次，那麼就認為是非法IP。識別了非法IP後，把IP同時放入本地緩存和分佈式緩存中。非法IP再次訪問的時候，攔截器發現本地緩存（沒有則去分佈式緩存）有記錄這個IP，直接返回異常狀態，不會繼續執行正常業務邏輯。
Guava本地緩存集成Redis分佈式緩存
public abstract class AbstractCombineCache<K, V> { private static Logger LOGGER = LoggerFactory.getLogger(AbstractCombineCache.class); protected Cache<K, V> localCache; protected ICacheService cacheService; public AbstractCombineCache(Cache<K, V> localCache, ICacheService cacheService) { this.localCache = localCache; this.cacheService = cacheService; } public Cache<K, V> getLocalCache() { return localCache; } public ICacheService getCacheService() { return cacheService; } public V get(K key) { //只有LoadingCache對象才有get方法，如果本地緩存不存在key值， 會執行CacheLoader的load方法，從分佈式緩存中加載。 if (localCache instanceof LoadingCache) { try { return ((LoadingCache<K, V>) localCache).get(key); } catch (ExecutionException e) { LOGGER.error(String.format("cache key=%s loading error...", key), e); return null; } catch (CacheLoader.InvalidCacheLoadException e) { //分佈式緩存中不存在這個key LOGGER.error(String.format("cache key=%s loading fail...", key)); return null; } } else { return localCache.getIfPresent(key); } } public void put(K key, V value, int expire) { this.localCache.put(key, value); String cacheKey = key instanceof String ? (String) key : key.toString(); if (value instanceof String) { this.cacheService.setex(cacheKey, (String) value, expire); } else { this.cacheService.setexObject(cacheKey, value, expire); } } }
AbstractCombineCache這個抽象類封裝了guava本地緩存和redis分佈式緩存操作，可以降低分佈式緩存壓力。
防止惡意IP攻擊緩存服務
public class IPBlackCache extends AbstractCombineCache<String, Object> { private static Logger LOGGER = LoggerFactory.getLogger(IPBlackCache.class); private static final String IP_BLACK_KEY_PREFIX = "wmhipblack_"; private static final String REDIS_SUCCESS_STATUS = "1"; private static final String IP_RATE_LUA = "local ip_rate = redis.call('incr', KEYS[1])n" + " if tonumber(ip_rate) == 1 thenn" + " redis.call('expire', KEYS[1], ARGV[1])n" + " return 1n" + " elseif tonumber(ip_rate) > tonumber(ARGV[2]) thenn" + " return 0n" + " elsen" + " return 1n" + " endn"; public IPBlackCache(Cache<String, Object> localCache, ICacheService cacheService) { super(localCache, cacheService); } /** * @param ipKey IP * @param maxRate 最大速率 * @param expire 過期時間 */ public boolean ipAccess(String ipKey, int maxRate, int expire) { if (StringUtils.isBlank(ipKey)) { return true; } String cacheKey = IP_BLACK_KEY_PREFIX + ipKey; return REDIS_SUCCESS_STATUS.equals( this.cacheService.eval( IP_RATE_LUA , Arrays.asList(cacheKey) , Arrays.asList(String.valueOf(expire), String.valueOf(maxRate)) ).toString() ); } /** * @param ipKey IP */ public void removeIpAccess(String ipKey) { if (StringUtils.isBlank(ipKey)) { return; } String cacheKey = IP_BLACK_KEY_PREFIX + ipKey; try { this.cacheService.del(cacheKey); } catch (Exception e) { LOGGER.error(String.format("%s, ip access remove error...", ipKey), e); } } }
沒有錯，IP_RATE_LUA 這個lua腳本和上面說的限流方案對應的lua腳本是一樣的。
IPBlackCache繼承了AbstractCombineCache，構造函數需要guava的本地Cache對象和redis分佈式緩存服務ICacheService 對象。
ipAccess方法用來判斷當前ip訪問次數是否在一定時間內已經達到了最大訪問頻次。
removeIpAccess方法是直接移除當前ip訪問頻次統計的key值。
防止惡意IP攻擊緩存配置類
@Configuration public class IPBlackCacheConfig { private static final String IPBLACK_LOCAL_CACHE_NAME = "ip-black-cache"; private static Logger LOGGER = LoggerFactory.getLogger(IPBlackCacheConfig.class); @Autowired private LimitConstants limitConstants; @Bean public IPBlackCache ipBlackCache(@Autowired ICacheService cacheService) { GuavaCacheBuilder cacheBuilder = new GuavaCacheBuilder<String, Object>(IPBLACK_LOCAL_CACHE_NAME); cacheBuilder.setCacheBuilder( CacheBuilder.newBuilder() .initialCapacity(100) .maximumSize(10000) .concurrencyLevel(10) .expireAfterWrite(limitConstants.getIpBlackExpire(), TimeUnit.SECONDS) .removalListener((RemovalListener<String, Object>) notification -> { String curTime = LocalDateTime.now().toString(); LOGGER.info(notification.getKey() + " 本地緩存移除時間：" + curTime); try { cacheService.del(notification.getKey()); LOGGER.info(notification.getKey() + " 分佈式緩存移除時間：" + curTime); } catch (Exception e) { LOGGER.error(notification.getKey() + " 分佈式緩存移除異常...", e); } }) ); cacheBuilder.setCacheLoader(new CacheLoader<String, Object>() { @Override public Object load(String key) { try { Object obj = cacheService.getString(key); LOGGER.info(String.format("從分佈式緩存中加載key=%s, value=%s", key, obj)); return obj; } catch (Exception e) { LOGGER.error(key + " 從分佈式緩存加載異常...", e); return null; } } }); Cache<String, Object> localCache = cacheBuilder.build(); IPBlackCache ipBlackCache = new IPBlackCache(localCache, cacheService); return ipBlackCache; } }
注入redis分佈式緩存服務ICacheService對象。
通過GuavaCacheBuilder構建guava本地Cache對象，指定初始容量（initialCapacity）、最大容量（maximumSize）、並發級別、key過期時間、key移除監聽器。最終要的是CacheLoader這個參數，是幹什麼用的呢？如果GuavaCacheBuilder指定了CacheLoader對象，那麼最終創建的guava本地Cache對象是LoadingCache類型（參考AbstractCombineCache類的get方法），LoadingCache對象的get方法首先從內存中獲取key對應的value，如果內存中不存在這個key則調用CacheLoader對象的load方法加載key對應的value值，加載成功後放入內存中。
最後通過ICacheService對象和guava本地Cache對象創建IPBlackCache（防止惡意IP攻擊緩存服務）對象。
攔截器里惡意IP校驗
定義一個註解，標註在指定方法上，攔截器里會識別這個註解。
@Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) public @interface IPBlackLimit { //統計時間內最大速率 int maxRate(); //頻次統計時間 int duration(); //方法名稱 String method() default StringUtils.EMPTY; }
攔截器里加入ipAccess方法，校驗遠程IP是否為惡意攻擊的IP。
/** * @param method 需要校驗的方法 * @param remoteAddr 遠程IP */ private boolean ipAccess(Method method, String remoteAddr) { if (StringUtils.isBlank(remoteAddr) || !AnnotatedElementUtils.isAnnotated(method, IPBlackLimit.class)) { return true; } IPBlackLimit ipBlackLimit = AnnotatedElementUtils.getMergedAnnotation(method, IPBlackLimit.class); try { String ip = remoteAddr.split(",")[0].trim(); String cacheKey = "cipb_" + (StringUtils.isBlank(ipBlackLimit.method()) ? ip : String.format("%s_%s", ip, ipBlackLimit.method())); String beginAccessTime = (String) ipBlackCache.get(cacheKey); if (StringUtils.isNotBlank(beginAccessTime)) { LocalDateTime beginTime = LocalDateTime.parse(beginAccessTime, DateTimeFormatter.ISO_LOCAL_DATE_TIME), endTime = LocalDateTime.now(); Duration duration = Duration.between(beginTime, endTime); if (duration.getSeconds() >= limitConstants.getIpBlackExpire()) { ipBlackCache.getLocalCache().invalidate(cacheKey); return true; } else { return false; } } boolean access = ipBlackCache.ipAccess(cacheKey, ipBlackLimit.maxRate(), ipBlackLimit.duration()); if (!access) { ipBlackCache.removeIpAccess(cacheKey); String curTime = LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME); ipBlackCache.put(cacheKey, curTime, limitConstants.getIpBlackExpire()); } return access; } catch (Exception e) { LOGGER.error(String.format("method=%sï¼remoteAddr=%s, ip access check error.", method.getName(), remoteAddr), e); return true; } }
remoteAddr取的是X-Forwarded-For對應的值。利用remoteAddr構造cacheKey參數，通過IPBlackCache判斷cacheKey是否存在。
如果是cacheKey存在的請求，判斷黑名單IP限制是否已經到達有效期，如果已經超過有效期則清除本地緩存和分佈式緩存的cacheKey，請求合法；如果沒有超過有效期則請求非法。
否則是cacheKey不存在的請求，使用IPBlackCache對象的ipAccess方法統計一定時間內的訪問頻次，如果頻次超過最大限制，表明是非法請求IP，需要往IPBlackCache對象寫入“cacheKey=當前時間”。
總結
本文的兩種方案都使用redis incr命令，如果不是特殊業務場景，redis的key要指定過期時間，嚴格來講需要保證incr和expire兩個命令的原子性，所以使用lua腳本方式。如果沒有那麼嚴格，完全可以先setex（設置key，value，過期時間），然後再incr（註：incr不會更新key的有效期）。本文的設計方案僅供參考，並不能應用於所有的業務場景。
