openvibers/openvibe.live Commits
Live's money calls carry the person's own subject assertion (#620)
Changes
@@@ -175,7 +175,7 @@ Minimum `.env`: `BASE_URL`, `JWT_SECRET`, `OV_NETWORK_URL`, `OV_NETWORK_INTERNAL- [docs/restream-branding.md](docs/restream-branding.md) — branding guide for restream channels.<!-- versions:start -->--- openvibe-contracts: v0.294.0--- openvibe-sdk: v0.82.0++- openvibe-contracts: v0.297.0++- openvibe-sdk: v0.90.0- openvibe-shared: v4.115.1<!-- versions:end -->@@@ -5,8 +5,8 @@"runtime": "node 22, express, PostgreSQL, node-media-server, mediasoup; no build step","port": 3000,"domain": "openvibe.live (www, ingest, whip)",-- "contracts": "openvibe-contracts v0.294.0",-- "sdk": "openvibe-sdk v0.82.0",++ "contracts": "openvibe-contracts v0.297.0",++ "sdk": "openvibe-sdk v0.90.0","shared": "openvibe-shared v4.115.1","capabilities": ["live.chat_context.read",@@@ -23,8 +23,8 @@"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.294.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1","pg": "^8.23.1","sanitize-html": "^2.17.7",@@@ -2840,9 +2840,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.294.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.294.0",-- "integrity": "sha512-CBDymrE4VrgslAAgCTCvecNrVcaevhBcIrb4nU4ApkJasst+6xt3RdR9lktsdHuhBeipdlKAfaxLAFmRadkKqA==",++ "version": "0.297.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",++ "integrity": "sha512-FycRElRNzeFLEfzVUKIyE3ieKphUrQbAbsprnvs0Uu7ZdlBKJG6wQV/9gD77Neu1Etcsyezf9Boyq+MsjuraVA==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -2856,9 +2856,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.82.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",-- "integrity": "sha512-oJarlzFGEkqYY0eFumIRjJUqMpOT70kgbhgeDtA8QHnuMZR3VIGQubM2UlOaqXgkxD5uhdZ/G6jLHbjvUhqTLQ==",++ "version": "0.90.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0",++ "integrity": "sha512-uOcYG+w5BVKePmSmCsmhFe6E1CNTr7/vLUqyOLSzsB9Z8D3VNbWsPCwOd5W0eY+3Dd4YhYIf4iT0QTIj9dinGg==","license": "MIT","bin": {"openvibe-download": "bin/openvibe-download.js"@@@ -42,8 +42,8 @@"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.294.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1","pg": "^8.23.1","sanitize-html": "^2.17.7",@@@ -27,6 +27,7 @@ const crypto = require('crypto');const db = require('../db/database');const config = require('../config');const billing = require('./billing-client');++const { assertionFor } = require('../net/subject-assertion');const { BillingCallError, subjectFor, ref, api } = billing;const base = () => config.baseUrl.replace(/\/+$/, '');@@@ -49,6 +50,10 @@ function refOf(out) {* calling Billing; a key already used with another body is refused; anything else is (re)sent.*/async function perform({ action, method = 'POST', path, body, key, liveUserId = null, liveRef = null, trace }) {++ // The person's own subject assertion (ADR-003, the 2026-11-06 amendment; plan H1): Billing checks++ // it against the person the money is for. Live mints it from the person's own token on the++ // request; without that token there is none, and the call travels the old way (the flag decides).++ const assertion = await assertionFor(trace, personOf(body));const d = db.getDb();const requestJson = JSON.stringify(body || {});const row = await d.prepare('SELECT * FROM billing_actions WHERE idempotency_key = ?').get(key);@@@ -62,7 +67,7 @@ async function perform({ action, method = 'POST', path, body, key, liveUserId =const id = (await d.prepare('SELECT id FROM billing_actions WHERE idempotency_key = ?').get(key)).id;await d.prepare('UPDATE billing_actions SET attempts = attempts + 1, updated_at = ov_now() WHERE id = ?').run(id);try {-- const out = await billing.request(method, path, { body, idempotencyKey: key, trace });++ const out = await billing.request(method, path, { body, idempotencyKey: key, trace, assertion });await d.prepare(`UPDATE billing_actions SET status = 'done', http_status = 200, billing_ref = ?, response_json = ?, error = NULL, updated_at = ov_now() WHERE id = ?`).run(refOf(out), JSON.stringify(out), id);return { out, replayed: !!out._replayed, actionId: id };@@@ -131,6 +136,17 @@ function normalizeMessage(message, max) {const streamTarget = (streamId) => (streamId ? { service: 'live', type: 'stream', id: String(streamId) } : undefined);++/** The person a money body is about: the payer, the subscriber, the cash-out's owner. */++function personOf(body) {++ const b = body || {};++ for (const field of ['from', 'subject', 'subscriber', 'on_behalf_of']) {++ const v = b[field];++ const id = v && typeof v === 'object' ? v.id : v;++ if (typeof id === 'string' && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(id)) return id;++ }++ return null;++}++/** POST /api/funds/donate under Billing. Same result shape as vibes.donate() plus `balance`. */async function donate(req, { toUserId, streamId, amount, message, goalId = null }) {const vibes = require('./vibes');@@@ -220,7 +236,7 @@ async function recycle(req, amount) {/** GET /api/funds/balance shape. Throws BillingCallError when Billing cannot answer. */async function balance(liveUserId, trace) {const sid = await subjectFor(liveUserId);-- const b = await api.balance(sid, { trace });++ const b = await api.balance(sid, { trace, assertion: await assertionFor(trace, sid) });const per = Number(b.bits_per_usd) || 100;const usd = (n) => (Math.round(Number(n) || 0) / per).toFixed(2);return {@@@ -232,9 +248,9 @@ async function balance(liveUserId, trace) {const LIVE_TYPE = { cashout_request: 'cashout', cashout_denied: 'refund', cashout_paid: 'cashout_paid' };/** GET /api/funds/history rows in Live's transaction shape (the dashboard renders these). */--async function history(liveUserId, limit = 50) {++async function history(liveUserId, limit = 50, trace = null) {const sid = await subjectFor(liveUserId);-- const page = await api.transactions(sid, { limit: Math.min(200, Math.max(1, limit)) });++ const page = await api.transactions(sid, { limit: Math.min(200, Math.max(1, limit)) }, { trace, assertion: await assertionFor(trace, sid) });const txns = page.transactions || [];const people = await billing.liveUsersForSubjects(txns.flatMap((t) => [t.from_subject, t.to_subject]));return txns.map((t) => {@@@ -311,12 +327,12 @@ function vipClient() {});return _vip;}--async function entitlementOf(member, creator) {++async function entitlementOf(member, creator, trace = null) {if (process.env.LIVE_VIP_ENTITLEMENTS !== 'off') {const e = await vipClient().checkEntitlement({ subject: member, creator, product: 'live' });if (e && e.status !== 'unknown') return { active: !!e.active, via: 'vip' };}-- const e = await api.entitlement(member, creator);++ const e = await api.entitlement(member, creator, { trace, assertion: await assertionFor(trace, member) });return { active: !!(e && e.active), via: 'billing' };}@@@ -325,7 +341,7 @@ function viewerOrNull(fn) { return fn().catch((e) => { if (e instanceof BillingCasync function subscribe(req, { streamer, provider, autoRenewRaw }) {const subscriber = await subjectFor(req.user.id);const to = await subjectFor(streamer.id, { role: 'recipient' });-- const ent = await api.entitlement(subscriber, to, { trace: req.headers });++ const ent = await api.entitlement(subscriber, to, { trace: req.headers, assertion: await assertionFor(req.headers, subscriber) });if (ent && ent.active) return { status: 409, body: { error: 'Already subscribed' } };if (provider === 'bucks') {@@@ -406,9 +422,9 @@ function presentSub(s, people) {};}--async function mySubscriptions(liveUserId) {++async function mySubscriptions(liveUserId, trace = null) {const sid = await subjectFor(liveUserId);-- const out = await api.listSubscriptions({ subscriber: sid });++ const out = await api.listSubscriptions({ subscriber: sid }, { trace, assertion: await assertionFor(trace, sid) });const subs = out.subscriptions || [];const people = await billing.liveUsersForSubjects(subs.map((s) => s.streamer && s.streamer.id));return subs.map((s) => presentSub(s, people));@@@ -423,12 +439,12 @@ function tierOnePrice(t) {return cents > 0 ? cents : null;}--async function channelState(streamer, viewer) {++async function channelState(streamer, viewer, trace = null) {const to = await viewerOrNull(() => subjectFor(streamer.id, { role: 'recipient' }));if (!to) return { subscribed: false, subscriberCount: 0 };const me = viewer ? await viewerOrNull(() => subjectFor(viewer.id)) : null;const [ent, list, tiers] = await Promise.all([-- me ? await entitlementOf(me, to) : Promise.resolve(null),++ me ? await entitlementOf(me, to, trace) : Promise.resolve(null),api.listSubscriptions({ streamer: to, status: 'active' }),api.tiers(to).catch(() => null),]);@@@ -439,7 +455,7 @@ async function channelState(streamer, viewer) {async function cancelSubscription(req, { id, streamerId }) {const sid = await subjectFor(req.user.id);-- const out = await api.listSubscriptions({ subscriber: sid }, { trace: req.headers });++ const out = await api.listSubscriptions({ subscriber: sid }, { trace: req.headers, assertion: await assertionFor(req.headers, sid) });const subs = (out.subscriptions || []).filter((s) => s.subscriber && s.subscriber.id === sid);let sub = subs.find((s) => String(s.id) === String(id));if (!sub && streamerId) {@@@ -52,10 +52,11 @@ function connectionNeverMade(err) {}/**-- * One call. opts: { body, idempotencyKey, trace (incoming request headers, for traceparent) }.++ * One call. opts: { body, idempotencyKey, trace (incoming request headers, for traceparent),++ * assertion (the person's own subject assertion, ADR-003's 2026-11-06 amendment) }.* Resolves the parsed JSON body of a 2xx (plus `_replayed` when Billing replayed the key).*/--async function request(method, path, { body, idempotencyKey, trace, retried = false } = {}) {++async function request(method, path, { body, idempotencyKey, trace, assertion, retried = false } = {}) {const write = method !== 'GET';if (write && !KEY_RE.test(String(idempotencyKey || ''))) throw new Error(`billing-client: ${method} ${path} needs a valid Idempotency-Key`);let auth;@@@ -66,6 +67,8 @@ async function request(method, path, { body, idempotencyKey, trace, retried = fa}const headers = { Accept: 'application/json', ...auth, ...http.outboundHeaders(http.requestContext(trace || {})) };if (write) headers['Idempotency-Key'] = idempotencyKey;++ // The person's own subject assertion: Billing checks it against the person the money is for.++ if (assertion) headers['X-OV-Subject-Assertion'] = String(assertion);let payload;if (body !== undefined) { headers['Content-Type'] = 'application/json'; payload = JSON.stringify(body); }let res;@@@ -81,7 +84,7 @@ async function request(method, path, { body, idempotencyKey, trace, retried = fatry { json = text ? JSON.parse(text) : {}; } catch { /* not JSON */ }if (res.status === 401 && !retried) { // token rotated/expired under us: once more with a fresh oneprincipal.invalidate(AUDIENCE);-- return await request(method, path, { body, idempotencyKey, trace, retried: true });++ return await request(method, path, { body, idempotencyKey, trace, assertion, retried: true });}if (res.ok) {const out = json || {};@@@ -77,7 +77,7 @@ router.post('/subscribe', requireAuth, money.guardWrite, async (req, res) => {// My subscriptionsrouter.get('/subscriptions/mine', requireAuth, async (req, res) => {-- try { return res.json({ subscriptions: await billingActions.mySubscriptions(req.user.id) }); }++ try { return res.json({ subscriptions: await billingActions.mySubscriptions(req.user.id, req.headers) }); }catch (err) {if (billingActions.sendError(res, err, { unavailable: 'Subscriptions unavailable right now — the billing service is not answering.' })) return;return res.status(503).json({ error: 'Subscriptions unavailable right now', unavailable: true });@@@ -97,7 +97,7 @@ router.get('/channel/:username', optionalAuth, async (req, res) => {const streamer = await db.getUserByUsername(req.params.username);if (!streamer) return res.status(404).json({ error: 'Not found' });let subscribed = false, subscriberCount = null, billingUnavailable = false, billingPriceCents = null;-- try { ({ subscribed, subscriberCount, priceCents: billingPriceCents } = await billingActions.channelState(streamer, req.user || null)); }++ try { ({ subscribed, subscriberCount, priceCents: billingPriceCents } = await billingActions.channelState(streamer, req.user || null, req.headers)); }catch { billingUnavailable = true; }const priceUsd = billingPriceCents ? billingPriceCents / 100 : await pay._num('sub_price_usd', 4.99);let powerchat = { direct: false, site: false };@@@ -169,7 +169,7 @@ router.post('/recycle', requireAuth, money.guardWrite, async (req, res) => {// ── Transaction History ──────────────────────────────────────router.get('/history', requireAuth, async (req, res) => {const limit = Math.min(parseInt(req.query.limit || '50'), 200);-- try { return res.json({ transactions: await billingActions.history(req.user.id, limit) }); }++ try { return res.json({ transactions: await billingActions.history(req.user.id, limit, req.headers) }); }catch (err) {if (billingActions.sendError(res, err, { unavailable: 'Vibes history unavailable right now — the billing service is not answering.' })) return;return res.status(503).json({ error: 'Vibes history unavailable right now', unavailable: true });nnew file mode 100644@@@ -0,0 +1,63 @@++'use strict';++++/**++ * The person's own subject assertion for Live's money calls (ADR-003, the 2026-11-06 amendment; plan H1).++ *++ * Live names the person it acts for — the person who signed in — and Billing believes it. On a money++ * route that is only Live's word: a compromised Live could tip in anyone's name or read anyone's++ * receipts. A subject assertion is the person's own answer to the same question: Network signs their++ * access token into a 60-second proof (ES256) that names them, and Billing checks that proof instead++ * of Live's word.++ *++ * Live holds the person's Network access token on the request (the session cookie `ov_token`, or the++ * Authorization header), so it can ask Network to sign it. Without that token there is nothing to++ * sign and no assertion is minted: the call then travels the old way, which Billing's++ * BILLING_REQUIRE_SUBJECT_ASSERTION flag decides on.++ *++ * The assertion is minted per call and never cached: Billing refuses a replayed one (the jti is++ * remembered), so two money calls must not share it.++ */++const { createSubjectAssertionMinter } = require('openvibe-sdk/assertions');++++const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++const CLIENT_ID = process.env.OV_OAUTH_CLIENT_ID || 'live';++const CLIENT_SECRET = process.env.OV_OAUTH_CLIENT_SECRET || '';++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++++/** The person's own Network access token on this request, or null (no session, or a Live API token). */++function personToken(headers) {++ const h = headers || {};++ const auth = String(h.authorization || '');++ if (/^Bearer\s+/i.test(auth)) {++ const token = auth.slice(7).trim();++ // A Live API token (hbt_…) is not a Network session; Network would refuse it anyway.++ if (token && !/^hbt_/.test(token)) return token;++ }++ // The session cookie (the person's Network token, JS-readable for the navbar).++ for (const part of String(h.cookie || '').split(';')) {++ const idx = part.indexOf('=');++ if (idx < 0) continue;++ if (part.slice(0, idx).trim() !== 'ov_token') continue;++ const value = decodeURIComponent(part.slice(idx + 1).trim());++ if (value) return value;++ }++ return null;++}++++const minter = createSubjectAssertionMinter({++ internalUrl: NETWORK_INTERNAL_URL,++ oauth: { clientId: CLIENT_ID, clientSecret: CLIENT_SECRET },++});++++/**++ * The assertion for the person this request acts for, or null when it cannot be had (no session,++ * no grant, Network down). It never throws: a money call must not fail because a proof is missing.++ */++async function assertionFor(headers, subject) {++ if (!subject || !SUBJECT_RE.test(String(subject))) return null;++ const userToken = personToken(headers);++ if (!userToken) return null;++ return await minter.mint({ userToken, subject, audience: 'openvibe.billing' });++}++++module.exports = { personToken, assertionFor, SUBJECT_RE };@@@ -14,6 +14,7 @@*/const http = require('http');const crypto = require('crypto');++const jwt = require('jsonwebtoken');const { serviceAuth, http: ovHttp, ids } = require('openvibe-contracts');const read = (req) => new Promise((resolve) => { const c = []; req.on('data', (d) => c.push(d)); req.on('end', () => resolve(Buffer.concat(c).toString('utf8'))); });@@@ -21,6 +22,9 @@ const send = (res, status, obj, headers = {}) => { res.writeHead(status, { 'Contconst problem = (res, status, code, detail, extra) => { res.writeHead(status, { 'Content-Type': 'application/problem+json' }); res.end(JSON.stringify(ovHttp.problem(status, code, { detail, extra }))); };const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve(`http://127.0.0.1:${server.address().port}`)));++/** The issuer Network's tokens carry in the tests. */++const ISSUER = 'https://openvibe.network';++const LIVE_BILLING_GRANTS = ['billing.intent.create', 'billing.transfer.create', 'billing.balance.read', 'billing.cashout.request', 'billing.subscription.manage', 'billing.entitlement.check'];async function startNetwork() {@@@ -38,6 +42,27 @@ async function startNetwork() {const claims = { iss: 'https://openvibe.network', sub: `svc:${body.get('client_id')}`, actor_type: 'service', aud: [aud], cap: grants[aud] || [], iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(6).toString('hex')}` };return send(res, 200, { access_token: serviceAuth.signServiceToken(claims, keys.privateKey), expires_in: 300 });}++ if (req.url === '/internal/subject-assertions') {++ // The mint route (ADR-003, the 2026-11-06 amendment): a first-party service hands over the++ // person's own token and gets back a 60-second assertion naming the person.++ const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.network' });++ if (!v.ok) return send(res, 401, { code: v.code });++ if (!/^svc:[a-z][a-z0-9-]{1,39}$/.test(String(v.claims.sub || ''))) return send(res, 403, { code: 'assertion.not_a_service' });++ const body = JSON.parse(raw || '{}');++ if (!body.user_token) return send(res, 401, { code: 'assertion.no_user_token' });++ let claims;++ try { claims = jwt.verify(String(body.user_token), keys.publicKey, { issuer: ISSUER }); } catch { return send(res, 401, { code: 'assertion.invalid_user_token' }); }++ const subject = claims.subject_id;++ if (body.subject && body.subject !== subject) return send(res, 403, { code: 'assertion.subject_mismatch' });++ const audience = Array.isArray(body.audience) ? body.audience : [body.audience];++ const forServices = [...new Set([v.claims.sub, ...(Array.isArray(body.for) ? body.for : [])])];++ const jti = `as_${crypto.randomBytes(6).toString('hex')}`;++ const now = Math.floor(Date.now() / 1000);++ const b64 = (b) => Buffer.from(b).toString('base64url');++ const header = b64(JSON.stringify({ alg: 'ES256', typ: 'JWT', kid: 'ov-network-assertion-1' }));++ const payload = b64(JSON.stringify({ iss: ISSUER, sub: subject, aud: audience, act: v.claims.sub, for: forServices, jti, iat: now, exp: now + 60 }));++ return send(res, 200, { assertion: `${header}.${payload}.${b64('signed')}`, subject, audience: body.audience, act: v.claims.sub, for: forServices, jti, expires_in: 60 });++ }if (req.url === '/internal/identity/resolve-batch') {const { ids: want } = JSON.parse(raw || '{}');return send(res, 200, { results: Object.fromEntries((want || []).map((id) => [id, legacy[id] ? { subject: { type: 'user', id: legacy[id] } } : null])) });@@@ -45,7 +70,9 @@ async function startNetwork() {send(res, 404, {});});const url = await listen(server);-- return { url, publicKey: keys.publicKey, grants, legacy, tokens, close: () => new Promise((r) => server.close(r)) };++ /** A Network user token (the session a person's browser holds). */++ const signUser = (subject, username = 'someone') => jwt.sign({ sub: String(subject), subject_id: subject, username, display_name: username, role: 'user' }, keys.privateKey, { algorithm: 'RS256', issuer: ISSUER, expiresIn: '1h' });++ return { url, publicKey: keys.publicKey, grants, legacy, tokens, signUser, close: () => new Promise((r) => server.close(r)) };}const CAP = [@@@ -172,7 +199,7 @@ async function startBilling({ network }) {const auth = String(req.headers.authorization || '');const v = serviceAuth.verifyServiceToken(auth.slice(7), { publicKey: network.publicKey, audience: 'openvibe.billing' });const rule = CAP.find(([m, re]) => m === req.method && re.test(path));-- const call = { method: req.method, path, query: Object.fromEntries(u.searchParams), key: req.headers['idempotency-key'] || null, body, cap: rule ? rule[2] : undefined, principal: v.ok ? v.claims.sub : null, traceparent: req.headers.traceparent || null };++ const call = { method: req.method, path, query: Object.fromEntries(u.searchParams), key: req.headers['idempotency-key'] || null, body, cap: rule ? rule[2] : undefined, principal: v.ok ? v.claims.sub : null, traceparent: req.headers.traceparent || null, assertion: req.headers['x-ov-subject-assertion'] || null };calls.push(call);if (state.delayMs) await new Promise((r) => setTimeout(r, state.delayMs));if (!v.ok) return problem(res, 401, v.code, v.reason);nnew file mode 100644@@@ -0,0 +1,118 @@++'use strict';++/**++ * The person's own subject assertion on Live's money calls (ADR-003, the 2026-11-06 amendment; plan H1).++ *++ * Live names the person it acts for, and Billing believes it. On a money route that is only Live's++ * word, so Live asks Network to sign the person's own token into a 60-second assertion and carries++ * it on (X-OV-Subject-Assertion). Without the person's token there is nothing to sign, and the call++ * travels the old way — which Billing's BILLING_REQUIRE_SUBJECT_ASSERTION flag decides on.++ */++const assert = require('assert');++const http = require('http');++const { startNetwork, startBilling } = require('./billing-stub');++++process.env.NODE_ENV = 'test';++process.env.OV_OAUTH_CLIENT_ID = 'live';++process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';++process.env.OV_BILLING_TIMEOUT_MS = '1500';++const quiet = console.log;++console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };++console.warn = () => {};++console.error = () => {};++++const SID = {++ ann: 'usr_01JAB2C3D4E5F6G7H8J9K0MNA1',++ bob: 'usr_01JAB2C3D4E5F6G7H8J9K0MNB2',++};++++let failures = 0;++async function check(name, fn) {++ try { await fn(); quiet(' ✓', name); }++ catch (e) { failures++; quiet(' ✗', name, '\n ', (e.stack || String(e)).split('\n').slice(0, 9).join('\n ')); }++}++++/** The claims of an assertion (the stub signs one; the signature is not what this test is about). */++const claimsOf = (assertion) => JSON.parse(Buffer.from(String(assertion).split('.')[1], 'base64url').toString('utf8'));++++(async () => {++ const network = await startNetwork();++ const billing = await startBilling({ network });++ process.env.OV_NETWORK_INTERNAL_URL = network.url;++ process.env.OV_BILLING_INTERNAL_URL = billing.url;++++ const db = require('../server/db/database');++ await db.initDb();++ const raw = db.getDb();++ const addUser = async (id, username, role) => await raw.prepare(++ `INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) OVERRIDING SYSTEM VALUE++ VALUES (?, ?, ?, ?, 'x', ?, 0)`).run(id, username, username, `${username}@x`, role);++ await addUser(2, 'ann', 'user');++ await addUser(3, 'bob', 'streamer');++ const link = async (uid, sid) => await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (?, 'network', ?, ?)").run(uid, String(100 + uid), sid);++ await link(2, SID.ann); await link(3, SID.bob);++ await db.ensureChannel(3);++ const streamB = Number((await db.createStream({ user_id: 3, channel_id: (await db.getChannelByUserId(3)).id, title: 'B', protocol: 'webrtc' })).lastInsertRowid);++++ const auth = require('../server/auth/auth');++ const signIn = async (req) => { const id = Number(req.headers['x-test-user'] || 0); const u = id ? await db.getUserById(id) : null; if (u) req.user = u; return u; };++ auth.requireAuth = (req, res, next) => { signIn(req).then((u) => u ? next() : res.status(401).json({ error: 'Authentication required' })).catch(next); };++ auth.optionalAuth = (req, res, next) => { signIn(req).then(() => next()).catch(next); };++++ const express = require('express');++ const app = express();++ app.use(express.json());++ app.use((req, res, next) => { signIn(req).then(() => next()).catch(next); });++ app.use('/api/funds', require('../server/monetization/routes'));++ const server = http.createServer(app);++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const call = async (method, p, user, body, headers = {}) => {++ const res = await fetch(`http://127.0.0.1:${server.address().port}${p}`, {++ method, redirect: 'manual',++ headers: { 'content-type': 'application/json', ...(user ? { 'x-test-user': String(user) } : {}), ...headers },++ body: body ? JSON.stringify(body) : undefined,++ });++ const text = await res.text();++ let json = null; try { json = JSON.parse(text); } catch { /* */ }++ return { status: res.status, json, text };++ };++ const callsSince = (n) => billing.calls.slice(n);++++ billing.fund(SID.ann, 3000);++++ await check('a money call for a person carries the person\'s own assertion', async () => {++ const n = billing.calls.length;++ const r = await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 150 }, { cookie: `ov_token=${network.signUser(SID.ann, 'ann')}` });++ assert.strictEqual(r.status, 200, r.text);++ const c = callsSince(n);++ assert.strictEqual(c.length, 1, 'one Billing call');++ assert.ok(c[0].assertion, 'the call carried an assertion');++ const claims = claimsOf(c[0].assertion);++ assert.strictEqual(claims.sub, SID.ann, 'the assertion names the person');++ assert.deepStrictEqual(claims.aud, ['openvibe.billing']);++ assert.strictEqual(claims.act, 'svc:live', 'Live is the service it was minted for');++ });++++ await check('without the person\'s token the money call travels the old way (no assertion)', async () => {++ const n = billing.calls.length;++ const r = await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 100 });++ assert.strictEqual(r.status, 200, r.text);++ const c = callsSince(n);++ assert.strictEqual(c.length, 1, 'one Billing call');++ assert.strictEqual(c[0].assertion, null, 'no assertion without the person\'s token');++ assert.deepStrictEqual(c[0].body.from, { type: 'user', id: SID.ann }, 'the person is still named in the body');++ });++++ await check('two money calls get two assertions (a replayed one is refused)', async () => {++ const token = network.signUser(SID.ann, 'ann');++ const n = billing.calls.length;++ await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 100 }, { cookie: `ov_token=${token}` });++ await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 100 }, { cookie: `ov_token=${token}` });++ const c = callsSince(n);++ assert.strictEqual(c.length, 2);++ assert.ok(c[0].assertion && c[1].assertion);++ assert.notStrictEqual(c[0].assertion, c[1].assertion, 'each call has its own assertion');++ });++++ quiet(failures ? `\n${failures} failed` : '\nall passed');++ process.exit(failures ? 1 : 0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -175,7 +175,7 @@ Minimum `.env`: `BASE_URL`, `JWT_SECRET`, `OV_NETWORK_URL`, `OV_NETWORK_INTERNAL
@@ -175,7 +175,7 @@ Minimum `.env`: `BASE_URL`, `JWT_SECRET`, `OV_NETWORK_URL`, `OV_NETWORK_INTERNAL
- [docs/restream-branding.md](docs/restream-branding.md) — branding guide for restream channels.
- [docs/restream-branding.md](docs/restream-branding.md) — branding guide for restream channels.
<!-- versions:start -->
<!-- versions:start -->
-- openvibe-contracts: v0.294.0
-- openvibe-sdk: v0.82.0
+- openvibe-contracts: v0.297.0
+- openvibe-sdk: v0.90.0
- openvibe-shared: v4.115.1
- openvibe-shared: v4.115.1
<!-- versions:end -->
<!-- versions:end -->
@@ -5,8 +5,8 @@
@@ -5,8 +5,8 @@
"runtime": "node 22, express, PostgreSQL, node-media-server, mediasoup; no build step",
"runtime": "node 22, express, PostgreSQL, node-media-server, mediasoup; no build step",
"port": 3000,
"port": 3000,
"domain": "openvibe.live (www, ingest, whip)",
"domain": "openvibe.live (www, ingest, whip)",
- "contracts": "openvibe-contracts v0.294.0",
- "sdk": "openvibe-sdk v0.82.0",
+ "contracts": "openvibe-contracts v0.297.0",
+ "sdk": "openvibe-sdk v0.90.0",
"shared": "openvibe-shared v4.115.1",
"shared": "openvibe-shared v4.115.1",
"capabilities": [
"capabilities": [
"live.chat_context.read",
"live.chat_context.read",
@@ -23,8 +23,8 @@
@@ -23,8 +23,8 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.294.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1",
"pg": "^8.23.1",
"pg": "^8.23.1",
"sanitize-html": "^2.17.7",
"sanitize-html": "^2.17.7",
@@ -2840,9 +2840,9 @@
@@ -2840,9 +2840,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.294.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.294.0",
- "integrity": "sha512-CBDymrE4VrgslAAgCTCvecNrVcaevhBcIrb4nU4ApkJasst+6xt3RdR9lktsdHuhBeipdlKAfaxLAFmRadkKqA==",
+ "version": "0.297.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",
+ "integrity": "sha512-FycRElRNzeFLEfzVUKIyE3ieKphUrQbAbsprnvs0Uu7ZdlBKJG6wQV/9gD77Neu1Etcsyezf9Boyq+MsjuraVA==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -2856,9 +2856,9 @@
@@ -2856,9 +2856,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.82.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",
- "integrity": "sha512-oJarlzFGEkqYY0eFumIRjJUqMpOT70kgbhgeDtA8QHnuMZR3VIGQubM2UlOaqXgkxD5uhdZ/G6jLHbjvUhqTLQ==",
+ "version": "0.90.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0",
+ "integrity": "sha512-uOcYG+w5BVKePmSmCsmhFe6E1CNTr7/vLUqyOLSzsB9Z8D3VNbWsPCwOd5W0eY+3Dd4YhYIf4iT0QTIj9dinGg==",
"license": "MIT",
"license": "MIT",
"bin": {
"bin": {
"openvibe-download": "bin/openvibe-download.js"
"openvibe-download": "bin/openvibe-download.js"
@@ -42,8 +42,8 @@
@@ -42,8 +42,8 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
"onnxruntime-node": "^1.20.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.294.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.82.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.297.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.90.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.115.1",
"pg": "^8.23.1",
"pg": "^8.23.1",
"sanitize-html": "^2.17.7",
"sanitize-html": "^2.17.7",
@@ -27,6 +27,7 @@ const crypto = require('crypto');
@@ -27,6 +27,7 @@ const crypto = require('crypto');
const db = require('../db/database');
const db = require('../db/database');
const config = require('../config');
const config = require('../config');
const billing = require('./billing-client');
const billing = require('./billing-client');
+const { assertionFor } = require('../net/subject-assertion');
const { BillingCallError, subjectFor, ref, api } = billing;
const { BillingCallError, subjectFor, ref, api } = billing;
const base = () => config.baseUrl.replace(/\/+$/, '');
const base = () => config.baseUrl.replace(/\/+$/, '');
@@ -49,6 +50,10 @@ function refOf(out) {
@@ -49,6 +50,10 @@ function refOf(out) {
* calling Billing; a key already used with another body is refused; anything else is (re)sent.
* calling Billing; a key already used with another body is refused; anything else is (re)sent.
*/
*/
async function perform({ action, method = 'POST', path, body, key, liveUserId = null, liveRef = null, trace }) {
async function perform({ action, method = 'POST', path, body, key, liveUserId = null, liveRef = null, trace }) {
+ // The person's own subject assertion (ADR-003, the 2026-11-06 amendment; plan H1): Billing checks
+ // it against the person the money is for. Live mints it from the person's own token on the
+ // request; without that token there is none, and the call travels the old way (the flag decides).
+ const assertion = await assertionFor(trace, personOf(body));
const d = db.getDb();
const d = db.getDb();
const requestJson = JSON.stringify(body || {});
const requestJson = JSON.stringify(body || {});
const row = await d.prepare('SELECT * FROM billing_actions WHERE idempotency_key = ?').get(key);
const row = await d.prepare('SELECT * FROM billing_actions WHERE idempotency_key = ?').get(key);
@@ -62,7 +67,7 @@ async function perform({ action, method = 'POST', path, body, key, liveUserId =
@@ -62,7 +67,7 @@ async function perform({ action, method = 'POST', path, body, key, liveUserId =
const id = (await d.prepare('SELECT id FROM billing_actions WHERE idempotency_key = ?').get(key)).id;
const id = (await d.prepare('SELECT id FROM billing_actions WHERE idempotency_key = ?').get(key)).id;
await d.prepare('UPDATE billing_actions SET attempts = attempts + 1, updated_at = ov_now() WHERE id = ?').run(id);
await d.prepare('UPDATE billing_actions SET attempts = attempts + 1, updated_at = ov_now() WHERE id = ?').run(id);
try {
try {
- const out = await billing.request(method, path, { body, idempotencyKey: key, trace });
+ const out = await billing.request(method, path, { body, idempotencyKey: key, trace, assertion });
await d.prepare(`UPDATE billing_actions SET status = 'done', http_status = 200, billing_ref = ?, response_json = ?, error = NULL, updated_at = ov_now() WHERE id = ?`)
await d.prepare(`UPDATE billing_actions SET status = 'done', http_status = 200, billing_ref = ?, response_json = ?, error = NULL, updated_at = ov_now() WHERE id = ?`)
.run(refOf(out), JSON.stringify(out), id);
.run(refOf(out), JSON.stringify(out), id);
return { out, replayed: !!out._replayed, actionId: id };
return { out, replayed: !!out._replayed, actionId: id };
@@ -131,6 +136,17 @@ function normalizeMessage(message, max) {
@@ -131,6 +136,17 @@ function normalizeMessage(message, max) {
const streamTarget = (streamId) => (streamId ? { service: 'live', type: 'stream', id: String(streamId) } : undefined);
const streamTarget = (streamId) => (streamId ? { service: 'live', type: 'stream', id: String(streamId) } : undefined);
+/** The person a money body is about: the payer, the subscriber, the cash-out's owner. */
+function personOf(body) {
+ const b = body || {};
+ for (const field of ['from', 'subject', 'subscriber', 'on_behalf_of']) {
+ const v = b[field];
+ const id = v && typeof v === 'object' ? v.id : v;
+ if (typeof id === 'string' && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(id)) return id;
+ }
+ return null;
+}
+
/** POST /api/funds/donate under Billing. Same result shape as vibes.donate() plus `balance`. */
/** POST /api/funds/donate under Billing. Same result shape as vibes.donate() plus `balance`. */
async function donate(req, { toUserId, streamId, amount, message, goalId = null }) {
async function donate(req, { toUserId, streamId, amount, message, goalId = null }) {
const vibes = require('./vibes');
const vibes = require('./vibes');
@@ -220,7 +236,7 @@ async function recycle(req, amount) {
@@ -220,7 +236,7 @@ async function recycle(req, amount) {
/** GET /api/funds/balance shape. Throws BillingCallError when Billing cannot answer. */
/** GET /api/funds/balance shape. Throws BillingCallError when Billing cannot answer. */
async function balance(liveUserId, trace) {
async function balance(liveUserId, trace) {
const sid = await subjectFor(liveUserId);
const sid = await subjectFor(liveUserId);
- const b = await api.balance(sid, { trace });
+ const b = await api.balance(sid, { trace, assertion: await assertionFor(trace, sid) });
const per = Number(b.bits_per_usd) || 100;
const per = Number(b.bits_per_usd) || 100;
const usd = (n) => (Math.round(Number(n) || 0) / per).toFixed(2);
const usd = (n) => (Math.round(Number(n) || 0) / per).toFixed(2);
return {
return {
@@ -232,9 +248,9 @@ async function balance(liveUserId, trace) {
@@ -232,9 +248,9 @@ async function balance(liveUserId, trace) {
const LIVE_TYPE = { cashout_request: 'cashout', cashout_denied: 'refund', cashout_paid: 'cashout_paid' };
const LIVE_TYPE = { cashout_request: 'cashout', cashout_denied: 'refund', cashout_paid: 'cashout_paid' };
/** GET /api/funds/history rows in Live's transaction shape (the dashboard renders these). */
/** GET /api/funds/history rows in Live's transaction shape (the dashboard renders these). */
-async function history(liveUserId, limit = 50) {
+async function history(liveUserId, limit = 50, trace = null) {
const sid = await subjectFor(liveUserId);
const sid = await subjectFor(liveUserId);
- const page = await api.transactions(sid, { limit: Math.min(200, Math.max(1, limit)) });
+ const page = await api.transactions(sid, { limit: Math.min(200, Math.max(1, limit)) }, { trace, assertion: await assertionFor(trace, sid) });
const txns = page.transactions || [];
const txns = page.transactions || [];
const people = await billing.liveUsersForSubjects(txns.flatMap((t) => [t.from_subject, t.to_subject]));
const people = await billing.liveUsersForSubjects(txns.flatMap((t) => [t.from_subject, t.to_subject]));
return txns.map((t) => {
return txns.map((t) => {
@@ -311,12 +327,12 @@ function vipClient() {
@@ -311,12 +327,12 @@ function vipClient() {
});
});
return _vip;
return _vip;
}
}
-async function entitlementOf(member, creator) {
+async function entitlementOf(member, creator, trace = null) {
if (process.env.LIVE_VIP_ENTITLEMENTS !== 'off') {
if (process.env.LIVE_VIP_ENTITLEMENTS !== 'off') {
const e = await vipClient().checkEntitlement({ subject: member, creator, product: 'live' });
const e = await vipClient().checkEntitlement({ subject: member, creator, product: 'live' });
if (e && e.status !== 'unknown') return { active: !!e.active, via: 'vip' };
if (e && e.status !== 'unknown') return { active: !!e.active, via: 'vip' };
}
}
- const e = await api.entitlement(member, creator);
+ const e = await api.entitlement(member, creator, { trace, assertion: await assertionFor(trace, member) });
return { active: !!(e && e.active), via: 'billing' };
return { active: !!(e && e.active), via: 'billing' };
}
}
@@ -325,7 +341,7 @@ function viewerOrNull(fn) { return fn().catch((e) => { if (e instanceof BillingC
@@ -325,7 +341,7 @@ function viewerOrNull(fn) { return fn().catch((e) => { if (e instanceof BillingC
async function subscribe(req, { streamer, provider, autoRenewRaw }) {
async function subscribe(req, { streamer, provider, autoRenewRaw }) {
const subscriber = await subjectFor(req.user.id);
const subscriber = await subjectFor(req.user.id);
const to = await subjectFor(streamer.id, { role: 'recipient' });
const to = await subjectFor(streamer.id, { role: 'recipient' });
- const ent = await api.entitlement(subscriber, to, { trace: req.headers });
+ const ent = await api.entitlement(subscriber, to, { trace: req.headers, assertion: await assertionFor(req.headers, subscriber) });
if (ent && ent.active) return { status: 409, body: { error: 'Already subscribed' } };
if (ent && ent.active) return { status: 409, body: { error: 'Already subscribed' } };
if (provider === 'bucks') {
if (provider === 'bucks') {
@@ -406,9 +422,9 @@ function presentSub(s, people) {
@@ -406,9 +422,9 @@ function presentSub(s, people) {
};
};
}
}
-async function mySubscriptions(liveUserId) {
+async function mySubscriptions(liveUserId, trace = null) {
const sid = await subjectFor(liveUserId);
const sid = await subjectFor(liveUserId);
- const out = await api.listSubscriptions({ subscriber: sid });
+ const out = await api.listSubscriptions({ subscriber: sid }, { trace, assertion: await assertionFor(trace, sid) });
const subs = out.subscriptions || [];
const subs = out.subscriptions || [];
const people = await billing.liveUsersForSubjects(subs.map((s) => s.streamer && s.streamer.id));
const people = await billing.liveUsersForSubjects(subs.map((s) => s.streamer && s.streamer.id));
return subs.map((s) => presentSub(s, people));
return subs.map((s) => presentSub(s, people));
@@ -423,12 +439,12 @@ function tierOnePrice(t) {
@@ -423,12 +439,12 @@ function tierOnePrice(t) {
return cents > 0 ? cents : null;
return cents > 0 ? cents : null;
}
}
-async function channelState(streamer, viewer) {
+async function channelState(streamer, viewer, trace = null) {
const to = await viewerOrNull(() => subjectFor(streamer.id, { role: 'recipient' }));
const to = await viewerOrNull(() => subjectFor(streamer.id, { role: 'recipient' }));
if (!to) return { subscribed: false, subscriberCount: 0 };
if (!to) return { subscribed: false, subscriberCount: 0 };
const me = viewer ? await viewerOrNull(() => subjectFor(viewer.id)) : null;
const me = viewer ? await viewerOrNull(() => subjectFor(viewer.id)) : null;
const [ent, list, tiers] = await Promise.all([
const [ent, list, tiers] = await Promise.all([
- me ? await entitlementOf(me, to) : Promise.resolve(null),
+ me ? await entitlementOf(me, to, trace) : Promise.resolve(null),
api.listSubscriptions({ streamer: to, status: 'active' }),
api.listSubscriptions({ streamer: to, status: 'active' }),
api.tiers(to).catch(() => null),
api.tiers(to).catch(() => null),
]);
]);
@@ -439,7 +455,7 @@ async function channelState(streamer, viewer) {
@@ -439,7 +455,7 @@ async function channelState(streamer, viewer) {
async function cancelSubscription(req, { id, streamerId }) {
async function cancelSubscription(req, { id, streamerId }) {
const sid = await subjectFor(req.user.id);
const sid = await subjectFor(req.user.id);
- const out = await api.listSubscriptions({ subscriber: sid }, { trace: req.headers });
+ const out = await api.listSubscriptions({ subscriber: sid }, { trace: req.headers, assertion: await assertionFor(req.headers, sid) });
const subs = (out.subscriptions || []).filter((s) => s.subscriber && s.subscriber.id === sid);
const subs = (out.subscriptions || []).filter((s) => s.subscriber && s.subscriber.id === sid);
let sub = subs.find((s) => String(s.id) === String(id));
let sub = subs.find((s) => String(s.id) === String(id));
if (!sub && streamerId) {
if (!sub && streamerId) {
@@ -52,10 +52,11 @@ function connectionNeverMade(err) {
@@ -52,10 +52,11 @@ function connectionNeverMade(err) {
}
}
/**
/**
- * One call. opts: { body, idempotencyKey, trace (incoming request headers, for traceparent) }.
+ * One call. opts: { body, idempotencyKey, trace (incoming request headers, for traceparent),
+ * assertion (the person's own subject assertion, ADR-003's 2026-11-06 amendment) }.
* Resolves the parsed JSON body of a 2xx (plus `_replayed` when Billing replayed the key).
* Resolves the parsed JSON body of a 2xx (plus `_replayed` when Billing replayed the key).
*/
*/
-async function request(method, path, { body, idempotencyKey, trace, retried = false } = {}) {
+async function request(method, path, { body, idempotencyKey, trace, assertion, retried = false } = {}) {
const write = method !== 'GET';
const write = method !== 'GET';
if (write && !KEY_RE.test(String(idempotencyKey || ''))) throw new Error(`billing-client: ${method} ${path} needs a valid Idempotency-Key`);
if (write && !KEY_RE.test(String(idempotencyKey || ''))) throw new Error(`billing-client: ${method} ${path} needs a valid Idempotency-Key`);
let auth;
let auth;
@@ -66,6 +67,8 @@ async function request(method, path, { body, idempotencyKey, trace, retried = fa
@@ -66,6 +67,8 @@ async function request(method, path, { body, idempotencyKey, trace, retried = fa
}
}
const headers = { Accept: 'application/json', ...auth, ...http.outboundHeaders(http.requestContext(trace || {})) };
const headers = { Accept: 'application/json', ...auth, ...http.outboundHeaders(http.requestContext(trace || {})) };
if (write) headers['Idempotency-Key'] = idempotencyKey;
if (write) headers['Idempotency-Key'] = idempotencyKey;
+ // The person's own subject assertion: Billing checks it against the person the money is for.
+ if (assertion) headers['X-OV-Subject-Assertion'] = String(assertion);
let payload;
let payload;
if (body !== undefined) { headers['Content-Type'] = 'application/json'; payload = JSON.stringify(body); }
if (body !== undefined) { headers['Content-Type'] = 'application/json'; payload = JSON.stringify(body); }
let res;
let res;
@@ -81,7 +84,7 @@ async function request(method, path, { body, idempotencyKey, trace, retried = fa
@@ -81,7 +84,7 @@ async function request(method, path, { body, idempotencyKey, trace, retried = fa
try { json = text ? JSON.parse(text) : {}; } catch { /* not JSON */ }
try { json = text ? JSON.parse(text) : {}; } catch { /* not JSON */ }
if (res.status === 401 && !retried) { // token rotated/expired under us: once more with a fresh one
if (res.status === 401 && !retried) { // token rotated/expired under us: once more with a fresh one
principal.invalidate(AUDIENCE);
principal.invalidate(AUDIENCE);
- return await request(method, path, { body, idempotencyKey, trace, retried: true });
+ return await request(method, path, { body, idempotencyKey, trace, assertion, retried: true });
}
}
if (res.ok) {
if (res.ok) {
const out = json || {};
const out = json || {};
@@ -77,7 +77,7 @@ router.post('/subscribe', requireAuth, money.guardWrite, async (req, res) => {
@@ -77,7 +77,7 @@ router.post('/subscribe', requireAuth, money.guardWrite, async (req, res) => {
// My subscriptions
// My subscriptions
router.get('/subscriptions/mine', requireAuth, async (req, res) => {
router.get('/subscriptions/mine', requireAuth, async (req, res) => {
- try { return res.json({ subscriptions: await billingActions.mySubscriptions(req.user.id) }); }
+ try { return res.json({ subscriptions: await billingActions.mySubscriptions(req.user.id, req.headers) }); }
catch (err) {
catch (err) {
if (billingActions.sendError(res, err, { unavailable: 'Subscriptions unavailable right now — the billing service is not answering.' })) return;
if (billingActions.sendError(res, err, { unavailable: 'Subscriptions unavailable right now — the billing service is not answering.' })) return;
return res.status(503).json({ error: 'Subscriptions unavailable right now', unavailable: true });
return res.status(503).json({ error: 'Subscriptions unavailable right now', unavailable: true });
@@ -97,7 +97,7 @@ router.get('/channel/:username', optionalAuth, async (req, res) => {
@@ -97,7 +97,7 @@ router.get('/channel/:username', optionalAuth, async (req, res) => {
const streamer = await db.getUserByUsername(req.params.username);
const streamer = await db.getUserByUsername(req.params.username);
if (!streamer) return res.status(404).json({ error: 'Not found' });
if (!streamer) return res.status(404).json({ error: 'Not found' });
let subscribed = false, subscriberCount = null, billingUnavailable = false, billingPriceCents = null;
let subscribed = false, subscriberCount = null, billingUnavailable = false, billingPriceCents = null;
- try { ({ subscribed, subscriberCount, priceCents: billingPriceCents } = await billingActions.channelState(streamer, req.user || null)); }
+ try { ({ subscribed, subscriberCount, priceCents: billingPriceCents } = await billingActions.channelState(streamer, req.user || null, req.headers)); }
catch { billingUnavailable = true; }
catch { billingUnavailable = true; }
const priceUsd = billingPriceCents ? billingPriceCents / 100 : await pay._num('sub_price_usd', 4.99);
const priceUsd = billingPriceCents ? billingPriceCents / 100 : await pay._num('sub_price_usd', 4.99);
let powerchat = { direct: false, site: false };
let powerchat = { direct: false, site: false };
@@ -169,7 +169,7 @@ router.post('/recycle', requireAuth, money.guardWrite, async (req, res) => {
@@ -169,7 +169,7 @@ router.post('/recycle', requireAuth, money.guardWrite, async (req, res) => {
// ── Transaction History ──────────────────────────────────────
// ── Transaction History ──────────────────────────────────────
router.get('/history', requireAuth, async (req, res) => {
router.get('/history', requireAuth, async (req, res) => {
const limit = Math.min(parseInt(req.query.limit || '50'), 200);
const limit = Math.min(parseInt(req.query.limit || '50'), 200);
- try { return res.json({ transactions: await billingActions.history(req.user.id, limit) }); }
+ try { return res.json({ transactions: await billingActions.history(req.user.id, limit, req.headers) }); }
catch (err) {
catch (err) {
if (billingActions.sendError(res, err, { unavailable: 'Vibes history unavailable right now — the billing service is not answering.' })) return;
if (billingActions.sendError(res, err, { unavailable: 'Vibes history unavailable right now — the billing service is not answering.' })) return;
return res.status(503).json({ error: 'Vibes history unavailable right now', unavailable: true });
return res.status(503).json({ error: 'Vibes history unavailable right now', unavailable: true });
new file mode 100644
new file mode 100644
@@ -0,0 +1,63 @@
@@ -0,0 +1,63 @@
+'use strict';
+
+/**
+ * The person's own subject assertion for Live's money calls (ADR-003, the 2026-11-06 amendment; plan H1).
+ *
+ * Live names the person it acts for — the person who signed in — and Billing believes it. On a money
+ * route that is only Live's word: a compromised Live could tip in anyone's name or read anyone's
+ * receipts. A subject assertion is the person's own answer to the same question: Network signs their
+ * access token into a 60-second proof (ES256) that names them, and Billing checks that proof instead
+ * of Live's word.
+ *
+ * Live holds the person's Network access token on the request (the session cookie `ov_token`, or the
+ * Authorization header), so it can ask Network to sign it. Without that token there is nothing to
+ * sign and no assertion is minted: the call then travels the old way, which Billing's
+ * BILLING_REQUIRE_SUBJECT_ASSERTION flag decides on.
+ *
+ * The assertion is minted per call and never cached: Billing refuses a replayed one (the jti is
+ * remembered), so two money calls must not share it.
+ */
+const { createSubjectAssertionMinter } = require('openvibe-sdk/assertions');
+
+const NETWORK_INTERNAL_URL = (process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+const CLIENT_ID = process.env.OV_OAUTH_CLIENT_ID || 'live';
+const CLIENT_SECRET = process.env.OV_OAUTH_CLIENT_SECRET || '';
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+/** The person's own Network access token on this request, or null (no session, or a Live API token). */
+function personToken(headers) {
+ const h = headers || {};
+ const auth = String(h.authorization || '');
+ if (/^Bearer\s+/i.test(auth)) {
+ const token = auth.slice(7).trim();
+ // A Live API token (hbt_…) is not a Network session; Network would refuse it anyway.
+ if (token && !/^hbt_/.test(token)) return token;
+ }
+ // The session cookie (the person's Network token, JS-readable for the navbar).
+ for (const part of String(h.cookie || '').split(';')) {
+ const idx = part.indexOf('=');
+ if (idx < 0) continue;
+ if (part.slice(0, idx).trim() !== 'ov_token') continue;
+ const value = decodeURIComponent(part.slice(idx + 1).trim());
+ if (value) return value;
+ }
+ return null;
+}
+
+const minter = createSubjectAssertionMinter({
+ internalUrl: NETWORK_INTERNAL_URL,
+ oauth: { clientId: CLIENT_ID, clientSecret: CLIENT_SECRET },
+});
+
+/**
+ * The assertion for the person this request acts for, or null when it cannot be had (no session,
+ * no grant, Network down). It never throws: a money call must not fail because a proof is missing.
+ */
+async function assertionFor(headers, subject) {
+ if (!subject || !SUBJECT_RE.test(String(subject))) return null;
+ const userToken = personToken(headers);
+ if (!userToken) return null;
+ return await minter.mint({ userToken, subject, audience: 'openvibe.billing' });
+}
+
+module.exports = { personToken, assertionFor, SUBJECT_RE };
@@ -14,6 +14,7 @@
@@ -14,6 +14,7 @@
*/
*/
const http = require('http');
const http = require('http');
const crypto = require('crypto');
const crypto = require('crypto');
+const jwt = require('jsonwebtoken');
const { serviceAuth, http: ovHttp, ids } = require('openvibe-contracts');
const { serviceAuth, http: ovHttp, ids } = require('openvibe-contracts');
const read = (req) => new Promise((resolve) => { const c = []; req.on('data', (d) => c.push(d)); req.on('end', () => resolve(Buffer.concat(c).toString('utf8'))); });
const read = (req) => new Promise((resolve) => { const c = []; req.on('data', (d) => c.push(d)); req.on('end', () => resolve(Buffer.concat(c).toString('utf8'))); });
@@ -21,6 +22,9 @@ const send = (res, status, obj, headers = {}) => { res.writeHead(status, { 'Cont
@@ -21,6 +22,9 @@ const send = (res, status, obj, headers = {}) => { res.writeHead(status, { 'Cont
const problem = (res, status, code, detail, extra) => { res.writeHead(status, { 'Content-Type': 'application/problem+json' }); res.end(JSON.stringify(ovHttp.problem(status, code, { detail, extra }))); };
const problem = (res, status, code, detail, extra) => { res.writeHead(status, { 'Content-Type': 'application/problem+json' }); res.end(JSON.stringify(ovHttp.problem(status, code, { detail, extra }))); };
const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve(`http://127.0.0.1:${server.address().port}`)));
const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve(`http://127.0.0.1:${server.address().port}`)));
+/** The issuer Network's tokens carry in the tests. */
+const ISSUER = 'https://openvibe.network';
+
const LIVE_BILLING_GRANTS = ['billing.intent.create', 'billing.transfer.create', 'billing.balance.read', 'billing.cashout.request', 'billing.subscription.manage', 'billing.entitlement.check'];
const LIVE_BILLING_GRANTS = ['billing.intent.create', 'billing.transfer.create', 'billing.balance.read', 'billing.cashout.request', 'billing.subscription.manage', 'billing.entitlement.check'];
async function startNetwork() {
async function startNetwork() {
@@ -38,6 +42,27 @@ async function startNetwork() {
@@ -38,6 +42,27 @@ async function startNetwork() {
const claims = { iss: 'https://openvibe.network', sub: `svc:${body.get('client_id')}`, actor_type: 'service', aud: [aud], cap: grants[aud] || [], iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(6).toString('hex')}` };
const claims = { iss: 'https://openvibe.network', sub: `svc:${body.get('client_id')}`, actor_type: 'service', aud: [aud], cap: grants[aud] || [], iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(6).toString('hex')}` };
return send(res, 200, { access_token: serviceAuth.signServiceToken(claims, keys.privateKey), expires_in: 300 });
return send(res, 200, { access_token: serviceAuth.signServiceToken(claims, keys.privateKey), expires_in: 300 });
}
}
+ if (req.url === '/internal/subject-assertions') {
+ // The mint route (ADR-003, the 2026-11-06 amendment): a first-party service hands over the
+ // person's own token and gets back a 60-second assertion naming the person.
+ const v = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey: keys.publicKey, issuer: ISSUER, audience: 'openvibe.network' });
+ if (!v.ok) return send(res, 401, { code: v.code });
+ if (!/^svc:[a-z][a-z0-9-]{1,39}$/.test(String(v.claims.sub || ''))) return send(res, 403, { code: 'assertion.not_a_service' });
+ const body = JSON.parse(raw || '{}');
+ if (!body.user_token) return send(res, 401, { code: 'assertion.no_user_token' });
+ let claims;
+ try { claims = jwt.verify(String(body.user_token), keys.publicKey, { issuer: ISSUER }); } catch { return send(res, 401, { code: 'assertion.invalid_user_token' }); }
+ const subject = claims.subject_id;
+ if (body.subject && body.subject !== subject) return send(res, 403, { code: 'assertion.subject_mismatch' });
+ const audience = Array.isArray(body.audience) ? body.audience : [body.audience];
+ const forServices = [...new Set([v.claims.sub, ...(Array.isArray(body.for) ? body.for : [])])];
+ const jti = `as_${crypto.randomBytes(6).toString('hex')}`;
+ const now = Math.floor(Date.now() / 1000);
+ const b64 = (b) => Buffer.from(b).toString('base64url');
+ const header = b64(JSON.stringify({ alg: 'ES256', typ: 'JWT', kid: 'ov-network-assertion-1' }));
+ const payload = b64(JSON.stringify({ iss: ISSUER, sub: subject, aud: audience, act: v.claims.sub, for: forServices, jti, iat: now, exp: now + 60 }));
+ return send(res, 200, { assertion: `${header}.${payload}.${b64('signed')}`, subject, audience: body.audience, act: v.claims.sub, for: forServices, jti, expires_in: 60 });
+ }
if (req.url === '/internal/identity/resolve-batch') {
if (req.url === '/internal/identity/resolve-batch') {
const { ids: want } = JSON.parse(raw || '{}');
const { ids: want } = JSON.parse(raw || '{}');
return send(res, 200, { results: Object.fromEntries((want || []).map((id) => [id, legacy[id] ? { subject: { type: 'user', id: legacy[id] } } : null])) });
return send(res, 200, { results: Object.fromEntries((want || []).map((id) => [id, legacy[id] ? { subject: { type: 'user', id: legacy[id] } } : null])) });
@@ -45,7 +70,9 @@ async function startNetwork() {
@@ -45,7 +70,9 @@ async function startNetwork() {
send(res, 404, {});
send(res, 404, {});
});
});
const url = await listen(server);
const url = await listen(server);
- return { url, publicKey: keys.publicKey, grants, legacy, tokens, close: () => new Promise((r) => server.close(r)) };
+ /** A Network user token (the session a person's browser holds). */
+ const signUser = (subject, username = 'someone') => jwt.sign({ sub: String(subject), subject_id: subject, username, display_name: username, role: 'user' }, keys.privateKey, { algorithm: 'RS256', issuer: ISSUER, expiresIn: '1h' });
+ return { url, publicKey: keys.publicKey, grants, legacy, tokens, signUser, close: () => new Promise((r) => server.close(r)) };
}
}
const CAP = [
const CAP = [
@@ -172,7 +199,7 @@ async function startBilling({ network }) {
@@ -172,7 +199,7 @@ async function startBilling({ network }) {
const auth = String(req.headers.authorization || '');
const auth = String(req.headers.authorization || '');
const v = serviceAuth.verifyServiceToken(auth.slice(7), { publicKey: network.publicKey, audience: 'openvibe.billing' });
const v = serviceAuth.verifyServiceToken(auth.slice(7), { publicKey: network.publicKey, audience: 'openvibe.billing' });
const rule = CAP.find(([m, re]) => m === req.method && re.test(path));
const rule = CAP.find(([m, re]) => m === req.method && re.test(path));
- const call = { method: req.method, path, query: Object.fromEntries(u.searchParams), key: req.headers['idempotency-key'] || null, body, cap: rule ? rule[2] : undefined, principal: v.ok ? v.claims.sub : null, traceparent: req.headers.traceparent || null };
+ const call = { method: req.method, path, query: Object.fromEntries(u.searchParams), key: req.headers['idempotency-key'] || null, body, cap: rule ? rule[2] : undefined, principal: v.ok ? v.claims.sub : null, traceparent: req.headers.traceparent || null, assertion: req.headers['x-ov-subject-assertion'] || null };
calls.push(call);
calls.push(call);
if (state.delayMs) await new Promise((r) => setTimeout(r, state.delayMs));
if (state.delayMs) await new Promise((r) => setTimeout(r, state.delayMs));
if (!v.ok) return problem(res, 401, v.code, v.reason);
if (!v.ok) return problem(res, 401, v.code, v.reason);
new file mode 100644
new file mode 100644
@@ -0,0 +1,118 @@
@@ -0,0 +1,118 @@
+'use strict';
+/**
+ * The person's own subject assertion on Live's money calls (ADR-003, the 2026-11-06 amendment; plan H1).
+ *
+ * Live names the person it acts for, and Billing believes it. On a money route that is only Live's
+ * word, so Live asks Network to sign the person's own token into a 60-second assertion and carries
+ * it on (X-OV-Subject-Assertion). Without the person's token there is nothing to sign, and the call
+ * travels the old way — which Billing's BILLING_REQUIRE_SUBJECT_ASSERTION flag decides on.
+ */
+const assert = require('assert');
+const http = require('http');
+const { startNetwork, startBilling } = require('./billing-stub');
+
+process.env.NODE_ENV = 'test';
+process.env.OV_OAUTH_CLIENT_ID = 'live';
+process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';
+process.env.OV_BILLING_TIMEOUT_MS = '1500';
+const quiet = console.log;
+console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };
+console.warn = () => {};
+console.error = () => {};
+
+const SID = {
+ ann: 'usr_01JAB2C3D4E5F6G7H8J9K0MNA1',
+ bob: 'usr_01JAB2C3D4E5F6G7H8J9K0MNB2',
+};
+
+let failures = 0;
+async function check(name, fn) {
+ try { await fn(); quiet(' ✓', name); }
+ catch (e) { failures++; quiet(' ✗', name, '\n ', (e.stack || String(e)).split('\n').slice(0, 9).join('\n ')); }
+}
+
+/** The claims of an assertion (the stub signs one; the signature is not what this test is about). */
+const claimsOf = (assertion) => JSON.parse(Buffer.from(String(assertion).split('.')[1], 'base64url').toString('utf8'));
+
+(async () => {
+ const network = await startNetwork();
+ const billing = await startBilling({ network });
+ process.env.OV_NETWORK_INTERNAL_URL = network.url;
+ process.env.OV_BILLING_INTERNAL_URL = billing.url;
+
+ const db = require('../server/db/database');
+ await db.initDb();
+ const raw = db.getDb();
+ const addUser = async (id, username, role) => await raw.prepare(
+ `INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) OVERRIDING SYSTEM VALUE
+ VALUES (?, ?, ?, ?, 'x', ?, 0)`).run(id, username, username, `${username}@x`, role);
+ await addUser(2, 'ann', 'user');
+ await addUser(3, 'bob', 'streamer');
+ const link = async (uid, sid) => await raw.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (?, 'network', ?, ?)").run(uid, String(100 + uid), sid);
+ await link(2, SID.ann); await link(3, SID.bob);
+ await db.ensureChannel(3);
+ const streamB = Number((await db.createStream({ user_id: 3, channel_id: (await db.getChannelByUserId(3)).id, title: 'B', protocol: 'webrtc' })).lastInsertRowid);
+
+ const auth = require('../server/auth/auth');
+ const signIn = async (req) => { const id = Number(req.headers['x-test-user'] || 0); const u = id ? await db.getUserById(id) : null; if (u) req.user = u; return u; };
+ auth.requireAuth = (req, res, next) => { signIn(req).then((u) => u ? next() : res.status(401).json({ error: 'Authentication required' })).catch(next); };
+ auth.optionalAuth = (req, res, next) => { signIn(req).then(() => next()).catch(next); };
+
+ const express = require('express');
+ const app = express();
+ app.use(express.json());
+ app.use((req, res, next) => { signIn(req).then(() => next()).catch(next); });
+ app.use('/api/funds', require('../server/monetization/routes'));
+ const server = http.createServer(app);
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const call = async (method, p, user, body, headers = {}) => {
+ const res = await fetch(`http://127.0.0.1:${server.address().port}${p}`, {
+ method, redirect: 'manual',
+ headers: { 'content-type': 'application/json', ...(user ? { 'x-test-user': String(user) } : {}), ...headers },
+ body: body ? JSON.stringify(body) : undefined,
+ });
+ const text = await res.text();
+ let json = null; try { json = JSON.parse(text); } catch { /* */ }
+ return { status: res.status, json, text };
+ };
+ const callsSince = (n) => billing.calls.slice(n);
+
+ billing.fund(SID.ann, 3000);
+
+ await check('a money call for a person carries the person\'s own assertion', async () => {
+ const n = billing.calls.length;
+ const r = await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 150 }, { cookie: `ov_token=${network.signUser(SID.ann, 'ann')}` });
+ assert.strictEqual(r.status, 200, r.text);
+ const c = callsSince(n);
+ assert.strictEqual(c.length, 1, 'one Billing call');
+ assert.ok(c[0].assertion, 'the call carried an assertion');
+ const claims = claimsOf(c[0].assertion);
+ assert.strictEqual(claims.sub, SID.ann, 'the assertion names the person');
+ assert.deepStrictEqual(claims.aud, ['openvibe.billing']);
+ assert.strictEqual(claims.act, 'svc:live', 'Live is the service it was minted for');
+ });
+
+ await check('without the person\'s token the money call travels the old way (no assertion)', async () => {
+ const n = billing.calls.length;
+ const r = await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 100 });
+ assert.strictEqual(r.status, 200, r.text);
+ const c = callsSince(n);
+ assert.strictEqual(c.length, 1, 'one Billing call');
+ assert.strictEqual(c[0].assertion, null, 'no assertion without the person\'s token');
+ assert.deepStrictEqual(c[0].body.from, { type: 'user', id: SID.ann }, 'the person is still named in the body');
+ });
+
+ await check('two money calls get two assertions (a replayed one is refused)', async () => {
+ const token = network.signUser(SID.ann, 'ann');
+ const n = billing.calls.length;
+ await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 100 }, { cookie: `ov_token=${token}` });
+ await call('POST', '/api/funds/donate', 2, { streamer_id: 3, stream_id: streamB, amount: 100 }, { cookie: `ov_token=${token}` });
+ const c = callsSince(n);
+ assert.strictEqual(c.length, 2);
+ assert.ok(c[0].assertion && c[1].assertion);
+ assert.notStrictEqual(c[0].assertion, c[1].assertion, 'each call has its own assertion');
+ });
+
+ quiet(failures ? `\n${failures} failed` : '\nall passed');
+ process.exit(failures ? 1 : 0);
+})().catch((err) => { console.error(err); process.exit(1); });