From 78c676d2b932280661c263a780565b3f97245282 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 2 Oct 2026 17:05:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(worker-risk):=20=E9=A3=8E=E6=8E=A7?= =?UTF-8?q?=E5=90=8E=E5=8F=B0=E9=A1=B5=E9=9D=A2=EF=BC=88=E5=8F=AF=E8=B0=83?= =?UTF-8?q?=E9=98=88=E5=80=BC/=E5=AE=9E=E6=97=B6=E7=8A=B6=E6=80=81/?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E7=95=99=E7=97=95/=E4=BA=BA=E5=B7=A5?= =?UTF-8?q?=E5=B0=81=E7=A6=81=E8=A7=A3=E5=B0=81=EF=BC=89=EF=BC=8C=E5=B9=B6?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=A7=A3=E5=B0=81=E5=90=8E=E4=BB=8D=E8=A2=AB?= =?UTF-8?q?=20403=20=E7=9A=84=20bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 修复的 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 全绿 --- apps/backend/src/config/app-config-keys.ts | 1 + .../db/migrations/089_worker_risk_events.sql | 58 +++ .../middleware/worker-risk-control.test.ts | 143 ++++++ .../src/middleware/worker-risk-control.ts | 273 +++++++++-- .../worker-platform/worker-risk-repo.ts | 192 ++++++++ .../src/routes/admin/worker-platform.ts | 2 + .../src/routes/admin/worker-platform/risk.ts | 145 ++++++ .../admin-worker-risk-service.ts | 215 +++++++++ .../admin-worker-user-service.ts | 9 + .../src/services/worker-platform/index.ts | 1 + .../worker-risk-config-service.ts | 128 +++++ .../pages/admin/AdminWorkerPlatformPage.tsx | 3 + .../pages/admin/panels/RiskControlPanel.tsx | 450 ++++++++++++++++++ .../src/services/admin/worker-platform.ts | 49 ++ apps/frontend/src/types/worker-platform.ts | 66 +++ 15 files changed, 1706 insertions(+), 29 deletions(-) create mode 100644 apps/backend/src/db/migrations/089_worker_risk_events.sql create mode 100644 apps/backend/src/repositories/worker-platform/worker-risk-repo.ts create mode 100644 apps/backend/src/routes/admin/worker-platform/risk.ts create mode 100644 apps/backend/src/services/worker-platform/admin-worker-risk-service.ts create mode 100644 apps/backend/src/services/worker-platform/worker-risk-config-service.ts create mode 100644 apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx diff --git a/apps/backend/src/config/app-config-keys.ts b/apps/backend/src/config/app-config-keys.ts index cc7f4206..d3204884 100644 --- a/apps/backend/src/config/app-config-keys.ts +++ b/apps/backend/src/config/app-config-keys.ts @@ -7,6 +7,7 @@ export const APP_CONFIG_KEYS = { workerProductMatch: 'worker_product_match', workerAnnouncement: 'worker_announcement', workerHall: 'worker_hall', + workerRiskControl: 'worker_risk_control', workerMockShops: 'worker_mock_shops', kuaishouIndustrySource: 'kuaishou_industry_source', affiliateDash: 'affiliate_dash', diff --git a/apps/backend/src/db/migrations/089_worker_risk_events.sql b/apps/backend/src/db/migrations/089_worker_risk_events.sql new file mode 100644 index 00000000..6dba0996 --- /dev/null +++ b/apps/backend/src/db/migrations/089_worker_risk_events.sql @@ -0,0 +1,58 @@ +-- 089_worker_risk_events.sql —— 风控事件留痕（只记显著事件，不按请求写库）。 +-- +-- 背景：大厅阶梯式风控（middleware/worker-risk-control.ts）的状态原本只存在进程内存里， +-- 容器一重建就全部清零——生产上 10/02 一天重建过 3 次，被封的脚本可以重新从头开始； +-- 后台也没有任何入口可看，只能翻日志文件。 +-- +-- 关键取舍：**只记录显著事件**（进入二级惩罚、三级自动封禁、人工处置）， +-- 不记录每一次被拒请求。原因是被刷时实测可达 45~65 QPS，一请求一写会把数据库重新拖垮 +-- （而这正是最初要解决的问题）。被拒请求的实时计数仍来自进程内存的 tracker， +-- 由管理接口直接暴露，不落库。 +-- +-- 量级预期：每天几条到几百条，可长期保留。 + +CREATE TABLE IF NOT EXISTS worker_risk_events ( + id BIGSERIAL PRIMARY KEY, + event_type TEXT NOT NULL + CHECK (event_type IN ( + 'tier2_penalized', + 'tier3_autofrozen', + 'manual_banned', + 'manual_unbanned', + 'manual_ban_failed' + )), + -- 与中间件的跟踪维度一致：worker:（能解出 token）或 ip:（未登录/无 token） + target_key TEXT NOT NULL DEFAULT '', + worker_id BIGINT REFERENCES worker_users(id) ON DELETE SET NULL, + -- 冗余留存账号名：账号被删或被改名后仍要能追溯 + worker_username TEXT NOT NULL DEFAULT '', + ip TEXT NOT NULL DEFAULT '', + -- 触发时的现场快照，便于事后判断阈值是否合理 + violation_count INTEGER NOT NULL DEFAULT 0, + window_count INTEGER NOT NULL DEFAULT 0, + penalty_until TIMESTAMPTZ, + action TEXT NOT NULL DEFAULT 'auto' CHECK (action IN ('auto', 'manual')), + actor_user_id BIGINT, + actor_username TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL DEFAULT '', + detail_json JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- 事件列表按时间倒序分页 +CREATE INDEX IF NOT EXISTS idx_worker_risk_events_created_at + ON worker_risk_events (created_at DESC, id DESC); + +-- 按 IP 追溯：刷子最常轮换账号，IP 是更稳定的维度 +CREATE INDEX IF NOT EXISTS idx_worker_risk_events_ip + ON worker_risk_events (ip, created_at DESC) + WHERE ip <> ''; + +-- 按账号追溯 +CREATE INDEX IF NOT EXISTS idx_worker_risk_events_worker + ON worker_risk_events (worker_id, created_at DESC) + WHERE worker_id IS NOT NULL; + +-- 按事件类型筛选 +CREATE INDEX IF NOT EXISTS idx_worker_risk_events_type + ON worker_risk_events (event_type, created_at DESC); diff --git a/apps/backend/src/middleware/worker-risk-control.test.ts b/apps/backend/src/middleware/worker-risk-control.test.ts index a74cad8f..08213714 100644 --- a/apps/backend/src/middleware/worker-risk-control.test.ts +++ b/apps/backend/src/middleware/worker-risk-control.test.ts @@ -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() + } +}) diff --git a/apps/backend/src/middleware/worker-risk-control.ts b/apps/backend/src/middleware/worker-risk-control.ts index 137eb531..a189a1fd 100644 --- a/apps/backend/src/middleware/worker-risk-control.ts +++ b/apps/backend/src/middleware/worker-risk-control.ts @@ -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() 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 { + 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 => { + 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[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() diff --git a/apps/backend/src/repositories/worker-platform/worker-risk-repo.ts b/apps/backend/src/repositories/worker-platform/worker-risk-repo.ts new file mode 100644 index 00000000..83d6e566 --- /dev/null +++ b/apps/backend/src/repositories/worker-platform/worker-risk-repo.ts @@ -0,0 +1,192 @@ +import { query } from '../../db/client.js' + +/** + * 风控事件留痕的数据访问层。 + * + * 设计约束（重要）： + * - **只写显著事件**，不写每一次被拒请求。被刷时实测 45~65 QPS，按请求写库会把数据库拖垮， + * 而那正是风控要解决的问题。实时计数走进程内存，见 middleware/worker-risk-control.ts。 + * - 写入方（风控中间件）必须 **fire-and-forget 并吞掉异常**：留痕失败绝不能让 + * 大厅请求变成 500。仓储层因此不抛业务错误，只让 pg 错误向上冒泡给调用方去吞。 + */ + +export type WorkerRiskEventType = + | 'tier2_penalized' + | 'tier3_autofrozen' + | 'manual_banned' + | 'manual_unbanned' + | 'manual_ban_failed' + +export type WorkerRiskEventRow = { + id: string + event_type: WorkerRiskEventType + target_key: string + worker_id: string | null + worker_username: string + ip: string + violation_count: number + window_count: number + penalty_until: Date | null + action: 'auto' | 'manual' + actor_user_id: string | null + actor_username: string + reason: string + detail_json: Record | null + created_at: Date +} + +export type InsertWorkerRiskEventInput = { + eventType: WorkerRiskEventType + targetKey: string + workerId?: number | null + ip?: string + violationCount?: number + windowCount?: number + penaltyUntil?: string | null + action?: 'auto' | 'manual' + actorUserId?: number | null + actorUsername?: string + reason?: string + detail?: Record +} + +/** 写入一条风控事件。账号名用子查询就地补全，避免调用方多发一次查询。 */ +export async function insertWorkerRiskEvent(input: InsertWorkerRiskEventInput): Promise { + await query( + ` + INSERT INTO worker_risk_events ( + event_type, target_key, worker_id, worker_username, ip, + violation_count, window_count, penalty_until, + action, actor_user_id, actor_username, reason, detail_json, created_at + ) VALUES ( + $1, $2, $3, + COALESCE((SELECT username FROM worker_users WHERE id = $3), ''), + $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb, NOW() + ) + `, + [ + input.eventType, + String(input.targetKey || '').slice(0, 200), + input.workerId ?? null, + String(input.ip || '').slice(0, 64), + Math.max(0, Math.trunc(Number(input.violationCount) || 0)), + Math.max(0, Math.trunc(Number(input.windowCount) || 0)), + input.penaltyUntil ?? null, + input.action === 'manual' ? 'manual' : 'auto', + input.actorUserId ?? null, + String(input.actorUsername || '').slice(0, 100), + String(input.reason || '').slice(0, 500), + JSON.stringify(input.detail ?? {}), + ], + ) +} + +export type ListWorkerRiskEventsInput = { + limit: number + offset?: number | undefined + eventType?: string | undefined + ip?: string | undefined + workerId?: number | null | undefined + targetKey?: string | undefined + /** 只看某个时间点之后的事件，用于「最近 N 小时」筛选 */ + sinceIso?: string | undefined +} + +/** + * 分页查询风控事件。 + * + * 注意：worker_risk_events 是低频表（每天几条到几百条），因此这里用朴素的分页查询； + * 没有做成 keyset 分页。 + */ +export async function listWorkerRiskEvents( + input: ListWorkerRiskEventsInput, +): Promise { + const { where, params } = buildEventFilters(input) + const limit = Math.min(200, Math.max(1, Math.trunc(Number(input.limit) || 20))) + const offset = Math.max(0, Math.trunc(Number(input.offset) || 0)) + + const result = await query( + ` + SELECT * + FROM worker_risk_events + ${where} + ORDER BY created_at DESC, id DESC + LIMIT $${params.length + 1} OFFSET $${params.length + 2} + `, + [...params, limit, offset], + ) + + return result.rows +} + +export async function countWorkerRiskEvents( + input: Omit, +): Promise { + const { where, params } = buildEventFilters(input) + const result = await query<{ total: string }>( + `SELECT COUNT(*)::text AS total FROM worker_risk_events ${where}`, + params, + ) + return Number(result.rows[0]?.total || 0) +} + +/** 按事件类型统计近期分布，用于面板上的概览卡片。 */ +export async function summarizeWorkerRiskEvents(sinceIso: string) { + const result = await query<{ event_type: string; total: string }>( + ` + SELECT event_type, COUNT(*)::text AS total + FROM worker_risk_events + WHERE created_at >= $1 + GROUP BY event_type + ORDER BY COUNT(*) DESC + `, + [sinceIso], + ) + return result.rows.map((row) => ({ eventType: row.event_type, total: Number(row.total) })) +} + +function buildEventFilters(input: { + eventType?: string | undefined + ip?: string | undefined + workerId?: number | null | undefined + targetKey?: string | undefined + sinceIso?: string | undefined +}): { where: string; params: unknown[] } { + const conditions: string[] = [] + const params: unknown[] = [] + + const eventType = String(input.eventType || '').trim() + if (eventType) { + params.push(eventType) + conditions.push(`event_type = $${params.length}`) + } + + const ip = String(input.ip || '').trim() + if (ip) { + params.push(ip) + conditions.push(`ip = $${params.length}`) + } + + const workerId = Number(input.workerId) + if (Number.isInteger(workerId) && workerId > 0) { + params.push(workerId) + conditions.push(`worker_id = $${params.length}`) + } + + const targetKey = String(input.targetKey || '').trim() + if (targetKey) { + params.push(targetKey) + conditions.push(`target_key = $${params.length}`) + } + + const sinceIso = String(input.sinceIso || '').trim() + if (sinceIso) { + params.push(sinceIso) + conditions.push(`created_at >= $${params.length}`) + } + + return { + where: conditions.length ? `WHERE ${conditions.join(' AND ')}` : '', + params, + } +} diff --git a/apps/backend/src/routes/admin/worker-platform.ts b/apps/backend/src/routes/admin/worker-platform.ts index 55e4d6ad..f0f8c5ff 100644 --- a/apps/backend/src/routes/admin/worker-platform.ts +++ b/apps/backend/src/routes/admin/worker-platform.ts @@ -9,6 +9,7 @@ import adminWorkerPlatformAcceptanceRoutes from './worker-platform/acceptance.js import adminWorkerPlatformAfterSalesRoutes from './worker-platform/after-sales.js' import adminWorkerPlatformLeaderboardRoutes from './worker-platform/leaderboard.js' import adminWorkerPlatformConsignmentRoutes from './worker-platform/consignment.js' +import adminWorkerPlatformRiskRoutes from './worker-platform/risk.js' const router = Router() @@ -21,5 +22,6 @@ router.use(adminWorkerPlatformAfterSalesRoutes) router.use(adminWorkerPlatformAcceptanceRoutes) router.use(adminWorkerPlatformLeaderboardRoutes) router.use(adminWorkerPlatformConsignmentRoutes) +router.use(adminWorkerPlatformRiskRoutes) export default router diff --git a/apps/backend/src/routes/admin/worker-platform/risk.ts b/apps/backend/src/routes/admin/worker-platform/risk.ts new file mode 100644 index 00000000..13be6a21 --- /dev/null +++ b/apps/backend/src/routes/admin/worker-platform/risk.ts @@ -0,0 +1,145 @@ +import { Router } from 'express' +import type { Request } from 'express' + +import { + banAdminWorkerRisk, + getAdminWorkerRiskOverview, + listAdminWorkerRiskEvents, + releaseAdminWorkerRiskTracker, + saveAdminWorkerRiskConfig, + unbanAdminWorkerRisk, +} from '../../../services/worker-platform/index.js' +import type { JsonRecord } from '../../../types/json.js' +import { createJsonHandler, requireAdminRoles } from '../session.js' + +const router = Router() + +function actorOf(req: Request) { + return { + userId: req.adminSession?.userId ?? null, + username: req.adminSession?.username || '', + } +} + +/** Express 5 的 params 值类型是 string | string[]，统一收敛成 string。 */ +function paramOf(value: string | string[] | undefined): string { + if (Array.isArray(value)) return value[0] ?? '' + return value ?? '' +} + +router.get( + '/worker-platform/risk-control', + requireAdminRoles(['admin', 'operator', 'support']), + createJsonHandler(() => getAdminWorkerRiskOverview(), { + successMessage: 'ok', + errorMessage: '读取风控概览失败', + scope: '[admin/worker-platform/risk-control]', + }), +) + +router.post( + '/worker-platform/risk-control', + requireAdminRoles(['admin']), + createJsonHandler((req) => saveAdminWorkerRiskConfig((req.body || {}) as JsonRecord), { + successMessage: '风控参数已保存', + errorMessage: '保存风控参数失败', + scope: '[admin/worker-platform/risk-control]', + audit: (_req, data) => { + const result = data as JsonRecord + const config = (result.config || {}) as JsonRecord + return { + action: 'worker_risk_config_updated', + targetType: 'worker_risk_config', + targetId: 'worker_risk_control', + data: { + enabled: config.enabled !== false, + tier1MaxRequests: config.tier1MaxRequests, + tier2TriggerViolations: config.tier2TriggerViolations, + tier3TriggerViolations: config.tier3TriggerViolations, + }, + } + }, + }), +) + +router.get( + '/worker-platform/risk-control/events', + requireAdminRoles(['admin', 'operator', 'support']), + createJsonHandler((req) => listAdminWorkerRiskEvents((req.query || {}) as JsonRecord), { + successMessage: 'ok', + errorMessage: '读取风控事件失败', + scope: '[admin/worker-platform/risk-control/events]', + }), +) + +router.post( + '/worker-platform/risk-control/workers/:workerId/ban', + requireAdminRoles(['admin']), + createJsonHandler( + (req) => + banAdminWorkerRisk( + paramOf(req.params.workerId), + actorOf(req), + (req.body || {}) as JsonRecord, + ), + { + successMessage: '账号已封禁', + errorMessage: '封禁账号失败', + scope: '[admin/worker-platform/risk-control/ban]', + audit: (req, data) => ({ + action: 'worker_risk_manual_ban', + targetType: 'worker', + targetId: String(paramOf(req.params.workerId)), + data: { result: data } as Record, + }), + }, + ), +) + +router.post( + '/worker-platform/risk-control/workers/:workerId/unban', + requireAdminRoles(['admin']), + createJsonHandler( + (req) => + unbanAdminWorkerRisk( + paramOf(req.params.workerId), + actorOf(req), + (req.body || {}) as JsonRecord, + ), + { + successMessage: '账号已解封', + errorMessage: '解封账号失败', + scope: '[admin/worker-platform/risk-control/unban]', + audit: (req, data) => ({ + action: 'worker_risk_manual_unban', + targetType: 'worker', + targetId: String(paramOf(req.params.workerId)), + data: { result: data } as Record, + }), + }, + ), +) + +/** + * 只清内存风控状态、不改数据库账号状态。 + * 用于「账号本身正常，但内存里还留着旧的封禁/惩罚标记」这类不一致场景。 + */ +router.post( + '/worker-platform/risk-control/workers/:workerId/release', + requireAdminRoles(['admin']), + createJsonHandler( + (req) => releaseAdminWorkerRiskTracker(paramOf(req.params.workerId), actorOf(req)), + { + successMessage: '已释放内存风控状态', + errorMessage: '释放风控状态失败', + scope: '[admin/worker-platform/risk-control/release]', + audit: (req) => ({ + action: 'worker_risk_tracker_released', + targetType: 'worker', + targetId: String(paramOf(req.params.workerId)), + }), + }, + ), +) + +export default router diff --git a/apps/backend/src/services/worker-platform/admin-worker-risk-service.ts b/apps/backend/src/services/worker-platform/admin-worker-risk-service.ts new file mode 100644 index 00000000..25f36f67 --- /dev/null +++ b/apps/backend/src/services/worker-platform/admin-worker-risk-service.ts @@ -0,0 +1,215 @@ +import type { JsonObject } from '../../types/json.js' +import { createHttpError } from '../../utils/http.js' +import { nowIso } from '../../utils/time.js' +import { + listActiveWorkerRiskTrackers, + markWorkerRiskBanned, + releaseWorkerRiskTracker, +} from '../../middleware/worker-risk-control.js' +import { + countWorkerRiskEvents, + insertWorkerRiskEvent, + listWorkerRiskEvents, + summarizeWorkerRiskEvents, + type WorkerRiskEventRow, +} from '../../repositories/worker-platform/worker-risk-repo.js' +import { + getWorkerUserById, + updateWorkerUser, +} from '../../repositories/worker-platform/worker-repo.js' +import { revokeAllWorkerSessions } from '../../repositories/worker-platform/worker-session-repo.js' +import { normalizePage, normalizePageSize } from '../admin/admin-query-utils.js' +import { getWorkerRiskConfig, saveWorkerRiskConfig } from './worker-risk-config-service.js' + +/** + * 后台「接单平台 → 风控」的读写服务。 + * + * 数据来源分两层，界面上必须让运营能分清楚： + * - **实时层**：风控中间件的进程内存 tracker（封禁/惩罚中的 key）。重启即清零，不是历史累计。 + * - **留痕层**：worker_risk_events（只记显著事件）。持久、可追溯。 + */ + +export type WorkerRiskActor = { userId?: number | null; username?: string } + +const OVERVIEW_SUMMARY_WINDOW_MS = 24 * 60 * 60 * 1000 + +export async function getAdminWorkerRiskOverview() { + const config = getWorkerRiskConfig() + const sinceIso = new Date(Date.now() - OVERVIEW_SUMMARY_WINDOW_MS).toISOString() + const [summary, totalEvents] = await Promise.all([ + summarizeWorkerRiskEvents(sinceIso), + countWorkerRiskEvents({}), + ]) + + return { + config, + /** 进程内存中的实时状态；重启会清空，界面需如实标注 */ + activeTrackers: listActiveWorkerRiskTrackers().map((item) => ({ + ...item, + penalizedUntilIso: item.penalizedUntil ? new Date(item.penalizedUntil).toISOString() : null, + lastSeenAtIso: new Date(item.lastSeenAt).toISOString(), + })), + summary24h: summary, + totalEvents, + serverTimeIso: new Date().toISOString(), + } +} + +export async function saveAdminWorkerRiskConfig(payload: JsonObject = {}) { + const config = await saveWorkerRiskConfig(payload) + return { config } +} + +export async function listAdminWorkerRiskEvents(query: JsonObject = {}) { + const page = normalizePage(query.page) + const pageSize = normalizePageSize(query.pageSize) + const sinceHours = Number(query.sinceHours ?? query.since_hours) + + const filters = { + eventType: String(query.eventType ?? query.event_type ?? '').trim(), + ip: String(query.ip || '').trim(), + workerId: Number(query.workerId ?? query.worker_id) || null, + targetKey: String(query.targetKey ?? query.target_key ?? '').trim(), + sinceIso: + Number.isFinite(sinceHours) && sinceHours > 0 + ? new Date(Date.now() - sinceHours * 60 * 60 * 1000).toISOString() + : undefined, + } + + const [rows, total] = await Promise.all([ + listWorkerRiskEvents({ ...filters, limit: pageSize, offset: (page - 1) * pageSize }), + countWorkerRiskEvents(filters), + ]) + + return { + items: rows.map(mapWorkerRiskEvent), + total, + page, + pageSize, + } +} + +/** 人工封禁打手账号（与自动封禁等价，只是留痕标记为人工操作）。 */ +export async function banAdminWorkerRisk( + workerId: number | string, + actor: WorkerRiskActor = {}, + payload: JsonObject = {}, +) { + const id = normalizeWorkerId(workerId) + const worker = await getWorkerUserById(id) + if (!worker) { + throw createHttpError('接单账号不存在', { statusCode: 404, errorCode: 'worker_not_found' }) + } + + const now = nowIso() + const reason = String(payload.reason || '').trim() || '后台人工封禁：疑似外挂刷单或异常高频请求' + + await updateWorkerUser(id, { + status: 'frozen', + frozen_reason: reason, + frozen_at: now, + session_version: Number(worker.session_version || 1) + 1, + updated_at: now, + }) + await revokeAllWorkerSessions(id, now) + // 内存标记必须同步置位，否则会出现「数据库已 frozen、内存却仍放行」的反向不一致 + markWorkerRiskBanned(id) + + await insertWorkerRiskEvent({ + eventType: 'manual_banned', + targetKey: `worker:${id}`, + workerId: id, + ip: String(payload.ip || '').trim(), + reason, + action: 'manual', + actorUserId: actor.userId ?? null, + actorUsername: actor.username || '', + }) + + return { workerId: id, status: 'frozen', reason } +} + +/** 人工解封：恢复账号可用，并释放内存里的风控状态（否则仍会被 403 挡住）。 */ +export async function unbanAdminWorkerRisk( + workerId: number | string, + actor: WorkerRiskActor = {}, + payload: JsonObject = {}, +) { + const id = normalizeWorkerId(workerId) + const worker = await getWorkerUserById(id) + if (!worker) { + throw createHttpError('接单账号不存在', { statusCode: 404, errorCode: 'worker_not_found' }) + } + + const now = nowIso() + const reason = String(payload.reason || '').trim() || '后台人工解封' + + await updateWorkerUser(id, { + status: 'active', + frozen_reason: '', + frozen_at: null, + session_version: Number(worker.session_version || 1) + 1, + updated_at: now, + }) + await revokeAllWorkerSessions(id, now) + releaseWorkerRiskTracker({ workerId: id }) + + await insertWorkerRiskEvent({ + eventType: 'manual_unbanned', + targetKey: `worker:${id}`, + workerId: id, + ip: String(payload.ip || '').trim(), + reason, + action: 'manual', + actorUserId: actor.userId ?? null, + actorUsername: actor.username || '', + }) + + return { workerId: id, status: 'active', reason } +} + +/** 仅释放内存风控状态，不改数据库状态。用于处理「内存误封但账号本身正常」的情况。 */ +export async function releaseAdminWorkerRiskTracker( + workerId: number | string, + actor: WorkerRiskActor = {}, +) { + const id = normalizeWorkerId(workerId) + releaseWorkerRiskTracker({ workerId: id }) + await insertWorkerRiskEvent({ + eventType: 'manual_unbanned', + targetKey: `worker:${id}`, + workerId: id, + reason: '后台仅释放内存风控状态（账号状态未变更）', + action: 'manual', + actorUserId: actor.userId ?? null, + actorUsername: actor.username || '', + }) + return { workerId: id } +} + +function normalizeWorkerId(value: number | string): number { + const id = Number(value) + if (!Number.isInteger(id) || id <= 0) { + throw createHttpError('接单账号 ID 无效', { statusCode: 400, errorCode: 'worker_id_invalid' }) + } + return id +} + +function mapWorkerRiskEvent(row: WorkerRiskEventRow) { + return { + id: String(row.id), + eventType: row.event_type, + targetKey: row.target_key, + workerId: row.worker_id === null ? null : Number(row.worker_id), + workerUsername: row.worker_username, + ip: row.ip, + violationCount: Number(row.violation_count) || 0, + windowCount: Number(row.window_count) || 0, + penaltyUntilIso: row.penalty_until ? new Date(row.penalty_until).toISOString() : null, + action: row.action, + actorUserId: row.actor_user_id === null ? null : Number(row.actor_user_id), + actorUsername: row.actor_username, + reason: row.reason, + createdAtIso: new Date(row.created_at).toISOString(), + } +} diff --git a/apps/backend/src/services/worker-platform/admin-worker-user-service.ts b/apps/backend/src/services/worker-platform/admin-worker-user-service.ts index c4051c07..18daf92f 100644 --- a/apps/backend/src/services/worker-platform/admin-worker-user-service.ts +++ b/apps/backend/src/services/worker-platform/admin-worker-user-service.ts @@ -12,6 +12,7 @@ import { import type { JsonObject } from '../../types/json.js' import { createHttpError } from '../../utils/http.js' import { nowIso } from '../../utils/time.js' +import { releaseWorkerRiskTracker } from '../../middleware/worker-risk-control.js' import { normalizePage, normalizePageSize } from '../admin/admin-query-utils.js' import { hashWorkerPassword, @@ -108,6 +109,14 @@ export async function reviewAdminWorkerUser(workerId: number | string, payload: updated_at: now, }) await revokeAllWorkerSessions(worker.id, now) + + // 必须同步释放内存里的风控状态。 + // review 只允许把状态改成 active/rejected/disabled/pending_review（没有 frozen）， + // 所以走到这里就意味着「不再由风控冻结」。而风控中间件的 banned 检查发生在鉴权之前， + // 只改数据库的话，账号在库里已经恢复、请求却仍然被 403 挡在门外， + // 只能等容器重启 —— 这是生产上实际发生过的故障。 + releaseWorkerRiskTracker({ workerId: worker.id }) + return { worker: mapWorkerUser(updated || worker) } } diff --git a/apps/backend/src/services/worker-platform/index.ts b/apps/backend/src/services/worker-platform/index.ts index b354c19b..7ebeb7de 100644 --- a/apps/backend/src/services/worker-platform/index.ts +++ b/apps/backend/src/services/worker-platform/index.ts @@ -28,6 +28,7 @@ export * from './admin-worker-leaderboard-service.js' export * from './admin-kuaishou-match-service.js' export * from './admin-worker-config-service.js' export * from './admin-worker-user-service.js' +export * from './admin-worker-risk-service.js' export * from './admin-worker-detail-service.js' export * from './admin-work-order-query-service.js' export * from './admin-worker-feedback-service.js' diff --git a/apps/backend/src/services/worker-platform/worker-risk-config-service.ts b/apps/backend/src/services/worker-platform/worker-risk-config-service.ts new file mode 100644 index 00000000..19824216 --- /dev/null +++ b/apps/backend/src/services/worker-platform/worker-risk-config-service.ts @@ -0,0 +1,128 @@ +import { APP_CONFIG_KEYS } from '../../config/app-config-keys.js' +import type { JsonObject } from '../../types/json.js' +import { readAppConfigEntry, saveAppConfigEntry } from '../config/app-config-store.js' + +/** + * 大厅风控阈值配置（可在后台「接单平台 → 风控」页调整）。 + * + * 这些值原先硬编码在 middleware/worker-risk-control.ts 里，改一次要发版； + * 现在落到 app_config_entries，保存后热生效，无需重启。 + * + * 为什么需要本层的 memo：中间件每个大厅请求都要读一次配置，而 readAppConfigEntry + * 内部会对整个 JSON 做一次深拷贝 + normalize。被刷时实测 45~65 QPS， + * 每请求一次深拷贝是纯浪费，因此这里加一个极短 TTL 的进程内缓存。 + * 后台保存时会显式失效，所以正常情况下改完立即生效（TTL 只是兜底）。 + */ + +export type WorkerRiskConfig = { + /** 总开关。关闭后大厅不再限流（仅用于排障，正常应保持开启） */ + enabled: boolean + /** 一级限流：统计窗口（毫秒） */ + tier1WindowMs: number + /** 一级限流：窗口内允许的请求数，超过即 429 */ + tier1MaxRequests: number + /** 违规计数窗口（毫秒），在此窗口内累计的 429 次数用于升级判定 */ + violationWindowMs: number + /** 二级惩罚：违规次数达到该值即进入静默惩罚期 */ + tier2TriggerViolations: number + /** 二级惩罚时长（毫秒） */ + tier2PenaltyMs: number + /** 三级封号：惩罚期内仍继续请求，累计违规达到该值即自动封号 */ + tier3TriggerViolations: number +} + +const CONFIG_MEMO_TTL_MS = 5_000 + +let configMemo: { value: WorkerRiskConfig; expiresAt: number } | null = null + +export function createDefaultWorkerRiskConfig(): WorkerRiskConfig { + return { + enabled: true, + tier1WindowMs: 1_000, + tier1MaxRequests: 3, + violationWindowMs: 60_000, + tier2TriggerViolations: 10, + tier2PenaltyMs: 300_000, + tier3TriggerViolations: 30, + } +} + +export function getWorkerRiskConfig(): WorkerRiskConfig { + const now = Date.now() + if (configMemo && configMemo.expiresAt > now) { + return configMemo.value + } + + const value = readAppConfigEntry({ + configKey: APP_CONFIG_KEYS.workerRiskControl, + fallback: createDefaultWorkerRiskConfig, + normalize: normalizeWorkerRiskConfig, + }) + + configMemo = { value, expiresAt: now + CONFIG_MEMO_TTL_MS } + return value +} + +export async function saveWorkerRiskConfig(rawValue: unknown): Promise { + const saved = await saveAppConfigEntry({ + configKey: APP_CONFIG_KEYS.workerRiskControl, + value: rawValue, + normalize: normalizeWorkerRiskConfig, + }) + // 同进程内立即生效，不等 TTL 过期 + configMemo = { value: saved, expiresAt: Date.now() + CONFIG_MEMO_TTL_MS } + return saved +} + +/** 仅测试使用：清掉进程内缓存，让下次读取回到 app-config-store。 */ +export function clearWorkerRiskConfigMemoForTest(): void { + configMemo = null +} + +export function normalizeWorkerRiskConfig(rawValue: unknown): WorkerRiskConfig { + const fallback = createDefaultWorkerRiskConfig() + const source = isPlainObject(rawValue) ? rawValue : {} + + const enabled = + source.enabled === undefined + ? fallback.enabled + : source.enabled !== false && String(source.enabled) !== 'false' + + return { + enabled, + tier1WindowMs: pickInt(source.tier1WindowMs, 100, 60_000, fallback.tier1WindowMs), + // 上限 600/窗口：再高就失去限流意义（1 秒 600 次已属异常），避免误配把风控关成摆设 + tier1MaxRequests: pickInt(source.tier1MaxRequests, 1, 600, fallback.tier1MaxRequests), + violationWindowMs: pickInt( + source.violationWindowMs, + 1_000, + 3_600_000, + fallback.violationWindowMs, + ), + tier2TriggerViolations: pickInt( + source.tier2TriggerViolations, + 1, + 10_000, + fallback.tier2TriggerViolations, + ), + tier2PenaltyMs: pickInt(source.tier2PenaltyMs, 1_000, 86_400_000, fallback.tier2PenaltyMs), + tier3TriggerViolations: pickInt( + source.tier3TriggerViolations, + 1, + 100_000, + fallback.tier3TriggerViolations, + ), + } +} + +function pickInt(value: unknown, min: number, max: number, fallback: number): number { + const parsed = Number(value) + if (!Number.isFinite(parsed)) return fallback + const truncated = Math.trunc(parsed) + if (truncated < min || truncated > max) return fallback + return truncated +} + +function isPlainObject(value: unknown): value is JsonObject { + return Object.prototype.toString.call(value) === '[object Object]' +} diff --git a/apps/frontend/src/pages/admin/AdminWorkerPlatformPage.tsx b/apps/frontend/src/pages/admin/AdminWorkerPlatformPage.tsx index b27364eb..b70bc412 100644 --- a/apps/frontend/src/pages/admin/AdminWorkerPlatformPage.tsx +++ b/apps/frontend/src/pages/admin/AdminWorkerPlatformPage.tsx @@ -20,6 +20,7 @@ import LevelsPanel from './panels/LevelsPanel' import NotificationsPanel from './panels/NotificationsPanel' import ProductMatchPanel from './panels/ProductMatchPanel' import ProductRulesPanel from './panels/ProductRulesPanel' +import RiskControlPanel from './panels/RiskControlPanel' import WorkersPanel from './panels/WorkersPanel' import WorkOrdersPanel from './panels/WorkOrdersPanel' @@ -47,6 +48,7 @@ const TAB_QUERY_KEYS: Record = { notifications: [['admin-notifications']], announcement: [['admin-announcements']], 'hall-config': [['admin-hall-config']], + 'risk-control': [['admin-worker-risk-overview'], ['admin-worker-risk-events']], } export default function AdminWorkerPlatformPage() { @@ -184,6 +186,7 @@ export default function AdminWorkerPlatformPage() { { key: 'leaderboard', label: '排行榜', children: }, { key: 'finance', label: '资金', children: }, { key: 'levels', label: '等级权限', children: }, + { key: 'risk-control', label: '风控', children: }, ...(isAdmin ? [ { diff --git a/apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx b/apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx new file mode 100644 index 00000000..1d0c798d --- /dev/null +++ b/apps/frontend/src/pages/admin/panels/RiskControlPanel.tsx @@ -0,0 +1,450 @@ +import { useQuery, useQueryClient } from '@tanstack/react-query' +import { + Alert, + App, + Button, + Card, + Descriptions, + Form, + Input, + InputNumber, + Select, + Space, + Switch, + Table, + Tag, +} from 'antd' +import type { ColumnsType } from 'antd/es/table' +import { useEffect, useState } from 'react' + +import { + banAdminWorkerRisk, + fetchAdminWorkerRiskEvents, + fetchAdminWorkerRiskOverview, + releaseAdminWorkerRiskTracker, + saveAdminWorkerRiskConfig, + unbanAdminWorkerRisk, +} from '@/services/admin' +import type { + WorkerRiskActiveTracker, + WorkerRiskConfig, + WorkerRiskEvent, + WorkerRiskEventType, +} from '@/types/worker-platform' +import { formatDateTime } from '@/utils/date-time' +import { getAdminRole } from '@/utils/admin-auth' + +const EVENT_TYPE_LABEL: Record = { + tier2_penalized: { text: '二级惩罚', color: 'orange' }, + tier3_autofrozen: { text: '三级自动封号', color: 'red' }, + manual_banned: { text: '人工封禁', color: 'volcano' }, + manual_unbanned: { text: '人工解封', color: 'green' }, + manual_ban_failed: { text: '封禁失败', color: 'default' }, +} + +function formatRemaining(iso: string | null) { + if (!iso) return '-' + const remainMs = new Date(iso).getTime() - Date.now() + if (!Number.isFinite(remainMs) || remainMs <= 0) return '已结束' + const seconds = Math.ceil(remainMs / 1000) + if (seconds < 60) return `剩余 ${seconds} 秒` + return `剩余 ${Math.ceil(seconds / 60)} 分钟` +} + +export default function RiskControlPanel() { + const { message, modal } = App.useApp() + const queryClient = useQueryClient() + const [form] = Form.useForm() + const [saving, setSaving] = useState(false) + const [acting, setActing] = useState(false) + const [page, setPage] = useState(1) + const [pageSize, setPageSize] = useState(20) + const [eventType, setEventType] = useState('') + // 写接口在后端要求 admin 角色，非 admin 只读，避免点了才报错 + const canManage = getAdminRole() === 'admin' + const [ipFilter, setIpFilter] = useState('') + + // 实时层：内存状态，刷新频率高一些；重启会清零 + const overviewQuery = useQuery({ + queryKey: ['admin-worker-risk-overview'], + queryFn: fetchAdminWorkerRiskOverview, + refetchInterval: 15_000, + }) + const overview = overviewQuery.data?.data + + const eventsQuery = useQuery({ + queryKey: ['admin-worker-risk-events', eventType, ipFilter, page, pageSize], + queryFn: () => + fetchAdminWorkerRiskEvents({ + page, + pageSize, + eventType: eventType || undefined, + ip: ipFilter.trim() || undefined, + }), + }) + const events = eventsQuery.data?.data + + useEffect(() => { + if (overview?.config) { + form.setFieldsValue(overview.config) + } + }, [overview?.config, form]) + + async function refreshAll() { + await Promise.all([ + queryClient.invalidateQueries({ queryKey: ['admin-worker-risk-overview'] }), + queryClient.invalidateQueries({ queryKey: ['admin-worker-risk-events'] }), + ]) + } + + async function saveConfig(values: WorkerRiskConfig) { + setSaving(true) + try { + await saveAdminWorkerRiskConfig(values) + await refreshAll() + message.success('风控参数已保存并立即生效') + } catch (error) { + message.error(error instanceof Error ? error.message : '保存风控参数失败') + } finally { + setSaving(false) + } + } + + function confirmAction(title: string, content: string, run: () => Promise) { + modal.confirm({ + title, + content, + okButtonProps: { danger: title.includes('封禁') }, + onOk: async () => { + setActing(true) + try { + await run() + await refreshAll() + message.success(`${title}成功`) + } catch (error) { + message.error(error instanceof Error ? error.message : `${title}失败`) + } finally { + setActing(false) + } + }, + }) + } + + const trackerColumns: ColumnsType = [ + { + title: '跟踪对象', + dataIndex: 'key', + render: (_value, row) => ( + + {row.workerId ? `账号 #${row.workerId}` : '未登录（按 IP）'} + {row.ip || '-'} + + ), + }, + { + title: '当前状态', + dataIndex: 'banned', + width: 140, + render: (_value, row) => { + if (row.banned) return 已封禁 + if (row.penalizedUntilIso) return 惩罚中 + return 仅记录 + }, + }, + { + title: '惩罚剩余', + dataIndex: 'penalizedUntilIso', + width: 120, + render: (value: string | null) => formatRemaining(value), + }, + { title: '违规次数', dataIndex: 'violationCount', width: 100 }, + { title: '被拒请求', dataIndex: 'rejectedCount', width: 100 }, + { + title: '最近出现', + dataIndex: 'lastSeenAtIso', + width: 170, + render: (value: string) => formatDateTime(value), + }, + { + title: '操作', + key: 'actions', + width: 240, + render: (_value, row) => ( + + {row.workerId ? ( + <> + + + + + ) : ( + 无账号维度 + )} + + ), + }, + ] + + const eventColumns: ColumnsType = [ + { + title: '时间', + dataIndex: 'createdAtIso', + width: 170, + render: (value: string) => formatDateTime(value), + }, + { + title: '事件', + dataIndex: 'eventType', + width: 130, + render: (value: WorkerRiskEventType) => { + const meta = EVENT_TYPE_LABEL[value] || { text: value, color: 'default' } + return {meta.text} + }, + }, + { + title: '对象', + dataIndex: 'workerId', + render: (_value, row) => ( + + + {row.workerId ? `#${row.workerId} ${row.workerUsername || ''}` : '未登录（按 IP）'} + + {row.ip || '-'} + + ), + }, + { title: '违规次数', dataIndex: 'violationCount', width: 100 }, + { + title: '触发方式', + dataIndex: 'action', + width: 100, + render: (value: string) => + value === 'manual' ? 人工 : 自动, + }, + { + title: '操作人', + dataIndex: 'actorUsername', + width: 120, + render: (value: string) => value || '-', + }, + { title: '原因', dataIndex: 'reason', ellipsis: true }, + ] + + const summary = overview?.summary24h || [] + + return ( + + + + 保存后立即生效，无需重启} + > + + + + {summary.reduce((acc, item) => acc + item.total, 0)} + + {overview?.totalEvents ?? 0} + + +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + {canManage + ? '阈值调低更严格；调高会放宽限制，请谨慎' + : '只有 admin 角色可以修改阈值'} + + +
+ + + + overviewQuery.refetch()} + loading={overviewQuery.isFetching} + > + 刷新 + + } + > + + + + + + setIpFilter(event.target.value)} + onPressEnter={() => setPage(1)} + /> + + + + +
{ + setPage(nextPage) + setPageSize(nextPageSize) + }, + }} + /> + + + ) +} diff --git a/apps/frontend/src/services/admin/worker-platform.ts b/apps/frontend/src/services/admin/worker-platform.ts index 6f43cc33..01b4bb91 100644 --- a/apps/frontend/src/services/admin/worker-platform.ts +++ b/apps/frontend/src/services/admin/worker-platform.ts @@ -15,6 +15,9 @@ import type { WorkerDepositUnfreeze, WorkerFinanceConfig, WorkerHallConfig, + WorkerRiskConfig, + WorkerRiskOverview, + WorkerRiskEventList, WorkerFinanceRequest, WorkerCancelRequest, WorkerOrderFeedback, @@ -850,3 +853,49 @@ export function deductAdminWorkOrderPendingDeposit( payload, ) } + +// --------------------------------------------------------------------------- +// 大厅风控（接单平台 → 风控） +// --------------------------------------------------------------------------- + +export function fetchAdminWorkerRiskOverview() { + return apiGet('/api/v1/admin/worker-platform/risk-control') +} + +export function saveAdminWorkerRiskConfig(payload: WorkerRiskConfig) { + return apiPost<{ config: WorkerRiskConfig }>( + '/api/v1/admin/worker-platform/risk-control', + payload, + ) +} + +export function fetchAdminWorkerRiskEvents(params: { + page?: number + pageSize?: number + eventType?: string + ip?: string + workerId?: number + sinceHours?: number +}) { + return apiGet('/api/v1/admin/worker-platform/risk-control/events', params) +} + +export function banAdminWorkerRisk(workerId: number, reason?: string) { + return apiPost<{ workerId: number; status: string; reason: string }>( + `/api/v1/admin/worker-platform/risk-control/workers/${workerId}/ban`, + { reason }, + ) +} + +export function unbanAdminWorkerRisk(workerId: number, reason?: string) { + return apiPost<{ workerId: number; status: string; reason: string }>( + `/api/v1/admin/worker-platform/risk-control/workers/${workerId}/unban`, + { reason }, + ) +} + +export function releaseAdminWorkerRiskTracker(workerId: number) { + return apiPost<{ workerId: number }>( + `/api/v1/admin/worker-platform/risk-control/workers/${workerId}/release`, + ) +} diff --git a/apps/frontend/src/types/worker-platform.ts b/apps/frontend/src/types/worker-platform.ts index a4333ce6..bba31b9b 100644 --- a/apps/frontend/src/types/worker-platform.ts +++ b/apps/frontend/src/types/worker-platform.ts @@ -789,3 +789,69 @@ export type CollectLookupResponse = { orders: CollectLookupOrder[] pendingCount: number } + +/** 大厅风控阈值（后台可调，保存即热生效） */ +export type WorkerRiskConfig = { + enabled: boolean + tier1WindowMs: number + tier1MaxRequests: number + violationWindowMs: number + tier2TriggerViolations: number + tier2PenaltyMs: number + tier3TriggerViolations: number +} + +/** 进程内存中的实时风控项（重启即清零，不是历史累计） */ +export type WorkerRiskActiveTracker = { + key: string + workerId: number | null + ip: string + banned: boolean + penalizedUntil: number + penalizedUntilIso: string | null + violationCount: number + windowCount: number + rejectedCount: number + lastSeenAt: number + lastSeenAtIso: string +} + +export type WorkerRiskEventType = + | 'tier2_penalized' + | 'tier3_autofrozen' + | 'manual_banned' + | 'manual_unbanned' + | 'manual_ban_failed' + +/** 风控留痕事件（持久化，只记显著事件） */ +export type WorkerRiskEvent = { + id: string + eventType: WorkerRiskEventType + targetKey: string + workerId: number | null + workerUsername: string + ip: string + violationCount: number + windowCount: number + penaltyUntilIso: string | null + action: 'auto' | 'manual' + actorUserId: number | null + actorUsername: string + reason: string + createdAtIso: string +} + +export type WorkerRiskOverview = { + config: WorkerRiskConfig + activeTrackers: WorkerRiskActiveTracker[] + summary24h: Array<{ eventType: string; total: number }> + totalEvents: number + serverTimeIso: string +} + +export type WorkerRiskEventList = { + items: WorkerRiskEvent[] + total: number + page: number + pageSize: number +}