From d628941efd218d2afb345f2c68590730c21dc8fc Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Sun, 13 Sep 2026 21:54:12 +0700 Subject: [PATCH 1/4] feat(bridge): persist authenticated events for local OpenFn workflows Signed-off-by: Jeremi Joslin --- services/event-bridge/README.md | 111 ++++++++++++++++-- services/event-bridge/package.json | 6 +- services/event-bridge/src/inbox.js | 99 ++++++++++++++++ services/event-bridge/src/server.js | 59 +++++++--- services/event-bridge/src/worker.js | 103 +++++++++++++++++ services/event-bridge/test/bridge.test.js | 32 ++++++ services/event-bridge/test/cli-journey.mjs | 50 +++++++++ services/event-bridge/test/inbox.test.js | 124 +++++++++++++++++++++ 8 files changed, 559 insertions(+), 25 deletions(-) create mode 100644 services/event-bridge/src/inbox.js create mode 100644 services/event-bridge/src/worker.js create mode 100644 services/event-bridge/test/cli-journey.mjs create mode 100644 services/event-bridge/test/inbox.test.js diff --git a/services/event-bridge/README.md b/services/event-bridge/README.md index 1c05a9a..e9d3088 100644 --- a/services/event-bridge/README.md +++ b/services/event-bridge/README.md @@ -1,7 +1,7 @@ # BREG event bridge -This pilot service verifies BREG CloudEvents before submitting a work order to a -fixed OpenFn webhook. It uses Node.js 22.23.2 built-ins and has no package +This service verifies BREG CloudEvents before either committing a local CLI +workflow to a durable inbox or submitting a work order to a fixed OpenFn webhook. It uses Node.js 22.23.2 built-ins and has no package dependencies. Start with `node services/event-bridge/src/server.js` from the repository root. @@ -14,13 +14,14 @@ decoded. Signatures are compared in constant time. Duplicate signed headers, content encoding, wrong source/type/schema, stale delivery times, and invalid payloads are refused before forwarding. -The pilot accepts exactly the common BREG payload members `entity`, `recordId`, +The receiver accepts exactly the common BREG payload members `entity`, `recordId`, `revision`, `trigger`, `packageRevision`, and `values`. Entity, event schema, trigger, and projected field names must match configuration. Record/event ids must be canonical lowercase UUIDs, revision a positive safe integer, package revision a SHA-256 binding, and projected values strings of at most 1024 UTF-8 -bytes. This deliberately implements the pilot's string projection; it is not -a general JSON Schema validator or lifecycle-event receiver. +bytes. Reviewed `request_lifecycle` events also carry closed request transition +metadata. Per-event `entity` and `valueFields` override the common defaults. +This is a bounded string projection receiver, not a general JSON Schema validator. Configuration is required unless a default is listed: @@ -29,10 +30,12 @@ Configuration is required unless a default is listed: | `BREG_HMAC_KEY_FILE` | File containing exact HMAC key bytes, at least 32 bytes. | | `BREG_EXPECTED_SOURCE` | Exact CloudEvent source for the registry instance. | | `BREG_EXPECTED_ENTITY` | Exact payload entity, `farm` for this pilot. | -| `BREG_EXPECTED_EVENTS_FILE` | JSON object mapping each exact type to `{ "schema": "exact-dataschema", "trigger": "created" }`; `patched` is also supported. | +| `BREG_EXPECTED_EVENTS_FILE` | JSON object mapping each exact type to `{ "schema": "exact-dataschema", "trigger": "created" }`; `patched` and `request_lifecycle` are also supported. Optional per-event `entity` and `valueFields` override defaults; CLI mode requires `effect`. | | `BREG_ALLOWED_VALUE_FIELDS_FILE` | JSON array of exact projected string fields, `["local-identifier"]` for this pilot. | -| `OPENFN_WEBHOOK_URL` | Fixed webhook URL, with no credentials, query or fragment. | -| `OPENFN_API_KEY_FILE` | Exact printable API key for `x-api-key`; no trailing newline. | +| `OPENFN_DELIVERY_MODE` | Default `webhook`; use `cli` for durable local execution. | +| `OPENFN_INBOX_PATH` | Required in CLI mode. Persistent SQLite file in a private directory. | +| `OPENFN_WEBHOOK_URL` | Webhook mode only. Fixed URL, with no credentials, query or fragment. | +| `OPENFN_API_KEY_FILE` | Webhook mode only. Exact printable API key for `x-api-key`; no trailing newline. | | `ALLOW_HTTP` | Default false. Set exactly `true` only for an explicitly trusted internal HTTP deployment. HTTPS otherwise required. | | `PORT` | Default 8081. | | `MAX_BODY_BYTES` | Default 65536; maximum 1048576. | @@ -68,13 +71,13 @@ The authenticated OpenFn webhook receives this JSON envelope: } ``` -A 2xx OpenFn response must contain a UUID `work_order_id` and no `error` field. +In webhook mode, a 2xx OpenFn response must contain a UUID `work_order_id` and no `error` field. Lightning 2.18.2 does not return a `run_id` for asynchronous acceptance. Only after this check does the bridge return `202 {"status":"accepted"}` to BREG. This acknowledges work-order acceptance, not workflow completion. Upstream rejection, malformed/generic success responses, redirects, transport failures, or oversized responses produce 502; deadline expiry produces 504. There are no internal -retries, redirect following, or persistent bridge inbox. +retries, redirect following, or persistent bridge inbox in webhook mode. If acceptance occurs but the response is lost, BREG can submit another work order. The workflow and destination must tolerate this, including replay under @@ -86,3 +89,91 @@ administrators because it contains no independent BREG verifier. `GET /healthz` returns a static readiness response. Request payloads, headers, secrets, and upstream error bodies are never logged or reflected in failures. Run focused checks with `npm --prefix services/event-bridge test`. + +## Durable CLI mode + +Set `OPENFN_DELIVERY_MODE=cli` and `OPENFN_INBOX_PATH` in the receiver. Omit +webhook URL/API-key settings. Each accepted event type has one stable logical +`effect`, such as `notify-applicant-v1`. Run one receiver per configured source; +receivers and workers may share a local inbox. The receiver validates HMAC and +payload, commits SQLite with WAL and FULL synchronous mode, and returns 202. +It never waits for a workflow. Full or unavailable storage produces a refusal +so BREG can retry; a stopped worker does not delay registry decisions. + +The worker uses the same `OPENFN_INBOX_PATH` and a reviewed +`OPENFN_WORKFLOWS_FILE` mapping effects to jobs: + +```json +{ + "notify-applicant-v1": { + "job": "./jobs/notify.js", + "adaptor": "@openfn/language-common=/absolute/path/to/installed/common", + "configurationFile": "./private/notification.json" + } +} +``` + +Job and configuration paths resolve relative to this file. Use an absolute +installed adaptor path as shown. `configurationFile` is optional. Pin the CLI +and adaptors at installation; execution disables automatic installation and +adaptor-name expansion. `OPENFN_BINARY` selects the installed CLI executable +(default `openfn`). `OPENFN_JOB_TIMEOUT_MS` defaults to 60000, maximum 1800000. +Use a local filesystem, not a network share, and preserve the inbox directory +across ordinary restarts. Protect configuration files and the whole state +volume; SQLite and per-attempt state files are owner-only. + +```sh +npm --prefix services/event-bridge run worker -- run +npm --prefix services/event-bridge run worker -- status +npm --prefix services/event-bridge run worker -- replay +``` + +`once` executes at most one ready item. Status returns only opaque effect IDs, +state, attempt counts and generic failures. A workflow receives +`state.configuration`, the authenticated event envelope as `state.data`, and +`state.eventEffectId`. The latter is a stable SHA-256 identity over source, +event ID and effect. Retry generation and delivery attempt do not change it. +Reusing an identity with different event content is refused. Destination writes +must use this stable identity or a domain-stable report identity for idempotency. + +The CLI runs outside the receiver, with a bounded lease, deadline and private +temporary working directory. Exit code zero alone does not establish success: +OpenFn CLI 1.40.1 reports workflow failures in `state.errors`. The worker requires +a valid bounded result with no errors before marking the item successful. +It discards stdout/stderr, removes temporary input/output, and drops the logical +event payload after success while retaining the receipt needed for deduplication. +SQLite/WAL backups may retain old pages; protect the full volume as event data. +Tokens, raw Evidence and credentials must never be event projections. + +Failures retry after 1, 2, 4 and 8 seconds, then enter `dead` after five attempts. +An interrupted worker's expired lease also counts toward that bound. Fix the +underlying job, credentials or destination, inspect status, then replay the dead +item by its effect ID. Replay keeps the effect identity. Successful items cannot +be replayed by this command. Preserve receipts for the whole source replay +window; deleting the inbox also deletes destination deduplication history. + +SIGINT/SIGTERM lets the current job finish within its deadline and stops the +worker. Allow the supervisor at least the configured job timeout before force +termination. After a forced kill, stop its remaining CLI child process group before +restarting it, wait for the lease to expire, and remove only its abandoned +`attempt-*` directories. A destination may already have accepted an effect +before the worker records success. Delivery is therefore at least once; email +can be duplicated after such a crash. Destinations that support idempotency +should enforce it independently. Dead-letter recovery cannot undo an email. + +## Verify the local handoff + +Run `npm --prefix services/event-bridge run check` and +`npm --prefix services/event-bridge test`. The explicit CLI journey additionally +requires installed OpenFn CLI 1.40.1 and common adaptor 3.3.4: + +```sh +OPENFN_BINARY=/absolute/path/to/openfn \ +OPENFN_COMMON_PATH=/absolute/path/to/installed/common \ + npm --prefix services/event-bridge run test:cli +``` + +This exercises real CLI success, restart/replay, a zero-exit workflow error, +retry and transient-state cleanup. It installs nothing during execution. +See the maintained [OpenFn CLI usage](https://docs.openfn.org/documentation/cli-usage) +for installation and adaptor arguments. diff --git a/services/event-bridge/package.json b/services/event-bridge/package.json index 2c0c536..bf33f03 100644 --- a/services/event-bridge/package.json +++ b/services/event-bridge/package.json @@ -5,8 +5,10 @@ "type": "module", "scripts": { "start": "node src/server.js", - "check": "node --check src/server.js", - "test": "node --test test/*.test.js" + "worker": "node src/worker.js", + "check": "node --check src/server.js && node --check src/inbox.js && node --check src/worker.js", + "test": "node --test test/*.test.js", + "test:cli": "node test/cli-journey.mjs" }, "engines": { "node": ">=22.23.2" } } diff --git a/services/event-bridge/src/inbox.js b/services/event-bridge/src/inbox.js new file mode 100644 index 0000000..8e9e19b --- /dev/null +++ b/services/event-bridge/src/inbox.js @@ -0,0 +1,99 @@ +import { createHash, randomUUID } from 'node:crypto'; +import { chmodSync, mkdirSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; + +function canonical(value) { + if (Array.isArray(value)) return value.map(canonical); + if (value !== null && typeof value === 'object') { + return Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical(value[key])])); + } + return value; +} +const digest = value => createHash('sha256').update(JSON.stringify(canonical(value))).digest('hex'); + +// The receiver commits before acknowledging. Workers hold no transaction while +// executing a workflow. A crashed worker's lease eventually permits redelivery. +export class DurableInbox { + constructor(path, { maxPending = 10000 } = {}) { + if (!Number.isSafeInteger(maxPending) || maxPending < 1) throw new Error('invalid inbox limit'); + mkdirSync(dirname(resolve(path)), { recursive: true, mode: 0o700 }); + this.db = new DatabaseSync(path); + chmodSync(path, 0o600); + this.maxPending = maxPending; + this.db.exec(`PRAGMA busy_timeout=1000; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; + CREATE TABLE IF NOT EXISTS inbox ( + id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, effect TEXT NOT NULL, + envelope TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, ready_at INTEGER NOT NULL DEFAULT 0, + lease TEXT, lease_until INTEGER, failure TEXT, + CHECK(status IN ('pending','running','succeeded','dead')) + ); + CREATE INDEX IF NOT EXISTS inbox_ready ON inbox(status, ready_at);`); + } + + accept(envelope, effect) { + if (typeof effect !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(effect)) { + throw new Error('invalid workflow effect'); + } + const id = digest([envelope.event.source, envelope.event.id, effect]); + const fingerprint = digest({ event: envelope.event, data: envelope.data }); + this.db.exec('BEGIN IMMEDIATE'); + try { + const prior = this.db.prepare('SELECT fingerprint FROM inbox WHERE id=?').get(id); + if (prior && prior.fingerprint !== fingerprint) throw new Error('event identity conflict'); + if (!prior) { + const { count } = this.db.prepare("SELECT count(*) AS count FROM inbox WHERE status != 'succeeded'").get(); + if (count >= this.maxPending) throw new Error('inbox full'); + this.db.prepare('INSERT INTO inbox(id,fingerprint,effect,envelope) VALUES(?,?,?,?)') + .run(id, fingerprint, effect, JSON.stringify(envelope)); + } + this.db.exec('COMMIT'); + return id; + } catch (error) { this.db.exec('ROLLBACK'); throw error; } + } + + claim({ now = Date.now(), leaseMs = 65000, maxAttempts = 5 } = {}) { + if (!Number.isSafeInteger(leaseMs) || leaseMs < 1000) throw new Error('invalid lease'); + this.db.exec('BEGIN IMMEDIATE'); + try { + this.db.prepare(`UPDATE inbox SET status=CASE WHEN attempts>=? THEN 'dead' ELSE 'pending' END, + lease=NULL, lease_until=NULL, failure='worker_interrupted' + WHERE status='running' AND lease_until<=?`).run(maxAttempts, now); + const row = this.db.prepare("SELECT * FROM inbox WHERE status='pending' AND ready_at<=? ORDER BY ready_at, rowid LIMIT 1").get(now); + if (row) { + row.lease = randomUUID(); + row.attempts += 1; + this.db.prepare("UPDATE inbox SET status='running', attempts=?, lease=?, lease_until=? WHERE id=?") + .run(row.attempts, row.lease, now + leaseMs, row.id); + } + this.db.exec('COMMIT'); + return row ? { id: row.id, effect: row.effect, envelope: JSON.parse(row.envelope), + attempts: row.attempts, lease: row.lease } : undefined; + } catch (error) { this.db.exec('ROLLBACK'); throw error; } + } + + complete(item) { + return this.db.prepare(`UPDATE inbox SET status='succeeded', envelope='{}', lease=NULL, lease_until=NULL, failure=NULL + WHERE id=? AND status='running' AND lease=?`).run(item.id, item.lease).changes === 1; + } + + fail(item, { now = Date.now(), maxAttempts = 5, retryMs = 1000 } = {}) { + const status = item.attempts >= maxAttempts ? 'dead' : 'pending'; + return this.db.prepare(`UPDATE inbox SET status=?, ready_at=?, lease=NULL, lease_until=NULL, failure='workflow_failed' + WHERE id=? AND status='running' AND lease=?`) + .run(status, now + retryMs * 2 ** Math.min(item.attempts - 1, 10), item.id, item.lease).changes === 1; + } + + replay(id) { + return this.db.prepare("UPDATE inbox SET status='pending', attempts=0, ready_at=0, failure=NULL WHERE id=? AND status='dead'") + .run(id).changes === 1; + } + + // Operator status deliberately omits projected data and credential material. + status() { + return this.db.prepare('SELECT id,effect,status,attempts,ready_at,lease_until,failure FROM inbox ORDER BY rowid').all(); + } + + close() { this.db.close(); } +} diff --git a/services/event-bridge/src/server.js b/services/event-bridge/src/server.js index 7617b10..f274a6f 100644 --- a/services/event-bridge/src/server.js +++ b/services/event-bridge/src/server.js @@ -3,6 +3,7 @@ import { readFileSync } from 'node:fs'; import http from 'node:http'; import https from 'node:https'; import { pathToFileURL } from 'node:url'; +import { DurableInbox } from './inbox.js'; export const EVENT_PATH = '/events/breg'; const SIGNED_HEADERS = [ @@ -27,24 +28,33 @@ function positive(value, fallback, maximum) { export function loadConfig(env = process.env) { try { const hmacKey = readFileSync(env.BREG_HMAC_KEY_FILE); - const apiKey = readFileSync(env.OPENFN_API_KEY_FILE, 'utf8'); + const deliveryMode = env.OPENFN_DELIVERY_MODE ?? 'webhook'; + if (!['webhook', 'cli'].includes(deliveryMode)) throw new Error('invalid delivery mode'); + const apiKey = deliveryMode === 'webhook' ? readFileSync(env.OPENFN_API_KEY_FILE, 'utf8') : undefined; const expectedEvents = JSON.parse(readFileSync(env.BREG_EXPECTED_EVENTS_FILE, 'utf8')); const allowedValueFields = JSON.parse(readFileSync(env.BREG_ALLOWED_VALUE_FIELDS_FILE, 'utf8')); - const url = new URL(env.OPENFN_WEBHOOK_URL); - if (hmacKey.length < 32 || !/^[\x21-\x7e]+$/.test(apiKey) || + const url = deliveryMode === 'webhook' ? new URL(env.OPENFN_WEBHOOK_URL) : undefined; + if (hmacKey.length < 32 || (apiKey !== undefined && !/^[\x21-\x7e]+$/.test(apiKey)) || !env.BREG_EXPECTED_SOURCE || !env.BREG_EXPECTED_ENTITY || !object(expectedEvents) || Object.keys(expectedEvents).length === 0 || Object.entries(expectedEvents).some(([type, binding]) => !type || !object(binding) || - typeof binding.schema !== 'string' || !binding.schema || !['created', 'patched'].includes(binding.trigger)) || + typeof binding.schema !== 'string' || !binding.schema || !['created', 'patched', 'request_lifecycle'].includes(binding.trigger) || + (binding.entity !== undefined && (typeof binding.entity !== 'string' || !binding.entity)) || + (binding.valueFields !== undefined && (!Array.isArray(binding.valueFields) || + binding.valueFields.some(field => typeof field !== 'string' || !field) || + new Set(binding.valueFields).size !== binding.valueFields.length)) || + (deliveryMode === 'cli' && (typeof binding.effect !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(binding.effect)))) || !Array.isArray(allowedValueFields) || allowedValueFields.length === 0 || allowedValueFields.some(field => typeof field !== 'string' || !field) || new Set(allowedValueFields).size !== allowedValueFields.length || - url.username || url.password || url.hash || url.search || - (url.protocol !== 'https:' && !(url.protocol === 'http:' && env.ALLOW_HTTP === 'true'))) { + (deliveryMode === 'cli' && (!env.OPENFN_INBOX_PATH || env.OPENFN_WEBHOOK_URL || env.OPENFN_API_KEY_FILE)) || + (url && (url.username || url.password || url.hash || url.search || + (url.protocol !== 'https:' && !(url.protocol === 'http:' && env.ALLOW_HTTP === 'true'))))) { throw new Error('invalid configuration'); } return { - hmacKey, apiKey, expectedEvents, allowedValueFields, + hmacKey, apiKey, expectedEvents, allowedValueFields, deliveryMode, + inboxPath: env.OPENFN_INBOX_PATH, expectedSource: env.BREG_EXPECTED_SOURCE, expectedEntity: env.BREG_EXPECTED_ENTITY, openfnUrl: url, port: positive(env.PORT, 8081, 65535), maxBodyBytes: positive(env.MAX_BODY_BYTES, 65536, 1048576), @@ -123,21 +133,40 @@ function verifyPayload(body, config, binding) { let data; try { data = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(body)); } catch { throw new Refusal(400, 'invalid_event'); } + const fields = binding.valueFields ?? config.allowedValueFields; const keys = ['entity', 'recordId', 'revision', 'trigger', 'packageRevision', 'values']; + if (binding.trigger === 'request_lifecycle') keys.push('request'); if (!object(data) || Object.keys(data).length !== keys.length || keys.some(key => !Object.hasOwn(data, key)) || - data.entity !== config.expectedEntity || typeof data.recordId !== 'string' || !UUID.test(data.recordId) || + data.entity !== (binding.entity ?? config.expectedEntity) || typeof data.recordId !== 'string' || !UUID.test(data.recordId) || !Number.isSafeInteger(data.revision) || data.revision < 1 || typeof data.packageRevision !== 'string' || !DIGEST.test(data.packageRevision) || - !['created', 'patched'].includes(data.trigger) || + !['created', 'patched', 'request_lifecycle'].includes(data.trigger) || data.trigger !== binding.trigger || !object(data.values) || - Object.keys(data.values).length !== config.allowedValueFields.length || - config.allowedValueFields.some(field => !Object.hasOwn(data.values, field) || + Object.keys(data.values).length !== fields.length || + fields.some(field => !Object.hasOwn(data.values, field) || typeof data.values[field] !== 'string' || Buffer.byteLength(data.values[field]) > 1024)) { throw new Refusal(400, 'invalid_event'); } + if (binding.trigger === 'request_lifecycle') verifyLifecycle(data.request); return data; } +function verifyLifecycle(value) { + const keys = ['proposalVersion', 'workflowRevision', 'transition', 'fromState', 'toState', + 'stage', 'reasonPresent', 'effectDigest', 'deduplicationKey']; + const text = v => typeof v === 'string' && v.length > 0 && Buffer.byteLength(v) <= 128; + if (!object(value) || keys.some(key => !Object.hasOwn(value, key)) || + Object.keys(value).some(key => !keys.includes(key) && key !== 'reason') || + !['proposalVersion', 'workflowRevision'].every(key => Number.isSafeInteger(value[key]) && value[key] >= 1) || + !['transition', 'fromState', 'toState'].every(key => text(value[key])) || + !(value.stage === null || text(value.stage)) || typeof value.reasonPresent !== 'boolean' || + !(value.effectDigest === null || (typeof value.effectDigest === 'string' && DIGEST.test(value.effectDigest))) || + typeof value.deduplicationKey !== 'string' || !DIGEST.test(value.deduplicationKey) || + (Object.hasOwn(value, 'reason') && (typeof value.reason !== 'string' || Buffer.byteLength(value.reason) > 8192))) { + throw new Refusal(400, 'invalid_event'); + } +} + function forward(config, envelope) { return new Promise((resolve, reject) => { const bytes = Buffer.from(JSON.stringify(envelope)); @@ -180,6 +209,7 @@ function respond(response, status, code) { } export function createBridge(config) { + const inbox = config.deliveryMode === 'cli' ? new DurableInbox(config.inboxPath) : undefined; const server = http.createServer({ maxHeaderSize: 16384 }, async (request, response) => { try { if (request.url === '/healthz' && request.method === 'GET') return respond(response, 200, 'ok'); @@ -192,15 +222,18 @@ export function createBridge(config) { !timingSafeEqual(Buffer.from(provided), Buffer.from(expected))) throw new Refusal(401, 'invalid_signature'); const data = verifyPayload(body, config, binding); const h = request.headers; - await forward(config, { + const envelope = { event: Object.fromEntries(['specversion', 'id', 'source', 'type', 'time', 'dataschema'].map(key => [key, h[`ce-${key}`]])), delivery: { generation: Number(h['x-registry-event-generation']), attempt: Number(h['x-registry-delivery-attempt']), time: h['x-registry-delivery-time'], idempotencyKey: h['idempotency-key'] }, data, - }); + }; + if (inbox) inbox.accept(envelope, binding.effect); + else await forward(config, envelope); respond(response, 202, 'accepted'); } catch (error) { respond(response, error instanceof Refusal ? error.status : 503, error instanceof Refusal ? error.message : 'unavailable'); } }); + if (inbox) server.on('close', () => inbox.close()); server.requestTimeout = 10000; server.headersTimeout = 5000; return server; diff --git a/services/event-bridge/src/worker.js b/services/event-bridge/src/worker.js new file mode 100644 index 0000000..ebde161 --- /dev/null +++ b/services/event-bridge/src/worker.js @@ -0,0 +1,103 @@ +import { spawn } from 'node:child_process'; +import { readFileSync } from 'node:fs'; +import { mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'; +import { dirname, join, resolve } from 'node:path'; +import { pathToFileURL } from 'node:url'; +import { DurableInbox } from './inbox.js'; + +export function loadWorkerConfig(env = process.env) { + try { + const workflowsPath = resolve(env.OPENFN_WORKFLOWS_FILE); + const workflows = JSON.parse(readFileSync(workflowsPath, 'utf8')); + if (!workflows || Array.isArray(workflows) || typeof workflows !== 'object' || + !Object.keys(workflows).length || !env.OPENFN_INBOX_PATH) throw new Error(); + for (const [effect, binding] of Object.entries(workflows)) { + if (!/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(effect) || !binding || + typeof binding.job !== 'string' || !binding.job || + typeof binding.adaptor !== 'string' || !binding.adaptor || + (binding.configurationFile !== undefined && typeof binding.configurationFile !== 'string') || + Object.keys(binding).some(key => !['job', 'adaptor', 'configurationFile'].includes(key))) throw new Error(); + binding.job = resolve(dirname(workflowsPath), binding.job); + if (binding.configurationFile) binding.configurationFile = resolve(dirname(workflowsPath), binding.configurationFile); + } + const timeoutMs = Number(env.OPENFN_JOB_TIMEOUT_MS ?? 60000); + if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 100 || timeoutMs > 1800000) throw new Error(); + return { inboxPath: resolve(env.OPENFN_INBOX_PATH), workflows, + binary: env.OPENFN_BINARY ?? 'openfn', timeoutMs }; + } catch { throw new Error('invalid worker configuration'); } +} + +export async function runCli(item, config) { + const binding = config.workflows[item.effect]; + if (!binding) throw new Error('workflow unavailable'); + const configuration = binding.configurationFile + ? JSON.parse(readFileSync(binding.configurationFile, 'utf8')) : {}; + const directory = await mkdtemp(join(dirname(config.inboxPath), 'attempt-')); + try { + const state = join(directory, 'state.json'); + const output = join(directory, 'output.json'); + await writeFile(state, JSON.stringify({ configuration, data: item.envelope, + eventEffectId: item.id }), { mode: 0o600 }); + await new Promise((resolveRun, rejectRun) => { + // Reviewed executable/job/adaptor paths only. Event data never becomes + // shell syntax, CLI arguments, a job name, or log output. + const child = spawn(config.binary, [binding.job, '-a', binding.adaptor, + '-s', state, '-o', output, '--no-cache-steps', + '--no-autoinstall', '--no-expand-adaptors', '--repo-dir', join(directory, 'repo'), + '--timeout', String(config.timeoutMs)], + { cwd: directory, shell: false, detached: process.platform !== 'win32', stdio: 'ignore' }); + const timer = setTimeout(() => { + try { + if (process.platform === 'win32') child.kill('SIGKILL'); + else process.kill(-child.pid, 'SIGKILL'); + } catch { /* Child may have exited before the deadline callback. */ } + }, config.timeoutMs); + child.once('error', () => { clearTimeout(timer); rejectRun(new Error('workflow unavailable')); }); + child.once('close', code => { + clearTimeout(timer); + code === 0 ? resolveRun() : rejectRun(new Error('workflow failed')); + }); + }); + // CLI 1.40.1 exits zero for workflow errors and records them in state.errors. + // A missing, oversized, invalid or failed result must remain retryable. + if ((await stat(output)).size > 16 * 1024 * 1024) throw new Error('workflow failed'); + const result = JSON.parse(await readFile(output, 'utf8')); + if (!result || typeof result !== 'object' || Array.isArray(result) || + (Object.hasOwn(result, 'errors') && (!result.errors || typeof result.errors !== 'object' || + Array.isArray(result.errors) || Object.keys(result.errors).length > 0))) { + throw new Error('workflow failed'); + } + } finally { await rm(directory, { recursive: true, force: true }); } +} + +export async function runOnce(inbox, config, execute = runCli) { + const item = inbox.claim({ leaseMs: config.timeoutMs + 5000 }); + if (!item) return false; + try { + await execute(item, config); + inbox.complete(item); + } catch { inbox.fail(item); } + return true; +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + let inbox; + try { + const config = loadWorkerConfig(); + inbox = new DurableInbox(config.inboxPath); + const [command = 'run', id] = process.argv.slice(2); + if (command === 'status') process.stdout.write(`${JSON.stringify(inbox.status())}\n`); + else if (command === 'replay' && /^[0-9a-f]{64}$/.test(id ?? '')) { + if (!inbox.replay(id)) throw new Error('replay unavailable'); + process.stdout.write('queued\n'); + } else if (command === 'once') await runOnce(inbox, config); + else if (command === 'run') { + let stopping = false; + for (const signal of ['SIGINT', 'SIGTERM']) process.on(signal, () => { stopping = true; }); + while (!stopping) { + if (!await runOnce(inbox, config)) await new Promise(resolvePoll => setTimeout(resolvePoll, 500)); + } + } else throw new Error('invalid worker command'); + } catch { process.stderr.write('worker unavailable\n'); process.exitCode = 1; } + finally { inbox?.close(); } +} diff --git a/services/event-bridge/test/bridge.test.js b/services/event-bridge/test/bridge.test.js index b50c1b2..f2c769b 100644 --- a/services/event-bridge/test/bridge.test.js +++ b/services/event-bridge/test/bridge.test.js @@ -6,6 +6,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test } from 'node:test'; import { createBridge, loadConfig, EVENT_PATH, webhookSignature } from '../src/server.js'; +import { DurableInbox } from '../src/inbox.js'; const key = Buffer.from('a-synthetic-hmac-key-at-least-32-bytes'); const source = 'urn:registrystack:registry:agricultural-holdings:instance:pilot'; @@ -187,3 +188,34 @@ test('configuration uses exact secret bytes, explicit HTTP trust and safe errors assert.equal(loadConfig({ ...env, ALLOW_HTTP: 'true' }).hmacKey.at(-1), 10); assert.throws(() => loadConfig({ ...env, OPENFN_WEBHOOK_URL: 'https://secret:canary@example.org/' }), /^Error: invalid bridge configuration$/); }); + + +test('CLI delivery authenticates before durable acceptance and handles lifecycle events without upstream work', async t => { + const directory = await mkdtemp(join(tmpdir(), 'bridge-durable-')); + const path = join(directory, 'inbox.sqlite'); + const { url, calls } = await fixture(t, () => { throw new Error('CLI mode must not forward'); }, { + deliveryMode: 'cli', inboxPath: path, + expectedEvents: { 'farm-created-v1': { schema, trigger: 'request_lifecycle', effect: 'notify', + entity: 'correction-request', valueFields: ['record-reference'] } }, + }); + t.after(() => rm(directory, { recursive: true, force: true })); + const payload = { ...data, entity: 'correction-request', trigger: 'request_lifecycle', + values: { 'record-reference': 'synthetic-reference' }, + request: { proposalVersion: 1, workflowRevision: 2, transition: 'request_revision', + fromState: 'submitted', toState: 'needs_changes', stage: 'review', reasonPresent: false, + effectDigest: `sha256:${'d'.repeat(64)}`, deduplicationKey: `sha256:${'e'.repeat(64)}` } }; + const invalid = delivery({}, payload); + invalid.headers['x-registry-signature'] = `v1=${'x'.repeat(43)}`; + assert.equal((await fetch(`${url}${EVENT_PATH}`, { method: 'POST', ...invalid })).status, 401); + const persisted = new DurableInbox(path); + try { + assert.equal(persisted.status().length, 0); + for (const generation of ['1', '2']) { + const request = delivery({ 'x-registry-event-generation': generation }, payload); + assert.equal((await fetch(`${url}${EVENT_PATH}`, { method: 'POST', ...request })).status, 202); + assert.equal(persisted.status().length, 1, '202 means durable acceptance, even before worker starts'); + } + assert.equal(calls.length, 0); + assert.equal(persisted.claim().envelope.data.request.toState, 'needs_changes'); + } finally { persisted.close(); } +}); diff --git a/services/event-bridge/test/cli-journey.mjs b/services/event-bridge/test/cli-journey.mjs new file mode 100644 index 0000000..d7819fd --- /dev/null +++ b/services/event-bridge/test/cli-journey.mjs @@ -0,0 +1,50 @@ +// Explicit integration check against an installed CLI and exact local adaptor. +// No network installation or hosted OpenFn instance is used during execution. +import assert from 'node:assert/strict'; +import { mkdtemp, writeFile, rm, readdir } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { DurableInbox } from '../src/inbox.js'; +import { runOnce } from '../src/worker.js'; + +if (!process.env.OPENFN_BINARY || !process.env.OPENFN_COMMON_PATH) { + throw new Error('Set OPENFN_BINARY and OPENFN_COMMON_PATH to installed CLI and common adaptor paths'); +} +const directory = await mkdtemp(join(tmpdir(), 'registry-real-cli-')); +const path = join(directory, 'inbox.sqlite'); +let inbox = new DurableInbox(path); +try { + const job = join(directory, 'job.js'); + const success = `fn(state => { + if (state.data.data.values.reference !== 'SYNTHETIC-ONLY' || !state.eventEffectId) { + throw new Error('invalid handoff'); + } + return state; + });`; + await writeFile(job, success); + const envelope = { event: { source: 'urn:synthetic:cli-proof', id: '1' }, + data: { values: { reference: 'SYNTHETIC-ONLY' } } }; + const config = { inboxPath: path, binary: resolve(process.env.OPENFN_BINARY), timeoutMs: 20000, + workflows: { notify: { job, + adaptor: `@openfn/language-common=${resolve(process.env.OPENFN_COMMON_PATH)}` } } }; + const id = inbox.accept(envelope, 'notify'); + await runOnce(inbox, config); + assert.equal(inbox.status()[0].status, 'succeeded'); + inbox.close(); + inbox = new DurableInbox(path); + assert.equal(inbox.accept(envelope, 'notify'), id); + assert.equal(await runOnce(inbox, config), false); + await writeFile(job, `fn(() => { throw new Error('SYNTHETIC_EXPECTED_FAILURE'); });`); + inbox.accept({ ...envelope, event: { ...envelope.event, id: '2' } }, 'notify'); + await runOnce(inbox, config); + assert.equal(inbox.status()[1].status, 'pending'); + await writeFile(job, success); + await new Promise(resolveRetry => setTimeout(resolveRetry, 1100)); + await runOnce(inbox, config); + assert.equal(inbox.status()[1].status, 'succeeded'); + assert.equal((await readdir(directory)).some(file => file.startsWith('attempt-')), false); + console.log('PASS: real CLI success, restart/replay, zero-exit workflow failure, retry and transient cleanup'); +} finally { + inbox.close(); + await rm(directory, { recursive: true, force: true }); +} diff --git a/services/event-bridge/test/inbox.test.js b/services/event-bridge/test/inbox.test.js new file mode 100644 index 0000000..810e711 --- /dev/null +++ b/services/event-bridge/test/inbox.test.js @@ -0,0 +1,124 @@ +import assert from 'node:assert/strict'; +import { mkdtemp, rm, writeFile, chmod, readdir, stat } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { DurableInbox } from '../src/inbox.js'; +import { runOnce, runCli } from '../src/worker.js'; + +const envelope = { event: { source: 'urn:synthetic:registry', id: 'event-1' }, + delivery: { generation: 1 }, data: { values: { reference: 'synthetic-record' } } }; + +async function fixture(t, options) { + const directory = await mkdtemp(join(tmpdir(), 'registry-inbox-')); + const path = join(directory, 'inbox.sqlite'); + const inbox = new DurableInbox(path, options); + t.after(async () => { inbox.close(); await rm(directory, { recursive: true, force: true }); }); + return { inbox, path, directory }; +} + +test('commit survives restart; new delivery generation retains one effect and conflicting payload is refused', async t => { + const { inbox, path } = await fixture(t); + assert.equal((await stat(path)).mode & 0o777, 0o600); + const id = inbox.accept(envelope, 'notify'); + const restarted = new DurableInbox(path); + try { + assert.equal(restarted.accept({ ...envelope, delivery: { generation: 2 } }, 'notify'), id); + assert.throws(() => restarted.accept({ ...envelope, data: { values: {} } }, 'notify'), /identity conflict/); + assert.equal(restarted.status().length, 1); + const item = restarted.claim(); + assert.deepEqual(item.envelope, envelope); + assert.equal(restarted.complete(item), true); + assert.equal(restarted.accept(envelope, 'notify'), id); + assert.equal(restarted.claim(), undefined); + assert.notEqual(restarted.accept(envelope, 'intake'), id); + } finally { restarted.close(); } +}); + +test('concurrent workers cannot claim one live lease; stale completion cannot overwrite recovery', async t => { + const { inbox, path } = await fixture(t); + inbox.accept(envelope, 'notify'); + const other = new DurableInbox(path); + try { + const first = inbox.claim({ now: 0, leaseMs: 1000 }); + assert.equal(other.claim({ now: 999 }), undefined); + const recovered = other.claim({ now: 1000 }); + assert.equal(recovered.id, first.id); + assert.equal(recovered.attempts, 2); + assert.notEqual(recovered.lease, first.lease); + assert.equal(inbox.complete(first), false); + assert.equal(other.complete(recovered), true); + } finally { other.close(); } +}); + +test('bounded retries reach dead-letter and explicit replay preserves effect identity', async t => { + const { inbox } = await fixture(t); + const id = inbox.accept(envelope, 'notify'); + const first = inbox.claim({ now: 0 }); + inbox.fail(first, { now: 0, maxAttempts: 2, retryMs: 100 }); + assert.equal(inbox.claim({ now: 99 }), undefined); + const second = inbox.claim({ now: 100 }); + inbox.fail(second, { now: 100, maxAttempts: 2 }); + assert.equal(inbox.status()[0].status, 'dead'); + assert.equal(inbox.claim({ now: 100000 }), undefined); + assert.equal(inbox.replay(id), true); + assert.equal(inbox.replay(id), false); + const replay = inbox.claim({ now: 100001 }); + assert.equal(replay.id, id); + assert.equal(replay.attempts, 1); +}); + +test('repeated worker crashes also exhaust bounded attempts', async t => { + const { inbox } = await fixture(t); + inbox.accept(envelope, 'notify'); + assert.ok(inbox.claim({ now: 0, leaseMs: 1000, maxAttempts: 2 })); + assert.ok(inbox.claim({ now: 1000, leaseMs: 1000, maxAttempts: 2 })); + assert.equal(inbox.claim({ now: 2000, maxAttempts: 2 }), undefined); + assert.equal(inbox.status()[0].status, 'dead'); +}); + +test('capacity failure does not persist or evict an accepted pending event', async t => { + const { inbox } = await fixture(t, { maxPending: 1 }); + inbox.accept(envelope, 'notify'); + assert.throws(() => inbox.accept(envelope, 'intake'), /inbox full/); + assert.equal(inbox.status().length, 1); + assert.ok(inbox.accept(envelope, 'notify')); +}); + +test('runner stores only redacted failure state and permits independent workflow recovery', async t => { + const { inbox } = await fixture(t); + inbox.accept(envelope, 'notify'); + const config = { timeoutMs: 1000 }; + await runOnce(inbox, config, async () => { throw new Error('SYNTHETIC-SECRET-CANARY'); }); + assert.equal(inbox.status()[0].status, 'pending'); + assert.equal(JSON.stringify(inbox.status()).includes('SYNTHETIC-SECRET-CANARY'), false); + inbox.accept(envelope, 'intake'); + await runOnce(inbox, config, async item => { assert.equal(item.effect, 'intake'); }); + assert.equal(inbox.status()[1].status, 'succeeded'); +}); + +test('CLI handoff uses protected state files, bounds execution and removes transient state', async t => { + const { inbox, path, directory } = await fixture(t); + inbox.accept(envelope, 'notify'); + const executable = join(directory, 'synthetic-cli'); + await writeFile(executable, `#!${process.execPath}\nconst fs=require('node:fs'); + const args=process.argv.slice(2); const state=JSON.parse(fs.readFileSync(args[args.indexOf('-s')+1])); + if(state.data.event.id!=='event-1'||state.configuration.key!=='synthetic-secret'||!state.eventEffectId)process.exit(1); + fs.writeFileSync(args[args.indexOf('-o')+1], JSON.stringify(state)); + process.stdout.write('SYNTHETIC-PRIVATE-OUTPUT'); process.stderr.write('SYNTHETIC-PRIVATE-ERROR'); + `); + await chmod(executable, 0o700); + const credentials = join(directory, 'credentials.json'); + await writeFile(credentials, JSON.stringify({ key: 'synthetic-secret' }), { mode: 0o600 }); + const config = { inboxPath: path, binary: executable, timeoutMs: 1000, + workflows: { notify: { job: 'synthetic.js', adaptor: 'common@1.0.0', configurationFile: credentials } } }; + await runOnce(inbox, config); + assert.equal(inbox.status()[0].status, 'succeeded'); + assert.equal((await readdir(directory)).some(file => file.startsWith('attempt-')), false); + await writeFile(executable, `#!${process.execPath}\nconst fs=require('node:fs'); const args=process.argv.slice(2); + fs.writeFileSync(args[args.indexOf('-o')+1],JSON.stringify({errors:{'job-1':{message:'SYNTHETIC-PRIVATE'}}}));`); + await assert.rejects(runCli({ effect: 'notify', id: 'synthetic', envelope }, config), /workflow failed/); + await writeFile(executable, `#!${process.execPath}\nsetInterval(()=>{},1000);`); + await assert.rejects(runCli({ effect: 'notify', id: 'synthetic', envelope }, { ...config, timeoutMs: 100 })); + assert.equal((await readdir(directory)).some(file => file.startsWith('attempt-')), false); +}); From aec7739ea1f3c3403dc4e40a2a3e97d60bec2c60 Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Sun, 13 Sep 2026 21:59:20 +0700 Subject: [PATCH 2/4] fix(bridge): accept events without projected fields Signed-off-by: Jeremi Joslin --- services/event-bridge/src/server.js | 2 +- services/event-bridge/test/bridge.test.js | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/services/event-bridge/src/server.js b/services/event-bridge/src/server.js index f274a6f..e034fab 100644 --- a/services/event-bridge/src/server.js +++ b/services/event-bridge/src/server.js @@ -44,7 +44,7 @@ export function loadConfig(env = process.env) { binding.valueFields.some(field => typeof field !== 'string' || !field) || new Set(binding.valueFields).size !== binding.valueFields.length)) || (deliveryMode === 'cli' && (typeof binding.effect !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(binding.effect)))) || - !Array.isArray(allowedValueFields) || allowedValueFields.length === 0 || + !Array.isArray(allowedValueFields) || allowedValueFields.some(field => typeof field !== 'string' || !field) || new Set(allowedValueFields).size !== allowedValueFields.length || (deliveryMode === 'cli' && (!env.OPENFN_INBOX_PATH || env.OPENFN_WEBHOOK_URL || env.OPENFN_API_KEY_FILE)) || diff --git a/services/event-bridge/test/bridge.test.js b/services/event-bridge/test/bridge.test.js index f2c769b..da78899 100644 --- a/services/event-bridge/test/bridge.test.js +++ b/services/event-bridge/test/bridge.test.js @@ -187,6 +187,8 @@ test('configuration uses exact secret bytes, explicit HTTP trust and safe errors assert.throws(() => loadConfig(env), /^Error: invalid bridge configuration$/); assert.equal(loadConfig({ ...env, ALLOW_HTTP: 'true' }).hmacKey.at(-1), 10); assert.throws(() => loadConfig({ ...env, OPENFN_WEBHOOK_URL: 'https://secret:canary@example.org/' }), /^Error: invalid bridge configuration$/); + await writeFile(join(dir, 'fields'), '[]'); + assert.deepEqual(loadConfig({ ...env, ALLOW_HTTP: 'true' }).allowedValueFields, []); }); From 16a508d1032a224e003312a5ffda150e8f41394d Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Sun, 13 Sep 2026 22:08:30 +0700 Subject: [PATCH 3/4] feat(bridge): compose multiple pinned workflow adaptors Signed-off-by: Jeremi Joslin --- services/event-bridge/README.md | 4 +++- services/event-bridge/src/worker.js | 10 +++++++--- services/event-bridge/test/inbox.test.js | 17 ++++++++++++++++- 3 files changed, 26 insertions(+), 5 deletions(-) diff --git a/services/event-bridge/README.md b/services/event-bridge/README.md index e9d3088..e95a19a 100644 --- a/services/event-bridge/README.md +++ b/services/event-bridge/README.md @@ -114,7 +114,9 @@ The worker uses the same `OPENFN_INBOX_PATH` and a reviewed ``` Job and configuration paths resolve relative to this file. Use an absolute -installed adaptor path as shown. `configurationFile` is optional. Pin the CLI +installed adaptor path as shown. `configurationFile` is optional. Use +`adaptors: ["package=/path", "second-package=/path"]` instead of `adaptor` +when a job composes two or more adaptors; never set both forms. Pin the CLI and adaptors at installation; execution disables automatic installation and adaptor-name expansion. `OPENFN_BINARY` selects the installed CLI executable (default `openfn`). `OPENFN_JOB_TIMEOUT_MS` defaults to 60000, maximum 1800000. diff --git a/services/event-bridge/src/worker.js b/services/event-bridge/src/worker.js index ebde161..b9a3e4c 100644 --- a/services/event-bridge/src/worker.js +++ b/services/event-bridge/src/worker.js @@ -12,11 +12,14 @@ export function loadWorkerConfig(env = process.env) { if (!workflows || Array.isArray(workflows) || typeof workflows !== 'object' || !Object.keys(workflows).length || !env.OPENFN_INBOX_PATH) throw new Error(); for (const [effect, binding] of Object.entries(workflows)) { + const adaptors = binding?.adaptors ?? (binding?.adaptor ? [binding.adaptor] : []); if (!/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(effect) || !binding || typeof binding.job !== 'string' || !binding.job || - typeof binding.adaptor !== 'string' || !binding.adaptor || + (binding.adaptor !== undefined && binding.adaptors !== undefined) || + !Array.isArray(adaptors) || adaptors.length === 0 || + adaptors.some(adaptor => typeof adaptor !== 'string' || !adaptor) || (binding.configurationFile !== undefined && typeof binding.configurationFile !== 'string') || - Object.keys(binding).some(key => !['job', 'adaptor', 'configurationFile'].includes(key))) throw new Error(); + Object.keys(binding).some(key => !['job', 'adaptor', 'adaptors', 'configurationFile'].includes(key))) throw new Error(); binding.job = resolve(dirname(workflowsPath), binding.job); if (binding.configurationFile) binding.configurationFile = resolve(dirname(workflowsPath), binding.configurationFile); } @@ -41,7 +44,8 @@ export async function runCli(item, config) { await new Promise((resolveRun, rejectRun) => { // Reviewed executable/job/adaptor paths only. Event data never becomes // shell syntax, CLI arguments, a job name, or log output. - const child = spawn(config.binary, [binding.job, '-a', binding.adaptor, + const adaptors = binding.adaptors ?? [binding.adaptor]; + const child = spawn(config.binary, [binding.job, ...adaptors.flatMap(adaptor => ['-a', adaptor]), '-s', state, '-o', output, '--no-cache-steps', '--no-autoinstall', '--no-expand-adaptors', '--repo-dir', join(directory, 'repo'), '--timeout', String(config.timeoutMs)], diff --git a/services/event-bridge/test/inbox.test.js b/services/event-bridge/test/inbox.test.js index 810e711..0dcf833 100644 --- a/services/event-bridge/test/inbox.test.js +++ b/services/event-bridge/test/inbox.test.js @@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test } from 'node:test'; import { DurableInbox } from '../src/inbox.js'; -import { runOnce, runCli } from '../src/worker.js'; +import { loadWorkerConfig, runOnce, runCli } from '../src/worker.js'; const envelope = { event: { source: 'urn:synthetic:registry', id: 'event-1' }, delivery: { generation: 1 }, data: { values: { reference: 'synthetic-record' } } }; @@ -122,3 +122,18 @@ test('CLI handoff uses protected state files, bounds execution and removes trans await assert.rejects(runCli({ effect: 'notify', id: 'synthetic', envelope }, { ...config, timeoutMs: 100 })); assert.equal((await readdir(directory)).some(file => file.startsWith('attempt-')), false); }); + +test('reviewed workflow configuration binds one or several exact adaptor paths without ambiguity', async t => { + const { path, directory } = await fixture(t); + const file = join(directory, 'workflows.json'); + const env = { OPENFN_INBOX_PATH: path, OPENFN_WORKFLOWS_FILE: file }; + const binding = { job: './job.js', adaptors: ['evidence=/installed/evidence', 'casework=/installed/casework'] }; + await writeFile(file, JSON.stringify({ intake: binding })); + const config = loadWorkerConfig(env); + assert.equal(config.workflows.intake.job, join(directory, 'job.js')); + assert.deepEqual(config.workflows.intake.adaptors, binding.adaptors); + await writeFile(file, JSON.stringify({ intake: { ...binding, adaptor: 'common=/installed/common' } })); + assert.throws(() => loadWorkerConfig(env), /invalid worker configuration/); + await writeFile(file, JSON.stringify({ intake: { ...binding, adaptors: [] } })); + assert.throws(() => loadWorkerConfig(env), /invalid worker configuration/); +}); From 4914519050740933c4bae384631452cdcecc8e7e Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 14 Sep 2026 01:30:13 +0700 Subject: [PATCH 4/4] Allow exact configured BREG event ingress path Signed-off-by: Codex --- services/event-bridge/README.md | 5 ++++- services/event-bridge/src/server.js | 10 +++++++--- services/event-bridge/test/bridge.test.js | 18 ++++++++++++++++-- 3 files changed, 27 insertions(+), 6 deletions(-) diff --git a/services/event-bridge/README.md b/services/event-bridge/README.md index e95a19a..2882dc6 100644 --- a/services/event-bridge/README.md +++ b/services/event-bridge/README.md @@ -5,7 +5,9 @@ workflow to a durable inbox or submitting a work order to a fixed OpenFn webhook dependencies. Start with `node services/event-bridge/src/server.js` from the repository root. -`POST /events/breg` is the only ingress route. The method, path without query, +`POST /events/breg` is the default ingress route. Set `BREG_EVENT_PATH` to one +fixed `/events/` path when an authored BREG destination uses another +route. The method, exact path without query, content type, all signed headers, and bounded exact body bytes are covered by the BREG `breg-webhook-signature-v1` HMAC contract: unsigned 64-bit big-endian byte lengths precede each field in the order defined in Registry Stack's @@ -29,6 +31,7 @@ Configuration is required unless a default is listed: | --- | --- | | `BREG_HMAC_KEY_FILE` | File containing exact HMAC key bytes, at least 32 bytes. | | `BREG_EXPECTED_SOURCE` | Exact CloudEvent source for the registry instance. | +| `BREG_EVENT_PATH` | Optional exact signed ingress path, default `/events/breg`; one lowercase name under `/events/`, with letters, digits or hyphens. | | `BREG_EXPECTED_ENTITY` | Exact payload entity, `farm` for this pilot. | | `BREG_EXPECTED_EVENTS_FILE` | JSON object mapping each exact type to `{ "schema": "exact-dataschema", "trigger": "created" }`; `patched` and `request_lifecycle` are also supported. Optional per-event `entity` and `valueFields` override defaults; CLI mode requires `effect`. | | `BREG_ALLOWED_VALUE_FIELDS_FILE` | JSON array of exact projected string fields, `["local-identifier"]` for this pilot. | diff --git a/services/event-bridge/src/server.js b/services/event-bridge/src/server.js index e034fab..c3261cb 100644 --- a/services/event-bridge/src/server.js +++ b/services/event-bridge/src/server.js @@ -28,13 +28,15 @@ function positive(value, fallback, maximum) { export function loadConfig(env = process.env) { try { const hmacKey = readFileSync(env.BREG_HMAC_KEY_FILE); + const eventPath = env.BREG_EVENT_PATH ?? EVENT_PATH; const deliveryMode = env.OPENFN_DELIVERY_MODE ?? 'webhook'; if (!['webhook', 'cli'].includes(deliveryMode)) throw new Error('invalid delivery mode'); const apiKey = deliveryMode === 'webhook' ? readFileSync(env.OPENFN_API_KEY_FILE, 'utf8') : undefined; const expectedEvents = JSON.parse(readFileSync(env.BREG_EXPECTED_EVENTS_FILE, 'utf8')); const allowedValueFields = JSON.parse(readFileSync(env.BREG_ALLOWED_VALUE_FIELDS_FILE, 'utf8')); const url = deliveryMode === 'webhook' ? new URL(env.OPENFN_WEBHOOK_URL) : undefined; - if (hmacKey.length < 32 || (apiKey !== undefined && !/^[\x21-\x7e]+$/.test(apiKey)) || + if (hmacKey.length < 32 || !/^\/events\/[a-z0-9][a-z0-9-]{0,63}$/.test(eventPath) || + (apiKey !== undefined && !/^[\x21-\x7e]+$/.test(apiKey)) || !env.BREG_EXPECTED_SOURCE || !env.BREG_EXPECTED_ENTITY || !object(expectedEvents) || Object.keys(expectedEvents).length === 0 || Object.entries(expectedEvents).some(([type, binding]) => !type || !object(binding) || @@ -56,6 +58,7 @@ export function loadConfig(env = process.env) { hmacKey, apiKey, expectedEvents, allowedValueFields, deliveryMode, inboxPath: env.OPENFN_INBOX_PATH, expectedSource: env.BREG_EXPECTED_SOURCE, expectedEntity: env.BREG_EXPECTED_ENTITY, + eventPath, openfnUrl: url, port: positive(env.PORT, 8081, 65535), maxBodyBytes: positive(env.MAX_BODY_BYTES, 65536, 1048576), maxDeliverySkewSeconds: positive(env.MAX_DELIVERY_SKEW_SECONDS, 300, 3600), @@ -210,14 +213,15 @@ function respond(response, status, code) { export function createBridge(config) { const inbox = config.deliveryMode === 'cli' ? new DurableInbox(config.inboxPath) : undefined; + const eventPath = config.eventPath ?? EVENT_PATH; const server = http.createServer({ maxHeaderSize: 16384 }, async (request, response) => { try { if (request.url === '/healthz' && request.method === 'GET') return respond(response, 200, 'ok'); - if (request.url !== EVENT_PATH || request.method !== 'POST') return respond(response, 404, 'not_found'); + if (request.url !== eventPath || request.method !== 'POST') return respond(response, 404, 'not_found'); const binding = verifyHeaders(request, config); const body = await readBody(request, config.maxBodyBytes); const provided = request.headers['x-registry-signature']; - const expected = webhookSignature(config.hmacKey, request.headers, body); + const expected = webhookSignature(config.hmacKey, request.headers, body, 'POST', eventPath); if (!/^v1=[A-Za-z0-9_-]{43}$/.test(provided) || !timingSafeEqual(Buffer.from(provided), Buffer.from(expected))) throw new Refusal(401, 'invalid_signature'); const data = verifyPayload(body, config, binding); diff --git a/services/event-bridge/test/bridge.test.js b/services/event-bridge/test/bridge.test.js index da78899..bd77b50 100644 --- a/services/event-bridge/test/bridge.test.js +++ b/services/event-bridge/test/bridge.test.js @@ -30,7 +30,7 @@ function sign(headers, body, path = EVENT_PATH, method = 'POST') { return `v1=${createHmac('sha256', key).update(Buffer.concat(parts)).digest('base64url')}`; } -function delivery(changes = {}, value = data) { +function delivery(changes = {}, value = data, path = EVENT_PATH) { const body = Buffer.from(JSON.stringify(value)); const headers = { 'content-type': 'application/json', 'ce-specversion': '1.0', 'ce-id': '22222222-2222-4222-8222-222222222222', 'ce-source': source, 'ce-type': 'farm-created-v1', @@ -38,7 +38,7 @@ function delivery(changes = {}, value = data) { 'x-registry-event-generation': '1', 'x-registry-delivery-attempt': '1', 'x-registry-delivery-time': new Date().toISOString(), 'idempotency-key': `sha256:${'c'.repeat(64)}`, ...changes }; - headers['x-registry-signature'] = sign(headers, body); + headers['x-registry-signature'] = sign(headers, body, path); return { headers, body }; } @@ -87,6 +87,17 @@ test('forwards verified envelope only after real work-order acceptance; old even assert.equal(JSON.stringify(calls[0]).includes(key.toString()), false); }); +test('configured event path is exact and bound into the HMAC', async t => { + const eventPath = '/events/laboratory'; + const { url, calls } = await fixture(t, undefined, { eventPath }); + assert.equal((await fetch(`${url}${EVENT_PATH}`, { method: 'POST', ...delivery() })).status, 404); + assert.equal((await fetch(`${url}${eventPath}?other=1`, { method: 'POST', ...delivery({}, data, eventPath) })).status, 404); + assert.equal((await fetch(`${url}${eventPath}`, { method: 'POST', ...delivery() })).status, 401); + assert.equal(calls.length, 0); + assert.equal((await fetch(`${url}${eventPath}`, { method: 'POST', ...delivery({}, data, eventPath) })).status, 202); + assert.equal(calls.length, 1); +}); + test('rejects body and every signed header tampering without upstream effects', async t => { const { url, calls } = await fixture(t); for (const field of ['ce-specversion', 'ce-id', 'ce-source', 'ce-type', 'ce-time', 'ce-dataschema', @@ -186,6 +197,9 @@ test('configuration uses exact secret bytes, explicit HTTP trust and safe errors BREG_EXPECTED_SOURCE: source, BREG_EXPECTED_ENTITY: 'farm', OPENFN_WEBHOOK_URL: 'http://openfn:4000/i/pilot' }; assert.throws(() => loadConfig(env), /^Error: invalid bridge configuration$/); assert.equal(loadConfig({ ...env, ALLOW_HTTP: 'true' }).hmacKey.at(-1), 10); + assert.equal(loadConfig({ ...env, ALLOW_HTTP: 'true', BREG_EVENT_PATH: '/events/laboratory' }).eventPath, '/events/laboratory'); + assert.throws(() => loadConfig({ ...env, ALLOW_HTTP: 'true', BREG_EVENT_PATH: '/events/laboratory?other=1' }), + /^Error: invalid bridge configuration$/); assert.throws(() => loadConfig({ ...env, OPENFN_WEBHOOK_URL: 'https://secret:canary@example.org/' }), /^Error: invalid bridge configuration$/); await writeFile(join(dir, 'fields'), '[]'); assert.deepEqual(loadConfig({ ...env, ALLOW_HTTP: 'true' }).allowedValueFields, []);