feat(risk-control): 新增 IP/账号手动封禁，并收口财务同步的全表更新

- 风控：worker_risk_bans 表 + 前置封禁中间件（登录一并拦截）+ 后台列表/新增/解封 - 财务：成本快照补商品维度收口到本批 order_item_id，不再整表扫 899MB 的 order_items - 性能：order_items 覆盖索引（迁移 110）、大厅 pageSize 上限 20、已读/上传接口限流
2026-10-30 20:15:14 +08:00
parent 1b170d3f45
commit fb485f7e36
17 changed files with 1373 additions and 18 deletions
@@ -0,0 +1,138 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import type { Request, Response } from 'express'
+
+import {
+  createWorkerBanGuardMiddleware,
+  normalizeWorkerIp,
+  resetWorkerBanGuardForTest,
+  type WorkerBanLoader,
+} from './worker-ban-guard.js'
+
+function buildMockReq(options: { ip?: string; workerId?: number }): 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,
+    method: 'GET',
+    originalUrl: '/api/v1/worker/hall/orders',
+    socket: { remoteAddress: options.ip || '127.0.0.1' },
+  } as unknown as Request
+}
+
+function buildMockRes() {
+  const state = { status: 200, payload: null as unknown }
+  const res = {
+    status(code: number) {
+      state.status = code
+      return res
+    },
+    setHeader() {
+      return res
+    },
+    json(body: unknown) {
+      state.payload = body
+      return res
+    },
+  } as unknown as Response
+  return {
+    get status() {
+      return state.status
+    },
+    get payload() {
+      return state.payload
+    },
+    res,
+  }
+}
+
+function errorCodeOf(payload: unknown): string {
+  const body = payload as { error?: { errorCode?: string }; errorCode?: string } | null
+  return body?.error?.errorCode || body?.errorCode || ''
+}
+
+test('IP 封禁：命中即 403，且不放行到后续中间件', async () => {
+  resetWorkerBanGuardForTest()
+  const loadBans: WorkerBanLoader = async () => [
+    { target_type: 'ip', target_value: '::ffff:203.0.113.9' },
+  ]
+  const middleware = createWorkerBanGuardMiddleware({ loadBans })
+  const req = buildMockReq({ ip: '203.0.113.9' })
+  const res = buildMockRes()
+  let nextCalled = false
+
+  await middleware(req, res.res, () => {
+    nextCalled = true
+  })
+
+  assert.equal(nextCalled, false, '被封 IP 不能放行')
+  assert.equal(res.status, 403)
+  assert.equal(errorCodeOf(res.payload), 'worker_ip_banned')
+})
+
+test('账号封禁：按 token 里的 workerId 拦截（即使换了 IP）', async () => {
+  resetWorkerBanGuardForTest()
+  const loadBans: WorkerBanLoader = async () => [
+    { target_type: 'worker', target_value: '42', worker_id: '42' },
+  ]
+  const middleware = createWorkerBanGuardMiddleware({ loadBans })
+  const req = buildMockReq({ ip: '198.51.100.7', workerId: 42 })
+  const res = buildMockRes()
+  let nextCalled = false
+
+  await middleware(req, res.res, () => {
+    nextCalled = true
+  })
+
+  assert.equal(nextCalled, false, '被封账号不能放行')
+  assert.equal(res.status, 403)
+  assert.equal(errorCodeOf(res.payload), 'worker_account_banned')
+})
+
+test('未命中封禁时正常放行', async () => {
+  resetWorkerBanGuardForTest()
+  const middleware = createWorkerBanGuardMiddleware({
+    loadBans: async () => [{ target_type: 'ip', target_value: '203.0.113.9' }],
+  })
+  const req = buildMockReq({ ip: '198.51.100.7', workerId: 7 })
+  const res = buildMockRes()
+  let nextCalled = false
+
+  await middleware(req, res.res, () => {
+    nextCalled = true
+  })
+
+  assert.equal(nextCalled, true)
+  assert.equal(res.status, 200)
+})
+
+test('加载封禁列表失败时 fail-open（不能因为数据库抖动挡死整个接单端）', async () => {
+  resetWorkerBanGuardForTest()
+  const middleware = createWorkerBanGuardMiddleware({
+    loadBans: async () => {
+      throw new Error('db down')
+    },
+  })
+  const req = buildMockReq({ ip: '203.0.113.9' })
+  const res = buildMockRes()
+  let nextCalled = false
+
+  await middleware(req, res.res, () => {
+    nextCalled = true
+  })
+
+  assert.equal(nextCalled, true)
+})
+
+test('normalizeWorkerIp 归一化 IPv6-mapped 与方括号写法', () => {
+  assert.equal(normalizeWorkerIp('::ffff:203.0.113.9'), '203.0.113.9')
+  assert.equal(normalizeWorkerIp('[2001:DB8::1]'), '2001:db8::1')
+  assert.equal(normalizeWorkerIp(' 203.0.113.9 '), '203.0.113.9')
+})
@@ -0,0 +1,170 @@
+import type { NextFunction, Request, Response } from 'express'
+
+import { listActiveWorkerRiskBans } from '../repositories/worker-platform/worker-risk-ban-repo.js'
+import { buildErrorPayload, createHttpError } from '../utils/http.js'
+import { logError, logWarn } from '../utils/logger.js'
+import { extractFastWorkerId, resolveClientIp } from './worker-risk-control.js'
+
+/**
+ * 手动封禁（IP / 账号）拦截中间件。
+ *
+ * 与 worker-risk-control 的分工：
+ *   · worker-risk-control：按**行为**自动升级（限流 → 惩罚 → 自动封号），状态在进程内存；
+ *   · 本中间件：按**人工决定**拦截，事实存在 worker_risk_bans 表，进程重启不丢。
+ *
+ * 性能约束（这是被刷场景下最热的代码）：
+ *   · 每个请求只做两次 Set/Map 命中，绝不查库；
+ *   · 全量封禁列表进程内缓存 30 秒，后台增删封禁时立即失效（同进程）；
+ *   · 加载失败 **fail-open**：宁可短暂漏挡，也不能因为一次数据库抖动把整个接单端挡死。
+ *
+ * 拦截范围：挂在整个 /api/v1/worker 路由之前，因此登录、大厅、订单、上传都会被挡；
+ * 被封 IP 连登录都进不来——这正是运营想要的"现在就挡住它"。
+ */
+
+type WorkerBanSnapshot = {
+  ips: Set<string>
+  workers: Set<number>
+  expiresAt: number
+}
+
+const CACHE_TTL_MS = 30_000
+
+/**
+ * 归一化 IP：去掉 IPv6-mapped 前缀与方括号、统一小写。
+ * 反向代理（Caddy）对 IPv4 请求可能给出 `::ffff:1.2.3.4`，不归一化会导致封了却挡不住。
+ */
+export function normalizeWorkerIp(value: unknown): string {
+  return String(value || '')
+    .trim()
+    .replace(/^\[|\]$/g, '')
+    .replace(/^::ffff:/i, '')
+    .toLowerCase()
+}
+
+let snapshot: WorkerBanSnapshot | null = null
+let inflightLoad: Promise<void> | null = null
+
+/** 后台增删封禁后调用，让本进程立即生效，不必等 TTL。 */
+export function invalidateWorkerBanCache(): void {
+  snapshot = null
+}
+
+async function loadSnapshot(): Promise<void> {
+  const rows = await listActiveWorkerRiskBans()
+  const ips = new Set<string>()
+  const workers = new Set<number>()
+  for (const row of rows) {
+    if (row.target_type === 'ip') {
+      const ip = normalizeWorkerIp(row.target_value)
+      if (ip) ips.add(ip)
+      continue
+    }
+    const workerId = Number(row.worker_id || row.target_value)
+    if (Number.isInteger(workerId) && workerId > 0) workers.add(workerId)
+  }
+  snapshot = { ips, workers, expiresAt: Date.now() + CACHE_TTL_MS }
+}
+
+async function ensureSnapshot(): Promise<WorkerBanSnapshot | null> {
+  const now = Date.now()
+  if (snapshot && snapshot.expiresAt > now) return snapshot
+  if (!inflightLoad) {
+    inflightLoad = loadSnapshot()
+      .catch((error) => {
+        // fail-open：记录一次，保留旧的快照（如果有），不要阻塞请求
+        logError('[worker/ban-guard]', '加载手动封禁列表失败，本进程暂时放行', {
+          error: error instanceof Error ? error.message : String(error),
+        })
+        if (snapshot) snapshot.expiresAt = Date.now() + CACHE_TTL_MS
+      })
+      .finally(() => {
+        inflightLoad = null
+      })
+  }
+  await inflightLoad
+  return snapshot
+}
+
+/** 同一目标 60 秒内只打一条日志，避免被封的脚本刷爆日志盘。 */
+const lastLogAt = new Map<string, number>()
+const LOG_INTERVAL_MS = 60_000
+
+function shouldLog(key: string): boolean {
+  const now = Date.now()
+  const last = lastLogAt.get(key) || 0
+  if (now - last < LOG_INTERVAL_MS) return false
+  lastLogAt.set(key, now)
+  if (lastLogAt.size > 5_000) lastLogAt.clear()
+  return true
+}
+
+export type WorkerBanLoader = () => Promise<
+  Array<{ target_type: 'ip' | 'worker'; target_value: string; worker_id?: string | null }>
+>
+
+export function createWorkerBanGuardMiddleware(deps: { loadBans?: WorkerBanLoader } = {}) {
+  // 测试注入用：不传则走真实仓储 + 30 秒缓存
+  const injected = deps.loadBans
+  return async (req: Request, res: Response, next: NextFunction): Promise<void> => {
+    const ip = normalizeWorkerIp(resolveClientIp(req))
+    const workerId = extractFastWorkerId(req)
+
+    let bannedIp = false
+    let bannedWorker = false
+
+    if (injected) {
+      try {
+        const rows = await injected()
+        bannedIp = rows.some(
+          (row) => row.target_type === 'ip' && normalizeWorkerIp(row.target_value) === ip,
+        )
+        bannedWorker =
+          workerId !== null &&
+          rows.some(
+            (row) =>
+              row.target_type === 'worker' &&
+              Number(row.worker_id || row.target_value) === workerId,
+          )
+      } catch {
+        bannedIp = false
+        bannedWorker = false
+      }
+    } else {
+      const current = await ensureSnapshot()
+      bannedIp = Boolean(current?.ips.has(ip))
+      bannedWorker = workerId !== null && Boolean(current?.workers.has(workerId))
+    }
+
+    if (!bannedIp && !bannedWorker) {
+      next()
+      return
+    }
+
+    const targetKey = bannedIp ? `ip:${ip}` : `worker:${workerId}`
+    if (shouldLog(targetKey)) {
+      logWarn('[worker/ban-guard]', '命中手动封禁，拒绝请求', {
+        targetKey,
+        ip,
+        workerId,
+        method: req.method,
+        originalUrl: req.originalUrl,
+      })
+    }
+
+    const error = createHttpError(
+      bannedIp ? '该网络出口已被后台封禁，无法访问接单端' : '该接单账号已被后台手动封禁',
+      {
+        statusCode: 403,
+        errorCode: bannedIp ? 'worker_ip_banned' : 'worker_account_banned',
+      },
+    )
+    res.status(403).json(buildErrorPayload(error, '已被后台手动封禁'))
+  }
+}
+
+/** 测试辅助：清空缓存与日志节流表。 */
+export function resetWorkerBanGuardForTest(): void {
+  snapshot = null
+  inflightLoad = null
+  lastLogAt.clear()
+}