openvibers/openvibe.watch Commits
Email in: a webhook watch can have an email address of its own (A3) (#197)
Changes
@@@ -98,6 +98,11 @@ VALKEY_PREFIX=#WATCH_HOOK_EXAMPLE_SECRET=# A receiver answers within this, or the attempt is retried on the backoff.#WATCH_WEBHOOK_TIMEOUT_MS=10000++# Email in (agent plan A3): a webhook watch may have an address at WATCH_MAIL_DOMAIN. Cloudflare Email Routing hands each++# message to the mail Worker (edge/mail-worker.mjs, deployed by deploy/mail-worker.js), which posts it to /hooks/email++# signed with this secret (at least 32 characters; the Worker's INGRESS_SECRET). Unset: no email in.++#WATCH_MAIL_INGRESS_SECRET=++#WATCH_MAIL_DOMAIN=openvibe.watch# OpenVibe.Actor's internal address, for actor-task and actor-agent actions (Watch needs actor.task.create and actor.agent.run grants).#WATCH_ACTOR_URL=http://127.0.0.1:4950@@@ -3,6 +3,13 @@What shipped on OpenVibe.Watch, newest first. The plan is T18 ("Watch complete"); each entry namesthe step it landed.++## 2026-11-03 — email in (agent plan A3)++++- A webhook watch can have an email address of its own (`in-…@openvibe.watch`): mail sent there runs its check on the++ sender, subject and text. Cloudflare Email Routing hands each message to the mail Worker (`edge/mail-worker.mjs`),++ which posts it to `POST /hooks/email`, signed; an unknown address or a paused watch bounces the mail++ (`server/webhook-sources.js`, `server/mail-parse.js`, migration `0016_webhook_email_address`).++## 2026-11-01 — the condition in words, a quiet test before arming, and the evidence in every alert- The watch list and each watch page say the watch's structured condition in one sentence — what it reads, where,@@@ -219,6 +219,24 @@ product (`server/webhook-sources.js`, `server/carriers/webhook.js`).("Got it") at once, with the check run right after on the interaction (`value_path: "data.name"` reads a slashcommand's name). A bad signature answers 401, as Discord checks.- Cases: `test/webhook-providers.test.js`.++- **Email in (agent plan A3).** A webhook watch can have an email address of its own instead of a secret: "Wake it by++ email?" on its page, or `POST /api/v1/watches/:id/webhook-secret` with `{ "provider": "email" }`, answers++ `{ address: "in-<26 characters>@openvibe.watch" }`. The address is the key (anyone who has it can wake the watch);++ asking again makes a new one and retires the old, and a webhook secret replaces it.++ - Cloudflare Email Routing receives the domain's mail (SPF, DKIM, DMARC checked there) and its catch-all rule hands++ every message to the mail Worker ([edge/mail-worker.mjs](edge/mail-worker.mjs)). The Worker posts++ `{ to, from, raw (base64, at most 180 KB), truncated, size }` to `POST /hooks/email` with++ `X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256 of "<time>.<body>">` (`WATCH_MAIL_INGRESS_SECRET`, within++ five minutes).++ - Watch finds the watch by the address, reads the message ([server/mail-parse.js](server/mail-parse.js): headers,++ encoded words, multipart, base64 and quoted-printable, HTML-only mail as text) and runs the check on++ `{ from, from_name, to, subject, date, message_id, text, text_truncated, attachments }` (`value_path: "subject"`).++ `from` is the envelope sender Email Routing checked; `header_from` is what the message says.++ - An unknown address answers 404 and a paused watch 409, and the Worker refuses the mail, so its sender gets a bounce.++ Over the hour's budget the mail is taken and not checked; a busy watch is tried once more after 1.5 s.++ - Deploy: `ssh host "sudo sed -n 's/^WATCH_MAIL_INGRESS_SECRET=//p' /etc/openvibe/watch.env" | node deploy/mail-worker.js --secret-stdin`++ uploads the Worker, turns Email Routing on for the zone and points the catch-all at the Worker.++ - Cases: `test/email-in.test.js`.- **What the extraction reads.** A JSON delivery is parsed and becomes the document (`value_path: "order.total"`); anyother is read as text (`text`, `regex`). A delivery that says JSON but is not records `parse_error`.- **Answers.** 200 `{ watch_id, state, triggered }` after the check. 401 for a missing or wrong signature. 404 for annnew file mode 100644@@@ -0,0 +1,79 @@++#!/usr/bin/env node++'use strict';++/**++ * Deploy Watch's email in (agent plan A3) on Cloudflare: the mail Worker (edge/mail-worker.mjs) with its bindings,++ * Email Routing on the zone, and the catch-all rule that hands every message to the Worker.++ *++ * ssh host "sudo sed -n 's/^WATCH_MAIL_INGRESS_SECRET=//p' /etc/openvibe/watch.env" \++ * | node deploy/mail-worker.js --secret-stdin [--zone openvibe.watch] [--watch-url https://openvibe.watch]++ *++ * The Cloudflare token comes from CF_API_TOKEN_FILE (default ~/.config/cloudflare-token-watch-mail): Workers Scripts++ * Write and Account Settings Read on the account; Email Routing Rules Write, DNS Write, Zone Settings Write and Zone++ * Read on the zone. The ingress secret is read from stdin (Watch's WATCH_MAIL_INGRESS_SECRET) and never printed.++ * Without --secret-stdin the Worker's code is updated and its secret left as it is. Prints only what it did.++ */++const fs = require('fs');++const os = require('os');++const path = require('path');++++const args = process.argv.slice(2);++const opt = (name, def) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] ? args[i + 1] : def; };++const ZONE = opt('zone', 'openvibe.watch');++const WATCH_URL = opt('watch-url', 'https://openvibe.watch');++const SCRIPT = opt('name', 'openvibe-watch-mail');++const TOKEN = fs.readFileSync(process.env.CF_API_TOKEN_FILE || path.join(os.homedir(), '.config', 'cloudflare-token-watch-mail'), 'utf8').trim();++const API = 'https://api.cloudflare.com/client/v4';++++async function cf(method, p, body, { form = false } = {}) {++ const res = await fetch(`${API}${p}`, {++ method,++ headers: { Authorization: `Bearer ${TOKEN}`, ...(body && !form ? { 'Content-Type': 'application/json' } : {}) },++ body: body ? (form ? body : JSON.stringify(body)) : undefined,++ });++ const out = await res.json().catch(() => ({}));++ if (!out.success) throw new Error(`${method} ${p}: ${res.status} ${JSON.stringify(out.errors || out).slice(0, 400)}`);++ return out.result;++}++++async function readStdin() {++ const chunks = [];++ for await (const c of process.stdin) chunks.push(c);++ return Buffer.concat(chunks).toString('utf8').trim();++}++++(async () => {++ const secret = args.includes('--secret-stdin') ? await readStdin() : null;++ if (secret !== null && secret.length < 32) throw new Error('the ingress secret is missing or shorter than 32 characters');++++ const zone = (await cf('GET', `/zones?name=${encodeURIComponent(ZONE)}`))[0];++ if (!zone) throw new Error(`no zone ${ZONE}`);++ const account = zone.account.id;++++ // 1. The Worker: its code and WATCH_URL; the secret binding is kept unless a new one is given.++ const code = fs.readFileSync(path.join(__dirname, '..', 'edge', 'mail-worker.mjs'));++ const bindings = [{ type: 'plain_text', name: 'WATCH_URL', text: WATCH_URL }];++ if (secret) bindings.push({ type: 'secret_text', name: 'INGRESS_SECRET', text: secret });++ else bindings.push({ type: 'inherit', name: 'INGRESS_SECRET' });++ const form = new FormData();++ form.append('metadata', new Blob([JSON.stringify({ main_module: 'mail-worker.mjs', compatibility_date: '2026-10-01', bindings })], { type: 'application/json' }));++ form.append('mail-worker.mjs', new Blob([code], { type: 'application/javascript+module' }), 'mail-worker.mjs');++ await cf('PUT', `/accounts/${account}/workers/scripts/${SCRIPT}`, form, { form: true });++ console.log(`worker ${SCRIPT}: uploaded (${code.length} bytes)${secret ? ', ingress secret set' : ', ingress secret kept'}`);++++ // 2. Email Routing on the zone (its MX and SPF records), once.++ const routing = await cf('GET', `/zones/${zone.id}/email/routing`);++ if (!routing.enabled) {++ await cf('POST', `/zones/${zone.id}/email/routing/enable`, {});++ console.log(`email routing: enabled on ${ZONE}`);++ } else console.log(`email routing: already on (${routing.status || 'ready'})`);++++ // 3. Every address of the zone goes to the Worker; Watch says which ones are a watch's.++ await cf('PUT', `/zones/${zone.id}/email/routing/rules/catch_all`, {++ name: 'OpenVibe.Watch email in', enabled: true,++ matchers: [{ type: 'all' }],++ actions: [{ type: 'worker', value: [SCRIPT] }],++ });++ console.log(`catch-all: every address at ${ZONE} → worker ${SCRIPT}`);++ const dns = await cf('GET', `/zones/${zone.id}/email/routing/dns`).catch(() => null);++ if (Array.isArray(dns)) console.log(`email routing records: ${dns.map((r) => `${r.type} ${r.name}`).join(', ')}`);++})().catch((err) => { console.error(`mail-worker deploy failed: ${err.message}`); process.exit(1); });nnew file mode 100644@@@ -0,0 +1,80 @@++/**++ * OpenVibe.Watch email in (agent plan A3): the Cloudflare Email Worker behind openvibe.watch's catch-all rule.++ *++ * Email Routing receives every message for the domain (its MX), checks SPF, DKIM and DMARC, and hands it here. The++ * Worker posts it to Watch, signed, and refuses the mail when Watch knows no watch at that address, so the sender gets++ * a bounce instead of silence.++ *++ * POST ${WATCH_URL}/hooks/email++ * X-OpenVibe-Mail-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256(INGRESS_SECRET, t + "." + body)>++ * { "to", "from", "raw": <the message, base64, at most 180 KB>, "truncated": <bool>, "size": <bytes> }++ *++ * Bindings: WATCH_URL (plain text, https://openvibe.watch) and INGRESS_SECRET (secret; Watch's++ * WATCH_MAIL_INGRESS_SECRET). No dependencies. Deployed by deploy/mail-worker.sh.++ */++++const MAX_RAW = 180 * 1024; // base64 makes it 4/3 bigger, and Watch takes 256 KB a delivery++++async function readRaw(stream, limit) {++ const reader = stream.getReader();++ const chunks = [];++ let size = 0;++ let truncated = false;++ for (;;) {++ const { done, value } = await reader.read();++ if (done) break;++ if (size + value.length > limit) {++ chunks.push(value.subarray(0, limit - size));++ size = limit;++ truncated = true;++ await reader.cancel();++ break;++ }++ chunks.push(value);++ size += value.length;++ }++ const out = new Uint8Array(size);++ let at = 0;++ for (const c of chunks) { out.set(c, at); at += c.length; }++ return { bytes: out, truncated };++}++++function base64(bytes) {++ let s = '';++ const step = 0x8000;++ for (let i = 0; i < bytes.length; i += step) s += String.fromCharCode.apply(null, bytes.subarray(i, i + step));++ return btoa(s);++}++++async function sign(secret, data) {++ const key = await crypto.subtle.importKey('raw', new TextEncoder().encode(secret), { name: 'HMAC', hash: 'SHA-256' }, false, ['sign']);++ const mac = new Uint8Array(await crypto.subtle.sign('HMAC', key, new TextEncoder().encode(data)));++ return Array.from(mac, (b) => b.toString(16).padStart(2, '0')).join('');++}++++export default {++ async email(message, env) {++ const { bytes, truncated } = await readRaw(message.raw, MAX_RAW);++ const body = JSON.stringify({ to: message.to, from: message.from, raw: base64(bytes), truncated, size: message.rawSize });++ const url = `${String(env.WATCH_URL || 'https://openvibe.watch').replace(/\/+$/, '')}/hooks/email`;++ const post = async () => {++ const t = String(Math.floor(Date.now() / 1000));++ const signature = `t=${t},v1=${await sign(env.INGRESS_SECRET, `${t}.${body}`)}`;++ return fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-OpenVibe-Mail-Signature': signature, 'User-Agent': 'OpenVibe.Watch mail-worker' }, body });++ };++ let res;++ try {++ res = await post();++ // The watch's previous check is still running: once more, a moment later.++ if (res.status === 503) { await new Promise((r) => setTimeout(r, 1500)); res = await post(); }++ } catch {++ // Watch unreachable: Email Routing has no "try later", so the mail is refused with a reason the sender sees.++ message.setReject('OpenVibe.Watch is not answering right now; please send it again later.');++ return;++ }++ if (res.status === 404) message.setReject('No watch has this address.');++ else if (res.status === 409) message.setReject('This watch is paused.');++ else if (res.status === 413) message.setReject('The message is too large.');++ // 200, 429 (over the hour's budget), 503 twice (busy), 5xx: taken; Watch logged what happened.++ },++};nnew file mode 100644@@@ -0,0 +1,4 @@++-- phase: expand++-- OpenVibe.Watch email in (agent plan A3): a watch's email address (provider 'email') is its key, kept in++-- webhook_secrets.secret as the address's local part (in-<26 characters>). Mail to the address finds its watch by it.++CREATE UNIQUE INDEX IF NOT EXISTS webhook_secrets_email_address ON webhook_secrets (secret) WHERE provider = 'email';@@@ -91,6 +91,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,}// Webhook sources (plan T18 step 5, server/webhook-sources.js): public, signed with the watch's own secret, read raw// (the signature covers the exact bytes), so no body parser runs before it.++ // Email in (A3): the Cloudflare mail Worker posts each message here, signed with WATCH_MAIL_INGRESS_SECRET.++ if (webhookSources) app.post('/hooks/email', (req, res) => { webhookSources.receiveMail(req, res).catch((err) => { log.error(`[webhook-sources] mail: ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });if (webhookSources) app.post('/hooks/:id', (req, res) => { webhookSources.receive(req, res).catch((err) => { log.error(`[webhook-sources] ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });app.use('/api', express.json({ limit: '256kb', type: ['application/json', 'application/*+json'] }));@@@ -93,6 +93,12 @@ function load(env = process.env) {webhookSources: {maxPerWatchPerHour: int(env.WATCH_WEBHOOK_MAX_PER_HOUR, 60),},++ // Email in (agent plan A3, server/webhook-sources.js): addresses at WATCH_MAIL_DOMAIN; the Cloudflare mail Worker++ // signs what it posts to /hooks/email with WATCH_MAIL_INGRESS_SECRET. Without the secret there is no email in.++ mailIn: {++ domain: (env.WATCH_MAIL_DOMAIN || 'openvibe.watch').trim().toLowerCase(),++ ingressSecret: env.WATCH_MAIL_INGRESS_SECRET || null,++ },// Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired// watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).@@@ -826,6 +826,8 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: CREATE_Sconst runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });const acts = actions ? await actions.recent(row.id, 10) : [];const hookProvider = v.source.kind === 'webhook' && ctx.webhookSources && ctx.webhookSources.providerOf ? await ctx.webhookSources.providerOf(row.id) : null;++ const hookAddress = hookProvider === 'email' && ctx.webhookSources.addressOf ? await ctx.webhookSources.addressOf(row.id) : null;++ const mailIn = !!(config.mailIn && config.mailIn.ingressSecret);const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`: a.kind === 'codes-run'@@@ -891,10 +893,14 @@ ${ran ? notice(html`The check ran: <strong>${W.stateWords(ran)}</strong>.`, ran${testRun ? testCheckNotice(testRun, testObs, v) : ''}<dl class="facts"><dt>Status</dt><dd>${badge(unarmed ? 'Not armed yet' : W.statusWords(v.status), unarmed ? 'warn' : W.statusKind(v.status))}</dd>--${v.source.kind === 'webhook'++${v.source.kind === 'webhook' && hookAddress++ ? html`<dt>Email address</dt><dd><code class="copyable">${hookAddress}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Make a new address</button></form><br><span class="muted small">Mail sent here wakes the watch. Anyone with the address can, so give it only to senders you mean; a new address retires this one.</span>++<details class="hook-provider"><summary>Use a webhook instead?</summary><p class="muted small">A webhook address with a secret of Watch's own replaces the email address, which stops working at once.</p><form method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a webhook secret</button></form></details></dd>`++ : v.source.kind === 'webhook'? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">${hookProvider ? `Signed by ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hookProvider] || hookProvider} with the key you gave. A new secret of Watch's own replaces it.` : 'The secret was shown once, when the watch was made. A new one replaces it at once.'}</span><details class="hook-provider"><summary>Signed by Slack, Stripe or Discord?</summary><p class="muted small">They sign with a key of their own: paste it here and Watch checks their signature instead of its own. It is kept to check each delivery and never shown again.</p>--<form method="post" action="/watches/${v.id}/webhook-secret"><label>Sender <select name="provider"><option value="slack">Slack (signing secret)</option><option value="stripe">Stripe (endpoint secret, whsec_…)</option><option value="discord">Discord (application public key)</option></select></label> <label>Its signing secret <input name="secret" type="password" autocomplete="off" required maxlength="220"></label> <button class="btn" type="submit">Use it</button></form></details></dd>`++<form method="post" action="/watches/${v.id}/webhook-secret"><label>Sender <select name="provider"><option value="slack">Slack (signing secret)</option><option value="stripe">Stripe (endpoint secret, whsec_…)</option><option value="discord">Discord (application public key)</option></select></label> <label>Its signing secret <input name="secret" type="password" autocomplete="off" required maxlength="220"></label> <button class="btn" type="submit">Use it</button></form></details>${mailIn ? html`++<details class="hook-provider"><summary>Wake it by email?</summary><p class="muted small">The watch gets an address of its own at ${config.mailIn.domain}: mail sent there runs the check with its sender, subject and text. It replaces the webhook secret.</p><form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Give it an email address</button></form></details>` : ''}</dd>`: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd><dt>Actions</dt><dd>${configuredActions}</dd>@@@ -999,7 +1005,20 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:res.set('Referrer-Policy', 'no-referrer');return page(req, res, {title: `${v.name}: webhook address`, private: true, crumbs: [{ label: 'Your watches', href: '/watches' }, { label: v.name, href: `/watches/${v.id}` }],-- body: hook.provider ? html`<h1>Your webhook is ready</h1>++ body: hook.provider === 'email' ? html`<h1>Your watch has an email address</h1>++${notice('Mail sent to this address wakes the watch. The address is the key: anyone who has it can wake the watch, so give it only to senders you mean.', 'ok')}++<dl class="facts">++<dt>Address</dt><dd><code class="copyable">${hook.address}</code></dd>++</dl>++<h2>Sending to it</h2>++<ul class="facts">++<li><strong>From a mailbox:</strong> make a filter that forwards the mail you care about (a sender, a subject) to this address.</li>++<li><strong>From a service:</strong> put the address where it sends its notices (an alert, a form, a store's order mail).</li>++<li><strong>What the watch reads:</strong> <code>from</code>, <code>from_name</code>, <code>to</code>, <code>subject</code>, <code>date</code>, <code>message_id</code>, <code>text</code> (the first 20,000 characters), <code>attachments</code> (how many). For example <code>value_path: "subject"</code> with "contains" <code>invoice</code>.</li>++<li>Mail to an address no watch has is refused, so its sender gets a bounce; a paused watch refuses it too.</li>++</ul>++<p>${W.inWords(v)}</p>++<p><a class="btn" href="/watches/${v.id}">Back to the watch</a></p>` : hook.provider ? html`<h1>Your webhook is ready</h1>${notice(`Deliveries to this address are checked with the ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hook.provider]} key you gave. It is not shown again.`, 'ok')}<dl class="facts"><dt>Address</dt><dd><code class="copyable">${hook.url}</code></dd>nnew file mode 100644@@@ -0,0 +1,147 @@++'use strict';++/**++ * The parts of an email a watch reads (agent plan A3, email in): who sent it, to whom, the subject, the date, the++ * message id and the text. Enough MIME for the mail people actually send: folded headers, RFC 2047 encoded words,++ * multipart (nested, up to four deep), base64 and quoted-printable bodies, and the common charsets. The first++ * text/plain part is the text; a message with only HTML gets its HTML as plain text. Attachments are counted, not kept.++ *++ * parseMail(raw: Buffer, { maxText }) → { from, to, subject, date, message_id, text, text_truncated, attachments }++ */++++const MAX_DEPTH = 4;++++function decoderFor(charset) {++ const label = String(charset || 'utf-8').trim().toLowerCase().replace(/^"|"$/g, '');++ try { return new TextDecoder(label === 'us-ascii' || label === 'ascii' ? 'utf-8' : label); } catch { return new TextDecoder('utf-8'); }++}++++/** Split a raw part into its header block and body (at the first empty line, CRLF or LF). */++function splitPart(buf) {++ const s = buf.toString('latin1');++ const m = /\r?\n\r?\n/.exec(s);++ if (!m) return { head: s, body: Buffer.alloc(0) };++ return { head: s.slice(0, m.index), body: buf.subarray(m.index + m[0].length) };++}++++/** Header block → Map of lower-case name → raw value (first one wins), with folded lines unfolded. */++function headersOf(head) {++ const out = new Map();++ for (const line of head.replace(/\r?\n[ \t]+/g, ' ').split(/\r?\n/)) {++ const i = line.indexOf(':');++ if (i <= 0) continue;++ const name = line.slice(0, i).trim().toLowerCase();++ if (!out.has(name)) out.set(name, line.slice(i + 1).trim());++ }++ return out;++}++++/** RFC 2047: =?charset?B|Q?text?= words, decoded; the whitespace between two encoded words goes. */++function decodeWords(value) {++ const raw = Buffer.from(String(value || ''), 'latin1').toString('latin1');++ return raw.replace(/(=\?[^?\s]+\?[bBqQ]\?[^?\s]*\?=)\s+(?==\?[^?\s]+\?[bBqQ]\?[^?\s]*\?=)/g, '$1')++ .replace(/=\?([^?\s]+)\?([bBqQ])\?([^?\s]*)\?=/g, (_, charset, enc, text) => {++ try {++ const bytes = enc.toUpperCase() === 'B'++ ? Buffer.from(text, 'base64')++ : Buffer.from(text.replace(/_/g, ' ').replace(/=([0-9A-Fa-f]{2})/g, (__, h) => String.fromCharCode(parseInt(h, 16))), 'latin1');++ return decoderFor(charset.split('*')[0]).decode(bytes);++ } catch { return text; }++ });++}++++/** A header's text: raw 8-bit bytes read as UTF-8 when they are UTF-8 (what most senders mean), then encoded words. */++function headerText(value) {++ let s = String(value || '');++ if (/[\x80-\xff]/.test(s)) {++ const u = Buffer.from(s, 'latin1').toString('utf8');++ if (!u.includes('\ufffd')) s = u;++ }++ return decodeWords(s);++}++++/** `type/subtype; a=b; c="d"` → { type, params } */++function contentType(value) {++ const [type, ...rest] = String(value || 'text/plain').split(';');++ const params = {};++ for (const p of rest) {++ const i = p.indexOf('=');++ if (i > 0) params[p.slice(0, i).trim().toLowerCase()] = p.slice(i + 1).trim().replace(/^"|"$/g, '');++ }++ return { type: type.trim().toLowerCase() || 'text/plain', params };++}++++function decodeBody(body, encoding) {++ const enc = String(encoding || '').trim().toLowerCase();++ if (enc === 'base64') return Buffer.from(body.toString('latin1').replace(/[^A-Za-z0-9+/=]/g, ''), 'base64');++ if (enc === 'quoted-printable') {++ const s = body.toString('latin1').replace(/=\r?\n/g, '');++ return Buffer.from(s.replace(/=([0-9A-Fa-f]{2})/g, (_, h) => String.fromCharCode(parseInt(h, 16))), 'latin1');++ }++ return body;++}++++/** HTML as plain text: scripts and styles dropped, breaks kept, tags gone, the common entities read. */++function htmlText(html) {++ return String(html)++ .replace(/<(script|style|head)[^>]*>[\s\S]*?<\/\1>/gi, ' ')++ .replace(/<br\s*\/?>/gi, '\n').replace(/<\/(p|div|li|tr|h[1-6])>/gi, '\n')++ .replace(/<[^>]+>/g, ' ')++ .replace(/ /gi, ' ').replace(/</gi, '<').replace(/>/gi, '>').replace(/"/gi, '"').replace(/'|'/gi, "'")++ .replace(/&#(\d{1,6});/g, (_, n) => { const c = Number(n); return c > 0 && c < 0x110000 ? String.fromCodePoint(c) : ' '; })++ .replace(/&/gi, '&')++ .replace(/[ \t]+\n/g, '\n').replace(/\n[ \t]+/g, '\n').replace(/\n{3,}/g, '\n\n').replace(/[ \t]{2,}/g, ' ').trim();++}++++/** Walk a part (and its children): the first text/plain, else the first text/html, and how many attachments. */++function walk(buf, depth, found) {++ const { head, body } = splitPart(buf);++ const h = headersOf(head);++ const ct = contentType(h.get('content-type'));++ const disposition = String(h.get('content-disposition') || '').toLowerCase();++ if (ct.type.startsWith('multipart/') && ct.params.boundary && depth < MAX_DEPTH) {++ const marker = `--${ct.params.boundary}`;++ const s = body.toString('latin1');++ const pieces = s.split(marker).slice(1);++ for (const piece of pieces) {++ if (piece.startsWith('--')) break; // the closing boundary++ walk(Buffer.from(piece.replace(/^\r?\n/, ''), 'latin1'), depth + 1, found);++ }++ return;++ }++ if (disposition.startsWith('attachment') || (!ct.type.startsWith('text/') && !ct.type.startsWith('multipart/'))) { found.attachments++; return; }++ if (ct.type === 'text/plain' && found.plain == null) found.plain = decoderFor(ct.params.charset).decode(decodeBody(body, h.get('content-transfer-encoding')));++ else if (ct.type === 'text/html' && found.html == null) found.html = decoderFor(ct.params.charset).decode(decodeBody(body, h.get('content-transfer-encoding')));++}++++/** "Ann <[email protected]>" or "[email protected]" → "[email protected]" (lower-cased domain kept as written). */++function addressOf(value) {++ const s = headerText(value);++ const m = /<([^<>\s]+@[^<>\s]+)>/.exec(s) || /([^\s<>",;]+@[^\s<>",;]+)/.exec(s);++ return m ? m[1] : '';++}++++function parseMail(raw, { maxText = 20000 } = {}) {++ const buf = Buffer.isBuffer(raw) ? raw : Buffer.from(String(raw || ''), 'utf8');++ const { head } = splitPart(buf);++ const h = headersOf(head);++ const found = { plain: null, html: null, attachments: 0 };++ walk(buf, 0, found);++ let text = found.plain != null ? found.plain : (found.html != null ? htmlText(found.html) : '');++ text = text.replace(/\r\n/g, '\n').trim();++ const truncated = text.length > maxText;++ if (truncated) text = text.slice(0, maxText);++ const date = h.get('date') ? new Date(h.get('date')) : null;++ return {++ from: addressOf(h.get('from')),++ from_name: headerText(h.get('from') || '').replace(/<[^>]*>/, '').trim().replace(/^"|"$/g, '').trim().slice(0, 200),++ to: addressOf(h.get('to')),++ subject: headerText(h.get('subject') || '').slice(0, 500),++ date: date && !Number.isNaN(date.getTime()) ? date.toISOString() : null,++ message_id: String(h.get('message-id') || '').replace(/^<|>$/g, '').slice(0, 300),++ text,++ text_truncated: truncated,++ attachments: found.attachments,++ };++}++++module.exports = { parseMail, htmlText, decodeWords };@@@ -24,12 +24,22 @@* secret anyway). Discord wants an interaction reply within 3 seconds: a PING gets* { type: 1 }, anything else { type: 4, ephemeral "Got it" } at once, and the check runs* right after, with the interaction as its input.++ * Email in (A3) POST …/webhook-secret { provider: 'email' } gives the watch an address of its own,++ * in-<26 characters>@openvibe.watch (WATCH_MAIL_DOMAIN). Cloudflare Email Routing hands every message++ * for the domain to a Worker (edge/mail-worker.js), which posts it to POST /hooks/email signed with++ * WATCH_MAIL_INGRESS_SECRET (X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256(secret,++ * time + "." + body)>, within five minutes). The address finds the watch; the message is read++ * (server/mail-parse.js) and the check runs with { from, from_name, to, subject, date, message_id,++ * text, text_truncated, attachments } as its JSON input. An unknown address answers 404, and the++ * Worker refuses the mail, so the sender gets a bounce. The address is the key: anyone who has it++ * can wake the watch, and making a new one retires the old.* Answers 200 { state, triggered } after the check; 404 for an unknown id or a watch that is not a webhook* watch; 401 for a missing or wrong signature (the same answer either way); 409 for a paused watch;* 429 over the watch's hourly budget (WATCH_WEBHOOK_MAX_PER_HOUR, default 60) with Retry-After;* 503 with Retry-After: 1 while the watch's previous check still runs; 413 over 256 KB.*/const crypto = require('crypto');++const { parseMail } = require('./mail-parse');const SECRET_RE = /^whk_[A-Za-z0-9_-]{43}$/;const TOLERANCE_SEC = 300;@@@ -40,6 +50,10 @@ const PROVIDERS = Object.freeze({discord: { name: 'Discord', secret: /^[0-9a-f]{64}$/, how: 'Discord signs each interaction with your application\'s key (X-Signature-Ed25519); paste its public key.' },});const MAX_BODY = 256 * 1024;++// An email address of a watch's own: its local part is the key (Crockford base32, 26 characters, 130 bits).++const MAIL_LOCAL_RE = /^in-[0-9a-hjkmnp-tv-z]{26}$/;++const B32 = '0123456789abcdefghjkmnpqrstvwxyz';++const newMailLocal = () => `in-${Array.from(crypto.randomBytes(26), (b) => B32[b & 31]).join('')}`;const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;const endpointFor = (watchId) => `/internal/hooks/${watchId}`;@@@ -110,12 +124,17 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.noconst maxPerHour = Number.isInteger(config.webhookSources && config.webhookSources.maxPerWatchPerHour) && config.webhookSources.maxPerWatchPerHour > 0? config.webhookSources.maxPerWatchPerHour : 60;const counts = new Map(); // watch id → { hour, n }++ const mail = {++ domain: String((config.mailIn && config.mailIn.domain) || 'openvibe.watch').toLowerCase(),++ ingressSecret: (config.mailIn && config.mailIn.ingressSecret) || null,++ };const st = {set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider) VALUES (?, ?, ?, ?)ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = excluded.provider`),get: db.prepare('SELECT secret, provider FROM webhook_secrets WHERE watch_id = ?'),drop: db.prepare('DELETE FROM webhook_secrets WHERE watch_id = ?'),watch: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),++ byMail: db.prepare("SELECT watch_id FROM webhook_secrets WHERE provider = 'email' AND secret = ?"),};/**@@@ -125,6 +144,12 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no*/async function issue(watchId, { provider = null, secret: theirs = null } = {}) {const url = `${config.baseUrl}/hooks/${watchId}`;++ if (provider === 'email') {++ if (!mail.ingressSecret) return { error: 'email in is not set up on this server' };++ const local = newMailLocal();++ await st.set.run(watchId, local, now(), 'email');++ return { url, provider: 'email', address: `${local}@${mail.domain}`, signature: 'Mail sent to this address wakes the watch: the address is the key.' };++ }if (provider != null && provider !== '') {const p = PROVIDERS[provider];if (!p) return { error: `provider is one of ${Object.keys(PROVIDERS).join(', ')}` };@@@ -138,6 +163,11 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.noreturn { url, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, Authorization: Bearer <secret>, or Telegram\'s X-Telegram-Bot-Api-Secret-Token' };}async function forget(watchId) { await st.drop.run(watchId); }++ /** The watch's email address, when it has one (shown to its owner on the watch's page). */++ async function addressOf(watchId) {++ const row = await st.get.get(watchId);++ return row && row.provider === 'email' && MAIL_LOCAL_RE.test(row.secret) ? `${row.secret}@${mail.domain}` : null;++ }/** Whose secret signs this watch's deliveries: null for Watch's own, else 'slack' or 'stripe'. */async function providerOf(watchId) { return ((await st.get.get(watchId)) || {}).provider || null; }@@@ -215,7 +245,51 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.noreturn res.json({ watch_id: id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });}-- return { issue, forget, providerOf, receive, endpointFor, verified, SECRET_RE, PROVIDERS };++ /**++ * POST /hooks/email — a message from the mail Worker (edge/mail-worker.js): { to, from, raw (base64), truncated },++ * signed with WATCH_MAIL_INGRESS_SECRET. Answers as /hooks/:id does; 404 makes the Worker refuse the mail.++ */++ async function receiveMail(req, res) {++ let raw;++ try { raw = await readRaw(req); } catch (err) { return problem(res, err.status || 400, err.status === 413 ? 'watch.too_large' : 'watch.bad_request', err.status === 413 ? 'a delivery is at most 256 KB' : 'unreadable body'); }++ if (!mail.ingressSecret) return problem(res, 404, 'watch.not_found', 'email in is not set up on this server');++ if (!mailSigned(mail.ingressSecret, raw, req.headers, now())) return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed by the mail ingress');++ let msg;++ try { msg = JSON.parse(raw.toString('utf8')); } catch { return problem(res, 400, 'watch.bad_request', 'the delivery is not JSON'); }++ const to = String((msg && msg.to) || '').trim().toLowerCase();++ const at = to.lastIndexOf('@');++ const local = at > 0 ? to.slice(0, at) : '';++ const found = at > 0 && to.slice(at + 1) === mail.domain && MAIL_LOCAL_RE.test(local) ? await st.byMail.get(local) : null;++ const row = found ? await st.watch.get(found.watch_id) : null;++ const source = row ? (typeof row.source === 'string' ? JSON.parse(row.source) : row.source) : null;++ if (!row || !source || source.kind !== 'webhook') return problem(res, 404, 'watch.not_found', 'no watch has this address');++ if (row.status !== 'active') return problem(res, 409, 'watch.disabled', `the watch is ${row.status}`);++ if (!underBudget(row.id)) return problem(res, 429, 'watch.rate_limited', `at most ${maxPerHour} deliveries an hour wake one watch`, { 'Retry-After': String(retryAfter()) });++ const parsed = parseMail(Buffer.from(String(msg.raw || ''), 'base64'));++ // The envelope is what Email Routing checked; the headers are what the sender wrote.++ const input = { ...parsed, from: String(msg.from || parsed.from || '').slice(0, 320), to, header_from: parsed.from, truncated: !!msg.truncated };++ let out;++ try {++ out = await check.run(row.id, { trigger: 'webhook', input: { body: Buffer.from(JSON.stringify(input)), content_type: 'application/json' } });++ } catch (err) {++ log.error(`[webhook-sources] mail ${row.id}: ${err.message}`);++ return problem(res, 500, 'watch.internal', 'the check failed; try again');++ }++ if (out && out.busy) return problem(res, 503, 'watch.busy', 'the watch\'s previous check is still running', { 'Retry-After': '1' });++ res.setHeader('Cache-Control', 'no-store');++ return res.json({ watch_id: row.id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });++ }++++ return { issue, forget, providerOf, addressOf, receive, receiveMail, endpointFor, verified, SECRET_RE, PROVIDERS };++}++++/** The mail Worker's signature: X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256(secret, time + "." + body)>. */++function mailSigned(secret, raw, headers, nowMs = Date.now()) {++ const parts = String(headers['x-openvibe-mail-signature'] || '').split(',').map((x) => x.trim().split('='));++ const t = (parts.find(([k]) => k === 't') || [])[1];++ if (!secret || !fresh(t, nowMs)) return false;++ const want = hmac(secret, Buffer.concat([Buffer.from(`${t}.`), raw]));++ return parts.filter(([k]) => k === 'v1').some(([, v]) => safeEqual(v || '', want));}--module.exports = { createWebhookSources, verified, endpointFor, newSecret, PROVIDERS };++module.exports = { createWebhookSources, verified, mailSigned, endpointFor, newSecret, newMailLocal, MAIL_LOCAL_RE, PROVIDERS };nnew file mode 100644@@@ -0,0 +1,212 @@++'use strict';++/**++ * Email in (agent plan A3): a webhook watch can have an email address of its own. Cloudflare Email Routing hands each++ * message for openvibe.watch to the mail Worker (edge/mail-worker.mjs), which posts it to POST /hooks/email signed++ * with WATCH_MAIL_INGRESS_SECRET; the address finds the watch, the message is read (server/mail-parse.js) and the++ * check runs on { from, subject, text, … }. An unknown address is 404 (the Worker bounces the mail); a paused watch 409;++ * a bad or old signature 401. Without the ingress secret there is no email in.++ */++const assert = require('assert');++const crypto = require('crypto');++const fs = require('fs');++const path = require('path');++const { boot, request, userToken, suite, newUser } = require('./helpers');++const { parseMail, htmlText } = require('../server/mail-parse');++const { mailSigned, MAIL_LOCAL_RE, newMailLocal } = require('../server/webhook-sources');++++const t = suite('email-in');++const INGRESS = 'mail-ingress-test-secret-0123456789';++let svc;++let bare;++const me = newUser();++const ME = userToken(me);++const OTHER = userToken(newUser());++const api = (base, method, p, opts = {}) => request(base, method, p, opts);++const hmac = (secret, data) => crypto.createHmac('sha256', secret).update(data).digest('hex');++const nowSec = () => Math.floor(Date.now() / 1000);++const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);++const def = (name) => ({ name, source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json', value_path: 'subject' }, condition: { op: 'contains', value: 'invoice' }, action: [{ kind: 'notification' }] });++++const CRLF = (s) => s.replace(/\r?\n/g, '\r\n');++const mailRaw = (to, subject = 'Your invoice #42', text = 'Total: 12.00 EUR') => CRLF(`From: "Ann Example" <[email protected]>++To: ${to}++Subject: ${subject}++Date: Mon, 03 Nov 2026 05:00:00 +0000++Message-ID: <[email protected]>++MIME-Version: 1.0++Content-Type: text/plain; charset=utf-8++++${text}++`);++const deliver = (base, to, raw, { secret = INGRESS, at = String(nowSec()), from = '[email protected]' } = {}) => {++ const body = JSON.stringify({ to, from, raw: Buffer.from(raw).toString('base64'), truncated: false, size: raw.length });++ return fetch(`${base}/hooks/email`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', 'X-OpenVibe-Mail-Signature': `t=${at},v1=${hmac(secret, `${at}.${body}`)}` } });++};++let watch;++let address;++++t('the reader: headers, encoded words, multipart, quoted-printable, base64, HTML-only, attachments', () => {++ const plain = parseMail(Buffer.from(mailRaw('[email protected]')));++ assert.strictEqual(plain.from, '[email protected]');++ assert.strictEqual(plain.from_name, 'Ann Example');++ assert.strictEqual(plain.subject, 'Your invoice #42');++ assert.strictEqual(plain.date, '2026-11-03T05:00:00.000Z');++ assert.strictEqual(plain.message_id, '[email protected]');++ assert.strictEqual(plain.text, 'Total: 12.00 EUR');++ const multi = parseMail(Buffer.from(CRLF(`From: =?UTF-8?B?w4Ruw6k=?= <[email protected]>++To: [email protected]++Subject: =?utf-8?Q?Caf=C3=A9_order?= =?utf-8?Q?_shipped?=++Content-Type: multipart/mixed; boundary="outer"++++--outer++Content-Type: multipart/alternative; boundary=inner++++--inner++Content-Type: text/plain; charset=utf-8++Content-Transfer-Encoding: quoted-printable++++Caf=C3=A9 order #7 has shipped. A long line that was =++wrapped.++--inner++Content-Type: text/html; charset=utf-8++++<p>Café order</p>++--inner--++--outer++Content-Type: application/pdf; name="label.pdf"++Content-Disposition: attachment; filename="label.pdf"++Content-Transfer-Encoding: base64++++JVBERi0xLjQK++--outer--++`)));++ assert.strictEqual(multi.from_name, 'Äné');++ assert.strictEqual(multi.subject, 'Café order shipped', 'adjacent encoded words join');++ assert.strictEqual(multi.text, 'Café order #7 has shipped. A long line that was wrapped.');++ assert.strictEqual(multi.attachments, 1);++ const html = parseMail(Buffer.from(CRLF(`From: [email protected]++Subject: Receipt++Content-Type: text/html; charset=iso-8859-1++Content-Transfer-Encoding: base64++++${Buffer.from('<html><head><style>p{}</style></head><body><p>Thanks!</p><p>Total & tax: 5 \xa3</p></body></html>', 'latin1').toString('base64')}++`)));++ assert.strictEqual(html.text, 'Thanks!\nTotal & tax: 5 £');++ assert.strictEqual(htmlText('<script>x()</script>a<br>b'), 'a\nb');++ const long = parseMail(Buffer.from(mailRaw('x@y', 's', 'z'.repeat(30000))));++ assert.strictEqual(long.text.length, 20000);++ assert.strictEqual(long.text_truncated, true);++ assert.deepStrictEqual(parseMail(Buffer.from('garbage')).subject, '');++});++++t('the signature: t and v1 over "<t>.<body>", within five minutes', () => {++ const body = Buffer.from('{"to":"x"}');++ const ts = String(nowSec());++ assert.strictEqual(mailSigned(INGRESS, body, { 'x-openvibe-mail-signature': `t=${ts},v1=${hmac(INGRESS, `${ts}.${body}`)}` }), true);++ assert.strictEqual(mailSigned(INGRESS, body, { 'x-openvibe-mail-signature': `t=${ts},v1=${hmac('other', `${ts}.${body}`)}` }), false);++ const old = String(nowSec() - 400);++ assert.strictEqual(mailSigned(INGRESS, body, { 'x-openvibe-mail-signature': `t=${old},v1=${hmac(INGRESS, `${old}.${body}`)}` }), false);++ assert.strictEqual(mailSigned(null, body, {}), false);++ assert.match(newMailLocal(), MAIL_LOCAL_RE);++ assert.notStrictEqual(newMailLocal(), newMailLocal());++});++++t('boot', async () => {++ svc = await boot({ env: { WATCH_MAIL_INGRESS_SECRET: INGRESS } });++ bare = await boot();++});++++t('without the ingress secret there is no email in', async () => {++ const made = await api(bare.base, 'POST', '/api/v1/watches', { token: ME, body: def('no mail') });++ assert.strictEqual(made.status, 201);++ const r = await api(bare.base, 'POST', `/api/v1/watches/${made.body.watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });++ assert.strictEqual(r.status, 422);++ assert.match(r.body.detail, /email in is not set up/);++ assert.strictEqual((await deliver(bare.base, '[email protected]', mailRaw('x'))).status, 404);++});++++t('the owner gives a webhook watch an address; it is shown on the watch\'s page; another person cannot', async () => {++ const made = await api(svc.base, 'POST', '/api/v1/watches', { token: ME, body: def('invoices') });++ assert.strictEqual(made.status, 201, JSON.stringify(made.body));++ watch = made.body.watch;++ assert.strictEqual((await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: OTHER, body: { provider: 'email' } })).status, 404);++ const r = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });++ assert.strictEqual(r.status, 200, JSON.stringify(r.body));++ address = r.body.webhook.address;++ assert.match(address, /^in-[0-9a-hjkmnp-tv-z]{26}@openvibe\.watch$/);++ assert.strictEqual(r.body.webhook.provider, 'email');++ assert.strictEqual(await svc.webhookSources.addressOf(watch.id), address);++ assert.strictEqual(await svc.webhookSources.providerOf(watch.id), 'email');++});++++t('mail to the address runs the check on its subject; the envelope sender is the "from"', async () => {++ const res = await deliver(svc.base, address.toUpperCase(), mailRaw(address), { from: '[email protected]' });++ assert.strictEqual(res.status, 200);++ assert.deepStrictEqual(await res.json(), { watch_id: watch.id, state: 'condition_met', triggered: true });++ const obs = await svc.db.prepare('SELECT * FROM observations WHERE watch_id = ? ORDER BY rid DESC').get(watch.id);++ assert.ok(obs, 'an observation was recorded');++ assert.match(JSON.stringify(obs), /Your invoice #42/);++ const quiet = await deliver(svc.base, address, mailRaw(address, 'Lunch on Friday?'));++ assert.strictEqual((await quiet.json()).triggered, false, 'a subject without "invoice" does not meet the condition');++ assert.strictEqual((await runs(watch.id)).length, 2);++});++++t('refused: a bad or old signature (401), an unknown address or domain (404), a paused watch (409)', async () => {++ assert.strictEqual((await deliver(svc.base, address, mailRaw(address), { secret: 'wrong' })).status, 401);++ assert.strictEqual((await deliver(svc.base, address, mailRaw(address), { at: String(nowSec() - 600) })).status, 401);++ assert.strictEqual((await deliver(svc.base, '[email protected]', mailRaw('x'))).status, 404);++ assert.strictEqual((await deliver(svc.base, address.replace('@openvibe.watch', '@elsewhere.example'), mailRaw('x'))).status, 404);++ assert.strictEqual((await deliver(svc.base, '[email protected]', mailRaw('x'))).status, 404);++ assert.strictEqual((await deliver(svc.base, watch.id + '@openvibe.watch', mailRaw('x'))).status, 404, 'a watch id is not an address');++ // The webhook address no longer takes Watch's own secret: the email address replaced it.++ const hook = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body: '{}', headers: { 'Content-Type': 'application/json', Authorization: 'Bearer whk_x' } });++ assert.strictEqual(hook.status, 401);++ await svc.db.prepare("UPDATE watches SET status = 'paused' WHERE id = ?").run(watch.id);++ assert.strictEqual((await deliver(svc.base, address, mailRaw(address))).status, 409);++ await svc.db.prepare("UPDATE watches SET status = 'active' WHERE id = ?").run(watch.id);++});++++t('a new address retires the old one; a webhook secret replaces the address', async () => {++ const again = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });++ const next = again.body.webhook.address;++ assert.notStrictEqual(next, address);++ assert.strictEqual((await deliver(svc.base, address, mailRaw(address))).status, 404, 'the old address is gone');++ assert.strictEqual((await deliver(svc.base, next, mailRaw(next))).status, 200);++ const own = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME });++ assert.match(own.body.webhook.secret, /^whk_/);++ assert.strictEqual(await svc.webhookSources.addressOf(watch.id), null);++ assert.strictEqual((await deliver(svc.base, next, mailRaw(next))).status, 404);++});++++t('the page offers an address only where email in is set up, and shows it on the watch', async () => {++ const pages = fs.readFileSync(path.join(__dirname, '..', 'server', 'http', 'pages.js'), 'utf8');++ assert.match(pages, /\$\{mailIn \? html`\n<details class="hook-provider"><summary>Wake it by email\?<\/summary>/);++ assert.match(pages, /<dt>Email address<\/dt><dd><code class="copyable">\$\{hookAddress\}<\/code>/);++ assert.match(pages, /hook\.provider === 'email' \? html`<h1>Your watch has an email address<\/h1>/);++});++++t('the Worker posts what Email Routing hands it, signed, and bounces an unknown address', async () => {++ const mod = await import(path.join(__dirname, '..', 'edge', 'mail-worker.mjs'));++ const fresh = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });++ const to = fresh.body.webhook.address;++ const msg = (rcpt) => {++ const raw = Buffer.from(mailRaw(rcpt));++ return { to: rcpt, from: '[email protected]', rawSize: raw.length, raw: new Blob([raw]).stream(), rejected: null, setReject(r) { this.rejected = r; } };++ };++ const env = { WATCH_URL: svc.base, INGRESS_SECRET: INGRESS };++ const before = (await runs(watch.id)).length;++ const ok = msg(to);++ await mod.default.email(ok, env);++ assert.strictEqual(ok.rejected, null, 'taken');++ assert.strictEqual((await runs(watch.id)).length, before + 1, 'the check ran');++ const unknown = msg('[email protected]');++ await mod.default.email(unknown, env);++ assert.strictEqual(unknown.rejected, 'No watch has this address.');++ const down = msg(to);++ await mod.default.email(down, { WATCH_URL: 'http://127.0.0.1:1', INGRESS_SECRET: INGRESS });++ assert.match(down.rejected, /not answering/);++});++++t('shutdown', async () => { await svc.stop(); await bare.stop(); });++++t.run();
@@ -98,6 +98,11 @@ VALKEY_PREFIX=
@@ -98,6 +98,11 @@ VALKEY_PREFIX=
#WATCH_HOOK_EXAMPLE_SECRET=
#WATCH_HOOK_EXAMPLE_SECRET=
# A receiver answers within this, or the attempt is retried on the backoff.
# A receiver answers within this, or the attempt is retried on the backoff.
#WATCH_WEBHOOK_TIMEOUT_MS=10000
#WATCH_WEBHOOK_TIMEOUT_MS=10000
+# Email in (agent plan A3): a webhook watch may have an address at WATCH_MAIL_DOMAIN. Cloudflare Email Routing hands each
+# message to the mail Worker (edge/mail-worker.mjs, deployed by deploy/mail-worker.js), which posts it to /hooks/email
+# signed with this secret (at least 32 characters; the Worker's INGRESS_SECRET). Unset: no email in.
+#WATCH_MAIL_INGRESS_SECRET=
+#WATCH_MAIL_DOMAIN=openvibe.watch
# OpenVibe.Actor's internal address, for actor-task and actor-agent actions (Watch needs actor.task.create and actor.agent.run grants).
# OpenVibe.Actor's internal address, for actor-task and actor-agent actions (Watch needs actor.task.create and actor.agent.run grants).
#WATCH_ACTOR_URL=http://127.0.0.1:4950
#WATCH_ACTOR_URL=http://127.0.0.1:4950
@@ -3,6 +3,13 @@
@@ -3,6 +3,13 @@
What shipped on OpenVibe.Watch, newest first. The plan is T18 ("Watch complete"); each entry names
What shipped on OpenVibe.Watch, newest first. The plan is T18 ("Watch complete"); each entry names
the step it landed.
the step it landed.
+## 2026-11-03 — email in (agent plan A3)
+
+- A webhook watch can have an email address of its own (`in-…@openvibe.watch`): mail sent there runs its check on the
+ sender, subject and text. Cloudflare Email Routing hands each message to the mail Worker (`edge/mail-worker.mjs`),
+ which posts it to `POST /hooks/email`, signed; an unknown address or a paused watch bounces the mail
+ (`server/webhook-sources.js`, `server/mail-parse.js`, migration `0016_webhook_email_address`).
+
## 2026-11-01 — the condition in words, a quiet test before arming, and the evidence in every alert
## 2026-11-01 — the condition in words, a quiet test before arming, and the evidence in every alert
- The watch list and each watch page say the watch's structured condition in one sentence — what it reads, where,
- The watch list and each watch page say the watch's structured condition in one sentence — what it reads, where,
@@ -219,6 +219,24 @@ product (`server/webhook-sources.js`, `server/carriers/webhook.js`).
@@ -219,6 +219,24 @@ product (`server/webhook-sources.js`, `server/carriers/webhook.js`).
("Got it") at once, with the check run right after on the interaction (`value_path: "data.name"` reads a slash
("Got it") at once, with the check run right after on the interaction (`value_path: "data.name"` reads a slash
command's name). A bad signature answers 401, as Discord checks.
command's name). A bad signature answers 401, as Discord checks.
- Cases: `test/webhook-providers.test.js`.
- Cases: `test/webhook-providers.test.js`.
+- **Email in (agent plan A3).** A webhook watch can have an email address of its own instead of a secret: "Wake it by
+ email?" on its page, or `POST /api/v1/watches/:id/webhook-secret` with `{ "provider": "email" }`, answers
+ `{ address: "in-<26 characters>@openvibe.watch" }`. The address is the key (anyone who has it can wake the watch);
+ asking again makes a new one and retires the old, and a webhook secret replaces it.
+ - Cloudflare Email Routing receives the domain's mail (SPF, DKIM, DMARC checked there) and its catch-all rule hands
+ every message to the mail Worker ([edge/mail-worker.mjs](edge/mail-worker.mjs)). The Worker posts
+ `{ to, from, raw (base64, at most 180 KB), truncated, size }` to `POST /hooks/email` with
+ `X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256 of "<time>.<body>">` (`WATCH_MAIL_INGRESS_SECRET`, within
+ five minutes).
+ - Watch finds the watch by the address, reads the message ([server/mail-parse.js](server/mail-parse.js): headers,
+ encoded words, multipart, base64 and quoted-printable, HTML-only mail as text) and runs the check on
+ `{ from, from_name, to, subject, date, message_id, text, text_truncated, attachments }` (`value_path: "subject"`).
+ `from` is the envelope sender Email Routing checked; `header_from` is what the message says.
+ - An unknown address answers 404 and a paused watch 409, and the Worker refuses the mail, so its sender gets a bounce.
+ Over the hour's budget the mail is taken and not checked; a busy watch is tried once more after 1.5 s.
+ - Deploy: `ssh host "sudo sed -n 's/^WATCH_MAIL_INGRESS_SECRET=//p' /etc/openvibe/watch.env" | node deploy/mail-worker.js --secret-stdin`
+ uploads the Worker, turns Email Routing on for the zone and points the catch-all at the Worker.
+ - Cases: `test/email-in.test.js`.
- **What the extraction reads.** A JSON delivery is parsed and becomes the document (`value_path: "order.total"`); any
- **What the extraction reads.** A JSON delivery is parsed and becomes the document (`value_path: "order.total"`); any
other is read as text (`text`, `regex`). A delivery that says JSON but is not records `parse_error`.
other is read as text (`text`, `regex`). A delivery that says JSON but is not records `parse_error`.
- **Answers.** 200 `{ watch_id, state, triggered }` after the check. 401 for a missing or wrong signature. 404 for an
- **Answers.** 200 `{ watch_id, state, triggered }` after the check. 401 for a missing or wrong signature. 404 for an
new file mode 100644
new file mode 100644
@@ -0,0 +1,79 @@
@@ -0,0 +1,79 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * Deploy Watch's email in (agent plan A3) on Cloudflare: the mail Worker (edge/mail-worker.mjs) with its bindings,
+ * Email Routing on the zone, and the catch-all rule that hands every message to the Worker.
+ *
+ * ssh host "sudo sed -n 's/^WATCH_MAIL_INGRESS_SECRET=//p' /etc/openvibe/watch.env" \
+ * | node deploy/mail-worker.js --secret-stdin [--zone openvibe.watch] [--watch-url https://openvibe.watch]
+ *
+ * The Cloudflare token comes from CF_API_TOKEN_FILE (default ~/.config/cloudflare-token-watch-mail): Workers Scripts
+ * Write and Account Settings Read on the account; Email Routing Rules Write, DNS Write, Zone Settings Write and Zone
+ * Read on the zone. The ingress secret is read from stdin (Watch's WATCH_MAIL_INGRESS_SECRET) and never printed.
+ * Without --secret-stdin the Worker's code is updated and its secret left as it is. Prints only what it did.
+ */
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+const args = process.argv.slice(2);
+const opt = (name, def) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] ? args[i + 1] : def; };
+const ZONE = opt('zone', 'openvibe.watch');
+const WATCH_URL = opt('watch-url', 'https://openvibe.watch');
+const SCRIPT = opt('name', 'openvibe-watch-mail');
+const TOKEN = fs.readFileSync(process.env.CF_API_TOKEN_FILE || path.join(os.homedir(), '.config', 'cloudflare-token-watch-mail'), 'utf8').trim();
+const API = 'https://api.cloudflare.com/client/v4';
+
+async function cf(method, p, body, { form = false } = {}) {
+ const res = await fetch(`${API}${p}`, {
+ method,
+ headers: { Authorization: `Bearer ${TOKEN}`, ...(body && !form ? { 'Content-Type': 'application/json' } : {}) },
+ body: body ? (form ? body : JSON.stringify(body)) : undefined,
+ });
+ const out = await res.json().catch(() => ({}));
+ if (!out.success) throw new Error(`${method} ${p}: ${res.status} ${JSON.stringify(out.errors || out).slice(0, 400)}`);
+ return out.result;
+}
+
+async function readStdin() {
+ const chunks = [];
+ for await (const c of process.stdin) chunks.push(c);
+ return Buffer.concat(chunks).toString('utf8').trim();
+}
+
+(async () => {
+ const secret = args.includes('--secret-stdin') ? await readStdin() : null;
+ if (secret !== null && secret.length < 32) throw new Error('the ingress secret is missing or shorter than 32 characters');
+
+ const zone = (await cf('GET', `/zones?name=${encodeURIComponent(ZONE)}`))[0];
+ if (!zone) throw new Error(`no zone ${ZONE}`);
+ const account = zone.account.id;
+
+ // 1. The Worker: its code and WATCH_URL; the secret binding is kept unless a new one is given.
+ const code = fs.readFileSync(path.join(__dirname, '..', 'edge', 'mail-worker.mjs'));
+ const bindings = [{ type: 'plain_text', name: 'WATCH_URL', text: WATCH_URL }];
+ if (secret) bindings.push({ type: 'secret_text', name: 'INGRESS_SECRET', text: secret });
+ else bindings.push({ type: 'inherit', name: 'INGRESS_SECRET' });
+ const form = new FormData();
+ form.append('metadata', new Blob([JSON.stringify({ main_module: 'mail-worker.mjs', compatibility_date: '2026-10-01', bindings })], { type: 'application/json' }));
+ form.append('mail-worker.mjs', new Blob([code], { type: 'application/javascript+module' }), 'mail-worker.mjs');
+ await cf('PUT', `/accounts/${account}/workers/scripts/${SCRIPT}`, form, { form: true });
+ console.log(`worker ${SCRIPT}: uploaded (${code.length} bytes)${secret ? ', ingress secret set' : ', ingress secret kept'}`);
+
+ // 2. Email Routing on the zone (its MX and SPF records), once.
+ const routing = await cf('GET', `/zones/${zone.id}/email/routing`);
+ if (!routing.enabled) {
+ await cf('POST', `/zones/${zone.id}/email/routing/enable`, {});
+ console.log(`email routing: enabled on ${ZONE}`);
+ } else console.log(`email routing: already on (${routing.status || 'ready'})`);
+
+ // 3. Every address of the zone goes to the Worker; Watch says which ones are a watch's.
+ await cf('PUT', `/zones/${zone.id}/email/routing/rules/catch_all`, {
+ name: 'OpenVibe.Watch email in', enabled: true,
+ matchers: [{ type: 'all' }],
+ actions: [{ type: 'worker', value: [SCRIPT] }],
+ });
+ console.log(`catch-all: every address at ${ZONE} → worker ${SCRIPT}`);
+ const dns = await cf('GET', `/zones/${zone.id}/email/routing/dns`).catch(() => null);
+ if (Array.isArray(dns)) console.log(`email routing records: ${dns.map((r) => `${r.type} ${r.name}`).join(', ')}`);
+})().catch((err) => { console.error(`mail-worker deploy failed: ${err.message}`); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,80 @@
@@ -0,0 +1,80 @@
+/**
+ * OpenVibe.Watch email in (agent plan A3): the Cloudflare Email Worker behind openvibe.watch's catch-all rule.
+ *
+ * Email Routing receives every message for the domain (its MX), checks SPF, DKIM and DMARC, and hands it here. The
+ * Worker posts it to Watch, signed, and refuses the mail when Watch knows no watch at that address, so the sender gets
+ * a bounce instead of silence.
+ *
+ * POST ${WATCH_URL}/hooks/email
+ * X-OpenVibe-Mail-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256(INGRESS_SECRET, t + "." + body)>
+ * { "to", "from", "raw": <the message, base64, at most 180 KB>, "truncated": <bool>, "size": <bytes> }
+ *
+ * Bindings: WATCH_URL (plain text, https://openvibe.watch) and INGRESS_SECRET (secret; Watch's
+ * WATCH_MAIL_INGRESS_SECRET). No dependencies. Deployed by deploy/mail-worker.sh.
+ */
+
+const MAX_RAW = 180 * 1024; // base64 makes it 4/3 bigger, and Watch takes 256 KB a delivery
+
+async function readRaw(stream, limit) {
+ const reader = stream.getReader();
+ const chunks = [];
+ let size = 0;
+ let truncated = false;
+ for (;;) {
+ const { done, value } = await reader.read();
+ if (done) break;
+ if (size + value.length > limit) {
+ chunks.push(value.subarray(0, limit - size));
+ size = limit;
+ truncated = true;
+ await reader.cancel();
+ break;
+ }
+ chunks.push(value);
+ size += value.length;
+ }
+ const out = new Uint8Array(size);
+ let at = 0;
+ for (const c of chunks) { out.set(c, at); at += c.length; }
+ return { bytes: out, truncated };
+}
+
+function base64(bytes) {
+ let s = '';
+ const step = 0x8000;
+ for (let i = 0; i < bytes.length; i += step) s += String.fromCharCode.apply(null, bytes.subarray(i, i + step));
+ return btoa(s);
+}
+
+async function sign(secret, data) {
+ const key = await crypto.subtle.importKey('raw', new TextEncoder().encode(secret), { name: 'HMAC', hash: 'SHA-256' }, false, ['sign']);
+ const mac = new Uint8Array(await crypto.subtle.sign('HMAC', key, new TextEncoder().encode(data)));
+ return Array.from(mac, (b) => b.toString(16).padStart(2, '0')).join('');
+}
+
+export default {
+ async email(message, env) {
+ const { bytes, truncated } = await readRaw(message.raw, MAX_RAW);
+ const body = JSON.stringify({ to: message.to, from: message.from, raw: base64(bytes), truncated, size: message.rawSize });
+ const url = `${String(env.WATCH_URL || 'https://openvibe.watch').replace(/\/+$/, '')}/hooks/email`;
+ const post = async () => {
+ const t = String(Math.floor(Date.now() / 1000));
+ const signature = `t=${t},v1=${await sign(env.INGRESS_SECRET, `${t}.${body}`)}`;
+ return fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-OpenVibe-Mail-Signature': signature, 'User-Agent': 'OpenVibe.Watch mail-worker' }, body });
+ };
+ let res;
+ try {
+ res = await post();
+ // The watch's previous check is still running: once more, a moment later.
+ if (res.status === 503) { await new Promise((r) => setTimeout(r, 1500)); res = await post(); }
+ } catch {
+ // Watch unreachable: Email Routing has no "try later", so the mail is refused with a reason the sender sees.
+ message.setReject('OpenVibe.Watch is not answering right now; please send it again later.');
+ return;
+ }
+ if (res.status === 404) message.setReject('No watch has this address.');
+ else if (res.status === 409) message.setReject('This watch is paused.');
+ else if (res.status === 413) message.setReject('The message is too large.');
+ // 200, 429 (over the hour's budget), 503 twice (busy), 5xx: taken; Watch logged what happened.
+ },
+};
new file mode 100644
new file mode 100644
@@ -0,0 +1,4 @@
@@ -0,0 +1,4 @@
+-- phase: expand
+-- OpenVibe.Watch email in (agent plan A3): a watch's email address (provider 'email') is its key, kept in
+-- webhook_secrets.secret as the address's local part (in-<26 characters>). Mail to the address finds its watch by it.
+CREATE UNIQUE INDEX IF NOT EXISTS webhook_secrets_email_address ON webhook_secrets (secret) WHERE provider = 'email';
@@ -91,6 +91,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
@@ -91,6 +91,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
}
}
// Webhook sources (plan T18 step 5, server/webhook-sources.js): public, signed with the watch's own secret, read raw
// Webhook sources (plan T18 step 5, server/webhook-sources.js): public, signed with the watch's own secret, read raw
// (the signature covers the exact bytes), so no body parser runs before it.
// (the signature covers the exact bytes), so no body parser runs before it.
+ // Email in (A3): the Cloudflare mail Worker posts each message here, signed with WATCH_MAIL_INGRESS_SECRET.
+ if (webhookSources) app.post('/hooks/email', (req, res) => { webhookSources.receiveMail(req, res).catch((err) => { log.error(`[webhook-sources] mail: ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
if (webhookSources) app.post('/hooks/:id', (req, res) => { webhookSources.receive(req, res).catch((err) => { log.error(`[webhook-sources] ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
if (webhookSources) app.post('/hooks/:id', (req, res) => { webhookSources.receive(req, res).catch((err) => { log.error(`[webhook-sources] ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
app.use('/api', express.json({ limit: '256kb', type: ['application/json', 'application/*+json'] }));
app.use('/api', express.json({ limit: '256kb', type: ['application/json', 'application/*+json'] }));
@@ -93,6 +93,12 @@ function load(env = process.env) {
@@ -93,6 +93,12 @@ function load(env = process.env) {
webhookSources: {
webhookSources: {
maxPerWatchPerHour: int(env.WATCH_WEBHOOK_MAX_PER_HOUR, 60),
maxPerWatchPerHour: int(env.WATCH_WEBHOOK_MAX_PER_HOUR, 60),
},
},
+ // Email in (agent plan A3, server/webhook-sources.js): addresses at WATCH_MAIL_DOMAIN; the Cloudflare mail Worker
+ // signs what it posts to /hooks/email with WATCH_MAIL_INGRESS_SECRET. Without the secret there is no email in.
+ mailIn: {
+ domain: (env.WATCH_MAIL_DOMAIN || 'openvibe.watch').trim().toLowerCase(),
+ ingressSecret: env.WATCH_MAIL_INGRESS_SECRET || null,
+ },
// Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired
// Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired
// watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).
// watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).
@@ -826,6 +826,8 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: CREATE_S
@@ -826,6 +826,8 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: CREATE_S
const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });
const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });
const acts = actions ? await actions.recent(row.id, 10) : [];
const acts = actions ? await actions.recent(row.id, 10) : [];
const hookProvider = v.source.kind === 'webhook' && ctx.webhookSources && ctx.webhookSources.providerOf ? await ctx.webhookSources.providerOf(row.id) : null;
const hookProvider = v.source.kind === 'webhook' && ctx.webhookSources && ctx.webhookSources.providerOf ? await ctx.webhookSources.providerOf(row.id) : null;
+ const hookAddress = hookProvider === 'email' && ctx.webhookSources.addressOf ? await ctx.webhookSources.addressOf(row.id) : null;
+ const mailIn = !!(config.mailIn && config.mailIn.ingressSecret);
const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'
const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'
? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`
? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`
: a.kind === 'codes-run'
: a.kind === 'codes-run'
@@ -891,10 +893,14 @@ ${ran ? notice(html`The check ran: <strong>${W.stateWords(ran)}</strong>.`, ran
@@ -891,10 +893,14 @@ ${ran ? notice(html`The check ran: <strong>${W.stateWords(ran)}</strong>.`, ran
${testRun ? testCheckNotice(testRun, testObs, v) : ''}
${testRun ? testCheckNotice(testRun, testObs, v) : ''}
<dl class="facts">
<dl class="facts">
<dt>Status</dt><dd>${badge(unarmed ? 'Not armed yet' : W.statusWords(v.status), unarmed ? 'warn' : W.statusKind(v.status))}</dd>
<dt>Status</dt><dd>${badge(unarmed ? 'Not armed yet' : W.statusWords(v.status), unarmed ? 'warn' : W.statusKind(v.status))}</dd>
-${v.source.kind === 'webhook'
+${v.source.kind === 'webhook' && hookAddress
+ ? html`<dt>Email address</dt><dd><code class="copyable">${hookAddress}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Make a new address</button></form><br><span class="muted small">Mail sent here wakes the watch. Anyone with the address can, so give it only to senders you mean; a new address retires this one.</span>
+<details class="hook-provider"><summary>Use a webhook instead?</summary><p class="muted small">A webhook address with a secret of Watch's own replaces the email address, which stops working at once.</p><form method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a webhook secret</button></form></details></dd>`
+ : v.source.kind === 'webhook'
? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">${hookProvider ? `Signed by ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hookProvider] || hookProvider} with the key you gave. A new secret of Watch's own replaces it.` : 'The secret was shown once, when the watch was made. A new one replaces it at once.'}</span>
? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">${hookProvider ? `Signed by ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hookProvider] || hookProvider} with the key you gave. A new secret of Watch's own replaces it.` : 'The secret was shown once, when the watch was made. A new one replaces it at once.'}</span>
<details class="hook-provider"><summary>Signed by Slack, Stripe or Discord?</summary><p class="muted small">They sign with a key of their own: paste it here and Watch checks their signature instead of its own. It is kept to check each delivery and never shown again.</p>
<details class="hook-provider"><summary>Signed by Slack, Stripe or Discord?</summary><p class="muted small">They sign with a key of their own: paste it here and Watch checks their signature instead of its own. It is kept to check each delivery and never shown again.</p>
-<form method="post" action="/watches/${v.id}/webhook-secret"><label>Sender <select name="provider"><option value="slack">Slack (signing secret)</option><option value="stripe">Stripe (endpoint secret, whsec_…)</option><option value="discord">Discord (application public key)</option></select></label> <label>Its signing secret <input name="secret" type="password" autocomplete="off" required maxlength="220"></label> <button class="btn" type="submit">Use it</button></form></details></dd>`
+<form method="post" action="/watches/${v.id}/webhook-secret"><label>Sender <select name="provider"><option value="slack">Slack (signing secret)</option><option value="stripe">Stripe (endpoint secret, whsec_…)</option><option value="discord">Discord (application public key)</option></select></label> <label>Its signing secret <input name="secret" type="password" autocomplete="off" required maxlength="220"></label> <button class="btn" type="submit">Use it</button></form></details>${mailIn ? html`
+<details class="hook-provider"><summary>Wake it by email?</summary><p class="muted small">The watch gets an address of its own at ${config.mailIn.domain}: mail sent there runs the check with its sender, subject and text. It replaces the webhook secret.</p><form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Give it an email address</button></form></details>` : ''}</dd>`
: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}
: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}
<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd>
<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd>
<dt>Actions</dt><dd>${configuredActions}</dd>
<dt>Actions</dt><dd>${configuredActions}</dd>
@@ -999,7 +1005,20 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:
@@ -999,7 +1005,20 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:
res.set('Referrer-Policy', 'no-referrer');
res.set('Referrer-Policy', 'no-referrer');
return page(req, res, {
return page(req, res, {
title: `${v.name}: webhook address`, private: true, crumbs: [{ label: 'Your watches', href: '/watches' }, { label: v.name, href: `/watches/${v.id}` }],
title: `${v.name}: webhook address`, private: true, crumbs: [{ label: 'Your watches', href: '/watches' }, { label: v.name, href: `/watches/${v.id}` }],
- body: hook.provider ? html`<h1>Your webhook is ready</h1>
+ body: hook.provider === 'email' ? html`<h1>Your watch has an email address</h1>
+${notice('Mail sent to this address wakes the watch. The address is the key: anyone who has it can wake the watch, so give it only to senders you mean.', 'ok')}
+<dl class="facts">
+<dt>Address</dt><dd><code class="copyable">${hook.address}</code></dd>
+</dl>
+<h2>Sending to it</h2>
+<ul class="facts">
+<li><strong>From a mailbox:</strong> make a filter that forwards the mail you care about (a sender, a subject) to this address.</li>
+<li><strong>From a service:</strong> put the address where it sends its notices (an alert, a form, a store's order mail).</li>
+<li><strong>What the watch reads:</strong> <code>from</code>, <code>from_name</code>, <code>to</code>, <code>subject</code>, <code>date</code>, <code>message_id</code>, <code>text</code> (the first 20,000 characters), <code>attachments</code> (how many). For example <code>value_path: "subject"</code> with "contains" <code>invoice</code>.</li>
+<li>Mail to an address no watch has is refused, so its sender gets a bounce; a paused watch refuses it too.</li>
+</ul>
+<p>${W.inWords(v)}</p>
+<p><a class="btn" href="/watches/${v.id}">Back to the watch</a></p>` : hook.provider ? html`<h1>Your webhook is ready</h1>
${notice(`Deliveries to this address are checked with the ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hook.provider]} key you gave. It is not shown again.`, 'ok')}
${notice(`Deliveries to this address are checked with the ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hook.provider]} key you gave. It is not shown again.`, 'ok')}
<dl class="facts">
<dl class="facts">
<dt>Address</dt><dd><code class="copyable">${hook.url}</code></dd>
<dt>Address</dt><dd><code class="copyable">${hook.url}</code></dd>
new file mode 100644
new file mode 100644
@@ -0,0 +1,147 @@
@@ -0,0 +1,147 @@
+'use strict';
+/**
+ * The parts of an email a watch reads (agent plan A3, email in): who sent it, to whom, the subject, the date, the
+ * message id and the text. Enough MIME for the mail people actually send: folded headers, RFC 2047 encoded words,
+ * multipart (nested, up to four deep), base64 and quoted-printable bodies, and the common charsets. The first
+ * text/plain part is the text; a message with only HTML gets its HTML as plain text. Attachments are counted, not kept.
+ *
+ * parseMail(raw: Buffer, { maxText }) → { from, to, subject, date, message_id, text, text_truncated, attachments }
+ */
+
+const MAX_DEPTH = 4;
+
+function decoderFor(charset) {
+ const label = String(charset || 'utf-8').trim().toLowerCase().replace(/^"|"$/g, '');
+ try { return new TextDecoder(label === 'us-ascii' || label === 'ascii' ? 'utf-8' : label); } catch { return new TextDecoder('utf-8'); }
+}
+
+/** Split a raw part into its header block and body (at the first empty line, CRLF or LF). */
+function splitPart(buf) {
+ const s = buf.toString('latin1');
+ const m = /\r?\n\r?\n/.exec(s);
+ if (!m) return { head: s, body: Buffer.alloc(0) };
+ return { head: s.slice(0, m.index), body: buf.subarray(m.index + m[0].length) };
+}
+
+/** Header block → Map of lower-case name → raw value (first one wins), with folded lines unfolded. */
+function headersOf(head) {
+ const out = new Map();
+ for (const line of head.replace(/\r?\n[ \t]+/g, ' ').split(/\r?\n/)) {
+ const i = line.indexOf(':');
+ if (i <= 0) continue;
+ const name = line.slice(0, i).trim().toLowerCase();
+ if (!out.has(name)) out.set(name, line.slice(i + 1).trim());
+ }
+ return out;
+}
+
+/** RFC 2047: =?charset?B|Q?text?= words, decoded; the whitespace between two encoded words goes. */
+function decodeWords(value) {
+ const raw = Buffer.from(String(value || ''), 'latin1').toString('latin1');
+ return raw.replace(/(=\?[^?\s]+\?[bBqQ]\?[^?\s]*\?=)\s+(?==\?[^?\s]+\?[bBqQ]\?[^?\s]*\?=)/g, '$1')
+ .replace(/=\?([^?\s]+)\?([bBqQ])\?([^?\s]*)\?=/g, (_, charset, enc, text) => {
+ try {
+ const bytes = enc.toUpperCase() === 'B'
+ ? Buffer.from(text, 'base64')
+ : Buffer.from(text.replace(/_/g, ' ').replace(/=([0-9A-Fa-f]{2})/g, (__, h) => String.fromCharCode(parseInt(h, 16))), 'latin1');
+ return decoderFor(charset.split('*')[0]).decode(bytes);
+ } catch { return text; }
+ });
+}
+
+/** A header's text: raw 8-bit bytes read as UTF-8 when they are UTF-8 (what most senders mean), then encoded words. */
+function headerText(value) {
+ let s = String(value || '');
+ if (/[\x80-\xff]/.test(s)) {
+ const u = Buffer.from(s, 'latin1').toString('utf8');
+ if (!u.includes('\ufffd')) s = u;
+ }
+ return decodeWords(s);
+}
+
+/** `type/subtype; a=b; c="d"` → { type, params } */
+function contentType(value) {
+ const [type, ...rest] = String(value || 'text/plain').split(';');
+ const params = {};
+ for (const p of rest) {
+ const i = p.indexOf('=');
+ if (i > 0) params[p.slice(0, i).trim().toLowerCase()] = p.slice(i + 1).trim().replace(/^"|"$/g, '');
+ }
+ return { type: type.trim().toLowerCase() || 'text/plain', params };
+}
+
+function decodeBody(body, encoding) {
+ const enc = String(encoding || '').trim().toLowerCase();
+ if (enc === 'base64') return Buffer.from(body.toString('latin1').replace(/[^A-Za-z0-9+/=]/g, ''), 'base64');
+ if (enc === 'quoted-printable') {
+ const s = body.toString('latin1').replace(/=\r?\n/g, '');
+ return Buffer.from(s.replace(/=([0-9A-Fa-f]{2})/g, (_, h) => String.fromCharCode(parseInt(h, 16))), 'latin1');
+ }
+ return body;
+}
+
+/** HTML as plain text: scripts and styles dropped, breaks kept, tags gone, the common entities read. */
+function htmlText(html) {
+ return String(html)
+ .replace(/<(script|style|head)[^>]*>[\s\S]*?<\/\1>/gi, ' ')
+ .replace(/<br\s*\/?>/gi, '\n').replace(/<\/(p|div|li|tr|h[1-6])>/gi, '\n')
+ .replace(/<[^>]+>/g, ' ')
+ .replace(/ /gi, ' ').replace(/</gi, '<').replace(/>/gi, '>').replace(/"/gi, '"').replace(/'|'/gi, "'")
+ .replace(/&#(\d{1,6});/g, (_, n) => { const c = Number(n); return c > 0 && c < 0x110000 ? String.fromCodePoint(c) : ' '; })
+ .replace(/&/gi, '&')
+ .replace(/[ \t]+\n/g, '\n').replace(/\n[ \t]+/g, '\n').replace(/\n{3,}/g, '\n\n').replace(/[ \t]{2,}/g, ' ').trim();
+}
+
+/** Walk a part (and its children): the first text/plain, else the first text/html, and how many attachments. */
+function walk(buf, depth, found) {
+ const { head, body } = splitPart(buf);
+ const h = headersOf(head);
+ const ct = contentType(h.get('content-type'));
+ const disposition = String(h.get('content-disposition') || '').toLowerCase();
+ if (ct.type.startsWith('multipart/') && ct.params.boundary && depth < MAX_DEPTH) {
+ const marker = `--${ct.params.boundary}`;
+ const s = body.toString('latin1');
+ const pieces = s.split(marker).slice(1);
+ for (const piece of pieces) {
+ if (piece.startsWith('--')) break; // the closing boundary
+ walk(Buffer.from(piece.replace(/^\r?\n/, ''), 'latin1'), depth + 1, found);
+ }
+ return;
+ }
+ if (disposition.startsWith('attachment') || (!ct.type.startsWith('text/') && !ct.type.startsWith('multipart/'))) { found.attachments++; return; }
+ if (ct.type === 'text/plain' && found.plain == null) found.plain = decoderFor(ct.params.charset).decode(decodeBody(body, h.get('content-transfer-encoding')));
+ else if (ct.type === 'text/html' && found.html == null) found.html = decoderFor(ct.params.charset).decode(decodeBody(body, h.get('content-transfer-encoding')));
+}
+
+/** "Ann <[email protected]>" or "[email protected]" → "[email protected]" (lower-cased domain kept as written). */
+function addressOf(value) {
+ const s = headerText(value);
+ const m = /<([^<>\s]+@[^<>\s]+)>/.exec(s) || /([^\s<>",;]+@[^\s<>",;]+)/.exec(s);
+ return m ? m[1] : '';
+}
+
+function parseMail(raw, { maxText = 20000 } = {}) {
+ const buf = Buffer.isBuffer(raw) ? raw : Buffer.from(String(raw || ''), 'utf8');
+ const { head } = splitPart(buf);
+ const h = headersOf(head);
+ const found = { plain: null, html: null, attachments: 0 };
+ walk(buf, 0, found);
+ let text = found.plain != null ? found.plain : (found.html != null ? htmlText(found.html) : '');
+ text = text.replace(/\r\n/g, '\n').trim();
+ const truncated = text.length > maxText;
+ if (truncated) text = text.slice(0, maxText);
+ const date = h.get('date') ? new Date(h.get('date')) : null;
+ return {
+ from: addressOf(h.get('from')),
+ from_name: headerText(h.get('from') || '').replace(/<[^>]*>/, '').trim().replace(/^"|"$/g, '').trim().slice(0, 200),
+ to: addressOf(h.get('to')),
+ subject: headerText(h.get('subject') || '').slice(0, 500),
+ date: date && !Number.isNaN(date.getTime()) ? date.toISOString() : null,
+ message_id: String(h.get('message-id') || '').replace(/^<|>$/g, '').slice(0, 300),
+ text,
+ text_truncated: truncated,
+ attachments: found.attachments,
+ };
+}
+
+module.exports = { parseMail, htmlText, decodeWords };
@@ -24,12 +24,22 @@
@@ -24,12 +24,22 @@
* secret anyway). Discord wants an interaction reply within 3 seconds: a PING gets
* secret anyway). Discord wants an interaction reply within 3 seconds: a PING gets
* { type: 1 }, anything else { type: 4, ephemeral "Got it" } at once, and the check runs
* { type: 1 }, anything else { type: 4, ephemeral "Got it" } at once, and the check runs
* right after, with the interaction as its input.
* right after, with the interaction as its input.
+ * Email in (A3) POST …/webhook-secret { provider: 'email' } gives the watch an address of its own,
+ * in-<26 characters>@openvibe.watch (WATCH_MAIL_DOMAIN). Cloudflare Email Routing hands every message
+ * for the domain to a Worker (edge/mail-worker.js), which posts it to POST /hooks/email signed with
+ * WATCH_MAIL_INGRESS_SECRET (X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256(secret,
+ * time + "." + body)>, within five minutes). The address finds the watch; the message is read
+ * (server/mail-parse.js) and the check runs with { from, from_name, to, subject, date, message_id,
+ * text, text_truncated, attachments } as its JSON input. An unknown address answers 404, and the
+ * Worker refuses the mail, so the sender gets a bounce. The address is the key: anyone who has it
+ * can wake the watch, and making a new one retires the old.
* Answers 200 { state, triggered } after the check; 404 for an unknown id or a watch that is not a webhook
* Answers 200 { state, triggered } after the check; 404 for an unknown id or a watch that is not a webhook
* watch; 401 for a missing or wrong signature (the same answer either way); 409 for a paused watch;
* watch; 401 for a missing or wrong signature (the same answer either way); 409 for a paused watch;
* 429 over the watch's hourly budget (WATCH_WEBHOOK_MAX_PER_HOUR, default 60) with Retry-After;
* 429 over the watch's hourly budget (WATCH_WEBHOOK_MAX_PER_HOUR, default 60) with Retry-After;
* 503 with Retry-After: 1 while the watch's previous check still runs; 413 over 256 KB.
* 503 with Retry-After: 1 while the watch's previous check still runs; 413 over 256 KB.
*/
*/
const crypto = require('crypto');
const crypto = require('crypto');
+const { parseMail } = require('./mail-parse');
const SECRET_RE = /^whk_[A-Za-z0-9_-]{43}$/;
const SECRET_RE = /^whk_[A-Za-z0-9_-]{43}$/;
const TOLERANCE_SEC = 300;
const TOLERANCE_SEC = 300;
@@ -40,6 +50,10 @@ const PROVIDERS = Object.freeze({
@@ -40,6 +50,10 @@ const PROVIDERS = Object.freeze({
discord: { name: 'Discord', secret: /^[0-9a-f]{64}$/, how: 'Discord signs each interaction with your application\'s key (X-Signature-Ed25519); paste its public key.' },
discord: { name: 'Discord', secret: /^[0-9a-f]{64}$/, how: 'Discord signs each interaction with your application\'s key (X-Signature-Ed25519); paste its public key.' },
});
});
const MAX_BODY = 256 * 1024;
const MAX_BODY = 256 * 1024;
+// An email address of a watch's own: its local part is the key (Crockford base32, 26 characters, 130 bits).
+const MAIL_LOCAL_RE = /^in-[0-9a-hjkmnp-tv-z]{26}$/;
+const B32 = '0123456789abcdefghjkmnpqrstvwxyz';
+const newMailLocal = () => `in-${Array.from(crypto.randomBytes(26), (b) => B32[b & 31]).join('')}`;
const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;
const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;
const endpointFor = (watchId) => `/internal/hooks/${watchId}`;
const endpointFor = (watchId) => `/internal/hooks/${watchId}`;
@@ -110,12 +124,17 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -110,12 +124,17 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
const maxPerHour = Number.isInteger(config.webhookSources && config.webhookSources.maxPerWatchPerHour) && config.webhookSources.maxPerWatchPerHour > 0
const maxPerHour = Number.isInteger(config.webhookSources && config.webhookSources.maxPerWatchPerHour) && config.webhookSources.maxPerWatchPerHour > 0
? config.webhookSources.maxPerWatchPerHour : 60;
? config.webhookSources.maxPerWatchPerHour : 60;
const counts = new Map(); // watch id → { hour, n }
const counts = new Map(); // watch id → { hour, n }
+ const mail = {
+ domain: String((config.mailIn && config.mailIn.domain) || 'openvibe.watch').toLowerCase(),
+ ingressSecret: (config.mailIn && config.mailIn.ingressSecret) || null,
+ };
const st = {
const st = {
set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider) VALUES (?, ?, ?, ?)
set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider) VALUES (?, ?, ?, ?)
ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = excluded.provider`),
ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = excluded.provider`),
get: db.prepare('SELECT secret, provider FROM webhook_secrets WHERE watch_id = ?'),
get: db.prepare('SELECT secret, provider FROM webhook_secrets WHERE watch_id = ?'),
drop: db.prepare('DELETE FROM webhook_secrets WHERE watch_id = ?'),
drop: db.prepare('DELETE FROM webhook_secrets WHERE watch_id = ?'),
watch: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
watch: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
+ byMail: db.prepare("SELECT watch_id FROM webhook_secrets WHERE provider = 'email' AND secret = ?"),
};
};
/**
/**
@@ -125,6 +144,12 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -125,6 +144,12 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
*/
*/
async function issue(watchId, { provider = null, secret: theirs = null } = {}) {
async function issue(watchId, { provider = null, secret: theirs = null } = {}) {
const url = `${config.baseUrl}/hooks/${watchId}`;
const url = `${config.baseUrl}/hooks/${watchId}`;
+ if (provider === 'email') {
+ if (!mail.ingressSecret) return { error: 'email in is not set up on this server' };
+ const local = newMailLocal();
+ await st.set.run(watchId, local, now(), 'email');
+ return { url, provider: 'email', address: `${local}@${mail.domain}`, signature: 'Mail sent to this address wakes the watch: the address is the key.' };
+ }
if (provider != null && provider !== '') {
if (provider != null && provider !== '') {
const p = PROVIDERS[provider];
const p = PROVIDERS[provider];
if (!p) return { error: `provider is one of ${Object.keys(PROVIDERS).join(', ')}` };
if (!p) return { error: `provider is one of ${Object.keys(PROVIDERS).join(', ')}` };
@@ -138,6 +163,11 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -138,6 +163,11 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
return { url, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, Authorization: Bearer <secret>, or Telegram\'s X-Telegram-Bot-Api-Secret-Token' };
return { url, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, Authorization: Bearer <secret>, or Telegram\'s X-Telegram-Bot-Api-Secret-Token' };
}
}
async function forget(watchId) { await st.drop.run(watchId); }
async function forget(watchId) { await st.drop.run(watchId); }
+ /** The watch's email address, when it has one (shown to its owner on the watch's page). */
+ async function addressOf(watchId) {
+ const row = await st.get.get(watchId);
+ return row && row.provider === 'email' && MAIL_LOCAL_RE.test(row.secret) ? `${row.secret}@${mail.domain}` : null;
+ }
/** Whose secret signs this watch's deliveries: null for Watch's own, else 'slack' or 'stripe'. */
/** Whose secret signs this watch's deliveries: null for Watch's own, else 'slack' or 'stripe'. */
async function providerOf(watchId) { return ((await st.get.get(watchId)) || {}).provider || null; }
async function providerOf(watchId) { return ((await st.get.get(watchId)) || {}).provider || null; }
@@ -215,7 +245,51 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -215,7 +245,51 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
return res.json({ watch_id: id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });
return res.json({ watch_id: id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });
}
}
- return { issue, forget, providerOf, receive, endpointFor, verified, SECRET_RE, PROVIDERS };
+ /**
+ * POST /hooks/email — a message from the mail Worker (edge/mail-worker.js): { to, from, raw (base64), truncated },
+ * signed with WATCH_MAIL_INGRESS_SECRET. Answers as /hooks/:id does; 404 makes the Worker refuse the mail.
+ */
+ async function receiveMail(req, res) {
+ let raw;
+ try { raw = await readRaw(req); } catch (err) { return problem(res, err.status || 400, err.status === 413 ? 'watch.too_large' : 'watch.bad_request', err.status === 413 ? 'a delivery is at most 256 KB' : 'unreadable body'); }
+ if (!mail.ingressSecret) return problem(res, 404, 'watch.not_found', 'email in is not set up on this server');
+ if (!mailSigned(mail.ingressSecret, raw, req.headers, now())) return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed by the mail ingress');
+ let msg;
+ try { msg = JSON.parse(raw.toString('utf8')); } catch { return problem(res, 400, 'watch.bad_request', 'the delivery is not JSON'); }
+ const to = String((msg && msg.to) || '').trim().toLowerCase();
+ const at = to.lastIndexOf('@');
+ const local = at > 0 ? to.slice(0, at) : '';
+ const found = at > 0 && to.slice(at + 1) === mail.domain && MAIL_LOCAL_RE.test(local) ? await st.byMail.get(local) : null;
+ const row = found ? await st.watch.get(found.watch_id) : null;
+ const source = row ? (typeof row.source === 'string' ? JSON.parse(row.source) : row.source) : null;
+ if (!row || !source || source.kind !== 'webhook') return problem(res, 404, 'watch.not_found', 'no watch has this address');
+ if (row.status !== 'active') return problem(res, 409, 'watch.disabled', `the watch is ${row.status}`);
+ if (!underBudget(row.id)) return problem(res, 429, 'watch.rate_limited', `at most ${maxPerHour} deliveries an hour wake one watch`, { 'Retry-After': String(retryAfter()) });
+ const parsed = parseMail(Buffer.from(String(msg.raw || ''), 'base64'));
+ // The envelope is what Email Routing checked; the headers are what the sender wrote.
+ const input = { ...parsed, from: String(msg.from || parsed.from || '').slice(0, 320), to, header_from: parsed.from, truncated: !!msg.truncated };
+ let out;
+ try {
+ out = await check.run(row.id, { trigger: 'webhook', input: { body: Buffer.from(JSON.stringify(input)), content_type: 'application/json' } });
+ } catch (err) {
+ log.error(`[webhook-sources] mail ${row.id}: ${err.message}`);
+ return problem(res, 500, 'watch.internal', 'the check failed; try again');
+ }
+ if (out && out.busy) return problem(res, 503, 'watch.busy', 'the watch\'s previous check is still running', { 'Retry-After': '1' });
+ res.setHeader('Cache-Control', 'no-store');
+ return res.json({ watch_id: row.id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });
+ }
+
+ return { issue, forget, providerOf, addressOf, receive, receiveMail, endpointFor, verified, SECRET_RE, PROVIDERS };
+}
+
+/** The mail Worker's signature: X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256(secret, time + "." + body)>. */
+function mailSigned(secret, raw, headers, nowMs = Date.now()) {
+ const parts = String(headers['x-openvibe-mail-signature'] || '').split(',').map((x) => x.trim().split('='));
+ const t = (parts.find(([k]) => k === 't') || [])[1];
+ if (!secret || !fresh(t, nowMs)) return false;
+ const want = hmac(secret, Buffer.concat([Buffer.from(`${t}.`), raw]));
+ return parts.filter(([k]) => k === 'v1').some(([, v]) => safeEqual(v || '', want));
}
}
-module.exports = { createWebhookSources, verified, endpointFor, newSecret, PROVIDERS };
+module.exports = { createWebhookSources, verified, mailSigned, endpointFor, newSecret, newMailLocal, MAIL_LOCAL_RE, PROVIDERS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,212 @@
@@ -0,0 +1,212 @@
+'use strict';
+/**
+ * Email in (agent plan A3): a webhook watch can have an email address of its own. Cloudflare Email Routing hands each
+ * message for openvibe.watch to the mail Worker (edge/mail-worker.mjs), which posts it to POST /hooks/email signed
+ * with WATCH_MAIL_INGRESS_SECRET; the address finds the watch, the message is read (server/mail-parse.js) and the
+ * check runs on { from, subject, text, … }. An unknown address is 404 (the Worker bounces the mail); a paused watch 409;
+ * a bad or old signature 401. Without the ingress secret there is no email in.
+ */
+const assert = require('assert');
+const crypto = require('crypto');
+const fs = require('fs');
+const path = require('path');
+const { boot, request, userToken, suite, newUser } = require('./helpers');
+const { parseMail, htmlText } = require('../server/mail-parse');
+const { mailSigned, MAIL_LOCAL_RE, newMailLocal } = require('../server/webhook-sources');
+
+const t = suite('email-in');
+const INGRESS = 'mail-ingress-test-secret-0123456789';
+let svc;
+let bare;
+const me = newUser();
+const ME = userToken(me);
+const OTHER = userToken(newUser());
+const api = (base, method, p, opts = {}) => request(base, method, p, opts);
+const hmac = (secret, data) => crypto.createHmac('sha256', secret).update(data).digest('hex');
+const nowSec = () => Math.floor(Date.now() / 1000);
+const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);
+const def = (name) => ({ name, source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json', value_path: 'subject' }, condition: { op: 'contains', value: 'invoice' }, action: [{ kind: 'notification' }] });
+
+const CRLF = (s) => s.replace(/\r?\n/g, '\r\n');
+const mailRaw = (to, subject = 'Your invoice #42', text = 'Total: 12.00 EUR') => CRLF(`From: "Ann Example" <[email protected]>
+To: ${to}
+Subject: ${subject}
+Date: Mon, 03 Nov 2026 05:00:00 +0000
+Message-ID: <[email protected]>
+MIME-Version: 1.0
+Content-Type: text/plain; charset=utf-8
+
+${text}
+`);
+const deliver = (base, to, raw, { secret = INGRESS, at = String(nowSec()), from = '[email protected]' } = {}) => {
+ const body = JSON.stringify({ to, from, raw: Buffer.from(raw).toString('base64'), truncated: false, size: raw.length });
+ return fetch(`${base}/hooks/email`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', 'X-OpenVibe-Mail-Signature': `t=${at},v1=${hmac(secret, `${at}.${body}`)}` } });
+};
+let watch;
+let address;
+
+t('the reader: headers, encoded words, multipart, quoted-printable, base64, HTML-only, attachments', () => {
+ const plain = parseMail(Buffer.from(mailRaw('[email protected]')));
+ assert.strictEqual(plain.from, '[email protected]');
+ assert.strictEqual(plain.from_name, 'Ann Example');
+ assert.strictEqual(plain.subject, 'Your invoice #42');
+ assert.strictEqual(plain.date, '2026-11-03T05:00:00.000Z');
+ assert.strictEqual(plain.message_id, '[email protected]');
+ assert.strictEqual(plain.text, 'Total: 12.00 EUR');
+ const multi = parseMail(Buffer.from(CRLF(`From: =?UTF-8?B?w4Ruw6k=?= <[email protected]>
+To: [email protected]
+Subject: =?utf-8?Q?Caf=C3=A9_order?= =?utf-8?Q?_shipped?=
+Content-Type: multipart/mixed; boundary="outer"
+
+--outer
+Content-Type: multipart/alternative; boundary=inner
+
+--inner
+Content-Type: text/plain; charset=utf-8
+Content-Transfer-Encoding: quoted-printable
+
+Caf=C3=A9 order #7 has shipped. A long line that was =
+wrapped.
+--inner
+Content-Type: text/html; charset=utf-8
+
+<p>Café order</p>
+--inner--
+--outer
+Content-Type: application/pdf; name="label.pdf"
+Content-Disposition: attachment; filename="label.pdf"
+Content-Transfer-Encoding: base64
+
+JVBERi0xLjQK
+--outer--
+`)));
+ assert.strictEqual(multi.from_name, 'Äné');
+ assert.strictEqual(multi.subject, 'Café order shipped', 'adjacent encoded words join');
+ assert.strictEqual(multi.text, 'Café order #7 has shipped. A long line that was wrapped.');
+ assert.strictEqual(multi.attachments, 1);
+ const html = parseMail(Buffer.from(CRLF(`From: [email protected]
+Subject: Receipt
+Content-Type: text/html; charset=iso-8859-1
+Content-Transfer-Encoding: base64
+
+${Buffer.from('<html><head><style>p{}</style></head><body><p>Thanks!</p><p>Total & tax: 5 \xa3</p></body></html>', 'latin1').toString('base64')}
+`)));
+ assert.strictEqual(html.text, 'Thanks!\nTotal & tax: 5 £');
+ assert.strictEqual(htmlText('<script>x()</script>a<br>b'), 'a\nb');
+ const long = parseMail(Buffer.from(mailRaw('x@y', 's', 'z'.repeat(30000))));
+ assert.strictEqual(long.text.length, 20000);
+ assert.strictEqual(long.text_truncated, true);
+ assert.deepStrictEqual(parseMail(Buffer.from('garbage')).subject, '');
+});
+
+t('the signature: t and v1 over "<t>.<body>", within five minutes', () => {
+ const body = Buffer.from('{"to":"x"}');
+ const ts = String(nowSec());
+ assert.strictEqual(mailSigned(INGRESS, body, { 'x-openvibe-mail-signature': `t=${ts},v1=${hmac(INGRESS, `${ts}.${body}`)}` }), true);
+ assert.strictEqual(mailSigned(INGRESS, body, { 'x-openvibe-mail-signature': `t=${ts},v1=${hmac('other', `${ts}.${body}`)}` }), false);
+ const old = String(nowSec() - 400);
+ assert.strictEqual(mailSigned(INGRESS, body, { 'x-openvibe-mail-signature': `t=${old},v1=${hmac(INGRESS, `${old}.${body}`)}` }), false);
+ assert.strictEqual(mailSigned(null, body, {}), false);
+ assert.match(newMailLocal(), MAIL_LOCAL_RE);
+ assert.notStrictEqual(newMailLocal(), newMailLocal());
+});
+
+t('boot', async () => {
+ svc = await boot({ env: { WATCH_MAIL_INGRESS_SECRET: INGRESS } });
+ bare = await boot();
+});
+
+t('without the ingress secret there is no email in', async () => {
+ const made = await api(bare.base, 'POST', '/api/v1/watches', { token: ME, body: def('no mail') });
+ assert.strictEqual(made.status, 201);
+ const r = await api(bare.base, 'POST', `/api/v1/watches/${made.body.watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });
+ assert.strictEqual(r.status, 422);
+ assert.match(r.body.detail, /email in is not set up/);
+ assert.strictEqual((await deliver(bare.base, '[email protected]', mailRaw('x'))).status, 404);
+});
+
+t('the owner gives a webhook watch an address; it is shown on the watch\'s page; another person cannot', async () => {
+ const made = await api(svc.base, 'POST', '/api/v1/watches', { token: ME, body: def('invoices') });
+ assert.strictEqual(made.status, 201, JSON.stringify(made.body));
+ watch = made.body.watch;
+ assert.strictEqual((await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: OTHER, body: { provider: 'email' } })).status, 404);
+ const r = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.body));
+ address = r.body.webhook.address;
+ assert.match(address, /^in-[0-9a-hjkmnp-tv-z]{26}@openvibe\.watch$/);
+ assert.strictEqual(r.body.webhook.provider, 'email');
+ assert.strictEqual(await svc.webhookSources.addressOf(watch.id), address);
+ assert.strictEqual(await svc.webhookSources.providerOf(watch.id), 'email');
+});
+
+t('mail to the address runs the check on its subject; the envelope sender is the "from"', async () => {
+ const res = await deliver(svc.base, address.toUpperCase(), mailRaw(address), { from: '[email protected]' });
+ assert.strictEqual(res.status, 200);
+ assert.deepStrictEqual(await res.json(), { watch_id: watch.id, state: 'condition_met', triggered: true });
+ const obs = await svc.db.prepare('SELECT * FROM observations WHERE watch_id = ? ORDER BY rid DESC').get(watch.id);
+ assert.ok(obs, 'an observation was recorded');
+ assert.match(JSON.stringify(obs), /Your invoice #42/);
+ const quiet = await deliver(svc.base, address, mailRaw(address, 'Lunch on Friday?'));
+ assert.strictEqual((await quiet.json()).triggered, false, 'a subject without "invoice" does not meet the condition');
+ assert.strictEqual((await runs(watch.id)).length, 2);
+});
+
+t('refused: a bad or old signature (401), an unknown address or domain (404), a paused watch (409)', async () => {
+ assert.strictEqual((await deliver(svc.base, address, mailRaw(address), { secret: 'wrong' })).status, 401);
+ assert.strictEqual((await deliver(svc.base, address, mailRaw(address), { at: String(nowSec() - 600) })).status, 401);
+ assert.strictEqual((await deliver(svc.base, '[email protected]', mailRaw('x'))).status, 404);
+ assert.strictEqual((await deliver(svc.base, address.replace('@openvibe.watch', '@elsewhere.example'), mailRaw('x'))).status, 404);
+ assert.strictEqual((await deliver(svc.base, '[email protected]', mailRaw('x'))).status, 404);
+ assert.strictEqual((await deliver(svc.base, watch.id + '@openvibe.watch', mailRaw('x'))).status, 404, 'a watch id is not an address');
+ // The webhook address no longer takes Watch's own secret: the email address replaced it.
+ const hook = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body: '{}', headers: { 'Content-Type': 'application/json', Authorization: 'Bearer whk_x' } });
+ assert.strictEqual(hook.status, 401);
+ await svc.db.prepare("UPDATE watches SET status = 'paused' WHERE id = ?").run(watch.id);
+ assert.strictEqual((await deliver(svc.base, address, mailRaw(address))).status, 409);
+ await svc.db.prepare("UPDATE watches SET status = 'active' WHERE id = ?").run(watch.id);
+});
+
+t('a new address retires the old one; a webhook secret replaces the address', async () => {
+ const again = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });
+ const next = again.body.webhook.address;
+ assert.notStrictEqual(next, address);
+ assert.strictEqual((await deliver(svc.base, address, mailRaw(address))).status, 404, 'the old address is gone');
+ assert.strictEqual((await deliver(svc.base, next, mailRaw(next))).status, 200);
+ const own = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME });
+ assert.match(own.body.webhook.secret, /^whk_/);
+ assert.strictEqual(await svc.webhookSources.addressOf(watch.id), null);
+ assert.strictEqual((await deliver(svc.base, next, mailRaw(next))).status, 404);
+});
+
+t('the page offers an address only where email in is set up, and shows it on the watch', async () => {
+ const pages = fs.readFileSync(path.join(__dirname, '..', 'server', 'http', 'pages.js'), 'utf8');
+ assert.match(pages, /\$\{mailIn \? html`\n<details class="hook-provider"><summary>Wake it by email\?<\/summary>/);
+ assert.match(pages, /<dt>Email address<\/dt><dd><code class="copyable">\$\{hookAddress\}<\/code>/);
+ assert.match(pages, /hook\.provider === 'email' \? html`<h1>Your watch has an email address<\/h1>/);
+});
+
+t('the Worker posts what Email Routing hands it, signed, and bounces an unknown address', async () => {
+ const mod = await import(path.join(__dirname, '..', 'edge', 'mail-worker.mjs'));
+ const fresh = await api(svc.base, 'POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'email' } });
+ const to = fresh.body.webhook.address;
+ const msg = (rcpt) => {
+ const raw = Buffer.from(mailRaw(rcpt));
+ return { to: rcpt, from: '[email protected]', rawSize: raw.length, raw: new Blob([raw]).stream(), rejected: null, setReject(r) { this.rejected = r; } };
+ };
+ const env = { WATCH_URL: svc.base, INGRESS_SECRET: INGRESS };
+ const before = (await runs(watch.id)).length;
+ const ok = msg(to);
+ await mod.default.email(ok, env);
+ assert.strictEqual(ok.rejected, null, 'taken');
+ assert.strictEqual((await runs(watch.id)).length, before + 1, 'the check ran');
+ const unknown = msg('[email protected]');
+ await mod.default.email(unknown, env);
+ assert.strictEqual(unknown.rejected, 'No watch has this address.');
+ const down = msg(to);
+ await mod.default.email(down, { WATCH_URL: 'http://127.0.0.1:1', INGRESS_SECRET: INGRESS });
+ assert.match(down.rejected, /not answering/);
+});
+
+t('shutdown', async () => { await svc.stop(); await bare.stop(); });
+
+t.run();