openvibers/openvibe.watch Commits
Watch: webhook sources: a watch woken by a signed POST to /hooks/<id> (plan T18 step 5) (#41)
Changes
@@@ -89,7 +89,7 @@ every page and every form, never 403. A signed-in write must also come from thisThe cadence is a select (15 minutes, 1 hour, 6 hours, 1 day); 15 minutes is the shortest a person maychoose here (the contract's floor is 30 seconds, for a service). A validation error re-renders theform with the person's own input and the registry's own message next to the field it names. The--kinds whose carrier does not exist yet (webhook, run, node) and the `ai` extraction are **not++kinds whose carrier does not exist yet (run, node) and the push kinds (event, webhook) and the `ai` extraction are **notoffered** — and a watch of such a kind created through the API still answers `watch.carrier_unavailable`when checked from the site, fetching nothing. An event watch (made through the API) is checked when its eventarrives, so Check now answers `watch.event_awaited`.@@@ -130,7 +130,7 @@ This release carries the pull rungs and event sources:| `feed` | RSS 2.0 / RSS 1.0 (RDF) / Atom 1.0 → `{ items, latest, count }` | implemented || `api` | JSON/XML with `items_path` + `fields`, or the body itself | implemented || `event` | delivered by OpenVibe.Events (see [Event sources](#event-sources)) | implemented |--| `webhook` | delivered to `/internal/hooks/…` | step 5 |++| `webhook` | a signed POST to `/hooks/<watch id>` (see [Webhook sources](#webhook-sources)) | implemented || `run` | browser/code polling through OpenVibe.Run | step 6 || `node` | a probe through OpenVibe.Node | step 7 |@@@ -163,6 +163,28 @@ nothing while nothing happens (`server/event-sources.js`, `server/carriers/eventsubscriptions or another endpoint's. It is off without `EVENTS_URL`, the secret or the OAuth client secret.- **Check now** on an event watch answers 422 `watch.event_awaited`: there is no event to read.++## Webhook sources++++A watch whose source is `{ "kind": "webhook" }` (and `cadence: null`) is woken by a POST to++`https://openvibe.watch/hooks/<watch id>` from anything that can call a webhook: GitHub, a CI, a store, another OpenVibe++product (`server/webhook-sources.js`, `server/carriers/webhook.js`).++++- **The address is the watch's own.** Its stored endpoint is `/internal/hooks/<watch id>` (the form `watch.watch@1`++ names), whatever a caller sent, so a watch can never point at another watch's address.++- **The secret is shown once.** Creating the watch answers `{ watch, webhook: { url, secret } }`. The secret++ (`whk_…`) is never shown again; `POST /api/v1/watches/:id/webhook-secret` replaces it (the old one stops working at++ once). It is kept in full because checking a signature needs it (`webhook_secrets`), and it goes with its watch:++ deleting the watch or the account removes it.++- **Signing.** `X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the exact body with the secret>` (GitHub's form), or++ `Authorization: Bearer <secret>` for a sender that cannot sign. Compared in constant time.++- **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`.++- **Answers.** 200 `{ watch_id, state, triggered }` after the check. 401 for a missing or wrong signature. 404 for an++ unknown address or a watch that is not a webhook watch. 409 for a paused watch. 413 over 256 KB. 429 with++ `Retry-After` over `WATCH_WEBHOOK_MAX_PER_HOUR` (60) deliveries an hour. 503 with `Retry-After: 1` while the watch's++ previous check still runs.++- **Check now** on a webhook watch answers 422 `watch.webhook_awaited`: there is no delivery to read.++## A checkAn in-process scheduler ([server/scheduler.js](server/scheduler.js)) starts checks for active@@@ -260,7 +282,7 @@ an app or module may not.| Capability | Routes ||---|---|| `watch.watch.read` | `GET /api/v1/watches?status=&project_id=&before=&limit=`, `GET /api/v1/watches/:id` |--| `watch.watch.manage` | `POST /api/v1/watches`, `PATCH /api/v1/watches/:id`, `POST /api/v1/watches/:id/pause`, `POST /api/v1/watches/:id/resume`, `DELETE /api/v1/watches/:id` |++| `watch.watch.manage` | `POST /api/v1/watches`, `PATCH /api/v1/watches/:id`, `POST /api/v1/watches/:id/pause`, `POST /api/v1/watches/:id/resume`, `POST /api/v1/watches/:id/webhook-secret`, `DELETE /api/v1/watches/:id` || `watch.observation.read` | `GET /api/v1/watches/:id/observations?before=&limit=`, `GET /api/v1/watches/:id/checks?before=&limit=` || `watch.check.run` | `POST /api/v1/watches/:id/check` (404 unknown, 409 `watch.busy` while one runs, 409 `watch.disabled` for a paused watch, 422 `watch.carrier_unavailable`, or 422 `watch.event_awaited` for an event source) || `watch.resource.read` | `GET /api/v1/resources?project=&kind=&cursor=&limit=`, `GET /api/v1/resources/:ovrn` (first-party service token only) |@@@ -434,10 +456,11 @@ Called elsewhere, as the service principal `watch` (client credentials from Open422), cross-site writes refused, and hostile names and values escaped on every pageNot yet demonstrated: a check against a real site from a deployed service (nothing is deployed--yet), the webhook carrier (step 5), the Run rung (step 6) and Node probes (step 7) — each answers--`watch.carrier_unavailable` and fetches nothing today — and the action kinds beyond notifications. Event sources are--covered end to end by `test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route--and the subscription sync against a stand-in Events).++yet), the Run rung (step 6) and Node probes (step 7) — each answers `watch.carrier_unavailable` and fetches nothing++today — and the action kinds beyond notifications. Event sources are covered end to end by++`test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route and the subscription++sync against a stand-in Events), and webhook sources by `test/webhook-sources.test.js` (the secret shown once, both++signing forms, every refusal, the budget, rotation, text and JSON bodies, and removal).## Deploynnew file mode 100644@@@ -0,0 +1,11 @@++-- phase: expand++-- OpenVibe.Watch: webhook sources (plan T18 step 5, server/webhook-sources.js). A webhook watch is woken by a POST to++-- https://openvibe.watch/hooks/<watch id>, signed with the watch's own secret (HMAC-SHA256 of the body, or the secret++-- as a bearer token). The secret is shown once, when the watch is created or the secret rotated, and is needed here in++-- full to check a signature. It goes with its watch (removal, account deletion).++++CREATE TABLE webhook_secrets (++ watch_id text COLLATE "C" PRIMARY KEY,++ secret text NOT NULL,++ created_at bigint NOT NULL++);@@@ -33,6 +33,7 @@ async function extraErase(t, subjects, counts) {counts.add(counts.erased, 'check_runs', await t.exec(`DELETE FROM check_runs WHERE watch_id IN ${mine}`, [subjects]));counts.add(counts.erased, 'action_runs', await t.exec(`DELETE FROM action_runs WHERE watch_id IN ${mine}`, [subjects]));counts.add(counts.erased, 'watch_endpoint_state', await t.exec(`DELETE FROM watch_endpoint_state WHERE watch_id IN ${mine}`, [subjects]));++ counts.add(counts.erased, 'webhook_secrets', await t.exec(`DELETE FROM webhook_secrets WHERE watch_id IN ${mine}`, [subjects]));counts.add(counts.erased, 'watches', await t.exec('DELETE FROM watches WHERE owner_sub = ANY($1::text[])', [subjects]));}@@@ -22,7 +22,7 @@ const { http } = require('openvibe-contracts');const { CAPS } = require('../auth');const { RegistryError, STATUSES } = require('../registry');--function watchesRouter({ registry, auth, limits }) {++function watchesRouter({ registry, auth, limits, webhookSources = null }) {const router = express.Router();const read = [auth.requireCap(CAPS.read), limits.reads('watch.read')];const manage = [auth.requireCap(CAPS.manage), limits.budget('watch.write')];@@@ -64,7 +64,10 @@ function watchesRouter({ registry, auth, limits }) {router.post('/api/v1/watches', manage, async (req, res) => {try {const row = await registry.create(req.body, req.principal);-- res.status(201).json({ watch: registry.view(row) });++ const view = registry.view(row);++ // A webhook watch's address and secret, shown this once (server/webhook-sources.js).++ const webhook = webhookSources && view.source && view.source.kind === 'webhook' ? await webhookSources.issue(row.id) : null;++ res.status(201).json(webhook ? { watch: view, webhook } : { watch: view });} catch (err) { fail(res, req, err); }});@@@ -87,10 +90,25 @@ function watchesRouter({ registry, auth, limits }) {router.post('/api/v1/watches/:id/pause', manage, setStatus('paused'));router.post('/api/v1/watches/:id/resume', manage, setStatus('active'));++ // A new secret for a webhook watch (the old one stops working at once); also how a watch that became a webhook++ // watch gets its first.++ router.post('/api/v1/watches/:id/webhook-secret', manage, async (req, res) => {++ try {++ const row = await registry.get(req.params.id, req.principal);++ if (!row) return notFound(res, req);++ const view = registry.view(row);++ if (!webhookSources || !view.source || view.source.kind !== 'webhook') {++ return res.status(409).type('application/problem+json').send(JSON.stringify({ type: 'about:blank', title: 'watch.not_webhook', status: 409, code: 'watch.not_webhook', detail: 'only a webhook watch has a secret' }));++ }++ res.json({ watch_id: row.id, webhook: await webhookSources.issue(row.id) });++ } catch (err) { fail(res, req, err); }++ });++router.delete('/api/v1/watches/:id', manage, async (req, res) => {try {const out = await registry.remove(req.params.id, req.principal);if (!out) return notFound(res, req);++ if (webhookSources) await webhookSources.forget(out.id);res.json(out);} catch (err) { fail(res, req, err); }});@@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";--function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null }) {++function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null }) {const app = express();app.disable('x-powered-by');app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');@@@ -87,6 +87,9 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,// Every other event type is an event source's (plan T18 step 5, server/event-sources.js).app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets || [], send, log, onEvent: eventSources ? eventSources.onEvent : null }));}++ // 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.++ 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.get('/api/health', (_req, res) => {@@@ -122,7 +125,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log });// OpenVibe.Services' resource index (watch.resource.read, service tokens only): read on loopback; the public vhost 404s it.app.use(resourceIndexRouter({ db, auth }));-- app.use(watchesRouter({ registry, auth, limits }));++ app.use(watchesRouter({ registry, auth, limits, webhookSources }));app.use(checksRouter({ registry, check, observations, auth, limits }));app.use('/api', (req, res) => http.sendProblem(res, 404, 'watch.not_found', { detail: `no route ${req.method} ${req.path}`, ctx: req.ov }));@@@ -12,7 +12,8 @@* http feed api implemented here (step 3)* event step 5: OpenVibe.Events delivers a matching event and the check reads it (carriers/event.js,* server/event-sources.js); a check with no event (Check now) answers watch.event_awaited-- * webhook later in step 5++ * webhook step 5: a signed POST to /hooks/<watch id> runs the check with the delivery (carriers/webhook.js,++ * server/webhook-sources.js); a check with no delivery (Check now) answers watch.webhook_awaited* run step 6* node step 7*@@@ -25,13 +26,13 @@ const { createHttpCarrier } = require('./http');const { createFeedCarrier } = require('./feed');const { createApiCarrier } = require('./api');const { createEventCarrier } = require('./event');++const { createWebhookCarrier } = require('./webhook');/** The documented order. A source kind fixes its carrier; this is the order a watch would prefer. */const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api'];/** The step of plan T18 that carries each kind this release does not. */const LATER = {-- webhook: { step: 5, note: 'a webhook source is delivered to /internal/hooks/, never polled' },run: { step: 6, note: 'browser polling goes through OpenVibe.Run only after every cheaper rung was unavailable' },node: { step: 7, note: 'a node probe observes a local/LAN thing through OpenVibe.Node' },};@@@ -53,6 +54,7 @@ function createCarriers({ fetcher, config, log = console }) {feed: createFeedCarrier({ fetcher, config, log }),api: createApiCarrier({ fetcher, config, log }),event: createEventCarrier(),++ webhook: createWebhookCarrier(),};/**@@@ -64,6 +66,9 @@ function createCarriers({ fetcher, config, log = console }) {if (kind === 'event' && !input) {return { unavailable: { code: 'watch.event_awaited', carrier: 'event', detail: `an event source is checked when a ${watch.source.pattern} event arrives from OpenVibe.Events. Nothing was fetched.` } };}++ if (kind === 'webhook' && !input) {++ return { unavailable: { code: 'watch.webhook_awaited', carrier: 'webhook', detail: 'a webhook source is checked when a signed delivery reaches its /hooks/ address. Nothing was fetched.' } };++ }const carrier = carriers[kind];if (!carrier) return { unavailable: unavailable(kind) };return { name: carrier.name, run: carrier.run };nnew file mode 100644@@@ -0,0 +1,29 @@++'use strict';++/**++ * The webhook carrier (plan T18 step 5): a webhook source is pushed to, never polled. A signed POST to++ * /hooks/<watch id> (server/webhook-sources.js) runs the check with the delivery as its input; the extraction reads++ * the body: parsed JSON as the document when the sender said JSON, the text otherwise.++ *++ * Without a delivery (Check now) there is nothing to read: carriers/index.js answers watch.webhook_awaited first.++ */++const crypto = require('crypto');++++function createWebhookCarrier() {++ return {++ name: 'webhook',++ async run({ input }) {++ const text = Buffer.isBuffer(input.body) ? input.body.toString('utf8') : String(input.body || '');++ const contentType = input.content_type || 'application/octet-stream';++ let document;++ if (/\bjson\b/i.test(contentType)) {++ try { document = JSON.parse(text); } catch { return { state: 'parse_error', error_code: 'unreadable', detail: 'the delivery said JSON but its body is not JSON' }; }++ }++ return {++ state: 'ok', body: text, content_type: contentType, ...(document !== undefined ? { document } : {}),++ bytes: Buffer.byteLength(text), raw_body_hash: crypto.createHash('sha256').update(text).digest('hex'),++ };++ },++ };++}++++module.exports = { createWebhookCarrier };@@@ -86,6 +86,10 @@ function load(env = process.env) {maxPerWatchPerHour: int(env.WATCH_EVENT_MAX_PER_HOUR, 60),syncMs: int(env.WATCH_EVENT_SYNC_MS, 10 * 60_000),},++ // Webhook sources (server/webhook-sources.js): at most WATCH_WEBHOOK_MAX_PER_HOUR deliveries wake one watch.++ webhookSources: {++ maxPerWatchPerHour: int(env.WATCH_WEBHOOK_MAX_PER_HOUR, 60),++ },// 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).@@@ -24,6 +24,7 @@ const condition = require('./condition'); // { evaluate, hasChanged, holds,const { createCheck } = require('./check');const { createScheduler } = require('./scheduler');const { createEventSources } = require('./event-sources');++const { createWebhookSources } = require('./webhook-sources');const { createServiceOutbox } = require('openvibe-sdk/events');const { validate } = require('openvibe-contracts');const { jwksClient, createServiceTokenClient } = require('openvibe-sdk/auth');@@@ -79,6 +80,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });registry.onEventSourceChange(() => eventSources.kick());++ const webhookSources = createWebhookSources({ db, registry, check, config, now, log });const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });const auth = createAuth({ config, log });// The public site: sessions in Watch's own database, the Network key to verify a sign-in token@@@ -92,7 +94,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst accountSend = givenSend || (config.oauth.clientSecret? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }): null);-- const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources });++ const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources });// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background// on an unref'd timer, keeping the last good keys through a Network outage.keys.start();@@@ -141,7 +143,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchif (!givenDb) await db.close();}-- return { config, db, registry, eventSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };++ return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };}/**@@@ -254,10 +254,13 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxconst owned = await st.countOwned.get(owner);if (Number(owned.n) >= maxWatchesPerOwner) throw new RegistryError(`an owner may hold at most ${maxWatchesPerOwner} watches`);const t = now();++ const id = `wch_${ids.ulid(t)}`;++ // A webhook watch's address is its own (server/webhook-sources.js): whatever endpoint a caller named is replaced.++ const sourceOf = (s) => (s && s.kind === 'webhook' ? { kind: 'webhook', endpoint: `/internal/hooks/${id}` } : s);const rec = {name: req.name || defaultName(req.source),status: req.status || 'active',-- source: req.source,++ source: sourceOf(req.source),cadence: req.cadence == null ? null : req.cadence,extraction: req.extraction,comparison: req.comparison == null ? null : req.comparison,@@@ -268,7 +271,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxlabels: req.labels || {},};const row = {-- id: `wch_${ids.ulid(t)}`,++ id,project_id: null,owner_sub: owner,name: rec.name,@@@ -304,6 +307,8 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxif (!row) return null;if (!changes || typeof changes !== 'object' || Array.isArray(changes)) throw new RegistryError('body must be a watch object');const current = fromRow(row);++ // A webhook source always answers at this watch's own address, never another's.++ if (changes.source && changes.source.kind === 'webhook') changes = { ...changes, source: { kind: 'webhook', endpoint: `/internal/hooks/${row.id}` } };const merged = { ...current, ...changes };delete merged.id; // an id never changesdelete merged.owner; // ownership never changes (a watch is the subject's)nnew file mode 100644@@@ -0,0 +1,118 @@++'use strict';++/**++ * Webhook sources (plan T18 step 5): a watch whose source is { kind: 'webhook' } is woken by a POST to++ * https://openvibe.watch/hooks/<watch id> (its stored endpoint is /internal/hooks/<watch id>, as watch.watch@1 names++ * it), never on a cadence. Any service that can call a webhook (GitHub, a CI, a store, another OpenVibe product) can++ * wake a watch, and the watch's extraction, condition and actions take it from there.++ *++ * The secret whk_<43 characters>, made when the watch is created and shown once in that answer, or again by++ * POST /api/v1/watches/:id/webhook-secret (which replaces it). It is needed in full to check a++ * signature, so it is stored as it is (webhook_secrets), and it goes with its watch.++ * Signing X-Hub-Signature-256: sha256=<hex HMAC-SHA256(secret, raw body)> (GitHub's form; many senders use++ * it), or Authorization: Bearer <secret> for a sender that cannot sign. Compared in constant time.++ * 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 SECRET_RE = /^whk_[A-Za-z0-9_-]{43}$/;++const MAX_BODY = 256 * 1024;++++const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;++const endpointFor = (watchId) => `/internal/hooks/${watchId}`;++++function safeEqual(a, b) {++ const x = Buffer.from(String(a));++ const y = Buffer.from(String(b));++ return x.length === y.length && crypto.timingSafeEqual(x, y);++}++++/** Does the delivery carry the watch's secret: a matching body signature, or the secret as a bearer token? */++function verified(secret, raw, headers) {++ if (!secret) return false;++ const sig = String(headers['x-hub-signature-256'] || '');++ if (sig.startsWith('sha256=')) {++ const want = `sha256=${crypto.createHmac('sha256', secret).update(raw).digest('hex')}`;++ if (safeEqual(sig, want)) return true;++ }++ const auth = String(headers.authorization || '');++ if (/^Bearer\s+/i.test(auth) && safeEqual(auth.replace(/^Bearer\s+/i, '').trim(), secret)) return true;++ return false;++}++++function createWebhookSources({ db, registry, check, config, now = () => Date.now(), log = console }) {++ const 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 st = {++ set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at) VALUES (?, ?, ?)++ ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at`),++ get: db.prepare('SELECT secret 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'),++ };++++ /** The public address and a fresh secret for a webhook watch (replacing any it had). Shown once. */++ async function issue(watchId) {++ const secret = newSecret();++ await st.set.run(watchId, secret, now());++ return { url: `${config.baseUrl}/hooks/${watchId}`, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, or Authorization: Bearer <secret>' };++ }++ async function forget(watchId) { await st.drop.run(watchId); }++++ function retryAfter() { return 3600 - Math.floor((now() / 1000) % 3600); }++ function underBudget(id) {++ const hour = Math.floor(now() / 3600_000);++ const c = counts.get(id);++ if (!c || c.hour !== hour) { counts.set(id, { hour, n: 1 }); return true; }++ if (c.n >= maxPerHour) return false;++ c.n++;++ return true;++ }++++ const problem = (res, status, code, detail, headers = {}) => {++ for (const [k, v] of Object.entries(headers)) res.setHeader(k, v);++ res.setHeader('Cache-Control', 'no-store');++ res.status(status).type('application/problem+json').send(JSON.stringify({ type: 'about:blank', title: code, status, code, detail }));++ };++ // Over the cap: a declared length is refused before reading; an undeclared one stops being kept and is drained, so the++ // sender still gets its 413 instead of a dropped connection.++ const readRaw = (req) => new Promise((resolve, reject) => {++ if (Number(req.headers['content-length']) > MAX_BODY) { req.resume(); reject(Object.assign(new Error('too large'), { status: 413 })); return; }++ const chunks = [];++ let size = 0;++ req.on('data', (c) => { size += c.length; if (size <= MAX_BODY) chunks.push(c); });++ req.on('end', () => (size > MAX_BODY ? reject(Object.assign(new Error('too large'), { status: 413 })) : resolve(Buffer.concat(chunks))));++ req.on('error', reject);++ });++++ /** POST /hooks/:id — the public receiver. */++ async function receive(req, res) {++ const id = String(req.params.id || '');++ 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'); }++ const row = /^wch_[0-9A-HJKMNP-TV-Z]{26}$/.test(id) ? await st.watch.get(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 webhook watch at this address');++ const secret = (await st.get.get(id) || {}).secret || null;++ if (!verified(secret, raw, req.headers)) return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed with this watch\'s secret');++ if (row.status !== 'active') return problem(res, 409, 'watch.disabled', `the watch is ${row.status}`);++ if (!underBudget(id)) return problem(res, 429, 'watch.rate_limited', `at most ${maxPerHour} deliveries an hour wake one watch`, { 'Retry-After': String(retryAfter()) });++ let out;++ try {++ out = await check.run(id, { trigger: 'webhook', input: { body: raw, content_type: req.headers['content-type'] || null } });++ } catch (err) {++ log.error(`[webhook-sources] ${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: id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });++ }++++ return { issue, forget, receive, endpointFor, verified, SECRET_RE };++}++++module.exports = { createWebhookSources, verified, endpointFor, newSecret };@@@ -68,12 +68,16 @@ t('the export carries the person\'s watches and no session', async () => {t('the deletion removes the person\'s watches with their checks and observations, and their sessions; once', async () => {const event = ev('evt_01JZ0000000000000000000D01', 'network.account.deleted', { deletion_id: 'del_01JZ0000000000000000000DE1', subject: me });++ // A webhook secret hanging under the person's watch goes with it (server/webhook-sources.js).++ const mineId = await svc.db.value('SELECT id FROM watches WHERE owner_sub = $1', [me]);++ await svc.db.prepare('INSERT INTO webhook_secrets (watch_id, secret, created_at) VALUES (?, ?, ?)').run(mineId, 'whk_test', Date.now());const r = await deliver(event);assert.deepStrictEqual([r.status, r.json.outcome], [200, 'erased'], JSON.stringify(r.json));assert.strictEqual(await count('SELECT count(*) FROM watches WHERE owner_sub = $1', [me]), 0);assert.strictEqual(await count('SELECT count(*) FROM web_sessions WHERE subject = $1', [me]), 0);assert.strictEqual(await count('SELECT count(*) FROM check_runs WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned check runs');assert.strictEqual(await count('SELECT count(*) FROM observations WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned observations');++ assert.strictEqual(await count('SELECT count(*) FROM webhook_secrets WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned webhook secret');assert.strictEqual(await count('SELECT count(*) FROM watches WHERE owner_sub = $1', [other]), 1, 'someone else\'s watch stays');const conf = stub.calls.filter((c) => c.url === '/internal/account-deletions/del_01JZ0000000000000000000DE1/confirmations');assert.strictEqual(conf.length, 1);@@@ -176,12 +176,12 @@ t('a manual check runs one check and answers the released check-run; observation// a watch whose carrier is a later step answers 422 and never fetchesconst later = await api('POST', '/api/v1/watches', {token: ME,-- body: mine({ source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json' } }),++ body: mine({ source: { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, cadence: { every_sec: 900 }, extraction: { kind: 'json' } }),});const refused = await api('POST', `/api/v1/watches/${later.body.watch.id}/check`, { token: ME });assert.strictEqual(refused.status, 422);assert.strictEqual(refused.body.code, 'watch.carrier_unavailable');-- assert.match(refused.body.detail, /step 5/);++ assert.match(refused.body.detail, /step 7/);// a paused watch cannot be checked by hand eitherawait api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });const paused = await api('POST', `/api/v1/watches/${watch.id}/check`, { token: ME });@@@ -186,9 +186,8 @@ t('a pattern that hits its deadline is a recorded check error, never a hang', asprice = '19.99';});--t('a webhook, node or run source is accepted but never fetched (steps 5-7); an event source waits for its event', async () => {++t('a node or run source is accepted but never fetched (steps 6-7); event and webhook sources wait to be pushed to', async () => {const cases = [-- ['webhook', { kind: 'webhook', endpoint: '/internal/hooks/abc' }, { op: 'exists' }],['node', { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, { op: 'exists' }],['run', { kind: 'run', runtime: 'browser', url: `${web.origin}/price`, steps: [{ goto: 'https://example.org' }] }, { op: 'exists' }],];@@@ -196,14 +195,14 @@ t('a webhook, node or run source is accepted but never fetched (steps 5-7); an efor (const [kind, source, condition] of cases) {// an event or a webhook is pushed to and never polls; node and run sources poll through// OpenVibe.Node and OpenVibe.Run, so they carry a cadence — and still nothing is fetched-- const cadence = (kind === 'event' || kind === 'webhook') ? null : { every_sec: 900 };++ const cadence = { every_sec: 900 };const w = await create(person, { source, cadence, extraction: { kind: 'text' }, condition });assert.strictEqual((await row(w.id)).status, 'active');const out = await check(w.id, 'manual');assert.strictEqual(out.run.state, 'skipped', `${kind}: ${out.run.detail}`);assert.strictEqual(out.run.error_code, 'watch.carrier_unavailable');assert.strictEqual(out.run.carrier, kind);-- assert.match(out.run.detail, /step [567]/);++ assert.match(out.run.detail, /step [67]/);assert.strictEqual(out.unavailable.code, 'watch.carrier_unavailable');assert.strictEqual((await observations(w.id)).length, 0);const [fail] = (await payloads('watch.check.failed')).slice(-1);nnew file mode 100644@@@ -0,0 +1,123 @@++'use strict';++/**++ * Webhook sources (plan T18 step 5): a webhook watch is woken by a POST to /hooks/<watch id> signed with its own++ * secret (HMAC-SHA256 of the body, GitHub's form, or the secret as a bearer token), never on a cadence. The secret is++ * shown once at creation (and on rotation); the address is always the watch's own; deliveries have an hourly budget.++ */++const assert = require('assert');++const crypto = require('crypto');++const { boot, request, userToken, suite, newUser } = require('./helpers');++const { verified } = require('../server/webhook-sources');++++const t = suite('webhook-sources');++let svc;++const me = newUser();++const ME = userToken(me);++const OTHER = userToken(newUser());++++const def = (over = {}) => ({++ name: 'shop orders',++ source: { kind: 'webhook', endpoint: '/internal/hooks/someone-elses' },++ cadence: null,++ extraction: { kind: 'json', value_path: 'order.total' },++ condition: { op: 'gt', value: 100 },++ action: [{ kind: 'notification' }],++ ...over,++});++const api = (method, p, opts = {}) => request(svc.base, method, p, opts);++const sign = (secret, body) => `sha256=${crypto.createHmac('sha256', secret).update(body).digest('hex')}`;++const deliver = (id, body, headers = {}) => fetch(`${svc.base}/hooks/${id}`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...headers } });++const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);++++let watch;++let secret;++++t('boot', async () => {++ svc = await boot({ env: { WATCH_WEBHOOK_MAX_PER_HOUR: '4' } });++});++++t('creating a webhook watch answers its address and secret once; the endpoint is the watch\'s own', async () => {++ const made = await api('POST', '/api/v1/watches', { token: ME, body: def() });++ assert.strictEqual(made.status, 201, JSON.stringify(made.body));++ watch = made.body.watch;++ assert.deepStrictEqual(watch.source, { kind: 'webhook', endpoint: `/internal/hooks/${watch.id}` }, 'never another watch\'s address');++ assert.strictEqual(made.body.webhook.url, `${svc.config.baseUrl}/hooks/${watch.id}`);++ assert.match(made.body.webhook.secret, /^whk_[A-Za-z0-9_-]{43}$/);++ secret = made.body.webhook.secret;++ const read = await api('GET', `/api/v1/watches/${watch.id}`, { token: ME });++ assert.strictEqual(read.body.webhook, undefined, 'a read never shows the secret');++ assert.ok(!JSON.stringify(read.body).includes(secret));++});++++t('a signed delivery runs the check with its JSON body as the document; the condition fires', async () => {++ const body = JSON.stringify({ order: { id: 'o1', total: 250 } });++ const res = await deliver(watch.id, body, { 'X-Hub-Signature-256': sign(secret, body) });++ assert.strictEqual(res.status, 200);++ const out = await res.json();++ assert.deepStrictEqual(out, { watch_id: watch.id, state: 'condition_met', triggered: true });++ const [r] = await runs(watch.id);++ assert.deepStrictEqual([r.trigger, r.carrier], ['webhook', 'webhook']);++ // The bearer form, for a sender that cannot sign.++ const small = JSON.stringify({ order: { id: 'o2', total: 20 } });++ const res2 = await deliver(watch.id, small, { Authorization: `Bearer ${secret}` });++ assert.deepStrictEqual(await res2.json(), { watch_id: watch.id, state: 'changed', triggered: false });++});++++t('refused: a wrong or missing signature (401), an unknown or non-webhook address (404), a paused watch (409)', async () => {++ const body = JSON.stringify({ order: { total: 999 } });++ assert.strictEqual((await deliver(watch.id, body)).status, 401);++ assert.strictEqual((await deliver(watch.id, body, { 'X-Hub-Signature-256': sign('whk_wrong', body) })).status, 401);++ assert.strictEqual((await deliver(watch.id, body, { 'X-Hub-Signature-256': sign(secret, `${body} `) })).status, 401, 'the signature covers the exact bytes');++ assert.strictEqual((await deliver(watch.id, body, { Authorization: 'Bearer whk_wrong' })).status, 401);++ assert.strictEqual((await deliver('wch_01JZ0000000000000000000000', body, { Authorization: `Bearer ${secret}` })).status, 404);++ assert.strictEqual((await deliver('not-an-id', body)).status, 404);++ const page = await svc.registry.create({ ...def(), source: { kind: 'http', url: 'https://example.org/' }, cadence: { every_sec: 900 }, extraction: { kind: 'text' } }, { sub: me });++ assert.strictEqual((await deliver(page.id, body, { Authorization: `Bearer ${secret}` })).status, 404, 'a page watch has no webhook');++ await api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });++ assert.strictEqual((await deliver(watch.id, body, { Authorization: `Bearer ${secret}` })).status, 409);++ await api('POST', `/api/v1/watches/${watch.id}/resume`, { token: ME });++ assert.strictEqual((await runs(watch.id)).length, 2, 'no refused delivery ran a check');++});++++t('the hourly budget answers 429 with Retry-After; a body over 256 KB is 413', async () => {++ const body = JSON.stringify({ order: { total: 1 } });++ const statuses = [];++ for (let i = 0; i < 3; i++) statuses.push((await deliver(watch.id, body, { Authorization: `Bearer ${secret}` })).status);++ assert.deepStrictEqual(statuses, [200, 200, 429], 'two deliveries were already used this hour (budget 4)');++ const limited = await deliver(watch.id, body, { Authorization: `Bearer ${secret}` });++ assert.ok(Number(limited.headers.get('retry-after')) > 0);++ const big = 'x'.repeat(300 * 1024);++ assert.strictEqual((await deliver(watch.id, big, { Authorization: `Bearer ${secret}` })).status, 413);++});++++t('rotating replaces the secret at once; only the owner may; Check now waits for a delivery', async () => {++ const rotated = await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME });++ assert.strictEqual(rotated.status, 200);++ assert.notStrictEqual(rotated.body.webhook.secret, secret);++ assert.strictEqual((await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: OTHER })).status, 404);++ const out = await svc.check.run(watch.id, { trigger: 'manual' });++ assert.strictEqual(out.run.error_code, 'watch.webhook_awaited');++ assert.strictEqual(verified(secret, Buffer.from('{}'), { authorization: `Bearer ${secret}` }), true, 'the helper itself');++ const stored = await svc.db.prepare('SELECT secret FROM webhook_secrets WHERE watch_id = ?').get(watch.id);++ assert.strictEqual(stored.secret, rotated.body.webhook.secret, 'the old secret is gone');++ secret = rotated.body.webhook.secret;++});++++t('a non-JSON delivery is read as text; JSON that is not JSON is a parse error, recorded', async () => {++ const text = await svc.registry.create(def({ name: 'plain', extraction: { kind: 'text' }, condition: { op: 'contains', value: 'paid' } }), { sub: me });++ const { secret: s } = await svc.webhookSources.issue(text.id);++ const ok = await fetch(`${svc.base}/hooks/${text.id}`, { method: 'POST', body: 'order paid', headers: { 'Content-Type': 'text/plain', Authorization: `Bearer ${s}` } });++ assert.strictEqual((await ok.json()).triggered, true);++ const bad = await fetch(`${svc.base}/hooks/${text.id}`, { method: 'POST', body: '{nope', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${s}` } });++ assert.strictEqual((await bad.json()).state, 'parse_error');++});++++t('deleting the watch forgets its secret', async () => {++ assert.strictEqual((await api('DELETE', `/api/v1/watches/${watch.id}`, { token: ME })).status, 200);++ assert.strictEqual(await svc.db.prepare('SELECT secret FROM webhook_secrets WHERE watch_id = ?').get(watch.id), undefined);++ assert.strictEqual((await deliver(watch.id, '{}', { Authorization: `Bearer ${secret}` })).status, 404);++});++++t('stop', async () => { await svc.stop(); });++++t.run();
@@ -89,7 +89,7 @@ every page and every form, never 403. A signed-in write must also come from this
@@ -89,7 +89,7 @@ every page and every form, never 403. A signed-in write must also come from this
The cadence is a select (15 minutes, 1 hour, 6 hours, 1 day); 15 minutes is the shortest a person may
The cadence is a select (15 minutes, 1 hour, 6 hours, 1 day); 15 minutes is the shortest a person may
choose here (the contract's floor is 30 seconds, for a service). A validation error re-renders the
choose here (the contract's floor is 30 seconds, for a service). A validation error re-renders the
form with the person's own input and the registry's own message next to the field it names. The
form with the person's own input and the registry's own message next to the field it names. The
-kinds whose carrier does not exist yet (webhook, run, node) and the `ai` extraction are **not
+kinds whose carrier does not exist yet (run, node) and the push kinds (event, webhook) and the `ai` extraction are **not
offered** — and a watch of such a kind created through the API still answers `watch.carrier_unavailable`
offered** — and a watch of such a kind created through the API still answers `watch.carrier_unavailable`
when checked from the site, fetching nothing. An event watch (made through the API) is checked when its event
when checked from the site, fetching nothing. An event watch (made through the API) is checked when its event
arrives, so Check now answers `watch.event_awaited`.
arrives, so Check now answers `watch.event_awaited`.
@@ -130,7 +130,7 @@ This release carries the pull rungs and event sources:
@@ -130,7 +130,7 @@ This release carries the pull rungs and event sources:
| `feed` | RSS 2.0 / RSS 1.0 (RDF) / Atom 1.0 → `{ items, latest, count }` | implemented |
| `feed` | RSS 2.0 / RSS 1.0 (RDF) / Atom 1.0 → `{ items, latest, count }` | implemented |
| `api` | JSON/XML with `items_path` + `fields`, or the body itself | implemented |
| `api` | JSON/XML with `items_path` + `fields`, or the body itself | implemented |
| `event` | delivered by OpenVibe.Events (see [Event sources](#event-sources)) | implemented |
| `event` | delivered by OpenVibe.Events (see [Event sources](#event-sources)) | implemented |
-| `webhook` | delivered to `/internal/hooks/…` | step 5 |
+| `webhook` | a signed POST to `/hooks/<watch id>` (see [Webhook sources](#webhook-sources)) | implemented |
| `run` | browser/code polling through OpenVibe.Run | step 6 |
| `run` | browser/code polling through OpenVibe.Run | step 6 |
| `node` | a probe through OpenVibe.Node | step 7 |
| `node` | a probe through OpenVibe.Node | step 7 |
@@ -163,6 +163,28 @@ nothing while nothing happens (`server/event-sources.js`, `server/carriers/event
@@ -163,6 +163,28 @@ nothing while nothing happens (`server/event-sources.js`, `server/carriers/event
subscriptions or another endpoint's. It is off without `EVENTS_URL`, the secret or the OAuth client secret.
subscriptions or another endpoint's. It is off without `EVENTS_URL`, the secret or the OAuth client secret.
- **Check now** on an event watch answers 422 `watch.event_awaited`: there is no event to read.
- **Check now** on an event watch answers 422 `watch.event_awaited`: there is no event to read.
+## Webhook sources
+
+A watch whose source is `{ "kind": "webhook" }` (and `cadence: null`) is woken by a POST to
+`https://openvibe.watch/hooks/<watch id>` from anything that can call a webhook: GitHub, a CI, a store, another OpenVibe
+product (`server/webhook-sources.js`, `server/carriers/webhook.js`).
+
+- **The address is the watch's own.** Its stored endpoint is `/internal/hooks/<watch id>` (the form `watch.watch@1`
+ names), whatever a caller sent, so a watch can never point at another watch's address.
+- **The secret is shown once.** Creating the watch answers `{ watch, webhook: { url, secret } }`. The secret
+ (`whk_…`) is never shown again; `POST /api/v1/watches/:id/webhook-secret` replaces it (the old one stops working at
+ once). It is kept in full because checking a signature needs it (`webhook_secrets`), and it goes with its watch:
+ deleting the watch or the account removes it.
+- **Signing.** `X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the exact body with the secret>` (GitHub's form), or
+ `Authorization: Bearer <secret>` for a sender that cannot sign. Compared in constant time.
+- **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`.
+- **Answers.** 200 `{ watch_id, state, triggered }` after the check. 401 for a missing or wrong signature. 404 for an
+ unknown address or a watch that is not a webhook watch. 409 for a paused watch. 413 over 256 KB. 429 with
+ `Retry-After` over `WATCH_WEBHOOK_MAX_PER_HOUR` (60) deliveries an hour. 503 with `Retry-After: 1` while the watch's
+ previous check still runs.
+- **Check now** on a webhook watch answers 422 `watch.webhook_awaited`: there is no delivery to read.
+
## A check
## A check
An in-process scheduler ([server/scheduler.js](server/scheduler.js)) starts checks for active
An in-process scheduler ([server/scheduler.js](server/scheduler.js)) starts checks for active
@@ -260,7 +282,7 @@ an app or module may not.
@@ -260,7 +282,7 @@ an app or module may not.
| Capability | Routes |
| Capability | Routes |
|---|---|
|---|---|
| `watch.watch.read` | `GET /api/v1/watches?status=&project_id=&before=&limit=`, `GET /api/v1/watches/:id` |
| `watch.watch.read` | `GET /api/v1/watches?status=&project_id=&before=&limit=`, `GET /api/v1/watches/:id` |
-| `watch.watch.manage` | `POST /api/v1/watches`, `PATCH /api/v1/watches/:id`, `POST /api/v1/watches/:id/pause`, `POST /api/v1/watches/:id/resume`, `DELETE /api/v1/watches/:id` |
+| `watch.watch.manage` | `POST /api/v1/watches`, `PATCH /api/v1/watches/:id`, `POST /api/v1/watches/:id/pause`, `POST /api/v1/watches/:id/resume`, `POST /api/v1/watches/:id/webhook-secret`, `DELETE /api/v1/watches/:id` |
| `watch.observation.read` | `GET /api/v1/watches/:id/observations?before=&limit=`, `GET /api/v1/watches/:id/checks?before=&limit=` |
| `watch.observation.read` | `GET /api/v1/watches/:id/observations?before=&limit=`, `GET /api/v1/watches/:id/checks?before=&limit=` |
| `watch.check.run` | `POST /api/v1/watches/:id/check` (404 unknown, 409 `watch.busy` while one runs, 409 `watch.disabled` for a paused watch, 422 `watch.carrier_unavailable`, or 422 `watch.event_awaited` for an event source) |
| `watch.check.run` | `POST /api/v1/watches/:id/check` (404 unknown, 409 `watch.busy` while one runs, 409 `watch.disabled` for a paused watch, 422 `watch.carrier_unavailable`, or 422 `watch.event_awaited` for an event source) |
| `watch.resource.read` | `GET /api/v1/resources?project=&kind=&cursor=&limit=`, `GET /api/v1/resources/:ovrn` (first-party service token only) |
| `watch.resource.read` | `GET /api/v1/resources?project=&kind=&cursor=&limit=`, `GET /api/v1/resources/:ovrn` (first-party service token only) |
@@ -434,10 +456,11 @@ Called elsewhere, as the service principal `watch` (client credentials from Open
@@ -434,10 +456,11 @@ Called elsewhere, as the service principal `watch` (client credentials from Open
422), cross-site writes refused, and hostile names and values escaped on every page
422), cross-site writes refused, and hostile names and values escaped on every page
Not yet demonstrated: a check against a real site from a deployed service (nothing is deployed
Not yet demonstrated: a check against a real site from a deployed service (nothing is deployed
-yet), the webhook carrier (step 5), the Run rung (step 6) and Node probes (step 7) — each answers
-`watch.carrier_unavailable` and fetches nothing today — and the action kinds beyond notifications. Event sources are
-covered end to end by `test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route
-and the subscription sync against a stand-in Events).
+yet), the Run rung (step 6) and Node probes (step 7) — each answers `watch.carrier_unavailable` and fetches nothing
+today — and the action kinds beyond notifications. Event sources are covered end to end by
+`test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route and the subscription
+sync against a stand-in Events), and webhook sources by `test/webhook-sources.test.js` (the secret shown once, both
+signing forms, every refusal, the budget, rotation, text and JSON bodies, and removal).
## Deploy
## Deploy
new file mode 100644
new file mode 100644
@@ -0,0 +1,11 @@
@@ -0,0 +1,11 @@
+-- phase: expand
+-- OpenVibe.Watch: webhook sources (plan T18 step 5, server/webhook-sources.js). A webhook watch is woken by a POST to
+-- https://openvibe.watch/hooks/<watch id>, signed with the watch's own secret (HMAC-SHA256 of the body, or the secret
+-- as a bearer token). The secret is shown once, when the watch is created or the secret rotated, and is needed here in
+-- full to check a signature. It goes with its watch (removal, account deletion).
+
+CREATE TABLE webhook_secrets (
+ watch_id text COLLATE "C" PRIMARY KEY,
+ secret text NOT NULL,
+ created_at bigint NOT NULL
+);
@@ -33,6 +33,7 @@ async function extraErase(t, subjects, counts) {
@@ -33,6 +33,7 @@ async function extraErase(t, subjects, counts) {
counts.add(counts.erased, 'check_runs', await t.exec(`DELETE FROM check_runs WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'check_runs', await t.exec(`DELETE FROM check_runs WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'action_runs', await t.exec(`DELETE FROM action_runs WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'action_runs', await t.exec(`DELETE FROM action_runs WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'watch_endpoint_state', await t.exec(`DELETE FROM watch_endpoint_state WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'watch_endpoint_state', await t.exec(`DELETE FROM watch_endpoint_state WHERE watch_id IN ${mine}`, [subjects]));
+ counts.add(counts.erased, 'webhook_secrets', await t.exec(`DELETE FROM webhook_secrets WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'watches', await t.exec('DELETE FROM watches WHERE owner_sub = ANY($1::text[])', [subjects]));
counts.add(counts.erased, 'watches', await t.exec('DELETE FROM watches WHERE owner_sub = ANY($1::text[])', [subjects]));
}
}
@@ -22,7 +22,7 @@ const { http } = require('openvibe-contracts');
@@ -22,7 +22,7 @@ const { http } = require('openvibe-contracts');
const { CAPS } = require('../auth');
const { CAPS } = require('../auth');
const { RegistryError, STATUSES } = require('../registry');
const { RegistryError, STATUSES } = require('../registry');
-function watchesRouter({ registry, auth, limits }) {
+function watchesRouter({ registry, auth, limits, webhookSources = null }) {
const router = express.Router();
const router = express.Router();
const read = [auth.requireCap(CAPS.read), limits.reads('watch.read')];
const read = [auth.requireCap(CAPS.read), limits.reads('watch.read')];
const manage = [auth.requireCap(CAPS.manage), limits.budget('watch.write')];
const manage = [auth.requireCap(CAPS.manage), limits.budget('watch.write')];
@@ -64,7 +64,10 @@ function watchesRouter({ registry, auth, limits }) {
@@ -64,7 +64,10 @@ function watchesRouter({ registry, auth, limits }) {
router.post('/api/v1/watches', manage, async (req, res) => {
router.post('/api/v1/watches', manage, async (req, res) => {
try {
try {
const row = await registry.create(req.body, req.principal);
const row = await registry.create(req.body, req.principal);
- res.status(201).json({ watch: registry.view(row) });
+ const view = registry.view(row);
+ // A webhook watch's address and secret, shown this once (server/webhook-sources.js).
+ const webhook = webhookSources && view.source && view.source.kind === 'webhook' ? await webhookSources.issue(row.id) : null;
+ res.status(201).json(webhook ? { watch: view, webhook } : { watch: view });
} catch (err) { fail(res, req, err); }
} catch (err) { fail(res, req, err); }
});
});
@@ -87,10 +90,25 @@ function watchesRouter({ registry, auth, limits }) {
@@ -87,10 +90,25 @@ function watchesRouter({ registry, auth, limits }) {
router.post('/api/v1/watches/:id/pause', manage, setStatus('paused'));
router.post('/api/v1/watches/:id/pause', manage, setStatus('paused'));
router.post('/api/v1/watches/:id/resume', manage, setStatus('active'));
router.post('/api/v1/watches/:id/resume', manage, setStatus('active'));
+ // A new secret for a webhook watch (the old one stops working at once); also how a watch that became a webhook
+ // watch gets its first.
+ router.post('/api/v1/watches/:id/webhook-secret', manage, async (req, res) => {
+ try {
+ const row = await registry.get(req.params.id, req.principal);
+ if (!row) return notFound(res, req);
+ const view = registry.view(row);
+ if (!webhookSources || !view.source || view.source.kind !== 'webhook') {
+ return res.status(409).type('application/problem+json').send(JSON.stringify({ type: 'about:blank', title: 'watch.not_webhook', status: 409, code: 'watch.not_webhook', detail: 'only a webhook watch has a secret' }));
+ }
+ res.json({ watch_id: row.id, webhook: await webhookSources.issue(row.id) });
+ } catch (err) { fail(res, req, err); }
+ });
+
router.delete('/api/v1/watches/:id', manage, async (req, res) => {
router.delete('/api/v1/watches/:id', manage, async (req, res) => {
try {
try {
const out = await registry.remove(req.params.id, req.principal);
const out = await registry.remove(req.params.id, req.principal);
if (!out) return notFound(res, req);
if (!out) return notFound(res, req);
+ if (webhookSources) await webhookSources.forget(out.id);
res.json(out);
res.json(out);
} catch (err) { fail(res, req, err); }
} catch (err) { fail(res, req, err); }
});
});
@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv
@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv
+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "
+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "
+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";
+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";
-function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null }) {
+function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null, eventSources = null, webhookSources = null }) {
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
@@ -87,6 +87,9 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
@@ -87,6 +87,9 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
// Every other event type is an event source's (plan T18 step 5, server/event-sources.js).
// Every other event type is an event source's (plan T18 step 5, server/event-sources.js).
app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets || [], send, log, onEvent: eventSources ? eventSources.onEvent : null }));
app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets || [], send, log, onEvent: eventSources ? eventSources.onEvent : null }));
}
}
+ // 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.
+ 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'] }));
app.get('/api/health', (_req, res) => {
app.get('/api/health', (_req, res) => {
@@ -122,7 +125,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
@@ -122,7 +125,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log });
const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log });
// OpenVibe.Services' resource index (watch.resource.read, service tokens only): read on loopback; the public vhost 404s it.
// OpenVibe.Services' resource index (watch.resource.read, service tokens only): read on loopback; the public vhost 404s it.
app.use(resourceIndexRouter({ db, auth }));
app.use(resourceIndexRouter({ db, auth }));
- app.use(watchesRouter({ registry, auth, limits }));
+ app.use(watchesRouter({ registry, auth, limits, webhookSources }));
app.use(checksRouter({ registry, check, observations, auth, limits }));
app.use(checksRouter({ registry, check, observations, auth, limits }));
app.use('/api', (req, res) => http.sendProblem(res, 404, 'watch.not_found', { detail: `no route ${req.method} ${req.path}`, ctx: req.ov }));
app.use('/api', (req, res) => http.sendProblem(res, 404, 'watch.not_found', { detail: `no route ${req.method} ${req.path}`, ctx: req.ov }));
@@ -12,7 +12,8 @@
@@ -12,7 +12,8 @@
* http feed api implemented here (step 3)
* http feed api implemented here (step 3)
* event step 5: OpenVibe.Events delivers a matching event and the check reads it (carriers/event.js,
* event step 5: OpenVibe.Events delivers a matching event and the check reads it (carriers/event.js,
* server/event-sources.js); a check with no event (Check now) answers watch.event_awaited
* server/event-sources.js); a check with no event (Check now) answers watch.event_awaited
- * webhook later in step 5
+ * webhook step 5: a signed POST to /hooks/<watch id> runs the check with the delivery (carriers/webhook.js,
+ * server/webhook-sources.js); a check with no delivery (Check now) answers watch.webhook_awaited
* run step 6
* run step 6
* node step 7
* node step 7
*
*
@@ -25,13 +26,13 @@ const { createHttpCarrier } = require('./http');
@@ -25,13 +26,13 @@ const { createHttpCarrier } = require('./http');
const { createFeedCarrier } = require('./feed');
const { createFeedCarrier } = require('./feed');
const { createApiCarrier } = require('./api');
const { createApiCarrier } = require('./api');
const { createEventCarrier } = require('./event');
const { createEventCarrier } = require('./event');
+const { createWebhookCarrier } = require('./webhook');
/** The documented order. A source kind fixes its carrier; this is the order a watch would prefer. */
/** The documented order. A source kind fixes its carrier; this is the order a watch would prefer. */
const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api'];
const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api'];
/** The step of plan T18 that carries each kind this release does not. */
/** The step of plan T18 that carries each kind this release does not. */
const LATER = {
const LATER = {
- webhook: { step: 5, note: 'a webhook source is delivered to /internal/hooks/, never polled' },
run: { step: 6, note: 'browser polling goes through OpenVibe.Run only after every cheaper rung was unavailable' },
run: { step: 6, note: 'browser polling goes through OpenVibe.Run only after every cheaper rung was unavailable' },
node: { step: 7, note: 'a node probe observes a local/LAN thing through OpenVibe.Node' },
node: { step: 7, note: 'a node probe observes a local/LAN thing through OpenVibe.Node' },
};
};
@@ -53,6 +54,7 @@ function createCarriers({ fetcher, config, log = console }) {
@@ -53,6 +54,7 @@ function createCarriers({ fetcher, config, log = console }) {
feed: createFeedCarrier({ fetcher, config, log }),
feed: createFeedCarrier({ fetcher, config, log }),
api: createApiCarrier({ fetcher, config, log }),
api: createApiCarrier({ fetcher, config, log }),
event: createEventCarrier(),
event: createEventCarrier(),
+ webhook: createWebhookCarrier(),
};
};
/**
/**
@@ -64,6 +66,9 @@ function createCarriers({ fetcher, config, log = console }) {
@@ -64,6 +66,9 @@ function createCarriers({ fetcher, config, log = console }) {
if (kind === 'event' && !input) {
if (kind === 'event' && !input) {
return { unavailable: { code: 'watch.event_awaited', carrier: 'event', detail: `an event source is checked when a ${watch.source.pattern} event arrives from OpenVibe.Events. Nothing was fetched.` } };
return { unavailable: { code: 'watch.event_awaited', carrier: 'event', detail: `an event source is checked when a ${watch.source.pattern} event arrives from OpenVibe.Events. Nothing was fetched.` } };
}
}
+ if (kind === 'webhook' && !input) {
+ return { unavailable: { code: 'watch.webhook_awaited', carrier: 'webhook', detail: 'a webhook source is checked when a signed delivery reaches its /hooks/ address. Nothing was fetched.' } };
+ }
const carrier = carriers[kind];
const carrier = carriers[kind];
if (!carrier) return { unavailable: unavailable(kind) };
if (!carrier) return { unavailable: unavailable(kind) };
return { name: carrier.name, run: carrier.run };
return { name: carrier.name, run: carrier.run };
new file mode 100644
new file mode 100644
@@ -0,0 +1,29 @@
@@ -0,0 +1,29 @@
+'use strict';
+/**
+ * The webhook carrier (plan T18 step 5): a webhook source is pushed to, never polled. A signed POST to
+ * /hooks/<watch id> (server/webhook-sources.js) runs the check with the delivery as its input; the extraction reads
+ * the body: parsed JSON as the document when the sender said JSON, the text otherwise.
+ *
+ * Without a delivery (Check now) there is nothing to read: carriers/index.js answers watch.webhook_awaited first.
+ */
+const crypto = require('crypto');
+
+function createWebhookCarrier() {
+ return {
+ name: 'webhook',
+ async run({ input }) {
+ const text = Buffer.isBuffer(input.body) ? input.body.toString('utf8') : String(input.body || '');
+ const contentType = input.content_type || 'application/octet-stream';
+ let document;
+ if (/\bjson\b/i.test(contentType)) {
+ try { document = JSON.parse(text); } catch { return { state: 'parse_error', error_code: 'unreadable', detail: 'the delivery said JSON but its body is not JSON' }; }
+ }
+ return {
+ state: 'ok', body: text, content_type: contentType, ...(document !== undefined ? { document } : {}),
+ bytes: Buffer.byteLength(text), raw_body_hash: crypto.createHash('sha256').update(text).digest('hex'),
+ };
+ },
+ };
+}
+
+module.exports = { createWebhookCarrier };
@@ -86,6 +86,10 @@ function load(env = process.env) {
@@ -86,6 +86,10 @@ function load(env = process.env) {
maxPerWatchPerHour: int(env.WATCH_EVENT_MAX_PER_HOUR, 60),
maxPerWatchPerHour: int(env.WATCH_EVENT_MAX_PER_HOUR, 60),
syncMs: int(env.WATCH_EVENT_SYNC_MS, 10 * 60_000),
syncMs: int(env.WATCH_EVENT_SYNC_MS, 10 * 60_000),
},
},
+ // Webhook sources (server/webhook-sources.js): at most WATCH_WEBHOOK_MAX_PER_HOUR deliveries wake one watch.
+ webhookSources: {
+ maxPerWatchPerHour: int(env.WATCH_WEBHOOK_MAX_PER_HOUR, 60),
+ },
// 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).
@@ -24,6 +24,7 @@ const condition = require('./condition'); // { evaluate, hasChanged, holds,
@@ -24,6 +24,7 @@ const condition = require('./condition'); // { evaluate, hasChanged, holds,
const { createCheck } = require('./check');
const { createCheck } = require('./check');
const { createScheduler } = require('./scheduler');
const { createScheduler } = require('./scheduler');
const { createEventSources } = require('./event-sources');
const { createEventSources } = require('./event-sources');
+const { createWebhookSources } = require('./webhook-sources');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { validate } = require('openvibe-contracts');
const { validate } = require('openvibe-contracts');
const { jwksClient, createServiceTokenClient } = require('openvibe-sdk/auth');
const { jwksClient, createServiceTokenClient } = require('openvibe-sdk/auth');
@@ -79,6 +80,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -79,6 +80,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });
const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });
registry.onEventSourceChange(() => eventSources.kick());
registry.onEventSourceChange(() => eventSources.kick());
+ const webhookSources = createWebhookSources({ db, registry, check, config, now, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const auth = createAuth({ config, log });
const auth = createAuth({ config, log });
// The public site: sessions in Watch's own database, the Network key to verify a sign-in token
// The public site: sessions in Watch's own database, the Network key to verify a sign-in token
@@ -92,7 +94,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -92,7 +94,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const accountSend = givenSend || (config.oauth.clientSecret
const accountSend = givenSend || (config.oauth.clientSecret
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
: null);
: null);
- const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources });
+ const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend, eventSources, webhookSources });
// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background
// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background
// on an unref'd timer, keeping the last good keys through a Network outage.
// on an unref'd timer, keeping the last good keys through a Network outage.
keys.start();
keys.start();
@@ -141,7 +143,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -141,7 +143,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (!givenDb) await db.close();
if (!givenDb) await db.close();
}
}
- return { config, db, registry, eventSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
+ return { config, db, registry, eventSources, webhookSources, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
}
}
/**
/**
@@ -254,10 +254,13 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -254,10 +254,13 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
const owned = await st.countOwned.get(owner);
const owned = await st.countOwned.get(owner);
if (Number(owned.n) >= maxWatchesPerOwner) throw new RegistryError(`an owner may hold at most ${maxWatchesPerOwner} watches`);
if (Number(owned.n) >= maxWatchesPerOwner) throw new RegistryError(`an owner may hold at most ${maxWatchesPerOwner} watches`);
const t = now();
const t = now();
+ const id = `wch_${ids.ulid(t)}`;
+ // A webhook watch's address is its own (server/webhook-sources.js): whatever endpoint a caller named is replaced.
+ const sourceOf = (s) => (s && s.kind === 'webhook' ? { kind: 'webhook', endpoint: `/internal/hooks/${id}` } : s);
const rec = {
const rec = {
name: req.name || defaultName(req.source),
name: req.name || defaultName(req.source),
status: req.status || 'active',
status: req.status || 'active',
- source: req.source,
+ source: sourceOf(req.source),
cadence: req.cadence == null ? null : req.cadence,
cadence: req.cadence == null ? null : req.cadence,
extraction: req.extraction,
extraction: req.extraction,
comparison: req.comparison == null ? null : req.comparison,
comparison: req.comparison == null ? null : req.comparison,
@@ -268,7 +271,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -268,7 +271,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
labels: req.labels || {},
labels: req.labels || {},
};
};
const row = {
const row = {
- id: `wch_${ids.ulid(t)}`,
+ id,
project_id: null,
project_id: null,
owner_sub: owner,
owner_sub: owner,
name: rec.name,
name: rec.name,
@@ -304,6 +307,8 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -304,6 +307,8 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
if (!row) return null;
if (!row) return null;
if (!changes || typeof changes !== 'object' || Array.isArray(changes)) throw new RegistryError('body must be a watch object');
if (!changes || typeof changes !== 'object' || Array.isArray(changes)) throw new RegistryError('body must be a watch object');
const current = fromRow(row);
const current = fromRow(row);
+ // A webhook source always answers at this watch's own address, never another's.
+ if (changes.source && changes.source.kind === 'webhook') changes = { ...changes, source: { kind: 'webhook', endpoint: `/internal/hooks/${row.id}` } };
const merged = { ...current, ...changes };
const merged = { ...current, ...changes };
delete merged.id; // an id never changes
delete merged.id; // an id never changes
delete merged.owner; // ownership never changes (a watch is the subject's)
delete merged.owner; // ownership never changes (a watch is the subject's)
new file mode 100644
new file mode 100644
@@ -0,0 +1,118 @@
@@ -0,0 +1,118 @@
+'use strict';
+/**
+ * Webhook sources (plan T18 step 5): a watch whose source is { kind: 'webhook' } is woken by a POST to
+ * https://openvibe.watch/hooks/<watch id> (its stored endpoint is /internal/hooks/<watch id>, as watch.watch@1 names
+ * it), never on a cadence. Any service that can call a webhook (GitHub, a CI, a store, another OpenVibe product) can
+ * wake a watch, and the watch's extraction, condition and actions take it from there.
+ *
+ * The secret whk_<43 characters>, made when the watch is created and shown once in that answer, or again by
+ * POST /api/v1/watches/:id/webhook-secret (which replaces it). It is needed in full to check a
+ * signature, so it is stored as it is (webhook_secrets), and it goes with its watch.
+ * Signing X-Hub-Signature-256: sha256=<hex HMAC-SHA256(secret, raw body)> (GitHub's form; many senders use
+ * it), or Authorization: Bearer <secret> for a sender that cannot sign. Compared in constant time.
+ * 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 SECRET_RE = /^whk_[A-Za-z0-9_-]{43}$/;
+const MAX_BODY = 256 * 1024;
+
+const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;
+const endpointFor = (watchId) => `/internal/hooks/${watchId}`;
+
+function safeEqual(a, b) {
+ const x = Buffer.from(String(a));
+ const y = Buffer.from(String(b));
+ return x.length === y.length && crypto.timingSafeEqual(x, y);
+}
+
+/** Does the delivery carry the watch's secret: a matching body signature, or the secret as a bearer token? */
+function verified(secret, raw, headers) {
+ if (!secret) return false;
+ const sig = String(headers['x-hub-signature-256'] || '');
+ if (sig.startsWith('sha256=')) {
+ const want = `sha256=${crypto.createHmac('sha256', secret).update(raw).digest('hex')}`;
+ if (safeEqual(sig, want)) return true;
+ }
+ const auth = String(headers.authorization || '');
+ if (/^Bearer\s+/i.test(auth) && safeEqual(auth.replace(/^Bearer\s+/i, '').trim(), secret)) return true;
+ return false;
+}
+
+function createWebhookSources({ db, registry, check, config, now = () => Date.now(), log = console }) {
+ const 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 st = {
+ set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at) VALUES (?, ?, ?)
+ ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at`),
+ get: db.prepare('SELECT secret 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'),
+ };
+
+ /** The public address and a fresh secret for a webhook watch (replacing any it had). Shown once. */
+ async function issue(watchId) {
+ const secret = newSecret();
+ await st.set.run(watchId, secret, now());
+ return { url: `${config.baseUrl}/hooks/${watchId}`, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, or Authorization: Bearer <secret>' };
+ }
+ async function forget(watchId) { await st.drop.run(watchId); }
+
+ function retryAfter() { return 3600 - Math.floor((now() / 1000) % 3600); }
+ function underBudget(id) {
+ const hour = Math.floor(now() / 3600_000);
+ const c = counts.get(id);
+ if (!c || c.hour !== hour) { counts.set(id, { hour, n: 1 }); return true; }
+ if (c.n >= maxPerHour) return false;
+ c.n++;
+ return true;
+ }
+
+ const problem = (res, status, code, detail, headers = {}) => {
+ for (const [k, v] of Object.entries(headers)) res.setHeader(k, v);
+ res.setHeader('Cache-Control', 'no-store');
+ res.status(status).type('application/problem+json').send(JSON.stringify({ type: 'about:blank', title: code, status, code, detail }));
+ };
+ // Over the cap: a declared length is refused before reading; an undeclared one stops being kept and is drained, so the
+ // sender still gets its 413 instead of a dropped connection.
+ const readRaw = (req) => new Promise((resolve, reject) => {
+ if (Number(req.headers['content-length']) > MAX_BODY) { req.resume(); reject(Object.assign(new Error('too large'), { status: 413 })); return; }
+ const chunks = [];
+ let size = 0;
+ req.on('data', (c) => { size += c.length; if (size <= MAX_BODY) chunks.push(c); });
+ req.on('end', () => (size > MAX_BODY ? reject(Object.assign(new Error('too large'), { status: 413 })) : resolve(Buffer.concat(chunks))));
+ req.on('error', reject);
+ });
+
+ /** POST /hooks/:id — the public receiver. */
+ async function receive(req, res) {
+ const id = String(req.params.id || '');
+ 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'); }
+ const row = /^wch_[0-9A-HJKMNP-TV-Z]{26}$/.test(id) ? await st.watch.get(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 webhook watch at this address');
+ const secret = (await st.get.get(id) || {}).secret || null;
+ if (!verified(secret, raw, req.headers)) return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed with this watch\'s secret');
+ if (row.status !== 'active') return problem(res, 409, 'watch.disabled', `the watch is ${row.status}`);
+ if (!underBudget(id)) return problem(res, 429, 'watch.rate_limited', `at most ${maxPerHour} deliveries an hour wake one watch`, { 'Retry-After': String(retryAfter()) });
+ let out;
+ try {
+ out = await check.run(id, { trigger: 'webhook', input: { body: raw, content_type: req.headers['content-type'] || null } });
+ } catch (err) {
+ log.error(`[webhook-sources] ${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: id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });
+ }
+
+ return { issue, forget, receive, endpointFor, verified, SECRET_RE };
+}
+
+module.exports = { createWebhookSources, verified, endpointFor, newSecret };
@@ -68,12 +68,16 @@ t('the export carries the person\'s watches and no session', async () => {
@@ -68,12 +68,16 @@ t('the export carries the person\'s watches and no session', async () => {
t('the deletion removes the person\'s watches with their checks and observations, and their sessions; once', async () => {
t('the deletion removes the person\'s watches with their checks and observations, and their sessions; once', async () => {
const event = ev('evt_01JZ0000000000000000000D01', 'network.account.deleted', { deletion_id: 'del_01JZ0000000000000000000DE1', subject: me });
const event = ev('evt_01JZ0000000000000000000D01', 'network.account.deleted', { deletion_id: 'del_01JZ0000000000000000000DE1', subject: me });
+ // A webhook secret hanging under the person's watch goes with it (server/webhook-sources.js).
+ const mineId = await svc.db.value('SELECT id FROM watches WHERE owner_sub = $1', [me]);
+ await svc.db.prepare('INSERT INTO webhook_secrets (watch_id, secret, created_at) VALUES (?, ?, ?)').run(mineId, 'whk_test', Date.now());
const r = await deliver(event);
const r = await deliver(event);
assert.deepStrictEqual([r.status, r.json.outcome], [200, 'erased'], JSON.stringify(r.json));
assert.deepStrictEqual([r.status, r.json.outcome], [200, 'erased'], JSON.stringify(r.json));
assert.strictEqual(await count('SELECT count(*) FROM watches WHERE owner_sub = $1', [me]), 0);
assert.strictEqual(await count('SELECT count(*) FROM watches WHERE owner_sub = $1', [me]), 0);
assert.strictEqual(await count('SELECT count(*) FROM web_sessions WHERE subject = $1', [me]), 0);
assert.strictEqual(await count('SELECT count(*) FROM web_sessions WHERE subject = $1', [me]), 0);
assert.strictEqual(await count('SELECT count(*) FROM check_runs WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned check runs');
assert.strictEqual(await count('SELECT count(*) FROM check_runs WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned check runs');
assert.strictEqual(await count('SELECT count(*) FROM observations WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned observations');
assert.strictEqual(await count('SELECT count(*) FROM observations WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned observations');
+ assert.strictEqual(await count('SELECT count(*) FROM webhook_secrets WHERE watch_id NOT IN (SELECT id FROM watches)'), 0, 'no orphaned webhook secret');
assert.strictEqual(await count('SELECT count(*) FROM watches WHERE owner_sub = $1', [other]), 1, 'someone else\'s watch stays');
assert.strictEqual(await count('SELECT count(*) FROM watches WHERE owner_sub = $1', [other]), 1, 'someone else\'s watch stays');
const conf = stub.calls.filter((c) => c.url === '/internal/account-deletions/del_01JZ0000000000000000000DE1/confirmations');
const conf = stub.calls.filter((c) => c.url === '/internal/account-deletions/del_01JZ0000000000000000000DE1/confirmations');
assert.strictEqual(conf.length, 1);
assert.strictEqual(conf.length, 1);
@@ -176,12 +176,12 @@ t('a manual check runs one check and answers the released check-run; observation
@@ -176,12 +176,12 @@ t('a manual check runs one check and answers the released check-run; observation
// a watch whose carrier is a later step answers 422 and never fetches
// a watch whose carrier is a later step answers 422 and never fetches
const later = await api('POST', '/api/v1/watches', {
const later = await api('POST', '/api/v1/watches', {
token: ME,
token: ME,
- body: mine({ source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json' } }),
+ body: mine({ source: { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, cadence: { every_sec: 900 }, extraction: { kind: 'json' } }),
});
});
const refused = await api('POST', `/api/v1/watches/${later.body.watch.id}/check`, { token: ME });
const refused = await api('POST', `/api/v1/watches/${later.body.watch.id}/check`, { token: ME });
assert.strictEqual(refused.status, 422);
assert.strictEqual(refused.status, 422);
assert.strictEqual(refused.body.code, 'watch.carrier_unavailable');
assert.strictEqual(refused.body.code, 'watch.carrier_unavailable');
- assert.match(refused.body.detail, /step 5/);
+ assert.match(refused.body.detail, /step 7/);
// a paused watch cannot be checked by hand either
// a paused watch cannot be checked by hand either
await api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });
await api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });
const paused = await api('POST', `/api/v1/watches/${watch.id}/check`, { token: ME });
const paused = await api('POST', `/api/v1/watches/${watch.id}/check`, { token: ME });
@@ -186,9 +186,8 @@ t('a pattern that hits its deadline is a recorded check error, never a hang', as
@@ -186,9 +186,8 @@ t('a pattern that hits its deadline is a recorded check error, never a hang', as
price = '19.99';
price = '19.99';
});
});
-t('a webhook, node or run source is accepted but never fetched (steps 5-7); an event source waits for its event', async () => {
+t('a node or run source is accepted but never fetched (steps 6-7); event and webhook sources wait to be pushed to', async () => {
const cases = [
const cases = [
- ['webhook', { kind: 'webhook', endpoint: '/internal/hooks/abc' }, { op: 'exists' }],
['node', { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, { op: 'exists' }],
['node', { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, { op: 'exists' }],
['run', { kind: 'run', runtime: 'browser', url: `${web.origin}/price`, steps: [{ goto: 'https://example.org' }] }, { op: 'exists' }],
['run', { kind: 'run', runtime: 'browser', url: `${web.origin}/price`, steps: [{ goto: 'https://example.org' }] }, { op: 'exists' }],
];
];
@@ -196,14 +195,14 @@ t('a webhook, node or run source is accepted but never fetched (steps 5-7); an e
@@ -196,14 +195,14 @@ t('a webhook, node or run source is accepted but never fetched (steps 5-7); an e
for (const [kind, source, condition] of cases) {
for (const [kind, source, condition] of cases) {
// an event or a webhook is pushed to and never polls; node and run sources poll through
// an event or a webhook is pushed to and never polls; node and run sources poll through
// OpenVibe.Node and OpenVibe.Run, so they carry a cadence — and still nothing is fetched
// OpenVibe.Node and OpenVibe.Run, so they carry a cadence — and still nothing is fetched
- const cadence = (kind === 'event' || kind === 'webhook') ? null : { every_sec: 900 };
+ const cadence = { every_sec: 900 };
const w = await create(person, { source, cadence, extraction: { kind: 'text' }, condition });
const w = await create(person, { source, cadence, extraction: { kind: 'text' }, condition });
assert.strictEqual((await row(w.id)).status, 'active');
assert.strictEqual((await row(w.id)).status, 'active');
const out = await check(w.id, 'manual');
const out = await check(w.id, 'manual');
assert.strictEqual(out.run.state, 'skipped', `${kind}: ${out.run.detail}`);
assert.strictEqual(out.run.state, 'skipped', `${kind}: ${out.run.detail}`);
assert.strictEqual(out.run.error_code, 'watch.carrier_unavailable');
assert.strictEqual(out.run.error_code, 'watch.carrier_unavailable');
assert.strictEqual(out.run.carrier, kind);
assert.strictEqual(out.run.carrier, kind);
- assert.match(out.run.detail, /step [567]/);
+ assert.match(out.run.detail, /step [67]/);
assert.strictEqual(out.unavailable.code, 'watch.carrier_unavailable');
assert.strictEqual(out.unavailable.code, 'watch.carrier_unavailable');
assert.strictEqual((await observations(w.id)).length, 0);
assert.strictEqual((await observations(w.id)).length, 0);
const [fail] = (await payloads('watch.check.failed')).slice(-1);
const [fail] = (await payloads('watch.check.failed')).slice(-1);
new file mode 100644
new file mode 100644
@@ -0,0 +1,123 @@
@@ -0,0 +1,123 @@
+'use strict';
+/**
+ * Webhook sources (plan T18 step 5): a webhook watch is woken by a POST to /hooks/<watch id> signed with its own
+ * secret (HMAC-SHA256 of the body, GitHub's form, or the secret as a bearer token), never on a cadence. The secret is
+ * shown once at creation (and on rotation); the address is always the watch's own; deliveries have an hourly budget.
+ */
+const assert = require('assert');
+const crypto = require('crypto');
+const { boot, request, userToken, suite, newUser } = require('./helpers');
+const { verified } = require('../server/webhook-sources');
+
+const t = suite('webhook-sources');
+let svc;
+const me = newUser();
+const ME = userToken(me);
+const OTHER = userToken(newUser());
+
+const def = (over = {}) => ({
+ name: 'shop orders',
+ source: { kind: 'webhook', endpoint: '/internal/hooks/someone-elses' },
+ cadence: null,
+ extraction: { kind: 'json', value_path: 'order.total' },
+ condition: { op: 'gt', value: 100 },
+ action: [{ kind: 'notification' }],
+ ...over,
+});
+const api = (method, p, opts = {}) => request(svc.base, method, p, opts);
+const sign = (secret, body) => `sha256=${crypto.createHmac('sha256', secret).update(body).digest('hex')}`;
+const deliver = (id, body, headers = {}) => fetch(`${svc.base}/hooks/${id}`, { method: 'POST', body, headers: { 'Content-Type': 'application/json', ...headers } });
+const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);
+
+let watch;
+let secret;
+
+t('boot', async () => {
+ svc = await boot({ env: { WATCH_WEBHOOK_MAX_PER_HOUR: '4' } });
+});
+
+t('creating a webhook watch answers its address and secret once; the endpoint is the watch\'s own', async () => {
+ const made = await api('POST', '/api/v1/watches', { token: ME, body: def() });
+ assert.strictEqual(made.status, 201, JSON.stringify(made.body));
+ watch = made.body.watch;
+ assert.deepStrictEqual(watch.source, { kind: 'webhook', endpoint: `/internal/hooks/${watch.id}` }, 'never another watch\'s address');
+ assert.strictEqual(made.body.webhook.url, `${svc.config.baseUrl}/hooks/${watch.id}`);
+ assert.match(made.body.webhook.secret, /^whk_[A-Za-z0-9_-]{43}$/);
+ secret = made.body.webhook.secret;
+ const read = await api('GET', `/api/v1/watches/${watch.id}`, { token: ME });
+ assert.strictEqual(read.body.webhook, undefined, 'a read never shows the secret');
+ assert.ok(!JSON.stringify(read.body).includes(secret));
+});
+
+t('a signed delivery runs the check with its JSON body as the document; the condition fires', async () => {
+ const body = JSON.stringify({ order: { id: 'o1', total: 250 } });
+ const res = await deliver(watch.id, body, { 'X-Hub-Signature-256': sign(secret, body) });
+ assert.strictEqual(res.status, 200);
+ const out = await res.json();
+ assert.deepStrictEqual(out, { watch_id: watch.id, state: 'condition_met', triggered: true });
+ const [r] = await runs(watch.id);
+ assert.deepStrictEqual([r.trigger, r.carrier], ['webhook', 'webhook']);
+ // The bearer form, for a sender that cannot sign.
+ const small = JSON.stringify({ order: { id: 'o2', total: 20 } });
+ const res2 = await deliver(watch.id, small, { Authorization: `Bearer ${secret}` });
+ assert.deepStrictEqual(await res2.json(), { watch_id: watch.id, state: 'changed', triggered: false });
+});
+
+t('refused: a wrong or missing signature (401), an unknown or non-webhook address (404), a paused watch (409)', async () => {
+ const body = JSON.stringify({ order: { total: 999 } });
+ assert.strictEqual((await deliver(watch.id, body)).status, 401);
+ assert.strictEqual((await deliver(watch.id, body, { 'X-Hub-Signature-256': sign('whk_wrong', body) })).status, 401);
+ assert.strictEqual((await deliver(watch.id, body, { 'X-Hub-Signature-256': sign(secret, `${body} `) })).status, 401, 'the signature covers the exact bytes');
+ assert.strictEqual((await deliver(watch.id, body, { Authorization: 'Bearer whk_wrong' })).status, 401);
+ assert.strictEqual((await deliver('wch_01JZ0000000000000000000000', body, { Authorization: `Bearer ${secret}` })).status, 404);
+ assert.strictEqual((await deliver('not-an-id', body)).status, 404);
+ const page = await svc.registry.create({ ...def(), source: { kind: 'http', url: 'https://example.org/' }, cadence: { every_sec: 900 }, extraction: { kind: 'text' } }, { sub: me });
+ assert.strictEqual((await deliver(page.id, body, { Authorization: `Bearer ${secret}` })).status, 404, 'a page watch has no webhook');
+ await api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });
+ assert.strictEqual((await deliver(watch.id, body, { Authorization: `Bearer ${secret}` })).status, 409);
+ await api('POST', `/api/v1/watches/${watch.id}/resume`, { token: ME });
+ assert.strictEqual((await runs(watch.id)).length, 2, 'no refused delivery ran a check');
+});
+
+t('the hourly budget answers 429 with Retry-After; a body over 256 KB is 413', async () => {
+ const body = JSON.stringify({ order: { total: 1 } });
+ const statuses = [];
+ for (let i = 0; i < 3; i++) statuses.push((await deliver(watch.id, body, { Authorization: `Bearer ${secret}` })).status);
+ assert.deepStrictEqual(statuses, [200, 200, 429], 'two deliveries were already used this hour (budget 4)');
+ const limited = await deliver(watch.id, body, { Authorization: `Bearer ${secret}` });
+ assert.ok(Number(limited.headers.get('retry-after')) > 0);
+ const big = 'x'.repeat(300 * 1024);
+ assert.strictEqual((await deliver(watch.id, big, { Authorization: `Bearer ${secret}` })).status, 413);
+});
+
+t('rotating replaces the secret at once; only the owner may; Check now waits for a delivery', async () => {
+ const rotated = await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME });
+ assert.strictEqual(rotated.status, 200);
+ assert.notStrictEqual(rotated.body.webhook.secret, secret);
+ assert.strictEqual((await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: OTHER })).status, 404);
+ const out = await svc.check.run(watch.id, { trigger: 'manual' });
+ assert.strictEqual(out.run.error_code, 'watch.webhook_awaited');
+ assert.strictEqual(verified(secret, Buffer.from('{}'), { authorization: `Bearer ${secret}` }), true, 'the helper itself');
+ const stored = await svc.db.prepare('SELECT secret FROM webhook_secrets WHERE watch_id = ?').get(watch.id);
+ assert.strictEqual(stored.secret, rotated.body.webhook.secret, 'the old secret is gone');
+ secret = rotated.body.webhook.secret;
+});
+
+t('a non-JSON delivery is read as text; JSON that is not JSON is a parse error, recorded', async () => {
+ const text = await svc.registry.create(def({ name: 'plain', extraction: { kind: 'text' }, condition: { op: 'contains', value: 'paid' } }), { sub: me });
+ const { secret: s } = await svc.webhookSources.issue(text.id);
+ const ok = await fetch(`${svc.base}/hooks/${text.id}`, { method: 'POST', body: 'order paid', headers: { 'Content-Type': 'text/plain', Authorization: `Bearer ${s}` } });
+ assert.strictEqual((await ok.json()).triggered, true);
+ const bad = await fetch(`${svc.base}/hooks/${text.id}`, { method: 'POST', body: '{nope', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${s}` } });
+ assert.strictEqual((await bad.json()).state, 'parse_error');
+});
+
+t('deleting the watch forgets its secret', async () => {
+ assert.strictEqual((await api('DELETE', `/api/v1/watches/${watch.id}`, { token: ME })).status, 200);
+ assert.strictEqual(await svc.db.prepare('SELECT secret FROM webhook_secrets WHERE watch_id = ?').get(watch.id), undefined);
+ assert.strictEqual((await deliver(watch.id, '{}', { Authorization: `Bearer ${secret}` })).status, 404);
+});
+
+t('stop', async () => { await svc.stop(); });
+
+t.run();