openvibers/openvibe.events Commits
Realtime tickets (roadmap WS-E task 3, WS-F task 1; ADR-005 amendment 2, Contracts 0.61.0 identity.realtime-ticket-claims@1): /realtime/stream?ticket= opens a stream as the signed-in person from any OpenVibe site without a third-party cookie. The ticket is Network's two-minute RS256 JWT (iss <issuer>/realtime, sub usr_, aud openvibe.events only, typ and purpose realtime, lifetime at most 300 s, jti rtk_); each is accepted once (ticket.used), never as Bearer or cookie, never logged; ticket.invalid / ticket.expired otherwise. A user JWT carrying typ or purpose is no session here. test/realtime-tickets.test.js: a ticket for Alice never receives Bob's network.notification.created on any pattern (live or replay), user:<id> is refused as a pattern, expired/wrong purpose/typ/audience/issuer/key/lifetime/reused/session-JWT tickets refused, resume from last_event_id with a fresh ticket, retention gap reported first
Changes
@@@ -165,12 +165,20 @@ app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.ra## Realtime (SSE)```js--const es = new EventSource('https://events.openvibe.network/realtime/stream?topics=live.stream.*,network.notification.*', { withCredentials: true });++const es = new EventSource('https://events.openvibe.network/realtime/stream?topics=live.stream.*', { withCredentials: true });++// A signed-in person's own events from any OpenVibe site: a fresh ticket from Network for every (re)connect.++const { ticket, stream_url, topics } = await (await fetch('https://openvibe.network/api/v1/realtime/ticket', { method: 'POST', headers: { Authorization: `Bearer ${networkJwt}` } })).json();++const mine = new EventSource(`${stream_url}?topics=${topics.join(',')}&ticket=${ticket}${lastSeq != null ? `&last_event_id=${lastSeq}` : ''}`);es.onmessage = (m) => { const { seq, event } = JSON.parse(m.data); };es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });```--- Auth: the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.event.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.++- Auth: a **realtime ticket** (`?ticket=`), the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.event.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.++- Realtime tickets (ADR-005 amendment 2): a page on any OpenVibe site cannot count on a cookie of events.openvibe.network (third-party there), and an EventSource cannot send a header. So it asks Network for a ticket (`POST https://openvibe.network/api/v1/realtime/ticket`, answering `network.realtime-ticket-result@1`) and opens `/realtime/stream?topics=network.notification.*&ticket=<ticket>` without credentials. The ticket is an RS256 JWT signed with Network's key (`identity.realtime-ticket-claims@1`): `iss <OV_NETWORK_ISSUER>/realtime`, `sub <usr_>`, `aud [openvibe.events]`, `typ` and `purpose` `realtime`, a lifetime of at most 300 s (Network mints 120 s) and `jti rtk_…`.++ - Events accepts each ticket once, never as Bearer or cookie, and never logs it. The refusals are 401: `ticket.invalid`, `ticket.expired` and `ticket.used`.++ - A reconnect asks for a new ticket and resumes with `last_event_id`.++ - Conversely, a user JWT that carries `typ` or `purpose` (a ticket, a FedCM assertion) is not a session here.++- A person's topic: `network.notification.created` (Network's outbox; visibility `subject`, subject the recipient). `topics=network.notification.*` streams a person's own notifications and nobody else's. There is no `user:<id>` topic: it is not a valid pattern (400 `realtime.bad_topic`), and subject visibility already does its job.- Visibility: `public` events go to anyone subscribed to the topic; `subject` events only to the user whose subject id (`usr_…`) is the event's `actor.id` or its user `subject.id`; `internal` events never reach a browser. A guessed topic yields nothing.- Resume: the SSE `id` is the seq, so the browser's automatic `Last-Event-ID` (or `?last_event_id=`) replays what was missed. A cursor older than retention first gets `event: gap` (`{ reason, from_seq, to_seq }`), as does a replay that hits `REALTIME_REPLAY_MAX`.- Public replay window: browsers (signed out or signed in) are replayed `public` events received in the last `REALTIME_PUBLIC_REPLAY_SECONDS` (300; 0 = none), enough to ride out a reconnect. Older public events are not replayed: the stream opens with `event: gap` (`reason: "public_window"`, `window_seconds`), and the client refetches state from the owning service. `subject` events addressed to the viewer, and service viewers, keep the whole retention. This keeps the stream from paging through a month of public history, chat lines included; nothing in the network needs more (no browser surface replays public events today).@@@ -4,7 +4,12 @@** - Service principals: RS256 client-credentials tokens from OpenVibe.Network (audience* openvibe.events), verified with openvibe-contracts serviceAuth.verifyServiceToken.-- * - Browsers: Network user JWTs (RS256, same key), cookie `ov_token` or Bearer.++ * - Browsers: Network user JWTs (RS256, same key), cookie `ov_token` or Bearer; or, on++ * /realtime/stream only, a realtime ticket (?ticket=): a two-minute, single-use RS256 JWT Network++ * signs for the signed-in person (identity.realtime-ticket-claims@1, ADR-005 amendment 2), so a++ * page on any OpenVibe site can open an EventSource without a third-party cookie. A ticket is++ * never a session (issuer <network>/realtime, typ, audience openvibe.events only) and a session++ * token is never a ticket.** - Developer apps (ADR-014): app tokens (sub app:app_<ULID>, project_id, env) holding* events.app.publish / events.app.read / events.app.subscribe, on the routes that accept them@@@ -127,12 +132,55 @@ function verifyUserJwt(token, { publicKey, issuer, audiences, now = Date.now() }const aud = Array.isArray(claims.aud) ? claims.aud : [claims.aud];if (!aud.some(a => audiences.includes(a))) return null;if (claims.actor_type || /^(svc|app|mod):/.test(String(claims.sub))) return null;++ // A typed token (a realtime ticket, a FedCM assertion) is never a session.++ if (claims.typ !== undefined || claims.purpose !== undefined) return null;return claims;} catch {return null;}}++const TICKET_LIFETIME_MAX_S = 300;++const TICKET_SKEW_S = 30;++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const TICKET_JTI_RE = /^rtk_[0-9a-f]{24}$/;++++/**++ * A realtime ticket (identity.realtime-ticket-claims@1). Returns { ok: true, claims } or++ * { ok: false, code, reason }. Checks the signature, iss (<issuer>/realtime), aud (openvibe.events),++ * typ and purpose (realtime), sub (a usr_ subject), jti, expiry and a lifetime of at most 300 s.++ * Single use is the caller's (createAuth keeps the jtis it accepted).++ */++function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {++ const bad = (reason) => ({ ok: false, code: 'ticket.invalid', reason });++ if (!publicKey) return { ok: false, code: 'token.unavailable', reason: 'signing key not loaded yet' };++ if (typeof token !== 'string' || token.length > 4096) return bad('not a ticket');++ const parts = token.split('.');++ if (parts.length !== 3) return bad('not a ticket');++ let header; let claims;++ try {++ header = b64json(parts[0]);++ claims = b64json(parts[1]);++ } catch {++ return bad('not a ticket');++ }++ if (!header || header.alg !== 'RS256' || !claims || typeof claims !== 'object') return bad('not a ticket');++ let signed = false;++ try { signed = crypto.verify('RSA-SHA256', Buffer.from(`${parts[0]}.${parts[1]}`), publicKey, Buffer.from(parts[2], 'base64url')); } catch { signed = false; }++ if (!signed) return bad('signature does not verify');++ if (claims.iss !== `${issuer}/realtime`) return bad('not a realtime ticket (issuer)');++ if (claims.typ !== 'realtime' || claims.purpose !== 'realtime') return bad('not a realtime ticket (purpose)');++ const aud = Array.isArray(claims.aud) ? claims.aud : [claims.aud];++ if (aud.length !== 1 || aud[0] !== 'openvibe.events') return bad('not for openvibe.events');++ if (!SUBJECT_RE.test(String(claims.sub)) || !TICKET_JTI_RE.test(String(claims.jti))) return bad('bad claims');++ const t = Math.floor(now / 1000);++ if (!Number.isInteger(claims.iat) || !Number.isInteger(claims.exp)) return bad('bad claims');++ if (claims.exp - claims.iat > TICKET_LIFETIME_MAX_S || claims.exp <= claims.iat) return bad('lifetime over 300 s');++ if (claims.iat > t + TICKET_SKEW_S) return bad('issued in the future');++ if (claims.exp <= t) return { ok: false, code: 'ticket.expired', reason: 'the ticket expired; ask Network for a new one' };++ return { ok: true, claims };++}++function bearer(req) {const h = String(req.headers.authorization || '');return h.startsWith('Bearer ') ? h.slice(7).trim() : null;@@@ -156,6 +204,16 @@ function serviceSlug(sub) {}function createAuth({ config, keys, store = null }) {++ // Realtime tickets are single use: the jtis accepted while they are still valid (in memory; a++ // restart forgets them, and a ticket lives two minutes at most anyway).++ const usedTickets = new Map(); // jti -> exp (ms)++ function spendTicket(jti, expSec, now = Date.now()) {++ if (usedTickets.size > 5000) for (const [k, exp] of usedTickets) if (exp <= now) usedTickets.delete(k);++ if (usedTickets.has(jti)) return false;++ usedTickets.set(jti, expSec * 1000);++ return true;++ }++// Token lifetimes are judged on wall-clock time, never on the worker's (injectable) clock.function verifyService(token, { acceptSandbox = false } = {}) {const publicKey = keys.get();@@@ -222,10 +280,18 @@ function createAuth({ config, keys, store = null }) {/*** Who is on the other end of a realtime connection:* { kind: 'service', sub } | { kind: 'user', subjectId, sub } | { kind: 'anonymous' }-- * or { error: { status, code, detail } }. A Bearer token must verify; an unverifiable cookie-- * (expired session) degrades to anonymous so the browser still gets public events.++ * or { error: { status, code, detail } }. A realtime ticket (?ticket=) must verify and be unused;++ * then nothing else is looked at. A Bearer token must verify; an unverifiable cookie (expired++ * session) degrades to anonymous so the browser still gets public events. The ticket is never logged.*/function realtimeViewer(req) {++ const q = req.query || {};++ if (q.ticket !== undefined) {++ const r = verifyRealtimeTicket(Array.isArray(q.ticket) ? null : q.ticket, { publicKey: keys.get(), issuer: config.issuer });++ if (!r.ok) return { error: { status: r.code === 'token.unavailable' ? 503 : 401, code: r.code, detail: r.reason } };++ if (!spendTicket(r.claims.jti, r.claims.exp)) return { error: { status: 401, code: 'ticket.used', detail: 'a ticket opens one stream; ask Network for a new one' } };++ return { kind: 'user', subjectId: r.claims.sub, sub: r.claims.sub, via: 'ticket' };++ }const token = bearer(req);if (token) {const svc = verifyService(token);@@@ -253,4 +319,4 @@ function createAuth({ config, keys, store = null }) {return { verifyService, verifyUser, requireCap, appOrService, realtimeViewer };}--module.exports = { CAPS, createKeyStore, createAuth, hasCap, allows, verifyUserJwt, serviceSlug, bearer, cookie };++module.exports = { CAPS, createKeyStore, createAuth, hasCap, allows, verifyUserJwt, verifyRealtimeTicket, serviceSlug, bearer, cookie };@@@ -4,11 +4,17 @@** GET /realtime/stream?topics=live.stream.*,network.notification.**++ * Who is asking (server/auth.js realtimeViewer): a realtime ticket (?ticket=, two minutes, single use,++ * minted by Network for the signed-in person: how pages on every OpenVibe site open a stream without a++ * third-party cookie), a Bearer user or service token, the ov_token cookie, or nobody.++ ** Every message is `id: <seq>` + `data: {"seq":n,"event":<envelope>}`. Visibility decides who sees* an event:* public anyone subscribed to a matching topic (signed-out visitors too, unless disabled)* subject only the user whose subject id is the event's actor.id or its subject.id (subject-- * type user); a guessed topic yields nothing for anyone else++ * type user); a guessed topic yields nothing for anyone else. A person's topic is an++ * event type with this visibility (network.notification.*, ADR-005 amendment 2), never++ * a `user:<id>` name (not a valid pattern)* internal service principals (token with events.event.read) only, never a browser* Developer-app events (app.<project_key>.*, events.app.publish) are never streamed here, to anyone.*@@@ -45,6 +45,18 @@ function userToken({ subjectId = ids.newId('user'), aud = ['openvibe.live', 'ope}, key);}++/**++ * A realtime ticket as Network mints it (identity.realtime-ticket-claims@1): two minutes, single use,++ * issuer <network>/realtime, typ and purpose realtime, audience openvibe.events only. Override anything.++ */++function realtimeTicket({ subjectId = ids.newId('user'), iat, exp, key = privateKey, ...over } = {}) {++ const now = Math.floor(Date.now() / 1000);++ return serviceAuth.signServiceToken({++ iss: `${ISSUER}/realtime`, sub: subjectId, aud: ['openvibe.events'], typ: 'realtime', purpose: 'realtime',++ iat: iat ?? now, exp: exp ?? (iat ?? now) + 120, jti: `rtk_${crypto.randomBytes(12).toString('hex')}`, ...over,++ }, key);++}++function manualClock(t = Date.now()) {return { t, now() { return this.t; }, advance(ms) { this.t += ms; return this.t; } };}@@@ -199,6 +211,6 @@ function suite(name) {}module.exports = {-- ISSUER, privateKey, publicKey, silent, serviceToken, appToken, userToken, manualClock, tmpDir, boot, request,++ ISSUER, privateKey, publicKey, silent, serviceToken, appToken, userToken, realtimeTicket, manualClock, tmpDir, boot, request,envelope, subscriber, sse, sleep, suite,};nnew file mode 100644@@@ -0,0 +1,175 @@++'use strict';++// Realtime tickets (ADR-005 amendment 2; roadmap WS-E task 3, WS-F task 1): a page on any OpenVibe site++// opens /realtime/stream?ticket=… with a two-minute, single-use ticket Network signed for the signed-in++// person, and sees what that person's session would: public events and subject events addressed to them.++// A ticket for Alice never receives Bob's network.notification.created, on any pattern; expired, foreign,++// reused and session tokens are refused as tickets, and a ticket is never a session. A stream resumes++// from last_event_id with a fresh ticket, and a cursor past retention is reported as a gap.++const assert = require('assert');++const crypto = require('crypto');++const { ids } = require('openvibe-contracts');++const { boot, request, serviceToken, userToken, realtimeTicket, envelope, sse, sleep, suite, ISSUER } = require('./helpers');++++const t = suite('realtime-tickets');++let h;++const network = serviceToken('network', ['events.event.publish']);++const live = serviceToken('live', ['events.event.publish']);++const alice = ids.newId('user');++const bob = ids.newId('user');++const open = [];++const TOPIC = 'network.notification.*';++++async function stream(q, headers) {++ const c = await sse(h.base, `/realtime/stream?${q}`, { headers });++ open.push(c);++ return c;++}++async function publish(token, env) {++ const r = await request(h.base, 'POST', '/api/v1/events', { token, body: env });++ assert.strictEqual(r.status, 201, r.text);++ return r.body.seq;++}++/** network.notification.created for `who`, as Network's outbox writes it. */++const notified = (who, unread = 1) => envelope('network', {++ event_type: 'network.notification.created', visibility: 'subject', actor: { type: 'system', id: 'network' },++ subject: { type: 'user', id: who }, priority: 'low',++ payload: { notification_id: crypto.randomUUID(), type: 'STREAM_LIVE', category: 'stream', priority: 'normal', service: 'live', created_at: new Date().toISOString(), unread_count: unread },++});++const refused = async (q, code, headers) => {++ const r = await request(h.base, 'GET', `/realtime/stream?${q}`, { headers });++ assert.strictEqual(r.status, 401, `${q.slice(0, 60)}: ${r.text}`);++ assert.strictEqual(r.body.code, code);++ assert.ok(!r.text.includes('ticket=') && !r.text.includes('eyJ'), 'the ticket is not echoed');++ return r;++};++const quiet = () => sleep(150);++++t('boot', async () => { h = await boot({ env: { REALTIME_HEARTBEAT_MS: '100' } }); });++++t('a ticket opens a stream as its person: their subject events and public ones, nobody else\'s', async () => {++ const a = await stream(`topics=${TOPIC},live.stream.*&ticket=${realtimeTicket({ subjectId: alice })}`);++ const b = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: bob })}`);++ assert.strictEqual(a.status, 200);++ await sleep(30);++ assert.ok(a.comments.includes('connected user'));++ const toAlice = await publish(network, notified(alice));++ const toBob = await publish(network, notified(bob));++ const pub = await publish(live, envelope('live', { visibility: 'public' }));++ await a.waitFor(c => c.events().length === 2);++ await b.waitFor(c => c.events().length === 1);++ await quiet();++ assert.deepStrictEqual(a.events().map(e => e.seq), [toAlice, pub]);++ assert.deepStrictEqual(b.events().map(e => e.seq), [toBob]);++ assert.strictEqual(a.events()[0].event.payload.unread_count, 1);++ a.close(); b.close();++});++++t('a guessed user:<other> topic yields nothing: refused as a pattern, and no pattern reaches Bob\'s events', async () => {++ let r = await request(h.base, 'GET', `/realtime/stream?topics=user:${bob}&ticket=${realtimeTicket({ subjectId: alice })}`);++ assert.strictEqual(r.status, 400);++ assert.strictEqual(r.body.code, 'realtime.bad_topic');++ const guess = await stream(`topics=${TOPIC},*,network.*,*.created&ticket=${realtimeTicket({ subjectId: alice })}`);++ const anon = await stream(`topics=${TOPIC},*`);++ await publish(network, notified(bob));++ await publish(network, envelope('network', { event_type: 'network.notification.created', visibility: 'subject', actor: { type: 'user', id: bob }, subject: { type: 'user', id: bob } }));++ await publish(network, envelope('network', { event_type: 'network.wallet.credited', visibility: 'internal', subject: { type: 'user', id: alice } }));++ await quiet();++ assert.deepStrictEqual(guess.events(), [], 'Alice\'s ticket sees none of Bob\'s events and no internal one');++ assert.deepStrictEqual(anon.events(), []);++ // Replay too: from before those events, Alice's ticket still gets none of them.++ const replay = await stream(`topics=*&ticket=${realtimeTicket({ subjectId: alice })}&last_event_id=0`);++ await quiet();++ assert.ok(replay.events().every(e => e.event.subject.id !== bob && e.event.visibility !== 'internal'), 'replay never includes Bob\'s or internal events');++ guess.close(); anon.close(); replay.close();++});++++t('refused as a ticket: expired, wrong purpose, typ or audience, a session\'s issuer, too long-lived, foreign key, reused, a session JWT', async () => {++ const now = Math.floor(Date.now() / 1000);++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, iat: now - 300, exp: now - 180 })}`, 'ticket.expired');++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, purpose: 'session' })}`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, typ: undefined })}`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, aud: ['openvibe.network'] })}`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, aud: ['openvibe.events', 'openvibe.live'] })}`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, iss: ISSUER })}`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, iat: now, exp: now + 3600 })}`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: '57' })}`, 'ticket.invalid');++ const other = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, privateKeyEncoding: { type: 'pkcs8', format: 'pem' }, publicKeyEncoding: { type: 'spki', format: 'pem' } });++ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, key: other.privateKey })}`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=not-a-ticket`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=`, 'ticket.invalid');++ await refused(`topics=${TOPIC}&ticket=${userToken({ subjectId: alice, aud: ['openvibe.events'] })}`, 'ticket.invalid');++++ // Single use: the second stream with the same ticket is refused, whoever asks.++ const once = realtimeTicket({ subjectId: alice });++ const first = await stream(`topics=${TOPIC}&ticket=${once}`);++ assert.strictEqual(first.status, 200);++ await refused(`topics=${TOPIC}&ticket=${once}`, 'ticket.used');++ first.close();++ await refused(`topics=${TOPIC}&ticket=${once}`, 'ticket.used');++});++++t('a ticket is never a session: refused as Bearer, anonymous as a cookie', async () => {++ const tk = realtimeTicket({ subjectId: alice });++ const r = await request(h.base, 'GET', `/realtime/stream?topics=${TOPIC}`, { token: tk });++ assert.strictEqual(r.status, 401);++ assert.strictEqual(r.body.code, 'token.invalid');++ const c = await stream(`topics=${TOPIC}`, { Cookie: `ov_token=${realtimeTicket({ subjectId: alice })}` });++ await publish(network, notified(alice));++ await quiet();++ assert.ok(c.comments.includes('connected anonymous'), 'a ticket in the cookie is no session');++ assert.deepStrictEqual(c.events(), []);++ c.close();++ // A session JWT still works as before (Bearer and cookie), and a ticket beats a Bearer when both come.++ const s = await stream(`topics=${TOPIC}`, { Authorization: `Bearer ${userToken({ subjectId: alice })}` });++ await sleep(30);++ assert.ok(s.comments.includes('connected user'));++ s.close();++ const both = await request(h.base, 'GET', `/realtime/stream?topics=${TOPIC}&ticket=expired.not.valid`, { token: userToken({ subjectId: alice }) });++ assert.strictEqual(both.status, 401, 'a bad ticket is refused even beside a good session');++});++++t('disconnect and resume from the cursor with a fresh ticket: the missed events, in order, then live', async () => {++ const a1 = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}`);++ const seen = await publish(network, notified(alice, 1));++ await a1.waitFor(c => c.events().length === 1);++ const cursor = a1.events()[0].seq;++ assert.strictEqual(cursor, seen);++ a1.close(); // the tab went to sleep, or Events restarted++ await sleep(30);++ const missed1 = await publish(network, notified(alice, 2));++ await publish(network, notified(bob, 9));++ const missed2 = await publish(network, notified(alice, 3));++ const a2 = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}&last_event_id=${cursor}`);++ await a2.waitFor(c => c.events().length === 2);++ assert.deepStrictEqual(a2.events().map(e => e.seq), [missed1, missed2]);++ assert.deepStrictEqual(a2.events().map(e => e.event.payload.unread_count), [2, 3]);++ assert.deepStrictEqual(a2.gaps(), []);++ const now = await publish(network, notified(alice, 4));++ await a2.waitFor(c => c.events().length === 3);++ assert.strictEqual(a2.events()[2].seq, now);++ a2.close();++});++++t('a cursor older than retention is reported as a gap before the events kept', async () => {++ const old = await publish(network, notified(alice, 5));++ h.store.prune({ retentionDays: 30, now: Date.now() + 31 * 86400000 });++ const kept = await publish(network, notified(alice, 6));++ const c = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}&last_event_id=${old - 1}`);++ await c.waitFor(x => x.events().length === 1);++ assert.strictEqual(c.gaps().length, 1);++ assert.strictEqual(c.gaps()[0].reason, 'retention');++ assert.ok(c.messages.findIndex(m => m.event === 'gap') < c.messages.findIndex(m => !m.event), 'the gap comes first');++ assert.strictEqual(c.events()[0].seq, kept);++ c.close();++});++++t('CORS: a ticket stream needs no credentials, from https://*.openvibe.* origins', async () => {++ const c = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}`, { Origin: 'https://openvibe.live' });++ assert.strictEqual(c.status, 200);++ assert.strictEqual(c.headers['access-control-allow-origin'], 'https://openvibe.live');++ c.close();++});++++t('stop', async () => { for (const c of open) c.close(); await h.stop(); });++++t.run();
@@ -165,12 +165,20 @@ app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.ra
@@ -165,12 +165,20 @@ app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.ra
## Realtime (SSE)
## Realtime (SSE)
```js
```js
-const es = new EventSource('https://events.openvibe.network/realtime/stream?topics=live.stream.*,network.notification.*', { withCredentials: true });
+const es = new EventSource('https://events.openvibe.network/realtime/stream?topics=live.stream.*', { withCredentials: true });
+// A signed-in person's own events from any OpenVibe site: a fresh ticket from Network for every (re)connect.
+const { ticket, stream_url, topics } = await (await fetch('https://openvibe.network/api/v1/realtime/ticket', { method: 'POST', headers: { Authorization: `Bearer ${networkJwt}` } })).json();
+const mine = new EventSource(`${stream_url}?topics=${topics.join(',')}&ticket=${ticket}${lastSeq != null ? `&last_event_id=${lastSeq}` : ''}`);
es.onmessage = (m) => { const { seq, event } = JSON.parse(m.data); };
es.onmessage = (m) => { const { seq, event } = JSON.parse(m.data); };
es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });
es.addEventListener('gap', (m) => { /* events were missed: refetch state */ });
```
```
-- Auth: the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.event.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.
+- Auth: a **realtime ticket** (`?ticket=`), the Network `ov_token` cookie or a Bearer user JWT; a service token with `events.event.read`; or nobody (public events only, `REALTIME_ALLOW_ANONYMOUS`). An expired cookie degrades to anonymous; a bad Bearer is a 401.
+- Realtime tickets (ADR-005 amendment 2): a page on any OpenVibe site cannot count on a cookie of events.openvibe.network (third-party there), and an EventSource cannot send a header. So it asks Network for a ticket (`POST https://openvibe.network/api/v1/realtime/ticket`, answering `network.realtime-ticket-result@1`) and opens `/realtime/stream?topics=network.notification.*&ticket=<ticket>` without credentials. The ticket is an RS256 JWT signed with Network's key (`identity.realtime-ticket-claims@1`): `iss <OV_NETWORK_ISSUER>/realtime`, `sub <usr_>`, `aud [openvibe.events]`, `typ` and `purpose` `realtime`, a lifetime of at most 300 s (Network mints 120 s) and `jti rtk_…`.
+ - Events accepts each ticket once, never as Bearer or cookie, and never logs it. The refusals are 401: `ticket.invalid`, `ticket.expired` and `ticket.used`.
+ - A reconnect asks for a new ticket and resumes with `last_event_id`.
+ - Conversely, a user JWT that carries `typ` or `purpose` (a ticket, a FedCM assertion) is not a session here.
+- A person's topic: `network.notification.created` (Network's outbox; visibility `subject`, subject the recipient). `topics=network.notification.*` streams a person's own notifications and nobody else's. There is no `user:<id>` topic: it is not a valid pattern (400 `realtime.bad_topic`), and subject visibility already does its job.
- Visibility: `public` events go to anyone subscribed to the topic; `subject` events only to the user whose subject id (`usr_…`) is the event's `actor.id` or its user `subject.id`; `internal` events never reach a browser. A guessed topic yields nothing.
- Visibility: `public` events go to anyone subscribed to the topic; `subject` events only to the user whose subject id (`usr_…`) is the event's `actor.id` or its user `subject.id`; `internal` events never reach a browser. A guessed topic yields nothing.
- Resume: the SSE `id` is the seq, so the browser's automatic `Last-Event-ID` (or `?last_event_id=`) replays what was missed. A cursor older than retention first gets `event: gap` (`{ reason, from_seq, to_seq }`), as does a replay that hits `REALTIME_REPLAY_MAX`.
- Resume: the SSE `id` is the seq, so the browser's automatic `Last-Event-ID` (or `?last_event_id=`) replays what was missed. A cursor older than retention first gets `event: gap` (`{ reason, from_seq, to_seq }`), as does a replay that hits `REALTIME_REPLAY_MAX`.
- Public replay window: browsers (signed out or signed in) are replayed `public` events received in the last `REALTIME_PUBLIC_REPLAY_SECONDS` (300; 0 = none), enough to ride out a reconnect. Older public events are not replayed: the stream opens with `event: gap` (`reason: "public_window"`, `window_seconds`), and the client refetches state from the owning service. `subject` events addressed to the viewer, and service viewers, keep the whole retention. This keeps the stream from paging through a month of public history, chat lines included; nothing in the network needs more (no browser surface replays public events today).
- Public replay window: browsers (signed out or signed in) are replayed `public` events received in the last `REALTIME_PUBLIC_REPLAY_SECONDS` (300; 0 = none), enough to ride out a reconnect. Older public events are not replayed: the stream opens with `event: gap` (`reason: "public_window"`, `window_seconds`), and the client refetches state from the owning service. `subject` events addressed to the viewer, and service viewers, keep the whole retention. This keeps the stream from paging through a month of public history, chat lines included; nothing in the network needs more (no browser surface replays public events today).
@@ -4,7 +4,12 @@
@@ -4,7 +4,12 @@
*
*
* - Service principals: RS256 client-credentials tokens from OpenVibe.Network (audience
* - Service principals: RS256 client-credentials tokens from OpenVibe.Network (audience
* openvibe.events), verified with openvibe-contracts serviceAuth.verifyServiceToken.
* openvibe.events), verified with openvibe-contracts serviceAuth.verifyServiceToken.
- * - Browsers: Network user JWTs (RS256, same key), cookie `ov_token` or Bearer.
+ * - Browsers: Network user JWTs (RS256, same key), cookie `ov_token` or Bearer; or, on
+ * /realtime/stream only, a realtime ticket (?ticket=): a two-minute, single-use RS256 JWT Network
+ * signs for the signed-in person (identity.realtime-ticket-claims@1, ADR-005 amendment 2), so a
+ * page on any OpenVibe site can open an EventSource without a third-party cookie. A ticket is
+ * never a session (issuer <network>/realtime, typ, audience openvibe.events only) and a session
+ * token is never a ticket.
*
*
* - Developer apps (ADR-014): app tokens (sub app:app_<ULID>, project_id, env) holding
* - Developer apps (ADR-014): app tokens (sub app:app_<ULID>, project_id, env) holding
* events.app.publish / events.app.read / events.app.subscribe, on the routes that accept them
* events.app.publish / events.app.read / events.app.subscribe, on the routes that accept them
@@ -127,12 +132,55 @@ function verifyUserJwt(token, { publicKey, issuer, audiences, now = Date.now() }
@@ -127,12 +132,55 @@ function verifyUserJwt(token, { publicKey, issuer, audiences, now = Date.now() }
const aud = Array.isArray(claims.aud) ? claims.aud : [claims.aud];
const aud = Array.isArray(claims.aud) ? claims.aud : [claims.aud];
if (!aud.some(a => audiences.includes(a))) return null;
if (!aud.some(a => audiences.includes(a))) return null;
if (claims.actor_type || /^(svc|app|mod):/.test(String(claims.sub))) return null;
if (claims.actor_type || /^(svc|app|mod):/.test(String(claims.sub))) return null;
+ // A typed token (a realtime ticket, a FedCM assertion) is never a session.
+ if (claims.typ !== undefined || claims.purpose !== undefined) return null;
return claims;
return claims;
} catch {
} catch {
return null;
return null;
}
}
}
}
+const TICKET_LIFETIME_MAX_S = 300;
+const TICKET_SKEW_S = 30;
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const TICKET_JTI_RE = /^rtk_[0-9a-f]{24}$/;
+
+/**
+ * A realtime ticket (identity.realtime-ticket-claims@1). Returns { ok: true, claims } or
+ * { ok: false, code, reason }. Checks the signature, iss (<issuer>/realtime), aud (openvibe.events),
+ * typ and purpose (realtime), sub (a usr_ subject), jti, expiry and a lifetime of at most 300 s.
+ * Single use is the caller's (createAuth keeps the jtis it accepted).
+ */
+function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {
+ const bad = (reason) => ({ ok: false, code: 'ticket.invalid', reason });
+ if (!publicKey) return { ok: false, code: 'token.unavailable', reason: 'signing key not loaded yet' };
+ if (typeof token !== 'string' || token.length > 4096) return bad('not a ticket');
+ const parts = token.split('.');
+ if (parts.length !== 3) return bad('not a ticket');
+ let header; let claims;
+ try {
+ header = b64json(parts[0]);
+ claims = b64json(parts[1]);
+ } catch {
+ return bad('not a ticket');
+ }
+ if (!header || header.alg !== 'RS256' || !claims || typeof claims !== 'object') return bad('not a ticket');
+ let signed = false;
+ try { signed = crypto.verify('RSA-SHA256', Buffer.from(`${parts[0]}.${parts[1]}`), publicKey, Buffer.from(parts[2], 'base64url')); } catch { signed = false; }
+ if (!signed) return bad('signature does not verify');
+ if (claims.iss !== `${issuer}/realtime`) return bad('not a realtime ticket (issuer)');
+ if (claims.typ !== 'realtime' || claims.purpose !== 'realtime') return bad('not a realtime ticket (purpose)');
+ const aud = Array.isArray(claims.aud) ? claims.aud : [claims.aud];
+ if (aud.length !== 1 || aud[0] !== 'openvibe.events') return bad('not for openvibe.events');
+ if (!SUBJECT_RE.test(String(claims.sub)) || !TICKET_JTI_RE.test(String(claims.jti))) return bad('bad claims');
+ const t = Math.floor(now / 1000);
+ if (!Number.isInteger(claims.iat) || !Number.isInteger(claims.exp)) return bad('bad claims');
+ if (claims.exp - claims.iat > TICKET_LIFETIME_MAX_S || claims.exp <= claims.iat) return bad('lifetime over 300 s');
+ if (claims.iat > t + TICKET_SKEW_S) return bad('issued in the future');
+ if (claims.exp <= t) return { ok: false, code: 'ticket.expired', reason: 'the ticket expired; ask Network for a new one' };
+ return { ok: true, claims };
+}
+
function bearer(req) {
function bearer(req) {
const h = String(req.headers.authorization || '');
const h = String(req.headers.authorization || '');
return h.startsWith('Bearer ') ? h.slice(7).trim() : null;
return h.startsWith('Bearer ') ? h.slice(7).trim() : null;
@@ -156,6 +204,16 @@ function serviceSlug(sub) {
@@ -156,6 +204,16 @@ function serviceSlug(sub) {
}
}
function createAuth({ config, keys, store = null }) {
function createAuth({ config, keys, store = null }) {
+ // Realtime tickets are single use: the jtis accepted while they are still valid (in memory; a
+ // restart forgets them, and a ticket lives two minutes at most anyway).
+ const usedTickets = new Map(); // jti -> exp (ms)
+ function spendTicket(jti, expSec, now = Date.now()) {
+ if (usedTickets.size > 5000) for (const [k, exp] of usedTickets) if (exp <= now) usedTickets.delete(k);
+ if (usedTickets.has(jti)) return false;
+ usedTickets.set(jti, expSec * 1000);
+ return true;
+ }
+
// Token lifetimes are judged on wall-clock time, never on the worker's (injectable) clock.
// Token lifetimes are judged on wall-clock time, never on the worker's (injectable) clock.
function verifyService(token, { acceptSandbox = false } = {}) {
function verifyService(token, { acceptSandbox = false } = {}) {
const publicKey = keys.get();
const publicKey = keys.get();
@@ -222,10 +280,18 @@ function createAuth({ config, keys, store = null }) {
@@ -222,10 +280,18 @@ function createAuth({ config, keys, store = null }) {
/**
/**
* Who is on the other end of a realtime connection:
* Who is on the other end of a realtime connection:
* { kind: 'service', sub } | { kind: 'user', subjectId, sub } | { kind: 'anonymous' }
* { kind: 'service', sub } | { kind: 'user', subjectId, sub } | { kind: 'anonymous' }
- * or { error: { status, code, detail } }. A Bearer token must verify; an unverifiable cookie
- * (expired session) degrades to anonymous so the browser still gets public events.
+ * or { error: { status, code, detail } }. A realtime ticket (?ticket=) must verify and be unused;
+ * then nothing else is looked at. A Bearer token must verify; an unverifiable cookie (expired
+ * session) degrades to anonymous so the browser still gets public events. The ticket is never logged.
*/
*/
function realtimeViewer(req) {
function realtimeViewer(req) {
+ const q = req.query || {};
+ if (q.ticket !== undefined) {
+ const r = verifyRealtimeTicket(Array.isArray(q.ticket) ? null : q.ticket, { publicKey: keys.get(), issuer: config.issuer });
+ if (!r.ok) return { error: { status: r.code === 'token.unavailable' ? 503 : 401, code: r.code, detail: r.reason } };
+ if (!spendTicket(r.claims.jti, r.claims.exp)) return { error: { status: 401, code: 'ticket.used', detail: 'a ticket opens one stream; ask Network for a new one' } };
+ return { kind: 'user', subjectId: r.claims.sub, sub: r.claims.sub, via: 'ticket' };
+ }
const token = bearer(req);
const token = bearer(req);
if (token) {
if (token) {
const svc = verifyService(token);
const svc = verifyService(token);
@@ -253,4 +319,4 @@ function createAuth({ config, keys, store = null }) {
@@ -253,4 +319,4 @@ function createAuth({ config, keys, store = null }) {
return { verifyService, verifyUser, requireCap, appOrService, realtimeViewer };
return { verifyService, verifyUser, requireCap, appOrService, realtimeViewer };
}
}
-module.exports = { CAPS, createKeyStore, createAuth, hasCap, allows, verifyUserJwt, serviceSlug, bearer, cookie };
+module.exports = { CAPS, createKeyStore, createAuth, hasCap, allows, verifyUserJwt, verifyRealtimeTicket, serviceSlug, bearer, cookie };
@@ -4,11 +4,17 @@
@@ -4,11 +4,17 @@
*
*
* GET /realtime/stream?topics=live.stream.*,network.notification.*
* GET /realtime/stream?topics=live.stream.*,network.notification.*
*
*
+ * Who is asking (server/auth.js realtimeViewer): a realtime ticket (?ticket=, two minutes, single use,
+ * minted by Network for the signed-in person: how pages on every OpenVibe site open a stream without a
+ * third-party cookie), a Bearer user or service token, the ov_token cookie, or nobody.
+ *
* Every message is `id: <seq>` + `data: {"seq":n,"event":<envelope>}`. Visibility decides who sees
* Every message is `id: <seq>` + `data: {"seq":n,"event":<envelope>}`. Visibility decides who sees
* an event:
* an event:
* public anyone subscribed to a matching topic (signed-out visitors too, unless disabled)
* public anyone subscribed to a matching topic (signed-out visitors too, unless disabled)
* subject only the user whose subject id is the event's actor.id or its subject.id (subject
* subject only the user whose subject id is the event's actor.id or its subject.id (subject
- * type user); a guessed topic yields nothing for anyone else
+ * type user); a guessed topic yields nothing for anyone else. A person's topic is an
+ * event type with this visibility (network.notification.*, ADR-005 amendment 2), never
+ * a `user:<id>` name (not a valid pattern)
* internal service principals (token with events.event.read) only, never a browser
* internal service principals (token with events.event.read) only, never a browser
* Developer-app events (app.<project_key>.*, events.app.publish) are never streamed here, to anyone.
* Developer-app events (app.<project_key>.*, events.app.publish) are never streamed here, to anyone.
*
*
@@ -45,6 +45,18 @@ function userToken({ subjectId = ids.newId('user'), aud = ['openvibe.live', 'ope
@@ -45,6 +45,18 @@ function userToken({ subjectId = ids.newId('user'), aud = ['openvibe.live', 'ope
}, key);
}, key);
}
}
+/**
+ * A realtime ticket as Network mints it (identity.realtime-ticket-claims@1): two minutes, single use,
+ * issuer <network>/realtime, typ and purpose realtime, audience openvibe.events only. Override anything.
+ */
+function realtimeTicket({ subjectId = ids.newId('user'), iat, exp, key = privateKey, ...over } = {}) {
+ const now = Math.floor(Date.now() / 1000);
+ return serviceAuth.signServiceToken({
+ iss: `${ISSUER}/realtime`, sub: subjectId, aud: ['openvibe.events'], typ: 'realtime', purpose: 'realtime',
+ iat: iat ?? now, exp: exp ?? (iat ?? now) + 120, jti: `rtk_${crypto.randomBytes(12).toString('hex')}`, ...over,
+ }, key);
+}
+
function manualClock(t = Date.now()) {
function manualClock(t = Date.now()) {
return { t, now() { return this.t; }, advance(ms) { this.t += ms; return this.t; } };
return { t, now() { return this.t; }, advance(ms) { this.t += ms; return this.t; } };
}
}
@@ -199,6 +211,6 @@ function suite(name) {
@@ -199,6 +211,6 @@ function suite(name) {
}
}
module.exports = {
module.exports = {
- ISSUER, privateKey, publicKey, silent, serviceToken, appToken, userToken, manualClock, tmpDir, boot, request,
+ ISSUER, privateKey, publicKey, silent, serviceToken, appToken, userToken, realtimeTicket, manualClock, tmpDir, boot, request,
envelope, subscriber, sse, sleep, suite,
envelope, subscriber, sse, sleep, suite,
};
};
new file mode 100644
new file mode 100644
@@ -0,0 +1,175 @@
@@ -0,0 +1,175 @@
+'use strict';
+// Realtime tickets (ADR-005 amendment 2; roadmap WS-E task 3, WS-F task 1): a page on any OpenVibe site
+// opens /realtime/stream?ticket=… with a two-minute, single-use ticket Network signed for the signed-in
+// person, and sees what that person's session would: public events and subject events addressed to them.
+// A ticket for Alice never receives Bob's network.notification.created, on any pattern; expired, foreign,
+// reused and session tokens are refused as tickets, and a ticket is never a session. A stream resumes
+// from last_event_id with a fresh ticket, and a cursor past retention is reported as a gap.
+const assert = require('assert');
+const crypto = require('crypto');
+const { ids } = require('openvibe-contracts');
+const { boot, request, serviceToken, userToken, realtimeTicket, envelope, sse, sleep, suite, ISSUER } = require('./helpers');
+
+const t = suite('realtime-tickets');
+let h;
+const network = serviceToken('network', ['events.event.publish']);
+const live = serviceToken('live', ['events.event.publish']);
+const alice = ids.newId('user');
+const bob = ids.newId('user');
+const open = [];
+const TOPIC = 'network.notification.*';
+
+async function stream(q, headers) {
+ const c = await sse(h.base, `/realtime/stream?${q}`, { headers });
+ open.push(c);
+ return c;
+}
+async function publish(token, env) {
+ const r = await request(h.base, 'POST', '/api/v1/events', { token, body: env });
+ assert.strictEqual(r.status, 201, r.text);
+ return r.body.seq;
+}
+/** network.notification.created for `who`, as Network's outbox writes it. */
+const notified = (who, unread = 1) => envelope('network', {
+ event_type: 'network.notification.created', visibility: 'subject', actor: { type: 'system', id: 'network' },
+ subject: { type: 'user', id: who }, priority: 'low',
+ payload: { notification_id: crypto.randomUUID(), type: 'STREAM_LIVE', category: 'stream', priority: 'normal', service: 'live', created_at: new Date().toISOString(), unread_count: unread },
+});
+const refused = async (q, code, headers) => {
+ const r = await request(h.base, 'GET', `/realtime/stream?${q}`, { headers });
+ assert.strictEqual(r.status, 401, `${q.slice(0, 60)}: ${r.text}`);
+ assert.strictEqual(r.body.code, code);
+ assert.ok(!r.text.includes('ticket=') && !r.text.includes('eyJ'), 'the ticket is not echoed');
+ return r;
+};
+const quiet = () => sleep(150);
+
+t('boot', async () => { h = await boot({ env: { REALTIME_HEARTBEAT_MS: '100' } }); });
+
+t('a ticket opens a stream as its person: their subject events and public ones, nobody else\'s', async () => {
+ const a = await stream(`topics=${TOPIC},live.stream.*&ticket=${realtimeTicket({ subjectId: alice })}`);
+ const b = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: bob })}`);
+ assert.strictEqual(a.status, 200);
+ await sleep(30);
+ assert.ok(a.comments.includes('connected user'));
+ const toAlice = await publish(network, notified(alice));
+ const toBob = await publish(network, notified(bob));
+ const pub = await publish(live, envelope('live', { visibility: 'public' }));
+ await a.waitFor(c => c.events().length === 2);
+ await b.waitFor(c => c.events().length === 1);
+ await quiet();
+ assert.deepStrictEqual(a.events().map(e => e.seq), [toAlice, pub]);
+ assert.deepStrictEqual(b.events().map(e => e.seq), [toBob]);
+ assert.strictEqual(a.events()[0].event.payload.unread_count, 1);
+ a.close(); b.close();
+});
+
+t('a guessed user:<other> topic yields nothing: refused as a pattern, and no pattern reaches Bob\'s events', async () => {
+ let r = await request(h.base, 'GET', `/realtime/stream?topics=user:${bob}&ticket=${realtimeTicket({ subjectId: alice })}`);
+ assert.strictEqual(r.status, 400);
+ assert.strictEqual(r.body.code, 'realtime.bad_topic');
+ const guess = await stream(`topics=${TOPIC},*,network.*,*.created&ticket=${realtimeTicket({ subjectId: alice })}`);
+ const anon = await stream(`topics=${TOPIC},*`);
+ await publish(network, notified(bob));
+ await publish(network, envelope('network', { event_type: 'network.notification.created', visibility: 'subject', actor: { type: 'user', id: bob }, subject: { type: 'user', id: bob } }));
+ await publish(network, envelope('network', { event_type: 'network.wallet.credited', visibility: 'internal', subject: { type: 'user', id: alice } }));
+ await quiet();
+ assert.deepStrictEqual(guess.events(), [], 'Alice\'s ticket sees none of Bob\'s events and no internal one');
+ assert.deepStrictEqual(anon.events(), []);
+ // Replay too: from before those events, Alice's ticket still gets none of them.
+ const replay = await stream(`topics=*&ticket=${realtimeTicket({ subjectId: alice })}&last_event_id=0`);
+ await quiet();
+ assert.ok(replay.events().every(e => e.event.subject.id !== bob && e.event.visibility !== 'internal'), 'replay never includes Bob\'s or internal events');
+ guess.close(); anon.close(); replay.close();
+});
+
+t('refused as a ticket: expired, wrong purpose, typ or audience, a session\'s issuer, too long-lived, foreign key, reused, a session JWT', async () => {
+ const now = Math.floor(Date.now() / 1000);
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, iat: now - 300, exp: now - 180 })}`, 'ticket.expired');
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, purpose: 'session' })}`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, typ: undefined })}`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, aud: ['openvibe.network'] })}`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, aud: ['openvibe.events', 'openvibe.live'] })}`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, iss: ISSUER })}`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, iat: now, exp: now + 3600 })}`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: '57' })}`, 'ticket.invalid');
+ const other = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, privateKeyEncoding: { type: 'pkcs8', format: 'pem' }, publicKeyEncoding: { type: 'spki', format: 'pem' } });
+ await refused(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice, key: other.privateKey })}`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=not-a-ticket`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=`, 'ticket.invalid');
+ await refused(`topics=${TOPIC}&ticket=${userToken({ subjectId: alice, aud: ['openvibe.events'] })}`, 'ticket.invalid');
+
+ // Single use: the second stream with the same ticket is refused, whoever asks.
+ const once = realtimeTicket({ subjectId: alice });
+ const first = await stream(`topics=${TOPIC}&ticket=${once}`);
+ assert.strictEqual(first.status, 200);
+ await refused(`topics=${TOPIC}&ticket=${once}`, 'ticket.used');
+ first.close();
+ await refused(`topics=${TOPIC}&ticket=${once}`, 'ticket.used');
+});
+
+t('a ticket is never a session: refused as Bearer, anonymous as a cookie', async () => {
+ const tk = realtimeTicket({ subjectId: alice });
+ const r = await request(h.base, 'GET', `/realtime/stream?topics=${TOPIC}`, { token: tk });
+ assert.strictEqual(r.status, 401);
+ assert.strictEqual(r.body.code, 'token.invalid');
+ const c = await stream(`topics=${TOPIC}`, { Cookie: `ov_token=${realtimeTicket({ subjectId: alice })}` });
+ await publish(network, notified(alice));
+ await quiet();
+ assert.ok(c.comments.includes('connected anonymous'), 'a ticket in the cookie is no session');
+ assert.deepStrictEqual(c.events(), []);
+ c.close();
+ // A session JWT still works as before (Bearer and cookie), and a ticket beats a Bearer when both come.
+ const s = await stream(`topics=${TOPIC}`, { Authorization: `Bearer ${userToken({ subjectId: alice })}` });
+ await sleep(30);
+ assert.ok(s.comments.includes('connected user'));
+ s.close();
+ const both = await request(h.base, 'GET', `/realtime/stream?topics=${TOPIC}&ticket=expired.not.valid`, { token: userToken({ subjectId: alice }) });
+ assert.strictEqual(both.status, 401, 'a bad ticket is refused even beside a good session');
+});
+
+t('disconnect and resume from the cursor with a fresh ticket: the missed events, in order, then live', async () => {
+ const a1 = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}`);
+ const seen = await publish(network, notified(alice, 1));
+ await a1.waitFor(c => c.events().length === 1);
+ const cursor = a1.events()[0].seq;
+ assert.strictEqual(cursor, seen);
+ a1.close(); // the tab went to sleep, or Events restarted
+ await sleep(30);
+ const missed1 = await publish(network, notified(alice, 2));
+ await publish(network, notified(bob, 9));
+ const missed2 = await publish(network, notified(alice, 3));
+ const a2 = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}&last_event_id=${cursor}`);
+ await a2.waitFor(c => c.events().length === 2);
+ assert.deepStrictEqual(a2.events().map(e => e.seq), [missed1, missed2]);
+ assert.deepStrictEqual(a2.events().map(e => e.event.payload.unread_count), [2, 3]);
+ assert.deepStrictEqual(a2.gaps(), []);
+ const now = await publish(network, notified(alice, 4));
+ await a2.waitFor(c => c.events().length === 3);
+ assert.strictEqual(a2.events()[2].seq, now);
+ a2.close();
+});
+
+t('a cursor older than retention is reported as a gap before the events kept', async () => {
+ const old = await publish(network, notified(alice, 5));
+ h.store.prune({ retentionDays: 30, now: Date.now() + 31 * 86400000 });
+ const kept = await publish(network, notified(alice, 6));
+ const c = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}&last_event_id=${old - 1}`);
+ await c.waitFor(x => x.events().length === 1);
+ assert.strictEqual(c.gaps().length, 1);
+ assert.strictEqual(c.gaps()[0].reason, 'retention');
+ assert.ok(c.messages.findIndex(m => m.event === 'gap') < c.messages.findIndex(m => !m.event), 'the gap comes first');
+ assert.strictEqual(c.events()[0].seq, kept);
+ c.close();
+});
+
+t('CORS: a ticket stream needs no credentials, from https://*.openvibe.* origins', async () => {
+ const c = await stream(`topics=${TOPIC}&ticket=${realtimeTicket({ subjectId: alice })}`, { Origin: 'https://openvibe.live' });
+ assert.strictEqual(c.status, 200);
+ assert.strictEqual(c.headers['access-control-allow-origin'], 'https://openvibe.live');
+ c.close();
+});
+
+t('stop', async () => { for (const c of open) c.close(); await h.stop(); });
+
+t.run();