feat(worker-risk): 风控后台页面(可调阈值/实时状态/事件留痕/人工封禁解封),并修复解封后仍被 403 的 bug

## 修复的 bug(生产已发生)

风控三级封号把 tracker.banned 置为 true 后,banned 检查发生在鉴权之前,
而它只存在于进程内存。后台把账号改回 active 时数据库变了、内存没变,
于是账号在库里已经正常、请求却永远 403,只能等容器重启才恢复。

修法两道:
- **数据库是唯一事实来源**:banned 改为带 30 秒 TTL 的缓存,到点回查
  worker_users.status;已解封(或账号已删)则清除标记并放行。
  对被判 banned 的 key 最多每 30 秒查一次,即便对方仍以 65 QPS 刷也只是
  每 30 秒一次查询。查询失败时**保守维持封禁**,不因数据库抖动放行刷子。
- **后台改状态时主动释放**:review 只允许设 active/rejected/disabled/pending_review
  (没有 frozen),所以走到那里就意味着不再由风控冻结,直接释放内存状态。

顺带修掉一个连带缺陷:解封时只清了 violationCount,没清滑动窗口计数,
账号刚恢复就会被上一轮残留的 windowCount 立刻限流、并在 60 秒内二次封禁。
现在整轮窗口状态一并清零。

## 新增后台页面

接单平台 → 风控(/admin/worker-platform?tab=risk-control),三块:
- 阈值配置:6 个阈值可视化编辑,落 app_config_entries,保存即热生效,无需重启
- 当前实时风控:进程内存状态(已封禁/惩罚中/仅记录),支持人工解封、封禁、仅释放内存状态
- 风控事件留痕:worker_risk_events 分页查询,可按类型/IP 筛选

## 关键取舍

只记**显著事件**(二级惩罚、三级封号、人工处置),不记每一次被拒请求。
被刷时实测 45~65 QPS,按请求写库正是最初要解决的问题。实时计数仍走进程内存。
事件写入为 fire-and-forget 并吞异常,留痕失败绝不影响大厅请求。

## 验证

- 新增 6 个用例覆盖解封立即生效、对账自愈、TTL 不重复查库、对账失败保守维持、阈值可配置
- 迁移 089 已在临时库实测:表结构/4 索引/CHECK 约束/外键均正确,非法 event_type 被拒
- 后端 557 测试 554 通过 0 失败;前后端 typecheck/lint/format/build 全绿
This commit is contained in:
yml2213
2026-10-02 17:05:25 +08:00
parent adaf4ed240
commit 78c676d2b9
15 changed files with 1706 additions and 29 deletions
@@ -1,9 +1,16 @@
import test from 'node:test'
import assert from 'node:assert/strict'
import type { Request, Response } from 'express'
import { APP_CONFIG_KEYS } from '../config/app-config-keys.js'
import { setAppConfigCacheEntry } from '../services/config/app-config-store.js'
import {
clearWorkerRiskConfigMemoForTest,
createDefaultWorkerRiskConfig,
} from '../services/worker-platform/worker-risk-config-service.js'
import {
createWorkerHallRiskMiddleware,
extractFastWorkerId,
releaseWorkerRiskTracker,
resetWorkerRiskTrackersForTest,
getWorkerRiskTrackerForTest,
} from './worker-risk-control.js'
@@ -140,3 +147,139 @@ test('阶梯式风控: 在惩罚期内持续暴力刷接口,违规达 30 次
await middleware(req, bannedRes.res, next)
assert.equal(bannedRes.status, 403, '封号后请求应返回 403 Forbidden')
})
/**
* 以下三个用例覆盖同一个生产 bug:
* 后台把被风控冻结的账号改回 active 之后,内存里的 tracker.banned 仍是 true,
* 而 banned 检查发生在鉴权之前 —— 数据库里账号已经正常,人却永远接不了单,
* 只能等容器重启。修复有两道:后台解封时主动释放 + 与数据库定期对账自愈。
*/
async function driveToBanned(workerId: number, ip: string) {
resetWorkerRiskTrackersForTest()
const middleware = createWorkerHallRiskMiddleware()
const req = buildMockReq({ workerId, ip })
for (let i = 0; i < 35; i++) {
const { res } = buildMockRes()
await middleware(req, res, () => {})
}
const tracker = getWorkerRiskTrackerForTest(`worker:${workerId}`)
assert.equal(tracker?.banned, true, '前置条件:应已被判定封禁')
return req
}
test('风控修复: 后台解封后必须能立即接单(内存标记需一并释放)', async () => {
const req = await driveToBanned(88801, '10.0.0.1')
// 后台把账号改回 active 时会调用它
releaseWorkerRiskTracker({ workerId: 88801, ip: '10.0.0.1' })
assert.equal(getWorkerRiskTrackerForTest('worker:88801'), undefined, '内存标记应被清除')
let passed = 0
const middleware = createWorkerHallRiskMiddleware()
const mockRes = buildMockRes()
await middleware(req, mockRes.res, () => {
passed++
})
assert.equal(passed, 1, '解封后请求应放行,而不是继续 403')
assert.equal(mockRes.status, 200)
})
test('风控修复: 数据库已解封时,内存封禁标记应自动失效(对账自愈)', async () => {
const req = await driveToBanned(88802, '10.0.0.2')
// TTL 内不重复查库:driveToBanned 期间已完成一次对账,这里先确认它的短路行为
const tracker = getWorkerRiskTrackerForTest('worker:88802')
assert.ok(tracker, '应存在 tracker')
assert.ok(tracker.bannedVerifiedAt > 0, '应已记录过一次对账时间')
// 模拟「对账间隔已过」:真实环境等 30 秒即可,测试里直接把时间戳归零
tracker.bannedVerifiedAt = 0
// 模拟「数据库里已经是正常账号」:不调用 release,只让对账查到 active
let passed = 0
const middleware = createWorkerHallRiskMiddleware({
lookupWorker: async () => ({ status: 'active' }),
})
const mockRes = buildMockRes()
await middleware(req, mockRes.res, () => {
passed++
})
assert.equal(passed, 1, '对账发现已解封后应放行')
assert.equal(getWorkerRiskTrackerForTest('worker:88802')?.banned, false, '封禁标记应被清掉')
})
test('风控修复: 对账间隔内的重复请求不重复查库(TTL 生效)', async () => {
const req = await driveToBanned(88805, '10.0.0.5')
const tracker = getWorkerRiskTrackerForTest('worker:88805')
assert.ok(tracker, '应存在 tracker')
// 归零以模拟「对账间隔已过」,让第一次请求真正触发查库
tracker.bannedVerifiedAt = 0
let lookupCalls = 0
const middleware = createWorkerHallRiskMiddleware({
lookupWorker: async () => {
lookupCalls++
return { status: 'frozen' }
},
})
for (let i = 0; i < 5; i++) {
const mockRes = buildMockRes()
await middleware(req, mockRes.res, () => {})
}
assert.equal(lookupCalls, 1, '同一 key 在 30 秒窗口内只应对账一次,避免被刷时打爆数据库')
})
test('风控修复: 对账查询失败时必须保守维持封禁(不能因数据库抖动放行刷子)', async () => {
const req = await driveToBanned(88803, '10.0.0.3')
let passed = 0
const middleware = createWorkerHallRiskMiddleware({
lookupWorker: async () => {
throw new Error('db down')
},
})
const mockRes = buildMockRes()
await middleware(req, mockRes.res, () => {
passed++
})
assert.equal(passed, 0, '对账失败不得放行')
assert.equal(mockRes.status, 403, '对账失败应维持 403')
})
test('风控阈值可后台配置: tier1MaxRequests 生效且保存后无需重启', async () => {
resetWorkerRiskTrackersForTest()
try {
// 直接写进程内配置缓存,等价于后台保存后的状态
setAppConfigCacheEntry(APP_CONFIG_KEYS.workerRiskControl, {
...createDefaultWorkerRiskConfig(),
tier1MaxRequests: 1,
})
clearWorkerRiskConfigMemoForTest()
const middleware = createWorkerHallRiskMiddleware()
const req = buildMockReq({ workerId: 88804, ip: '10.0.0.4' })
let passed = 0
const first = buildMockRes()
await middleware(req, first.res, () => {
passed++
})
const second = buildMockRes()
await middleware(req, second.res, () => {
passed++
})
assert.equal(passed, 1, '阈值为 1 时第二次请求应被拒')
assert.equal(second.status, 429)
} finally {
// 还原默认值,避免污染其他用例
setAppConfigCacheEntry(APP_CONFIG_KEYS.workerRiskControl, createDefaultWorkerRiskConfig())
clearWorkerRiskConfigMemoForTest()
}
})
@@ -2,16 +2,24 @@ 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 { getWorkerUserById, updateWorkerUser } from '../repositories/worker-platform/worker-repo.js'
import { revokeAllWorkerSessions } from '../repositories/worker-platform/worker-session-repo.js'
import { insertWorkerRiskEvent } from '../repositories/worker-platform/worker-risk-repo.js'
import { getWorkerRiskConfig } from '../services/worker-platform/worker-risk-config-service.js'
/**
* 打手端阶梯式风控配置与状态机
*
* 针对外挂/脚本毫秒级死循环狂刷接单大厅、挤占单线程 Node 事件循环与数据库连接池的场景:
* Tier 1 (平滑限流): 1 秒内超过 3 次请求 -> 429 快速拒绝,避免穿透后续 DB
* Tier 2 (风控强化): 1 分钟内累计触发 10 次 429 -> 进入 5 分钟静默惩罚期,秒拒且 Retry-After: 300
* Tier 3 (恶意封号): 在惩罚期内仍不停止、累计违规达 30 次 -> 判定恶意不改,自动更新数据库 status='frozen' 并作废 session
* Tier 1 (平滑限流): 窗口内超过 N 次请求 -> 429 快速拒绝,避免穿透后续 DB
* Tier 2 (风控强化): 违规窗口内累计触发 M 次 -> 进入静默惩罚期,秒拒且带 Retry-After
* Tier 3 (恶意封号): 在惩罚期内仍不停止、累计违规达 K 次 -> 自动封禁账号并作废 session
*
* 阈值全部来自 worker-risk-config-service(后台可调,保存即热生效),不再硬编码。
*
* 状态仍然是**进程内存**(重启清零),这是刻意的:按请求更新内存计数是纳秒级操作,
* 落库会瞬间打爆数据库。内存状态通过 listActiveWorkerRiskTrackers() 暴露给后台实时展示;
* 只有「显著事件」(进入二级惩罚、三级封号、人工处置)才异步写 worker_risk_events 留痕。
*/
interface WorkerRiskTracker {
@@ -25,20 +33,40 @@ interface WorkerRiskTracker {
penalizedUntil: number
// 是否已被 Tier 3 自动封禁
banned: boolean
// 上次与数据库对账「是否仍处于 frozen」的时间戳,0 表示尚未对账
bannedVerifiedAt: number
// 便于后台展示的追溯信息(最近一次见到的身份/IP 与时间)
lastWorkerId: number | null
lastIp: string
lastSeenAt: number
// 累计被拒请求数(仅用于展示,不落库)
rejectedCount: number
}
const riskTrackers = new Map<string, WorkerRiskTracker>()
let lastCleanupAt = 0
// 常量配置
const TIER1_WINDOW_MS = 1_000
const TIER1_MAX_REQUESTS = 3 // 正常人手点击每秒不会超过 3 次
/**
* 内存封禁标记与数据库的对账间隔。
*
* 为什么必须有:`banned` 只是数据库 `worker_users.status === 'frozen'` 的缓存。
* 若把它当成永久事实,后台把账号改回 active 后,中间件仍会在鉴权之前直接 403
* ——数据库里已经是正常账号,人却永远接不了单,只能等容器重启。
* 因此这里按 TTL 回查数据库;数据库是唯一事实来源。
*
* 代价可控:只有被判定为 banned 的 key 才会走到这里,且每个 key 最多每 30 秒查一次。
* 即便该账号正以 65 QPS 持续刷接口,对数据库也只是每 30 秒一次查询。
*/
const BAN_RECHECK_INTERVAL_MS = 30_000
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 秒)
/**
* 查账号状态的函数签名。抽成类型 + 可注入,是为了让「数据库已解封、内存仍封禁」
* 这条修复路径能被测试覆盖(测试环境没有 Postgres)。生产不传即用真实仓储。
*/
export type WorkerRiskWorkerLookup = (workerId: number) => Promise<{ status?: string } | null>
const TIER3_TRIGGER_VIOLATIONS = 30 // 累计触发 30 次且多次警告无效 -> 直接封号
/** 后台展示用的 tracker 快照上限,避免极端情况下返回超大数组 */
const TRACKER_SNAPSHOT_LIMIT = 200
/** 从 Authorization 请求头极速无 DB 提取 workerId (JWT payload base64 解码,微秒级) */
export function extractFastWorkerId(req: Request): number | null {
@@ -83,46 +111,123 @@ function getTargetKeys(req: Request): { primaryKey: string; workerId: number | n
function getOrCreateTracker(key: string, now: number): WorkerRiskTracker {
let tracker = riskTrackers.get(key)
if (!tracker) {
const config = getWorkerRiskConfig()
tracker = {
windowCount: 0,
windowResetAt: now + TIER1_WINDOW_MS,
windowResetAt: now + config.tier1WindowMs,
violationCount: 0,
violationResetAt: now + VIOLATION_WINDOW_MS,
violationResetAt: now + config.violationWindowMs,
penalizedUntil: 0,
banned: false,
bannedVerifiedAt: 0,
lastWorkerId: null,
lastIp: '',
lastSeenAt: now,
rejectedCount: 0,
}
riskTrackers.set(key, tracker)
}
return tracker
}
/**
* 判断仍被封禁的 key 是否真的还该被封。
*
* 返回 true 表示维持封禁;返回 false 表示数据库侧已解封(或账号已不存在),
* 调用方应清掉内存标记并放行。
*
* 查询失败时**保守维持封禁**——风控宁可误挡,也不能因为一次数据库抖动把刷子放进来。
*/
async function isWorkerStillFrozen(
tracker: WorkerRiskTracker,
now: number,
lookupWorker: WorkerRiskWorkerLookup,
): Promise<boolean> {
if (tracker.bannedVerifiedAt && now - tracker.bannedVerifiedAt < BAN_RECHECK_INTERVAL_MS) {
return true
}
tracker.bannedVerifiedAt = now
// banned 只会由 Tier 3 设置,而 Tier 3 要求能解出 workerId,因此这里必然有 workerId;
// 兜底返回 true 以免出现无法校验却放行的情况。
const workerId = tracker.lastWorkerId
if (!workerId) return true
try {
const worker = await lookupWorker(workerId)
if (!worker) {
// 账号已被删除,封禁自然失效
tracker.banned = false
return false
}
if (worker.status === 'frozen') return true
// 解封后必须把整轮窗口状态一起清零,只清 banned 是不够的:
// 上一轮刷出来的 windowCount / violationCount 仍在窗口内,账号刚恢复就会立刻
// 被 Tier 1 限流、并在 60 秒内再次凑满 Tier 2 触发条件而二次封禁。
tracker.banned = false
tracker.penalizedUntil = 0
tracker.violationCount = 0
tracker.violationResetAt = 0
tracker.windowCount = 0
tracker.windowResetAt = 0
tracker.rejectedCount = 0
return false
} catch (err) {
logError('[worker/risk-control]', '封禁状态对账失败,暂时维持封禁', { workerId, err })
return true
}
}
/** 接单大厅与高频打手接口的风控拦截中间件 */
export function createWorkerHallRiskMiddleware() {
export function createWorkerHallRiskMiddleware(
deps: { lookupWorker?: WorkerRiskWorkerLookup } = {},
) {
const lookupWorker: WorkerRiskWorkerLookup = deps.lookupWorker ?? getWorkerUserById
return async (req: Request, res: Response, next: NextFunction): Promise<void> => {
const config = getWorkerRiskConfig()
if (!config.enabled) {
next()
return
}
const now = Date.now()
cleanupExpiredTrackers(now)
const { primaryKey, workerId, ip } = getTargetKeys(req)
const tracker = getOrCreateTracker(primaryKey, now)
if (workerId) tracker.lastWorkerId = workerId
if (ip) tracker.lastIp = ip
tracker.lastSeenAt = now
// 1. 已被封号直接拦截
// 1. 被封号则拦截。但必须先与数据库对账:`banned` 只是数据库 status='frozen' 的缓存,
// 后台解封(或直接改库)后内存标记若不失效,账号在数据库里已是 active、
// 人却仍然接不了单——生产上已实际发生,只能等容器重启才恢复。
if (tracker.banned) {
const error = createHttpError('接单账号已被系统安全风控封禁', {
statusCode: 403,
errorCode: 'worker_account_frozen',
if (await isWorkerStillFrozen(tracker, now, lookupWorker)) {
tracker.rejectedCount += 1
const error = createHttpError('接单账号已被系统安全风控封禁', {
statusCode: 403,
errorCode: 'worker_account_frozen',
})
res.status(403).json(buildErrorPayload(error, '接单账号已被封禁'))
return
}
logWarn('[worker/risk-control]', '封禁标记与数据库不一致,已自动解除并放行', {
primaryKey,
workerId,
})
res.status(403).json(buildErrorPayload(error, '接单账号已被封禁'))
return
}
// 2. 检查是否处于 Tier 2 惩罚期
if (tracker.penalizedUntil > now) {
tracker.violationCount += 1
tracker.rejectedCount += 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) {
if (tracker.violationCount >= config.tier3TriggerViolations && workerId && !tracker.banned) {
tracker.banned = true
await executeAutoFreezeWorker(workerId, ip, tracker.violationCount)
const error = createHttpError('恶意高频刷接口多次不改,接单账号已被系统自动永久封禁', {
@@ -152,37 +257,53 @@ export function createWorkerHallRiskMiddleware() {
// 3. 正常滑动窗口限流检查 (Tier 1)
if (now > tracker.windowResetAt) {
tracker.windowCount = 0
tracker.windowResetAt = now + TIER1_WINDOW_MS
tracker.windowResetAt = now + config.tier1WindowMs
}
tracker.windowCount += 1
if (tracker.windowCount <= TIER1_MAX_REQUESTS) {
if (tracker.windowCount <= config.tier1MaxRequests) {
next()
return
}
// 4. 触发 Tier 1 超限:记录违规
tracker.rejectedCount += 1
if (now > tracker.violationResetAt) {
tracker.violationCount = 0
tracker.violationResetAt = now + VIOLATION_WINDOW_MS
tracker.violationResetAt = now + config.violationWindowMs
}
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)
if (tracker.violationCount >= config.tier2TriggerViolations) {
tracker.penalizedUntil = now + config.tier2PenaltyMs
const retryAfter = Math.ceil(config.tier2PenaltyMs / 1000)
res.setHeader('Retry-After', String(retryAfter))
logWarn('[worker/risk-control]', '打手频繁触发限流,进入二级风控惩罚锁定 (5分钟)', {
const penaltyMinutes = Math.round(config.tier2PenaltyMs / 60_000)
logWarn('[worker/risk-control]', '打手频繁触发限流,进入二级风控惩罚锁定', {
primaryKey,
workerId,
ip,
violationCount: tracker.violationCount,
penaltyMinutes,
})
// 显著事件留痕(低频;被拒请求本身不落库)
recordWorkerRiskEvent({
eventType: 'tier2_penalized',
targetKey: primaryKey,
workerId,
ip,
violationCount: tracker.violationCount,
windowCount: tracker.windowCount,
penaltyUntil: new Date(tracker.penalizedUntil).toISOString(),
reason: `违规 ${tracker.violationCount} 次,进入 ${penaltyMinutes} 分钟静默惩罚期`,
detail: { penaltyMinutes, config },
})
const error = createHttpError(
'检测到连续异常高频请求,已触发 5 分钟安全风控限制,请停止刷单脚本',
`检测到连续异常高频请求,已触发 ${penaltyMinutes} 分钟安全风控限制,请停止刷单脚本`,
{
statusCode: 429,
errorCode: 'risk_penalized',
@@ -229,11 +350,105 @@ async function executeAutoFreezeWorker(workerId: number, ip: string, violationCo
violationCount,
reason,
})
recordWorkerRiskEvent({
eventType: 'tier3_autofrozen',
targetKey: `worker:${workerId}`,
workerId,
ip,
violationCount,
reason,
})
} catch (err) {
logError('[worker/risk-control]', '自动封禁打手账号执行异常', { workerId, ip, err })
}
}
/**
* 风控事件留痕。**fire-and-forget**:留痕失败绝不能影响请求结果,
* 更不允许因为一次数据库抖动把大厅请求变成 500。
*/
function recordWorkerRiskEvent(input: Parameters<typeof insertWorkerRiskEvent>[0]): void {
void insertWorkerRiskEvent(input).catch((err) => {
logError('[worker/risk-control]', '风控事件留痕失败', { err, eventType: input.eventType })
})
}
export type WorkerRiskTrackerSnapshot = {
key: string
workerId: number | null
ip: string
banned: boolean
penalizedUntil: number
violationCount: number
windowCount: number
rejectedCount: number
lastSeenAt: number
}
/**
* 后台展示用:当前仍在生效(封禁 / 惩罚中 / 本轮有违规)的内存风控项。
* 这些数据不落库,进程重启即清零——后台需如实标注,避免被误读为历史累计。
*/
export function listActiveWorkerRiskTrackers(): WorkerRiskTrackerSnapshot[] {
const now = Date.now()
const items: WorkerRiskTrackerSnapshot[] = []
for (const [key, tracker] of riskTrackers.entries()) {
if (!tracker.banned && tracker.penalizedUntil <= now && tracker.violationCount <= 0) {
continue
}
items.push({
key,
workerId: tracker.lastWorkerId,
ip: tracker.lastIp,
banned: tracker.banned,
penalizedUntil: tracker.penalizedUntil,
violationCount: tracker.violationCount,
windowCount: tracker.windowCount,
rejectedCount: tracker.rejectedCount,
lastSeenAt: tracker.lastSeenAt,
})
}
return items.sort((a, b) => b.lastSeenAt - a.lastSeenAt).slice(0, TRACKER_SNAPSHOT_LIMIT)
}
/**
* 解除某个打手(及可选 IP)的内存风控状态。
*
* 后台解封账号时**必须**连同调用:把数据库改回 active 只解决鉴权层,
* 内存里的 banned / 惩罚标记不会自己消失(30 秒对账虽能兜底恢复,但后台点击后应立即生效)。
*/
export function releaseWorkerRiskTracker(input: {
workerId?: number | null
ip?: string | null
}): void {
const workerId = Number(input.workerId)
if (Number.isInteger(workerId) && workerId > 0) {
riskTrackers.delete(`worker:${workerId}`)
}
const ip = String(input.ip || '').trim()
if (ip) {
riskTrackers.delete(`ip:${ip}`)
}
}
/**
* 后台人工封禁:同时置位内存标记,避免出现「数据库已 frozen、内存却仍放行」的反向不一致。
* 仅支持按账号封禁——按 IP 封禁没有可对账的事实来源,会永久生效,因此不开放。
*/
export function markWorkerRiskBanned(workerId: number): void {
const id = Number(workerId)
if (!Number.isInteger(id) || id <= 0) return
const now = Date.now()
const tracker = getOrCreateTracker(`worker:${id}`, now)
tracker.banned = true
tracker.bannedVerifiedAt = now
tracker.lastWorkerId = id
tracker.lastSeenAt = now
}
/** 测试辅助:清空风控记录 */
export function resetWorkerRiskTrackersForTest(): void {
riskTrackers.clear()