mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-11-14 00:01:03 +00:00
feat(mail-ingest): read forwarded mails from an IMAP box, a signed webhook and the TREK Cloud relay
The IMAP job reads the box every two minutes through the SSRF guard, keeps a uid cursor and moves read mails to TREK/processed. The webhook takes raw MIME signed with HMAC-SHA256 over a timestamp, the recipient and the mail, and is a 404 unless it is the transport. The relay job collects a managed install's mails.
This commit is contained in:
Maurice
@@ -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;
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<ImapMailRef[]>;
|
||||
/** One mail's raw bytes; null for a uid that is gone. */
|
||||
source(uid: number): Promise<Buffer | null>;
|
||||
/** Makes the folder (made of these parts, in the server's own delimiter) unless it exists; answers its path. */
|
||||
ensureFolder(parts: readonly string[]): Promise<string>;
|
||||
move(uids: readonly number[], folder: string): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<ImapSession> {
|
||||
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<string> {
|
||||
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<ImapMailRef[]> {
|
||||
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<Buffer | null> {
|
||||
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<string> {
|
||||
const created = await this.client.mailboxCreate([...parts]);
|
||||
return created.path;
|
||||
}
|
||||
|
||||
async move(uids: readonly number[], folder: string): Promise<void> {
|
||||
if (uids.length === 0) return;
|
||||
await this.client.messageMove(uids.join(','), folder, { uid: true });
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
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);
|
||||
}
|
||||
@@ -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<number> {
|
||||
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<void> {
|
||||
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}`);
|
||||
}
|
||||
}
|
||||
@@ -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<MailIngestInboundResponse> {
|
||||
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<Buffer> {
|
||||
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);
|
||||
});
|
||||
}
|
||||
@@ -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<typeof relayInboxSchema>['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<number> {
|
||||
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<boolean> {
|
||||
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';
|
||||
}
|
||||
}
|
||||
@@ -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<unknown>;
|
||||
/** `POST /v1/instance/inbox/:id/ack`: the mail is taken in (or never will be), and the relay drops it. */
|
||||
ack(id: string): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<MailIngestAdminStatusResponse> {
|
||||
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<MailIngestAdminTestResponse> {
|
||||
const transport = await this.settings.transport();
|
||||
const error = await this.problemWith(transport);
|
||||
return { transport, ok: error === null, error };
|
||||
}
|
||||
|
||||
private async problemWith(transport: MailIngestTransportSetting): Promise<string | null> {
|
||||
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<string | null> {
|
||||
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', '');
|
||||
}
|
||||
}
|
||||
@@ -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<MailIntakeRefusal, [status: number, error: string]> = {
|
||||
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<string> {
|
||||
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<MailIngestInboundResponse> {
|
||||
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'));
|
||||
}
|
||||
@@ -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 {}
|
||||
|
||||
@@ -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<Record<string, unknown>>()),
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<string | null> => 'a'.repeat(32)),
|
||||
imapConfig: vi.fn(async (): Promise<typeof IMAP | null> => 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<ImapSession> => 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<ReturnType<typeof build>>;
|
||||
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<typeof signed>, 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' });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user