#!/usr/bin/env node import crypto from 'node:crypto'; import { existsSync, readFileSync } from 'node:fs'; import { mkdir, readFile, rename, rm, writeFile } from 'node:fs/promises'; import path from 'node:path'; import { pathToFileURL } from 'node:url'; import { DatabaseSync } from 'node:sqlite'; import { createDomain, createInboundFolder, createImportedInboundMessage, getDomainByName, getUser, getUserByLogin, hasImportedInboundMessage, initDatabase, updateDomain, upsertImportedInboundMailbox } from '../src/db.js'; import { createDkimKeyPair } from '../src/dkim.js'; import { isLegacyPasswordHash } from '../src/password-hash.js'; import { importVestaSnapshot, readVestaSnapshotMetadata } from '../src/vesta-maildir-import.js'; const checkpointVersion = 1; const checkpointMessageInterval = 100; const checkpointTimeIntervalMs = 1000; const defaultDbApi = { createDomain, createInboundFolder, createImportedInboundMessage, getDomainByName, getUser, getUserByLogin, hasImportedInboundMessage, initDatabase, updateDomain, upsertImportedInboundMailbox }; export class VestaImportCliError extends Error { constructor(category, message) { super(message); this.name = 'VestaImportCliError'; this.category = category; } } export function parseVestaImportArguments(argv, { env = process.env, cwd = process.cwd() } = {}) { const values = {}; let help = false; for (let index = 0; index < argv.length; index += 1) { const argument = String(argv[index] || ''); if (argument === '--help' || argument === '-h') { help = true; continue; } if (argument === '--dry-run') { values.dryRun = true; continue; } const inline = argument.match(/^--([a-z-]+)=(.*)$/); const name = inline?.[1] || argument.match(/^--([a-z-]+)$/)?.[1] || ''; if (!['snapshot', 'data-dir', 'user', 'source', 'checkpoint'].includes(name)) { throw new VestaImportCliError('usage', '导入参数不正确。'); } const value = inline ? inline[2] : argv[++index]; if (value === undefined || String(value).trim() === '') { throw new VestaImportCliError('usage', '导入参数缺少值。'); } values[toCamelCase(name)] = String(value).trim(); } if (help) return { help: true }; const snapshot = values.snapshot ? path.resolve(cwd, values.snapshot) : ''; const dataDir = path.resolve(cwd, values.dataDir || env.DATA_DIR || 'data'); const user = String(values.user || '').trim(); const source = normalizeImportSource(values.source); const checkpoint = path.resolve( cwd, values.checkpoint || path.join(dataDir, 'vesta-maildir-import.checkpoint.json') ); if (!snapshot || !user || !source) { throw new VestaImportCliError('usage', '必须指定快照、目标用户和导入来源。'); } return { help: false, snapshot, dataDir, user, source, checkpoint, dryRun: Boolean(values.dryRun) }; } export function loadMailHubEnvironment({ env = process.env, cwd = process.cwd() } = {}) { const file = path.join(cwd, '.env'); if (!existsSync(file)) return env; for (const line of readFileSync(file, 'utf8').split(/\r?\n/)) { const clean = line.trim(); if (!clean || clean.startsWith('#')) continue; const separator = clean.indexOf('='); if (separator === -1) continue; const key = clean.slice(0, separator).trim(); let value = clean.slice(separator + 1).trim(); if (!key || key in env) continue; if ((value.startsWith('"') && value.endsWith('"')) || (value.startsWith("'") && value.endsWith("'"))) { value = value.slice(1, -1); } env[key] = value; } return env; } export async function runVestaMaildirImport(options, { env = process.env, output = process.stdout, clock = Date.now, dbApi = defaultDbApi, inspectTarget = inspectMailHubTarget, readSnapshot = readVestaSnapshotMetadata, importSnapshot = importVestaSnapshot, hashSupported = isLegacyPasswordHash, signal = null } = {}) { const normalized = normalizeRunOptions(options); const inventory = inspectTarget({ dataDir: normalized.dataDir, user: normalized.user }); const snapshot = await readSnapshot({ root: normalized.snapshot }); const defaults = domainDefaults(env); const checkpointIdentity = { version: checkpointVersion, source: normalized.source, targetUserId: inventory.user.id, snapshotId: snapshotIdentifier(normalized.snapshot, snapshot) }; const previousCheckpoint = normalized.dryRun ? null : await readImportCheckpoint(normalized.checkpoint, checkpointIdentity); const preflight = preflightVestaImport(snapshot, inventory, { hashSupported, defaults, allowExistingMailboxes: Boolean(previousCheckpoint) }); writeCountLine(output, 'preflight', { domains: preflight.domains, new_domains: preflight.newDomains, mailboxes: preflight.mailboxes, existing_mailboxes: preflight.existingMailboxes, hashes: preflight.hashes, supported_hashes: preflight.supportedHashes, unsupported_hashes: preflight.unsupportedHashes, missing_hashes: preflight.missingHashes, conflicts: preflight.conflicts, invalid_defaults: preflight.invalidDefaults, warnings: preflight.warningCount }); if (preflight.conflicts || preflight.unsupportedHashes || preflight.missingHashes || preflight.invalidDefaults) { throw new VestaImportCliError('preflight', 'Vesta 导入预检失败。'); } if (normalized.dryRun) { const report = await importSnapshot({ root: normalized.snapshot, dryRun: true, signal }); const summary = sanitizeImportReport(report); writeCompletionLine(output, summary); return { preflight, summary, checkpoint: null }; } dbApi.initDatabase(normalized.dataDir, mailHubSecret(env)); const targetUser = resolveDbUser(dbApi, normalized.user); if (!targetUser || targetUser.id !== inventory.user.id || targetUser.status === 'disabled') { throw new VestaImportCliError('target', '目标用户不可用。'); } // Always rescan the snapshot and let the database source key deduplicate it. // A file checkpoint cannot prove that earlier rows survived a DB rollback. const resumeCheckpoint = previousCheckpoint ? { ...previousCheckpoint, lastSourceKey: '', processed: 0, bytes: 0, complete: false } : null; const progress = { imported: 0, skipped: 0, processed: Number(resumeCheckpoint?.processed || 0), bytes: Number(resumeCheckpoint?.bytes || 0) }; const checkpointWriter = createAtomicCheckpointWriter({ filePath: normalized.checkpoint, identity: checkpointIdentity, initial: { ...(resumeCheckpoint || {}), warningCount: preflight.warningCount }, clock }); await checkpointWriter.flush({ force: true, complete: false }); const adapter = createMailHubImportAdapter({ dbApi, user: targetUser, source: normalized.source, defaults, progress }); let report; try { report = await importSnapshot({ root: normalized.snapshot, adapter, resumeAfterSourceKey: '', signal, onCheckpoint: async (sourceKey, context) => { progress.processed += 1; progress.bytes += Number(context?.message?.size || context?.message?.rawMessageBytes?.length || 0); const flushed = await checkpointWriter.record(sourceKey, progress); if (flushed) { writeCountLine(output, 'progress', { processed: progress.processed, imported: progress.imported, skipped: progress.skipped, bytes: progress.bytes }); } } }); } catch (error) { await checkpointWriter.flush({ force: true, complete: false }); throw error; } const summary = sanitizeImportReport(report); await checkpointWriter.flush({ force: true, complete: true, report: summary, fallbackSourceKey: report.lastSourceKey || resumeCheckpoint?.lastSourceKey || '' }); writeCompletionLine(output, summary); return { preflight, summary, checkpoint: checkpointWriter.snapshot() }; } export function inspectMailHubTarget({ dataDir, user }) { const databaseFile = path.join(path.resolve(dataDir), 'mailhub.sqlite'); if (!existsSync(databaseFile)) throw new VestaImportCliError('target', '目标数据库不存在。'); let database; try { database = new DatabaseSync(databaseFile, { readOnly: true }); const target = parseTargetUser(user); const userRow = target.id ? database.prepare('SELECT id, username, email, status FROM users WHERE id = ?').get(target.id) : database.prepare('SELECT id, username, email, status FROM users WHERE lower(username) = ? OR lower(email) = ?').get(target.login, target.login); if (!userRow || userRow.status === 'disabled') throw new VestaImportCliError('target', '目标用户不可用。'); const domains = database .prepare('SELECT id, user_id, lower(domain) AS domain FROM domains') .all() .map((row) => ({ id: Number(row.id), userId: Number(row.user_id), domain: row.domain })); const mailboxes = database .prepare(` SELECT m.id, m.user_id, lower(m.address) AS address, lower(d.domain) AS domain, m.deleted_at FROM inbound_mailboxes m JOIN domains d ON d.id = m.domain_id `) .all() .map((row) => ({ id: Number(row.id), userId: Number(row.user_id), address: row.address, domain: row.domain, deletedAt: row.deleted_at || null })); return { user: { id: Number(userRow.id), username: userRow.username, email: userRow.email, status: userRow.status }, domains, mailboxes }; } catch (error) { if (error instanceof VestaImportCliError) throw error; throw new VestaImportCliError('target', '目标数据库无法读取。'); } finally { database?.close(); } } export function preflightVestaImport(snapshot, inventory, { hashSupported = isLegacyPasswordHash, defaults = {}, allowExistingMailboxes = false } = {}) { const targetUserId = Number(inventory.user.id); const existingDomains = new Map(inventory.domains.map((domain) => [domain.domain, domain])); const existingMailboxes = new Map(inventory.mailboxes.map((mailbox) => [mailbox.address, mailbox])); const sourceDomains = new Set(); const sourceMailboxes = new Set(); let conflicts = 0; let newDomains = 0; let hashes = 0; let supportedHashes = 0; let unsupportedHashes = 0; let missingHashes = 0; let existingMailboxCount = 0; for (const domain of snapshot.domains) { const name = normalizeDomainName(domain.domain || domain.name); if (!name || sourceDomains.has(name)) conflicts += 1; else sourceDomains.add(name); const existing = existingDomains.get(name); if (existing && existing.userId !== targetUserId) conflicts += 1; if (!existing) newDomains += 1; } for (const mailbox of snapshot.mailboxes) { const address = normalizeMailboxAddress(mailbox.address); if (!address) { conflicts += 1; } else { if (sourceMailboxes.has(address)) conflicts += 1; else sourceMailboxes.add(address); const mailboxDomain = address.slice(address.lastIndexOf('@') + 1); if (!sourceDomains.has(mailboxDomain)) conflicts += 1; const existing = existingMailboxes.get(address); if (existing) { existingMailboxCount += 1; if ( existing.userId !== targetUserId || existing.deletedAt || !allowExistingMailboxes ) conflicts += 1; } } const passwordHash = String(mailbox.passwordHash || mailbox.legacyPasswordHash || '').trim(); if (!passwordHash) { missingHashes += 1; continue; } hashes += 1; if (hashSupported(passwordHash)) supportedHashes += 1; else unsupportedHashes += 1; } return { domains: snapshot.domains.length, newDomains, mailboxes: snapshot.mailboxes.length, existingMailboxes: existingMailboxCount, hashes, supportedHashes, unsupportedHashes, missingHashes, conflicts, invalidDefaults: newDomains > 0 && !validNewDomainDefaults(defaults) ? 1 : 0, warningCount: Array.isArray(snapshot.warnings) ? snapshot.warnings.length : 0 }; } export function createMailHubImportAdapter({ dbApi, user, source, defaults, progress }) { return { async ensureDomain(sourceDomain) { const domainName = normalizeDomainName(sourceDomain.domain || sourceDomain.name); const catchAllAddress = String(sourceDomain.catchAllAddress ?? sourceDomain.catchAll ?? '').trim(); const existing = dbApi.getDomainByName(domainName); if (existing) { if (Number(existing.userId) !== Number(user.id)) { throw new VestaImportCliError('conflict', '现有域名归属冲突。'); } return dbApi.updateDomain(existing.id, user.id, { catchAllAddress }); } if (!validNewDomainDefaults(defaults)) { throw new VestaImportCliError('configuration', '新域名默认配置不可用。'); } const keys = createDkimKeyPair(); const created = dbApi.createDomain(user.id, { domain: domainName, selector: defaultSelector(), verificationToken: crypto.randomBytes(18).toString('hex'), dkimPublic: keys.publicKey, dkimPrivate: keys.privateKey, senderHost: defaults.senderHost, sendingIp: defaults.sendingIp, spfExtra: defaults.spfExtra, dmarcPolicy: defaults.dmarcPolicy, dmarcRua: defaults.dmarcRua }); return dbApi.updateDomain(created.id, user.id, { catchAllAddress }); }, async ensureMailbox(mailbox) { return dbApi.upsertImportedInboundMailbox(user.id, { address: mailbox.address, displayName: mailbox.displayName, passwordHash: mailbox.passwordHash || mailbox.legacyPasswordHash || '', aliases: mailbox.aliases, forwardTo: mailbox.forwardTo, keepForwarded: mailbox.keepForwarded ?? !mailbox.forwardOnly, quotaMb: mailbox.quotaMb, status: mailbox.suspended || mailbox.status === 'suspended' ? 'disabled' : 'active' }); }, async ensureFolder(mailbox, folder) { return dbApi.createInboundFolder(mailbox, folder); }, async hasMessage(sourceKey) { const exists = dbApi.hasImportedInboundMessage(source, sourceKey); if (exists) progress.skipped += 1; return exists; }, async createMessage(message, context) { const result = dbApi.createImportedInboundMessage(context.mailbox, { ...message, importSource: source, sourceKey: message.sourceKey }); if (result.created) progress.imported += 1; else progress.skipped += 1; return result; } }; } export function createAtomicCheckpointWriter({ filePath, identity, initial = null, clock = Date.now, messageInterval = checkpointMessageInterval, timeIntervalMs = checkpointTimeIntervalMs }) { let pending = 0; let lastFlushAt = clock(); let state = { ...identity, lastSourceKey: String(initial?.lastSourceKey || ''), processed: Number(initial?.processed || 0), bytes: Number(initial?.bytes || 0), complete: Boolean(initial?.complete), warningCount: Number(initial?.warningCount || 0) }; async function flush({ force = false, complete = state.complete, report = null, fallbackSourceKey = '' } = {}) { if (!force && pending < messageInterval && clock() - lastFlushAt < timeIntervalMs) return false; if (!pending && !force) return false; state = { ...state, lastSourceKey: state.lastSourceKey || String(fallbackSourceKey || ''), complete: Boolean(complete), warningCount: Number(report?.warningCount ?? state.warningCount ?? 0), updatedAt: new Date(clock()).toISOString() }; await atomicWriteJson(filePath, state); pending = 0; lastFlushAt = clock(); return true; } return { async record(sourceKey, progress) { state = { ...state, lastSourceKey: String(sourceKey || state.lastSourceKey || ''), processed: Number(progress.processed || 0), bytes: Number(progress.bytes || 0), complete: false }; pending += 1; if (pending < messageInterval && clock() - lastFlushAt < timeIntervalMs) return false; return await flush(); }, flush, snapshot() { return { ...state }; } }; } export async function readImportCheckpoint(filePath, identity) { let checkpoint; try { checkpoint = JSON.parse(await readFile(filePath, 'utf8')); } catch (error) { if (error?.code === 'ENOENT') return null; throw new VestaImportCliError('checkpoint', '导入断点无法读取。'); } if ( checkpoint?.version !== identity.version || checkpoint?.source !== identity.source || Number(checkpoint?.targetUserId) !== Number(identity.targetUserId) || checkpoint?.snapshotId !== identity.snapshotId || typeof checkpoint?.lastSourceKey !== 'string' ) { throw new VestaImportCliError('checkpoint', '导入断点与本次任务不匹配。'); } return checkpoint; } export function sanitizeImportReport(report) { return { dryRun: Boolean(report?.dryRun), domains: Number(report?.domains || 0), mailboxes: Number(report?.mailboxes || 0), messages: Number(report?.messages || 0), bytes: Number(report?.bytes || 0), plannedMessages: Number(report?.plannedMessages || 0), importedMessages: Number(report?.importedMessages || 0), skippedMessages: Number(report?.skippedMessages || 0), resumeSkippedMessages: Number(report?.resumeSkippedMessages || 0), warningCount: Number(report?.warningCount ?? (Array.isArray(report?.warnings) ? report.warnings.length : 0)) }; } export function usageText() { return [ 'Usage: node scripts/import-vesta-maildir.js --snapshot PATH --data-dir PATH --user USER --source NAME [--checkpoint FILE] [--dry-run]', '', 'USER accepts a username/email, or an explicit numeric id in the form id:123.', 'Run --dry-run first. The checkpoint is written only during a real import.' ].join('\n'); } async function main() { let options; try { loadMailHubEnvironment(); options = parseVestaImportArguments(process.argv.slice(2)); if (options.help) { process.stdout.write(`${usageText()}\n`); return; } const controller = new AbortController(); const abort = () => controller.abort(new Error('Vesta 导入已取消。')); process.once('SIGINT', abort); process.once('SIGTERM', abort); try { await runVestaMaildirImport(options, { signal: controller.signal }); } finally { process.off('SIGINT', abort); process.off('SIGTERM', abort); } } catch (error) { const category = safeCategory(error?.category); process.stderr.write(`failed errors=1 category=${category}\n`); process.exitCode = 1; } } function normalizeRunOptions(options = {}) { const source = normalizeImportSource(options.source); const snapshot = path.resolve(String(options.snapshot || '')); const dataDir = path.resolve(String(options.dataDir || '')); const checkpoint = path.resolve(String(options.checkpoint || path.join(dataDir, 'vesta-maildir-import.checkpoint.json'))); const user = String(options.user || '').trim(); if (!source || !user || !options.snapshot || !options.dataDir) { throw new VestaImportCliError('usage', '导入参数不完整。'); } return { source, snapshot, dataDir, checkpoint, user, dryRun: Boolean(options.dryRun) }; } function resolveDbUser(dbApi, identifier) { const target = parseTargetUser(identifier); return target.id ? dbApi.getUser(target.id) : dbApi.getUserByLogin(target.login); } function parseTargetUser(value) { const clean = String(value || '').trim().toLowerCase(); const explicitId = clean.match(/^id:(\d+)$/); return explicitId ? { id: Number(explicitId[1]), login: '' } : { id: 0, login: clean }; } function domainDefaults(env) { const policy = String(env.DMARC_POLICY || 'none').trim().toLowerCase(); return { senderHost: String(env.MAIL_HOSTNAME || 'mailhub.local').trim().toLowerCase(), sendingIp: String(env.SENDING_IP || '').trim(), spfExtra: String(env.DEFAULT_SPF_MECHANISMS || 'include:spf.mailjet.com').trim(), dmarcPolicy: ['none', 'quarantine', 'reject'].includes(policy) ? policy : 'none', dmarcRua: String(env.DMARC_RUA || '').trim() }; } function validNewDomainDefaults(defaults) { return validHostname(defaults.senderHost) && Boolean(String(defaults.sendingIp || '').trim()); } function validHostname(value) { const hostname = String(value || '').trim().toLowerCase().replace(/\.$/, ''); if (!hostname || hostname.length > 253) return false; return hostname.split('.').every((label) => ( label.length > 0 && label.length <= 63 && /^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$/.test(label) )); } function mailHubSecret(env) { if (env.SESSION_SECRET) return String(env.SESSION_SECRET); return crypto .createHash('sha256') .update(`${env.ADMIN_PASSWORD || 'change-this-admin-password'}:${env.API_TOKEN || ''}`) .digest('hex'); } function defaultSelector() { const date = new Date(); return `mh${date.getUTCFullYear()}${String(date.getUTCMonth() + 1).padStart(2, '0')}`; } function snapshotIdentifier(snapshotPath, snapshot = {}) { const fingerprint = { path: path.resolve(snapshotPath), domains: Array.isArray(snapshot.domains) ? snapshot.domains : [], mailboxes: Array.isArray(snapshot.mailboxes) ? snapshot.mailboxes : [], warnings: Array.isArray(snapshot.warnings) ? snapshot.warnings : [] }; return crypto .createHash('sha256') .update(JSON.stringify(sortJsonValue(fingerprint))) .digest('hex'); } function sortJsonValue(value) { if (Array.isArray(value)) return value.map(sortJsonValue); if (!value || typeof value !== 'object') return value; return Object.fromEntries( Object.keys(value) .sort() .map((key) => [key, sortJsonValue(value[key])]) ); } function normalizeImportSource(value) { const source = String(value || '').trim().toLowerCase(); if (!source || source.length > 120 || !/^[a-z0-9][a-z0-9._:@/-]*$/.test(source)) return ''; return source; } function normalizeDomainName(value) { const domain = String(value || '').trim().toLowerCase(); return /^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$/.test(domain) ? domain : ''; } function normalizeMailboxAddress(value) { const address = String(value || '').trim().toLowerCase(); return /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(address) ? address : ''; } function toCamelCase(value) { return String(value).replace(/-([a-z])/g, (_match, letter) => letter.toUpperCase()); } function writeCompletionLine(output, summary) { writeCountLine(output, 'complete', { dry_run: summary.dryRun ? 1 : 0, domains: summary.domains, mailboxes: summary.mailboxes, messages: summary.messages, bytes: summary.bytes, planned: summary.plannedMessages, imported: summary.importedMessages, skipped: summary.skippedMessages, resume_skipped: summary.resumeSkippedMessages, warnings: summary.warningCount }); } function writeCountLine(output, event, values) { const fields = Object.entries(values).map(([key, value]) => `${key}=${Number(value || 0)}`); output.write(`${event} ${fields.join(' ')}\n`); } function safeCategory(value) { const clean = String(value || 'import').trim().toLowerCase(); return /^[a-z][a-z0-9_-]{0,31}$/.test(clean) ? clean : 'import'; } async function atomicWriteJson(filePath, value) { const directory = path.dirname(filePath); await mkdir(directory, { recursive: true }); const temporary = path.join( directory, `.${path.basename(filePath)}.${process.pid}.${crypto.randomBytes(6).toString('hex')}.tmp` ); try { await writeFile(temporary, `${JSON.stringify(value)}\n`, { encoding: 'utf8', mode: 0o600, flag: 'wx' }); await rename(temporary, filePath); } catch (error) { await rm(temporary, { force: true }); throw error; } } const invokedUrl = process.argv[1] ? pathToFileURL(path.resolve(process.argv[1])).href : ''; if (import.meta.url === invokedUrl) await main();