openvibers/openvibe.watch Commits
Watch: node sources — check something on your own network from your machine (probe through Bot), with a form template (#58)
Changes
@@@ -132,7 +132,15 @@ This release carries the pull rungs and event sources:| `event` | delivered by OpenVibe.Events (see [Event sources](#event-sources)) | implemented || `webhook` | a signed POST to `/hooks/<watch id>` (see [Webhook sources](#webhook-sources)) | implemented || `run` | browser/code polling through OpenVibe.Run | step 6 |--| `node` | a probe through OpenVibe.Node | step 7 |++| `node` | a probe from the owner's own machine through OpenVibe.Bot and OpenVibe.Node (see below) | implemented (step 7) |++++**Node sources** (plan T18 step 7, contracts 0.167.0, [server/carriers/node.js](server/carriers/node.js)): `{ kind: 'node', node_id: 'dev_…', probe: { target, timeout_s } }`. The check runs from the owner's own machine:++- **The call**: Watch asks OpenVibe.Bot (`WATCH_BOT_URL`, default `http://127.0.0.1:4630`) with its own token (`bot.computer.control`) and the owner in `X-OV-Subject`. Bot answers only for that person's machine, only where they turned on "Let my watches check things from this computer", and Watch may send nothing but the read-only `probe`.++- **Targets**: the machine's OpenVibe Node reaches only its own network (private, loopback or link-local addresses).++- **http(s) targets**: the answer's body is extracted like a page, and a status other than 200/203 is an `http_error`.++- **tcp targets**: the document is `{ open, status: 'up' | 'down', ms }`.++- **Owners**: an agent-owned watch is never probed (`node_person_only`).++- **The form**: its "Something on your network goes down" template makes one: a machine id, a host:port, and the condition `status` eq `down`.A kind whose carrier does not exist yet is **accepted by the registry but never fetched**: thecheck is recorded `skipped` with `watch.carrier_unavailable` and the reason names the step. That is@@@ -468,8 +476,8 @@ Called elsewhere, as the service principal `watch` (client credentials from Open422), cross-site writes refused, and hostile names and values escaped on every pageNot yet demonstrated: a check against a real site from a deployed service (nothing is deployed--yet), the Run rung (step 6) and Node probes (step 7) — each answers `watch.carrier_unavailable` and fetches nothing--today — and the action kinds beyond notifications. Event sources are covered end to end by++yet), the Run rung (step 6), which answers `watch.carrier_unavailable` and fetches nothing++today (node sources are covered by `test/node-carrier.test.js` against a stand-in Bot) — and the action kinds beyond notifications. Event sources are covered end to end by`test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route and the subscriptionsync against a stand-in Events), and webhook sources by `test/webhook-sources.test.js` (the secret shown once, bothsigning forms, every refusal, the budget, rotation, text and JSON bodies, and removal).@@@ -12,7 +12,7 @@"dotenv": "^16.4.7","express": "^4.21.2","fast-xml-parser": "^4.5.3",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.167.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3","pg": "^8.23.0"@@@ -769,9 +769,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.165.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.0",-- "integrity": "sha512-IfBo0pI0lAqSGvtjA7PWjzf2apWMOrY1wAvHB9miGUgVOsjlLbkE8ND8SSutExoLGmZRsf+pePcCw2QbGjTjIA==",++ "version": "0.167.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.167.0",++ "integrity": "sha512-CVV3IX+jS6WJM5+7lYS3KK8ZJUNkTyBY55Pal3CG0hZlSDpIqGa/wFw1uSVgoo4vE4vkljE6hIjz6cGvopLrnQ==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -24,7 +24,7 @@"dotenv": "^16.4.7","express": "^4.21.2","fast-xml-parser": "^4.5.3",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.167.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3","pg": "^8.23.0"@@@ -14,8 +14,8 @@* server/event-sources.js); a check with no event (Check now) answers watch.event_awaited* webhook step 5: a signed POST to /hooks/<watch id> runs the check with the delivery (carriers/webhook.js,* server/webhook-sources.js); a check with no delivery (Check now) answers watch.webhook_awaited++ * node step 7: a probe from the owner's own machine, through OpenVibe.Bot (carriers/node.js, contracts 0.167)* run step 6-- * node step 7** A kind whose carrier is not built yet is accepted by the registry (a watch may be defined before* its transport exists) but is NEVER fetched: choose() answers `watch.carrier_unavailable`, the@@@ -27,14 +27,14 @@ const { createFeedCarrier } = require('./feed');const { createApiCarrier } = require('./api');const { createEventCarrier } = require('./event');const { createWebhookCarrier } = require('./webhook');++const { createNodeCarrier } = require('./node');/** The documented order. A source kind fixes its carrier; this is the order a watch would prefer. */--const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api'];++const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api', 'node'];/** The step of plan T18 that carries each kind this release does not. */const LATER = {run: { step: 6, note: 'browser polling goes through OpenVibe.Run only after every cheaper rung was unavailable' },-- node: { step: 7, note: 'a node probe observes a local/LAN thing through OpenVibe.Node' },};function unavailable(kind) {@@@ -48,13 +48,14 @@ function unavailable(kind) {};}--function createCarriers({ fetcher, config, log = console }) {++function createCarriers({ fetcher, config, log = console, machines = null }) {const carriers = {http: createHttpCarrier({ fetcher, config, log }),feed: createFeedCarrier({ fetcher, config, log }),api: createApiCarrier({ fetcher, config, log }),event: createEventCarrier(),webhook: createWebhookCarrier(),++ node: createNodeCarrier({ machines }),};/**nnew file mode 100644@@@ -0,0 +1,52 @@++'use strict';++/**++ * The node carrier (plan T18 step 7): a check made FROM the owner's own machine, of something on that machine's own++ * network (a NAS's status page, a printer, a port on the router), through OpenVibe.Bot and the machine's OpenVibe Node++ * (the computer plugin's read-only `probe`). Watch never reaches the person's network itself.++ *++ * source: { kind: 'node', node_id: 'dev_…', probe: { target: 'http(s)://…' | 'tcp://host:port', timeout_s? } }++ *++ * An http(s) probe states its body (one GET from the machine, no redirect followed): the body and content type are++ * extracted like any page, and a status other than 200/203 is an http_error. A tcp probe states the document++ * { open, ms, error? } (as JSON too, so json and jsonpath extractions read it). The owner must be a person: Bot acts++ * only for a machine's owner.++ */++const { sha256 } = require('../util');++++const PERSON = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const base = { http_status: null, bytes: null, raw_body_hash: null, etag: null, last_modified: null, body: null, content_type: null, detail: null, error_code: null, notBefore: null };++++function createNodeCarrier({ machines }) {++ return {++ name: 'node',++ run: async ({ watch, owner }) => {++ if (!machines) return { ...base, state: 'disabled', error_code: 'node_unavailable', detail: 'Watch has no OpenVibe.Bot client here (no service credentials)' };++ if (!PERSON.test(String(owner || ''))) return { ...base, state: 'disabled', error_code: 'node_person_only', detail: 'a node source checks from a person\'s own machine: the watch must belong to a person' };++ const { node_id: machineId, probe } = watch.source;++ let answer;++ try {++ answer = await machines.probe({ subjectId: owner, machineId, probe });++ } catch (err) {++ const code = err && err.code ? String(err.code) : 'node_unreachable';++ const offline = code === 'bot.device_offline';++ return { ...base, state: offline ? 'timeout' : 'http_error', error_code: code, detail: String((err && (err.detail || err.message)) || 'OpenVibe.Bot did not answer').slice(0, 500) };++ }++ if (!answer || answer.result !== 'ack' || !answer.data) {++ return { ...base, state: answer && answer.result === 'expired' ? 'timeout' : 'http_error', error_code: `node_${(answer && answer.result) || 'no_answer'}`, detail: String((answer && answer.reason) || 'the machine did not answer the probe').slice(0, 500) };++ }++ const d = answer.data;++ if (typeof d.open === 'boolean') {++ // `status` says it in words, so a condition can compare it ({ op: 'eq', value: 'down' }).++ const document = { open: d.open, status: d.open ? 'up' : 'down', ms: Number.isFinite(d.ms) ? d.ms : null, ...(d.error ? { error: String(d.error) } : {}) };++ const body = JSON.stringify(document);++ return { ...base, state: 'ok', document, body, content_type: 'application/json', bytes: body.length, raw_body_hash: sha256(body) };++ }++ const body = typeof d.body === 'string' ? d.body : '';++ const out = { ...base, http_status: Number.isInteger(d.status) ? d.status : null, content_type: d.content_type || null, etag: d.etag || null, last_modified: d.last_modified || null, bytes: body.length };++ if (out.http_status !== 200 && out.http_status !== 203) return { ...out, state: 'http_error', error_code: `http_${out.http_status}`, detail: `HTTP ${out.http_status} from ${probe.target} (seen from the machine)` };++ return { ...out, state: 'ok', body, raw_body_hash: sha256(body) };++ },++ };++}++++module.exports = { createNodeCarrier };@@@ -222,7 +222,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obserconst state = url ? (await st.endpoint.get(watch.id, url)) || {} : {};let out;try {-- out = await choice.run({ watch, state, now, input });++ out = await choice.run({ watch, state, now, input, owner: row.owner_sub });} catch (err) {log.error(`[check] ${watch.id}: carrier ${carrier} crashed: ${err.stack || err}`);out = { state: 'http_error', error_code: 'internal', detail: 'the carrier failed unexpectedly' };@@@ -105,6 +105,8 @@ function load(env = process.env) {runUrl: (env.WATCH_RUN_URL || 'http://127.0.0.1:4920').replace(/\/+$/, ''),// OpenVibe.Codes' internal address, for codes-run actions (a coding task on the owner's own machine).codesUrl: (env.WATCH_CODES_URL || 'http://127.0.0.1:4900').replace(/\/+$/, ''),++ // OpenVibe.Bot's internal address, for node sources (a probe from the owner's own machine).++ botUrl: (env.WATCH_BOT_URL || 'http://127.0.0.1:4630').replace(/\/+$/, ''),},// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins@@@ -85,6 +85,16 @@ const TEMPLATES = {defaultOp: 'changed',extraction: 'json',},++ lan: {++ card: 'Something on your network goes down',++ cardBlurb: 'A NAS, a printer, a camera or a game server at home: your own computer checks that its port answers, and Watch tells you when it stops. Nothing on your network is opened to the internet.',++ title: 'Watch something on your own network',++ kind: 'node',++ kindWords: 'a device on your own network',++ ops: ['eq'],++ extraction: 'json',++ valuePath: 'status',++ },};const TEMPLATE_NAMES = Object.keys(TEMPLATES);// The public events the event template offers (Contracts: each is `public`, so any watch may read it): which payload@@@ -140,7 +150,7 @@ function createPageRoutes(ctx) {extraction: t.extraction, selector: '', value_path: t.valuePath || '',op: t.defaultOp || t.ops[0], value: '', for_sec: '',cadence: '3600', notify: 'on', agent_on: '', agent_id: '', agent_note: '', codes_on: '', codes_repo: '', codes_brief: '',-- event: EVENT_CHOICES[0].value, event_filter: '',++ event: EVENT_CHOICES[0].value, event_filter: '', machine: '', lan_address: '',};}@@@ -161,6 +171,7 @@ function createPageRoutes(ctx) {const kind = watch.source && watch.source.kind;if (kind === 'webhook') return 'webhook';if (kind === 'event') return 'event';++ if (kind === 'node') return 'lan';if (kind === 'feed') return 'feed';if (kind === 'api') return 'json';const e = watch.extraction || {};@@@ -192,6 +203,8 @@ function createPageRoutes(ctx) {codes_repo: (codesAction(watch.action) || {}).repo || '',codes_brief: (codesAction(watch.action) || {}).brief || '',event: (watch.source && watch.source.kind === 'event' && watch.source.pattern) || EVENT_CHOICES[0].value,++ machine: (watch.source && watch.source.kind === 'node' && watch.source.node_id) || '',++ lan_address: (watch.source && watch.source.kind === 'node' && watch.source.probe && String(watch.source.probe.target || '').replace(/^tcp:\/\//, '')) || '',event_filter: (() => {const c = eventChoice(watch.source && watch.source.pattern);const f = watch.source && watch.source.filter;@@@ -276,6 +289,28 @@ function createPageRoutes(ctx) {if (name) body.name = name;return { body };}++ // A watch on the person's own network: their machine checks that host:port answers (carriers/node.js).++ if (template === 'lan') {++ const name = String(fields.name == null ? '' : fields.name).trim();++ if (name.length > 120) errors.name = 'A name can be at most 120 characters.';++ const machine = String(fields.machine || '').trim();++ if (!/^dev_[0-9A-HJKMNP-TV-Z]{26}$/.test(machine)) errors.machine = 'Give your machine\'s id from openvibe.bot/machines (dev_ followed by 26 characters).';++ const address = String(fields.lan_address || '').trim();++ if (!/^(\[[0-9A-Fa-f:.]+\]|[A-Za-z0-9.-]{1,253}):\d{1,5}$/.test(address) || Number(address.split(':').pop()) > 65535) errors.lan_address = 'Give the device\'s address and port, for example 192.168.1.20:80 or nas.local:5000.';++ const cadenceSec = INT(fields.cadence, 3600);++ if (cadenceSec < W.MIN_CADENCE_SEC) errors.cadence = 'The shortest cadence here is 15 minutes.';++ if (cadenceSec > 2592000) errors.cadence = 'The longest cadence is 30 days.';++ if (Object.keys(errors).length) return { errors };++ const body = {++ source: { kind: 'node', node_id: machine, probe: { target: `tcp://${address}`, timeout_s: 10 } },++ cadence: { every_sec: cadenceSec, jitter_sec: 0 },++ extraction: { kind: 'json', value_path: 'status' },++ condition: { op: 'eq', value: 'down' },++ action,++ };++ if (name) body.name = name;++ return { body };++ }// A webhook watch has no address to fetch and no cadence: it is pushed to (server/webhook-sources.js).if (template === 'webhook') {const name = String(fields.name == null ? '' : fields.name).trim();@@@ -392,14 +427,18 @@ ${error ? html`<span class="field-error" role="alert">${error}</span>` : ''}: html`<input type="hidden" name="op" value="${t.ops[0]}"><span class="fixed">fires when it ${OPS[t.ops[0]]}</span>`;const needsValue = t.ops.some((op) => NEEDS_VALUE.has(op));-- const conditionBlock = html`<fieldset><legend>When should it fire?</legend>++ const conditionBlock = template === 'lan'++ ? html`<fieldset><legend>When should it fire?</legend>++<p class="muted small">When the address stops accepting connections, seen from your machine. It fires once each time it goes down.</p>++</fieldset>`++ : html`<fieldset><legend>When should it fire?</legend>${field('Condition', opControl, { name: 'op', error: errors.op })}${needsValue ? field('Value', html`<input type="text" id="${idOf('value')}" name="value" value="${fields.value}" maxlength="200">`, { name: 'value', error: errors.value, hint: 'The value to compare the reading against. Numbers are compared as numbers.' }) : ''}${(template === 'price' || template === 'json') ? field('Hold for (seconds)', html`<input type="number" id="${idOf('for_sec')}" name="for_sec" value="${fields.for_sec}" min="0" max="2592000" inputmode="numeric">`, { name: 'for_sec', error: errors.for_sec, hint: 'Blank fires as soon as the condition is met; a number makes it wait that long and stay true.' }) : ''}</fieldset>`;let extractionBlock;-- if (template === 'event') {++ if (template === 'event' || template === 'lan') {extractionBlock = '';} else if (template === 'webhook') {extractionBlock = html`<fieldset><legend>What should it read?</legend>@@@ -426,7 +465,13 @@ ${field('Value path', html`<input type="text" id="${idOf('value_path')}" name="vconst pushed = template === 'webhook' || template === 'event';const chosen = eventChoice(fields.event) || EVENT_CHOICES[0];-- const whatBlock = template === 'event'++ const whatBlock = template === 'lan'++ ? html`<fieldset><legend>What to watch</legend>++${field('Name', html`<input type="text" id="${idOf('name')}" name="name" value="${fields.name}" maxlength="120" placeholder="e.g. The NAS">`, { name: 'name', error: errors.name, hint: 'Optional. Without one, Watch names it after the address.' })}++${field('Your machine', html`<input type="text" id="${idOf('machine')}" name="machine" value="${fields.machine}" maxlength="30" placeholder="dev_…" required>`, { name: 'machine', error: errors.machine, hint: html`A computer on that network with OpenVibe Node and its computer plugin: copy its id from <a href="https://openvibe.bot/machines">openvibe.bot/machines</a>, where you also turn on “Let my watches check things from this computer”.` })}++${field('Address and port', html`<input type="text" id="${idOf('lan_address')}" name="lan_address" value="${fields.lan_address}" maxlength="260" placeholder="192.168.1.20:80" required>`, { name: 'lan_address', error: errors.lan_address, hint: 'On your own network only (a private or local address): your machine checks that it accepts a connection.' })}++</fieldset>`++ : template === 'event'? html`<fieldset><legend>What to watch</legend>${field('Name', html`<input type="text" id="${idOf('name')}" name="name" value="${fields.name}" maxlength="120" placeholder="e.g. When goosely goes live">`, { name: 'name', error: errors.name, hint: 'Optional. Without one, Watch names it after the event.' })}${field('When', select('event', EVENT_CHOICES.map((e) => ({ value: e.value, label: e.label })), fields.event), { name: 'event', error: errors.event, hint: 'It fires the moment OpenVibe.Events delivers it, once per occurrence. Nothing is polled.' })}@@@ -18,6 +18,7 @@ const { createPoster } = require('./net/post');const { createActorTasks } = require('./actor');const { createProjects, createRunJobs } = require('./run-jobs');const { createCodesTasks } = require('./codes-tasks');++const { createMachines } = require('./machines');const { createFetcher } = require('./net/fetcher');const { createSpacer } = require('./spacer');const { createCarriers } = require('./carriers');@@@ -43,7 +44,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, post: givenPost = null, actor: givenActor = null, run: givenRun = null, codes: givenCodes = null, projects: givenProjects = 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, actor: givenActor = null, run: givenRun = null, codes: givenCodes = null, projects: givenProjects = null, machines: givenMachines = 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 });@@@ -61,7 +62,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst client = config.oauth.clientSecret? createClient({tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),-- baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl, run: config.actions.runUrl, codes: config.actions.codesUrl }, fetch: fetchImpl,++ baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl, run: config.actions.runUrl, codes: config.actions.codesUrl, bot: config.actions.botUrl }, fetch: fetchImpl,}): null;// A watch that names a project: Network says whether its owner is an admin or owner of it, when it is saved (Run checks@@@ -75,7 +76,9 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst fetcher = createFetcher({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.fetch.timeoutMs, maxBytes: config.fetch.maxBytes, maxRedirects: config.fetch.maxRedirects });const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });// Every carrier goes out through the spacer, so two watches over one host never hammer it.-- const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });++ // A node source probes from its owner's own machine through OpenVibe.Bot (carriers/node.js).++ const machines = givenMachines || (client ? createMachines(client, { baseUrl: config.actions.botUrl }) : null);++ const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log, machines });const notifications = client ? createNotificationsClient(client) : null;const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;// Webhook actions leave through the same guard as every fetch (server/net/post.js).nnew file mode 100644@@@ -0,0 +1,26 @@++'use strict';++/**++ * A person's own machine on OpenVibe.Bot, for a watch's `node` source (plan T18 step 7, contracts 0.167.0).++ *++ * const machines = createMachines(client, { baseUrl }); // client: openvibe-sdk/core createClient with Watch's token++ * await machines.probe({ subjectId, machineId, probe }) → bot.command-result@1 ({ result, data?, reason?, code? })++ *++ * probe() POSTs one `probe` (bot.command@1: { target, timeout_s }) to Bot's machine API with Watch's own token++ * (bot.computer.control, audience openvibe.bot) and the watch's owner in X-OV-Subject. Bot sends it only to that person's++ * own machine, only where they turned on "Let my watches check things from this computer", and Watch may send nothing++ * but probes. The machine itself reaches only its own network (private, loopback or link-local addresses). A refusal by++ * Bot arrives as openvibe-sdk OpenVibeError (status, code); the machine's own refusal is result "nack" with a reason.++ */++++function createMachines(client, { baseUrl }) {++ async function probe({ subjectId, machineId, probe: value }) {++ const body = { kind: 'probe', value: { target: value.target, ...(Number.isInteger(value.timeout_s) ? { timeout_s: value.timeout_s } : {}) }, ms: 45000 };++ return await client.json({++ service: 'bot', baseUrl, audience: 'openvibe.bot', method: 'POST', path: `/api/v1/machines/${encodeURIComponent(machineId)}/commands`,++ headers: { 'X-OV-Subject': subjectId }, json: body, retries: 0, idempotencyKey: false,++ });++ }++ return { probe };++}++++module.exports = { createMachines };@@@ -114,6 +114,15 @@ function validateRequest(body, { create = false } = {}) {if (cadence.every_sec < MIN_EVERY_SEC) throw new RegistryError(`cadence.every_sec must be at least ${MIN_EVERY_SEC} seconds (15 minutes)`);}if (PUSHED.has(kind) && cadence) throw new RegistryError(`a ${kind} source is pushed to and never polls: cadence must be null`);++ if (kind === 'node') {++ // A probe from the owner's own machine (carriers/node.js): the machine by its OpenVibe.Bot id, and one target on++ // its network. The machine refuses anything that is not private, loopback or link-local.++ if (!/^dev_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(body.source.node_id || ''))) throw new RegistryError('/source/node_id must be one of your machines on OpenVibe.Bot (dev_…)');++ const probe = body.source.probe || {};++ if (typeof probe.target !== 'string' || !/^(https?|tcp):\/\/[^\s]{3,2040}$/.test(probe.target)) throw new RegistryError('/source/probe/target must be http(s)://… or tcp://host:port on the machine\'s own network');++ if (probe.timeout_s !== undefined && (!Number.isInteger(probe.timeout_s) || probe.timeout_s < 1 || probe.timeout_s > 30)) throw new RegistryError('/source/probe/timeout_s is 1 to 30');++ for (const k of Object.keys(probe)) if (!['target', 'timeout_s'].includes(k)) throw new RegistryError(`/source/probe/${k} is not known (target, timeout_s)`);++ }// A HEAD source states no body, so only a document/record extraction can read one: refuse// the misconfiguration here instead of recording `parse_error` on every check.if (body.source.method === 'HEAD' && !['json', 'jsonpath'].includes((body.extraction || {}).kind)) {@@@ -84,12 +84,19 @@ function inWords(watch) {what = e.value_path ? html`${extractionWords(watch.extraction)} of the newest item of the feed` : 'the newest item of the feed';} else if (source.kind === 'webhook') {what = path ? html`the value at <code>${path}</code> in a webhook delivery` : (e.kind === 'text' ? 'what a webhook delivery says' : 'a webhook delivery');++ } else if (source.kind === 'node') {++ const target = (source.probe && source.probe.target) || '';++ what = /^tcp:/.test(target) ? html`<b>${host(target)}</b>, seen from your computer,` : html`${extractionWords(watch.extraction)} on <b>${host(target)}</b>, seen from your computer,`;} else if (source.kind === 'event') {what = path ? html`the value at <code>${path}</code> of a <b>${source.pattern}</b> event` : html`a <b>${source.pattern}</b> event`;} else {what = html`${extractionWords(watch.extraction)} on <b>${host(source.url)}</b>`;}const every = watch.cadence ? `, checked ${everyWords(watch.cadence.every_sec)}` : '';++ const c = watch.condition || {};++ if (source.kind === 'node' && /^tcp:/.test((source.probe && source.probe.target) || '') && c.op === 'eq' && c.value === 'down') {++ return html`Tell me when <b>${host(source.probe.target)}</b> stops answering, seen from your computer${every}.`;++ }return html`Tell me when ${what} ${conditionWords(watch.condition)}${every}.`;}@@@ -176,12 +176,12 @@ t('a manual check runs one check and answers the released check-run; observation// a watch whose carrier is a later step answers 422 and never fetchesconst later = await api('POST', '/api/v1/watches', {token: ME,-- body: mine({ source: { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, cadence: { every_sec: 900 }, extraction: { kind: 'json' } }),++ body: mine({ source: { kind: 'run', runtime: 'browser', url: 'https://example.com/', steps: [{ open: 'https://example.com/' }] }, cadence: { every_sec: 900 }, extraction: { kind: 'json' } }),});const refused = await api('POST', `/api/v1/watches/${later.body.watch.id}/check`, { token: ME });assert.strictEqual(refused.status, 422);assert.strictEqual(refused.body.code, 'watch.carrier_unavailable');-- assert.match(refused.body.detail, /step 7/);++ assert.match(refused.body.detail, /step 6/);// a paused watch cannot be checked by hand eitherawait api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });const paused = await api('POST', `/api/v1/watches/${watch.id}/check`, { token: ME });@@@ -186,15 +186,14 @@ t('a pattern that hits its deadline is a recorded check error, never a hang', asprice = '19.99';});--t('a node or run source is accepted but never fetched (steps 6-7); event and webhook sources wait to be pushed to', async () => {++t('a run source is accepted but never fetched (step 6; node sources probe through Bot since step 7, test/node-carrier.test.js)', async () => {const cases = [-- ['node', { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, { op: 'exists' }],['run', { kind: 'run', runtime: 'browser', url: `${web.origin}/price`, steps: [{ goto: 'https://example.org' }] }, { op: 'exists' }],];const before = web.requests.length;for (const [kind, source, condition] of cases) {-- // an event or a webhook is pushed to and never polls; node and run sources poll through-- // OpenVibe.Node and OpenVibe.Run, so they carry a cadence — and still nothing is fetched++ // an event or a webhook is pushed to and never polls; a run source polls through OpenVibe.Run, so it carries a++ // cadence — and still nothing is fetchedconst cadence = { every_sec: 900 };const w = await create(person, { source, cadence, extraction: { kind: 'text' }, condition });assert.strictEqual((await row(w.id)).status, 'active');@@@ -202,7 +201,7 @@ t('a node or run source is accepted but never fetched (steps 6-7); event and webassert.strictEqual(out.run.state, 'skipped', `${kind}: ${out.run.detail}`);assert.strictEqual(out.run.error_code, 'watch.carrier_unavailable');assert.strictEqual(out.run.carrier, kind);-- assert.match(out.run.detail, /step [67]/);++ assert.match(out.run.detail, /step 6/);assert.strictEqual(out.unavailable.code, 'watch.carrier_unavailable');assert.strictEqual((await observations(w.id)).length, 0);const [fail] = (await payloads('watch.check.failed')).slice(-1);@@@ -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, post = null, actor = null, run = null, codes = null, projects = null } = {}) {++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null, run = null, codes = null, projects = null, machines = 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 } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(codes ? { codes } : {}), ...(projects ? { projects } : {}), ...(now ? { now } : {}) });++ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(codes ? { codes } : {}), ...(projects ? { projects } : {}), ...(machines ? { machines } : {}), ...(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,98 @@++'use strict';++/**++ * The node carrier (plan T18 step 7, contracts 0.167.0): a watch checks something on its owner's own network from++ * their machine, through OpenVibe.Bot's probe. The source is checked when it is saved (a dev_ machine, an++ * http(s)/tcp target, nothing else); the probe goes to Bot as the owner; an http answer's body is extracted like a page,++ * a tcp answer's { open, ms } is a JSON document; Bot's refusal and the machine's are recorded failures; an agent-owned++ * watch is never probed.++ */++const assert = require('assert');++const { boot, suite, newUser } = require('./helpers');++++const t = suite('node-carrier');++const MACHINE = 'dev_01M4XTHNFFAWWZQHB4XTMSN1YB';++const PERSON = { sub: newUser() };++const calls = [];++let reply = null;++const machines = {++ probe: async (args) => {++ calls.push(args);++ if (reply instanceof Error) throw reply;++ return reply;++ },++};++let svc;++++const create = async (source, extraction = { kind: 'json', value_path: 'up' }, owner = PERSON) => (await svc.registry.create({++ name: 'nas', source, cadence: { every_sec: 900 }, extraction, condition: { op: 'eq', value: 'never-matches' }, action: [{ kind: 'notification' }],++}, owner)).id;++const lastRun = async (id) => (await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid DESC').get(id));++const lastValue = async (id) => {++ const o = await svc.db.prepare('SELECT * FROM observations WHERE watch_id = ? ORDER BY rid DESC').get(id);++ return o ? (typeof o.value === 'string' ? JSON.parse(o.value) : o.value) : undefined;++};++++t('boot with a stand-in for Bot\'s machine API', async () => {++ svc = await boot({ machines });++});++++t('the source is checked when it is saved', async () => {++ const bad = async (source, re) => {++ await assert.rejects(create(source), (e) => re.test(e.message));++ };++ await bad({ kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { target: 'http://192.168.1.20/' } }, /node_id/);++ await bad({ kind: 'node', node_id: MACHINE, probe: { target: 'ftp://192.168.1.20/' } }, /target/);++ await bad({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/', method: 'POST' } }, /method is not known/);++ await bad({ kind: 'node', node_id: MACHINE, probe: { target: 'tcp://192.168.1.20:22', timeout_s: 99 } }, /timeout_s/);++});++++t('an http probe goes to Bot as the owner and its body is extracted like a page', async () => {++ reply = { result: 'ack', data: { status: 200, content_type: 'application/json', etag: '"v1"', body: '{"up": true}', ms: 4 } };++ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/status', timeout_s: 5 } });++ await svc.check.run(id, { trigger: 'manual' });++ assert.deepStrictEqual(calls.at(-1), { subjectId: PERSON.sub, machineId: MACHINE, probe: { target: 'http://192.168.1.20/status', timeout_s: 5 } });++ const run = await lastRun(id);++ assert.deepStrictEqual([run.carrier, run.http_status], ['node', 200]);++ assert.strictEqual(await lastValue(id), true);++ reply = { result: 'ack', data: { status: 503, body: 'busy', ms: 3 } };++ await svc.check.run(id, { trigger: 'manual' });++ const failed = await lastRun(id);++ assert.deepStrictEqual([failed.state, failed.error_code], ['http_error', 'http_503']);++});++++t('a tcp probe states { open, ms } as a JSON document', async () => {++ reply = { result: 'ack', data: { open: false, ms: 2001, error: 'timed out' } };++ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'tcp://192.168.1.5:22' } }, { kind: 'json', value_path: 'open' });++ await svc.check.run(id, { trigger: 'manual' });++ assert.strictEqual((await lastRun(id)).state !== 'http_error', true);++ assert.strictEqual(await lastValue(id), false);++});++++t('Bot\'s refusal and the machine\'s are recorded failures with their reasons', async () => {++ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/' } });++ reply = Object.assign(new Error('the owner has not let their watches check things from this computer'), { code: 'bot.agent_not_allowed', status: 403 });++ await svc.check.run(id, { trigger: 'manual' });++ let run = await lastRun(id);++ assert.deepStrictEqual([run.state, run.error_code], ['http_error', 'bot.agent_not_allowed']);++ assert.match(run.detail, /watches/);++ reply = { result: 'nack', reason: 'a probe reaches this machine\'s own network only' };++ await svc.check.run(id, { trigger: 'manual' });++ run = await lastRun(id);++ assert.deepStrictEqual([run.state, run.error_code], ['http_error', 'node_nack']);++ reply = Object.assign(new Error('the machine is offline'), { code: 'bot.device_offline', status: 409 });++ await svc.check.run(id, { trigger: 'manual' });++ assert.strictEqual((await lastRun(id)).state, 'timeout');++});++++t('a watch owned by an agent is never probed', async () => {++ const before = calls.length;++ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/' } }, undefined, { sub: 'agt_01M4XTHNFFAWWZQHB4XTMSN1YB' });++ await svc.check.run(id, { trigger: 'manual' });++ const run = await lastRun(id);++ assert.deepStrictEqual([run.state, run.error_code], ['disabled', 'node_person_only']);++ assert.strictEqual(calls.length, before);++});++++t('done', async () => { await svc.stop(); });++++t.run();@@@ -226,6 +226,33 @@ t('the event template: a channel going live (or a blog post) wakes the watch, wiassert.strictEqual(bad.status, 422, 'only the offered public events');});++t('the network template: a device at home stops answering, checked from the person\'s own machine', async () => {++ // A second person, so this test's writes do not count against the first one's budget of 30 a minute.++ const owner = svc.network.addUser('lan-owner');++ const own = await svc.signIn(owner);++ const create = (fields) => svc.get('/watches/new', { cookie: own, form: fields });++ const ME = userToken(owner.subject);++ const session = own;++ const form = await svc.get('/watches/new?template=lan', { cookie: session });++ assert.strictEqual(form.status, 200);++ assert.match(form.text, /name="machine"/);++ assert.match(form.text, /name="lan_address"/);++ assert.match(form.text, /openvibe\.bot\/machines/);++ const made = await create({ template: 'lan', name: 'The NAS', machine: 'dev_01M4XTHNFFAWWZQHB4XTMSN1YB', lan_address: '192.168.1.20:5000', cadence: '3600', notify: 'on' });++ assert.strictEqual(made.status, 303, made.text);++ const w = (await request(svc.base, 'GET', `/api/v1/watches/${indexOf(made)}`, { token: ME })).body.watch;++ assert.deepStrictEqual(w.source, { kind: 'node', node_id: 'dev_01M4XTHNFFAWWZQHB4XTMSN1YB', probe: { target: 'tcp://192.168.1.20:5000', timeout_s: 10 } });++ assert.deepStrictEqual([w.extraction, w.condition], [{ kind: 'json', value_path: 'status' }, { op: 'eq', value: 'down' }]);++ const page = await svc.get(`/watches/${indexOf(made)}`, { cookie: session });++ assert.match(page.text, /192\.168\.1\.20:5000<\/b> stops answering, seen from your computer/);++ const edit = await svc.get(`/watches/${indexOf(made)}/edit`, { cookie: session });++ assert.match(edit.text, /value="192\.168\.1\.20:5000"/);++ const bad = await create({ template: 'lan', machine: 'nope', lan_address: 'not an address', cadence: '3600' });++ assert.strictEqual(bad.status, 422);++ assert.match(bad.text, /openvibe\.bot\/machines \(dev_/);++ assert.match(bad.text, /address and port/);++});++t('validation errors re-render the form with the registry\'s message and the person\'s input', async () => {const badUrl = await create({ template: 'page', name: 'Keep me', url: 'not-a-url', selector: '.x', cadence: '3600' });assert.strictEqual(badUrl.status, 422);
@@ -132,7 +132,15 @@ This release carries the pull rungs and event sources:
@@ -132,7 +132,15 @@ This release carries the pull rungs and event sources:
| `event` | delivered by OpenVibe.Events (see [Event sources](#event-sources)) | implemented |
| `event` | delivered by OpenVibe.Events (see [Event sources](#event-sources)) | implemented |
| `webhook` | a signed POST to `/hooks/<watch id>` (see [Webhook sources](#webhook-sources)) | implemented |
| `webhook` | a signed POST to `/hooks/<watch id>` (see [Webhook sources](#webhook-sources)) | implemented |
| `run` | browser/code polling through OpenVibe.Run | step 6 |
| `run` | browser/code polling through OpenVibe.Run | step 6 |
-| `node` | a probe through OpenVibe.Node | step 7 |
+| `node` | a probe from the owner's own machine through OpenVibe.Bot and OpenVibe.Node (see below) | implemented (step 7) |
+
+**Node sources** (plan T18 step 7, contracts 0.167.0, [server/carriers/node.js](server/carriers/node.js)): `{ kind: 'node', node_id: 'dev_…', probe: { target, timeout_s } }`. The check runs from the owner's own machine:
+- **The call**: Watch asks OpenVibe.Bot (`WATCH_BOT_URL`, default `http://127.0.0.1:4630`) with its own token (`bot.computer.control`) and the owner in `X-OV-Subject`. Bot answers only for that person's machine, only where they turned on "Let my watches check things from this computer", and Watch may send nothing but the read-only `probe`.
+- **Targets**: the machine's OpenVibe Node reaches only its own network (private, loopback or link-local addresses).
+- **http(s) targets**: the answer's body is extracted like a page, and a status other than 200/203 is an `http_error`.
+- **tcp targets**: the document is `{ open, status: 'up' | 'down', ms }`.
+- **Owners**: an agent-owned watch is never probed (`node_person_only`).
+- **The form**: its "Something on your network goes down" template makes one: a machine id, a host:port, and the condition `status` eq `down`.
A kind whose carrier does not exist yet is **accepted by the registry but never fetched**: the
A kind whose carrier does not exist yet is **accepted by the registry but never fetched**: the
check is recorded `skipped` with `watch.carrier_unavailable` and the reason names the step. That is
check is recorded `skipped` with `watch.carrier_unavailable` and the reason names the step. That is
@@ -468,8 +476,8 @@ Called elsewhere, as the service principal `watch` (client credentials from Open
@@ -468,8 +476,8 @@ Called elsewhere, as the service principal `watch` (client credentials from Open
422), cross-site writes refused, and hostile names and values escaped on every page
422), cross-site writes refused, and hostile names and values escaped on every page
Not yet demonstrated: a check against a real site from a deployed service (nothing is deployed
Not yet demonstrated: a check against a real site from a deployed service (nothing is deployed
-yet), the Run rung (step 6) and Node probes (step 7) — each answers `watch.carrier_unavailable` and fetches nothing
-today — and the action kinds beyond notifications. Event sources are covered end to end by
+yet), the Run rung (step 6), which answers `watch.carrier_unavailable` and fetches nothing
+today (node sources are covered by `test/node-carrier.test.js` against a stand-in Bot) — and the action kinds beyond notifications. Event sources are covered end to end by
`test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route and the subscription
`test/event-sources.test.js` (matching, visibility, once per event, the budget, the signed route and the subscription
sync against a stand-in Events), and webhook sources by `test/webhook-sources.test.js` (the secret shown once, both
sync against a stand-in Events), and webhook sources by `test/webhook-sources.test.js` (the secret shown once, both
signing forms, every refusal, the budget, rotation, text and JSON bodies, and removal).
signing forms, every refusal, the budget, rotation, text and JSON bodies, and removal).
@@ -12,7 +12,7 @@
@@ -12,7 +12,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.167.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -769,9 +769,9 @@
@@ -769,9 +769,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.165.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.0",
- "integrity": "sha512-IfBo0pI0lAqSGvtjA7PWjzf2apWMOrY1wAvHB9miGUgVOsjlLbkE8ND8SSutExoLGmZRsf+pePcCw2QbGjTjIA==",
+ "version": "0.167.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.167.0",
+ "integrity": "sha512-CVV3IX+jS6WJM5+7lYS3KK8ZJUNkTyBY55Pal3CG0hZlSDpIqGa/wFw1uSVgoo4vE4vkljE6hIjz6cGvopLrnQ==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -24,7 +24,7 @@
@@ -24,7 +24,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.165.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.167.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -14,8 +14,8 @@
@@ -14,8 +14,8 @@
* server/event-sources.js); a check with no event (Check now) answers watch.event_awaited
* server/event-sources.js); a check with no event (Check now) answers watch.event_awaited
* webhook step 5: a signed POST to /hooks/<watch id> runs the check with the delivery (carriers/webhook.js,
* webhook step 5: a signed POST to /hooks/<watch id> runs the check with the delivery (carriers/webhook.js,
* server/webhook-sources.js); a check with no delivery (Check now) answers watch.webhook_awaited
* server/webhook-sources.js); a check with no delivery (Check now) answers watch.webhook_awaited
+ * node step 7: a probe from the owner's own machine, through OpenVibe.Bot (carriers/node.js, contracts 0.167)
* run step 6
* run step 6
- * node step 7
*
*
* A kind whose carrier is not built yet is accepted by the registry (a watch may be defined before
* A kind whose carrier is not built yet is accepted by the registry (a watch may be defined before
* its transport exists) but is NEVER fetched: choose() answers `watch.carrier_unavailable`, the
* its transport exists) but is NEVER fetched: choose() answers `watch.carrier_unavailable`, the
@@ -27,14 +27,14 @@ const { createFeedCarrier } = require('./feed');
@@ -27,14 +27,14 @@ const { createFeedCarrier } = require('./feed');
const { createApiCarrier } = require('./api');
const { createApiCarrier } = require('./api');
const { createEventCarrier } = require('./event');
const { createEventCarrier } = require('./event');
const { createWebhookCarrier } = require('./webhook');
const { createWebhookCarrier } = require('./webhook');
+const { createNodeCarrier } = require('./node');
/** The documented order. A source kind fixes its carrier; this is the order a watch would prefer. */
/** The documented order. A source kind fixes its carrier; this is the order a watch would prefer. */
-const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api'];
+const PREFERENCE = ['event', 'webhook', 'http', 'feed', 'api', 'node'];
/** The step of plan T18 that carries each kind this release does not. */
/** The step of plan T18 that carries each kind this release does not. */
const LATER = {
const LATER = {
run: { step: 6, note: 'browser polling goes through OpenVibe.Run only after every cheaper rung was unavailable' },
run: { step: 6, note: 'browser polling goes through OpenVibe.Run only after every cheaper rung was unavailable' },
- node: { step: 7, note: 'a node probe observes a local/LAN thing through OpenVibe.Node' },
};
};
function unavailable(kind) {
function unavailable(kind) {
@@ -48,13 +48,14 @@ function unavailable(kind) {
@@ -48,13 +48,14 @@ function unavailable(kind) {
};
};
}
}
-function createCarriers({ fetcher, config, log = console }) {
+function createCarriers({ fetcher, config, log = console, machines = null }) {
const carriers = {
const carriers = {
http: createHttpCarrier({ fetcher, config, log }),
http: createHttpCarrier({ fetcher, config, log }),
feed: createFeedCarrier({ fetcher, config, log }),
feed: createFeedCarrier({ fetcher, config, log }),
api: createApiCarrier({ fetcher, config, log }),
api: createApiCarrier({ fetcher, config, log }),
event: createEventCarrier(),
event: createEventCarrier(),
webhook: createWebhookCarrier(),
webhook: createWebhookCarrier(),
+ node: createNodeCarrier({ machines }),
};
};
/**
/**
new file mode 100644
new file mode 100644
@@ -0,0 +1,52 @@
@@ -0,0 +1,52 @@
+'use strict';
+/**
+ * The node carrier (plan T18 step 7): a check made FROM the owner's own machine, of something on that machine's own
+ * network (a NAS's status page, a printer, a port on the router), through OpenVibe.Bot and the machine's OpenVibe Node
+ * (the computer plugin's read-only `probe`). Watch never reaches the person's network itself.
+ *
+ * source: { kind: 'node', node_id: 'dev_…', probe: { target: 'http(s)://…' | 'tcp://host:port', timeout_s? } }
+ *
+ * An http(s) probe states its body (one GET from the machine, no redirect followed): the body and content type are
+ * extracted like any page, and a status other than 200/203 is an http_error. A tcp probe states the document
+ * { open, ms, error? } (as JSON too, so json and jsonpath extractions read it). The owner must be a person: Bot acts
+ * only for a machine's owner.
+ */
+const { sha256 } = require('../util');
+
+const PERSON = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const base = { http_status: null, bytes: null, raw_body_hash: null, etag: null, last_modified: null, body: null, content_type: null, detail: null, error_code: null, notBefore: null };
+
+function createNodeCarrier({ machines }) {
+ return {
+ name: 'node',
+ run: async ({ watch, owner }) => {
+ if (!machines) return { ...base, state: 'disabled', error_code: 'node_unavailable', detail: 'Watch has no OpenVibe.Bot client here (no service credentials)' };
+ if (!PERSON.test(String(owner || ''))) return { ...base, state: 'disabled', error_code: 'node_person_only', detail: 'a node source checks from a person\'s own machine: the watch must belong to a person' };
+ const { node_id: machineId, probe } = watch.source;
+ let answer;
+ try {
+ answer = await machines.probe({ subjectId: owner, machineId, probe });
+ } catch (err) {
+ const code = err && err.code ? String(err.code) : 'node_unreachable';
+ const offline = code === 'bot.device_offline';
+ return { ...base, state: offline ? 'timeout' : 'http_error', error_code: code, detail: String((err && (err.detail || err.message)) || 'OpenVibe.Bot did not answer').slice(0, 500) };
+ }
+ if (!answer || answer.result !== 'ack' || !answer.data) {
+ return { ...base, state: answer && answer.result === 'expired' ? 'timeout' : 'http_error', error_code: `node_${(answer && answer.result) || 'no_answer'}`, detail: String((answer && answer.reason) || 'the machine did not answer the probe').slice(0, 500) };
+ }
+ const d = answer.data;
+ if (typeof d.open === 'boolean') {
+ // `status` says it in words, so a condition can compare it ({ op: 'eq', value: 'down' }).
+ const document = { open: d.open, status: d.open ? 'up' : 'down', ms: Number.isFinite(d.ms) ? d.ms : null, ...(d.error ? { error: String(d.error) } : {}) };
+ const body = JSON.stringify(document);
+ return { ...base, state: 'ok', document, body, content_type: 'application/json', bytes: body.length, raw_body_hash: sha256(body) };
+ }
+ const body = typeof d.body === 'string' ? d.body : '';
+ const out = { ...base, http_status: Number.isInteger(d.status) ? d.status : null, content_type: d.content_type || null, etag: d.etag || null, last_modified: d.last_modified || null, bytes: body.length };
+ if (out.http_status !== 200 && out.http_status !== 203) return { ...out, state: 'http_error', error_code: `http_${out.http_status}`, detail: `HTTP ${out.http_status} from ${probe.target} (seen from the machine)` };
+ return { ...out, state: 'ok', body, raw_body_hash: sha256(body) };
+ },
+ };
+}
+
+module.exports = { createNodeCarrier };
@@ -222,7 +222,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
@@ -222,7 +222,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
const state = url ? (await st.endpoint.get(watch.id, url)) || {} : {};
const state = url ? (await st.endpoint.get(watch.id, url)) || {} : {};
let out;
let out;
try {
try {
- out = await choice.run({ watch, state, now, input });
+ out = await choice.run({ watch, state, now, input, owner: row.owner_sub });
} catch (err) {
} catch (err) {
log.error(`[check] ${watch.id}: carrier ${carrier} crashed: ${err.stack || err}`);
log.error(`[check] ${watch.id}: carrier ${carrier} crashed: ${err.stack || err}`);
out = { state: 'http_error', error_code: 'internal', detail: 'the carrier failed unexpectedly' };
out = { state: 'http_error', error_code: 'internal', detail: 'the carrier failed unexpectedly' };
@@ -105,6 +105,8 @@ function load(env = process.env) {
@@ -105,6 +105,8 @@ function load(env = process.env) {
runUrl: (env.WATCH_RUN_URL || 'http://127.0.0.1:4920').replace(/\/+$/, ''),
runUrl: (env.WATCH_RUN_URL || 'http://127.0.0.1:4920').replace(/\/+$/, ''),
// OpenVibe.Codes' internal address, for codes-run actions (a coding task on the owner's own machine).
// OpenVibe.Codes' internal address, for codes-run actions (a coding task on the owner's own machine).
codesUrl: (env.WATCH_CODES_URL || 'http://127.0.0.1:4900').replace(/\/+$/, ''),
codesUrl: (env.WATCH_CODES_URL || 'http://127.0.0.1:4900').replace(/\/+$/, ''),
+ // OpenVibe.Bot's internal address, for node sources (a probe from the owner's own machine).
+ botUrl: (env.WATCH_BOT_URL || 'http://127.0.0.1:4630').replace(/\/+$/, ''),
},
},
// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins
// Who may use which named secret (server/secret-grants.js): a WATCH_CRED_* credential (owners and the origins
@@ -85,6 +85,16 @@ const TEMPLATES = {
@@ -85,6 +85,16 @@ const TEMPLATES = {
defaultOp: 'changed',
defaultOp: 'changed',
extraction: 'json',
extraction: 'json',
},
},
+ lan: {
+ card: 'Something on your network goes down',
+ cardBlurb: 'A NAS, a printer, a camera or a game server at home: your own computer checks that its port answers, and Watch tells you when it stops. Nothing on your network is opened to the internet.',
+ title: 'Watch something on your own network',
+ kind: 'node',
+ kindWords: 'a device on your own network',
+ ops: ['eq'],
+ extraction: 'json',
+ valuePath: 'status',
+ },
};
};
const TEMPLATE_NAMES = Object.keys(TEMPLATES);
const TEMPLATE_NAMES = Object.keys(TEMPLATES);
// The public events the event template offers (Contracts: each is `public`, so any watch may read it): which payload
// The public events the event template offers (Contracts: each is `public`, so any watch may read it): which payload
@@ -140,7 +150,7 @@ function createPageRoutes(ctx) {
@@ -140,7 +150,7 @@ function createPageRoutes(ctx) {
extraction: t.extraction, selector: '', value_path: t.valuePath || '',
extraction: t.extraction, selector: '', value_path: t.valuePath || '',
op: t.defaultOp || t.ops[0], value: '', for_sec: '',
op: t.defaultOp || t.ops[0], value: '', for_sec: '',
cadence: '3600', notify: 'on', agent_on: '', agent_id: '', agent_note: '', codes_on: '', codes_repo: '', codes_brief: '',
cadence: '3600', notify: 'on', agent_on: '', agent_id: '', agent_note: '', codes_on: '', codes_repo: '', codes_brief: '',
- event: EVENT_CHOICES[0].value, event_filter: '',
+ event: EVENT_CHOICES[0].value, event_filter: '', machine: '', lan_address: '',
};
};
}
}
@@ -161,6 +171,7 @@ function createPageRoutes(ctx) {
@@ -161,6 +171,7 @@ function createPageRoutes(ctx) {
const kind = watch.source && watch.source.kind;
const kind = watch.source && watch.source.kind;
if (kind === 'webhook') return 'webhook';
if (kind === 'webhook') return 'webhook';
if (kind === 'event') return 'event';
if (kind === 'event') return 'event';
+ if (kind === 'node') return 'lan';
if (kind === 'feed') return 'feed';
if (kind === 'feed') return 'feed';
if (kind === 'api') return 'json';
if (kind === 'api') return 'json';
const e = watch.extraction || {};
const e = watch.extraction || {};
@@ -192,6 +203,8 @@ function createPageRoutes(ctx) {
@@ -192,6 +203,8 @@ function createPageRoutes(ctx) {
codes_repo: (codesAction(watch.action) || {}).repo || '',
codes_repo: (codesAction(watch.action) || {}).repo || '',
codes_brief: (codesAction(watch.action) || {}).brief || '',
codes_brief: (codesAction(watch.action) || {}).brief || '',
event: (watch.source && watch.source.kind === 'event' && watch.source.pattern) || EVENT_CHOICES[0].value,
event: (watch.source && watch.source.kind === 'event' && watch.source.pattern) || EVENT_CHOICES[0].value,
+ machine: (watch.source && watch.source.kind === 'node' && watch.source.node_id) || '',
+ lan_address: (watch.source && watch.source.kind === 'node' && watch.source.probe && String(watch.source.probe.target || '').replace(/^tcp:\/\//, '')) || '',
event_filter: (() => {
event_filter: (() => {
const c = eventChoice(watch.source && watch.source.pattern);
const c = eventChoice(watch.source && watch.source.pattern);
const f = watch.source && watch.source.filter;
const f = watch.source && watch.source.filter;
@@ -276,6 +289,28 @@ function createPageRoutes(ctx) {
@@ -276,6 +289,28 @@ function createPageRoutes(ctx) {
if (name) body.name = name;
if (name) body.name = name;
return { body };
return { body };
}
}
+ // A watch on the person's own network: their machine checks that host:port answers (carriers/node.js).
+ if (template === 'lan') {
+ const name = String(fields.name == null ? '' : fields.name).trim();
+ if (name.length > 120) errors.name = 'A name can be at most 120 characters.';
+ const machine = String(fields.machine || '').trim();
+ if (!/^dev_[0-9A-HJKMNP-TV-Z]{26}$/.test(machine)) errors.machine = 'Give your machine\'s id from openvibe.bot/machines (dev_ followed by 26 characters).';
+ const address = String(fields.lan_address || '').trim();
+ if (!/^(\[[0-9A-Fa-f:.]+\]|[A-Za-z0-9.-]{1,253}):\d{1,5}$/.test(address) || Number(address.split(':').pop()) > 65535) errors.lan_address = 'Give the device\'s address and port, for example 192.168.1.20:80 or nas.local:5000.';
+ const cadenceSec = INT(fields.cadence, 3600);
+ if (cadenceSec < W.MIN_CADENCE_SEC) errors.cadence = 'The shortest cadence here is 15 minutes.';
+ if (cadenceSec > 2592000) errors.cadence = 'The longest cadence is 30 days.';
+ if (Object.keys(errors).length) return { errors };
+ const body = {
+ source: { kind: 'node', node_id: machine, probe: { target: `tcp://${address}`, timeout_s: 10 } },
+ cadence: { every_sec: cadenceSec, jitter_sec: 0 },
+ extraction: { kind: 'json', value_path: 'status' },
+ condition: { op: 'eq', value: 'down' },
+ action,
+ };
+ if (name) body.name = name;
+ return { body };
+ }
// A webhook watch has no address to fetch and no cadence: it is pushed to (server/webhook-sources.js).
// A webhook watch has no address to fetch and no cadence: it is pushed to (server/webhook-sources.js).
if (template === 'webhook') {
if (template === 'webhook') {
const name = String(fields.name == null ? '' : fields.name).trim();
const name = String(fields.name == null ? '' : fields.name).trim();
@@ -392,14 +427,18 @@ ${error ? html`<span class="field-error" role="alert">${error}</span>` : ''}
@@ -392,14 +427,18 @@ ${error ? html`<span class="field-error" role="alert">${error}</span>` : ''}
: html`<input type="hidden" name="op" value="${t.ops[0]}"><span class="fixed">fires when it ${OPS[t.ops[0]]}</span>`;
: html`<input type="hidden" name="op" value="${t.ops[0]}"><span class="fixed">fires when it ${OPS[t.ops[0]]}</span>`;
const needsValue = t.ops.some((op) => NEEDS_VALUE.has(op));
const needsValue = t.ops.some((op) => NEEDS_VALUE.has(op));
- const conditionBlock = html`<fieldset><legend>When should it fire?</legend>
+ const conditionBlock = template === 'lan'
+ ? html`<fieldset><legend>When should it fire?</legend>
+<p class="muted small">When the address stops accepting connections, seen from your machine. It fires once each time it goes down.</p>
+</fieldset>`
+ : html`<fieldset><legend>When should it fire?</legend>
${field('Condition', opControl, { name: 'op', error: errors.op })}
${field('Condition', opControl, { name: 'op', error: errors.op })}
${needsValue ? field('Value', html`<input type="text" id="${idOf('value')}" name="value" value="${fields.value}" maxlength="200">`, { name: 'value', error: errors.value, hint: 'The value to compare the reading against. Numbers are compared as numbers.' }) : ''}
${needsValue ? field('Value', html`<input type="text" id="${idOf('value')}" name="value" value="${fields.value}" maxlength="200">`, { name: 'value', error: errors.value, hint: 'The value to compare the reading against. Numbers are compared as numbers.' }) : ''}
${(template === 'price' || template === 'json') ? field('Hold for (seconds)', html`<input type="number" id="${idOf('for_sec')}" name="for_sec" value="${fields.for_sec}" min="0" max="2592000" inputmode="numeric">`, { name: 'for_sec', error: errors.for_sec, hint: 'Blank fires as soon as the condition is met; a number makes it wait that long and stay true.' }) : ''}
${(template === 'price' || template === 'json') ? field('Hold for (seconds)', html`<input type="number" id="${idOf('for_sec')}" name="for_sec" value="${fields.for_sec}" min="0" max="2592000" inputmode="numeric">`, { name: 'for_sec', error: errors.for_sec, hint: 'Blank fires as soon as the condition is met; a number makes it wait that long and stay true.' }) : ''}
</fieldset>`;
</fieldset>`;
let extractionBlock;
let extractionBlock;
- if (template === 'event') {
+ if (template === 'event' || template === 'lan') {
extractionBlock = '';
extractionBlock = '';
} else if (template === 'webhook') {
} else if (template === 'webhook') {
extractionBlock = html`<fieldset><legend>What should it read?</legend>
extractionBlock = html`<fieldset><legend>What should it read?</legend>
@@ -426,7 +465,13 @@ ${field('Value path', html`<input type="text" id="${idOf('value_path')}" name="v
@@ -426,7 +465,13 @@ ${field('Value path', html`<input type="text" id="${idOf('value_path')}" name="v
const pushed = template === 'webhook' || template === 'event';
const pushed = template === 'webhook' || template === 'event';
const chosen = eventChoice(fields.event) || EVENT_CHOICES[0];
const chosen = eventChoice(fields.event) || EVENT_CHOICES[0];
- const whatBlock = template === 'event'
+ const whatBlock = template === 'lan'
+ ? html`<fieldset><legend>What to watch</legend>
+${field('Name', html`<input type="text" id="${idOf('name')}" name="name" value="${fields.name}" maxlength="120" placeholder="e.g. The NAS">`, { name: 'name', error: errors.name, hint: 'Optional. Without one, Watch names it after the address.' })}
+${field('Your machine', html`<input type="text" id="${idOf('machine')}" name="machine" value="${fields.machine}" maxlength="30" placeholder="dev_…" required>`, { name: 'machine', error: errors.machine, hint: html`A computer on that network with OpenVibe Node and its computer plugin: copy its id from <a href="https://openvibe.bot/machines">openvibe.bot/machines</a>, where you also turn on “Let my watches check things from this computer”.` })}
+${field('Address and port', html`<input type="text" id="${idOf('lan_address')}" name="lan_address" value="${fields.lan_address}" maxlength="260" placeholder="192.168.1.20:80" required>`, { name: 'lan_address', error: errors.lan_address, hint: 'On your own network only (a private or local address): your machine checks that it accepts a connection.' })}
+</fieldset>`
+ : template === 'event'
? html`<fieldset><legend>What to watch</legend>
? html`<fieldset><legend>What to watch</legend>
${field('Name', html`<input type="text" id="${idOf('name')}" name="name" value="${fields.name}" maxlength="120" placeholder="e.g. When goosely goes live">`, { name: 'name', error: errors.name, hint: 'Optional. Without one, Watch names it after the event.' })}
${field('Name', html`<input type="text" id="${idOf('name')}" name="name" value="${fields.name}" maxlength="120" placeholder="e.g. When goosely goes live">`, { name: 'name', error: errors.name, hint: 'Optional. Without one, Watch names it after the event.' })}
${field('When', select('event', EVENT_CHOICES.map((e) => ({ value: e.value, label: e.label })), fields.event), { name: 'event', error: errors.event, hint: 'It fires the moment OpenVibe.Events delivers it, once per occurrence. Nothing is polled.' })}
${field('When', select('event', EVENT_CHOICES.map((e) => ({ value: e.value, label: e.label })), fields.event), { name: 'event', error: errors.event, hint: 'It fires the moment OpenVibe.Events delivers it, once per occurrence. Nothing is polled.' })}
@@ -18,6 +18,7 @@ const { createPoster } = require('./net/post');
@@ -18,6 +18,7 @@ const { createPoster } = require('./net/post');
const { createActorTasks } = require('./actor');
const { createActorTasks } = require('./actor');
const { createProjects, createRunJobs } = require('./run-jobs');
const { createProjects, createRunJobs } = require('./run-jobs');
const { createCodesTasks } = require('./codes-tasks');
const { createCodesTasks } = require('./codes-tasks');
+const { createMachines } = require('./machines');
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');
@@ -43,7 +44,7 @@ const { createApp } = require('./app');
@@ -43,7 +44,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, post: givenPost = null, actor: givenActor = null, run: givenRun = null, codes: givenCodes = null, projects: givenProjects = 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, actor: givenActor = null, run: givenRun = null, codes: givenCodes = null, projects: givenProjects = null, machines: givenMachines = 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 });
@@ -61,7 +62,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -61,7 +62,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const client = config.oauth.clientSecret
const client = config.oauth.clientSecret
? createClient({
? createClient({
tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),
tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),
- baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl, run: config.actions.runUrl, codes: config.actions.codesUrl }, fetch: fetchImpl,
+ baseUrls: { network: config.networkInternalUrl, actor: config.actions.actorUrl, run: config.actions.runUrl, codes: config.actions.codesUrl, bot: config.actions.botUrl }, fetch: fetchImpl,
})
})
: null;
: null;
// A watch that names a project: Network says whether its owner is an admin or owner of it, when it is saved (Run checks
// A watch that names a project: Network says whether its owner is an admin or owner of it, when it is saved (Run checks
@@ -75,7 +76,9 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -75,7 +76,9 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const fetcher = createFetcher({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.fetch.timeoutMs, maxBytes: config.fetch.maxBytes, maxRedirects: config.fetch.maxRedirects });
const fetcher = createFetcher({ guard, userAgent: config.fetch.userAgent, timeoutMs: config.fetch.timeoutMs, maxBytes: config.fetch.maxBytes, maxRedirects: config.fetch.maxRedirects });
const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
// Every carrier goes out through the spacer, so two watches over one host never hammer it.
// Every carrier goes out through the spacer, so two watches over one host never hammer it.
- const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
+ // A node source probes from its owner's own machine through OpenVibe.Bot (carriers/node.js).
+ const machines = givenMachines || (client ? createMachines(client, { baseUrl: config.actions.botUrl }) : null);
+ const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log, machines });
const notifications = client ? createNotificationsClient(client) : null;
const notifications = client ? createNotificationsClient(client) : null;
const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;
const actorTasks = client ? createActorTasks(client, { baseUrl: config.actions.actorUrl }) : null;
// Webhook actions leave through the same guard as every fetch (server/net/post.js).
// Webhook actions leave through the same guard as every fetch (server/net/post.js).
new file mode 100644
new file mode 100644
@@ -0,0 +1,26 @@
@@ -0,0 +1,26 @@
+'use strict';
+/**
+ * A person's own machine on OpenVibe.Bot, for a watch's `node` source (plan T18 step 7, contracts 0.167.0).
+ *
+ * const machines = createMachines(client, { baseUrl }); // client: openvibe-sdk/core createClient with Watch's token
+ * await machines.probe({ subjectId, machineId, probe }) → bot.command-result@1 ({ result, data?, reason?, code? })
+ *
+ * probe() POSTs one `probe` (bot.command@1: { target, timeout_s }) to Bot's machine API with Watch's own token
+ * (bot.computer.control, audience openvibe.bot) and the watch's owner in X-OV-Subject. Bot sends it only to that person's
+ * own machine, only where they turned on "Let my watches check things from this computer", and Watch may send nothing
+ * but probes. The machine itself reaches only its own network (private, loopback or link-local addresses). A refusal by
+ * Bot arrives as openvibe-sdk OpenVibeError (status, code); the machine's own refusal is result "nack" with a reason.
+ */
+
+function createMachines(client, { baseUrl }) {
+ async function probe({ subjectId, machineId, probe: value }) {
+ const body = { kind: 'probe', value: { target: value.target, ...(Number.isInteger(value.timeout_s) ? { timeout_s: value.timeout_s } : {}) }, ms: 45000 };
+ return await client.json({
+ service: 'bot', baseUrl, audience: 'openvibe.bot', method: 'POST', path: `/api/v1/machines/${encodeURIComponent(machineId)}/commands`,
+ headers: { 'X-OV-Subject': subjectId }, json: body, retries: 0, idempotencyKey: false,
+ });
+ }
+ return { probe };
+}
+
+module.exports = { createMachines };
@@ -114,6 +114,15 @@ function validateRequest(body, { create = false } = {}) {
@@ -114,6 +114,15 @@ function validateRequest(body, { create = false } = {}) {
if (cadence.every_sec < MIN_EVERY_SEC) throw new RegistryError(`cadence.every_sec must be at least ${MIN_EVERY_SEC} seconds (15 minutes)`);
if (cadence.every_sec < MIN_EVERY_SEC) throw new RegistryError(`cadence.every_sec must be at least ${MIN_EVERY_SEC} seconds (15 minutes)`);
}
}
if (PUSHED.has(kind) && cadence) throw new RegistryError(`a ${kind} source is pushed to and never polls: cadence must be null`);
if (PUSHED.has(kind) && cadence) throw new RegistryError(`a ${kind} source is pushed to and never polls: cadence must be null`);
+ if (kind === 'node') {
+ // A probe from the owner's own machine (carriers/node.js): the machine by its OpenVibe.Bot id, and one target on
+ // its network. The machine refuses anything that is not private, loopback or link-local.
+ if (!/^dev_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(body.source.node_id || ''))) throw new RegistryError('/source/node_id must be one of your machines on OpenVibe.Bot (dev_…)');
+ const probe = body.source.probe || {};
+ if (typeof probe.target !== 'string' || !/^(https?|tcp):\/\/[^\s]{3,2040}$/.test(probe.target)) throw new RegistryError('/source/probe/target must be http(s)://… or tcp://host:port on the machine\'s own network');
+ if (probe.timeout_s !== undefined && (!Number.isInteger(probe.timeout_s) || probe.timeout_s < 1 || probe.timeout_s > 30)) throw new RegistryError('/source/probe/timeout_s is 1 to 30');
+ for (const k of Object.keys(probe)) if (!['target', 'timeout_s'].includes(k)) throw new RegistryError(`/source/probe/${k} is not known (target, timeout_s)`);
+ }
// A HEAD source states no body, so only a document/record extraction can read one: refuse
// A HEAD source states no body, so only a document/record extraction can read one: refuse
// the misconfiguration here instead of recording `parse_error` on every check.
// the misconfiguration here instead of recording `parse_error` on every check.
if (body.source.method === 'HEAD' && !['json', 'jsonpath'].includes((body.extraction || {}).kind)) {
if (body.source.method === 'HEAD' && !['json', 'jsonpath'].includes((body.extraction || {}).kind)) {
@@ -84,12 +84,19 @@ function inWords(watch) {
@@ -84,12 +84,19 @@ function inWords(watch) {
what = e.value_path ? html`${extractionWords(watch.extraction)} of the newest item of the feed` : 'the newest item of the feed';
what = e.value_path ? html`${extractionWords(watch.extraction)} of the newest item of the feed` : 'the newest item of the feed';
} else if (source.kind === 'webhook') {
} else if (source.kind === 'webhook') {
what = path ? html`the value at <code>${path}</code> in a webhook delivery` : (e.kind === 'text' ? 'what a webhook delivery says' : 'a webhook delivery');
what = path ? html`the value at <code>${path}</code> in a webhook delivery` : (e.kind === 'text' ? 'what a webhook delivery says' : 'a webhook delivery');
+ } else if (source.kind === 'node') {
+ const target = (source.probe && source.probe.target) || '';
+ what = /^tcp:/.test(target) ? html`<b>${host(target)}</b>, seen from your computer,` : html`${extractionWords(watch.extraction)} on <b>${host(target)}</b>, seen from your computer,`;
} else if (source.kind === 'event') {
} else if (source.kind === 'event') {
what = path ? html`the value at <code>${path}</code> of a <b>${source.pattern}</b> event` : html`a <b>${source.pattern}</b> event`;
what = path ? html`the value at <code>${path}</code> of a <b>${source.pattern}</b> event` : html`a <b>${source.pattern}</b> event`;
} else {
} else {
what = html`${extractionWords(watch.extraction)} on <b>${host(source.url)}</b>`;
what = html`${extractionWords(watch.extraction)} on <b>${host(source.url)}</b>`;
}
}
const every = watch.cadence ? `, checked ${everyWords(watch.cadence.every_sec)}` : '';
const every = watch.cadence ? `, checked ${everyWords(watch.cadence.every_sec)}` : '';
+ const c = watch.condition || {};
+ if (source.kind === 'node' && /^tcp:/.test((source.probe && source.probe.target) || '') && c.op === 'eq' && c.value === 'down') {
+ return html`Tell me when <b>${host(source.probe.target)}</b> stops answering, seen from your computer${every}.`;
+ }
return html`Tell me when ${what} ${conditionWords(watch.condition)}${every}.`;
return html`Tell me when ${what} ${conditionWords(watch.condition)}${every}.`;
}
}
@@ -176,12 +176,12 @@ t('a manual check runs one check and answers the released check-run; observation
@@ -176,12 +176,12 @@ t('a manual check runs one check and answers the released check-run; observation
// a watch whose carrier is a later step answers 422 and never fetches
// a watch whose carrier is a later step answers 422 and never fetches
const later = await api('POST', '/api/v1/watches', {
const later = await api('POST', '/api/v1/watches', {
token: ME,
token: ME,
- body: mine({ source: { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, cadence: { every_sec: 900 }, extraction: { kind: 'json' } }),
+ body: mine({ source: { kind: 'run', runtime: 'browser', url: 'https://example.com/', steps: [{ open: 'https://example.com/' }] }, cadence: { every_sec: 900 }, extraction: { kind: 'json' } }),
});
});
const refused = await api('POST', `/api/v1/watches/${later.body.watch.id}/check`, { token: ME });
const refused = await api('POST', `/api/v1/watches/${later.body.watch.id}/check`, { token: ME });
assert.strictEqual(refused.status, 422);
assert.strictEqual(refused.status, 422);
assert.strictEqual(refused.body.code, 'watch.carrier_unavailable');
assert.strictEqual(refused.body.code, 'watch.carrier_unavailable');
- assert.match(refused.body.detail, /step 7/);
+ assert.match(refused.body.detail, /step 6/);
// a paused watch cannot be checked by hand either
// a paused watch cannot be checked by hand either
await api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });
await api('POST', `/api/v1/watches/${watch.id}/pause`, { token: ME });
const paused = await api('POST', `/api/v1/watches/${watch.id}/check`, { token: ME });
const paused = await api('POST', `/api/v1/watches/${watch.id}/check`, { token: ME });
@@ -186,15 +186,14 @@ t('a pattern that hits its deadline is a recorded check error, never a hang', as
@@ -186,15 +186,14 @@ t('a pattern that hits its deadline is a recorded check error, never a hang', as
price = '19.99';
price = '19.99';
});
});
-t('a node or run source is accepted but never fetched (steps 6-7); event and webhook sources wait to be pushed to', async () => {
+t('a run source is accepted but never fetched (step 6; node sources probe through Bot since step 7, test/node-carrier.test.js)', async () => {
const cases = [
const cases = [
- ['node', { kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { path: '/tmp/x' } }, { op: 'exists' }],
['run', { kind: 'run', runtime: 'browser', url: `${web.origin}/price`, steps: [{ goto: 'https://example.org' }] }, { op: 'exists' }],
['run', { kind: 'run', runtime: 'browser', url: `${web.origin}/price`, steps: [{ goto: 'https://example.org' }] }, { op: 'exists' }],
];
];
const before = web.requests.length;
const before = web.requests.length;
for (const [kind, source, condition] of cases) {
for (const [kind, source, condition] of cases) {
- // an event or a webhook is pushed to and never polls; node and run sources poll through
- // OpenVibe.Node and OpenVibe.Run, so they carry a cadence — and still nothing is fetched
+ // an event or a webhook is pushed to and never polls; a run source polls through OpenVibe.Run, so it carries a
+ // cadence — and still nothing is fetched
const cadence = { every_sec: 900 };
const cadence = { every_sec: 900 };
const w = await create(person, { source, cadence, extraction: { kind: 'text' }, condition });
const w = await create(person, { source, cadence, extraction: { kind: 'text' }, condition });
assert.strictEqual((await row(w.id)).status, 'active');
assert.strictEqual((await row(w.id)).status, 'active');
@@ -202,7 +201,7 @@ t('a node or run source is accepted but never fetched (steps 6-7); event and web
@@ -202,7 +201,7 @@ t('a node or run source is accepted but never fetched (steps 6-7); event and web
assert.strictEqual(out.run.state, 'skipped', `${kind}: ${out.run.detail}`);
assert.strictEqual(out.run.state, 'skipped', `${kind}: ${out.run.detail}`);
assert.strictEqual(out.run.error_code, 'watch.carrier_unavailable');
assert.strictEqual(out.run.error_code, 'watch.carrier_unavailable');
assert.strictEqual(out.run.carrier, kind);
assert.strictEqual(out.run.carrier, kind);
- assert.match(out.run.detail, /step [67]/);
+ assert.match(out.run.detail, /step 6/);
assert.strictEqual(out.unavailable.code, 'watch.carrier_unavailable');
assert.strictEqual(out.unavailable.code, 'watch.carrier_unavailable');
assert.strictEqual((await observations(w.id)).length, 0);
assert.strictEqual((await observations(w.id)).length, 0);
const [fail] = (await payloads('watch.check.failed')).slice(-1);
const [fail] = (await payloads('watch.check.failed')).slice(-1);
@@ -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, post = null, actor = null, run = null, codes = null, projects = null } = {}) {
+async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null, post = null, actor = null, run = null, codes = null, projects = null, machines = 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 } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(codes ? { codes } : {}), ...(projects ? { projects } : {}), ...(now ? { now } : {}) });
+ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(post ? { post } : {}), ...(actor ? { actor } : {}), ...(run ? { run } : {}), ...(codes ? { codes } : {}), ...(projects ? { projects } : {}), ...(machines ? { machines } : {}), ...(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,98 @@
@@ -0,0 +1,98 @@
+'use strict';
+/**
+ * The node carrier (plan T18 step 7, contracts 0.167.0): a watch checks something on its owner's own network from
+ * their machine, through OpenVibe.Bot's probe. The source is checked when it is saved (a dev_ machine, an
+ * http(s)/tcp target, nothing else); the probe goes to Bot as the owner; an http answer's body is extracted like a page,
+ * a tcp answer's { open, ms } is a JSON document; Bot's refusal and the machine's are recorded failures; an agent-owned
+ * watch is never probed.
+ */
+const assert = require('assert');
+const { boot, suite, newUser } = require('./helpers');
+
+const t = suite('node-carrier');
+const MACHINE = 'dev_01M4XTHNFFAWWZQHB4XTMSN1YB';
+const PERSON = { sub: newUser() };
+const calls = [];
+let reply = null;
+const machines = {
+ probe: async (args) => {
+ calls.push(args);
+ if (reply instanceof Error) throw reply;
+ return reply;
+ },
+};
+let svc;
+
+const create = async (source, extraction = { kind: 'json', value_path: 'up' }, owner = PERSON) => (await svc.registry.create({
+ name: 'nas', source, cadence: { every_sec: 900 }, extraction, condition: { op: 'eq', value: 'never-matches' }, action: [{ kind: 'notification' }],
+}, owner)).id;
+const lastRun = async (id) => (await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid DESC').get(id));
+const lastValue = async (id) => {
+ const o = await svc.db.prepare('SELECT * FROM observations WHERE watch_id = ? ORDER BY rid DESC').get(id);
+ return o ? (typeof o.value === 'string' ? JSON.parse(o.value) : o.value) : undefined;
+};
+
+t('boot with a stand-in for Bot\'s machine API', async () => {
+ svc = await boot({ machines });
+});
+
+t('the source is checked when it is saved', async () => {
+ const bad = async (source, re) => {
+ await assert.rejects(create(source), (e) => re.test(e.message));
+ };
+ await bad({ kind: 'node', node_id: 'nod_01J0000000000000000000000', probe: { target: 'http://192.168.1.20/' } }, /node_id/);
+ await bad({ kind: 'node', node_id: MACHINE, probe: { target: 'ftp://192.168.1.20/' } }, /target/);
+ await bad({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/', method: 'POST' } }, /method is not known/);
+ await bad({ kind: 'node', node_id: MACHINE, probe: { target: 'tcp://192.168.1.20:22', timeout_s: 99 } }, /timeout_s/);
+});
+
+t('an http probe goes to Bot as the owner and its body is extracted like a page', async () => {
+ reply = { result: 'ack', data: { status: 200, content_type: 'application/json', etag: '"v1"', body: '{"up": true}', ms: 4 } };
+ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/status', timeout_s: 5 } });
+ await svc.check.run(id, { trigger: 'manual' });
+ assert.deepStrictEqual(calls.at(-1), { subjectId: PERSON.sub, machineId: MACHINE, probe: { target: 'http://192.168.1.20/status', timeout_s: 5 } });
+ const run = await lastRun(id);
+ assert.deepStrictEqual([run.carrier, run.http_status], ['node', 200]);
+ assert.strictEqual(await lastValue(id), true);
+ reply = { result: 'ack', data: { status: 503, body: 'busy', ms: 3 } };
+ await svc.check.run(id, { trigger: 'manual' });
+ const failed = await lastRun(id);
+ assert.deepStrictEqual([failed.state, failed.error_code], ['http_error', 'http_503']);
+});
+
+t('a tcp probe states { open, ms } as a JSON document', async () => {
+ reply = { result: 'ack', data: { open: false, ms: 2001, error: 'timed out' } };
+ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'tcp://192.168.1.5:22' } }, { kind: 'json', value_path: 'open' });
+ await svc.check.run(id, { trigger: 'manual' });
+ assert.strictEqual((await lastRun(id)).state !== 'http_error', true);
+ assert.strictEqual(await lastValue(id), false);
+});
+
+t('Bot\'s refusal and the machine\'s are recorded failures with their reasons', async () => {
+ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/' } });
+ reply = Object.assign(new Error('the owner has not let their watches check things from this computer'), { code: 'bot.agent_not_allowed', status: 403 });
+ await svc.check.run(id, { trigger: 'manual' });
+ let run = await lastRun(id);
+ assert.deepStrictEqual([run.state, run.error_code], ['http_error', 'bot.agent_not_allowed']);
+ assert.match(run.detail, /watches/);
+ reply = { result: 'nack', reason: 'a probe reaches this machine\'s own network only' };
+ await svc.check.run(id, { trigger: 'manual' });
+ run = await lastRun(id);
+ assert.deepStrictEqual([run.state, run.error_code], ['http_error', 'node_nack']);
+ reply = Object.assign(new Error('the machine is offline'), { code: 'bot.device_offline', status: 409 });
+ await svc.check.run(id, { trigger: 'manual' });
+ assert.strictEqual((await lastRun(id)).state, 'timeout');
+});
+
+t('a watch owned by an agent is never probed', async () => {
+ const before = calls.length;
+ const id = await create({ kind: 'node', node_id: MACHINE, probe: { target: 'http://192.168.1.20/' } }, undefined, { sub: 'agt_01M4XTHNFFAWWZQHB4XTMSN1YB' });
+ await svc.check.run(id, { trigger: 'manual' });
+ const run = await lastRun(id);
+ assert.deepStrictEqual([run.state, run.error_code], ['disabled', 'node_person_only']);
+ assert.strictEqual(calls.length, before);
+});
+
+t('done', async () => { await svc.stop(); });
+
+t.run();
@@ -226,6 +226,33 @@ t('the event template: a channel going live (or a blog post) wakes the watch, wi
@@ -226,6 +226,33 @@ t('the event template: a channel going live (or a blog post) wakes the watch, wi
assert.strictEqual(bad.status, 422, 'only the offered public events');
assert.strictEqual(bad.status, 422, 'only the offered public events');
});
});
+t('the network template: a device at home stops answering, checked from the person\'s own machine', async () => {
+ // A second person, so this test's writes do not count against the first one's budget of 30 a minute.
+ const owner = svc.network.addUser('lan-owner');
+ const own = await svc.signIn(owner);
+ const create = (fields) => svc.get('/watches/new', { cookie: own, form: fields });
+ const ME = userToken(owner.subject);
+ const session = own;
+ const form = await svc.get('/watches/new?template=lan', { cookie: session });
+ assert.strictEqual(form.status, 200);
+ assert.match(form.text, /name="machine"/);
+ assert.match(form.text, /name="lan_address"/);
+ assert.match(form.text, /openvibe\.bot\/machines/);
+ const made = await create({ template: 'lan', name: 'The NAS', machine: 'dev_01M4XTHNFFAWWZQHB4XTMSN1YB', lan_address: '192.168.1.20:5000', cadence: '3600', notify: 'on' });
+ assert.strictEqual(made.status, 303, made.text);
+ const w = (await request(svc.base, 'GET', `/api/v1/watches/${indexOf(made)}`, { token: ME })).body.watch;
+ assert.deepStrictEqual(w.source, { kind: 'node', node_id: 'dev_01M4XTHNFFAWWZQHB4XTMSN1YB', probe: { target: 'tcp://192.168.1.20:5000', timeout_s: 10 } });
+ assert.deepStrictEqual([w.extraction, w.condition], [{ kind: 'json', value_path: 'status' }, { op: 'eq', value: 'down' }]);
+ const page = await svc.get(`/watches/${indexOf(made)}`, { cookie: session });
+ assert.match(page.text, /192\.168\.1\.20:5000<\/b> stops answering, seen from your computer/);
+ const edit = await svc.get(`/watches/${indexOf(made)}/edit`, { cookie: session });
+ assert.match(edit.text, /value="192\.168\.1\.20:5000"/);
+ const bad = await create({ template: 'lan', machine: 'nope', lan_address: 'not an address', cadence: '3600' });
+ assert.strictEqual(bad.status, 422);
+ assert.match(bad.text, /openvibe\.bot\/machines \(dev_/);
+ assert.match(bad.text, /address and port/);
+});
+
t('validation errors re-render the form with the registry\'s message and the person\'s input', async () => {
t('validation errors re-render the form with the registry\'s message and the person\'s input', async () => {
const badUrl = await create({ template: 'page', name: 'Keep me', url: 'not-a-url', selector: '.x', cadence: '3600' });
const badUrl = await create({ template: 'page', name: 'Keep me', url: 'not-a-url', selector: '.x', cadence: '3600' });
assert.strictEqual(badUrl.status, 422);
assert.strictEqual(badUrl.status, 422);