短信验证码接口防刷实战：Redis 限流 3 策略与 5 分钟 10 次拦截

短信验证码接口防刷实战：Redis 限流 3 策略与 5 分钟 10 次拦截

短信验证码作为现代应用中最常见的身份验证手段之一，其安全性直接关系到用户账户和资金安全。然而，随着黑产技术的不断升级，短信验证码接口正成为恶意攻击者的重点目标。本文将深入探讨基于 Redis 的三种高效限流策略，并提供可落地的工程方案，帮助开发者构建坚固的防刷体系。

1. 短信验证码接口面临的安全挑战

在电商平台的实际运营中，我们曾遭遇过一次典型的短信验证码接口攻击。攻击者利用脚本在短时间内对注册接口发起数万次请求，导致短信费用激增并引发系统告警。事后分析发现，这类攻击通常具有以下特征：

• 高频请求：单 IP 每秒发起数十次验证码请求
• 号码轮换：使用虚拟号码或临时号码进行批量注册
• 分布式攻击：通过代理池分散请求来源 IP

常见攻击类型对比表：

提示：根据行业数据，未受保护的短信接口平均每月会产生 $5000+ 的无效短信费用，而完善的防护方案可将这一数字降低至 $50 以内。

2. Redis 限流核心策略

2.1 基于 IP 的滑动窗口限流

我们首先实现最基础的 IP 限流策略，采用滑动窗口算法确保精准控制：

def check_ip_limit(ip): current_time = int(time.time()) window_size = 300 # 5分钟窗口 max_requests = 10 # 使用有序集合存储请求时间戳 key = f"sms:ip:{ip}" redis.zremrangebyscore(key, 0, current_time - window_size) request_count = redis.zcard(key) if request_count >= max_requests: return False redis.zadd(key, {current_time: current_time}) redis.expire(key, window_size) return True

关键参数说明：

• window_size：300秒（5分钟）时间窗口
• max_requests：窗口内允许的最大请求数
• zremrangebyscore：清理过期请求记录
• zadd：添加当前请求时间戳

2.2 手机号+设备指纹双因素限流

单纯依赖 IP 限制容易被绕过，我们引入设备指纹增强防护：

public boolean checkMobileLimit(String mobile, String deviceId) { String key = "sms:mobile:" + mobile; long current = System.currentTimeMillis() / 1000; long window = 300; // 5分钟 // 获取已有记录 Map<String,String> data = redis.hgetAll(key); if (data.containsKey(deviceId)) { long lastTime = Long.parseLong(data.get(deviceId)); if (current - lastTime < 60) { // 60秒冷却期 return false; } } // 更新记录 redis.hset(key, deviceId, String.valueOf(current)); redis.expire(key, window); return true; }

设备指纹生成策略：

1. 客户端收集：屏幕分辨率、OS版本、字体列表等
2. 服务端加工：通过SHA256生成唯一指纹
3. 持久化存储：关联用户行为画像

2.3 分级动态阈值策略

针对不同风险等级实施差异化限流：

风险等级判定矩阵：

func GetRiskLevel(ip, deviceID string) int { // 检查IP信誉 ipScore := redis.ZScore("ip:reputation", ip) // 检查设备历史 deviceCount := redis.HLen("device:" + deviceID) // 综合判定 switch { case ipScore > 80 || deviceCount > 5: return 0 // 低风险 case ipScore < 30: return 2 // 高风险 default: return 1 // 中风险 } }

3. 工程实现与优化

3.1 分布式限流架构

系统组件图：

1. API网关：前置流量过滤
2. Redis集群：中央限流计数器
3. 风控服务：实时规则计算
4. 日志分析：离线规则优化

# Redis集群配置示例 redis-cli --cluster create \ 192.168.1.101:6379 \ 192.168.1.102:6379 \ 192.168.1.103:6379 \ --cluster-replicas 1

3.2 性能优化技巧

1. Pipeline批量操作：

with redis.pipeline() as pipe: for ip in ip_list: pipe.zadd(f"sms:ip:{ip}", {timestamp: timestamp}) pipe.expire(f"sms:ip:{ip}", 300) pipe.execute()

2. Lua脚本原子操作：

local key = KEYS[1] local window = tonumber(ARGV[1]) local max = tonumber(ARGV[2]) local now = tonumber(ARGV[3]) redis.call('ZREMRANGEBYSCORE', key, 0, now - window) local count = redis.call('ZCARD', key) if count >= max then return 0 end redis.call('ZADD', key, now, now) redis.call('EXPIRE', key, window) return 1

3. 本地缓存降级：

@Cacheable(value = "smsLimit", key = "#mobile") public boolean checkCache(String mobile) { // 本地缓存未命中时查询Redis return redisTemplate.opsForZSet().zCard("sms:"+mobile) < 5; }

4. 进阶防护策略

4.1 验证码生命周期管理

状态机设计：

stateDiagram [*] --> 未发送 未发送 --> 已发送: 发送成功 已发送 --> 已验证: 验证通过 已发送 --> 已失效: 超时未验证 已验证 --> [*] 已失效 --> [*]

4.2 智能风控规则

动态规则引擎配置：

rules: - name: 高频IP检测 condition: ip_count > 50 within 1m action: block_ip_1h priority: 1 - name: 虚拟号检测 condition: carrier == 'virtual' action: require_captcha priority: 2 - name: 设备异常 condition: device_age < 24h && req_count > 20 action: throttle_50% priority: 3

4.3 监控与告警体系

关键监控指标：

1. 请求成功率/失败率
2. 各渠道送达延迟
3. 异常请求模式识别
4. 费用消耗趋势

-- 异常请求分析查询 SELECT ip, COUNT(*) as requests, AVG(interval) as avg_interval, COUNT(DISTINCT mobile) as unique_mobiles FROM sms_logs WHERE time > NOW() - INTERVAL '5 minutes' GROUP BY ip HAVING COUNT(*) > 100 ORDER BY requests DESC;

5. 实战：5分钟10次拦截实现

完整示例代码整合：

class SmsAntiFlood: def __init__(self, redis_conn): self.redis = redis_conn def check_request(self, ip, mobile, device_id): # 分级检查 if not self._check_ip(ip): return False if not self._check_mobile(mobile, device_id): return False return True def _check_ip(self, ip): key = f"limit:ip:{ip}" current = int(time.time()) # 滑动窗口计数 with self.redis.pipeline() as pipe: pipe.zadd(key, {current: current}) pipe.zremrangebyscore(key, 0, current - 300) pipe.zcard(key) pipe.expire(key, 300) _, _, count, _ = pipe.execute() return count <= 10 def _check_mobile(self, mobile, device_id): key = f"limit:mobile:{mobile}" current = int(time.time()) # 设备级控制 with self.redis.pipeline() as pipe: pipe.hget(key, device_id) pipe.hset(key, device_id, current) pipe.expire(key, 300) last_time, _, _ = pipe.execute() if last_time and current - int(last_time) < 60: return False return True

压力测试结果：

在实际项目中，我们通过这套方案将恶意短信请求降低了99.2%，同时保证了正常用户的顺畅体验。关键在于持续监控和调整参数，比如：

• 根据业务时段动态调整窗口大小
• 对海外IP实施更严格的策略
• 建立IP信誉库实现长效防护