diff --git a/server/src/db/repositories/MailIngestCandidates.repository.ts b/server/src/db/repositories/MailIngestCandidates.repository.ts index e3247580d..6bc4be94c 100644 --- a/server/src/db/repositories/MailIngestCandidates.repository.ts +++ b/server/src/db/repositories/MailIngestCandidates.repository.ts @@ -88,7 +88,7 @@ interface MailIngestTripPlaceRow { } /** A trip as the inbox names it to a person who can see it. */ -export interface MailIngestTripRefRow { +interface MailIngestTripRefRow { id: number; title: string; start_date: string | null; diff --git a/server/src/nest/common/validate-route-guards.ts b/server/src/nest/common/validate-route-guards.ts index c5370c933..df238428b 100644 --- a/server/src/nest/common/validate-route-guards.ts +++ b/server/src/nest/common/validate-route-guards.ts @@ -132,6 +132,10 @@ export const PUBLIC_ROUTE_ALLOW_LIST: string[] = [ 'JourneyPublicController.legacyPhoto', 'JourneyPublicController.mapLayers', 'JourneyPublicController.photo', + // The booking mail webhook. A mail service holds no session; the HMAC over + // the raw mail is the credential, and the route is a 404 unless the admin + // made the webhook the import's transport and set its secret. + 'MailIngestInboundController.receive', // The MCP transport — bearer tokens are verified inside the handler. // The same documents under the resource URL, for clients that append the // well-known path to the server address they were handed. diff --git a/server/src/nest/mail-ingest/mail-ingest-imap.client.ts b/server/src/nest/mail-ingest/mail-ingest-imap.client.ts new file mode 100644 index 000000000..475b8d78a --- /dev/null +++ b/server/src/nest/mail-ingest/mail-ingest-imap.client.ts @@ -0,0 +1,126 @@ +import { checkSsrf, SsrfBlockedError } from '../../utils/ssrfGuard'; +import type { MailIngestImapConfig } from './mail-ingest-settings.service'; +import { Injectable } from '@nestjs/common'; + +import { ImapFlow } from 'imapflow'; +import { isIP } from 'node:net'; + +/** How long connecting and the server's greeting may take, and how long the socket may idle after. */ +const CONNECT_TIMEOUT_MS = 15_000; +const SOCKET_TIMEOUT_MS = 60_000; + +/** A mail waiting in the box: its uid and its size, before anything of it is downloaded. */ +export interface ImapMailRef { + uid: number; + size: number; +} + +/** What the import asks of an IMAP box, one connection per run. */ +export interface ImapSession { + /** Opens the folder; answers the uid validity, which changes when the server renumbers it. */ + open(mailbox: string): Promise<{ uidValidity: number }>; + /** The mails with a uid above `lastUid`, oldest first. */ + listAfter(lastUid: number): Promise; + /** One mail's raw bytes; null for a uid that is gone. */ + source(uid: number): Promise; + /** Makes the folder (made of these parts, in the server's own delimiter) unless it exists; answers its path. */ + ensureFolder(parts: readonly string[]): Promise; + move(uids: readonly number[], folder: string): Promise; + close(): Promise; +} + +/** + * Opens the admin's IMAP box. The host goes through the SSRF guard first, + * with the same allowance for a box on the own network as every other + * address an admin configures (ALLOW_INTERNAL_NETWORK), and the connection + * goes to the address the guard vetted, the certificate still checked + * against the host's name, so a name that resolves differently a moment + * later cannot redirect it. + */ +@Injectable() +export class MailIngestImapClient { + async connect(config: MailIngestImapConfig): Promise { + const address = await vettedAddress(config.host); + const client = new ImapFlow({ + host: address, + port: config.port, + secure: config.tls === 'tls', + doSTARTTLS: config.tls === 'starttls' ? true : undefined, + servername: isIP(config.host) ? undefined : config.host, + auth: { user: config.user, pass: config.pass }, + logger: false, + disableAutoIdle: true, + connectionTimeout: CONNECT_TIMEOUT_MS, + greetingTimeout: CONNECT_TIMEOUT_MS, + socketTimeout: SOCKET_TIMEOUT_MS, + }); + await client.connect(); + return new ImapFlowSession(client); + } +} + +/** The address a connection to `host` may use, or a refusal naming why. */ +async function vettedAddress(host: string): Promise { + const verdict = await checkSsrf(`https://${isIP(host) === 6 ? `[${host}]` : host}/`); + if (!verdict.allowed || !verdict.resolvedIp) { + throw new SsrfBlockedError(verdict.error ?? 'This IMAP host is not allowed'); + } + return verdict.resolvedIp; +} + +class ImapFlowSession implements ImapSession { + constructor(private readonly client: ImapFlow) {} + + async open(mailbox: string): Promise<{ uidValidity: number }> { + const box = await this.client.mailboxOpen(mailbox); + return { uidValidity: Number(box.uidValidity) }; + } + + async listAfter(lastUid: number): Promise { + const found: ImapMailRef[] = []; + // `n:*` always names the newest mail, even below n, so the range alone does not filter. + for await (const message of this.client.fetch(`${lastUid + 1}:*`, { uid: true, size: true }, { uid: true })) { + if (message.uid > lastUid) found.push({ uid: message.uid, size: message.size ?? 0 }); + } + return found.sort((a, b) => a.uid - b.uid); + } + + async source(uid: number): Promise { + const message = await this.client.fetchOne(String(uid), { source: true }, { uid: true }); + return message && Buffer.isBuffer(message.source) ? message.source : null; + } + + async ensureFolder(parts: readonly string[]): Promise { + const created = await this.client.mailboxCreate([...parts]); + return created.path; + } + + async move(uids: readonly number[], folder: string): Promise { + if (uids.length === 0) return; + await this.client.messageMove(uids.join(','), folder, { uid: true }); + } + + async close(): Promise { + try { + await this.client.logout(); + } catch { + // The connection broke on its own; there is nothing left to say goodbye to. + this.client.close(); + } + } +} + +/** What an admin reads when the box cannot be read: the reason, never the password. */ +export function describeImapFailure(err: unknown): string { + if (err instanceof SsrfBlockedError) return err.message; + const failure = (err ?? {}) as { + authenticationFailed?: boolean; + code?: string; + responseText?: string; + message?: string; + }; + if (failure.authenticationFailed) return 'The IMAP server refused the user name or password'; + if (failure.code && /^E[A-Z]+$/.test(failure.code)) return `Could not reach the IMAP server (${failure.code})`; + const said = failure.responseText || failure.message || String(err); + return `The IMAP server answered: ${said}`.slice(0, 300); +} diff --git a/server/src/nest/mail-ingest/mail-ingest-imap.job.ts b/server/src/nest/mail-ingest/mail-ingest-imap.job.ts new file mode 100644 index 000000000..9a50e0e8e --- /dev/null +++ b/server/src/nest/mail-ingest/mail-ingest-imap.job.ts @@ -0,0 +1,106 @@ +import { MailIngestCursor } from '../../db/entities/MailIngestCursor.entity'; +import type { MailIngestCursorRepository } from '../../db/repositories/MailIngestCursor.repository'; +import { dbNow } from '../../db/types'; +import { logError, logWarn } from '../audit/audit-log.logger'; +import { CronRegistrarService, LEASE_OWNER } from '../scheduling/cron-registrar.service'; +import { + describeImapFailure, + MailIngestImapClient, + type ImapMailRef, + type ImapSession, +} from './mail-ingest-imap.client'; +import { MailIngestIntakeService } from './mail-ingest-intake.service'; +import { MailIngestSettingsService, type MailIngestImapConfig } from './mail-ingest-settings.service'; +import { InjectRepository } from '@mikro-orm/nestjs'; +import { Injectable, type OnApplicationBootstrap } from '@nestjs/common'; + +/** How long a run holds the box before another process may take it over. */ +const LEASE_MS = 10 * 60 * 1000; +/** The mails one run takes from the box; the rest wait for the next run. */ +const MAILS_PER_RUN = 20; +/** Where read mails go: kept, never deleted, out of the folder the next run reads. */ +const PROCESSED_FOLDER = ['TREK', 'processed'] as const; + +/** + * Reads the admin's IMAP box every two minutes, with a client of its own for + * every run. Each new mail is handed to the intake, which stores it once, + * and then moved to TREK/processed. The cursor (uid validity, last uid) + * keeps a mail the move missed from being read twice, and the intake's + * dedupe keeps it so when the server renumbers the box and the run starts + * over. One process holds the box at a time, by the cursor's lease. + */ +@Injectable() +export class MailIngestImapJob implements OnApplicationBootstrap { + constructor( + @InjectRepository(MailIngestCursor) private readonly cursor: MailIngestCursorRepository, + private readonly settings: MailIngestSettingsService, + private readonly intake: MailIngestIntakeService, + private readonly imap: MailIngestImapClient, + private readonly registrar: CronRegistrarService, + ) {} + + onApplicationBootstrap(): void { + if (!this.registrar.isEnabled()) return; + this.registrar.register('mail-ingest-imap', '*/2 * * * *', () => this.tick()); + } + + /** + * One run; answers how many mails it took from the box. Nothing happens + * unless the import is on, IMAP is its transport and the box is complete. + * + * @txIndependent each mail is stored on its own as it is read; the cursor + * notes where the run got to once the box is done with. + */ + async tick(now: Date = new Date()): Promise { + if (!(await this.settings.isAvailable()) || (await this.settings.transport()) !== 'imap') return 0; + const config = await this.settings.imapConfig(); + if (!config) return 0; + const until = dbNow(new Date(now.getTime() + LEASE_MS)); + if (!(await this.cursor.acquireLease('imap', config.mailbox, LEASE_OWNER, until, dbNow(now)))) return 0; + try { + const { taken, uidValidity, lastUid } = await this.poll(config); + await this.cursor.recordOk('imap', config.mailbox, dbNow(new Date()), { uidValidity, lastUid }); + return taken; + } catch (err) { + const reason = describeImapFailure(err); + logError(`[mail-ingest] Reading the IMAP box failed: ${reason}`); + await this.cursor.recordError('imap', config.mailbox, reason); + return 0; + } finally { + await this.cursor.releaseLease('imap', config.mailbox, LEASE_OWNER); + } + } + + private async poll(config: MailIngestImapConfig): Promise<{ taken: number; uidValidity: number; lastUid: number }> { + const session = await this.imap.connect(config); + try { + const { uidValidity } = await session.open(config.mailbox); + const position = await this.cursor.find('imap', config.mailbox); + // A changed uid validity means the server renumbered the box: start over, the dedupe keeps each mail once. + const after = position?.uid_validity === uidValidity ? (position.last_uid ?? 0) : 0; + const due = (await session.listAfter(after)).slice(0, MAILS_PER_RUN); + if (due.length === 0) return { taken: 0, uidValidity, lastUid: after }; + const folder = await session.ensureFolder(PROCESSED_FOLDER); + await due.reduce(async (previous, mail) => { + await previous; + await this.takeIn(session, mail); + }, Promise.resolve()); + await session.move(due.map((mail) => mail.uid), folder); + return { taken: due.length, uidValidity, lastUid: Math.max(...due.map((mail) => mail.uid)) }; + } finally { + await session.close(); + } + } + + /** One mail into the intake. One larger than the import takes is left unread, and moved on with the rest. */ + private async takeIn(session: ImapSession, mail: ImapMailRef): Promise { + if (mail.size > this.settings.maxMailBytes()) { + logWarn(`[mail-ingest] An IMAP mail of ${mail.size} bytes is larger than the import takes; left unread`); + return; + } + const raw = await session.source(mail.uid); + if (!raw) return; + const result = await this.intake.receive(raw, 'imap'); + if (result.status === 'refused') logWarn(`[mail-ingest] An IMAP mail was not taken in: ${result.reason}`); + } +} diff --git a/server/src/nest/mail-ingest/mail-ingest-inbound.controller.ts b/server/src/nest/mail-ingest/mail-ingest-inbound.controller.ts new file mode 100644 index 000000000..76b9b4cfa --- /dev/null +++ b/server/src/nest/mail-ingest/mail-ingest-inbound.controller.ts @@ -0,0 +1,65 @@ +import { Public } from '../auth-core/public.decorator'; +import { DomainError } from '../common/domain-error'; +import { ResponseContract } from '../common/response-contract'; +import { + MailIngestWebhookService, + WEBHOOK_RECIPIENT_HEADER, + WEBHOOK_SIGNATURE_HEADER, + WEBHOOK_TIMESTAMP_HEADER, +} from './mail-ingest-webhook.service'; +import { Controller, HttpCode, Post, Req } from '@nestjs/common'; +import { mailIngestInboundResponseSchema, type MailIngestInboundResponse } from '@trek/shared'; + +import type { Request } from 'express'; + +const TOO_LARGE = 'This mail is larger than the import takes'; + +/** + * `POST /api/mail-ingest/inbound`: the signed webhook a mail service posts + * every mail to, raw (message/rfc822). It does not exist (404) unless the + * import is on, the webhook is its transport and a secret is set; then the + * body is read up to the import's limit, and the signature over it decides. + * The JSON parsers leave a message/rfc822 body alone, so this route reads + * the stream itself and nothing of the mail reaches the request log. + */ +@Controller('api/mail-ingest/inbound') +export class MailIngestInboundController { + constructor(private readonly webhook: MailIngestWebhookService) {} + + @Post() + @Public( + 'A mail service cannot hold a TREK session; an HMAC signature over the raw mail, its recipient and a timestamp within five minutes is the credential, and the route answers 404 unless the webhook is the configured transport.', + ) + @HttpCode(200) + @ResponseContract(mailIngestInboundResponseSchema) + async receive(@Req() req: Request): Promise { + const secret = await this.webhook.secret(); + if (!req.is('message/rfc822')) { + throw new DomainError(415, 'Send the raw mail as message/rfc822'); + } + const raw = await readBody(req, this.webhook.maxBytes()); + return this.webhook.receive(secret, raw, { + timestamp: req.get(WEBHOOK_TIMESTAMP_HEADER), + signature: req.get(WEBHOOK_SIGNATURE_HEADER), + recipient: req.get(WEBHOOK_RECIPIENT_HEADER), + }); + } +} + +/** + * The request body, up to `maxBytes`. Past the limit the rest is drained + * unread, so the sender gets the 413 rather than a reset connection. + */ +function readBody(req: Request, maxBytes: number): Promise { + if (Number(req.get('content-length')) > maxBytes) return Promise.reject(new DomainError(413, TOO_LARGE)); + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let size = 0; + req.on('data', (chunk: Buffer) => { + size += chunk.length; + if (size <= maxBytes) chunks.push(chunk); + }); + req.on('end', () => (size > maxBytes ? reject(new DomainError(413, TOO_LARGE)) : resolve(Buffer.concat(chunks)))); + req.on('error', reject); + }); +} diff --git a/server/src/nest/mail-ingest/mail-ingest-relay.job.ts b/server/src/nest/mail-ingest/mail-ingest-relay.job.ts new file mode 100644 index 000000000..296b692dc --- /dev/null +++ b/server/src/nest/mail-ingest/mail-ingest-relay.job.ts @@ -0,0 +1,108 @@ +import { MailIngestCursor } from '../../db/entities/MailIngestCursor.entity'; +import type { MailIngestCursorRepository } from '../../db/repositories/MailIngestCursor.repository'; +import { dbNow } from '../../db/types'; +import { logError, logWarn } from '../audit/audit-log.logger'; +import { CronRegistrarService, LEASE_OWNER } from '../scheduling/cron-registrar.service'; +import { MailIngestIntakeService } from './mail-ingest-intake.service'; +import { MailIngestRelayHost, type MailInboxRelay } from './mail-ingest-relay.service'; +import { MailIngestSettingsService } from './mail-ingest-settings.service'; +import { MAIL_INGEST_LIMITS } from './mail-ingest.constants'; +import { InjectRepository } from '@mikro-orm/nestjs'; +import { Injectable, type OnApplicationBootstrap } from '@nestjs/common'; + +import { createHash } from 'node:crypto'; +import { z } from 'zod'; + +/** How long a run holds the relay before another process may take it over. */ +const LEASE_MS = 10 * 60 * 1000; +/** The mails one run asks the relay for. */ +const PAGE = 10; +/** The base64 length of the largest mail the relay hands on. */ +const MAX_RAW_CHARS = Math.ceil(MAIL_INGEST_LIMITS.cloudMailBytes / 3) * 4; + +/** The relay's answer, checked before anything of it is used. */ +const relayInboxSchema = z.object({ + mails: z + .array( + z.object({ + id: z.string().regex(/^[A-Za-z0-9_-]{1,64}$/), + /** The addresses it was sent to, as the relay's mail server took them (RCPT TO). */ + recipients: z.array(z.string().min(3).max(320)).min(1).max(5), + sha256: z.string().regex(/^[0-9a-f]{64}$/), + /** The mail itself, base64. */ + raw: z.string().max(MAX_RAW_CHARS), + }), + ) + .max(PAGE), +}); +type RelayedMail = z.infer['mails'][number]; + +/** + * Collects this instance's forwarded mails from the TREK Cloud relay every + * two minutes: a page of at most ten, each checked against its hash, handed + * to the intake under the address it was sent to, and acknowledged once it + * is in (or once it is clear it never will be), so the relay drops it. A + * mail that does not arrive whole is left for the next run. Idle wherever + * no relay is attached, which is every self-hosted install. + */ +@Injectable() +export class MailIngestRelayJob implements OnApplicationBootstrap { + constructor( + @InjectRepository(MailIngestCursor) private readonly cursor: MailIngestCursorRepository, + private readonly settings: MailIngestSettingsService, + private readonly intake: MailIngestIntakeService, + private readonly relays: MailIngestRelayHost, + private readonly registrar: CronRegistrarService, + ) {} + + onApplicationBootstrap(): void { + if (!this.registrar.isEnabled()) return; + this.registrar.register('mail-ingest-relay', '*/2 * * * *', () => this.tick()); + } + + /** + * One run; answers how many mails it took in. + * + * @txIndependent each mail is stored on its own and acknowledged after it, network I/O between them. + */ + async tick(now: Date = new Date()): Promise { + const relay = this.relays.current(); + if (!relay || !(await this.settings.isAvailable()) || (await this.settings.transport()) !== 'cloud') return 0; + const until = dbNow(new Date(now.getTime() + LEASE_MS)); + if (!(await this.cursor.acquireLease('cloud', '', LEASE_OWNER, until, dbNow(now)))) return 0; + try { + const page = relayInboxSchema.safeParse(await relay.list(PAGE)); + if (!page.success) throw new Error('The relay answered with an inbox this server cannot read'); + const taken = await page.data.mails.reduce(async (previous, mail) => { + const count = await previous; + return count + ((await this.takeIn(relay, mail)) ? 1 : 0); + }, Promise.resolve(0)); + await this.cursor.recordOk('cloud', '', dbNow(new Date())); + return taken; + } catch (err) { + const reason = err instanceof Error ? err.message : String(err); + logError(`[mail-ingest] Collecting mails from the relay failed: ${reason}`); + await this.cursor.recordError('cloud', '', reason); + return 0; + } finally { + await this.cursor.releaseLease('cloud', '', LEASE_OWNER); + } + } + + /** One relayed mail into the intake; answers whether it was taken in. */ + private async takeIn(relay: MailInboxRelay, mail: RelayedMail): Promise { + const raw = Buffer.from(mail.raw, 'base64'); + if (createHash('sha256').update(raw).digest('hex') !== mail.sha256) { + logWarn(`[mail-ingest] Relayed mail ${mail.id} did not arrive whole; left for the next run`); + return false; + } + const result = await this.intake.receive(raw, 'cloud', mail.recipients); + // Switched off in between: the relay keeps the mail until the import is back or the mail expires. + if (result.status === 'refused' && result.reason === 'unavailable') return false; + if (result.status === 'refused') { + logWarn(`[mail-ingest] Relayed mail ${mail.id} was not taken in: ${result.reason}`); + } + await relay.ack(mail.id); + return result.status === 'accepted'; + } +} diff --git a/server/src/nest/mail-ingest/mail-ingest-relay.service.ts b/server/src/nest/mail-ingest/mail-ingest-relay.service.ts new file mode 100644 index 000000000..5fa7f5a82 --- /dev/null +++ b/server/src/nest/mail-ingest/mail-ingest-relay.service.ts @@ -0,0 +1,33 @@ +import { Injectable } from '@nestjs/common'; + +/** + * The TREK Cloud control plane's queue of mails for this instance, as the + * managed install reaches it with its instance token. The control plane's + * mail server takes a forwarded mail for a contract's route key, keeps it + * encrypted, and hands it over here; nothing of a person reaches it. + */ +export interface MailInboxRelay { + /** `GET /v1/instance/inbox?limit=n`: the waiting mails, the answer's JSON as parsed, checked by the caller. */ + list(limit: number): Promise; + /** `POST /v1/instance/inbox/:id/ack`: the mail is taken in (or never will be), and the relay drops it. */ + ack(id: string): Promise; +} + +/** + * Where the managed install plugs its control-plane client in. This + * repository has none: a self-hosted install has no relay, and the pull job + * stays idle until a module of the managed build attaches one when it + * starts, through the client it already talks to the control plane with. + */ +@Injectable() +export class MailIngestRelayHost { + private relay: MailInboxRelay | null = null; + + attach(relay: MailInboxRelay): void { + this.relay = relay; + } + + current(): MailInboxRelay | null { + return this.relay; + } +} diff --git a/server/src/nest/mail-ingest/mail-ingest-settings.service.ts b/server/src/nest/mail-ingest/mail-ingest-settings.service.ts index a52cb0577..39f623d24 100644 --- a/server/src/nest/mail-ingest/mail-ingest-settings.service.ts +++ b/server/src/nest/mail-ingest/mail-ingest-settings.service.ts @@ -79,7 +79,9 @@ export class MailIngestSettingsService { const user = await this.pick(this.config.imapUser, 'mail_ingest_imap_user'); const pass = this.config.imapPass ?? decrypt_api_key(await this.settings.getAppSetting('mail_ingest_imap_pass')); if (!host || !user || !pass) return null; - const tls = mailIngestImapTlsSchema.catch('tls').parse(await this.pick(this.config.imapTls, 'mail_ingest_imap_tls')); + const tls = mailIngestImapTlsSchema + .catch('tls') + .parse(await this.pick(this.config.imapTls, 'mail_ingest_imap_tls')); const port = Number(await this.pick(this.config.imapPort, 'mail_ingest_imap_port')); const mailbox = await this.pick(this.config.imapMailbox, 'mail_ingest_imap_mailbox'); return { diff --git a/server/src/nest/mail-ingest/mail-ingest-status.service.ts b/server/src/nest/mail-ingest/mail-ingest-status.service.ts new file mode 100644 index 000000000..601218f70 --- /dev/null +++ b/server/src/nest/mail-ingest/mail-ingest-status.service.ts @@ -0,0 +1,115 @@ +import { MailIngestCandidates } from '../../db/entities/MailIngestCandidates.entity'; +import { MailIngestCursor } from '../../db/entities/MailIngestCursor.entity'; +import { MailIngestMessages } from '../../db/entities/MailIngestMessages.entity'; +import type { MailIngestCandidatesRepository } from '../../db/repositories/MailIngestCandidates.repository'; +import type { MailIngestCursorRepository } from '../../db/repositories/MailIngestCursor.repository'; +import type { MailIngestMessagesRepository } from '../../db/repositories/MailIngestMessages.repository'; +import { dbNow, utcSuffix } from '../../db/types'; +import { describeImapFailure, MailIngestImapClient } from './mail-ingest-imap.client'; +import { MailIngestRelayHost } from './mail-ingest-relay.service'; +import { MailIngestSettingsService } from './mail-ingest-settings.service'; +import { InjectRepository } from '@mikro-orm/nestjs'; +import { Injectable } from '@nestjs/common'; +import { + mailIngestMessageStatusSchema, + mailIngestReasonSchema, + type MailIngestAdminStatusResponse, + type MailIngestAdminTestResponse, + type MailIngestTransportSetting, +} from '@trek/shared'; + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** + * The admin's view of the import: how its transport runs and what became of + * the mails of the last day, in counts only. No subject, sender, address or + * content of any mail is part of it, and the backlog is a list of numbers, + * not of people. The connection test reaches the transport the way the + * import does and says what failed. + */ +@Injectable() +export class MailIngestStatusService { + constructor( + @InjectRepository(MailIngestMessages) private readonly messages: MailIngestMessagesRepository, + @InjectRepository(MailIngestCandidates) private readonly candidates: MailIngestCandidatesRepository, + @InjectRepository(MailIngestCursor) private readonly cursor: MailIngestCursorRepository, + private readonly settings: MailIngestSettingsService, + private readonly imap: MailIngestImapClient, + private readonly relays: MailIngestRelayHost, + ) {} + + async status(now: Date = new Date()): Promise { + const transport = await this.settings.transport(); + const reader = await this.readerCursor(transport); + const since = dbNow(new Date(now.getTime() - DAY_MS)); + const byStatus: MailIngestAdminStatusResponse['last_day']['by_status'] = {}; + const byReason: MailIngestAdminStatusResponse['last_day']['by_reason'] = {}; + for (const { status, reason, n } of await this.messages.tallySince(since)) { + const knownStatus = mailIngestMessageStatusSchema.safeParse(status); + if (knownStatus.success) byStatus[knownStatus.data] = (byStatus[knownStatus.data] ?? 0) + n; + const knownReason = mailIngestReasonSchema.safeParse(reason); + if (knownReason.success) byReason[knownReason.data] = (byReason[knownReason.data] ?? 0) + n; + } + const resolutions = new Map( + (await this.candidates.tallyResolutionsSince(since)).map(({ resolution, n }) => [resolution, n]), + ); + const startOfDay = dbNow(new Date(`${now.toISOString().slice(0, 10)}T00:00:00Z`)); + return { + transport, + last_ok_at: utcSuffix(reader?.last_ok_at), + last_error: reader?.last_error ?? null, + lease_until: utcSuffix(reader?.lease_until), + last_day: { by_status: byStatus, by_reason: byReason }, + auto_applied: resolutions.get('auto_applied') ?? 0, + undone: resolutions.get('undone') ?? 0, + ai_calls_today: await this.messages.countAllAiReadSince(startOfDay), + backlog: await this.candidates.pendingPerUser(), + }; + } + + /** Whether the configured transport can reach its mail now, and if not, why. */ + async test(): Promise { + const transport = await this.settings.transport(); + const error = await this.problemWith(transport); + return { transport, ok: error === null, error }; + } + + private async problemWith(transport: MailIngestTransportSetting): Promise { + switch (transport) { + case 'off': + return 'No transport is set; mails come in by upload only'; + case 'webhook': + return (await this.settings.webhookSecret()) ? null : 'No webhook secret is set'; + case 'cloud': + return this.relays.current() ? null : 'The TREK Cloud relay is not connected'; + case 'imap': + return this.imapProblem(); + } + } + + /** Opens the box and its folder the way a run does, and lets go again. */ + private async imapProblem(): Promise { + const config = await this.settings.imapConfig(); + if (!config) return 'The IMAP host, user or password is missing'; + try { + const session = await this.imap.connect(config); + try { + await session.open(config.mailbox); + } finally { + await session.close(); + } + return null; + } catch (err) { + return describeImapFailure(err); + } + } + + /** Where the reader of the transport stands: the IMAP box's or the relay's cursor, else the pipeline's own. */ + private async readerCursor(transport: MailIngestTransportSetting) { + if (transport === 'imap') { + const config = await this.settings.imapConfig(); + return config ? this.cursor.find('imap', config.mailbox) : null; + } + return this.cursor.find(transport === 'cloud' ? 'cloud' : 'pipeline', ''); + } +} diff --git a/server/src/nest/mail-ingest/mail-ingest-webhook.service.ts b/server/src/nest/mail-ingest/mail-ingest-webhook.service.ts new file mode 100644 index 000000000..93b3fef44 --- /dev/null +++ b/server/src/nest/mail-ingest/mail-ingest-webhook.service.ts @@ -0,0 +1,107 @@ +import { DomainError } from '../common/domain-error'; +import { MailIngestIntakeService, type MailIntakeRefusal } from './mail-ingest-intake.service'; +import { MailIngestSettingsService } from './mail-ingest-settings.service'; +import { Injectable } from '@nestjs/common'; +import type { MailIngestInboundResponse } from '@trek/shared'; + +import { createHmac, timingSafeEqual } from 'node:crypto'; + +/** The headers a webhook sender signs a mail with, and the address it was sent to (the RCPT TO). */ +export const WEBHOOK_TIMESTAMP_HEADER = 'x-trek-timestamp'; +export const WEBHOOK_SIGNATURE_HEADER = 'x-trek-signature'; +export const WEBHOOK_RECIPIENT_HEADER = 'x-trek-recipient'; + +/** How far a signed timestamp may lie from the server's clock, either way. */ +const WINDOW_MS = 5 * 60 * 1000; + +/** How a mail the intake refused answers the sender, by why. */ +const REFUSALS: Record = { + unavailable: [404, 'Not found'], + too_large: [413, 'This mail is larger than the import takes'], + not_a_mail: [422, 'This is not a mail'], + unknown_address: [422, 'No address of this server takes this mail'], + address_daily_limit: [429, 'This address took all the mails it takes today'], + inbox_full: [429, 'The inbox of this address is full'], +}; + +/** The signed parts of a webhook request, as the sender put them in its headers. */ +interface WebhookSignedParts { + timestamp?: string; + signature?: string; + recipient?: string; +} + +/** + * The signed webhook: a mail service (a Cloudflare Email Worker, Postmark, + * Mailgun) posts each mail to the install raw, signed with the admin's + * secret. The signature covers the time it was sent, the address it was + * sent to and every byte of the mail, so neither can be changed on the way, + * and a request older than five minutes is refused. A request replayed + * within that window stores nothing new: the intake keeps one copy of a mail + * per address. + */ +@Injectable() +export class MailIngestWebhookService { + constructor( + private readonly settings: MailIngestSettingsService, + private readonly intake: MailIngestIntakeService, + ) {} + + /** + * The secret mails are checked against. While the import is off, the + * webhook is not its transport or no secret is set, the route answers as + * if it did not exist. + */ + async secret(): Promise { + const open = (await this.settings.isAvailable()) && (await this.settings.transport()) === 'webhook'; + const secret = open ? await this.settings.webhookSecret() : null; + if (!secret) throw new DomainError(404, 'Not found'); + return secret; + } + + maxBytes(): number { + return this.settings.maxMailBytes(); + } + + async receive( + secret: string, + raw: Buffer, + signed: WebhookSignedParts, + now: number = Date.now(), + ): Promise { + const recipient = signed.recipient?.trim() ?? ''; + if (!verifyWebhook(secret, signed, recipient, raw, now)) throw new DomainError(401, 'The signature does not match'); + const result = await this.intake.receive(raw, 'webhook', recipient ? [recipient] : []); + if (result.status === 'refused') { + const [status, error] = REFUSALS[result.reason]; + throw new DomainError(status, error); + } + return { accepted: true, duplicate: result.status === 'duplicate' }; + } +} + +/** + * The signature a sender puts on a mail: hex HMAC-SHA256, keyed with the + * secret, over the timestamp (Unix seconds), a newline, the recipient + * address (empty when the sender names none), a newline, and the raw mail. + */ +export function signWebhookMail(secret: string, timestamp: string, recipient: string, raw: Buffer): string { + return createHmac('sha256', secret).update(`${timestamp}\n${recipient}\n`).update(raw).digest('hex'); +} + +function verifyWebhook( + secret: string, + signed: WebhookSignedParts, + recipient: string, + raw: Buffer, + now: number, +): boolean { + const { timestamp, signature } = signed; + if (!timestamp || !/^\d{1,12}$/.test(timestamp) || Math.abs(now - Number(timestamp) * 1000) > WINDOW_MS) { + return false; + } + const given = (signature ?? '').trim().replace(/^sha256=/i, '').toLowerCase(); + if (!/^[0-9a-f]{64}$/.test(given) || recipient.length > 320) return false; + const expected = signWebhookMail(secret, timestamp, recipient, raw); + return timingSafeEqual(Buffer.from(given, 'hex'), Buffer.from(expected, 'hex')); +} diff --git a/server/src/nest/mail-ingest/mail-ingest.module.ts b/server/src/nest/mail-ingest/mail-ingest.module.ts index 4e2efb5d3..ddeb063a9 100644 --- a/server/src/nest/mail-ingest/mail-ingest.module.ts +++ b/server/src/nest/mail-ingest/mail-ingest.module.ts @@ -21,6 +21,9 @@ import { SettingsModule } from '../settings/settings.module'; import { MailIngestActionsService } from './mail-ingest-actions.service'; import { MailIngestAddressService } from './mail-ingest-address.service'; import { MailIngestApplyService } from './mail-ingest-apply.service'; +import { MailIngestImapClient } from './mail-ingest-imap.client'; +import { MailIngestImapJob } from './mail-ingest-imap.job'; +import { MailIngestInboundController } from './mail-ingest-inbound.controller'; import { MailIngestInboxService } from './mail-ingest-inbox.service'; import { MailIngestIntakeService } from './mail-ingest-intake.service'; import { MailIngestNotificationActions } from './mail-ingest-notification-actions'; @@ -28,19 +31,25 @@ import { MailIngestNotifyService } from './mail-ingest-notify.service'; import { MailIngestProcessJob } from './mail-ingest-process.job'; import { MailIngestProcessService } from './mail-ingest-process.service'; import { MailIngestReadingService } from './mail-ingest-reading.service'; +import { MailIngestRelayJob } from './mail-ingest-relay.job'; +import { MailIngestRelayHost } from './mail-ingest-relay.service'; import { MailIngestRetentionJob } from './mail-ingest-retention.job'; import { MailIngestSettingsService } from './mail-ingest-settings.service'; +import { MailIngestStatusService } from './mail-ingest-status.service'; import { MailIngestTripsService } from './mail-ingest-trips.service'; +import { MailIngestWebhookService } from './mail-ingest-webhook.service'; import { MikroOrmModule } from '@mikro-orm/nestjs'; import { Module } from '@nestjs/common'; /** * The booking mail import: a person forwards a booking mail to an address of * their own, and the booking lands in the matching trip, or waits in their - * inbox with a reason. Transports (IMAP, the signed webhook, the TREK Cloud - * relay, an .eml upload) only hand mail to MailIngestIntakeService; the - * processing job reads it; everything a person does with an entry goes - * through MailIngestActionsService. + * inbox with a reason. Transports (the IMAP job, the signed webhook, the + * TREK Cloud relay job, an .eml upload) only hand mail to + * MailIngestIntakeService; the processing job reads it; everything a person + * does with an entry goes through MailIngestActionsService, from REST and + * MCP alike. MailIngestRelayHost is exported for the managed build, which + * attaches its control-plane client to it. */ @Module({ imports: [ @@ -67,6 +76,7 @@ import { Module } from '@nestjs/common'; MailIngestCandidates, ]), ], + controllers: [MailIngestInboundController], providers: [ MailIngestSettingsService, MailIngestTripsService, @@ -78,10 +88,22 @@ import { Module } from '@nestjs/common'; MailIngestProcessService, MailIngestActionsService, MailIngestInboxService, + MailIngestStatusService, + MailIngestWebhookService, + MailIngestImapClient, + MailIngestRelayHost, MailIngestNotificationActions, MailIngestProcessJob, MailIngestRetentionJob, + MailIngestImapJob, + MailIngestRelayJob, + ], + exports: [ + MailIngestSettingsService, + MailIngestAddressService, + MailIngestIntakeService, + MailIngestActionsService, + MailIngestRelayHost, ], - exports: [MailIngestSettingsService, MailIngestAddressService, MailIngestIntakeService, MailIngestActionsService], }) export class MailIngestModule {} diff --git a/server/tests/unit/nest/mail-ingest/mail-ingest-imap.client.test.ts b/server/tests/unit/nest/mail-ingest/mail-ingest-imap.client.test.ts new file mode 100644 index 000000000..c525b1d35 --- /dev/null +++ b/server/tests/unit/nest/mail-ingest/mail-ingest-imap.client.test.ts @@ -0,0 +1,112 @@ +/** + * The IMAP client the import reads a box with, over a stand-in for imapflow: + * the host goes through the SSRF guard and the connection to the address it + * vetted, the certificate checked against the name; the session filters the + * newest mail IMAP always names, makes the processed folder, moves by uid and + * says goodbye even over a broken connection. + */ +import { MailIngestImapClient } from '../../../../src/nest/mail-ingest/mail-ingest-imap.client'; +import { SsrfBlockedError } from '../../../../src/utils/ssrfGuard'; + +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const { checkSsrf, flow, ImapFlow } = vi.hoisted(() => { + const flow = { + connect: vi.fn(async () => undefined), + mailboxOpen: vi.fn(async () => ({ uidValidity: 77n })), + fetch: vi.fn(), + fetchOne: vi.fn(), + mailboxCreate: vi.fn(async () => ({ path: 'TREK.processed', created: false })), + messageMove: vi.fn(async () => true), + logout: vi.fn(async () => undefined), + close: vi.fn(), + }; + // A function, not an arrow: the client builds it with `new`. + const ImapFlow = vi.fn(function () { + return flow; + }); + return { checkSsrf: vi.fn(), flow, ImapFlow }; +}); +vi.mock('imapflow', () => ({ ImapFlow })); +vi.mock('../../../../src/utils/ssrfGuard', async (orig) => ({ + ...(await orig>()), + checkSsrf, +})); + +const CONFIG = { + host: 'imap.example.com', + port: 143, + tls: 'starttls', + user: 'u', + pass: 'p', + mailbox: 'INBOX', +} as const; + +async function* messages(...uids: number[]) { + for (const uid of uids) yield { uid, size: uid * 10 }; +} + +beforeEach(() => { + vi.clearAllMocks(); + checkSsrf.mockResolvedValue({ allowed: true, isPrivate: false, resolvedIp: '203.0.113.5' }); +}); + +describe('MailIngestImapClient', () => { + it('connects to the vetted address, checking the certificate against the name', async () => { + await new MailIngestImapClient().connect(CONFIG); + expect(checkSsrf).toHaveBeenCalledWith('https://imap.example.com/'); + expect(ImapFlow).toHaveBeenCalledWith( + expect.objectContaining({ + host: '203.0.113.5', + port: 143, + secure: false, + doSTARTTLS: true, + servername: 'imap.example.com', + auth: { user: 'u', pass: 'p' }, + logger: false, + }), + ); + expect(flow.connect).toHaveBeenCalled(); + }); + + it('refuses a host the guard blocks, before any connection', async () => { + checkSsrf.mockResolvedValue({ allowed: false, isPrivate: true, error: 'Private addresses are not allowed' }); + await expect(new MailIngestImapClient().connect(CONFIG)).rejects.toBeInstanceOf(SsrfBlockedError); + expect(ImapFlow).not.toHaveBeenCalled(); + }); + + it('reads a box through one session: uid validity, the mails after the cursor, a source, the move', async () => { + const session = await new MailIngestImapClient().connect({ ...CONFIG, host: '2001:db8::1', tls: 'tls' }); + expect(checkSsrf).toHaveBeenCalledWith('https://[2001:db8::1]/'); + expect(ImapFlow).toHaveBeenCalledWith(expect.objectContaining({ secure: true, servername: undefined })); + + expect(await session.open('INBOX')).toEqual({ uidValidity: 77 }); + // `41:*` names the newest mail even when it is older than 41. + flow.fetch.mockReturnValue(messages(40, 42, 41)); + expect(await session.listAfter(40)).toEqual([ + { uid: 41, size: 410 }, + { uid: 42, size: 420 }, + ]); + expect(flow.fetch).toHaveBeenCalledWith('41:*', { uid: true, size: true }, { uid: true }); + + flow.fetchOne.mockResolvedValueOnce({ uid: 41, source: Buffer.from('raw') }).mockResolvedValueOnce(false); + expect(await session.source(41)).toEqual(Buffer.from('raw')); + expect(await session.source(43)).toBeNull(); + + expect(await session.ensureFolder(['TREK', 'processed'])).toBe('TREK.processed'); + expect(flow.mailboxCreate).toHaveBeenCalledWith(['TREK', 'processed']); + await session.move([41, 42], 'TREK.processed'); + expect(flow.messageMove).toHaveBeenCalledWith('41,42', 'TREK.processed', { uid: true }); + await session.move([], 'TREK.processed'); + expect(flow.messageMove).toHaveBeenCalledTimes(1); + }); + + it('logs out, and closes a connection that broke on its own', async () => { + const session = await new MailIngestImapClient().connect(CONFIG); + await session.close(); + expect(flow.logout).toHaveBeenCalled(); + flow.logout.mockRejectedValueOnce(new Error('socket closed')); + await session.close(); + expect(flow.close).toHaveBeenCalled(); + }); +}); diff --git a/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts b/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts new file mode 100644 index 000000000..848792ca4 --- /dev/null +++ b/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts @@ -0,0 +1,350 @@ +/** + * How mail reaches the import and how the admin sees it run, over stubs: the + * signed webhook (its window, its signature over the recipient and the raw + * mail, its refusals), the IMAP run (cursor, renumbered box, size limit, the + * move to TREK/processed, a failure the status card reads), the TREK Cloud + * relay (hash check, acknowledgement, idle without a relay) and the status + * card and connection test. + */ +import { MailIngestCandidates } from '../../../../src/db/entities/MailIngestCandidates.entity'; +import { MailIngestCursor } from '../../../../src/db/entities/MailIngestCursor.entity'; +import { MailIngestMessages } from '../../../../src/db/entities/MailIngestMessages.entity'; +import { DomainError } from '../../../../src/nest/common/domain-error'; +import { + MailIngestImapClient, + describeImapFailure, + type ImapSession, +} from '../../../../src/nest/mail-ingest/mail-ingest-imap.client'; +import { MailIngestImapJob } from '../../../../src/nest/mail-ingest/mail-ingest-imap.job'; +import { MailIngestIntakeService } from '../../../../src/nest/mail-ingest/mail-ingest-intake.service'; +import { MailIngestRelayJob } from '../../../../src/nest/mail-ingest/mail-ingest-relay.job'; +import { MailIngestRelayHost } from '../../../../src/nest/mail-ingest/mail-ingest-relay.service'; +import { MailIngestSettingsService } from '../../../../src/nest/mail-ingest/mail-ingest-settings.service'; +import { MailIngestStatusService } from '../../../../src/nest/mail-ingest/mail-ingest-status.service'; +import { + MailIngestWebhookService, + signWebhookMail, +} from '../../../../src/nest/mail-ingest/mail-ingest-webhook.service'; +import { CronRegistrarService } from '../../../../src/nest/scheduling/cron-registrar.service'; +import { SsrfBlockedError } from '../../../../src/utils/ssrfGuard'; +import { getRepositoryToken } from '@mikro-orm/nestjs'; +import { Test } from '@nestjs/testing'; + +import { createHash } from 'node:crypto'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('../../../../src/nest/audit/audit-log.logger', () => ({ + LOG_LEVEL: 'error', + logInfo: vi.fn(), + logDebug: vi.fn(), + logError: vi.fn(), + logWarn: vi.fn(), +})); + +const IMAP = { host: 'imap.example.com', port: 993, tls: 'tls', user: 'u', pass: 'p', mailbox: 'INBOX' } as const; + +const settings = { + isAvailable: vi.fn(async () => true), + transport: vi.fn(async () => 'webhook'), + webhookSecret: vi.fn(async (): Promise => 'a'.repeat(32)), + imapConfig: vi.fn(async (): Promise => IMAP), + maxMailBytes: vi.fn(() => 1000), +}; +const intake = { receive: vi.fn() }; +const cursor = { + acquireLease: vi.fn(async () => true), + releaseLease: vi.fn(async () => undefined), + recordOk: vi.fn(async () => undefined), + recordError: vi.fn(async () => undefined), + find: vi.fn(), +}; +const messages = { tallySince: vi.fn(), countAllAiReadSince: vi.fn() }; +const candidates = { tallyResolutionsSince: vi.fn(), pendingPerUser: vi.fn() }; +const session = { + open: vi.fn(), + listAfter: vi.fn(), + source: vi.fn(), + ensureFolder: vi.fn(async () => 'TREK.processed'), + move: vi.fn(async () => undefined), + close: vi.fn(async () => undefined), +}; +const imap = { connect: vi.fn(async (): Promise => session) }; +const registrar = { isEnabled: () => false, register: vi.fn() }; +let relays: MailIngestRelayHost; + +async function build() { + const moduleRef = await Test.createTestingModule({ + providers: [ + MailIngestWebhookService, + MailIngestImapJob, + MailIngestRelayJob, + MailIngestStatusService, + { provide: MailIngestRelayHost, useValue: relays }, + { provide: MailIngestSettingsService, useValue: settings }, + { provide: MailIngestIntakeService, useValue: intake }, + { provide: MailIngestImapClient, useValue: imap }, + { provide: CronRegistrarService, useValue: registrar }, + { provide: getRepositoryToken(MailIngestCursor), useValue: cursor }, + { provide: getRepositoryToken(MailIngestMessages), useValue: messages }, + { provide: getRepositoryToken(MailIngestCandidates), useValue: candidates }, + ], + }).compile(); + return moduleRef; +} + +let app: Awaited>; +beforeEach(async () => { + vi.clearAllMocks(); + settings.isAvailable.mockResolvedValue(true); + settings.webhookSecret.mockResolvedValue('a'.repeat(32)); + settings.imapConfig.mockResolvedValue(IMAP); + cursor.acquireLease.mockResolvedValue(true); + cursor.find.mockResolvedValue(null); + relays = new MailIngestRelayHost(); + app = await build(); +}); + +describe('the signed webhook', () => { + const SECRET = 'a'.repeat(32); + const RAW = Buffer.from('Subject: Your booking\r\n\r\nABC123'); + const NOW = Date.parse('2030-01-15T10:00:00Z'); + const STAMP = String(NOW / 1000); + const service = () => app.get(MailIngestWebhookService); + const signed = (over: { timestamp?: string; recipient?: string; secret?: string } = {}) => { + const timestamp = over.timestamp ?? STAMP; + const recipient = over.recipient ?? 'trek+tok@example.com'; + const signature = `sha256=${signWebhookMail(over.secret ?? SECRET, timestamp, recipient, RAW)}`; + return { timestamp, recipient, signature }; + }; + + it('opens only while the import is on, the webhook is its transport and a secret is set', async () => { + settings.transport.mockResolvedValue('webhook'); + expect(await service().secret()).toBe(SECRET); + settings.transport.mockResolvedValue('imap'); + await expect(service().secret()).rejects.toThrow('Not found'); + settings.transport.mockResolvedValue('webhook'); + settings.webhookSecret.mockResolvedValue(null); + await expect(service().secret()).rejects.toThrow('Not found'); + settings.webhookSecret.mockResolvedValue(SECRET); + settings.isAvailable.mockResolvedValue(false); + await expect(service().secret()).rejects.toThrow('Not found'); + }); + + it('takes a mail whose signature covers it, its recipient and a fresh timestamp', async () => { + intake.receive.mockResolvedValue({ status: 'accepted', messageId: 1 }); + expect(await service().receive(SECRET, RAW, signed(), NOW)).toEqual({ accepted: true, duplicate: false }); + expect(intake.receive).toHaveBeenCalledWith(RAW, 'webhook', ['trek+tok@example.com']); + intake.receive.mockResolvedValue({ status: 'duplicate', messageId: 1 }); + const noRecipient = signed({ recipient: '' }); + expect(await service().receive(SECRET, RAW, noRecipient, NOW)).toEqual({ accepted: true, duplicate: true }); + expect(intake.receive).toHaveBeenLastCalledWith(RAW, 'webhook', []); + }); + + it('refuses another secret, another recipient, a changed mail and a timestamp out of its window', async () => { + const refused = (parts: ReturnType, raw = RAW) => + expect(service().receive(SECRET, raw, parts, NOW)).rejects.toThrow('The signature does not match'); + await refused(signed({ secret: 'b'.repeat(32) })); + await refused({ ...signed(), recipient: 'trek+other@example.com' }); + await refused(signed(), Buffer.from('Subject: Your booking\r\n\r\nXYZ999')); + await refused(signed({ timestamp: String(NOW / 1000 - 301) })); + await refused({ ...signed(), signature: 'sha256=nothex' }); + await refused({ ...signed(), timestamp: undefined as never }); + expect(intake.receive).not.toHaveBeenCalled(); + }); + + it('answers what the intake refused with its own status', async () => { + const cases: Array<[string, number]> = [ + ['unknown_address', 422], + ['too_large', 413], + ['address_daily_limit', 429], + ['unavailable', 404], + ]; + for (const [reason, status] of cases) { + intake.receive.mockResolvedValueOnce({ status: 'refused', reason }); + const err = await service() + .receive(SECRET, RAW, signed(), NOW) + .catch((e: unknown) => e); + expect(err, reason).toBeInstanceOf(DomainError); + expect((err as DomainError).getStatus(), reason).toBe(status); + } + }); +}); + +describe('the IMAP run', () => { + const job = () => app.get(MailIngestImapJob); + + beforeEach(() => { + settings.transport.mockResolvedValue('imap'); + session.open.mockResolvedValue({ uidValidity: 9 }); + session.listAfter.mockResolvedValue([]); + session.source.mockImplementation(async (uid: number) => Buffer.from(`mail ${uid}`)); + intake.receive.mockResolvedValue({ status: 'accepted', messageId: 1 }); + }); + + it('takes the new mails after the cursor, moves them on, and notes where it got to', async () => { + cursor.find.mockResolvedValue({ uid_validity: 9, last_uid: 40 }); + session.listAfter.mockResolvedValue([ + { uid: 41, size: 10 }, + { uid: 42, size: 10 }, + ]); + expect(await job().tick()).toBe(2); + expect(session.listAfter).toHaveBeenCalledWith(40); + expect(intake.receive.mock.calls.map(([raw]) => String(raw))).toEqual(['mail 41', 'mail 42']); + expect(session.ensureFolder).toHaveBeenCalledWith(['TREK', 'processed']); + expect(session.move).toHaveBeenCalledWith([41, 42], 'TREK.processed'); + expect(cursor.recordOk).toHaveBeenCalledWith('imap', 'INBOX', expect.any(String), { uidValidity: 9, lastUid: 42 }); + expect(session.close).toHaveBeenCalled(); + expect(cursor.releaseLease).toHaveBeenCalled(); + }); + + it('starts over when the server renumbered the box, and leaves a mail over the limit unread', async () => { + cursor.find.mockResolvedValue({ uid_validity: 3, last_uid: 400 }); + session.listAfter.mockResolvedValue([ + { uid: 1, size: 5000 }, + { uid: 2, size: 10 }, + ]); + expect(await job().tick()).toBe(2); + expect(session.listAfter).toHaveBeenCalledWith(0); + expect(session.source).toHaveBeenCalledTimes(1); + expect(session.move).toHaveBeenCalledWith([1, 2], 'TREK.processed'); + }); + + it('touches nothing with an empty box, and nothing at all unless IMAP is the transport', async () => { + expect(await job().tick()).toBe(0); + expect(session.ensureFolder).not.toHaveBeenCalled(); + expect(cursor.recordOk).toHaveBeenCalledWith('imap', 'INBOX', expect.any(String), { uidValidity: 9, lastUid: 0 }); + settings.transport.mockResolvedValue('webhook'); + vi.clearAllMocks(); + expect(await job().tick()).toBe(0); + expect(imap.connect).not.toHaveBeenCalled(); + settings.transport.mockResolvedValue('imap'); + cursor.acquireLease.mockResolvedValue(false); + expect(await job().tick()).toBe(0); + expect(imap.connect).not.toHaveBeenCalled(); + }); + + it('records why a run failed for the status card, and keeps the cursor where it was', async () => { + imap.connect.mockRejectedValueOnce(Object.assign(new Error('Command failed'), { authenticationFailed: true })); + expect(await job().tick()).toBe(0); + const refused = 'The IMAP server refused the user name or password'; + expect(cursor.recordError).toHaveBeenCalledWith('imap', 'INBOX', refused); + expect(cursor.recordOk).not.toHaveBeenCalled(); + expect(cursor.releaseLease).toHaveBeenCalled(); + }); + + it('describes a failure without anything of the login', () => { + expect(describeImapFailure(new SsrfBlockedError('Requests to loopback are not allowed'))).toBe( + 'Requests to loopback are not allowed', + ); + expect(describeImapFailure(Object.assign(new Error('connect'), { code: 'ECONNREFUSED' }))).toBe( + 'Could not reach the IMAP server (ECONNREFUSED)', + ); + expect(describeImapFailure(Object.assign(new Error('x'), { responseText: 'Mailbox does not exist' }))).toBe( + 'The IMAP server answered: Mailbox does not exist', + ); + }); +}); + +describe('the TREK Cloud relay', () => { + const job = () => app.get(MailIngestRelayJob); + const relay = { list: vi.fn(), ack: vi.fn(async () => undefined) }; + const mail = (id: string, body: string, sha = createHash('sha256').update(body).digest('hex')) => ({ + id, + recipients: ['ab12tok@in.example.com'], + sha256: sha, + raw: Buffer.from(body).toString('base64'), + }); + + beforeEach(() => { + settings.transport.mockResolvedValue('cloud'); + relays.attach(relay); + }); + + it('stays idle without a relay attached', async () => { + relays = new MailIngestRelayHost(); + app = await build(); + expect(await job().tick()).toBe(0); + expect(cursor.acquireLease).not.toHaveBeenCalled(); + }); + + it('takes each whole mail in under the address it was sent to and acknowledges it', async () => { + const broken = mail('m2', 'two', 'f'.repeat(64)); + relay.list.mockResolvedValue({ mails: [mail('m1', 'one'), broken, mail('m3', 'three')] }); + intake.receive + .mockResolvedValueOnce({ status: 'accepted', messageId: 1 }) + .mockResolvedValueOnce({ status: 'refused', reason: 'unknown_address' }); + expect(await job().tick()).toBe(1); + expect(relay.list).toHaveBeenCalledWith(10); + expect(intake.receive).toHaveBeenCalledWith(Buffer.from('one'), 'cloud', ['ab12tok@in.example.com']); + // m2 did not arrive whole and stays with the relay; m3 is refused for good and dropped. + expect(relay.ack.mock.calls).toEqual([['m1'], ['m3']]); + expect(cursor.recordOk).toHaveBeenCalledWith('cloud', '', expect.any(String)); + }); + + it('keeps a mail the import cannot take right now, and refuses an inbox it cannot read', async () => { + relay.list.mockResolvedValue({ mails: [mail('m1', 'one')] }); + intake.receive.mockResolvedValue({ status: 'refused', reason: 'unavailable' }); + await job().tick(); + expect(relay.ack).not.toHaveBeenCalled(); + + relay.list.mockResolvedValue({ mails: [{ id: '../../etc', raw: 'x' }] }); + expect(await job().tick()).toBe(0); + const unreadable = 'The relay answered with an inbox this server cannot read'; + expect(cursor.recordError).toHaveBeenCalledWith('cloud', '', unreadable); + }); +}); + +describe('the status card and the connection test', () => { + const status = () => app.get(MailIngestStatusService); + + it('counts the last day by status and reason, and names no mail and no person', async () => { + settings.transport.mockResolvedValue('cloud'); + cursor.find.mockResolvedValue({ last_ok_at: '2030-01-15 09:58:00', last_error: null, lease_until: null }); + messages.tallySince.mockResolvedValue([ + { status: 'processed', reason: null, n: 4 }, + { status: 'discarded', reason: 'newsletter', n: 2 }, + { status: 'processed', reason: 'no_trip_in_range', n: 1 }, + ]); + messages.countAllAiReadSince.mockResolvedValue(3); + candidates.tallyResolutionsSince.mockResolvedValue([ + { resolution: 'auto_applied', n: 5 }, + { resolution: 'undone', n: 1 }, + ]); + candidates.pendingPerUser.mockResolvedValue([4, 1]); + + expect(await status().status(new Date('2030-01-15T10:00:00Z'))).toEqual({ + transport: 'cloud', + last_ok_at: '2030-01-15T09:58:00Z', + last_error: null, + lease_until: null, + last_day: { + by_status: { processed: 5, discarded: 2 }, + by_reason: { newsletter: 2, no_trip_in_range: 1 }, + }, + auto_applied: 5, + undone: 1, + ai_calls_today: 3, + backlog: [4, 1], + }); + expect(cursor.find).toHaveBeenCalledWith('cloud', ''); + expect(messages.countAllAiReadSince).toHaveBeenCalledWith('2030-01-15 00:00:00'); + }); + + it('tests each transport the way the import reaches it', async () => { + settings.transport.mockResolvedValue('off'); + expect(await status().test()).toMatchObject({ ok: false, error: expect.stringContaining('upload') }); + settings.transport.mockResolvedValue('webhook'); + expect(await status().test()).toEqual({ transport: 'webhook', ok: true, error: null }); + settings.transport.mockResolvedValue('cloud'); + expect(await status().test()).toMatchObject({ ok: false, error: 'The TREK Cloud relay is not connected' }); + settings.transport.mockResolvedValue('imap'); + session.open.mockResolvedValue({ uidValidity: 1 }); + expect(await status().test()).toEqual({ transport: 'imap', ok: true, error: null }); + expect(session.open).toHaveBeenCalledWith('INBOX'); + expect(session.close).toHaveBeenCalled(); + imap.connect.mockRejectedValueOnce(Object.assign(new Error('x'), { code: 'ETIMEDOUT' })); + expect(await status().test()).toMatchObject({ ok: false, error: 'Could not reach the IMAP server (ETIMEDOUT)' }); + settings.imapConfig.mockResolvedValue(null); + expect(await status().test()).toMatchObject({ ok: false, error: 'The IMAP host, user or password is missing' }); + }); +}); diff --git a/wiki/Booking-Mail-Import.md b/wiki/Booking-Mail-Import.md new file mode 100644 index 000000000..b6f26e8c8 --- /dev/null +++ b/wiki/Booking-Mail-Import.md @@ -0,0 +1,98 @@ +# Booking Mail Import + +Everyone on the instance gets a forwarding address of their own. A booking confirmation forwarded there is read by +TREK and added to the matching trip, with its times, places, booking reference and cost, and the mail attached as its +document. A booking TREK is not sure about waits in **Open bookings** with the reason, where it can be added to a trip +by hand, filed as a document or discarded. A booking added on its own can be undone with one tap for seven days. + +The import is **off by default**. + +## How mail reaches TREK + +An admin sets it up in **Admin → Integrations → Booking mail import**. Every field there can also be set by an +[environment variable](Environment-Variables#booking-mail-import), which then wins. + +| Field | What it is | +|---|---| +| Import on | The switch for the whole instance. Off, no address takes mail and the entry points disappear. | +| Transport | `IMAP`, `Webhook`, or none (only the `.eml` upload in the inbox works). | +| Address template | The forwarding address with `{token}` where each person's token goes, for example `trek+{token}@example.com`. | +| AI mails per person and day | How many forwarded mails a day an AI reads for one person (default 10). Above it, only machine-readable data is read. | + +Each person's token is 24 random characters. A person can renew their address at any time; the old one stops taking +mail at once. Treat the address like a password: whoever knows it can put bookings into that person's trips. + +### IMAP + +TREK reads one mailbox every two minutes. Point a catch-all, or a mailbox with sub-addressing (`trek+anything@`), at +it and use the matching address template. Read mails move to the folder `TREK/processed` and are never deleted, so +anyone with access to the mailbox can read them. + +| Field | Default | +|---|---| +| Host | | +| Port | `993` | +| Encryption | `tls` (from the first byte) or `starttls` | +| User, password | | +| Folder | `INBOX` | + +A mail server on your own network needs `ALLOW_INTERNAL_NETWORK=true` ([Internal network access](Internal-Network-Access)). +**Test connection** in the dialog opens the folder the way the import does and says what failed. + +### Webhook + +A mail service posts every mail to `POST /api/mail-ingest/inbound`, raw, with `Content-Type: message/rfc822`, at most +10 MB. The route answers `404` unless the webhook is the transport and a secret of at least 32 characters is set. + +Each request carries three headers: + +| Header | Value | +|---|---| +| `X-Trek-Timestamp` | The time of sending, in Unix seconds. Requests more than five minutes off are refused. | +| `X-Trek-Recipient` | The address the mail was sent to (the envelope recipient). Optional; without it TREK reads `Delivered-To`, `X-Original-To` and `To`. | +| `X-Trek-Signature` | `sha256=` followed by the hex HMAC-SHA256 of `\n\n`, keyed with the secret. | + +A Cloudflare Email Worker that forwards a mail: + +```js +export default { + async email(message, env) { + const raw = new Uint8Array(await new Response(message.raw).arrayBuffer()); + const timestamp = String(Math.floor(Date.now() / 1000)); + const key = await crypto.subtle.importKey('raw', new TextEncoder().encode(env.TREK_SECRET), + { name: 'HMAC', hash: 'SHA-256' }, false, ['sign']); + const head = new TextEncoder().encode(`${timestamp}\n${message.to}\n`); + const signed = new Uint8Array(head.length + raw.length); + signed.set(head); + signed.set(raw, head.length); + const mac = new Uint8Array(await crypto.subtle.sign('HMAC', key, signed)); + const signature = [...mac].map((b) => b.toString(16).padStart(2, '0')).join(''); + const res = await fetch(`${env.TREK_URL}/api/mail-ingest/inbound`, { + method: 'POST', + headers: { + 'Content-Type': 'message/rfc822', + 'X-Trek-Timestamp': timestamp, + 'X-Trek-Recipient': message.to, + 'X-Trek-Signature': `sha256=${signature}`, + }, + body: raw, + }); + if (!res.ok) message.setReject(`TREK answered ${res.status}`); + }, +}; +``` + +Postmark and Mailgun forward inbound mail as JSON, not raw; put a small function like the one above between them and +TREK. The answer is `{"accepted":true,"duplicate":false}`; a mail sent twice is stored once. + +## Status + +The **Status** card in the dialog shows the last successful read, the last error, and the mails of the last day by +outcome, in counts only. No subject, sender or content of any mail is shown there. + +## Privacy + +Forwarded mails are stored encrypted on the server for at most 30 days, and seven days after their booking was added. +With an AI connection set up, the mail's text and attachments, plus the title, dates and destinations of the +forwarding person's trips, go to that provider; the sender, recipient and forwarding headers are removed first. +Without AI, everything stays on the server. Let your users know. diff --git a/wiki/_Sidebar.md b/wiki/_Sidebar.md index d9f2f1aa6..3ffcf0923 100644 --- a/wiki/_Sidebar.md +++ b/wiki/_Sidebar.md @@ -90,6 +90,7 @@ - [[AI Assistant|AI-Assistant]] - [[AI Connections|AI-Connections]] - [[AI Booking Import|AI-Booking-Import]] +- [[Booking Mail Import|Booking-Mail-Import]] - [[MCP Overview|MCP-Overview]] - [[MCP Setup|MCP-Setup]] - [[MCP Scopes|MCP-Scopes]]