wmailhub/tests/webhook.test.ts

/** Webhook 测试(v1.1.0 三级模型):SSRF 校验、三级优先级、签名/key/自定义头、重试与退避 */
import { describe, expect, test } from "bun:test";
import { ingestEmail } from "../console/src/services/email_ingest.ts";
import { attemptDelivery, processDueDeliveries, resolveWebhookConfig } from "../console/src/services/webhook_delivery.ts";
import { hmacSha256Hex } from "../shared/src/index.ts";
import { authedFetch, loginCookie, makeTestEnv, TEST_MAIL } from "./helpers.ts";

/** # 函数
 * 起一个本地 HTTP 接收器,记录请求并可控返回状态
 */
function startReceiver(mode: "ok" | "fail500"): { url: string; requests: { headers: Headers; body: string }[]; stop: () => void } {
  const requests: { headers: Headers; body: string }[] = [];
  const server = Bun.serve({
    port: 0,
    fetch: async (req) => {
      requests.push({ headers: req.headers, body: await req.text() });
      if (mode === "fail500") return new Response("no", { status: 500 });
      return Response.json({ ok: true });
    },
  });
  return { url: `http://127.0.0.1:${server.port}/hook`, requests, stop: () => server.stop(true) };
}

/** # 函数
 * 直接给域名/邮箱行写 webhook 配置(key 以空占位,投递 key 传 "" 即可)
 */
async function setScopedWebhook(
  t: Awaited<ReturnType<typeof makeTestEnv>>,
  table: "mailboxes" | "domains",
  id: string,
  url: string | null,
  headers: Record<string, string> = {},
) {
  await t.env.DB.prepare(
    `UPDATE ${table} SET webhook_url = ?, webhook_headers = ?, webhook_key_enc = NULL WHERE id = ?`,
  )
    .bind(url, Object.keys(headers).length ? JSON.stringify(headers) : null, id)
    .run();
}

describe("Webhook(邮箱 -> 域名 -> 全局)", () => {
  test("保存配置时拒绝私网/localhost endpoint(SSRF)", async () => {
    const t = await makeTestEnv();
    // SSRF 防护在"未开启本地联调开关"时生效(生产默认态)
    delete (t.env as { WMAILHUB_ALLOW_PRIVATE_WEBHOOKS?: string }).WMAILHUB_ALLOW_PRIVATE_WEBHOOKS;
    const cookie = await loginCookie(t);
    for (const endpoint of [
      "http://127.0.0.1:8080/hook",
      "http://localhost/hook",
      "http://169.254.169.254/latest/meta-data",
      "http://10.0.0.5/hook",
      "http://[::1]/hook",
      "ftp://example.com/x",
    ]) {
      const resp = await authedFetch(t, cookie, "http://localhost/api/settings", {
        method: "PUT",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ webhook: { url: endpoint } }),
      });
      expect(resp.status).toBe(400);
      const resp2 = await authedFetch(t, cookie, `http://localhost/api/mailboxes/${t.mailboxId}/webhook`, {
        method: "PUT",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ url: endpoint }),
      });
      expect(resp2.status).toBe(400);
    }
  });

  test("全局 Webhook 投递:key 走 Authorization,签名与自定义头正确", async () => {
    const t = await makeTestEnv();
    const recv = startReceiver("ok");
    const cookie = await loginCookie(t);
    const save = await authedFetch(t, cookie, "http://localhost/api/settings", {
      method: "PUT",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        webhook: { url: recv.url, headers: { "X-Custom": "wmailhub" }, key: "wmh_global_key_1234567890" },
      }),
    });
    expect(save.status).toBe(200);

    const email = await ingestEmail(t.env, {
      raw: new TextEncoder().encode(TEST_MAIL),
      mailFrom: "a@b.c",
      rcptTo: "hello@example.test",
      accountId: t.accountId,
    });
    const dlv = await t.env.DB.prepare("SELECT id, target_type FROM webhook_deliveries").first<{
      id: string;
      target_type: string;
    }>();
    expect(dlv?.target_type).toBe("global");

    const stats = await processDueDeliveries(t.env, undefined, 20, true);
    expect(stats.success).toBe(1);

    const req = recv.requests[0];
    expect(req.headers.get("Authorization") ?? null).toBe("Bearer wmh_global_key_1234567890");
    expect(req.headers.get("X-Custom") ?? null).toBe("wmailhub");
    const ts = req.headers.get("X-WmailHub-Timestamp") ?? "";
    const sig = await hmacSha256Hex("wmh_global_key_1234567890", String(ts).concat(".", req.body));
    expect(req.headers.get("X-WmailHub-Signature") ?? null).toBe(sig);
    expect(req.headers.get("X-WmailHub-Event") ?? null).toBe("email.received");
    expect(req.headers.get("X-WmailHub-Delivery") ?? null).toBe(dlv?.id ?? null);
    const payload = JSON.parse(req.body) as Record<string, unknown>;
    expect(payload.event).toBe("email.received");
    expect(payload.id).toBe(email.id);
    expect(payload.mailbox).toBe("*@example.test");
    recv.stop();
  });

  test("三级优先级:邮箱 > 域名 > 全局,命中即止;逐级回退", async () => {
    const t = await makeTestEnv();
    const globalRecv = startReceiver("ok");
    const domainRecv = startReceiver("ok");
    const mailboxRecv = startReceiver("ok");

    // 三级全配
    await setScopedWebhook(t, "domains", t.domainId, domainRecv.url);
    await setScopedWebhook(t, "mailboxes", t.mailboxId, mailboxRecv.url);
    await t.env.DB.prepare(
      "INSERT INTO settings (key, value) VALUES ('global_webhook', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
    )
      .bind(JSON.stringify({ url: globalRecv.url, headers: {} }))
      .run();

    await ingestEmail(t.env, {
      raw: new TextEncoder().encode(TEST_MAIL),
      mailFrom: "a@b.c",
      rcptTo: "hello@example.test",
      accountId: t.accountId,
    });
    await processDueDeliveries(t.env, undefined, 20, true);
    expect(mailboxRecv.requests.length).toBe(1);
    expect(domainRecv.requests.length).toBe(0);
    expect(globalRecv.requests.length).toBe(0);
    const level1 = await t.env.DB.prepare("SELECT target_type FROM webhook_deliveries ORDER BY created_at DESC LIMIT 1").first<{ target_type: string }>();
    expect(level1?.target_type).toBe("mailbox");

    // 清除邮箱级 -> 回退域名级
    await setScopedWebhook(t, "mailboxes", t.mailboxId, null);
    expect((await resolveWebhookConfig(t.env, t.mailboxId))?.targetType).toBe("domain");
    await ingestEmail(t.env, {
      raw: new TextEncoder().encode(TEST_MAIL),
      mailFrom: "a@b.c",
      rcptTo: "hello@example.test",
      accountId: t.accountId,
    });
    await processDueDeliveries(t.env, undefined, 20, true);
    expect(domainRecv.requests.length).toBe(1);

    // 再清除域名级 -> 回退全局
    await setScopedWebhook(t, "domains", t.domainId, null);
    expect((await resolveWebhookConfig(t.env, t.mailboxId))?.targetType).toBe("global");
    await ingestEmail(t.env, {
      raw: new TextEncoder().encode(TEST_MAIL),
      mailFrom: "a@b.c",
      rcptTo: "hello@example.test",
      accountId: t.accountId,
    });
    await processDueDeliveries(t.env, undefined, 20, true);
    expect(globalRecv.requests.length).toBe(1);

    for (const r of [globalRecv, domainRecv, mailboxRecv]) r.stop();
  });

  test("三级都未配置:不入队投递", async () => {
    const t = await makeTestEnv();
    await ingestEmail(t.env, {
      raw: new TextEncoder().encode(TEST_MAIL),
      mailFrom: "a@b.c",
      rcptTo: "hello@example.test",
      accountId: t.accountId,
    });
    const count = await t.env.DB.prepare("SELECT COUNT(*) AS n FROM webhook_deliveries").first<{ n: number }>();
    expect(count?.n).toBe(0);
  });

  test("连续失败进入重试并最终 exhausted(退避递增)", async () => {
    const t = await makeTestEnv();
    // 直接连一个不可达地址,验证 attemptDelivery 失败路径
    const result = await attemptDelivery(
      { url: "http://127.0.0.1:1/hook", headers: {}, key: "" },
      "d_test",
      { event: "email.received" },
      undefined,
      true,
    );
    expect(result.ok).toBe(false);

    // 状态机:把全局 webhook 指向不可达地址,反复处理直到 exhausted
    const cookie = await loginCookie(t);
    await authedFetch(t, cookie, "http://localhost/api/settings", {
      method: "PUT",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ webhook: { url: "http://127.0.0.1:1/hook" } }),
    });
    await ingestEmail(t.env, {
      raw: new TextEncoder().encode(TEST_MAIL),
      mailFrom: "a@b.c",
      rcptTo: "hello@example.test",
      accountId: t.accountId,
    });
    const { WEBHOOK_MAX_ATTEMPTS } = await import("../shared/src/index.ts");
    const row0 = await t.env.DB.prepare(
      "SELECT id FROM webhook_deliveries WHERE status = 'pending' ORDER BY created_at DESC LIMIT 1",
    ).first<{ id: string }>();
    expect(row0).not.toBeNull();
    for (let i = 0; i < WEBHOOK_MAX_ATTEMPTS + 1; i++) {
      const stats = await processDueDeliveries(t.env, undefined, 20, true);
      if (stats.processed === 0) break;
      await t.env.DB.prepare("UPDATE webhook_deliveries SET next_retry_at = ? WHERE id = ?")
        .bind(new Date().toISOString(), row0!.id)
        .run();
    }
    const final = await t.env.DB.prepare("SELECT status, attempt_count FROM webhook_deliveries WHERE id = ?")
      .bind(row0!.id)
      .first<{ status: string; attempt_count: number }>();
    expect(final?.status).toBe("exhausted");
    expect(final?.attempt_count).toBe(WEBHOOK_MAX_ATTEMPTS);
  });
});