openvibers/openvibe.run Commits

A11: apply Link browsing choices to cloud desks (#216)

OP OpenVibers committed 4 days ago · c5146d4858 · 7 files, +195 −7

* Apply Link browsing choices to cloud desk sessions

* Keep ended sessions closed when a late desk reply arrives

Changes

ddiff --git a/docs/link-egress.md b/docs/link-egress.md
nnew file mode 100644
iindex 0000000..f2497d7
---- /dev/null
++++ b/docs/link-egress.md
@@@ -0,0 +1,19 @@
++# Cloud browsing through Link
++
++The desk client registers each canonical Run cloud session with Network before asking Bot to start it. Network
++chooses an online device owned by that same person, or returns the saved offline choice. A wait decision leaves
++the session waiting; a Network failure never silently selects a direct route. Personal-machine desks keep their
++existing path.
++
++The cloud Node reads the decision assigned to its own machine and desk, then starts the browser behind a local
++proxy in the desk's isolated network. Session credentials stay in the Node proxy process. Run stores only the
++display label, "Going out through: …", in `run_sessions.egress_label` and exposes it in placement reasons.
++
++Each desk check renews authorization for two minutes. Ending the desk revokes it before asking Bot to stop;
++if Network is unavailable, Bot still receives the stop and the credential expires without renewal. Link checks
++open streams every five seconds. Session credentials are neither public API fields nor events nor logs.
++
++Deploy Network #702 and Connect #76, then the updated Node and computer plugin on cloud hosts and home devices,
++before this Run integration. This is the common Run browser-desk path used by cloud session callers. Cloud
++products which do not yet create a Run session must adopt that path to inherit its owner and browsing policy.
++Migration 0016 is expand-only and reserves 0015 for the open blind-fill PR; its files are untouched.
ddiff --git a/migrations/0016_link_egress.sql b/migrations/0016_link_egress.sql
nnew file mode 100644
iindex 0000000..aaebad3
---- /dev/null
++++ b/migrations/0016_link_egress.sql
@@@ -0,0 +1,3 @@
++-- phase: expand
++-- 0015 is reserved by the open blind-fill PR. Only display text is stored; never proxy credentials.
++ALTER TABLE run_sessions ADD COLUMN egress_label text;
ddiff --git a/server/sessions/desks.js b/server/sessions/desks.js
iindex 187032e..8faaff8 100644
---- a/server/sessions/desks.js
++++ b/server/sessions/desks.js
@@@ -37,6 +37,9 @@ function createDeskClient({ config, fetchImpl = globalThis.fetch, now = () => Da
? createServiceTokenClient({ tokenUrl: `${config.network.internalUrl}/oauth/token`, clientId, clientSecret, audience: botAudience, scope: SCOPE, fetch: fetchImpl, now })
: null;
const enabled = Boolean(tokens && botUrl);
++ const egress = require('./link-egress').createLinkEgress({ config, fetchImpl, now });
++ const cloud = new Set();
++ const assigned = new Map();
const note = enabled ? `sessions lease desks on Bot: ${botUrl}, capability ${SCOPE}` : 'sessions are off: OV_OAUTH_CLIENT_SECRET is unset, so Run cannot mint its Bot token';
async function call(method, path, subject, body) {
@@@ -65,11 +68,43 @@ function createDeskClient({ config, fetchImpl = globalThis.fetch, now = () => Da
return {
enabled,
note,
-- async hosts(subject) { const out = await call('GET', '/api/v1/desk-hosts', subject); return out && Array.isArray(out.hosts) ? out.hosts : []; },
-- lease: (machine, subject, body) => call('POST', `/api/v1/machines/${m(machine)}/desk-leases`, subject, body),
-- get: (machine, desk, subject) => call('GET', `/api/v1/machines/${m(machine)}/desk-leases/${m(desk)}`, subject),
++ async hosts(subject) {
++ const out = await call('GET', '/api/v1/desk-hosts', subject);
++ const hosts = out && Array.isArray(out.hosts) ? out.hosts : [];
++ for (const h of hosts) { if (h.shared || h.trust !== 'user-owned') cloud.add(h.machine_id); }
++ return hosts;
++ },
++ async lease(machine, subject, body) {
++ // Only canonical Run sessions enter the per-session Link path. Bot remains the validator for other lease requests.
++ if (cloud.has(machine) && /^ses_[0-9A-HJKMNP-TV-Z]{26}$/.test(body.session || '')) {
++ await egress.prepare(machine, subject, body);
++ assigned.set(`${machine}:${body.desk}`, body.session);
++ }
++ try { return await call('POST', `/api/v1/machines/${m(machine)}/desk-leases`, subject, body); }
++ catch (e) { if (assigned.has(`${machine}:${body.desk}`)) await egress.end(body.session).catch(() => {}); throw e; }
++ },
++ async get(machine, desk, subject) {
++ const lease = await call('GET', `/api/v1/machines/${m(machine)}/desk-leases/${m(desk)}`, subject);
++ if (lease && (lease.shared || cloud.has(machine) || assigned.has(`${machine}:${desk}`))) {
++ assigned.set(`${machine}:${desk}`, lease.session);
++ await egress.renew(lease.session);
++ }
++ return lease;
++ },
act: (machine, desk, subject, cmd) => call('POST', `/api/v1/machines/${m(machine)}/desk-leases/${m(desk)}/commands`, subject, cmd),
-- end: (machine, desk, subject) => call('DELETE', `/api/v1/machines/${m(machine)}/desk-leases/${m(desk)}`, subject),
++ async end(machine, desk, subject) {
++ const key = `${machine}:${desk}`;
++ let session = assigned.get(key);
++ if (!session) {
++ const lease = await call('GET', `/api/v1/machines/${m(machine)}/desk-leases/${m(desk)}`, subject);
++ if (lease?.shared) session = lease.session;
++ }
++ // Ending the desk must still reach Bot if Network is briefly unavailable. The credential also
++ // expires within two minutes without renewal, and Link rechecks live streams every five seconds.
++ if (session) await egress.end(session).catch(() => {});
++ assigned.delete(key);
++ return call('DELETE', `/api/v1/machines/${m(machine)}/desk-leases/${m(desk)}`, subject);
++ },
status: () => ({ enabled, note, bot_url: botUrl, capability: SCOPE }),
};
}
ddiff --git a/server/sessions/link-egress.js b/server/sessions/link-egress.js
nnew file mode 100644
iindex 0000000..cc7c6d2
---- /dev/null
++++ b/server/sessions/link-egress.js
@@@ -0,0 +1,46 @@
++'use strict';
++
++const crypto = require('crypto');
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++const revokers = new Map();
++const labels = new Map(); // Display only; never an authorization cache or a credential store.
++
++function createLinkEgress({ config, fetchImpl = globalThis.fetch, now = () => Date.now() }) {
++ const tokens = config.oauth.clientSecret ? createServiceTokenClient({ tokenUrl: `${config.network.internalUrl}/oauth/token`,
++ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.network',
++ scope: 'network.registry.read', fetch: fetchImpl, now }) : null;
++ async function call(method, path, body) {
++ if (!tokens) throw Object.assign(new Error('Browsing connection unavailable'), { retryable: true });
++ for (let attempt = 0; attempt < 2; attempt++) {
++ const r = await fetchImpl(`${config.network.internalUrl}/internal/egress/sessions/${path}`, { method,
++ headers: { ...(await tokens.authHeaders()), 'Content-Type': 'application/json' },
++ ...(body ? { body: JSON.stringify(body) } : {}), signal: AbortSignal.timeout(10_000), redirect: 'error' });
++ if (r.status === 401 && attempt === 0) { tokens.invalidate(); continue; }
++ if (!r.ok) throw Object.assign(new Error('Browsing connection unavailable'), { code: 'run.egress.unavailable', retryable: true });
++ return r.status === 204 ? null : r.json();
++ }
++ }
++ return {
++ async prepare(machine, owner, body) {
++ const route = await call('PUT', encodeURIComponent(body.session), { owner, machine, desk: body.desk,
++ credential: crypto.randomBytes(32).toString('base64url'), expires_at: now() + 120_000 });
++ revokers.set(body.session, () => call('DELETE', encodeURIComponent(body.session)));
++ if (route.mode === 'wait') throw Object.assign(new Error('Waiting for my computer'), { code: 'run.egress.waiting', retryable: true });
++ if (!['link', 'direct'].includes(route.mode)) throw Object.assign(new Error('Browsing connection unavailable'), { retryable: true });
++ labels.set(body.session, route.mode === 'link' ? String(route.device_name || 'My computer').slice(0, 100) : "OpenVibe's address");
++ return route;
++ },
++ async renew(session) {
++ await call('POST', `${encodeURIComponent(session)}/renew`);
++ revokers.set(session, () => call('DELETE', encodeURIComponent(session)));
++ },
++ async end(session) { labels.delete(session); revokers.delete(session); await call('DELETE', encodeURIComponent(session)); },
++ };
++}
++
++function goingOutThrough(session) { const name = labels.get(session); return name ? `Going out through: ${name}` : null; }
++async function revokeKnownSession(session) {
++ const revoke = revokers.get(session); revokers.delete(session); labels.delete(session);
++ if (revoke) await revoke();
++}
++module.exports = { createLinkEgress, goingOutThrough, revokeKnownSession };
ddiff --git a/server/sessions/store.js b/server/sessions/store.js
iindex 8605d9e..0d23946 100644
---- a/server/sessions/store.js
++++ b/server/sessions/store.js
@@@ -14,7 +14,7 @@ const { ids } = require('openvibe-contracts');
const COLUMNS = ['id', 'owner', 'via', 'kind', 'state', 'width', 'height', 'ttl_s', 'idle_s', 'machine_asked', 'requirements', 'idempotency_key',
'compute_pool', 'compute_strategy', 'compute_decision', 'request_digest', 'machine', 'desk', 'trust', 'offer_id', 'reasons', 'created_at', 'started_at', 'last_used_at', 'ended_at', 'end_reason',
-- 'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'control_lease', 'template', 'updated_at'];
++ 'end_detail', 'metered_minutes', 'lease_checked_at', 'task_id', 'curtain', 'control_lease', 'template', 'egress_label', 'updated_at'];
const JSON_COLUMNS = new Set(['requirements', 'reasons', 'compute_decision', 'control_lease']);
const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
@@@ -54,7 +54,7 @@ function present(row, now = Date.now()) {
idle_s: Number(row.idle_s),
placement: placed ? {
machine: row.machine, desk: row.desk, trust: row.trust, offer_id: row.offer_id || null, region: null,
-- reasons: json(row.reasons, []).slice(0, 8).map((r) => String(r).slice(0, 200)),
++ reasons: [row.egress_label || require('./link-egress').goingOutThrough(row.id), ...json(row.reasons, [])].filter(Boolean).slice(0, 8).map((r) => String(r).slice(0, 200)),
} : null,
timings: { created_at: iso(row.created_at), started_at: iso(row.started_at), last_used_at: iso(row.last_used_at), ended_at: iso(row.ended_at) },
end: row.state === 'ended' ? { reason: row.end_reason || 'ended', detail: row.end_detail ? String(row.end_detail).slice(0, 500) : null } : null,
@@@ -109,10 +109,14 @@ const open = async (db, limit = 200) => (await db.many("SELECT * FROM run_sessio
const openCount = async (db, owner) => Number((await db.one("SELECT count(*)::int AS n FROM run_sessions WHERE owner = $1 AND state <> 'ended'", [owner])).n);
async function update(t, id, fields) {
++ const label = require('./link-egress').goingOutThrough(id);
++ if (label) fields = { ...fields, egress_label: label };
const cols = Object.keys(fields).filter((k) => COLUMNS.includes(k) && k !== 'id');
if (!cols.length) return get(t, id);
const vals = cols.map((c) => (JSON_COLUMNS.has(c) && fields[c] != null ? JSON.stringify(fields[c]) : fields[c]));
-- const row = await t.maybe(`UPDATE run_sessions SET ${cols.map((c, i) => `${c} = $${i + 2}`).join(', ')} WHERE id = $1 RETURNING *`, [id, ...vals]);
++ // A late placement reply must never revive a session after its owner ended it.
++ const row = await t.maybe(`UPDATE run_sessions SET ${cols.map((c, i) => `${c} = $${i + 2}`).join(', ')} WHERE id = $1 AND state <> 'ended' RETURNING *`, [id, ...vals]) || await t.maybe('SELECT * FROM run_sessions WHERE id = $1', [id]);
++ if (row?.state === 'ended') void require('./link-egress').revokeKnownSession(id).catch(() => {});
return fromDb(row);
}
ddiff --git a/test/link-egress-desk.test.js b/test/link-egress-desk.test.js
nnew file mode 100644
iindex 0000000..375415c
---- /dev/null
++++ b/test/link-egress-desk.test.js
@@@ -0,0 +1,50 @@
++'use strict';
++const assert = require('assert/strict');
++const { ids } = require('openvibe-contracts');
++const { createDeskClient } = require('../server/sessions/desks');
++const { testDb } = require('./helpers/db');
++const store = require('../server/sessions/store');
++
++(async () => {
++ const dbtest = await testDb();
++ try {
++ const machine = `dev_${ids.ulid()}`, owner = `usr_${ids.ulid()}`, session = `ses_${ids.ulid()}`;
++ const body = { session, desk: store.deskOf(session), kind: 'browser', width: 1280, height: 800 };
++ const calls = [];
++ let mode = 'wait';
++ const config = { oauth: { clientId: 'run', clientSecret: 'test-only' }, network: { internalUrl: 'https://network.test' },
++ dispatch: { botUrl: 'https://bot.test', botAudience: 'openvibe.bot', timeoutMs: 1000 } };
++ const client = createDeskClient({ config, fetchImpl: async (url, options) => {
++ const json = value => new Response(JSON.stringify(value), { status: 200 });
++ if (url.endsWith('/oauth/token')) return json({ access_token: 'test-only', token_type: 'Bearer', expires_in: 3600 });
++ calls.push({ url, method: options.method, body: options.body && JSON.parse(options.body) });
++ if (url.endsWith('/desk-hosts')) return json({ hosts: [{ machine_id: machine, trust: 'first-party', shared: true, online: true }] });
++ if (url.includes('/internal/egress/')) return options.method === 'PUT' ? json({ mode, device_name: 'My home PC' }) : new Response(null, { status: 204 });
++ return json({ ...body, shared: true });
++ } });
++ await client.hosts(owner);
++ await assert.rejects(client.lease(machine, owner, body), { code: 'run.egress.waiting' });
++ assert.equal(calls.filter(c => c.url.endsWith('/desk-leases')).length, 0, 'wait does not start a browser');
++ mode = 'link';
++ await client.lease(machine, owner, body);
++ assert.deepEqual(calls.find(c => c.url.endsWith('/desk-leases')).body, body, 'no proxy credential is sent through the desk command');
++ const at = Date.now();
++ await store.insert(dbtest.db, { id: session, owner: `user:${owner}`, kind: 'browser', state: 'placing', width: 1280, height: 800,
++ ttl_s: 3600, idle_s: 900, request_digest: 'test', reasons: [], created_at: at, updated_at: at });
++ await store.update(dbtest.db, session, { state: 'running', machine, desk: body.desk });
++ await client.get(machine, body.desk, owner);
++ assert.ok(calls.some(c => c.url.endsWith('/renew')));
++ await client.end(machine, body.desk, owner);
++ const row = await store.get(dbtest.db, session);
++ assert.equal(row.egress_label, 'Going out through: My home PC');
++ assert.equal(store.present(row).placement.reasons[0], row.egress_label, 'label survives the client cache');
++ const ends = calls.filter(c => c.method === 'DELETE');
++ assert.ok(ends[0].url.includes('/internal/egress/'), 'revoke browsing before ending the desk');
++ assert.ok(ends[1].url.includes('/desk-leases/'));
++ await store.update(dbtest.db, session, { state: 'ended', ended_at: at + 1000, end_reason: 'ended' });
++ const late = await store.update(dbtest.db, session, { state: 'running', machine, desk: body.desk });
++ assert.equal(late.state, 'ended', 'late desk responses cannot revive an ended session');
++ assert.equal((await store.get(dbtest.db, session)).state, 'ended');
++ console.log('Link desk integration: wait before browser, credential separation, renew, revoke, persisted label and terminal session state passed');
++ } finally { await dbtest.close(); }
++})().catch(e => { console.error(e); process.exitCode = 1; });
ddiff --git a/test/link-egress.test.js b/test/link-egress.test.js
nnew file mode 100644
iindex 0000000..a656f6f
---- /dev/null
++++ b/test/link-egress.test.js
@@@ -0,0 +1,31 @@
++'use strict';
++const assert = require('assert/strict');
++const { createLinkEgress, goingOutThrough } = require('../server/sessions/link-egress');
++(async () => {
++ const requests = [];
++ let mode = 'link', available = true;
++ const config = { network: { internalUrl: 'https://network.test' }, oauth: { clientId: 'run', clientSecret: 'test-only' } };
++ const client = createLinkEgress({ config, fetchImpl: async (url, options) => {
++ if (url.endsWith('/oauth/token')) return new Response(JSON.stringify({ access_token: 'test-service', token_type: 'Bearer', expires_in: 3600 }), { status: 200 });
++ requests.push({ url, options });
++ if (!available) return new Response('', { status: 503 });
++ return options.method === 'PUT' ? new Response(JSON.stringify({ mode, device_name: 'Home PC' }), { status: 200 }) : new Response(null, { status: 204 });
++ } });
++ const body = { session: 'ses_test', desk: 'desk-test' };
++ await client.prepare('dev_cloud', 'usr_owner', body);
++ assert.equal(goingOutThrough(body.session), 'Going out through: Home PC');
++ const sent = JSON.parse(requests[0].options.body);
++ assert.equal(sent.owner, 'usr_owner');
++ assert.equal(sent.machine, 'dev_cloud');
++ assert.equal(sent.credential.length, 43);
++ assert.ok(!goingOutThrough(body.session).includes(sent.credential));
++ mode = 'direct'; await client.prepare('dev_cloud', 'usr_owner', body);
++ assert.equal(goingOutThrough(body.session), "Going out through: OpenVibe's address");
++ mode = 'wait'; await assert.rejects(client.prepare('dev_cloud', 'usr_owner', body), { code: 'run.egress.waiting' });
++ available = false; await assert.rejects(client.prepare('dev_cloud', 'usr_owner', body), { code: 'run.egress.unavailable' });
++ available = true; await client.renew(body.session); await client.end(body.session);
++ assert.equal(goingOutThrough(body.session), null);
++ assert.ok(requests.some(r => r.url.endsWith('/ses_test/renew')));
++ assert.equal(requests.at(-1).options.method, 'DELETE');
++ console.log('Link egress: proxy decision, both offline choices, failed authorization, renewal and end passed');
++})().catch(e => { console.error(e); process.exitCode = 1; });