openvibers/openvibe.events Commits

Developer apps cannot disrupt first-party event flow

OP OpenVibers committed 23 Sept 2026 · cfbd21c398 · 7 files, +202 −28

- Loop guard: a first-party publish's chain depth counts first-party events only; an app publish
  counts first-party events plus its own project and environment. Before, an app that read a public
  event's trace_id could publish 8 repeats into that trace and every later first-party event in it
  was refused with 409 events.loop_detected (and dropped by the producer's outbox as permanent).
- App webhook delivery: the attempt settles on the response status and drops the body, and a
  deadline bounds the whole attempt; before, an endpoint dripping headers or body kept the attempt
  (and its worker slot) open indefinitely, since the socket timeout is only an idle timer.
- Worker: developer-app deliveries use at most EVENTS_APP_MAX_INFLIGHT (default half) of the
  in-flight slots, so critical-priority app deliveries to slow endpoints cannot starve first-party
  subscriptions.

Changes

ddiff --git a/.env.example b/.env.example
iindex e79f0f4..fe24d6a 100644
---- a/.env.example
++++ b/.env.example
@@@ -45,6 +45,8 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
  #EVENTS_WORKER=on
  #EVENTS_WORKER_INTERVAL_MS=500
  #EVENTS_MAX_INFLIGHT=20
++# Of which developer-app deliveries may use at most (default: half of EVENTS_MAX_INFLIGHT)
++#EVENTS_APP_MAX_INFLIGHT=10
  #EVENTS_DELIVERY_TIMEOUT_MS=10000
  #EVENTS_MAX_ATTEMPTS=8
  #EVENTS_BACKOFF_MS=1000,5000,30000,120000,600000,3600000
ddiff --git a/README.md b/README.md
iindex 6d89907..d05b786 100644
---- a/README.md
++++ b/README.md
@@@ -50,7 +50,7 @@ Errors are RFC 9457 problem+json (`openvibe-contracts` `http.problem`) with a st
  - `source` must be the calling service (`svc:live` publishes `source: "live"`), and `event_type` must start with a prefix that source owns (`live.*`, `media.*`, `network.*`, `community.*`, `chat.*`, `openre.*`, `billing.*`, `tips.*`, `vip.*`, `ai.*`, `games.*`, `tools.*`, `codes.*`, `host.*`; `EVENTS_SOURCE_PREFIXES` overrides).
  - Idempotent on `event_id`: a repeat answers `200 { event_id, seq, duplicate: true }` and is never stored twice (even after retention, for `EVENTS_RECEIPT_RETENTION_DAYS`).
  - Missing `trace_id` is taken from the request's `traceparent`; `priority` defaults to `important`, `visibility` to `internal`.
--- Loop guard: an event whose trace already carries 8 hops (the cross-service chain depth, or the same source/type/subject repeating in the trace) is refused with `409 events.loop_detected`.
++- Loop guard: an event whose trace already carries 8 hops (the cross-service chain depth, or the same source/type/subject repeating in the trace) is refused with `409 events.loop_detected`. The chain depth of a first-party publish counts first-party events only (developer-app events in the same trace never count toward it), so an app cannot poison a trace it has seen.
  - Each accepted event gets a global `seq` and is committed with one delivery row per matching subscription before anything is sent.
  
  ## Subscriptions and delivery
ddiff --git a/server/config.js b/server/config.js
iindex d294942..fc622dd 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -45,6 +45,7 @@ function load(env = process.env) {
      const nodeEnv = env.NODE_ENV || 'development';
      const isProduction = nodeEnv === 'production';
      const port = int(env.PORT, 4300);
++    const maxInflight = int(env.EVENTS_MAX_INFLIGHT, 20);
      return {
          port,
          host: env.HOST || '127.0.0.1',
@@@ -77,7 +78,9 @@ function load(env = process.env) {
          worker: {
              enabled: env.EVENTS_WORKER !== 'off',
              intervalMs: int(env.EVENTS_WORKER_INTERVAL_MS, 500),
--            maxInflight: int(env.EVENTS_MAX_INFLIGHT, 20),
++            maxInflight,
++            // Developer-app deliveries share at most this many of the slots (default: half).
++            maxAppInflight: Math.max(1, Math.min(maxInflight, int(env.EVENTS_APP_MAX_INFLIGHT, Math.floor(maxInflight / 2)))),
              timeoutMs: int(env.EVENTS_DELIVERY_TIMEOUT_MS, 10000),
              maxAttempts: int(env.EVENTS_MAX_ATTEMPTS, 8),
              // Wait after attempt n fails = backoffMs[n-1] (the last value repeats).
ddiff --git a/server/egress.js b/server/egress.js
iindex 5e9ef2f..655a6b3 100644
---- a/server/egress.js
++++ b/server/egress.js
@@@ -119,14 +119,15 @@ async function checkAppEndpoint(endpoint, { lookup } = {}) {
   * POST to an app endpoint through the guard. Returns { status } or throws (err.permanent = true
   * when the endpoint itself is refused). `requestImpl`/`isAllowed` exist for tests only.
   */
--function createGuardedPost({ lookup = dns.lookup, requestImpl = https.request, isAllowed = isPublicAddress, maxResponseBytes = 64 * 1024 } = {}) {
++function createGuardedPost({ lookup = dns.lookup, requestImpl = https.request, isAllowed = isPublicAddress } = {}) {
      return function guardedPost(endpoint, { headers, body, timeoutMs = 10000 }) {
          const p = parseAppEndpoint(endpoint);
          if (!p.ok) return Promise.reject(Object.assign(new Error(`endpoint not allowed: ${p.reason}`), { permanent: true }));
          if (p.literal && !isAllowed(p.host)) return Promise.reject(Object.assign(new Error('endpoint not allowed: not a public address'), { permanent: true }));
          return new Promise((resolve, reject) => {
              let settled = false;
--            const done = (fn, v) => { if (!settled) { settled = true; fn(v); } };
++            let deadline = null;
++            const done = (fn, v) => { if (!settled) { settled = true; clearTimeout(deadline); fn(v); } };
              // The socket's own lookup: the address we connect to is the one we checked.
              const guardedLookup = (hostname, opts, cb) => {
                  if (typeof opts === 'function') { cb = opts; opts = {}; }
@@@ -150,13 +151,16 @@ function createGuardedPost({ lookup = dns.lookup, requestImpl = https.request, i
                  servername: p.literal ? undefined : p.host,
                  timeout: timeoutMs,
              }, (res) => {
--                let n = 0;
--                res.on('data', (c) => { n += c.length; if (n > maxResponseBytes) res.destroy(); });
--                res.on('end', () => done(resolve, { status: res.statusCode }));
--                res.on('close', () => done(resolve, { status: res.statusCode }));
--                res.on('error', () => done(resolve, { status: res.statusCode }));
++                // The status is all a delivery needs: settle now and drop the body, so an endpoint
++                // that drips its response cannot keep this attempt (and its worker slot) open.
++                done(resolve, { status: res.statusCode });
++                res.destroy();
              });
--            req.on('timeout', () => { req.destroy(Object.assign(new Error('timeout'), { name: 'TimeoutError' })); });
++            const timeout = () => { req.destroy(Object.assign(new Error('timeout'), { name: 'TimeoutError' })); };
++            // `timeout` above is only the socket's idle timer; this bounds the whole attempt, however
++            // slowly the endpoint trickles its status line and headers.
++            deadline = setTimeout(timeout, timeoutMs);
++            req.on('timeout', timeout);
              req.on('error', (err) => {
                  if (err.code === 'EOV_NOT_PUBLIC' || (err.cause && err.cause.code === 'EOV_NOT_PUBLIC')) err.permanent = true;
                  done(reject, err);
ddiff --git a/server/store.js b/server/store.js
iindex 3d8d353..fdb05c5 100644
---- a/server/store.js
++++ b/server/store.js
@@@ -193,7 +193,12 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
          nextSeq: db.prepare("UPDATE sequences SET value = value + 1 WHERE name = 'events' RETURNING value"),
          lastSeq: db.prepare("SELECT value FROM sequences WHERE name = 'events'"),
          minSeq: db.prepare('SELECT MIN(seq) AS s FROM events'),
--        chainHops: db.prepare('SELECT MAX(hops) AS h FROM events WHERE trace_id = ? AND source != ?'),
++        // Hop depth counts only rows the publisher's own tenancy can have caused: a first-party
++        // publish ignores app events (an app that saw a public trace_id cannot poison that trace), an
++        // app publish counts first-party rows plus its own project and environment.
++        chainHops: db.prepare('SELECT MAX(hops) AS h FROM events WHERE trace_id = ? AND source != ? AND project_id IS NULL'),
++        chainHopsApp: db.prepare(`SELECT MAX(hops) AS h FROM events WHERE trace_id = ? AND source != ?
++            AND (project_id IS NULL OR (project_id = ? AND env = ?))`),
          selfRepeats: db.prepare(`SELECT COUNT(*) AS n FROM events WHERE trace_id = ? AND source = ? AND event_type = ?
              AND subject_type = ? AND subject_id = ? AND (request_id IS NULL OR request_id != ?)`),
          insertEvent: db.prepare(`INSERT INTO events (id, seq, event_type, version, source, actor, subject_type, subject_id,
@@@ -252,7 +257,10 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
              }
              // Loop guard: depth of the cross-service chain in this trace, and how often this exact
              // (source, type, subject) already happened in the trace from other publish calls.
--            const chain = (q.chainHops.get(env.trace_id, env.source).h || 0) + 1;
++            const chainRow = project
++                ? q.chainHopsApp.get(env.trace_id, env.source, project.projectId, project.env)
++                : q.chainHops.get(env.trace_id, env.source);
++            const chain = (chainRow.h || 0) + 1;
              const repeats = q.selfRepeats.get(env.trace_id, env.source, env.event_type, env.subject.type, env.subject.id, requestId || '').n + 1;
              const hops = Math.max(chain, repeats);
              if (hops > maxHops) {
@@@ -416,12 +424,14 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8 } = {}
       * Next due delivery per subscription (at most one each, so per-subscription concurrency stays 1),
       * ordered by priority class, then seq. `busy` are subscription ids with a delivery in flight.
       */
--    function dueDeliveries(now, limit, busy = []) {
++    function dueDeliveries(now, limit, busy = [], { firstPartyOnly = false } = {}) {
++        // app_project_id: set for developer-app subscriptions (the worker caps their share of slots).
          const rows = db.prepare(`
              SELECT * FROM (
--                SELECT d.*, ROW_NUMBER() OVER (PARTITION BY d.subscription_id ORDER BY d.priority, d.seq) AS rn
++                SELECT d.*, s.project_id AS app_project_id, ROW_NUMBER() OVER (PARTITION BY d.subscription_id ORDER BY d.priority, d.seq) AS rn
                  FROM deliveries d JOIN subscriptions s ON s.id = d.subscription_id
                  WHERE d.status IN ('pending', 'failed') AND d.next_attempt_at <= ? AND s.enabled = 1
++                ${firstPartyOnly ? 'AND s.project_id IS NULL' : ''}
              ) WHERE rn = 1 ORDER BY priority, seq LIMIT ?`).all(now, limit + busy.length);
          const skip = new Set(busy);
          return rows.filter(r => !skip.has(r.subscription_id)).slice(0, limit);
ddiff --git a/server/worker.js b/server/worker.js
iindex 667da27..c7ea1ff 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -14,7 +14,8 @@
   * address, redirects refused. A refused address is permanent (dead at once; replay requeues it).
   *
   * Scheduling: a setTimeout chain (never overlapping ticks); priority classes critical > important >
-- * low, then seq; one delivery in flight per subscription; at most maxInflight overall.
++ * low, then seq; one delivery in flight per subscription; at most maxInflight overall, of which at
++ * most maxAppInflight to developer-app endpoints (so apps can never hold every slot).
   */
  const crypto = require('crypto');
  const { sign } = require('../lib/client');
@@@ -27,6 +28,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
      const opts = config.worker;
      const busy = new Set();          // subscription ids with a delivery in flight
      const inflight = new Set();      // promises
++    let appInflight = 0;             // of which to developer-app endpoints (at most opts.maxAppInflight)
      let timer = null;
      let running = false;
      let lastTickAt = null;
@@@ -103,19 +105,36 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
          lastTickAt = Date.now();
          const room = opts.maxInflight - inflight.size;
          if (room <= 0) return 0;
--        const due = store.dueDeliveries(clock.now(), room, [...busy]);
--        for (const d of due) {
--            busy.add(d.subscription_id);
--            const p = send(d)
--                .catch(err => log.error(`[worker] delivery ${d.event_id} -> ${d.subscription_id} crashed: ${err.stack || err}`))
--                .finally(() => {
--                    busy.delete(d.subscription_id);
--                    inflight.delete(p);
--                    if (running) schedule(0);
--                });
--            inflight.add(p);
--        }
--        return due.length;
++        let started = 0;
++        let skippedApp = false;
++        const startAll = (due) => {
++            for (const d of due) {
++                if (started >= room) return;
++                const isApp = Boolean(d.app_project_id);
++                if (isApp && appInflight >= opts.maxAppInflight) { skippedApp = true; continue; }
++                start(d, isApp);
++                started++;
++            }
++        };
++        startAll(store.dueDeliveries(clock.now(), room, [...busy], { firstPartyOnly: appInflight >= opts.maxAppInflight }));
++        // App deliveries (which can outrank first-party ones by priority) filled the app share: the
++        // rest of the room goes to first-party subscriptions.
++        if (skippedApp && started < room) startAll(store.dueDeliveries(clock.now(), room - started, [...busy], { firstPartyOnly: true }));
++        return started;
++    }
++
++    function start(d, isApp) {
++        busy.add(d.subscription_id);
++        if (isApp) appInflight++;
++        const p = send(d)
++            .catch(err => log.error(`[worker] delivery ${d.event_id} -> ${d.subscription_id} crashed: ${err.stack || err}`))
++            .finally(() => {
++                busy.delete(d.subscription_id);
++                if (isApp) appInflight--;
++                inflight.delete(p);
++                if (running) schedule(0);
++            });
++        inflight.add(p);
      }
  
      function schedule(ms) {
ddiff --git a/test/apps-abuse.test.js b/test/apps-abuse.test.js
nnew file mode 100644
iindex 0000000..42797de
---- /dev/null
++++ b/test/apps-abuse.test.js
@@@ -0,0 +1,136 @@
++'use strict';
++// Developer apps must not be able to hurt first-party event flow: the loop guard's hop count is
++// kept per tenancy (an app cannot poison a first-party trace), and app deliveries cannot hold every
++// delivery slot (bounded time per attempt, a capped share of the in-flight slots).
++const assert = require('assert');
++const net = require('net');
++const nodeHttp = require('http');
++const { ids } = require('openvibe-contracts');
++const apps = require('../server/apps');
++const { createGuardedPost, isPublicAddress } = require('../server/egress');
++const { boot, request, serviceToken, appToken, envelope, subscriber, suite, sleep } = require('./helpers');
++
++const t = suite('apps-abuse');
++const ALL = ['events.app.publish', 'events.app.read', 'events.app.subscribe'];
++const prj = `prj_${ids.ulid()}`;
++const app = ids.newId('app');
++const key = apps.projectKey(prj);
++const tok = () => appToken({ appId: app, projectId: prj, env: 'sandbox', cap: ALL });
++const appEvent = (over = {}) => envelope('x', {
++    source: apps.appSource(app), event_type: `app.${key}.poke`, actor: { type: 'app', id: app }, subject: { type: 'thing', id: '1' }, ...over,
++});
++
++t('loop guard: an app cannot poison a first-party trace it has seen', async () => {
++    const h = await boot({ env: { EVENTS_APP_SANDBOX_PUBLISH_PER_MINUTE: '0' } });
++    try {
++        const live = serviceToken('live', ['events.event.publish']);
++        const media = serviceToken('media', ['events.event.publish']);
++        const trace = 'ab'.repeat(16);
++        // A public first-party event: every app can read it, trace_id included.
++        let r = await request(h.base, 'POST', '/api/v1/events', { token: live, body: envelope('live', { trace_id: trace, visibility: 'public' }) });
++        assert.strictEqual(r.status, 201, r.text);
++        // The app republishes into that trace until its own hop count reaches the maximum (8).
++        for (let i = 0; i < 8; i++) {
++            r = await request(h.base, 'POST', '/api/v1/events', { token: tok(), body: appEvent({ trace_id: trace }) });
++            assert.strictEqual(r.status, 201, r.text);
++        }
++        // A first-party consumer reacting to the live event in the same trace must still publish.
++        r = await request(h.base, 'POST', '/api/v1/events', {
++            token: media, body: envelope('media', { event_type: 'media.recording.started', trace_id: trace }),
++        });
++        assert.strictEqual(r.status, 201, `first-party publish refused: ${r.text}`);
++        // And a first-party loop is still caught.
++        let last;
++        for (let i = 0; i < 9; i++) {
++            last = await request(h.base, 'POST', '/api/v1/events', {
++                token: media, body: envelope('media', { event_type: 'media.recording.started', trace_id: trace }),
++            });
++        }
++        assert.strictEqual(last.status, 409); assert.strictEqual(last.body.code, 'events.loop_detected');
++    } finally {
++        await h.stop();
++    }
++});
++
++// ── Slow endpoints ──
++function fakeLookup(host, opts, cb) {
++    if (typeof opts === 'function') { cb = opts; opts = {}; }
++    return opts.all ? cb(null, [{ address: '127.0.0.1', family: 4 }]) : cb(null, '127.0.0.1', 4);
++}
++
++/** A raw TCP server that answers slowly: 'headers' drips header lines, 'body' drips the body. */
++async function dripServer(mode) {
++    const sockets = new Set();
++    const server = net.createServer((sock) => {
++        sockets.add(sock);
++        sock.on('error', () => {});
++        sock.once('data', () => {
++            if (mode === 'headers') sock.write('HTTP/1.1 200 OK\r\n');
++            else sock.write('HTTP/1.1 200 OK\r\nContent-Length: 100000\r\n\r\n');
++            const timer = setInterval(() => { if (!sock.destroyed) sock.write(mode === 'headers' ? 'X-Drip: 1\r\n' : 'x'); }, 50);
++            sock.on('close', () => { clearInterval(timer); sockets.delete(sock); });
++        });
++    });
++    await new Promise(r => server.listen(0, '127.0.0.1', r));
++    return {
++        port: server.address().port,
++        close: () => new Promise(r => { for (const s of sockets) s.destroy(); server.close(() => r()); }),
++    };
++}
++
++t('egress: a slow endpoint cannot hold a delivery past its timeout', async () => {
++    for (const mode of ['headers', 'body']) {
++        const srv = await dripServer(mode);
++        try {
++            const post = createGuardedPost({
++                lookup: fakeLookup,
++                requestImpl: (o, cb) => nodeHttp.request({ ...o, port: srv.port }, cb),
++                isAllowed: (ip) => ip === '127.0.0.1' || isPublicAddress(ip),
++            });
++            const started = Date.now();
++            const outcome = await Promise.race([
++                post('https://hooks.example.com/x', { headers: {}, body: '{}', timeoutMs: 300 }).then(v => ({ v }), e => ({ e })),
++                sleep(2000).then(() => ({ hung: true })),
++            ]);
++            assert.ok(!outcome.hung, `${mode}: the delivery was still in flight after 2 s (timeout 300 ms)`);
++            assert.ok(Date.now() - started < 1500, mode);
++        } finally {
++            await srv.close();
++        }
++    }
++});
++
++t('worker: app deliveries never take every in-flight slot', async () => {
++    const pending = [];
++    const appPost = () => new Promise((resolve) => pending.push(resolve));   // an app endpoint that never answers
++    const h = await boot({
++        appPost, dnsLookup: (host, opts, cb) => cb(null, [{ address: '93.184.216.34', family: 4 }]),
++        env: { EVENTS_MAX_INFLIGHT: '2', EVENTS_APP_SANDBOX_PUBLISH_PER_MINUTE: '0', EVENTS_APP_SANDBOX_MAX_SUBSCRIPTIONS: '0' },
++    });
++    const firstParty = await subscriber();
++    try {
++        for (const p of ['a', 'b', 'c']) {
++            const r = await request(h.base, 'POST', '/api/v1/subscriptions', { token: tok(), body: { topic_pattern: `app.${key}.*`, endpoint: `https://hooks.example.com/${p}` } });
++            assert.strictEqual(r.status, 201, r.text);
++        }
++        const s = await request(h.base, 'POST', '/api/v1/subscriptions', { token: serviceToken('chat', ['events.subscription.manage']), body: { topic_pattern: 'live.*', endpoint: firstParty.url } });
++        assert.strictEqual(s.status, 201, s.text);
++        // Critical app events are queued before the first-party one.
++        for (let i = 0; i < 3; i++) {
++            const r = await request(h.base, 'POST', '/api/v1/events', { token: tok(), body: appEvent({ priority: 'critical' }) });
++            assert.strictEqual(r.status, 201, r.text);
++        }
++        const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.publish']), body: envelope('live') });
++        assert.strictEqual(r.status, 201);
++        h.worker.dispatch();
++        for (let i = 0; i < 40 && !firstParty.calls.length; i++) await sleep(25);
++        assert.strictEqual(firstParty.calls.length, 1, 'the first-party delivery got a slot');
++        assert.ok(pending.length >= 1, 'app deliveries still run');
++    } finally {
++        for (const resolve of pending) resolve({ status: 204 });
++        await firstParty.close();
++        await h.stop();
++    }
++});
++
++t.run();