openvibers/openvibe.live Commits

Live calls Network as a service principal: coins and notification pushes use a scoped client-credentials token, the internal key stays as fallback

OP OpenVibers committed 22 Sept 2026 · 90aaaee262 · 7 files, +255 −11

Roadmap Wave 1 (ADR-003), openvibe-contracts v0.2.0 (pinned by release tarball).
- server/net/network-principal.js: headersFor(path) returns a cached, 5-minute token from
  Network's /oauth/token (Live's existing OAuth client id/secret) for the routes Network guards
  by capability (coins credit/debit/transfer, notifications push/push-bulk) and X-Internal-Key for
  everything else. No token (old Network, no grant, no secret) -> key, token attempts paused 5 min
- wallet-client and notify: a 401 on a token call is retried once with the key; wallet calls carry
  idempotency keys, so the retry can't double-apply
- identity-sync test no longer swaps servers on one port (undici reused the dead keep-alive socket
  on Node 22)

Changes

ddiff --git a/package-lock.json b/package-lock.json
iindex fe65d18e..b49f54cf 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -25,6 +25,7 @@
          "multer": "^1.4.5-lts.1",
          "node-media-server": "^2.7.1",
          "onnxruntime-node": "^1.20.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.2.0",
          "openvibe-shared": "file:./vendor/openvibe-shared",
          "sanitize-html": "^2.17.7",
          "sdp-transform": "^3.0.0",
@@@ -840,6 +841,39 @@
          "node": ">=12.0"
        }
      },
++    "node_modules/ajv": {
++      "version": "8.20.0",
++      "resolved": "https://registry.npmjs.org/ajv/-/ajv-8.20.0.tgz",
++      "integrity": "sha512-Thbli+OlOj+iMPYFBVBfJ3OmCAnaSyNn4M1vz9T6Gka5Jt9ba/HIR56joy65tY6kx/FCF5VXNB819Y7/GUrBGA==",
++      "license": "MIT",
++      "dependencies": {
++        "fast-deep-equal": "^3.1.3",
++        "fast-uri": "^3.0.1",
++        "json-schema-traverse": "^1.0.0",
++        "require-from-string": "^2.0.2"
++      },
++      "funding": {
++        "type": "github",
++        "url": "https://github.com/sponsors/epoberezkin"
++      }
++    },
++    "node_modules/ajv-formats": {
++      "version": "3.0.1",
++      "resolved": "https://registry.npmjs.org/ajv-formats/-/ajv-formats-3.0.1.tgz",
++      "integrity": "sha512-8iUql50EUR+uUcdRQ3HDqa6EVyo3docL8g5WJ3FNcWmu62IbkGUue/pEyLBW8VGKKucTPgqeks4fIU1DA4yowQ==",
++      "license": "MIT",
++      "dependencies": {
++        "ajv": "^8.0.0"
++      },
++      "peerDependencies": {
++        "ajv": "^8.0.0"
++      },
++      "peerDependenciesMeta": {
++        "ajv": {
++          "optional": true
++        }
++      }
++    },
      "node_modules/ansi-styles": {
        "version": "4.3.0",
        "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-4.3.0.tgz",
@@@ -1703,6 +1737,28 @@
          "node": ">=20"
        }
      },
++    "node_modules/fast-deep-equal": {
++      "version": "3.1.3",
++      "resolved": "https://registry.npmjs.org/fast-deep-equal/-/fast-deep-equal-3.1.3.tgz",
++      "integrity": "sha512-f3qQ9oQy9j2AhBe/H9VC91wLmKBCCU/gDOnKNAYG5hswO7BLKj09Hc5HYNz9cGI++xlpDCIgDaitVs03ATR84Q==",
++      "license": "MIT"
++    },
++    "node_modules/fast-uri": {
++      "version": "3.1.8",
++      "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.8.tgz",
++      "integrity": "sha512-GZMtZUTNRpOVIECoXwLNZS5xUGE+mVNbTB8h/7Rwh2TFWcBQiPzTgyZi05BF9UMZKkLJv8XBRJTlU7zg8+ZfMg==",
++      "funding": [
++        {
++          "type": "github",
++          "url": "https://github.com/sponsors/fastify"
++        },
++        {
++          "type": "opencollective",
++          "url": "https://opencollective.com/fastify"
++        }
++      ],
++      "license": "BSD-3-Clause"
++    },
      "node_modules/fetch-blob": {
        "version": "3.2.0",
        "resolved": "https://registry.npmjs.org/fetch-blob/-/fetch-blob-3.2.0.tgz",
@@@ -2160,6 +2216,12 @@
          "node": ">=0.10.0"
        }
      },
++    "node_modules/json-schema-traverse": {
++      "version": "1.0.0",
++      "resolved": "https://registry.npmjs.org/json-schema-traverse/-/json-schema-traverse-1.0.0.tgz",
++      "integrity": "sha512-NM8/P9n3XjXhIZn1lLhkFaACTOURQXjWhV4BA/RnOv8xvgqtqpAX9IO4mRQxSx1Rlo4tqzeqb0sOlruaOy3dug==",
++      "license": "MIT"
++    },
      "node_modules/jsonwebtoken": {
        "version": "9.0.3",
        "resolved": "https://registry.npmjs.org/jsonwebtoken/-/jsonwebtoken-9.0.3.tgz",
@@@ -2786,6 +2848,19 @@
          "onnxruntime-common": "1.27.0"
        }
      },
++    "node_modules/openvibe-contracts": {
++      "version": "0.2.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.2.0",
++      "integrity": "sha512-ARq8C+5x6CDimij62hOjqYa3nu4liwxPpPaACp3khb50YHeOPgMFJiXKZlbhsVTTE4xHe/FcMhkcRZAob3iTzw==",
++      "license": "AGPL-3.0-only",
++      "dependencies": {
++        "ajv": "^8.17.1",
++        "ajv-formats": "^3.0.1"
++      },
++      "engines": {
++        "node": ">=20"
++      }
++    },
      "node_modules/openvibe-shared": {
        "resolved": "vendor/openvibe-shared",
        "link": true
@@@ -3007,6 +3082,15 @@
        "integrity": "sha512-urBwgfrvVP/eAyXx4hluJivBKzuEbSQs9rKWCrCkbSxNv8mxPcUZKeuoF3Uy4mJl3Lwprp6yy5/39VWigZ4K6Q==",
        "license": "Apache-2.0"
      },
++    "node_modules/require-from-string": {
++      "version": "2.0.2",
++      "resolved": "https://registry.npmjs.org/require-from-string/-/require-from-string-2.0.2.tgz",
++      "integrity": "sha512-Xf0nWe6RseziFMu+Ap9biiUbmplq6S9/p+7w7YXP/JBHhrUDDUhwa+vANyubuqfZWTveU//DYVGsDG7RKL/vEw==",
++      "license": "MIT",
++      "engines": {
++        "node": ">=0.10.0"
++      }
++    },
      "node_modules/safe-buffer": {
        "version": "5.2.1",
        "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.2.1.tgz",
ddiff --git a/package.json b/package.json
iindex 4bc154fe..42635c3b 100644
---- a/package.json
++++ b/package.json
@@@ -1,7 +1,7 @@
  {
    "name": "live",
    "version": "1.0.0",
--  "description": "OpenVibe.Live — free & open live streaming. Part of the OpenVibe network (media by OpenVibe.Media, identity by OpenVibe.Network).",
++  "description": "OpenVibe.Live \u2014 free & open live streaming. Part of the OpenVibe network (media by OpenVibe.Media, identity by OpenVibe.Network).",
    "main": "server/index.js",
    "scripts": {
      "start": "node server/index.js",
@@@ -43,6 +43,7 @@
      "multer": "^1.4.5-lts.1",
      "node-media-server": "^2.7.1",
      "onnxruntime-node": "^1.20.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.2.0",
      "openvibe-shared": "file:./vendor/openvibe-shared",
      "sanitize-html": "^2.17.7",
      "sdp-transform": "^3.0.0",
ddiff --git a/server/monetization/wallet-client.js b/server/monetization/wallet-client.js
iindex 5e91e199..7575f567 100644
---- a/server/monetization/wallet-client.js
++++ b/server/monetization/wallet-client.js
@@@ -14,6 +14,7 @@
   */
  'use strict';
  const db = require('../db/database');
++const principal = require('../net/network-principal');
  
  const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
  const INTERNAL_API_KEY = process.env.INTERNAL_API_KEY || process.env.OV_INTERNAL_KEY || '';
@@@ -42,18 +43,25 @@ function networkUserId(localUserId) {
      }
  }
  
--async function _post(apiPath, body) {
++async function _post(apiPath, body, retried = false) {
      let res;
++    const auth = await principal.headersFor(apiPath);   // scoped service token, or X-Internal-Key
      try {
          res = await fetch(`${NETWORK_INTERNAL_URL}${apiPath}`, {
              method: 'POST',
--            headers: { 'X-Internal-Key': INTERNAL_API_KEY, 'Content-Type': 'application/json', Accept: 'application/json' },
++            headers: { ...auth, 'Content-Type': 'application/json', Accept: 'application/json' },
              body: JSON.stringify(body),
          });
      } catch (err) {
          throw new WalletError(`Network wallet unreachable: ${err.message}`, 0);
      }
      const json = await res.json().catch(() => null);
++    // A rejected token (expired key rotation, revoked grant) is retried once with the internal key.
++    // Idempotency keys make the retry safe: a credit that did land is not applied twice.
++    if (res.status === 401 && auth.Authorization && !retried) {
++        principal.tokenRejected(json && json.code);
++        return _post(apiPath, body, true);
++    }
      if (!res.ok) {
          throw new WalletError((json && json.error) || `wallet ${res.status}`, res.status, json);
      }
ddiff --git a/server/net/network-principal.js b/server/net/network-principal.js
nnew file mode 100644
iindex 00000000..40ca2b9a
---- /dev/null
++++ b/server/net/network-principal.js
@@@ -0,0 +1,56 @@
++'use strict';
++/**
++ * Live as a service principal toward OpenVibe.Network (roadmap Wave 1, ADR-003).
++ *
++ * headersFor(path) returns the auth headers for one internal call: a short-lived, capability-scoped
++ * token from Network's /oauth/token (client_credentials, using Live's existing OAuth client id and
++ * secret) for the routes Network guards by capability, and the shared X-Internal-Key for every other
++ * route. If a token can't be had (Network not upgraded yet, no grant, secret missing) the key is used
++ * and token attempts pause for a few minutes, so a Network outage never breaks coins or notifications.
++ * After a 401 on a token call, call tokenRejected() and retry once; the retry uses the key.
++ */
++const { serviceAuth } = require('openvibe-contracts');
++
++const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
++const INTERNAL_API_KEY = process.env.INTERNAL_API_KEY || process.env.OV_INTERNAL_KEY || '';
++const CLIENT_ID = process.env.OV_OAUTH_CLIENT_ID || 'live';
++const CLIENT_SECRET = process.env.OV_OAUTH_CLIENT_SECRET || '';
++const PAUSE_MS = 5 * 60 * 1000;
++
++// Network routes that accept a service token (OpenVibe.Network server/internal/routes.js TOKEN_ROUTES).
++const TOKEN_PATHS = new Set(['/internal/coins/credit', '/internal/coins/debit', '/internal/coins/transfer', '/internal/notifications/push', '/internal/notifications/push-bulk']);
++
++const tokens = CLIENT_SECRET ? serviceAuth.createTokenClient({
++    tokenUrl: `${NETWORK_INTERNAL_URL}/oauth/token`, clientId: CLIENT_ID, clientSecret: CLIENT_SECRET, audience: 'openvibe.network',
++}) : null;
++let pausedUntil = 0;
++let lastFailure = null;
++const stats = { token: 0, key: 0, tokenFailures: 0 };
++
++const legacyHeaders = () => (INTERNAL_API_KEY ? { 'X-Internal-Key': INTERNAL_API_KEY } : {});
++
++async function headersFor(path) {
++    if (tokens && TOKEN_PATHS.has(path) && Date.now() >= pausedUntil) {
++        try {
++            const h = await tokens.authHeaders();
++            stats.token++;
++            return h;
++        } catch (err) {
++            stats.tokenFailures++;
++            pausedUntil = Date.now() + PAUSE_MS;
++            if (lastFailure !== err.message) console.warn(`[Principal] no service token (${err.message}); using the internal key for ${PAUSE_MS / 60000} min`);
++            lastFailure = err.message;
++        }
++    }
++    stats.key++;
++    return legacyHeaders();
++}
++
++/** A token call came back 401: drop the cached token and use the key until the pause ends. */
++function tokenRejected(problemCode) {
++    if (tokens) tokens.invalidate();
++    pausedUntil = Date.now() + PAUSE_MS;
++    console.warn(`[Principal] Network refused the service token (${problemCode || '401'}); using the internal key for ${PAUSE_MS / 60000} min`);
++}
++
++module.exports = { headersFor, tokenRejected, TOKEN_PATHS, stats, _reset() { pausedUntil = 0; if (tokens) tokens.invalidate(); } };
ddiff --git a/server/utils/notify.js b/server/utils/notify.js
iindex 888519c9..09a183d2 100644
---- a/server/utils/notify.js
++++ b/server/utils/notify.js
@@@ -35,13 +35,19 @@ function toNetworkIds(liveUserIds) {
      return { ids: [...ids], unlinked };
  }
  
--function _post(path, body) {
--    if (!INTERNAL_API_KEY) return Promise.resolve(null);
--    return fetch(`${OV_NETWORK_INTERNAL_URL}${path}`, {
++async function _post(path, body, retried = false) {
++    if (!INTERNAL_API_KEY) return null;
++    const auth = await require('../net/network-principal').headersFor(path);  // service token where Network accepts one
++    const res = await fetch(`${OV_NETWORK_INTERNAL_URL}${path}`, {
          method: 'POST',
--        headers: { 'Content-Type': 'application/json', 'X-Internal-Key': INTERNAL_API_KEY },
++        headers: { 'Content-Type': 'application/json', ...auth },
          body: JSON.stringify(body),
      });
++    if (res.status === 401 && auth.Authorization && !retried) {
++        require('../net/network-principal').tokenRejected();
++        return _post(path, body, true);
++    }
++    return res;
  }
  
  /**
ddiff --git a/test/identity-sync.test.js b/test/identity-sync.test.js
iindex 5ca27031..53f1b416 100644
---- a/test/identity-sync.test.js
++++ b/test/identity-sync.test.js
@@@ -15,10 +15,12 @@ process.env.DB_PATH = path.join(tmp, 'live.db');
  process.env.INTERNAL_API_KEY = 'k-test';
  
  const received = [];
++const mode = { old: false };
  const network = http.createServer((req, res) => {
      let body = '';
      req.on('data', (c) => { body += c; });
      req.on('end', () => {
++        if (mode.old) { res.statusCode = 404; return res.end('{}'); }
          assert.strictEqual(req.headers['x-internal-key'], 'k-test');
          assert.strictEqual(req.url, '/internal/identity/legacy-map');
          const { entries } = JSON.parse(body);
@@@ -68,11 +70,9 @@ const network = http.createServer((req, res) => {
      assert.strictEqual(sync.subjectOf(uid), 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', 'a token for a different network id cannot rewrite the link');
  
      // Network without the endpoint yet: skip quietly.
--    network.close();
--    const old = http.createServer((req, res) => { res.statusCode = 404; res.end('{}'); });
--    await new Promise((r) => old.listen(Number(process.env.OV_NETWORK_INTERNAL_URL.split(':').pop()), '127.0.0.1', r));
++    mode.old = true;
      assert.deepStrictEqual(await sync.syncLegacyMap(), { skipped: 'network has no /internal/identity yet' });
--    old.close();
++    network.close();
  
      fs.rmSync(tmp, { recursive: true, force: true });
      console.log('identity sync: all checks passed');
ddiff --git a/test/network-principal.test.js b/test/network-principal.test.js
nnew file mode 100644
iindex 00000000..8fed15fa
---- /dev/null
++++ b/test/network-principal.test.js
@@@ -0,0 +1,89 @@
++'use strict';
++
++// Live as a service principal (roadmap Wave 1): coins and notification pushes go to Network with a
++// client-credentials token; every other internal call keeps the X-Internal-Key; a token that can't be
++// had or is refused falls back to the key without losing the call. Against a stub Network.
++
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++const { serviceAuth } = require('openvibe-contracts');
++
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-principal-'));
++process.env.DB_PATH = path.join(tmp, 'live.db');
++process.env.INTERNAL_API_KEY = 'legacy-key';
++process.env.OV_OAUTH_CLIENT_ID = 'live';
++process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';
++
++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 });
++const mode = { token: 'ok', coins: 'ok' };
++const seen = [];
++let tokenCalls = 0;
++
++const network = http.createServer((req, res) => {
++    let raw = '';
++    req.on('data', (c) => { raw += c; });
++    req.on('end', () => {
++        const send = (status, body, type = 'application/json') => { res.statusCode = status; res.setHeader('Content-Type', type); res.end(JSON.stringify(body)); };
++        if (req.url === '/oauth/token') {
++            tokenCalls++;
++            const f = new URLSearchParams(raw);
++            assert.strictEqual(f.get('grant_type'), 'client_credentials');
++            assert.strictEqual(f.get('client_secret'), 'live-secret');
++            if (mode.token !== 'ok') return send(400, { error: 'invalid_scope' });
++            const now = Math.floor(Date.now() / 1000);
++            return send(200, { access_token: serviceAuth.signServiceToken({ iss: 'https://openvibe.network', sub: 'svc:live', actor_type: 'service', aud: ['openvibe.network'], cap: ['network.coins.credit'], iat: now, exp: now + 300, jti: `tok_${tokenCalls}abcdefg` }, keys.privateKey), expires_in: 300 });
++        }
++        const auth = req.headers.authorization ? 'token' : req.headers['x-internal-key'] === 'legacy-key' ? 'key' : 'none';
++        seen.push({ url: req.url, auth });
++        if (auth === 'token' && mode.coins === 'refuse') return send(401, { code: 'token.bad_signature' }, 'application/problem+json');
++        if (req.url.startsWith('/internal/coins/')) return send(200, { balance: 10 });
++        return send(200, { ok: true });
++    });
++});
++
++(async () => {
++    await new Promise((r) => network.listen(0, '127.0.0.1', r));
++    process.env.OV_NETWORK_INTERNAL_URL = `http://127.0.0.1:${network.address().port}`;
++
++    const db = require('../server/db/database');
++    db.initDb();
++    const d = db.getDb();
++    const uid = d.prepare("INSERT INTO users (username, password_hash, stream_key) VALUES ('p', 'x', 'k1')").run().lastInsertRowid;
++    d.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id) VALUES (?, 'network', '41')").run(uid);
++    const wallet = require('../server/monetization/wallet-client');
++    const principal = require('../server/net/network-principal');
++    const notify = require('../server/utils/notify');
++
++    // 1. Coins go with a service token; the token is cached across calls.
++    assert.deepStrictEqual(await wallet.credit(uid, 5, 'test', 'idem-1'), { balance: 10 });
++    await wallet.credit(uid, 5, 'test', 'idem-2');
++    assert.deepStrictEqual(seen.map((s) => s.auth), ['token', 'token']);
++    assert.strictEqual(tokenCalls, 1, 'one token for both calls');
++
++    // 2. Routes Network doesn't guard by capability keep the key.
++    seen.length = 0;
++    await (async () => { notify.reportLinkedAccount({ id: uid, username: 'p' }); await new Promise((x) => setTimeout(x, 150)); })();
++    assert.deepStrictEqual(seen.map((s) => `${s.url}:${s.auth}`), ['/internal/link-account:key']);
++
++    // 3. A refused token is retried once with the key, and the key is used until the pause ends.
++    seen.length = 0; mode.coins = 'refuse';
++    assert.deepStrictEqual(await wallet.credit(uid, 5, 'test', 'idem-3'), { balance: 10 }, 'the call still succeeds');
++    assert.deepStrictEqual(seen.map((s) => s.auth), ['token', 'key']);
++    await wallet.credit(uid, 5, 'test', 'idem-4');
++    assert.strictEqual(seen[2].auth, 'key', 'paused: no token attempt');
++
++    // 4. Network that can't issue a token (not upgraded, no grant): quietly use the key.
++    principal._reset(); mode.coins = 'ok'; mode.token = 'fail'; seen.length = 0;
++    await wallet.debit(uid, 1, 'test', 'idem-5');
++    assert.deepStrictEqual(seen.map((s) => s.auth), ['key']);
++    assert.ok(principal.stats.tokenFailures >= 1);
++
++    network.close();
++    fs.rmSync(tmp, { recursive: true, force: true });
++    console.log('network principal: all checks passed');
++    process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });