openvibers/openvibe.watch Commits
Watch: webhook actions delivered (signed, SSRF-guarded, retried); named secrets need a grant (#33)
Changes
@@@ -93,6 +93,13 @@ VALKEY_PREFIX=# must start with WATCH_CRED_. Read at fetch time, never stored, never logged.#WATCH_CRED_EXAMPLE_API_KEY=--# Webhook secrets (step 5): a watch's source.endpoint is signed with the named variable. Names must--# start with WATCH_HOOK_ and hold at least 32 characters.++# Webhook keys: a webhook action signs its delivery with the variable its secret_env names. Names must++# start with WATCH_HOOK_ and the value hold at least 32 characters.#WATCH_HOOK_EXAMPLE_SECRET=++# A receiver answers within this, or the attempt is retried on the backoff.++#WATCH_WEBHOOK_TIMEOUT_MS=10000++++# Who may use which named secret (server/secret-grants.js). A name is not a capability: without a grant++# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.++# Owners are usr_ or agt_ subjects; origins are scheme://host[:port].++#WATCH_SECRET_GRANTS={"WATCH_CRED_EXAMPLE_API_KEY":{"owners":["usr_..."],"origins":["https://api.example.com"]},"WATCH_HOOK_EXAMPLE_SECRET":{"owners":["usr_..."]}}@@@ -177,7 +177,8 @@ A fired condition queues the watch's actions in the same transaction as the chec|---|---|| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` || `event` | `watch.watch.triggered` is emitted with every firing in the run's own commit, so that action is recorded `sent` as it is queued (the event names a person: an agent- or service-owned watch is `skipped`). Any other event type is not declared by Watch: `unavailable` |--| `webhook`, `actor-task`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |++| `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` |++| `actor-task`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.@@@ -319,9 +320,16 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:resolved address public at connect time (no DNS rebinding), each redirect hop re-checked, onedeadline and a byte cap per response. Exactly one path reaches the network — the carriers, throughthe fetcher — and only for a kind whose carrier exists. `test/carriers.test.js`.--- **Secrets.** A source's credential is the **name** of a `WATCH_CRED_*` variable, read at fetch-- time, sent only to the configured origin, never stored or logged. `OV_OAUTH_CLIENT_SECRET` and the-- credentials live in `/etc/openvibe/watch.env` (0600).++- **Secrets.** A source's credential is the **name** of a `WATCH_CRED_*` variable, and a webhook++ action's signing key the name of a `WATCH_HOOK_*` one. Both are read at use time, never stored or++ logged. A name is **not a capability**: the operator grants each one in `WATCH_SECRET_GRANTS`++ ([server/secret-grants.js](server/secret-grants.js)). A credential may be used only by the owners++ it is granted to, and sent only to the origins it is granted for. A webhook key may be used only by++ its owners. Otherwise anyone could name an operator's credential with a URL of their own, or have++ deliveries signed with another receiver's key. A name without a grant is usable by nobody. A save++ that names one is refused with 422 `watch.secret_not_granted` (the form shows why), and the carrier++ and the action worker check again at use. `OV_OAUTH_CLIENT_SECRET`, the credentials and the grants++ live in `/etc/openvibe/watch.env` (0600).- **Data.** An observation records what the source stated, never a guess; snapshots are capped andpruned by retention; a failed check never records an observation.@@@ -19,7 +19,16 @@* event `watch.watch.triggered` is emitted with every firing, in the run's own commit, so that action is* `sent` as it is queued (it needs a person as owner, as the event does). Any other event type is not* declared in Watch's manifest: `unavailable`.-- * webhook, actor-task, codes-run, run-job++ * webhook a POST of the firing (webhookFor below) to the action's https URL through the SSRF guard++ * (server/net/post.js), signed with the key its secret_env names exactly as OpenVibe.Events signs a++ * delivery (openvibe-sdk/events signDeliveryHeaders: X-OpenVibe-Timestamp + X-OpenVibe-Signature-V2,++ * which a receiver checks with the SDK's verifier). X-OpenVibe-Delivery-Id (`<check run>.<action++ * index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 and a URL++ * the guard refuses are `failed` at once. Anything else is retried on the same backoff, then `failed`.++ * The key must be granted to the watch's owner (WATCH_SECRET_GRANTS, server/secret-grants.js) and++ * set (32 characters or more), checked when it is queued and again at delivery: otherwise++ * `unavailable`.++ * actor-task, codes-run, run-job* `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and* the watch page says so rather than pretending they ran.*@@@ -29,13 +38,18 @@* status() says delivery is off.*/++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];const LEASE_MS = 2 * 60_000;const TITLE_MAX = 120;const MESSAGE_MAX = 300;--const LATER = new Set(['webhook', 'actor-task', 'codes-run', 'run-job']);++const LATER = new Set(['actor-task', 'codes-run', 'run-job']);const TRIGGERED = 'watch.watch.triggered';++const MIN_HOOK_SECRET = 32;++// A URL the guard refuses (server/net/guard.js) will be refused on every attempt: the delivery fails at once.++const REFUSED_URL = new Set(['bad_url', 'port_refused', 'address_refused']);const clip = (s, n) => { const t = String(s == null ? '' : s).replace(/\s+/g, ' ').trim(); return t.length > n ? `${t.slice(0, n - 1)}…` : t; };@@@ -64,7 +78,24 @@ function notificationFor({ config, watch, run, action, value, previousValue, obs};}--function createActions({ db, config, notifications = null, now = () => Date.now(), log = console }) {++/** The body a webhook action POSTs: the firing, as the watch.watch.triggered event states it. */++function webhookFor({ config, watch, run, value, previousValue, observedAt }) {++ return {++ type: TRIGGERED,++ watch: { id: watch.id, name: watch.name, url: `${config.baseUrl}/watches/${encodeURIComponent(watch.id)}` },++ check_run_id: run.id,++ condition: watch.condition || null,++ observation: {++ observed_at: observedAt ? new Date(observedAt).toISOString() : null,++ value: value === undefined ? null : value,++ previous_value: previousValue === undefined ? null : previousValue,++ },++ };++}++++function createActions({ db, config, notifications = null, post = null, now = () => Date.now(), log = console }) {++ // Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.++ const grants = () => config.secretGrants;const enabled = Boolean(notifications) && config.actions.enabled;let timer = null;let busy = null;@@@ -72,7 +103,7 @@ function createActions({ db, config, notifications = null, now = () => Date.now(const st = {insert: db.prepare(`INSERT INTO action_runs (watch_id, check_run_id, action_index, kind, owner_sub, request, state, last_error, created_at, finished_at)VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (check_run_id, action_index) DO NOTHING`),-- claim: db.prepare(`SELECT rid, watch_id, kind, owner_sub, request, attempts FROM action_runs++ claim: db.prepare(`SELECT rid, watch_id, check_run_id, action_index, kind, owner_sub, request, attempts FROM action_runsWHERE state = 'queued' AND next_attempt_at <= ? ORDER BY next_attempt_at, rid LIMIT ? FOR UPDATE SKIP LOCKED`),lease: db.prepare('UPDATE action_runs SET attempts = attempts + 1, next_attempt_at = ? WHERE rid = ?'),finish: db.prepare('UPDATE action_runs SET state = ?, last_error = ?, finished_at = ? WHERE rid = ?'),@@@ -98,6 +129,10 @@ function createActions({ db, config, notifications = null, now = () => Date.now(if (action.event_type !== TRIGGERED) { state = 'unavailable'; error = `watch.action_unavailable: ${action.event_type || 'an event'} is not an event Watch declares (only ${TRIGGERED})`; }else if (!USER_SUBJECT.test(String(owner || ''))) { state = 'skipped'; error = `watch.action_no_person: ${TRIGGERED} names a person, and this watch belongs to an agent or a service`; }else { state = 'sent'; error = 'emitted with the check run'; }++ } else if (kind === 'webhook') {++ const why = grants() ? grants().refusal(action.secret_env, owner) : `${action.secret_env} is not granted to any owner`;++ if (why) { state = 'unavailable'; error = `watch.action_secret_not_granted: ${why}`; }++ else request = { url: action.url, secret_env: action.secret_env, body: webhookFor({ config, watch, run, value, previousValue, observedAt }) };} else if (LATER.has(kind)) {state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;} else {@@@ -107,8 +142,36 @@ function createActions({ db, config, notifications = null, now = () => Date.now(}}-- async function deliver(row) {++ async function deliverWebhook(row, request, attempt) {++ const why = grants() ? grants().refusal(request.secret_env, row.owner_sub) : 'no grants';++ if (why) return { state: 'unavailable', error: `watch.action_secret_not_granted: ${why}` };++ const secret = config.secrets ? config.secrets[request.secret_env] : null;++ if (!secret || String(secret).length < MIN_HOOK_SECRET) {++ return { state: 'unavailable', error: `watch.action_secret_missing: ${request.secret_env} is not set (${MIN_HOOK_SECRET} characters or more)` };++ }++ if (!post) throw Object.assign(new Error('webhook delivery is not configured in this process'), { code: 'webhook_unconfigured' });++ const body = JSON.stringify(request.body);++ const headers = {++ ...signDeliveryHeaders(body, String(secret), { now: now() }),++ 'X-OpenVibe-Event-Type': TRIGGERED,++ 'X-OpenVibe-Delivery-Id': `${row.check_run_id}.${row.action_index}`,++ 'X-OpenVibe-Delivery-Attempt': String(attempt),++ };++ let res;++ try {++ res = await post(request.url, { headers, body });++ } catch (err) {++ if (err && REFUSED_URL.has(err.code)) return { state: 'failed', error: `watch.action_url_refused: ${err.message}` };++ throw err;++ }++ if (res.status >= 200 && res.status < 300) return { state: 'sent', error: null };++ if (res.status === 410) return { state: 'failed', error: 'watch.action_gone: the receiver answered 410 Gone' };++ throw Object.assign(new Error(`the receiver answered ${res.status}`), { code: `http_${res.status}` });++ }++++ async function deliver(row, attempt) {const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;++ if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };const out = await notifications.push({ subjectId: row.owner_sub, ...request });if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };@@@ -130,9 +193,11 @@ function createActions({ db, config, notifications = null, now = () => Date.now(for (const r of rows) {const attempt = Number(r.attempts) + 1;try {-- const res = await deliver(r);++ const res = await deliver(r, attempt);await st.finish.run(res.state, res.error, now(), r.rid);-- if (res.state === 'sent') out.sent++; else out.skipped++;++ if (res.state === 'sent') out.sent++;++ else if (res.state === 'failed') out.failed++;++ else out.skipped++;} catch (err) {const why = String((err && (err.code || err.message)) || err).slice(0, 300);if (attempt >= BACKOFF_MS.length + 1) {@@@ -178,4 +243,4 @@ function createActions({ db, config, notifications = null, now = () => Date.now(return { queue, tick, start, stop, status, recent, enabled };}--module.exports = { createActions, notificationFor, BACKOFF_MS };++module.exports = { createActions, notificationFor, webhookFor, BACKOFF_MS };@@@ -36,7 +36,7 @@ function createApiCarrier({ fetcher, config }) {run: async ({ watch, state }) => {const source = watch.source;const out = await conditionalGet({-- source, state, fetcher, secrets: config.secrets, accept: ACCEPT.api,++ source, state, fetcher, secrets: config.secretGrants.viewFor(watch, config.secrets), accept: ACCEPT.api,onlyChangeNeeded: onlyChange(watch),});if (out.state !== 'ok' || !out.body) return out;@@@ -125,7 +125,7 @@ function createFeedCarrier({ fetcher, config }) {name: 'feed',run: async ({ watch, state }) => {const out = await conditionalGet({-- source: watch.source, state, fetcher, secrets: config.secrets, accept: ACCEPT.feed,++ source: watch.source, state, fetcher, secrets: config.secretGrants.viewFor(watch, config.secrets), accept: ACCEPT.feed,onlyChangeNeeded: onlyChange(watch),});if (out.state !== 'ok' || !out.body) return out;@@@ -151,7 +151,7 @@ function createHttpCarrier({ fetcher, config }) {return {name: 'http',run: async ({ watch, state }) => conditionalGet({-- source: watch.source, state, fetcher, secrets: config.secrets, accept: ACCEPT.http,++ source: watch.source, state, fetcher, secrets: config.secretGrants.viewFor(watch, config.secrets), accept: ACCEPT.http,onlyChangeNeeded: onlyChange(watch),}),};@@@ -12,6 +12,7 @@* lives; nothing in this release executes it (steps 5-7).*/const pkg = require('../package.json');++const { parseGrants, createSecretGrants } = require('./secret-grants');const int = (v, d) => {const n = parseInt(v, 10);@@@ -85,8 +86,14 @@ function load(env = process.env) {enabled: env.WATCH_ACTIONS !== 'off',tickMs: int(env.WATCH_ACTIONS_TICK_MS, 5000),batch: int(env.WATCH_ACTIONS_BATCH, 20),++ // A webhook receiver answers within this, or the attempt is retried on the backoff.++ webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),},++ // Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins++ // it may be sent to) or a WATCH_HOOK_* webhook key (owners). No grant, no use.++ secretGrants: createSecretGrants({ grants: parseGrants(env.WATCH_SECRET_GRANTS) }),++// The check worker: every WATCH_TICK_MS, start checks for active watches whose turn has come.worker: {enabled: env.WATCH_WORKER !== 'off',@@@ -14,6 +14,7 @@ const { openDb } = require('./db');const { createRegistry } = require('./registry');const { createObservations } = require('./observations');const { createGuard } = require('./net/guard');++const { createPoster } = require('./net/post');const { createFetcher } = require('./net/fetcher');const { createSpacer } = require('./spacer');const { createCarriers } = require('./carriers');@@@ -36,7 +37,7 @@ const { createApp } = require('./app');const accountDataLib = require('./account-data');const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null } = {}) {++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null } = {}) {config = config || load();// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.const db = givenDb || await openDb(config, { log });@@@ -48,7 +49,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],validate: (env) => validate('events.event-envelope@1', env),});-- const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner });++ const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants });const observations = createObservations({ db, now });// The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte// cap, one deadline, conditional GET; per-host spacing in front of every request.@@@ -65,7 +66,9 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchbaseUrls: { network: config.networkInternalUrl }, fetch: fetchImpl,})): null;-- const actions = createActions({ db, config, notifications: givenNotifications || notifications, now, log });++ // Webhook actions leave through the same guard as every fetch (server/net/post.js).++ const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });++ const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, now, log });const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });const scheduler = createScheduler({ db, check, config, now, log });const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });nnew file mode 100644@@@ -0,0 +1,46 @@++'use strict';++/**++ * The one way a webhook action reaches a receiver (plan T18 step 5): POST a JSON body to a URL the watch's owner chose,++ * through the same SSRF guard as every fetch (server/net/guard.js: http(s) on the allowed ports, no credentials in the++ * URL, every resolved address public at connect time, so a DNS answer cannot change between check and connect).++ *++ * const post = createPoster({ guard, userAgent, timeoutMs });++ * const { status } = await post(url, { headers, body }); // throws on a refused URL, a timeout or a lost connection++ *++ * A redirect is never followed: a 3xx is the receiver's answer, like any other non-2xx. The response body is read only++ * up to a few KiB and discarded; nothing a receiver says is stored.++ */++const http = require('http');++const https = require('https');++++const MAX_RESPONSE_BYTES = 4096;++++function createPoster({ guard, userAgent, timeoutMs = 10_000 }) {++ return async function post(url, { headers = {}, body = '' } = {}) {++ const u = guard.checkUrl(url); // GuardError: bad_url, port_refused, address_refused++ const mod = u.protocol === 'https:' ? https : http;++ const payload = Buffer.from(String(body));++ return await new Promise((resolve, reject) => {++ let settled = false;++ const finish = (fn, v) => { if (!settled) { settled = true; fn(v); } };++ const req = mod.request(u, {++ method: 'POST',++ lookup: guard.lookup,++ timeout: timeoutMs,++ headers: { 'Content-Type': 'application/json', 'Content-Length': payload.length, 'User-Agent': userAgent, ...headers },++ }, (res) => {++ let seen = 0;++ res.on('data', (chunk) => { seen += chunk.length; if (seen > MAX_RESPONSE_BYTES) res.destroy(); });++ const done = () => finish(resolve, { status: res.statusCode });++ res.on('end', done);++ res.on('close', done);++ res.on('error', done);++ });++ req.on('timeout', () => req.destroy(Object.assign(new Error(`no answer within ${timeoutMs} ms`), { code: 'ETIMEDOUT' })));++ req.on('error', (err) => finish(reject, err));++ req.end(payload);++ });++ };++}++++module.exports = { createPoster };@@@ -122,7 +122,14 @@ function validateView(view) {return view;}--function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50 }) {++function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null }) {++ /** A named secret the owner may not use is refused when the watch is saved (server/secret-grants.js). */++ function checkSecrets(owner, rec) {++ if (!secretGrants) return;++ const bad = secretGrants.check(owner, rec);++ if (bad.length) throw new RegistryError(`${bad[0].path}: ${bad[0].reason}`, 'watch.secret_not_granted');++ }++const st = {get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),countOwned: db.prepare('SELECT COUNT(*) AS n FROM watches WHERE owner_sub = ? AND deleted_at IS NULL'),@@@ -236,6 +243,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outboxconst req = validateRequest(body, { create: true });const owner = ownerOf(principal);if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');++ checkSecrets(owner, req);// A per-owner cap, so one caller cannot fill the registry (the API and the form both come// through here). The number is a config value (WATCH_MAX_PER_OWNER).const owned = await st.countOwned.get(owner);@@@ -297,6 +305,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox// caller that sends one fails the schema check below rather than being silently ignored.if (changes.project_id === undefined) delete merged.project_id;validateRequest(merged);++ checkSecrets(row.owner_sub, merged);const sets = [];const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };for (const [k, v] of Object.entries(changes)) {nnew file mode 100644@@@ -0,0 +1,97 @@++'use strict';++/**++ * Who may use a named secret (plan T18). A watch names its secrets by environment variable and Watch reads the value++ * at use time, never storing it:++ *++ * source.auth.env WATCH_CRED_* a source's credential, sent with the watch's own request++ * action.secret_env WATCH_HOOK_* the key a webhook action signs its delivery with++ *++ * The NAME alone must not be a capability. Without a grant, anyone could create a watch that names an operator's++ * credential, point it at a URL of their own and receive the credential. They could also have their deliveries signed++ * with another receiver's key and replay them there. So the operator grants each name, in WATCH_SECRET_GRANTS (JSON):++ *++ * {"WATCH_CRED_ACME": {"owners": ["usr_…"], "origins": ["https://api.acme.example"]},++ * "WATCH_HOOK_ALEX": {"owners": ["usr_…"]}}++ *++ * A credential is used only by a granted owner and sent only to a granted origin (scheme, host and port, exactly). A++ * webhook key is used only by a granted owner. A name with no grant is usable by nobody, and that is the default.++ * The registry checks when a watch is saved (the API and the form say why), and the carriers and the action worker++ * check again at use, because a grant can be withdrawn. Malformed JSON grants nothing.++ */++++// A person (usr_) or one of their agents (agt_) may own a watch, so either may hold a grant.++const USER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;++const NAME = /^WATCH_(CRED|HOOK)_[A-Z0-9_]{2,60}$/;++++function originOf(u) {++ try {++ const x = new URL(String(u));++ return x.protocol === 'https:' || x.protocol === 'http:' ? x.origin : null;++ } catch { return null; }++}++++/** WATCH_SECRET_GRANTS → { name: { owners: Set, origins: Set } }. Anything malformed is dropped (never widened). */++function parseGrants(raw, log = console) {++ if (raw == null || String(raw).trim() === '') return {};++ let v;++ try { v = JSON.parse(raw); } catch {++ log.warn('[watch] WATCH_SECRET_GRANTS is not JSON: no watch can use a named secret');++ return {};++ }++ if (!v || typeof v !== 'object' || Array.isArray(v)) {++ log.warn('[watch] WATCH_SECRET_GRANTS must be an object of grants: no watch can use a named secret');++ return {};++ }++ const out = {};++ for (const [name, g] of Object.entries(v)) {++ if (!NAME.test(name) || !g || typeof g !== 'object') continue;++ const owners = (Array.isArray(g.owners) ? g.owners : []).map(String).filter((o) => USER.test(o));++ const origins = (Array.isArray(g.origins) ? g.origins : []).map(originOf).filter(Boolean);++ out[name] = { owners: new Set(owners), origins: new Set(origins) };++ }++ return out;++}++++function createSecretGrants({ grants = {} } = {}) {++ /** null when `owner` may use `name` (a credential: toward `url`), else why not, in one line. */++ function refusal(name, owner, url = undefined) {++ const g = grants[name];++ if (!g) return `${name} is not granted to any owner (WATCH_SECRET_GRANTS)`;++ if (!g.owners.has(String(owner || ''))) return `${name} is not granted to this owner`;++ if (url !== undefined) {++ const o = originOf(url);++ if (!o || !g.origins.has(o)) return `${name} may not be sent to ${o || 'that URL'}`;++ }++ return null;++ }++++ /** The named secrets of a watch (request-shaped) its owner may not use: [{ path, reason }], empty when all are. */++ function check(owner, rec) {++ const out = [];++ const source = rec && rec.source;++ const auth = source && source.auth;++ if (auth && auth.mode && auth.mode !== 'none' && auth.env) {++ const why = refusal(auth.env, owner, source.url || '');++ if (why) out.push({ path: '/source/auth/env', reason: why });++ }++ (Array.isArray(rec && rec.action) ? rec.action : []).forEach((a, i) => {++ if (a && a.kind === 'webhook' && a.secret_env) {++ const why = refusal(a.secret_env, owner);++ if (why) out.push({ path: `/action/${i}/secret_env`, reason: why });++ }++ });++ return out;++ }++++ /** The environment a carrier may read for this watch: its own credential, if granted toward its own origin. */++ function viewFor(watch, env) {++ const source = watch && watch.source;++ const auth = source && source.auth;++ if (!auth || !auth.env || !env) return {};++ return refusal(auth.env, watch.owner, source.url || '') ? {} : { [auth.env]: env[auth.env] };++ }++++ return { refusal, check, viewFor };++}++++module.exports = { parseGrants, createSecretGrants, originOf };@@@ -1,11 +1,15 @@'use strict';/**-- * Actions (plan T18 step 4): a condition that fires queues the watch's actions in the run's own transaction, once per++ * Actions (plan T18 steps 4-5): a condition that fires queues the watch's actions in the run's own transaction, once per* trigger; the worker delivers a notification to the owner through Network (openvibe-sdk/notifications), retries a* refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds-- * whose dispatchers are later steps as unavailable. The watch page says what happened.++ * whose dispatchers are later steps as unavailable. A webhook action POSTs the firing signed like an Events delivery,++ * with a delivery id that is the same on every attempt; a busy receiver is retried, 410 and a refused URL fail at once;++ * its key must be granted to the owner (refused at save, unavailable at delivery once withdrawn) and long enough. The++ * watch page says what happened.*/const assert = require('assert');++const { verifyDeliveryV2 } = require('openvibe-sdk/events');const { boot, site, suite, newUser, newAgent } = require('./helpers');const t = suite('actions');@@@ -25,6 +29,18 @@ const notifications = {const person = { sub: newUser() };const agent = { sub: newAgent() };++const stranger = { sub: newUser() };++const HOOK_KEY = 'k'.repeat(24) + 'Watch-hook-key-0123456789';++// The receiver, as server/net/post.js would reach it: every POST recorded; hookMode decides the answer.++const posts = [];++let hookMode = 'ok';++async function post(url, { headers, body }) {++ posts.push({ url, headers, body });++ if (hookMode === 'refused') throw Object.assign(new Error('hooks.example.org resolves to a non-public address'), { code: 'address_refused' });++ if (hookMode === 'down') throw Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' });++ return { status: { ok: 204, busy: 503, gone: 410 }[hookMode] };++}++const hook = (env = 'WATCH_HOOK_TEST') => ({ kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: env });const def = (overrides = {}) => ({name: 'Widget price',source: { kind: 'http', url: `${web.origin}/price`, format: null },@@@ -53,12 +69,20 @@ t('boot', async () => {'/price': () => ({ headers: { 'content-type': 'text/plain' }, body: price }),'/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },});-- svc = await boot({ notifications });++ svc = await boot({++ notifications, post,++ env: {++ WATCH_HOOK_TEST: HOOK_KEY, WATCH_HOOK_SHORT: 'too-short', WATCH_HOOK_THEIRS: HOOK_KEY,++ WATCH_SECRET_GRANTS: JSON.stringify({++ WATCH_HOOK_TEST: { owners: [person.sub] }, WATCH_HOOK_SHORT: { owners: [person.sub] }, WATCH_HOOK_THEIRS: { owners: [stranger.sub] },++ }),++ },++ });});t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {mode = 'ok'; pushes.length = 0;-- const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: 'WATCH_HOOK_TEST' }] });++ const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'actor-task', brief: 'look into it' }] });await setPrice('19.99');await check(w.id);assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');@@@ -66,7 +90,7 @@ t('a check that does not fire queues nothing; one that fires queues each actionconst out = await check(w.id);const rows = await acts(w.id);assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),-- [['notification', 'queued', out.run.id, 0], ['webhook', 'unavailable', out.run.id, 1]]);++ [['notification', 'queued', out.run.id, 0], ['actor-task', 'unavailable', out.run.id, 1]]);assert.match(rows[1].last_error, /watch\.action_unavailable/);const r = await svc.actions.tick();@@@ -141,10 +165,99 @@ t('an owner Network does not know is skipped; a muted category is delivered; anassert.strictEqual(pushes.length, before, 'nothing is pushed for an agent');});++t('a webhook action POSTs the firing, signed as an Events delivery; the receiver verifies it with the SDK', async () => {++ posts.length = 0; hookMode = 'ok';++ const w = await create(person, { name: 'hooked', action: [hook()] });++ const out = await fire(w);++ let [row] = await acts(w.id);++ assert.strictEqual(row.state, 'queued');++ const r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });++ assert.strictEqual(posts.length, 1);++ const p = posts[0];++ assert.strictEqual(p.url, 'https://hooks.example.org/w');++ const lower = Object.fromEntries(Object.entries(p.headers).map(([k, v]) => [k.toLowerCase(), v]));++ assert.strictEqual(lower['x-openvibe-event-type'], 'watch.watch.triggered');++ assert.strictEqual(lower['x-openvibe-delivery-id'], `${out.run.id}.0`);++ assert.strictEqual(lower['x-openvibe-delivery-attempt'], '1');++ assert.ok(verifyDeliveryV2(p.body, lower, HOOK_KEY), 'the signature verifies with the key');++ assert.ok(!verifyDeliveryV2(p.body, lower, 'x'.repeat(48)), 'and with no other key');++ assert.ok(!p.body.includes(HOOK_KEY), 'the key itself is never sent');++ const body = JSON.parse(p.body);++ assert.deepStrictEqual([body.type, body.watch.id, body.watch.name, body.check_run_id], ['watch.watch.triggered', w.id, 'hooked', out.run.id]);++ assert.strictEqual(body.watch.url, `https://openvibe.watch/watches/${w.id}`);++ assert.deepStrictEqual([body.observation.value, body.observation.previous_value], ['7.50', '19.99']);++ [row] = await acts(w.id);++ assert.deepStrictEqual([row.state, row.last_error], ['sent', null]);++ const stored = await svc.db.prepare('SELECT request FROM action_runs WHERE rid = ?').get(row.rid);++ assert.ok(!JSON.stringify(stored).includes(HOOK_KEY), 'the key is never stored');++});++++t('a busy receiver is retried on the backoff with the same delivery id; 410 and a refused URL fail at once', async () => {++ posts.length = 0; hookMode = 'busy';++ const w = await create(person, { name: 'busy', action: [hook()] });++ await fire(w);++ let r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 });++ let [row] = await acts(w.id);++ assert.deepStrictEqual([row.state, row.last_error], ['queued', 'http_503']);++ hookMode = 'down';++ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);++ r = await svc.actions.tick();++ assert.strictEqual(r.retried, 1, 'a lost connection is retried too');++ hookMode = 'ok';++ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);++ r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });++ assert.strictEqual(posts.length, 3);++ const ids = posts.map((x) => x.headers['X-OpenVibe-Delivery-Id']);++ assert.strictEqual(new Set(ids).size, 1, 'one delivery id across attempts');++ assert.deepStrictEqual(posts.map((x) => x.headers['X-OpenVibe-Delivery-Attempt']), ['1', '2', '3']);++++ for (const [mode, code] of [['gone', /action_gone/], ['refused', /action_url_refused/]]) {++ hookMode = mode;++ const g = await create(person, { name: `hook ${mode}`, action: [hook()] });++ await fire(g);++ r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 0, failed: 1 }, mode);++ [row] = await acts(g.id);++ assert.strictEqual(row.state, 'failed');++ assert.match(row.last_error, code);++ }++ hookMode = 'ok';++});++++t('a webhook key must be granted to the owner (refused at save, unavailable once withdrawn) and long enough', async () => {++ posts.length = 0; hookMode = 'ok';++ await assert.rejects(create(person, { name: 'theirs', action: [hook('WATCH_HOOK_THEIRS')] }),++ (e) => e.code === 'watch.secret_not_granted' && /\/action\/0\/secret_env: WATCH_HOOK_THEIRS is not granted to this owner/.test(e.message));++ await assert.rejects(create(person, { name: 'nobody', action: [hook('WATCH_HOOK_NOBODY')] }),++ (e) => e.code === 'watch.secret_not_granted' && /not granted to any owner/.test(e.message));++ const ok = await create(person, { name: 'patched' });++ await assert.rejects(svc.registry.patch(ok.id, { action: [hook('WATCH_HOOK_THEIRS')] }, person), (e) => e.code === 'watch.secret_not_granted', 'a patch is checked too');++++ const short = await create(person, { name: 'short key', action: [hook('WATCH_HOOK_SHORT')] });++ await fire(short);++ let r = await svc.actions.tick();++ assert.strictEqual(r.skipped, 1);++ assert.match((await acts(short.id))[0].last_error, /action_secret_missing: WATCH_HOOK_SHORT is not set \(32 characters or more\)/);++++ // Granted when it fired, withdrawn before delivery: the worker checks again.++ const later = await create(person, { name: 'withdrawn', action: [hook()] });++ await fire(later);++ const grants = svc.config.secretGrants;++ const { createSecretGrants } = require('../server/secret-grants');++ svc.config.secretGrants = createSecretGrants({ grants: {} });++ try {++ r = await svc.actions.tick();++ } finally { svc.config.secretGrants = grants; }++ assert.strictEqual(posts.length, 0, 'nothing was posted');++});++t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {const st = await svc.actions.status();assert.strictEqual(st.enabled, true);-- assert.ok(st.sent >= 3 && st.queued === 0 && st.failed === 0, JSON.stringify(st));++ assert.ok(st.sent >= 3 && st.queued === 0, JSON.stringify(st));const w = await create(person, { name: 'page check' });mode = 'ok';await fire(w);@@@ -6,14 +6,16 @@* address, a private host, a port the guard does not allow.*/const assert = require('assert');--const { boot, request, site, serviceToken, suite, rss } = require('./helpers');++const { boot, request, site, serviceToken, suite, rss, newUser } = require('./helpers');const t = suite('carriers');const OWNER = { sub: 'svc:carrier-test' };++// The person WATCH_CRED_TEST is granted to (WATCH_SECRET_GRANTS), toward the stub site's origin only.++const PERSON = { sub: newUser() };let svc;let web;--const watch = async (overrides = {}) => (await svc.registry.create({++const watch = async (overrides = {}, owner = OWNER) => (await svc.registry.create({name: 'carrier watch',source: { kind: 'http', url: `${web.origin}/page`, format: null },cadence: { every_sec: 900 },@@@ -21,7 +23,7 @@ const watch = async (overrides = {}) => (await svc.registry.create({condition: { op: 'eq', value: 'never-matches' },action: [{ kind: 'notification' }],...overrides,--}, OWNER)).id;++}, owner)).id;const check = async (id, trigger = 'manual') => await svc.check.run(id, { trigger });const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);@@@ -44,7 +46,7 @@ t('boot with a stub site', async () => {'/api.json': () => ({ headers: { 'content-type': 'application/json' }, body: JSON.stringify({ products: [{ sku: 'A', price: 19.5 }, { sku: 'B', price: 4 }] }) }),'/secret.json': (req) => ({ headers: { 'content-type': 'application/json' }, body: JSON.stringify({ saw: req.url }) }),});-- svc = await boot();++ svc = await boot({ env: { WATCH_SECRET_GRANTS: JSON.stringify({ WATCH_CRED_TEST: { owners: [PERSON.sub], origins: [web.origin] } }) } });});t('a conditional GET stores the validators, and the next check gets 304 → not_modified', async () => {@@@ -148,7 +150,7 @@ t('a credential is read from the environment by name, at fetch time, and never sconst id = await watch({source: { kind: 'api', url: `${web.origin}/secret.json`, format: 'json', auth: { mode: 'query', env: 'WATCH_CRED_TEST', param: 'k' } },extraction: { kind: 'json', value_path: 'saw' },-- });++ }, PERSON);// not set: the check is recorded disabled and nothing is fetchedconst before = web.hits('/secret.json').length;const missing = await check(id);@@@ -156,7 +158,8 @@ t('a credential is read from the environment by name, at fetch time, and never sassert.strictEqual(missing.run.error_code, 'credential_missing');assert.strictEqual(web.hits('/secret.json').length, before, 'no request was made');// the stored watch and its view carry the NAME, never a value-- const view = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: serviceToken('carrier-test', ['watch.watch.read']) });++ // Read as a first-party service acting for the owner (X-OV-Subject), the way the public site reads it.++ const view = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: serviceToken('carrier-test', ['watch.watch.read']), headers: { 'X-OV-Subject': PERSON.sub } });assert.strictEqual(view.status, 200);assert.deepStrictEqual(view.body.watch.source.auth, { mode: 'query', env: 'WATCH_CRED_TEST', param: 'k' });assert.ok(!view.text.includes('s3cret-value'));@@@ -171,6 +174,32 @@ t('a credential is read from the environment by name, at fetch time, and never sdelete svc.config.secrets.WATCH_CRED_TEST;});++t('a credential name is not a capability: only a granted owner may use it, and only toward a granted origin', async () => {++ const def = (url) => ({ source: { kind: 'api', url, format: 'json', auth: { mode: 'bearer', env: 'WATCH_CRED_TEST' } }, extraction: { kind: 'json', value_path: 'saw' } });++ // Anyone else naming it (a service, another person) is refused when the watch is saved.++ await assert.rejects(watch(def(`${web.origin}/secret.json`)), (e) => e.code === 'watch.secret_not_granted' && /\/source\/auth\/env: WATCH_CRED_TEST is not granted to this owner/.test(e.message));++ await assert.rejects(watch(def(`${web.origin}/secret.json`), { sub: newUser() }), (e) => e.code === 'watch.secret_not_granted');++ // The granted owner cannot send it anywhere else: the origin is part of the grant.++ await assert.rejects(watch(def('https://collector.example.org/steal'), PERSON), (e) => e.code === 'watch.secret_not_granted' && /may not be sent to https:\/\/collector\.example\.org/.test(e.message));++ const elsewhere = await watch({ source: { kind: 'api', url: `${web.origin}/secret.json`, format: 'json' }, extraction: { kind: 'json', value_path: 'saw' } }, PERSON);++ await assert.rejects(svc.registry.patch(elsewhere, def('https://collector.example.org/steal'), PERSON), (e) => e.code === 'watch.secret_not_granted', 'a patch is checked too');++ // Withdrawn after the watch was saved: the carrier sees no credential and sends nothing.++ const id = await watch(def(`${web.origin}/secret.json`), PERSON);++ svc.config.secrets.WATCH_CRED_TEST = 's3cret-value';++ const grants = svc.config.secretGrants;++ const { createSecretGrants } = require('../server/secret-grants');++ svc.config.secretGrants = createSecretGrants({ grants: {} });++ const before = web.hits('/secret.json').length;++ try {++ const out = await check(id);++ assert.deepStrictEqual([out.run.state, out.run.error_code], ['disabled', 'credential_missing']);++ } finally {++ svc.config.secretGrants = grants;++ delete svc.config.secrets.WATCH_CRED_TEST;++ }++ assert.strictEqual(web.hits('/secret.json').length, before, 'no request was made');++});++t('the feed carrier parses RSS and the extraction reads the latest entry', async () => {const id = await watch({source: { kind: 'feed', url: `${web.origin}/feed.xml`, format: 'rss' },@@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).*/--async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null } = {}) {++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null } = {}) {// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real// to fetch and verify against (no test touches the internet). Its URL is overridable via env.const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });@@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l});// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.const testdb = await require('./db').testDb();-- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(now ? { now } : {}) });++ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(now ? { now } : {}) });const base = `http://127.0.0.1:${h.server.address().port}`;/** A session cookie for a person, as /auth/callback would have set it. */nnew file mode 100644@@@ -0,0 +1,73 @@++'use strict';++/**++ * The webhook poster (server/net/post.js): a JSON POST through the SSRF guard. It sends the body and headers it is++ * given and answers the receiver's status; it never follows a redirect; a URL the guard refuses (an internal IP++ * literal, a name that resolves to one at connect time, a port off the list) is refused before or at connect; a++ * receiver that never answers times out.++ */++const assert = require('assert');++const http = require('http');++const { createGuard } = require('../server/net/guard');++const { createPoster } = require('../server/net/post');++const { suite } = require('./helpers');++++const t = suite('post');++let server;++let base;++const seen = [];++++t('a receiver', async () => {++ server = http.createServer((req, res) => {++ const chunks = [];++ req.on('data', (c) => chunks.push(c));++ req.on('end', () => {++ seen.push({ method: req.method, url: req.url, headers: req.headers, body: Buffer.concat(chunks).toString('utf8') });++ if (req.url === '/redirect') { res.writeHead(302, { location: 'http://127.0.0.1/elsewhere' }); return res.end(); }++ if (req.url === '/slow') return; // never answers++ res.writeHead(req.url === '/busy' ? 503 : 204);++ res.end();++ });++ });++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ base = `http://127.0.0.1:${server.address().port}`;++});++++t('it POSTs the body and headers through the guard and answers the status; a redirect is not followed', async () => {++ const guard = createGuard({ allowPrivateHosts: ['127.0.0.1'] });++ const post = createPoster({ guard, userAgent: 'OpenVibeWatch/test', timeoutMs: 2000 });++ const r = await post(`${base}/hook`, { headers: { 'X-OpenVibe-Event-Type': 'watch.watch.triggered' }, body: '{"a":1}' });++ assert.strictEqual(r.status, 204);++ const got = seen.pop();++ assert.deepStrictEqual([got.method, got.url, got.body], ['POST', '/hook', '{"a":1}']);++ assert.strictEqual(got.headers['content-type'], 'application/json');++ assert.strictEqual(got.headers['user-agent'], 'OpenVibeWatch/test');++ assert.strictEqual(got.headers['x-openvibe-event-type'], 'watch.watch.triggered');++ assert.strictEqual((await post(`${base}/busy`, { body: '{}' })).status, 503);++ const before = seen.length;++ assert.strictEqual((await post(`${base}/redirect`, { body: '{}' })).status, 302);++ assert.strictEqual(seen.length, before + 1, 'the Location was not fetched');++});++++t('internal addresses are refused: an IP literal before anything is sent, a resolved name at connect time', async () => {++ const strict = createGuard({});++ const post = createPoster({ guard: strict, userAgent: 'x', timeoutMs: 2000 });++ const before = seen.length;++ await assert.rejects(post(`${base}/hook`, { body: '{}' }), (e) => e.code === 'address_refused' || e.code === 'port_refused');++ await assert.rejects(post('http://169.254.169.254/latest/meta-data/', { body: '{}' }), (e) => e.code === 'address_refused');++ await assert.rejects(post('https://user:[email protected]/x', { body: '{}' }), (e) => e.code === 'bad_url');++ // A public-looking name whose DNS answer is internal: refused by the guard's lookup at connect.++ const rebinding = createGuard({ lookupImpl: (host, opts, cb) => cb(null, [{ address: '127.0.0.1', family: 4 }]) });++ const viaName = createPoster({ guard: rebinding, userAgent: 'x', timeoutMs: 2000 });++ await assert.rejects(viaName('https://hooks.example.org/w', { body: '{}' }), (e) => e.code === 'address_refused');++ assert.strictEqual(seen.length, before, 'nothing reached the receiver');++});++++t('a receiver that never answers times out', async () => {++ const guard = createGuard({ allowPrivateHosts: ['127.0.0.1'] });++ const post = createPoster({ guard, userAgent: 'x', timeoutMs: 150 });++ await assert.rejects(post(`${base}/slow`, { body: '{}' }), (e) => e.code === 'ETIMEDOUT');++});++++t('done', async () => { server.closeAllConnections?.(); await new Promise((r) => server.close(r)); });++++t.run();
@@ -93,6 +93,13 @@ VALKEY_PREFIX=
@@ -93,6 +93,13 @@ VALKEY_PREFIX=
# must start with WATCH_CRED_. Read at fetch time, never stored, never logged.
# must start with WATCH_CRED_. Read at fetch time, never stored, never logged.
#WATCH_CRED_EXAMPLE_API_KEY=
#WATCH_CRED_EXAMPLE_API_KEY=
-# Webhook secrets (step 5): a watch's source.endpoint is signed with the named variable. Names must
-# start with WATCH_HOOK_ and hold at least 32 characters.
+# Webhook keys: a webhook action signs its delivery with the variable its secret_env names. Names must
+# start with WATCH_HOOK_ and the value hold at least 32 characters.
#WATCH_HOOK_EXAMPLE_SECRET=
#WATCH_HOOK_EXAMPLE_SECRET=
+# A receiver answers within this, or the attempt is retried on the backoff.
+#WATCH_WEBHOOK_TIMEOUT_MS=10000
+
+# Who may use which named secret (server/secret-grants.js). A name is not a capability: without a grant
+# nobody can use it. A credential needs owners and the origins it may be sent to; a webhook key, owners.
+# Owners are usr_ or agt_ subjects; origins are scheme://host[:port].
+#WATCH_SECRET_GRANTS={"WATCH_CRED_EXAMPLE_API_KEY":{"owners":["usr_..."],"origins":["https://api.example.com"]},"WATCH_HOOK_EXAMPLE_SECRET":{"owners":["usr_..."]}}
@@ -177,7 +177,8 @@ A fired condition queues the watch's actions in the same transaction as the chec
@@ -177,7 +177,8 @@ A fired condition queues the watch's actions in the same transaction as the chec
|---|---|
|---|---|
| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` |
| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` |
| `event` | `watch.watch.triggered` is emitted with every firing in the run's own commit, so that action is recorded `sent` as it is queued (the event names a person: an agent- or service-owned watch is `skipped`). Any other event type is not declared by Watch: `unavailable` |
| `event` | `watch.watch.triggered` is emitted with every firing in the run's own commit, so that action is recorded `sent` as it is queued (the event names a person: an agent- or service-owned watch is `skipped`). Any other event type is not declared by Watch: `unavailable` |
-| `webhook`, `actor-task`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |
+| `webhook` | a POST of the firing (`type`, `watch {id, name, url}`, `check_run_id`, `condition`, `observation {observed_at, value, previous_value}`) to the action's https URL, through the same SSRF guard as every fetch, never following a redirect ([server/net/post.js](server/net/post.js)). It is signed with the key `secret_env` names exactly as OpenVibe.Events signs a delivery (`X-OpenVibe-Timestamp` + `X-OpenVibe-Signature-V2`; check it with openvibe-sdk/events `verifyDeliveryV2`). `X-OpenVibe-Delivery-Id` (`<check run>.<action index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 or a URL the guard refuses is `failed` at once. Anything else, or a lost connection, is retried on the backoff, then `failed`. The key must be granted to the owner (`WATCH_SECRET_GRANTS`, below) and set, 32 characters or more: otherwise `unavailable` |
+| `actor-task`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |
Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.
Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.
Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.
Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.
@@ -319,9 +320,16 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:
@@ -319,9 +320,16 @@ Reporting: [SECURITY.md](SECURITY.md). The rules the code keeps:
resolved address public at connect time (no DNS rebinding), each redirect hop re-checked, one
resolved address public at connect time (no DNS rebinding), each redirect hop re-checked, one
deadline and a byte cap per response. Exactly one path reaches the network — the carriers, through
deadline and a byte cap per response. Exactly one path reaches the network — the carriers, through
the fetcher — and only for a kind whose carrier exists. `test/carriers.test.js`.
the fetcher — and only for a kind whose carrier exists. `test/carriers.test.js`.
-- **Secrets.** A source's credential is the **name** of a `WATCH_CRED_*` variable, read at fetch
- time, sent only to the configured origin, never stored or logged. `OV_OAUTH_CLIENT_SECRET` and the
- credentials live in `/etc/openvibe/watch.env` (0600).
+- **Secrets.** A source's credential is the **name** of a `WATCH_CRED_*` variable, and a webhook
+ action's signing key the name of a `WATCH_HOOK_*` one. Both are read at use time, never stored or
+ logged. A name is **not a capability**: the operator grants each one in `WATCH_SECRET_GRANTS`
+ ([server/secret-grants.js](server/secret-grants.js)). A credential may be used only by the owners
+ it is granted to, and sent only to the origins it is granted for. A webhook key may be used only by
+ its owners. Otherwise anyone could name an operator's credential with a URL of their own, or have
+ deliveries signed with another receiver's key. A name without a grant is usable by nobody. A save
+ that names one is refused with 422 `watch.secret_not_granted` (the form shows why), and the carrier
+ and the action worker check again at use. `OV_OAUTH_CLIENT_SECRET`, the credentials and the grants
+ live in `/etc/openvibe/watch.env` (0600).
- **Data.** An observation records what the source stated, never a guess; snapshots are capped and
- **Data.** An observation records what the source stated, never a guess; snapshots are capped and
pruned by retention; a failed check never records an observation.
pruned by retention; a failed check never records an observation.
@@ -19,7 +19,16 @@
@@ -19,7 +19,16 @@
* event `watch.watch.triggered` is emitted with every firing, in the run's own commit, so that action is
* event `watch.watch.triggered` is emitted with every firing, in the run's own commit, so that action is
* `sent` as it is queued (it needs a person as owner, as the event does). Any other event type is not
* `sent` as it is queued (it needs a person as owner, as the event does). Any other event type is not
* declared in Watch's manifest: `unavailable`.
* declared in Watch's manifest: `unavailable`.
- * webhook, actor-task, codes-run, run-job
+ * webhook a POST of the firing (webhookFor below) to the action's https URL through the SSRF guard
+ * (server/net/post.js), signed with the key its secret_env names exactly as OpenVibe.Events signs a
+ * delivery (openvibe-sdk/events signDeliveryHeaders: X-OpenVibe-Timestamp + X-OpenVibe-Signature-V2,
+ * which a receiver checks with the SDK's verifier). X-OpenVibe-Delivery-Id (`<check run>.<action
+ * index>`) is the same on every attempt, so a receiver can drop a repeat. 2xx is `sent`. 410 and a URL
+ * the guard refuses are `failed` at once. Anything else is retried on the same backoff, then `failed`.
+ * The key must be granted to the watch's owner (WATCH_SECRET_GRANTS, server/secret-grants.js) and
+ * set (32 characters or more), checked when it is queued and again at delivery: otherwise
+ * `unavailable`.
+ * actor-task, codes-run, run-job
* `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and
* `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and
* the watch page says so rather than pretending they ran.
* the watch page says so rather than pretending they ran.
*
*
@@ -29,13 +38,18 @@
@@ -29,13 +38,18 @@
* status() says delivery is off.
* status() says delivery is off.
*/
*/
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+
const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];
const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];
const LEASE_MS = 2 * 60_000;
const LEASE_MS = 2 * 60_000;
const TITLE_MAX = 120;
const TITLE_MAX = 120;
const MESSAGE_MAX = 300;
const MESSAGE_MAX = 300;
-const LATER = new Set(['webhook', 'actor-task', 'codes-run', 'run-job']);
+const LATER = new Set(['actor-task', 'codes-run', 'run-job']);
const TRIGGERED = 'watch.watch.triggered';
const TRIGGERED = 'watch.watch.triggered';
+const MIN_HOOK_SECRET = 32;
+// A URL the guard refuses (server/net/guard.js) will be refused on every attempt: the delivery fails at once.
+const REFUSED_URL = new Set(['bad_url', 'port_refused', 'address_refused']);
const clip = (s, n) => { const t = String(s == null ? '' : s).replace(/\s+/g, ' ').trim(); return t.length > n ? `${t.slice(0, n - 1)}…` : t; };
const clip = (s, n) => { const t = String(s == null ? '' : s).replace(/\s+/g, ' ').trim(); return t.length > n ? `${t.slice(0, n - 1)}…` : t; };
@@ -64,7 +78,24 @@ function notificationFor({ config, watch, run, action, value, previousValue, obs
@@ -64,7 +78,24 @@ function notificationFor({ config, watch, run, action, value, previousValue, obs
};
};
}
}
-function createActions({ db, config, notifications = null, now = () => Date.now(), log = console }) {
+/** The body a webhook action POSTs: the firing, as the watch.watch.triggered event states it. */
+function webhookFor({ config, watch, run, value, previousValue, observedAt }) {
+ return {
+ type: TRIGGERED,
+ watch: { id: watch.id, name: watch.name, url: `${config.baseUrl}/watches/${encodeURIComponent(watch.id)}` },
+ check_run_id: run.id,
+ condition: watch.condition || null,
+ observation: {
+ observed_at: observedAt ? new Date(observedAt).toISOString() : null,
+ value: value === undefined ? null : value,
+ previous_value: previousValue === undefined ? null : previousValue,
+ },
+ };
+}
+
+function createActions({ db, config, notifications = null, post = null, now = () => Date.now(), log = console }) {
+ // Read at each use, not once: what a process was started with is what it enforces, for rows queued before it too.
+ const grants = () => config.secretGrants;
const enabled = Boolean(notifications) && config.actions.enabled;
const enabled = Boolean(notifications) && config.actions.enabled;
let timer = null;
let timer = null;
let busy = null;
let busy = null;
@@ -72,7 +103,7 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
@@ -72,7 +103,7 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
const st = {
const st = {
insert: db.prepare(`INSERT INTO action_runs (watch_id, check_run_id, action_index, kind, owner_sub, request, state, last_error, created_at, finished_at)
insert: db.prepare(`INSERT INTO action_runs (watch_id, check_run_id, action_index, kind, owner_sub, request, state, last_error, created_at, finished_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (check_run_id, action_index) DO NOTHING`),
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (check_run_id, action_index) DO NOTHING`),
- claim: db.prepare(`SELECT rid, watch_id, kind, owner_sub, request, attempts FROM action_runs
+ claim: db.prepare(`SELECT rid, watch_id, check_run_id, action_index, kind, owner_sub, request, attempts FROM action_runs
WHERE state = 'queued' AND next_attempt_at <= ? ORDER BY next_attempt_at, rid LIMIT ? FOR UPDATE SKIP LOCKED`),
WHERE state = 'queued' AND next_attempt_at <= ? ORDER BY next_attempt_at, rid LIMIT ? FOR UPDATE SKIP LOCKED`),
lease: db.prepare('UPDATE action_runs SET attempts = attempts + 1, next_attempt_at = ? WHERE rid = ?'),
lease: db.prepare('UPDATE action_runs SET attempts = attempts + 1, next_attempt_at = ? WHERE rid = ?'),
finish: db.prepare('UPDATE action_runs SET state = ?, last_error = ?, finished_at = ? WHERE rid = ?'),
finish: db.prepare('UPDATE action_runs SET state = ?, last_error = ?, finished_at = ? WHERE rid = ?'),
@@ -98,6 +129,10 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
@@ -98,6 +129,10 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
if (action.event_type !== TRIGGERED) { state = 'unavailable'; error = `watch.action_unavailable: ${action.event_type || 'an event'} is not an event Watch declares (only ${TRIGGERED})`; }
if (action.event_type !== TRIGGERED) { state = 'unavailable'; error = `watch.action_unavailable: ${action.event_type || 'an event'} is not an event Watch declares (only ${TRIGGERED})`; }
else if (!USER_SUBJECT.test(String(owner || ''))) { state = 'skipped'; error = `watch.action_no_person: ${TRIGGERED} names a person, and this watch belongs to an agent or a service`; }
else if (!USER_SUBJECT.test(String(owner || ''))) { state = 'skipped'; error = `watch.action_no_person: ${TRIGGERED} names a person, and this watch belongs to an agent or a service`; }
else { state = 'sent'; error = 'emitted with the check run'; }
else { state = 'sent'; error = 'emitted with the check run'; }
+ } else if (kind === 'webhook') {
+ const why = grants() ? grants().refusal(action.secret_env, owner) : `${action.secret_env} is not granted to any owner`;
+ if (why) { state = 'unavailable'; error = `watch.action_secret_not_granted: ${why}`; }
+ else request = { url: action.url, secret_env: action.secret_env, body: webhookFor({ config, watch, run, value, previousValue, observedAt }) };
} else if (LATER.has(kind)) {
} else if (LATER.has(kind)) {
state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;
state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;
} else {
} else {
@@ -107,8 +142,36 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
@@ -107,8 +142,36 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
}
}
}
}
- async function deliver(row) {
+ async function deliverWebhook(row, request, attempt) {
+ const why = grants() ? grants().refusal(request.secret_env, row.owner_sub) : 'no grants';
+ if (why) return { state: 'unavailable', error: `watch.action_secret_not_granted: ${why}` };
+ const secret = config.secrets ? config.secrets[request.secret_env] : null;
+ if (!secret || String(secret).length < MIN_HOOK_SECRET) {
+ return { state: 'unavailable', error: `watch.action_secret_missing: ${request.secret_env} is not set (${MIN_HOOK_SECRET} characters or more)` };
+ }
+ if (!post) throw Object.assign(new Error('webhook delivery is not configured in this process'), { code: 'webhook_unconfigured' });
+ const body = JSON.stringify(request.body);
+ const headers = {
+ ...signDeliveryHeaders(body, String(secret), { now: now() }),
+ 'X-OpenVibe-Event-Type': TRIGGERED,
+ 'X-OpenVibe-Delivery-Id': `${row.check_run_id}.${row.action_index}`,
+ 'X-OpenVibe-Delivery-Attempt': String(attempt),
+ };
+ let res;
+ try {
+ res = await post(request.url, { headers, body });
+ } catch (err) {
+ if (err && REFUSED_URL.has(err.code)) return { state: 'failed', error: `watch.action_url_refused: ${err.message}` };
+ throw err;
+ }
+ if (res.status >= 200 && res.status < 300) return { state: 'sent', error: null };
+ if (res.status === 410) return { state: 'failed', error: 'watch.action_gone: the receiver answered 410 Gone' };
+ throw Object.assign(new Error(`the receiver answered ${res.status}`), { code: `http_${res.status}` });
+ }
+
+ async function deliver(row, attempt) {
const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
+ if (row.kind === 'webhook') return await deliverWebhook(row, request, attempt);
if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };
if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };
const out = await notifications.push({ subjectId: row.owner_sub, ...request });
const out = await notifications.push({ subjectId: row.owner_sub, ...request });
if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };
if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };
@@ -130,9 +193,11 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
@@ -130,9 +193,11 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
for (const r of rows) {
for (const r of rows) {
const attempt = Number(r.attempts) + 1;
const attempt = Number(r.attempts) + 1;
try {
try {
- const res = await deliver(r);
+ const res = await deliver(r, attempt);
await st.finish.run(res.state, res.error, now(), r.rid);
await st.finish.run(res.state, res.error, now(), r.rid);
- if (res.state === 'sent') out.sent++; else out.skipped++;
+ if (res.state === 'sent') out.sent++;
+ else if (res.state === 'failed') out.failed++;
+ else out.skipped++;
} catch (err) {
} catch (err) {
const why = String((err && (err.code || err.message)) || err).slice(0, 300);
const why = String((err && (err.code || err.message)) || err).slice(0, 300);
if (attempt >= BACKOFF_MS.length + 1) {
if (attempt >= BACKOFF_MS.length + 1) {
@@ -178,4 +243,4 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
@@ -178,4 +243,4 @@ function createActions({ db, config, notifications = null, now = () => Date.now(
return { queue, tick, start, stop, status, recent, enabled };
return { queue, tick, start, stop, status, recent, enabled };
}
}
-module.exports = { createActions, notificationFor, BACKOFF_MS };
+module.exports = { createActions, notificationFor, webhookFor, BACKOFF_MS };
@@ -36,7 +36,7 @@ function createApiCarrier({ fetcher, config }) {
@@ -36,7 +36,7 @@ function createApiCarrier({ fetcher, config }) {
run: async ({ watch, state }) => {
run: async ({ watch, state }) => {
const source = watch.source;
const source = watch.source;
const out = await conditionalGet({
const out = await conditionalGet({
- source, state, fetcher, secrets: config.secrets, accept: ACCEPT.api,
+ source, state, fetcher, secrets: config.secretGrants.viewFor(watch, config.secrets), accept: ACCEPT.api,
onlyChangeNeeded: onlyChange(watch),
onlyChangeNeeded: onlyChange(watch),
});
});
if (out.state !== 'ok' || !out.body) return out;
if (out.state !== 'ok' || !out.body) return out;
@@ -125,7 +125,7 @@ function createFeedCarrier({ fetcher, config }) {
@@ -125,7 +125,7 @@ function createFeedCarrier({ fetcher, config }) {
name: 'feed',
name: 'feed',
run: async ({ watch, state }) => {
run: async ({ watch, state }) => {
const out = await conditionalGet({
const out = await conditionalGet({
- source: watch.source, state, fetcher, secrets: config.secrets, accept: ACCEPT.feed,
+ source: watch.source, state, fetcher, secrets: config.secretGrants.viewFor(watch, config.secrets), accept: ACCEPT.feed,
onlyChangeNeeded: onlyChange(watch),
onlyChangeNeeded: onlyChange(watch),
});
});
if (out.state !== 'ok' || !out.body) return out;
if (out.state !== 'ok' || !out.body) return out;
@@ -151,7 +151,7 @@ function createHttpCarrier({ fetcher, config }) {
@@ -151,7 +151,7 @@ function createHttpCarrier({ fetcher, config }) {
return {
return {
name: 'http',
name: 'http',
run: async ({ watch, state }) => conditionalGet({
run: async ({ watch, state }) => conditionalGet({
- source: watch.source, state, fetcher, secrets: config.secrets, accept: ACCEPT.http,
+ source: watch.source, state, fetcher, secrets: config.secretGrants.viewFor(watch, config.secrets), accept: ACCEPT.http,
onlyChangeNeeded: onlyChange(watch),
onlyChangeNeeded: onlyChange(watch),
}),
}),
};
};
@@ -12,6 +12,7 @@
@@ -12,6 +12,7 @@
* lives; nothing in this release executes it (steps 5-7).
* lives; nothing in this release executes it (steps 5-7).
*/
*/
const pkg = require('../package.json');
const pkg = require('../package.json');
+const { parseGrants, createSecretGrants } = require('./secret-grants');
const int = (v, d) => {
const int = (v, d) => {
const n = parseInt(v, 10);
const n = parseInt(v, 10);
@@ -85,8 +86,14 @@ function load(env = process.env) {
@@ -85,8 +86,14 @@ function load(env = process.env) {
enabled: env.WATCH_ACTIONS !== 'off',
enabled: env.WATCH_ACTIONS !== 'off',
tickMs: int(env.WATCH_ACTIONS_TICK_MS, 5000),
tickMs: int(env.WATCH_ACTIONS_TICK_MS, 5000),
batch: int(env.WATCH_ACTIONS_BATCH, 20),
batch: int(env.WATCH_ACTIONS_BATCH, 20),
+ // A webhook receiver answers within this, or the attempt is retried on the backoff.
+ webhookTimeoutMs: int(env.WATCH_WEBHOOK_TIMEOUT_MS, 10000),
},
},
+ // Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins
+ // it may be sent to) or a WATCH_HOOK_* webhook key (owners). No grant, no use.
+ secretGrants: createSecretGrants({ grants: parseGrants(env.WATCH_SECRET_GRANTS) }),
+
// The check worker: every WATCH_TICK_MS, start checks for active watches whose turn has come.
// The check worker: every WATCH_TICK_MS, start checks for active watches whose turn has come.
worker: {
worker: {
enabled: env.WATCH_WORKER !== 'off',
enabled: env.WATCH_WORKER !== 'off',
@@ -14,6 +14,7 @@ const { openDb } = require('./db');
@@ -14,6 +14,7 @@ const { openDb } = require('./db');
const { createRegistry } = require('./registry');
const { createRegistry } = require('./registry');
const { createObservations } = require('./observations');
const { createObservations } = require('./observations');
const { createGuard } = require('./net/guard');
const { createGuard } = require('./net/guard');
+const { createPoster } = require('./net/post');
const { createFetcher } = require('./net/fetcher');
const { createFetcher } = require('./net/fetcher');
const { createSpacer } = require('./spacer');
const { createSpacer } = require('./spacer');
const { createCarriers } = require('./carriers');
const { createCarriers } = require('./carriers');
@@ -36,7 +37,7 @@ const { createApp } = require('./app');
@@ -36,7 +37,7 @@ const { createApp } = require('./app');
const accountDataLib = require('./account-data');
const accountDataLib = require('./account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
-async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null } = {}) {
+async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null, post: givenPost = null } = {}) {
config = config || load();
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
const db = givenDb || await openDb(config, { log });
const db = givenDb || await openDb(config, { log });
@@ -48,7 +49,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -48,7 +49,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],
'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed'],
validate: (env) => validate('events.event-envelope@1', env),
validate: (env) => validate('events.event-envelope@1', env),
});
});
- const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner });
+ const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants });
const observations = createObservations({ db, now });
const observations = createObservations({ db, now });
// The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte
// The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte
// cap, one deadline, conditional GET; per-host spacing in front of every request.
// cap, one deadline, conditional GET; per-host spacing in front of every request.
@@ -65,7 +66,9 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -65,7 +66,9 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
baseUrls: { network: config.networkInternalUrl }, fetch: fetchImpl,
baseUrls: { network: config.networkInternalUrl }, fetch: fetchImpl,
}))
}))
: null;
: null;
- const actions = createActions({ db, config, notifications: givenNotifications || notifications, now, log });
+ // Webhook actions leave through the same guard as every fetch (server/net/post.js).
+ const post = givenPost || createPoster({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.actions.webhookTimeoutMs });
+ const actions = createActions({ db, config, notifications: givenNotifications || notifications, post, now, log });
const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
new file mode 100644
new file mode 100644
@@ -0,0 +1,46 @@
@@ -0,0 +1,46 @@
+'use strict';
+/**
+ * The one way a webhook action reaches a receiver (plan T18 step 5): POST a JSON body to a URL the watch's owner chose,
+ * through the same SSRF guard as every fetch (server/net/guard.js: http(s) on the allowed ports, no credentials in the
+ * URL, every resolved address public at connect time, so a DNS answer cannot change between check and connect).
+ *
+ * const post = createPoster({ guard, userAgent, timeoutMs });
+ * const { status } = await post(url, { headers, body }); // throws on a refused URL, a timeout or a lost connection
+ *
+ * A redirect is never followed: a 3xx is the receiver's answer, like any other non-2xx. The response body is read only
+ * up to a few KiB and discarded; nothing a receiver says is stored.
+ */
+const http = require('http');
+const https = require('https');
+
+const MAX_RESPONSE_BYTES = 4096;
+
+function createPoster({ guard, userAgent, timeoutMs = 10_000 }) {
+ return async function post(url, { headers = {}, body = '' } = {}) {
+ const u = guard.checkUrl(url); // GuardError: bad_url, port_refused, address_refused
+ const mod = u.protocol === 'https:' ? https : http;
+ const payload = Buffer.from(String(body));
+ return await new Promise((resolve, reject) => {
+ let settled = false;
+ const finish = (fn, v) => { if (!settled) { settled = true; fn(v); } };
+ const req = mod.request(u, {
+ method: 'POST',
+ lookup: guard.lookup,
+ timeout: timeoutMs,
+ headers: { 'Content-Type': 'application/json', 'Content-Length': payload.length, 'User-Agent': userAgent, ...headers },
+ }, (res) => {
+ let seen = 0;
+ res.on('data', (chunk) => { seen += chunk.length; if (seen > MAX_RESPONSE_BYTES) res.destroy(); });
+ const done = () => finish(resolve, { status: res.statusCode });
+ res.on('end', done);
+ res.on('close', done);
+ res.on('error', done);
+ });
+ req.on('timeout', () => req.destroy(Object.assign(new Error(`no answer within ${timeoutMs} ms`), { code: 'ETIMEDOUT' })));
+ req.on('error', (err) => finish(reject, err));
+ req.end(payload);
+ });
+ };
+}
+
+module.exports = { createPoster };
@@ -122,7 +122,14 @@ function validateView(view) {
@@ -122,7 +122,14 @@ function validateView(view) {
return view;
return view;
}
}
-function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50 }) {
+function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null }) {
+ /** A named secret the owner may not use is refused when the watch is saved (server/secret-grants.js). */
+ function checkSecrets(owner, rec) {
+ if (!secretGrants) return;
+ const bad = secretGrants.check(owner, rec);
+ if (bad.length) throw new RegistryError(`${bad[0].path}: ${bad[0].reason}`, 'watch.secret_not_granted');
+ }
+
const st = {
const st = {
get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
get: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
countOwned: db.prepare('SELECT COUNT(*) AS n FROM watches WHERE owner_sub = ? AND deleted_at IS NULL'),
countOwned: db.prepare('SELECT COUNT(*) AS n FROM watches WHERE owner_sub = ? AND deleted_at IS NULL'),
@@ -236,6 +243,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -236,6 +243,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
const req = validateRequest(body, { create: true });
const req = validateRequest(body, { create: true });
const owner = ownerOf(principal);
const owner = ownerOf(principal);
if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');
if (!owner) throw new RegistryError('no acting subject', 'watch.invalid');
+ checkSecrets(owner, req);
// A per-owner cap, so one caller cannot fill the registry (the API and the form both come
// A per-owner cap, so one caller cannot fill the registry (the API and the form both come
// through here). The number is a config value (WATCH_MAX_PER_OWNER).
// through here). The number is a config value (WATCH_MAX_PER_OWNER).
const owned = await st.countOwned.get(owner);
const owned = await st.countOwned.get(owner);
@@ -297,6 +305,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
@@ -297,6 +305,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
// caller that sends one fails the schema check below rather than being silently ignored.
// caller that sends one fails the schema check below rather than being silently ignored.
if (changes.project_id === undefined) delete merged.project_id;
if (changes.project_id === undefined) delete merged.project_id;
validateRequest(merged);
validateRequest(merged);
+ checkSecrets(row.owner_sub, merged);
const sets = [];
const sets = [];
const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };
const args = { id: row.id, updated_at: now(), updated_by: ownerOf(principal) };
for (const [k, v] of Object.entries(changes)) {
for (const [k, v] of Object.entries(changes)) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,97 @@
@@ -0,0 +1,97 @@
+'use strict';
+/**
+ * Who may use a named secret (plan T18). A watch names its secrets by environment variable and Watch reads the value
+ * at use time, never storing it:
+ *
+ * source.auth.env WATCH_CRED_* a source's credential, sent with the watch's own request
+ * action.secret_env WATCH_HOOK_* the key a webhook action signs its delivery with
+ *
+ * The NAME alone must not be a capability. Without a grant, anyone could create a watch that names an operator's
+ * credential, point it at a URL of their own and receive the credential. They could also have their deliveries signed
+ * with another receiver's key and replay them there. So the operator grants each name, in WATCH_SECRET_GRANTS (JSON):
+ *
+ * {"WATCH_CRED_ACME": {"owners": ["usr_…"], "origins": ["https://api.acme.example"]},
+ * "WATCH_HOOK_ALEX": {"owners": ["usr_…"]}}
+ *
+ * A credential is used only by a granted owner and sent only to a granted origin (scheme, host and port, exactly). A
+ * webhook key is used only by a granted owner. A name with no grant is usable by nobody, and that is the default.
+ * The registry checks when a watch is saved (the API and the form say why), and the carriers and the action worker
+ * check again at use, because a grant can be withdrawn. Malformed JSON grants nothing.
+ */
+
+// A person (usr_) or one of their agents (agt_) may own a watch, so either may hold a grant.
+const USER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;
+const NAME = /^WATCH_(CRED|HOOK)_[A-Z0-9_]{2,60}$/;
+
+function originOf(u) {
+ try {
+ const x = new URL(String(u));
+ return x.protocol === 'https:' || x.protocol === 'http:' ? x.origin : null;
+ } catch { return null; }
+}
+
+/** WATCH_SECRET_GRANTS → { name: { owners: Set, origins: Set } }. Anything malformed is dropped (never widened). */
+function parseGrants(raw, log = console) {
+ if (raw == null || String(raw).trim() === '') return {};
+ let v;
+ try { v = JSON.parse(raw); } catch {
+ log.warn('[watch] WATCH_SECRET_GRANTS is not JSON: no watch can use a named secret');
+ return {};
+ }
+ if (!v || typeof v !== 'object' || Array.isArray(v)) {
+ log.warn('[watch] WATCH_SECRET_GRANTS must be an object of grants: no watch can use a named secret');
+ return {};
+ }
+ const out = {};
+ for (const [name, g] of Object.entries(v)) {
+ if (!NAME.test(name) || !g || typeof g !== 'object') continue;
+ const owners = (Array.isArray(g.owners) ? g.owners : []).map(String).filter((o) => USER.test(o));
+ const origins = (Array.isArray(g.origins) ? g.origins : []).map(originOf).filter(Boolean);
+ out[name] = { owners: new Set(owners), origins: new Set(origins) };
+ }
+ return out;
+}
+
+function createSecretGrants({ grants = {} } = {}) {
+ /** null when `owner` may use `name` (a credential: toward `url`), else why not, in one line. */
+ function refusal(name, owner, url = undefined) {
+ const g = grants[name];
+ if (!g) return `${name} is not granted to any owner (WATCH_SECRET_GRANTS)`;
+ if (!g.owners.has(String(owner || ''))) return `${name} is not granted to this owner`;
+ if (url !== undefined) {
+ const o = originOf(url);
+ if (!o || !g.origins.has(o)) return `${name} may not be sent to ${o || 'that URL'}`;
+ }
+ return null;
+ }
+
+ /** The named secrets of a watch (request-shaped) its owner may not use: [{ path, reason }], empty when all are. */
+ function check(owner, rec) {
+ const out = [];
+ const source = rec && rec.source;
+ const auth = source && source.auth;
+ if (auth && auth.mode && auth.mode !== 'none' && auth.env) {
+ const why = refusal(auth.env, owner, source.url || '');
+ if (why) out.push({ path: '/source/auth/env', reason: why });
+ }
+ (Array.isArray(rec && rec.action) ? rec.action : []).forEach((a, i) => {
+ if (a && a.kind === 'webhook' && a.secret_env) {
+ const why = refusal(a.secret_env, owner);
+ if (why) out.push({ path: `/action/${i}/secret_env`, reason: why });
+ }
+ });
+ return out;
+ }
+
+ /** The environment a carrier may read for this watch: its own credential, if granted toward its own origin. */
+ function viewFor(watch, env) {
+ const source = watch && watch.source;
+ const auth = source && source.auth;
+ if (!auth || !auth.env || !env) return {};
+ return refusal(auth.env, watch.owner, source.url || '') ? {} : { [auth.env]: env[auth.env] };
+ }
+
+ return { refusal, check, viewFor };
+}
+
+module.exports = { parseGrants, createSecretGrants, originOf };
@@ -1,11 +1,15 @@
@@ -1,11 +1,15 @@
'use strict';
'use strict';
/**
/**
- * Actions (plan T18 step 4): a condition that fires queues the watch's actions in the run's own transaction, once per
+ * Actions (plan T18 steps 4-5): a condition that fires queues the watch's actions in the run's own transaction, once per
* trigger; the worker delivers a notification to the owner through Network (openvibe-sdk/notifications), retries a
* trigger; the worker delivers a notification to the owner through Network (openvibe-sdk/notifications), retries a
* refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds
* refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds
- * whose dispatchers are later steps as unavailable. The watch page says what happened.
+ * whose dispatchers are later steps as unavailable. A webhook action POSTs the firing signed like an Events delivery,
+ * with a delivery id that is the same on every attempt; a busy receiver is retried, 410 and a refused URL fail at once;
+ * its key must be granted to the owner (refused at save, unavailable at delivery once withdrawn) and long enough. The
+ * watch page says what happened.
*/
*/
const assert = require('assert');
const assert = require('assert');
+const { verifyDeliveryV2 } = require('openvibe-sdk/events');
const { boot, site, suite, newUser, newAgent } = require('./helpers');
const { boot, site, suite, newUser, newAgent } = require('./helpers');
const t = suite('actions');
const t = suite('actions');
@@ -25,6 +29,18 @@ const notifications = {
@@ -25,6 +29,18 @@ const notifications = {
const person = { sub: newUser() };
const person = { sub: newUser() };
const agent = { sub: newAgent() };
const agent = { sub: newAgent() };
+const stranger = { sub: newUser() };
+const HOOK_KEY = 'k'.repeat(24) + 'Watch-hook-key-0123456789';
+// The receiver, as server/net/post.js would reach it: every POST recorded; hookMode decides the answer.
+const posts = [];
+let hookMode = 'ok';
+async function post(url, { headers, body }) {
+ posts.push({ url, headers, body });
+ if (hookMode === 'refused') throw Object.assign(new Error('hooks.example.org resolves to a non-public address'), { code: 'address_refused' });
+ if (hookMode === 'down') throw Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' });
+ return { status: { ok: 204, busy: 503, gone: 410 }[hookMode] };
+}
+const hook = (env = 'WATCH_HOOK_TEST') => ({ kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: env });
const def = (overrides = {}) => ({
const def = (overrides = {}) => ({
name: 'Widget price',
name: 'Widget price',
source: { kind: 'http', url: `${web.origin}/price`, format: null },
source: { kind: 'http', url: `${web.origin}/price`, format: null },
@@ -53,12 +69,20 @@ t('boot', async () => {
@@ -53,12 +69,20 @@ t('boot', async () => {
'/price': () => ({ headers: { 'content-type': 'text/plain' }, body: price }),
'/price': () => ({ headers: { 'content-type': 'text/plain' }, body: price }),
'/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },
'/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },
});
});
- svc = await boot({ notifications });
+ svc = await boot({
+ notifications, post,
+ env: {
+ WATCH_HOOK_TEST: HOOK_KEY, WATCH_HOOK_SHORT: 'too-short', WATCH_HOOK_THEIRS: HOOK_KEY,
+ WATCH_SECRET_GRANTS: JSON.stringify({
+ WATCH_HOOK_TEST: { owners: [person.sub] }, WATCH_HOOK_SHORT: { owners: [person.sub] }, WATCH_HOOK_THEIRS: { owners: [stranger.sub] },
+ }),
+ },
+ });
});
});
t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {
t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {
mode = 'ok'; pushes.length = 0;
mode = 'ok'; pushes.length = 0;
- const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: 'WATCH_HOOK_TEST' }] });
+ const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'actor-task', brief: 'look into it' }] });
await setPrice('19.99');
await setPrice('19.99');
await check(w.id);
await check(w.id);
assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');
assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');
@@ -66,7 +90,7 @@ t('a check that does not fire queues nothing; one that fires queues each action
@@ -66,7 +90,7 @@ t('a check that does not fire queues nothing; one that fires queues each action
const out = await check(w.id);
const out = await check(w.id);
const rows = await acts(w.id);
const rows = await acts(w.id);
assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),
assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),
- [['notification', 'queued', out.run.id, 0], ['webhook', 'unavailable', out.run.id, 1]]);
+ [['notification', 'queued', out.run.id, 0], ['actor-task', 'unavailable', out.run.id, 1]]);
assert.match(rows[1].last_error, /watch\.action_unavailable/);
assert.match(rows[1].last_error, /watch\.action_unavailable/);
const r = await svc.actions.tick();
const r = await svc.actions.tick();
@@ -141,10 +165,99 @@ t('an owner Network does not know is skipped; a muted category is delivered; an
@@ -141,10 +165,99 @@ t('an owner Network does not know is skipped; a muted category is delivered; an
assert.strictEqual(pushes.length, before, 'nothing is pushed for an agent');
assert.strictEqual(pushes.length, before, 'nothing is pushed for an agent');
});
});
+t('a webhook action POSTs the firing, signed as an Events delivery; the receiver verifies it with the SDK', async () => {
+ posts.length = 0; hookMode = 'ok';
+ const w = await create(person, { name: 'hooked', action: [hook()] });
+ const out = await fire(w);
+ let [row] = await acts(w.id);
+ assert.strictEqual(row.state, 'queued');
+ const r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });
+ assert.strictEqual(posts.length, 1);
+ const p = posts[0];
+ assert.strictEqual(p.url, 'https://hooks.example.org/w');
+ const lower = Object.fromEntries(Object.entries(p.headers).map(([k, v]) => [k.toLowerCase(), v]));
+ assert.strictEqual(lower['x-openvibe-event-type'], 'watch.watch.triggered');
+ assert.strictEqual(lower['x-openvibe-delivery-id'], `${out.run.id}.0`);
+ assert.strictEqual(lower['x-openvibe-delivery-attempt'], '1');
+ assert.ok(verifyDeliveryV2(p.body, lower, HOOK_KEY), 'the signature verifies with the key');
+ assert.ok(!verifyDeliveryV2(p.body, lower, 'x'.repeat(48)), 'and with no other key');
+ assert.ok(!p.body.includes(HOOK_KEY), 'the key itself is never sent');
+ const body = JSON.parse(p.body);
+ assert.deepStrictEqual([body.type, body.watch.id, body.watch.name, body.check_run_id], ['watch.watch.triggered', w.id, 'hooked', out.run.id]);
+ assert.strictEqual(body.watch.url, `https://openvibe.watch/watches/${w.id}`);
+ assert.deepStrictEqual([body.observation.value, body.observation.previous_value], ['7.50', '19.99']);
+ [row] = await acts(w.id);
+ assert.deepStrictEqual([row.state, row.last_error], ['sent', null]);
+ const stored = await svc.db.prepare('SELECT request FROM action_runs WHERE rid = ?').get(row.rid);
+ assert.ok(!JSON.stringify(stored).includes(HOOK_KEY), 'the key is never stored');
+});
+
+t('a busy receiver is retried on the backoff with the same delivery id; 410 and a refused URL fail at once', async () => {
+ posts.length = 0; hookMode = 'busy';
+ const w = await create(person, { name: 'busy', action: [hook()] });
+ await fire(w);
+ let r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 });
+ let [row] = await acts(w.id);
+ assert.deepStrictEqual([row.state, row.last_error], ['queued', 'http_503']);
+ hookMode = 'down';
+ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);
+ r = await svc.actions.tick();
+ assert.strictEqual(r.retried, 1, 'a lost connection is retried too');
+ hookMode = 'ok';
+ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);
+ r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });
+ assert.strictEqual(posts.length, 3);
+ const ids = posts.map((x) => x.headers['X-OpenVibe-Delivery-Id']);
+ assert.strictEqual(new Set(ids).size, 1, 'one delivery id across attempts');
+ assert.deepStrictEqual(posts.map((x) => x.headers['X-OpenVibe-Delivery-Attempt']), ['1', '2', '3']);
+
+ for (const [mode, code] of [['gone', /action_gone/], ['refused', /action_url_refused/]]) {
+ hookMode = mode;
+ const g = await create(person, { name: `hook ${mode}`, action: [hook()] });
+ await fire(g);
+ r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 0, failed: 1 }, mode);
+ [row] = await acts(g.id);
+ assert.strictEqual(row.state, 'failed');
+ assert.match(row.last_error, code);
+ }
+ hookMode = 'ok';
+});
+
+t('a webhook key must be granted to the owner (refused at save, unavailable once withdrawn) and long enough', async () => {
+ posts.length = 0; hookMode = 'ok';
+ await assert.rejects(create(person, { name: 'theirs', action: [hook('WATCH_HOOK_THEIRS')] }),
+ (e) => e.code === 'watch.secret_not_granted' && /\/action\/0\/secret_env: WATCH_HOOK_THEIRS is not granted to this owner/.test(e.message));
+ await assert.rejects(create(person, { name: 'nobody', action: [hook('WATCH_HOOK_NOBODY')] }),
+ (e) => e.code === 'watch.secret_not_granted' && /not granted to any owner/.test(e.message));
+ const ok = await create(person, { name: 'patched' });
+ await assert.rejects(svc.registry.patch(ok.id, { action: [hook('WATCH_HOOK_THEIRS')] }, person), (e) => e.code === 'watch.secret_not_granted', 'a patch is checked too');
+
+ const short = await create(person, { name: 'short key', action: [hook('WATCH_HOOK_SHORT')] });
+ await fire(short);
+ let r = await svc.actions.tick();
+ assert.strictEqual(r.skipped, 1);
+ assert.match((await acts(short.id))[0].last_error, /action_secret_missing: WATCH_HOOK_SHORT is not set \(32 characters or more\)/);
+
+ // Granted when it fired, withdrawn before delivery: the worker checks again.
+ const later = await create(person, { name: 'withdrawn', action: [hook()] });
+ await fire(later);
+ const grants = svc.config.secretGrants;
+ const { createSecretGrants } = require('../server/secret-grants');
+ svc.config.secretGrants = createSecretGrants({ grants: {} });
+ try {
+ r = await svc.actions.tick();
+ } finally { svc.config.secretGrants = grants; }
+ assert.strictEqual(posts.length, 0, 'nothing was posted');
+});
+
t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {
t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {
const st = await svc.actions.status();
const st = await svc.actions.status();
assert.strictEqual(st.enabled, true);
assert.strictEqual(st.enabled, true);
- assert.ok(st.sent >= 3 && st.queued === 0 && st.failed === 0, JSON.stringify(st));
+ assert.ok(st.sent >= 3 && st.queued === 0, JSON.stringify(st));
const w = await create(person, { name: 'page check' });
const w = await create(person, { name: 'page check' });
mode = 'ok';
mode = 'ok';
await fire(w);
await fire(w);
@@ -6,14 +6,16 @@
@@ -6,14 +6,16 @@
* address, a private host, a port the guard does not allow.
* address, a private host, a port the guard does not allow.
*/
*/
const assert = require('assert');
const assert = require('assert');
-const { boot, request, site, serviceToken, suite, rss } = require('./helpers');
+const { boot, request, site, serviceToken, suite, rss, newUser } = require('./helpers');
const t = suite('carriers');
const t = suite('carriers');
const OWNER = { sub: 'svc:carrier-test' };
const OWNER = { sub: 'svc:carrier-test' };
+// The person WATCH_CRED_TEST is granted to (WATCH_SECRET_GRANTS), toward the stub site's origin only.
+const PERSON = { sub: newUser() };
let svc;
let svc;
let web;
let web;
-const watch = async (overrides = {}) => (await svc.registry.create({
+const watch = async (overrides = {}, owner = OWNER) => (await svc.registry.create({
name: 'carrier watch',
name: 'carrier watch',
source: { kind: 'http', url: `${web.origin}/page`, format: null },
source: { kind: 'http', url: `${web.origin}/page`, format: null },
cadence: { every_sec: 900 },
cadence: { every_sec: 900 },
@@ -21,7 +23,7 @@ const watch = async (overrides = {}) => (await svc.registry.create({
@@ -21,7 +23,7 @@ const watch = async (overrides = {}) => (await svc.registry.create({
condition: { op: 'eq', value: 'never-matches' },
condition: { op: 'eq', value: 'never-matches' },
action: [{ kind: 'notification' }],
action: [{ kind: 'notification' }],
...overrides,
...overrides,
-}, OWNER)).id;
+}, owner)).id;
const check = async (id, trigger = 'manual') => await svc.check.run(id, { trigger });
const check = async (id, trigger = 'manual') => await svc.check.run(id, { trigger });
const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);
const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);
@@ -44,7 +46,7 @@ t('boot with a stub site', async () => {
@@ -44,7 +46,7 @@ t('boot with a stub site', async () => {
'/api.json': () => ({ headers: { 'content-type': 'application/json' }, body: JSON.stringify({ products: [{ sku: 'A', price: 19.5 }, { sku: 'B', price: 4 }] }) }),
'/api.json': () => ({ headers: { 'content-type': 'application/json' }, body: JSON.stringify({ products: [{ sku: 'A', price: 19.5 }, { sku: 'B', price: 4 }] }) }),
'/secret.json': (req) => ({ headers: { 'content-type': 'application/json' }, body: JSON.stringify({ saw: req.url }) }),
'/secret.json': (req) => ({ headers: { 'content-type': 'application/json' }, body: JSON.stringify({ saw: req.url }) }),
});
});
- svc = await boot();
+ svc = await boot({ env: { WATCH_SECRET_GRANTS: JSON.stringify({ WATCH_CRED_TEST: { owners: [PERSON.sub], origins: [web.origin] } }) } });
});
});
t('a conditional GET stores the validators, and the next check gets 304 → not_modified', async () => {
t('a conditional GET stores the validators, and the next check gets 304 → not_modified', async () => {
@@ -148,7 +150,7 @@ t('a credential is read from the environment by name, at fetch time, and never s
@@ -148,7 +150,7 @@ t('a credential is read from the environment by name, at fetch time, and never s
const id = await watch({
const id = await watch({
source: { kind: 'api', url: `${web.origin}/secret.json`, format: 'json', auth: { mode: 'query', env: 'WATCH_CRED_TEST', param: 'k' } },
source: { kind: 'api', url: `${web.origin}/secret.json`, format: 'json', auth: { mode: 'query', env: 'WATCH_CRED_TEST', param: 'k' } },
extraction: { kind: 'json', value_path: 'saw' },
extraction: { kind: 'json', value_path: 'saw' },
- });
+ }, PERSON);
// not set: the check is recorded disabled and nothing is fetched
// not set: the check is recorded disabled and nothing is fetched
const before = web.hits('/secret.json').length;
const before = web.hits('/secret.json').length;
const missing = await check(id);
const missing = await check(id);
@@ -156,7 +158,8 @@ t('a credential is read from the environment by name, at fetch time, and never s
@@ -156,7 +158,8 @@ t('a credential is read from the environment by name, at fetch time, and never s
assert.strictEqual(missing.run.error_code, 'credential_missing');
assert.strictEqual(missing.run.error_code, 'credential_missing');
assert.strictEqual(web.hits('/secret.json').length, before, 'no request was made');
assert.strictEqual(web.hits('/secret.json').length, before, 'no request was made');
// the stored watch and its view carry the NAME, never a value
// the stored watch and its view carry the NAME, never a value
- const view = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: serviceToken('carrier-test', ['watch.watch.read']) });
+ // Read as a first-party service acting for the owner (X-OV-Subject), the way the public site reads it.
+ const view = await request(svc.base, 'GET', `/api/v1/watches/${id}`, { token: serviceToken('carrier-test', ['watch.watch.read']), headers: { 'X-OV-Subject': PERSON.sub } });
assert.strictEqual(view.status, 200);
assert.strictEqual(view.status, 200);
assert.deepStrictEqual(view.body.watch.source.auth, { mode: 'query', env: 'WATCH_CRED_TEST', param: 'k' });
assert.deepStrictEqual(view.body.watch.source.auth, { mode: 'query', env: 'WATCH_CRED_TEST', param: 'k' });
assert.ok(!view.text.includes('s3cret-value'));
assert.ok(!view.text.includes('s3cret-value'));
@@ -171,6 +174,32 @@ t('a credential is read from the environment by name, at fetch time, and never s
@@ -171,6 +174,32 @@ t('a credential is read from the environment by name, at fetch time, and never s
delete svc.config.secrets.WATCH_CRED_TEST;
delete svc.config.secrets.WATCH_CRED_TEST;
});
});
+t('a credential name is not a capability: only a granted owner may use it, and only toward a granted origin', async () => {
+ const def = (url) => ({ source: { kind: 'api', url, format: 'json', auth: { mode: 'bearer', env: 'WATCH_CRED_TEST' } }, extraction: { kind: 'json', value_path: 'saw' } });
+ // Anyone else naming it (a service, another person) is refused when the watch is saved.
+ await assert.rejects(watch(def(`${web.origin}/secret.json`)), (e) => e.code === 'watch.secret_not_granted' && /\/source\/auth\/env: WATCH_CRED_TEST is not granted to this owner/.test(e.message));
+ await assert.rejects(watch(def(`${web.origin}/secret.json`), { sub: newUser() }), (e) => e.code === 'watch.secret_not_granted');
+ // The granted owner cannot send it anywhere else: the origin is part of the grant.
+ await assert.rejects(watch(def('https://collector.example.org/steal'), PERSON), (e) => e.code === 'watch.secret_not_granted' && /may not be sent to https:\/\/collector\.example\.org/.test(e.message));
+ const elsewhere = await watch({ source: { kind: 'api', url: `${web.origin}/secret.json`, format: 'json' }, extraction: { kind: 'json', value_path: 'saw' } }, PERSON);
+ await assert.rejects(svc.registry.patch(elsewhere, def('https://collector.example.org/steal'), PERSON), (e) => e.code === 'watch.secret_not_granted', 'a patch is checked too');
+ // Withdrawn after the watch was saved: the carrier sees no credential and sends nothing.
+ const id = await watch(def(`${web.origin}/secret.json`), PERSON);
+ svc.config.secrets.WATCH_CRED_TEST = 's3cret-value';
+ const grants = svc.config.secretGrants;
+ const { createSecretGrants } = require('../server/secret-grants');
+ svc.config.secretGrants = createSecretGrants({ grants: {} });
+ const before = web.hits('/secret.json').length;
+ try {
+ const out = await check(id);
+ assert.deepStrictEqual([out.run.state, out.run.error_code], ['disabled', 'credential_missing']);
+ } finally {
+ svc.config.secretGrants = grants;
+ delete svc.config.secrets.WATCH_CRED_TEST;
+ }
+ assert.strictEqual(web.hits('/secret.json').length, before, 'no request was made');
+});
+
t('the feed carrier parses RSS and the extraction reads the latest entry', async () => {
t('the feed carrier parses RSS and the extraction reads the latest entry', async () => {
const id = await watch({
const id = await watch({
source: { kind: 'feed', url: `${web.origin}/feed.xml`, format: 'rss' },
source: { kind: 'feed', url: `${web.origin}/feed.xml`, format: 'rss' },
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
*/
*/
-async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null } = {}) {
+async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null } = {}) {
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
@@ -79,7 +79,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
});
});
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
const testdb = await require('./db').testDb();
const testdb = await require('./db').testDb();
- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(now ? { now } : {}) });
+ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(now ? { now } : {}) });
const base = `http://127.0.0.1:${h.server.address().port}`;
const base = `http://127.0.0.1:${h.server.address().port}`;
/** A session cookie for a person, as /auth/callback would have set it. */
/** A session cookie for a person, as /auth/callback would have set it. */
new file mode 100644
new file mode 100644
@@ -0,0 +1,73 @@
@@ -0,0 +1,73 @@
+'use strict';
+/**
+ * The webhook poster (server/net/post.js): a JSON POST through the SSRF guard. It sends the body and headers it is
+ * given and answers the receiver's status; it never follows a redirect; a URL the guard refuses (an internal IP
+ * literal, a name that resolves to one at connect time, a port off the list) is refused before or at connect; a
+ * receiver that never answers times out.
+ */
+const assert = require('assert');
+const http = require('http');
+const { createGuard } = require('../server/net/guard');
+const { createPoster } = require('../server/net/post');
+const { suite } = require('./helpers');
+
+const t = suite('post');
+let server;
+let base;
+const seen = [];
+
+t('a receiver', async () => {
+ server = http.createServer((req, res) => {
+ const chunks = [];
+ req.on('data', (c) => chunks.push(c));
+ req.on('end', () => {
+ seen.push({ method: req.method, url: req.url, headers: req.headers, body: Buffer.concat(chunks).toString('utf8') });
+ if (req.url === '/redirect') { res.writeHead(302, { location: 'http://127.0.0.1/elsewhere' }); return res.end(); }
+ if (req.url === '/slow') return; // never answers
+ res.writeHead(req.url === '/busy' ? 503 : 204);
+ res.end();
+ });
+ });
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ base = `http://127.0.0.1:${server.address().port}`;
+});
+
+t('it POSTs the body and headers through the guard and answers the status; a redirect is not followed', async () => {
+ const guard = createGuard({ allowPrivateHosts: ['127.0.0.1'] });
+ const post = createPoster({ guard, userAgent: 'OpenVibeWatch/test', timeoutMs: 2000 });
+ const r = await post(`${base}/hook`, { headers: { 'X-OpenVibe-Event-Type': 'watch.watch.triggered' }, body: '{"a":1}' });
+ assert.strictEqual(r.status, 204);
+ const got = seen.pop();
+ assert.deepStrictEqual([got.method, got.url, got.body], ['POST', '/hook', '{"a":1}']);
+ assert.strictEqual(got.headers['content-type'], 'application/json');
+ assert.strictEqual(got.headers['user-agent'], 'OpenVibeWatch/test');
+ assert.strictEqual(got.headers['x-openvibe-event-type'], 'watch.watch.triggered');
+ assert.strictEqual((await post(`${base}/busy`, { body: '{}' })).status, 503);
+ const before = seen.length;
+ assert.strictEqual((await post(`${base}/redirect`, { body: '{}' })).status, 302);
+ assert.strictEqual(seen.length, before + 1, 'the Location was not fetched');
+});
+
+t('internal addresses are refused: an IP literal before anything is sent, a resolved name at connect time', async () => {
+ const strict = createGuard({});
+ const post = createPoster({ guard: strict, userAgent: 'x', timeoutMs: 2000 });
+ const before = seen.length;
+ await assert.rejects(post(`${base}/hook`, { body: '{}' }), (e) => e.code === 'address_refused' || e.code === 'port_refused');
+ await assert.rejects(post('http://169.254.169.254/latest/meta-data/', { body: '{}' }), (e) => e.code === 'address_refused');
+ await assert.rejects(post('https://user:[email protected]/x', { body: '{}' }), (e) => e.code === 'bad_url');
+ // A public-looking name whose DNS answer is internal: refused by the guard's lookup at connect.
+ const rebinding = createGuard({ lookupImpl: (host, opts, cb) => cb(null, [{ address: '127.0.0.1', family: 4 }]) });
+ const viaName = createPoster({ guard: rebinding, userAgent: 'x', timeoutMs: 2000 });
+ await assert.rejects(viaName('https://hooks.example.org/w', { body: '{}' }), (e) => e.code === 'address_refused');
+ assert.strictEqual(seen.length, before, 'nothing reached the receiver');
+});
+
+t('a receiver that never answers times out', async () => {
+ const guard = createGuard({ allowPrivateHosts: ['127.0.0.1'] });
+ const post = createPoster({ guard, userAgent: 'x', timeoutMs: 150 });
+ await assert.rejects(post(`${base}/slow`, { body: '{}' }), (e) => e.code === 'ETIMEDOUT');
+});
+
+t('done', async () => { server.closeAllConnections?.(); await new Promise((r) => server.close(r)); });
+
+t.run();