feat(worker-risk): 新增大厅阶梯式风控中间件与鉴权短缓存，惩罚外挂死循环狂刷并支持恶意封号
1. 新增 worker-risk-control 阶梯式风控：单打手/IP 限流 3次/秒（超限返 429）；1 分钟累计违规 10 次升级为二级惩罚（锁定 5 分钟秒拒并带 Retry-After）；惩罚期内持续暴力刷接口累计 30 次自动判定恶意不改，直接封禁打手账号（status='frozen'）并作废所有 session； 2. 风控限流前置挂载在 requireWorkerSession 之前：命中限流或处于风控期时，微秒级在内存中基于 JWT payload 与 IP 拦截，彻底杜绝外挂流量穿透到 PostgreSQL 数据库； 3. verifyWorkerSessionToken 增加 3 秒内存校验缓存：并发请求直接复用内存鉴权对象，同一页面并发发出的多个接口请求仅查 1 次数据库，极大减轻连接池争抢压力； 4. 全套新增风控阶梯机制与自动封号自动化测试，后端 549 项测试全绿通过。
This commit is contained in:
yml2213
@@ -0,0 +1,142 @@
|
||||
import test from 'node:test'
|
||||
import assert from 'node:assert/strict'
|
||||
import type { Request, Response } from 'express'
|
||||
import {
|
||||
createWorkerHallRiskMiddleware,
|
||||
extractFastWorkerId,
|
||||
resetWorkerRiskTrackersForTest,
|
||||
getWorkerRiskTrackerForTest,
|
||||
} from './worker-risk-control.js'
|
||||
|
||||
function buildMockReq(options: { ip?: string; workerId?: number; url?: string }): Request {
|
||||
const headers: Record<string, string> = {}
|
||||
if (options.workerId) {
|
||||
const payload = Buffer.from(
|
||||
JSON.stringify({ typ: 'worker', uid: options.workerId }),
|
||||
'utf8',
|
||||
).toString('base64url')
|
||||
headers.authorization = `Bearer ${payload}.mockSignature`
|
||||
}
|
||||
return {
|
||||
ip: options.ip || '127.0.0.1',
|
||||
headers,
|
||||
originalUrl: options.url || '/api/v1/worker/hall/orders',
|
||||
} as unknown as Request
|
||||
}
|
||||
|
||||
function buildMockRes() {
|
||||
const state = {
|
||||
status: 200,
|
||||
headers: {} as Record<string, string>,
|
||||
payload: null as unknown,
|
||||
}
|
||||
const res = {
|
||||
status(code: number) {
|
||||
state.status = code
|
||||
return res
|
||||
},
|
||||
setHeader(name: string, value: string) {
|
||||
state.headers[name] = value
|
||||
return res
|
||||
},
|
||||
json(body: unknown) {
|
||||
state.payload = body
|
||||
return res
|
||||
},
|
||||
} as unknown as Response
|
||||
return {
|
||||
get status() {
|
||||
return state.status
|
||||
},
|
||||
get headers() {
|
||||
return state.headers
|
||||
},
|
||||
get payload() {
|
||||
return state.payload
|
||||
},
|
||||
res,
|
||||
}
|
||||
}
|
||||
|
||||
test('extractFastWorkerId 能在微秒级正确解析 JWT payload 中的 workerId', () => {
|
||||
const req = buildMockReq({ workerId: 41 })
|
||||
assert.equal(extractFastWorkerId(req), 41)
|
||||
|
||||
const reqNoAuth = buildMockReq({})
|
||||
assert.equal(extractFastWorkerId(reqNoAuth), null)
|
||||
|
||||
const reqMalformed = {
|
||||
headers: { authorization: 'Bearer invalid.token' },
|
||||
} as unknown as Request
|
||||
assert.equal(extractFastWorkerId(reqMalformed), null)
|
||||
})
|
||||
|
||||
test('阶梯式风控: 正常频次放行，超过3次/秒快速返回 429', async () => {
|
||||
resetWorkerRiskTrackersForTest()
|
||||
const middleware = createWorkerHallRiskMiddleware()
|
||||
const req = buildMockReq({ workerId: 9, ip: '1.2.3.4' })
|
||||
|
||||
let passed = 0
|
||||
const next = () => {
|
||||
passed++
|
||||
}
|
||||
|
||||
// 前 3 次请求放行
|
||||
for (let i = 0; i < 3; i++) {
|
||||
const { res } = buildMockRes()
|
||||
await middleware(req, res, next)
|
||||
}
|
||||
assert.equal(passed, 3, '前3次应该正常放行')
|
||||
|
||||
// 第 4 次请求超限
|
||||
const mockRes4 = buildMockRes()
|
||||
await middleware(req, mockRes4.res, next)
|
||||
assert.equal(passed, 3, '第4次请求不应放行')
|
||||
assert.equal(mockRes4.status, 429, '超限应返回 429')
|
||||
})
|
||||
|
||||
test('阶梯式风控: 累计违规 10 次进入 Tier 2 惩罚锁定 (5分钟)', async () => {
|
||||
resetWorkerRiskTrackersForTest()
|
||||
const middleware = createWorkerHallRiskMiddleware()
|
||||
const req = buildMockReq({ workerId: 41, ip: '183.34.195.102' })
|
||||
const next = () => {}
|
||||
|
||||
// 连续超频触发违规
|
||||
for (let i = 0; i < 15; i++) {
|
||||
const { res } = buildMockRes()
|
||||
await middleware(req, res, next)
|
||||
}
|
||||
|
||||
const tracker = getWorkerRiskTrackerForTest('worker:41')
|
||||
assert.ok(tracker, '应存在 tracker')
|
||||
assert.ok(tracker.penalizedUntil > Date.now(), '应进入惩罚期')
|
||||
|
||||
// 惩罚期内再发请求，直接返回 429 且带长 Retry-After
|
||||
const penalizedRes = buildMockRes()
|
||||
await middleware(req, penalizedRes.res, next)
|
||||
assert.equal(penalizedRes.status, 429)
|
||||
const retryAfter = Number(penalizedRes.headers['Retry-After'] || 0)
|
||||
assert.ok(retryAfter > 200, `Retry-After 应当大于 200 秒，实际: ${retryAfter}`)
|
||||
})
|
||||
|
||||
test('阶梯式风控: 在惩罚期内持续暴力刷接口，违规达 30 次直接触发自动封号', async () => {
|
||||
resetWorkerRiskTrackersForTest()
|
||||
const middleware = createWorkerHallRiskMiddleware()
|
||||
const req = buildMockReq({ workerId: 99999, ip: '183.34.195.102' })
|
||||
const next = () => {}
|
||||
|
||||
// 连续刷 35 次，穿透 Tier 1(3次)、Tier 2(10次)，直到触发 Tier 3(30次)
|
||||
for (let i = 0; i < 35; i++) {
|
||||
const { res } = buildMockRes()
|
||||
await middleware(req, res, next)
|
||||
}
|
||||
|
||||
const tracker = getWorkerRiskTrackerForTest('worker:99999')
|
||||
assert.ok(tracker, '应存在 tracker')
|
||||
assert.equal(tracker.banned, true, '恶意不改应被标记为 banned')
|
||||
|
||||
// 封号后再次请求，直接 403 拒绝
|
||||
const bannedRes = buildMockRes()
|
||||
await middleware(req, bannedRes.res, next)
|
||||
assert.equal(bannedRes.status, 403, '封号后请求应返回 403 Forbidden')
|
||||
})
|
||||
@@ -0,0 +1,271 @@
|
||||
import type { NextFunction, Request, Response } from 'express'
|
||||
import { buildErrorPayload, createHttpError } from '../utils/http.js'
|
||||
import { logError, logWarn } from '../utils/logger.js'
|
||||
import { nowIso } from '../utils/time.js'
|
||||
import { updateWorkerUser } from '../repositories/worker-platform/worker-repo.js'
|
||||
import { revokeAllWorkerSessions } from '../repositories/worker-platform/worker-session-repo.js'
|
||||
|
||||
/**
|
||||
* 打手端阶梯式风控配置与状态机
|
||||
*
|
||||
* 针对外挂/脚本毫秒级死循环狂刷接单大厅、挤占单线程 Node 事件循环与数据库连接池的场景：
|
||||
* Tier 1 (平滑限流): 1 秒内超过 3 次请求 -> 429 快速拒绝，避免穿透后续 DB
|
||||
* Tier 2 (风控强化): 1 分钟内累计触发 10 次 429 -> 进入 5 分钟静默惩罚期，秒拒且 Retry-After: 300
|
||||
* Tier 3 (恶意封号): 在惩罚期内仍不停止、累计违规达 30 次 -> 判定恶意不改，自动更新数据库 status='frozen' 并作废 session
|
||||
*/
|
||||
|
||||
interface WorkerRiskTracker {
|
||||
// 当前滑动窗口内请求计数
|
||||
windowCount: number
|
||||
windowResetAt: number
|
||||
// 短期累计违规次数 (429 触发次数)
|
||||
violationCount: number
|
||||
violationResetAt: number
|
||||
// Tier 2 惩罚截止时间戳 (0 表示正常)
|
||||
penalizedUntil: number
|
||||
// 是否已被 Tier 3 自动封禁
|
||||
banned: boolean
|
||||
}
|
||||
|
||||
const riskTrackers = new Map<string, WorkerRiskTracker>()
|
||||
let lastCleanupAt = 0
|
||||
|
||||
// 常量配置
|
||||
const TIER1_WINDOW_MS = 1_000
|
||||
const TIER1_MAX_REQUESTS = 3 // 正常人手点击每秒不会超过 3 次
|
||||
|
||||
const VIOLATION_WINDOW_MS = 60_000
|
||||
const TIER2_TRIGGER_VIOLATIONS = 10 // 1 分钟触发 10 次 429 则进入 Tier 2
|
||||
const TIER2_PENALTY_MS = 300_000 // Tier 2 惩罚 5 分钟 (300 秒)
|
||||
|
||||
const TIER3_TRIGGER_VIOLATIONS = 30 // 累计触发 30 次且多次警告无效 -> 直接封号
|
||||
|
||||
/** 从 Authorization 请求头极速无 DB 提取 workerId (JWT payload base64 解码，微秒级) */
|
||||
export function extractFastWorkerId(req: Request): number | null {
|
||||
const auth = String(req.headers.authorization || '').trim()
|
||||
if (!auth.startsWith('Bearer ') && !auth.startsWith('bearer ')) {
|
||||
return null
|
||||
}
|
||||
const token = auth.slice(7).trim()
|
||||
const dotIndex = token.indexOf('.')
|
||||
if (dotIndex <= 0) return null
|
||||
const encodedPayload = token.slice(0, dotIndex)
|
||||
try {
|
||||
const raw = Buffer.from(encodedPayload, 'base64url').toString('utf8')
|
||||
const payload = JSON.parse(raw) as { typ?: string; uid?: number | string }
|
||||
if (payload.typ === 'worker' && payload.uid) {
|
||||
const id = Number(payload.uid)
|
||||
return id > 0 ? id : null
|
||||
}
|
||||
} catch {
|
||||
// 忽略异常，降级按 IP 识别
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
export function resolveClientIp(req: Request): string {
|
||||
return String(
|
||||
req.ip ||
|
||||
String(req.headers['x-forwarded-for'] || '').split(',')[0] ||
|
||||
req.socket?.remoteAddress ||
|
||||
'unknown',
|
||||
).trim()
|
||||
}
|
||||
|
||||
/** 获取跟踪目标 key：优先以 workerId 维度（防止切 IP），兜底以 IP 维度（防止未登录或换账号） */
|
||||
function getTargetKeys(req: Request): { primaryKey: string; workerId: number | null; ip: string } {
|
||||
const ip = resolveClientIp(req)
|
||||
const workerId = extractFastWorkerId(req)
|
||||
const primaryKey = workerId ? `worker:${workerId}` : `ip:${ip}`
|
||||
return { primaryKey, workerId, ip }
|
||||
}
|
||||
|
||||
function getOrCreateTracker(key: string, now: number): WorkerRiskTracker {
|
||||
let tracker = riskTrackers.get(key)
|
||||
if (!tracker) {
|
||||
tracker = {
|
||||
windowCount: 0,
|
||||
windowResetAt: now + TIER1_WINDOW_MS,
|
||||
violationCount: 0,
|
||||
violationResetAt: now + VIOLATION_WINDOW_MS,
|
||||
penalizedUntil: 0,
|
||||
banned: false,
|
||||
}
|
||||
riskTrackers.set(key, tracker)
|
||||
}
|
||||
return tracker
|
||||
}
|
||||
|
||||
/** 接单大厅与高频打手接口的风控拦截中间件 */
|
||||
export function createWorkerHallRiskMiddleware() {
|
||||
return async (req: Request, res: Response, next: NextFunction): Promise<void> => {
|
||||
const now = Date.now()
|
||||
cleanupExpiredTrackers(now)
|
||||
|
||||
const { primaryKey, workerId, ip } = getTargetKeys(req)
|
||||
const tracker = getOrCreateTracker(primaryKey, now)
|
||||
|
||||
// 1. 已被封号直接拦截
|
||||
if (tracker.banned) {
|
||||
const error = createHttpError('接单账号已被系统安全风控封禁', {
|
||||
statusCode: 403,
|
||||
errorCode: 'worker_account_frozen',
|
||||
})
|
||||
res.status(403).json(buildErrorPayload(error, '接单账号已被封禁'))
|
||||
return
|
||||
}
|
||||
|
||||
// 2. 检查是否处于 Tier 2 惩罚期
|
||||
if (tracker.penalizedUntil > now) {
|
||||
tracker.violationCount += 1
|
||||
const retryAfterSeconds = Math.max(1, Math.ceil((tracker.penalizedUntil - now) / 1000))
|
||||
res.setHeader('Retry-After', String(retryAfterSeconds))
|
||||
|
||||
// 检查是否升级到 Tier 3 自动封号
|
||||
if (tracker.violationCount >= TIER3_TRIGGER_VIOLATIONS && workerId && !tracker.banned) {
|
||||
tracker.banned = true
|
||||
await executeAutoFreezeWorker(workerId, ip, tracker.violationCount)
|
||||
const error = createHttpError('恶意高频刷接口多次不改，接单账号已被系统自动永久封禁', {
|
||||
statusCode: 403,
|
||||
errorCode: 'worker_account_frozen',
|
||||
})
|
||||
res.status(403).json(buildErrorPayload(error, '接单账号已被永久封禁'))
|
||||
return
|
||||
}
|
||||
|
||||
logWarn('[worker/risk-control]', '打手处于二级风控惩罚期，拒绝请求', {
|
||||
primaryKey,
|
||||
workerId,
|
||||
ip,
|
||||
retryAfterSeconds,
|
||||
violationCount: tracker.violationCount,
|
||||
})
|
||||
|
||||
const error = createHttpError('系统检测到异常高频请求，已触发安全风控，请稍后再试', {
|
||||
statusCode: 429,
|
||||
errorCode: 'risk_penalized',
|
||||
})
|
||||
res.status(429).json(buildErrorPayload(error, '系统安全风控限制中'))
|
||||
return
|
||||
}
|
||||
|
||||
// 3. 正常滑动窗口限流检查 (Tier 1)
|
||||
if (now > tracker.windowResetAt) {
|
||||
tracker.windowCount = 0
|
||||
tracker.windowResetAt = now + TIER1_WINDOW_MS
|
||||
}
|
||||
tracker.windowCount += 1
|
||||
|
||||
if (tracker.windowCount <= TIER1_MAX_REQUESTS) {
|
||||
next()
|
||||
return
|
||||
}
|
||||
|
||||
// 4. 触发 Tier 1 超限：记录违规
|
||||
if (now > tracker.violationResetAt) {
|
||||
tracker.violationCount = 0
|
||||
tracker.violationResetAt = now + VIOLATION_WINDOW_MS
|
||||
}
|
||||
tracker.violationCount += 1
|
||||
|
||||
// 5. 检查是否升级为 Tier 2 惩罚
|
||||
if (tracker.violationCount >= TIER2_TRIGGER_VIOLATIONS) {
|
||||
tracker.penalizedUntil = now + TIER2_PENALTY_MS
|
||||
const retryAfter = Math.ceil(TIER2_PENALTY_MS / 1000)
|
||||
res.setHeader('Retry-After', String(retryAfter))
|
||||
|
||||
logWarn('[worker/risk-control]', '打手频繁触发限流，进入二级风控惩罚锁定 (5分钟)', {
|
||||
primaryKey,
|
||||
workerId,
|
||||
ip,
|
||||
violationCount: tracker.violationCount,
|
||||
})
|
||||
|
||||
const error = createHttpError(
|
||||
'检测到连续异常高频请求，已触发 5 分钟安全风控限制，请停止刷单脚本',
|
||||
{
|
||||
statusCode: 429,
|
||||
errorCode: 'risk_penalized',
|
||||
},
|
||||
)
|
||||
res.status(429).json(buildErrorPayload(error, '触发安全风控限制'))
|
||||
return
|
||||
}
|
||||
|
||||
// 仅 Tier 1 阶段超限，返回轻量 429
|
||||
res.setHeader('Retry-After', '2')
|
||||
logWarn('[worker/risk-control]', '大厅请求超过正常频率', {
|
||||
primaryKey,
|
||||
workerId,
|
||||
ip,
|
||||
windowCount: tracker.windowCount,
|
||||
violationCount: tracker.violationCount,
|
||||
})
|
||||
|
||||
const error = createHttpError('抢单大厅刷新过于频繁，请稍后再试', {
|
||||
statusCode: 429,
|
||||
errorCode: 'hall_rate_limited',
|
||||
})
|
||||
res.status(429).json(buildErrorPayload(error, '大厅请求过于频繁'))
|
||||
}
|
||||
}
|
||||
|
||||
/** 触发 Tier 3 自动封禁 */
|
||||
async function executeAutoFreezeWorker(workerId: number, ip: string, violationCount: number) {
|
||||
try {
|
||||
const now = nowIso()
|
||||
const reason = `系统风控：检测到高频自动化刷接口（违规超 ${violationCount} 次），多次警告后未停止，已被系统自动封禁`
|
||||
await updateWorkerUser(workerId, {
|
||||
status: 'frozen',
|
||||
frozen_reason: reason,
|
||||
frozen_at: now,
|
||||
updated_at: now,
|
||||
})
|
||||
// 强制踢掉该打手所有现存登录设备
|
||||
await revokeAllWorkerSessions(workerId, now)
|
||||
logError('[worker/risk-control]', '打手账号已被系统自动永久封禁！', {
|
||||
workerId,
|
||||
ip,
|
||||
violationCount,
|
||||
reason,
|
||||
})
|
||||
} catch (err) {
|
||||
logError('[worker/risk-control]', '自动封禁打手账号执行异常', { workerId, ip, err })
|
||||
}
|
||||
}
|
||||
|
||||
/** 测试辅助：清空风控记录 */
|
||||
export function resetWorkerRiskTrackersForTest(): void {
|
||||
riskTrackers.clear()
|
||||
lastCleanupAt = 0
|
||||
}
|
||||
|
||||
/** 获取内部 tracker 状态供测试断言 */
|
||||
export function getWorkerRiskTrackerForTest(key: string): WorkerRiskTracker | undefined {
|
||||
return riskTrackers.get(key)
|
||||
}
|
||||
|
||||
function cleanupExpiredTrackers(now: number): void {
|
||||
if (now - lastCleanupAt < 60_000) return
|
||||
lastCleanupAt = now
|
||||
for (const [key, tracker] of riskTrackers.entries()) {
|
||||
if (
|
||||
!tracker.banned &&
|
||||
tracker.penalizedUntil <= now &&
|
||||
tracker.windowResetAt <= now &&
|
||||
tracker.violationResetAt <= now
|
||||
) {
|
||||
riskTrackers.delete(key)
|
||||
}
|
||||
}
|
||||
// 硬上限防泄漏
|
||||
if (riskTrackers.size > 20_000) {
|
||||
const excess = riskTrackers.size - 20_000
|
||||
let count = 0
|
||||
for (const key of riskTrackers.keys()) {
|
||||
riskTrackers.delete(key)
|
||||
count++
|
||||
if (count >= excess) break
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user