wmailhub/console/src/routes/internal.ts

/**
 * /internal/email/* :Email Worker <-> Console 内部协议(docs/11 §7/§8/§9)
 *
 *   POST /internal/email/authorize   SMTP 阶段决策:accept | reject | temporary_error
 *   POST /internal/email/ingest      原始 MIME 入库(D1 + R2 + Webhook),仅 accept 后调用
 *
 * 两者都必须通过 HMAC 签名校验 + nonce 防重放 + credential/账号状态检查(docs/11 §8):
 * Console 校验 Worker ID、timestamp、nonce、body hash、HMAC、credential 状态,
 * 并确认请求 Worker 所属 Cloudflare 账号与收件域名一致。
 */

import { Hono } from "hono";
import type { Context, Next } from "hono";
import type { Env, AppVars } from "../env.ts";
import { verifyWorkerRequest } from "../../../shared/src/index.ts";
import { ingestEmail, IngestError, resolveMailbox } from "../services/email_ingest.ts";
import { decryptSecret } from "../db.ts";
import { nowIso } from "../util.ts";

type App = Hono<{ Bindings: Env; Variables: AppVars }>;
type Ctx = Context<{ Bindings: Env; Variables: AppVars }>;

/** # 成员变量: nonce 防重放窗口(与 timestamp 窗口一致) **/
const NONCE_TTL_SECONDS = 600;

/** # 成员变量: reject 状态码与原因(docs/11 §7;reason 不暴露内部信息) **/
const REJECT_REASONS: Record<string, string> = {
  DOMAIN_NOT_FOUND: "Mailbox does not exist",
  DOMAIN_DISABLED: "Receiving for this domain is not enabled",
  MAILBOX_NOT_FOUND: "Mailbox does not exist",
  MAILBOX_DISABLED: "Mailbox is disabled",
  RECEIVING_DISABLED: "Receiving is disabled",
};

/** # 成员变量: authorize 请求体 **/
interface AuthorizeBody {
  recipient?: string;
  sender?: string;
  message_id?: string;
}

/** # 成员变量: 决策结果 **/
export type EmailDecision =
  | { decision: "accept" }
  | { decision: "reject"; code: string; reason: string };

/** # 函数
 * HMAC 校验中间件:验签、时间窗、body hash、nonce 唯一性、credential/账号状态。
 * 原始请求体缓存到 c.var.rawBody 供路由处理器复用;workerAccountId 记录归属账号。
 */
async function workerAuth(c: Ctx, next: Next) {
  const raw = new Uint8Array(await c.req.arrayBuffer());
  const cred = await c.env.DB.prepare(
    `SELECT wc.id, wc.secret_enc, wc.cloudflare_account_id, ca.status AS account_status
     FROM worker_credentials wc JOIN cloudflare_accounts ca ON ca.id = wc.cloudflare_account_id
     WHERE wc.worker_id = ? AND wc.status = 'active' ORDER BY wc.created_at DESC LIMIT 1`,
  )
    .bind((c.req.header("Authorization") ?? "").replace(/^WmailHub-Worker\s+/, "").trim() || "\u0000")
    .first<{ id: string; secret_enc: string; cloudflare_account_id: string; account_status: string }>();
  if (!cred) return c.json({ error: "unknown worker or credential revoked" }, 401);

  const secret = await decryptSecret(c.env, cred.secret_enc);
  const verdict = await verifyWorkerRequest({
    secret,
    method: c.req.method,
    url: c.req.url,
    body: raw,
    headers: c.req.raw.headers,
  });
  if (!verdict.ok) return c.json({ error: verdict.error }, verdict.status);

  // nonce 防重放:主键冲突即拒绝
  try {
    await c.env.DB.prepare("INSERT INTO nonces (nonce, account_id, expires_at) VALUES (?, ?, ?)")
      .bind(
        verdict.nonce,
        cred.cloudflare_account_id,
        new Date((verdict.timestamp + NONCE_TTL_SECONDS) * 1000).toISOString(),
      )
      .run();
  } catch {
    return c.json({ error: "nonce replay detected" }, 401);
  }

  // 撤销立即生效:credential/账号状态在验签后复查
  const fresh = await c.env.DB.prepare(
    `SELECT wc.status AS cred_status, ca.status AS account_status
     FROM worker_credentials wc JOIN cloudflare_accounts ca ON ca.id = wc.cloudflare_account_id
     WHERE wc.id = ?`,
  )
    .bind(cred.id)
    .first<{ cred_status: string; account_status: string }>();
  if (!fresh || fresh.cred_status !== "active" || fresh.account_status === "revoked") {
    return c.json({ error: "credential revoked" }, 401);
  }

  c.set("rawBody", raw);
  c.set("workerCredentialId", cred.id);
  c.set("workerAccountId", cred.cloudflare_account_id);
  await next();
}

/** # 函数
 * 收件人决策逻辑(docs/11 §7):域名归属 -> 域名状态 -> 邮箱精确/ catch-all -> 邮箱状态。
 * 返回明确的 decision,绝不把 5xx/超时解释成邮箱不存在(那由 Worker 层处理)。
 */
export async function decideRecipient(
  env: Env,
  accountId: string,
  recipient: string,
): Promise<EmailDecision> {
  const addr = recipient.trim().toLowerCase();
  const at = addr.lastIndexOf("@");
  if (at <= 0 || at === addr.length - 1) {
    return { decision: "reject", code: "MAILBOX_NOT_FOUND", reason: REJECT_REASONS.MAILBOX_NOT_FOUND };
  }
  const domainName = addr.slice(at + 1);
  const domain = await env.DB.prepare(
    "SELECT id, status FROM domains WHERE name = ? AND account_id = ?",
  )
    .bind(domainName, accountId)
    .first<{ id: string; status: string }>();
  if (!domain) {
    return { decision: "reject", code: "DOMAIN_NOT_FOUND", reason: REJECT_REASONS.DOMAIN_NOT_FOUND };
  }
  if (domain.status !== "active") {
    return { decision: "reject", code: "DOMAIN_DISABLED", reason: REJECT_REASONS.DOMAIN_DISABLED };
  }

  // 邮箱:精确匹配优先,其次 catch-all;disabled 一律拒绝
  const exact = await env.DB.prepare(
    "SELECT id, status FROM mailboxes WHERE address = ? AND domain_id = ?",
  )
    .bind(addr, domain.id)
    .first<{ id: string; status: string }>();
  if (exact) {
    if (exact.status !== "active") {
      return { decision: "reject", code: "MAILBOX_DISABLED", reason: REJECT_REASONS.MAILBOX_DISABLED };
    }
    return { decision: "accept" };
  }
  const catchAll = await env.DB.prepare(
    "SELECT id, status FROM mailboxes WHERE domain_id = ? AND is_catch_all = 1 LIMIT 1",
  )
    .bind(domain.id)
    .first<{ id: string; status: string }>();
  if (catchAll) {
    if (catchAll.status !== "active") {
      return { decision: "reject", code: "MAILBOX_DISABLED", reason: REJECT_REASONS.MAILBOX_DISABLED };
    }
    return { decision: "accept" };
  }
  return { decision: "reject", code: "MAILBOX_NOT_FOUND", reason: REJECT_REASONS.MAILBOX_NOT_FOUND };
}

/** # 函数
 * 装配 Email Worker 内部路由
 */
export function internalRoutes(): App {
  const app = new Hono<{ Bindings: Env; Variables: AppVars }>();

  /** 第一阶段:SMTP 决策(不保存邮件) */
  app.post("/email/authorize", workerAuth, async (c) => {
    const accountId = c.get("workerAccountId")!;
    let body: AuthorizeBody;
    try {
      body = JSON.parse(new TextDecoder().decode(c.get("rawBody") ?? new Uint8Array())) as AuthorizeBody;
    } catch {
      return c.json({ error: "invalid body" }, 400);
    }
    const recipient = (body.recipient ?? "").trim();
    if (!recipient) return c.json({ error: "recipient required" }, 400);
    const verdict = await decideRecipient(c.env, accountId, recipient);
    // 未过期时记录 message_id 便于追踪(仅日志,不入库)
    if (body.message_id) {
      console.log(
        JSON.stringify({ t: nowIso(), msg: "email.authorize", decision: verdict.decision, recipient, message_id: body.message_id }),
      );
    }
    return c.json(verdict);
  });

  /** 第二阶段:raw MIME 入库(message/rfc822;收件人经信封头传递) */
  app.post("/email/ingest", workerAuth, async (c) => {
    const accountId = c.get("workerAccountId")!;
    const recipient = (c.req.header("X-WmailHub-Recipient") ?? "").trim();
    const sender = (c.req.header("X-WmailHub-Sender") ?? "").trim();
    if (!recipient) return c.json({ error: "missing X-WmailHub-Recipient" }, 400);

    // 动态邮箱:authorize 与 ingest 之间状态可能变化,再验一次收件人
    const verdict = await decideRecipient(c.env, accountId, recipient);
    if (verdict.decision === "reject") {
      return c.json({ error: "recipient no longer accepted", code: verdict.code }, 409);
    }

    // 邮箱关联兜底:authorize 已确认可接收,此处解析 mailbox(精确/catch-all)
    const mailboxId = await resolveMailbox(c.env, [recipient]);
    const raw = c.get("rawBody") ?? new Uint8Array();
    try {
      const result = await ingestEmail(c.env, { raw, mailFrom: sender, rcptTo: recipient, accountId });
      return c.json(
        { ok: true, id: result.id, mailbox_id: result.mailboxId, expires_at: result.expiresAt },
        202,
      );
    } catch (err) {
      if (err instanceof IngestError) return c.json({ error: err.message }, err.status as 413);
      return c.json({ error: err instanceof Error ? err.message : "ingest failed" }, 500);
    }
  });

  return app;
}