openvibers/openvibe.chat Commits
Revocation propagation (WS-B task 4, Contracts 0.39.0): Chat subscribes to network.user.token_valid_after, keeps the latest cutoff per subject, refuses Network tokens issued before it (locally and on the Live fallback) and closes the person's sockets opened with older tokens (4001, auth_revoked) within the event's delivery; bots' API-token sockets stay. test/revocation.test.js
Changes
@@@ -16,7 +16,7 @@"express": "^4.21.2","express-rate-limit": "^7.5.0","multer": "^2.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.35.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.39.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0","ws": "^8.18.0"@@@ -1062,9 +1062,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.35.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.35.0",-- "integrity": "sha512-5jkH9Gt9MVbYaxo/Fi1VSH0RwdPkbary7SOm2m9e5EvXYu8L/cJHMm6KrMKxa7MV6TJCATDeRpGLDOCaUeZmSw==",++ "version": "0.39.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.39.0",++ "integrity": "sha512-ek3tryU3pPwmFSg621gzr4ss1OE/gpufBnxk9XFDLSfUe2RuZpYVZCJNdQ0F/9GfA83Xop+xJW8+IjRyjFg6Yg==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -27,7 +27,7 @@"express": "^4.21.2","express-rate-limit": "^7.5.0","multer": "^2.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.35.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.39.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0","ws": "^8.18.0"@@@ -23,6 +23,7 @@ const config = require('../config');const db = require('../db/database');const serviceAuth = require('../net/service-auth');const ctx = require('../live-context');++const revocations = require('./revocations');const ROLE_RANK = { user: 0, streamer: 1, global_mod: 2, admin: 3 };const stats = { local: 0, live: 0, rejected: 0 };@@@ -76,22 +77,38 @@ function refuse(token) {/** Resolve a browser/bot token to a chat user (a fresh object) or null. */async function authenticate(token) {if (!token) return null;-- const viaLive = async () => { stats.live++; return ctx.authenticate(token); };++ const viaLive = async () => {++ stats.live++;++ const u = await ctx.authenticate(token);++ // Live resolved a Network token (first visit, another key): the same cutoff applies.++ const iat = u && u.auth_source !== 'api_token' ? tokenIat(token) : null;++ if (iat != null && iat * 1000 < revocations.cutoffFor(u.subject_id)) return refuse(token);++ return u;++ };if (String(token).startsWith('hbt_')) return viaLive();const key = await serviceAuth.ensureKey();if (!key) return viaLive();const r = verify(token, key);// Only a token signed by the key we hold is decided here; anything else is Live's to judge.if (!r.ok) return r.reason === 'signature' || r.reason === 'malformed' ? viaLive() : refuse(token);++ // Signed out everywhere, password changed, banned…: Network's cutoff for this person (WS-B task 4).++ if (revocations.isRevoked(r.claims)) return refuse(token);const user = userFor(r.claims);if (!user) return viaLive();stats.local++;return user;}++/** A JWT's iat (seconds), read without verifying (the token was verified when it was accepted); null otherwise. */++function tokenIat(token) {++ const parts = String(token || '').split('.');++ if (parts.length !== 3) return null;++ try { const c = b64json(parts[1]); return typeof c.iat === 'number' ? c.iat : null; } catch { return null; }++}++/** Why the last resolution of this token failed: 'invalid' (bad/expired) or 'unresolved' (no account). */function failureReason(token) {return (token && _reasons.get(tokenKey(token))) || ctx.authFailureReason(token);}--module.exports = { authenticate, failureReason, verify, userFor, stats: () => ({ ...stats }) };++module.exports = { authenticate, failureReason, verify, userFor, tokenIat, stats: () => ({ ...stats }) };nnew file mode 100644@@@ -0,0 +1,61 @@++/**++ * OpenVibe.Chat — per-person token cutoffs (roadmap WS-B task 4; Contracts 0.39.0++ * network.user.token_valid_after).++ *++ * When a person signs out everywhere, changes or resets their password, is banned or has their++ * sessions ended by staff, Network moves their cutoff and says so. Chat keeps the latest cutoff per++ * subject here, refuses a Network token issued before it (auth/network-session.js, Network's rule:++ * iat * 1000 < valid_after) and closes the sockets such tokens opened (chat-server revokeSubject).++ */++'use strict';++++const db = require('../db/database');++++let ready = false;++const cache = new Map(); // subject → valid_after ms (0 = none known)++++function ensureSchema() {++ if (ready) return;++ db.getDb().exec(`CREATE TABLE IF NOT EXISTS token_revocations (++ subject_id TEXT PRIMARY KEY,++ valid_after_ms INTEGER NOT NULL,++ reason TEXT,++ updated_at INTEGER NOT NULL++ )`);++ ready = true;++}++++/** The cutoff for a subject in ms, 0 when none. */++function cutoffFor(subject) {++ if (!subject) return 0;++ if (cache.has(subject)) return cache.get(subject);++ ensureSchema();++ const row = db.get('SELECT valid_after_ms FROM token_revocations WHERE subject_id = ?', [subject]);++ const ms = row ? Number(row.valid_after_ms) || 0 : 0;++ if (cache.size > 50000) cache.clear();++ cache.set(subject, ms);++ return ms;++}++++/** Keep the later cutoff (events can arrive out of order). Returns true when it moved forward. */++function record(subject, validAfterMs, reason = null, now = Date.now()) {++ ensureSchema();++ const current = cutoffFor(subject);++ if (!(validAfterMs > current)) return false;++ db.run(`INSERT INTO token_revocations (subject_id, valid_after_ms, reason, updated_at) VALUES (?, ?, ?, ?)++ ON CONFLICT(subject_id) DO UPDATE SET valid_after_ms = excluded.valid_after_ms, reason = excluded.reason, updated_at = excluded.updated_at++ WHERE excluded.valid_after_ms > token_revocations.valid_after_ms`, [subject, validAfterMs, reason, now]);++ cache.set(subject, validAfterMs);++ return true;++}++++/** Was a token with these claims issued before its subject's cutoff? */++function isRevoked(claims) {++ if (!claims || typeof claims.iat !== 'number') return false;++ return claims.iat * 1000 < cutoffFor(claims.subject_id);++}++++/** For tests: forget the in-memory cache (the table stays). */++function _reset() { cache.clear(); ready = false; }++++module.exports = { ensureSchema, cutoffFor, record, isRevoked, _reset };@@@ -23,6 +23,7 @@ const crypto = require('crypto');const db = require('../db/database');const ctx = require('../live-context');const { extractWsToken, authenticateWs } = require('../auth/auth');++const session = require('../auth/network-session');const { clientIpOf } = require('../net/client-ip');const permissions = require('../auth/permissions');const wordFilter = require('./word-filter');@@@ -283,7 +284,7 @@ class ChatServer {ws.off('close', onEarlyClose);ws.off('error', onEarlyClose);if (closedEarly || ws.readyState !== WebSocket.OPEN) return;-- this._registerConnection(ws, req, { ip, streamId, user, early });++ this._registerConnection(ws, req, { ip, streamId, user, early, tokenIat: user ? session.tokenIat(token) : null });})().catch((err) => {console.warn('[Chat] connection setup failed:', err.message);try { ws.close(1011, 'setup failed'); } catch {}@@@ -291,7 +292,7 @@ class ChatServer {}/** The rest of Live's handleConnection, once the socket's identity is known. */-- _registerConnection(ws, req, { ip, streamId, user, early }) {++ _registerConnection(ws, req, { ip, streamId, user, early, tokenIat = null }) {const perIp = this._ipSockets.get(ip) || 0;if (ip && ip !== 'unknown' && perIp >= MAX_CHAT_SOCKETS_PER_IP && !permissions.can(user, 'staff.limits.exempt')) {ws.close(4029, 'Too many connections');@@@ -317,6 +318,8 @@ class ChatServer {joinedAt: Date.now(),// Handle Live can address replies to (arena commands) through the bridge.connId: crypto.randomUUID(),++ // When the socket's token was issued (seconds), so a sign-out everywhere can close it.++ tokenIat,};this.clients.set(ws, clientInfo);@@@ -404,6 +407,7 @@ class ChatServer {if (user) {if (!client.user || client.user.id === user.id) {client.user = user;++ client.tokenIat = session.tokenIat(msg.token);client.anonId = null; // no longer anonymous} else {console.warn(`[Chat] Ignoring token identity mismatch for ${client.user.username} -> ${user.username}`);@@@ -2493,6 +2497,25 @@ class ChatServer {}}++ /**++ * Network moved this person's token cutoff (network.user.token_valid_after: signed out everywhere,++ * password changed, banned…). Close every socket they opened with an older Network token; the++ * browser reconnects and, holding no valid token, carries on as a guest. Bot sockets (hbt_ API++ * tokens) are not Network sessions and stay. Returns how many closed.++ */++ revokeSubject(subjectId, validAfterMs) {++ let closed = 0;++ for (const [ws, client] of this.clients) {++ const u = client.user;++ if (!u || !subjectId || u.subject_id !== subjectId || u.auth_source === 'api_token' || u._authSource === 'api_token') continue;++ if (client.tokenIat != null && client.tokenIat * 1000 >= validAfterMs) continue;++ try { if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'auth_revoked', reason: 'signed_out' })); } catch { /* closing anyway */ }++ try { ws.close(4001, 'Signed out'); } catch { /* already gone */ }++ closed++;++ }++ return closed;++ }++/*** Push a profile/identity update to all chat connections belonging to a user.* Refreshes cached client.user so subsequent messages use the new info.@@@ -37,10 +37,13 @@ const { viaProxy } = require('../net/service-auth');const deployNotice = require('../chat/deploy-notice');const prefs = require('../prefs/chat-preferences');const vipBadges = require('../vip/badges');++const revocations = require('../auth/revocations');++const ctx = require('../live-context');const CONSUMER = 'chat';const INBOX_TABLE = 'chat_event_inbox';--const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'vip.membership.changed']);++const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'network.user.token_valid_after', 'vip.membership.changed']);++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered@@@ -71,6 +74,27 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()if (p.namespace !== prefs.NAMESPACE) return 'ignored:namespace';return () => (prefs.handleEvent(event) ? 'invalidated' : 'unchanged');}++ if (event.event_type === 'network.user.token_valid_after') {++ // Signed out everywhere, password changed, banned…: refuse this person's older tokens and++ // close the sockets they opened (WS-B task 4).++ if (event.source !== 'network') return 'ignored:source';++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};++ const subject = p.subject && p.subject.id;++ const ms = Date.parse(p.valid_after);++ if (!SUBJECT_RE.test(String(subject || '')) || !Number.isFinite(ms)) return 'ignored:payload';++ return () => {++ const moved = revocations.record(subject, ms, typeof p.reason === 'string' ? p.reason.slice(0, 40) : null, now());++ if (!moved) return 'unchanged';++ return {++ outcome: 'revoked',++ after: () => {++ for (const r of db.all('SELECT id FROM ctx_users WHERE subject_id = ?', [subject])) ctx.invalidateUser(r.id);++ const closed = chatServer ? chatServer.revokeSubject(subject, ms) : 0;++ if (closed) log.log && log.log(`[Events consumer] ${p.reason || 'revoked'}: closed ${closed} socket(s)`);++ },++ };++ };++ }if (event.event_type === 'vip.membership.changed') {// A membership started, lapsed or was revoked: drop that member's cached badge answers for// that creator now instead of waiting out the cache TTL (VIP convergence).@@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {const results = await h.subscriptions.done;-- assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['vip.membership.changed', 'created']]);-- assert.strictEqual(stubEvents.subs.length, 3);++ assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['network.user.token_valid_after', 'created'], ['vip.membership.changed', 'created']]);++ assert.strictEqual(stubEvents.subs.length, 4);for (const s of stubEvents.subs) {assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');@@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operatoconst endpoint = subs.endpointFor(config, h.port);const posts = stubEvents.posts;let r = await subs.ensure({ config, endpoint });-- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);++ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists']);assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);r = await subs.ensure({ config, endpoint });-- assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['vip.membership.changed', 'exists', true]]);++ assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['network.user.token_valid_after', 'exists', true], ['vip.membership.changed', 'exists', true]]);assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });assert.strictEqual(stubEvents.subs[0].enabled, true);r = await subs.ensure({ config, endpoint, action: 'list' });-- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);++ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists']);});t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {nnew file mode 100644@@@ -0,0 +1,86 @@++'use strict';++/**++ * Revocation propagation in Chat (WS-B task 4, Contracts 0.39.0 network.user.token_valid_after):++ * a sign-out everywhere closes that person's chat socket within 5 seconds, their older token is refused++ * on REST and on a new socket, a token issued after the cutoff still works, other people and bots are++ * untouched, a stale (older) event changes nothing and a redelivery is a no-op.++ */++const assert = require('assert');++const crypto = require('crypto');++const { ids } = require('openvibe-contracts');++const { boot, suite, b64url } = require('./helpers');++++const t = suite('revocation');++let h, alice, bob, bot;++const now = () => Math.floor(Date.now() / 1000);++const jwt = (claims) => {++ const head = `${b64url(JSON.stringify({ alg: 'RS256', typ: 'JWT' }))}.${b64url(JSON.stringify(claims))}`;++ return `${head}.${b64url(crypto.sign('RSA-SHA256', Buffer.from(head), h.keys.privateKey))}`;++};++const tokenFor = (u, iatOffset = -60) => jwt({ iss: h.ISS, sub: String(u.id + 7000), subject_id: u.subject_id, username: u.username, role: u.role, iat: now() + iatOffset, exp: now() + 3600 });++const event = (subject, validAfter, reason = 'signed_out_everywhere') => ({++ event_id: ids.newId('event'), event_type: 'network.user.token_valid_after', version: 1, source: 'network',++ actor: { type: 'user', id: subject }, timestamp: new Date().toISOString(), visibility: 'internal',++ subject: { type: 'user', id: subject }, payload: { subject: { type: 'user', id: subject }, valid_after: new Date(validAfter).toISOString(), reason },++});++async function signedIn(token) {++ const ws = await h.ws({ bearer: token });++ ws.sendJson({ type: 'join', token });++ const auth = await ws.next((m) => m.type === 'auth');++ return { ws, auth };++}++const closed = (ws) => new Promise((resolve) => { if (ws.readyState === 3) return resolve({ code: ws._closeCode }); ws.once('close', (code) => resolve({ code })); });++++t('boot', async () => {++ h = await boot();++ alice = h.addUser('alice', { subject: ids.newId('user') });++ bob = h.addUser('bob', { subject: ids.newId('user') });++ bot = h.addUser('helperbot', { subject: alice.subject_id, apiScopes: ['chat'] });++ for (const u of [alice, bob]) h.ctx.upsertUser(h.live.users.get(u.id));++ assert.ok(require('../server/events/consumer').TOPICS.includes('network.user.token_valid_after'), 'Chat subscribes to it');++});++++t('a sign-out everywhere closes the person\'s socket within 5 seconds; others stay', async () => {++ const a = await signedIn(tokenFor(alice));++ const b = await signedIn(tokenFor(bob));++ assert.strictEqual(a.auth.core_username, 'alice');++ const started = Date.now();++ const out = h.eventsConsumer.apply(event(alice.subject_id, Date.now() - 5000));++ assert.strictEqual(out.outcome, 'revoked');++ const c = await closed(a.ws);++ assert.strictEqual(c.code, 4001);++ assert.ok(Date.now() - started < 5000, `closed in ${Date.now() - started} ms`);++ assert.ok(a.ws.all.some((m) => m.type === 'auth_revoked'), 'told why');++ assert.strictEqual(b.ws.readyState, 1, 'bob is untouched');++ b.ws.close();++});++++t('the old token is refused afterwards; a newer one works; a redelivery and an older cutoff change nothing', async () => {++ const old = tokenFor(alice, -60);++ assert.strictEqual((await h.http('GET', '/api/chat/search?q=x', { token: old })).status, 401);++ const again = await signedIn(old);++ assert.notStrictEqual(again.auth.core_username, 'alice', 'reconnects as a guest');++ again.ws.close();++ const fresh = await signedIn(tokenFor(alice, 0));++ assert.strictEqual(fresh.auth.core_username, 'alice', 'a token issued after the cutoff works');++ const ev = event(alice.subject_id, Date.now() - 120000, 'password_changed');++ assert.strictEqual(h.eventsConsumer.apply(ev).outcome, 'unchanged', 'an older cutoff never moves it back');++ assert.strictEqual(h.eventsConsumer.apply(ev).duplicate, true);++ assert.strictEqual(fresh.ws.readyState, 1);++ fresh.ws.close();++});++++t('bots on an API token and events from anyone but Network are left alone', async () => {++ const b = await signedIn(bot.token);++ assert.strictEqual(b.auth.core_username, 'helperbot');++ assert.strictEqual(h.eventsConsumer.apply(event(alice.subject_id, Date.now() + 1000)).outcome, 'revoked');++ await new Promise((r) => setTimeout(r, 200));++ assert.strictEqual(b.ws.readyState, 1, 'API-token socket stays');++ b.ws.close();++ const forged = { ...event(bob.subject_id, Date.now()), source: 'live' };++ assert.strictEqual(h.eventsConsumer.apply(forged).outcome, 'ignored:source');++ const bad = event(bob.subject_id, Date.now()); bad.payload.valid_after = 'soon';++ assert.strictEqual(h.eventsConsumer.apply(bad).outcome, 'ignored:payload');++});++++t.run(async () => { if (h && h.close) await h.close(); });
@@ -16,7 +16,7 @@
@@ -16,7 +16,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.35.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.39.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0",
"ws": "^8.18.0"
"ws": "^8.18.0"
@@ -1062,9 +1062,9 @@
@@ -1062,9 +1062,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.35.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.35.0",
- "integrity": "sha512-5jkH9Gt9MVbYaxo/Fi1VSH0RwdPkbary7SOm2m9e5EvXYu8L/cJHMm6KrMKxa7MV6TJCATDeRpGLDOCaUeZmSw==",
+ "version": "0.39.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.39.0",
+ "integrity": "sha512-ek3tryU3pPwmFSg621gzr4ss1OE/gpufBnxk9XFDLSfUe2RuZpYVZCJNdQ0F/9GfA83Xop+xJW8+IjRyjFg6Yg==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -27,7 +27,7 @@
@@ -27,7 +27,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.35.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.39.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0",
"ws": "^8.18.0"
"ws": "^8.18.0"
@@ -23,6 +23,7 @@ const config = require('../config');
@@ -23,6 +23,7 @@ const config = require('../config');
const db = require('../db/database');
const db = require('../db/database');
const serviceAuth = require('../net/service-auth');
const serviceAuth = require('../net/service-auth');
const ctx = require('../live-context');
const ctx = require('../live-context');
+const revocations = require('./revocations');
const ROLE_RANK = { user: 0, streamer: 1, global_mod: 2, admin: 3 };
const ROLE_RANK = { user: 0, streamer: 1, global_mod: 2, admin: 3 };
const stats = { local: 0, live: 0, rejected: 0 };
const stats = { local: 0, live: 0, rejected: 0 };
@@ -76,22 +77,38 @@ function refuse(token) {
@@ -76,22 +77,38 @@ function refuse(token) {
/** Resolve a browser/bot token to a chat user (a fresh object) or null. */
/** Resolve a browser/bot token to a chat user (a fresh object) or null. */
async function authenticate(token) {
async function authenticate(token) {
if (!token) return null;
if (!token) return null;
- const viaLive = async () => { stats.live++; return ctx.authenticate(token); };
+ const viaLive = async () => {
+ stats.live++;
+ const u = await ctx.authenticate(token);
+ // Live resolved a Network token (first visit, another key): the same cutoff applies.
+ const iat = u && u.auth_source !== 'api_token' ? tokenIat(token) : null;
+ if (iat != null && iat * 1000 < revocations.cutoffFor(u.subject_id)) return refuse(token);
+ return u;
+ };
if (String(token).startsWith('hbt_')) return viaLive();
if (String(token).startsWith('hbt_')) return viaLive();
const key = await serviceAuth.ensureKey();
const key = await serviceAuth.ensureKey();
if (!key) return viaLive();
if (!key) return viaLive();
const r = verify(token, key);
const r = verify(token, key);
// Only a token signed by the key we hold is decided here; anything else is Live's to judge.
// Only a token signed by the key we hold is decided here; anything else is Live's to judge.
if (!r.ok) return r.reason === 'signature' || r.reason === 'malformed' ? viaLive() : refuse(token);
if (!r.ok) return r.reason === 'signature' || r.reason === 'malformed' ? viaLive() : refuse(token);
+ // Signed out everywhere, password changed, banned…: Network's cutoff for this person (WS-B task 4).
+ if (revocations.isRevoked(r.claims)) return refuse(token);
const user = userFor(r.claims);
const user = userFor(r.claims);
if (!user) return viaLive();
if (!user) return viaLive();
stats.local++;
stats.local++;
return user;
return user;
}
}
+/** A JWT's iat (seconds), read without verifying (the token was verified when it was accepted); null otherwise. */
+function tokenIat(token) {
+ const parts = String(token || '').split('.');
+ if (parts.length !== 3) return null;
+ try { const c = b64json(parts[1]); return typeof c.iat === 'number' ? c.iat : null; } catch { return null; }
+}
+
/** Why the last resolution of this token failed: 'invalid' (bad/expired) or 'unresolved' (no account). */
/** Why the last resolution of this token failed: 'invalid' (bad/expired) or 'unresolved' (no account). */
function failureReason(token) {
function failureReason(token) {
return (token && _reasons.get(tokenKey(token))) || ctx.authFailureReason(token);
return (token && _reasons.get(tokenKey(token))) || ctx.authFailureReason(token);
}
}
-module.exports = { authenticate, failureReason, verify, userFor, stats: () => ({ ...stats }) };
+module.exports = { authenticate, failureReason, verify, userFor, tokenIat, stats: () => ({ ...stats }) };
new file mode 100644
new file mode 100644
@@ -0,0 +1,61 @@
@@ -0,0 +1,61 @@
+/**
+ * OpenVibe.Chat — per-person token cutoffs (roadmap WS-B task 4; Contracts 0.39.0
+ * network.user.token_valid_after).
+ *
+ * When a person signs out everywhere, changes or resets their password, is banned or has their
+ * sessions ended by staff, Network moves their cutoff and says so. Chat keeps the latest cutoff per
+ * subject here, refuses a Network token issued before it (auth/network-session.js, Network's rule:
+ * iat * 1000 < valid_after) and closes the sockets such tokens opened (chat-server revokeSubject).
+ */
+'use strict';
+
+const db = require('../db/database');
+
+let ready = false;
+const cache = new Map(); // subject → valid_after ms (0 = none known)
+
+function ensureSchema() {
+ if (ready) return;
+ db.getDb().exec(`CREATE TABLE IF NOT EXISTS token_revocations (
+ subject_id TEXT PRIMARY KEY,
+ valid_after_ms INTEGER NOT NULL,
+ reason TEXT,
+ updated_at INTEGER NOT NULL
+ )`);
+ ready = true;
+}
+
+/** The cutoff for a subject in ms, 0 when none. */
+function cutoffFor(subject) {
+ if (!subject) return 0;
+ if (cache.has(subject)) return cache.get(subject);
+ ensureSchema();
+ const row = db.get('SELECT valid_after_ms FROM token_revocations WHERE subject_id = ?', [subject]);
+ const ms = row ? Number(row.valid_after_ms) || 0 : 0;
+ if (cache.size > 50000) cache.clear();
+ cache.set(subject, ms);
+ return ms;
+}
+
+/** Keep the later cutoff (events can arrive out of order). Returns true when it moved forward. */
+function record(subject, validAfterMs, reason = null, now = Date.now()) {
+ ensureSchema();
+ const current = cutoffFor(subject);
+ if (!(validAfterMs > current)) return false;
+ db.run(`INSERT INTO token_revocations (subject_id, valid_after_ms, reason, updated_at) VALUES (?, ?, ?, ?)
+ ON CONFLICT(subject_id) DO UPDATE SET valid_after_ms = excluded.valid_after_ms, reason = excluded.reason, updated_at = excluded.updated_at
+ WHERE excluded.valid_after_ms > token_revocations.valid_after_ms`, [subject, validAfterMs, reason, now]);
+ cache.set(subject, validAfterMs);
+ return true;
+}
+
+/** Was a token with these claims issued before its subject's cutoff? */
+function isRevoked(claims) {
+ if (!claims || typeof claims.iat !== 'number') return false;
+ return claims.iat * 1000 < cutoffFor(claims.subject_id);
+}
+
+/** For tests: forget the in-memory cache (the table stays). */
+function _reset() { cache.clear(); ready = false; }
+
+module.exports = { ensureSchema, cutoffFor, record, isRevoked, _reset };
@@ -23,6 +23,7 @@ const crypto = require('crypto');
@@ -23,6 +23,7 @@ const crypto = require('crypto');
const db = require('../db/database');
const db = require('../db/database');
const ctx = require('../live-context');
const ctx = require('../live-context');
const { extractWsToken, authenticateWs } = require('../auth/auth');
const { extractWsToken, authenticateWs } = require('../auth/auth');
+const session = require('../auth/network-session');
const { clientIpOf } = require('../net/client-ip');
const { clientIpOf } = require('../net/client-ip');
const permissions = require('../auth/permissions');
const permissions = require('../auth/permissions');
const wordFilter = require('./word-filter');
const wordFilter = require('./word-filter');
@@ -283,7 +284,7 @@ class ChatServer {
@@ -283,7 +284,7 @@ class ChatServer {
ws.off('close', onEarlyClose);
ws.off('close', onEarlyClose);
ws.off('error', onEarlyClose);
ws.off('error', onEarlyClose);
if (closedEarly || ws.readyState !== WebSocket.OPEN) return;
if (closedEarly || ws.readyState !== WebSocket.OPEN) return;
- this._registerConnection(ws, req, { ip, streamId, user, early });
+ this._registerConnection(ws, req, { ip, streamId, user, early, tokenIat: user ? session.tokenIat(token) : null });
})().catch((err) => {
})().catch((err) => {
console.warn('[Chat] connection setup failed:', err.message);
console.warn('[Chat] connection setup failed:', err.message);
try { ws.close(1011, 'setup failed'); } catch {}
try { ws.close(1011, 'setup failed'); } catch {}
@@ -291,7 +292,7 @@ class ChatServer {
@@ -291,7 +292,7 @@ class ChatServer {
}
}
/** The rest of Live's handleConnection, once the socket's identity is known. */
/** The rest of Live's handleConnection, once the socket's identity is known. */
- _registerConnection(ws, req, { ip, streamId, user, early }) {
+ _registerConnection(ws, req, { ip, streamId, user, early, tokenIat = null }) {
const perIp = this._ipSockets.get(ip) || 0;
const perIp = this._ipSockets.get(ip) || 0;
if (ip && ip !== 'unknown' && perIp >= MAX_CHAT_SOCKETS_PER_IP && !permissions.can(user, 'staff.limits.exempt')) {
if (ip && ip !== 'unknown' && perIp >= MAX_CHAT_SOCKETS_PER_IP && !permissions.can(user, 'staff.limits.exempt')) {
ws.close(4029, 'Too many connections');
ws.close(4029, 'Too many connections');
@@ -317,6 +318,8 @@ class ChatServer {
@@ -317,6 +318,8 @@ class ChatServer {
joinedAt: Date.now(),
joinedAt: Date.now(),
// Handle Live can address replies to (arena commands) through the bridge.
// Handle Live can address replies to (arena commands) through the bridge.
connId: crypto.randomUUID(),
connId: crypto.randomUUID(),
+ // When the socket's token was issued (seconds), so a sign-out everywhere can close it.
+ tokenIat,
};
};
this.clients.set(ws, clientInfo);
this.clients.set(ws, clientInfo);
@@ -404,6 +407,7 @@ class ChatServer {
@@ -404,6 +407,7 @@ class ChatServer {
if (user) {
if (user) {
if (!client.user || client.user.id === user.id) {
if (!client.user || client.user.id === user.id) {
client.user = user;
client.user = user;
+ client.tokenIat = session.tokenIat(msg.token);
client.anonId = null; // no longer anonymous
client.anonId = null; // no longer anonymous
} else {
} else {
console.warn(`[Chat] Ignoring token identity mismatch for ${client.user.username} -> ${user.username}`);
console.warn(`[Chat] Ignoring token identity mismatch for ${client.user.username} -> ${user.username}`);
@@ -2493,6 +2497,25 @@ class ChatServer {
@@ -2493,6 +2497,25 @@ class ChatServer {
}
}
}
}
+ /**
+ * Network moved this person's token cutoff (network.user.token_valid_after: signed out everywhere,
+ * password changed, banned…). Close every socket they opened with an older Network token; the
+ * browser reconnects and, holding no valid token, carries on as a guest. Bot sockets (hbt_ API
+ * tokens) are not Network sessions and stay. Returns how many closed.
+ */
+ revokeSubject(subjectId, validAfterMs) {
+ let closed = 0;
+ for (const [ws, client] of this.clients) {
+ const u = client.user;
+ if (!u || !subjectId || u.subject_id !== subjectId || u.auth_source === 'api_token' || u._authSource === 'api_token') continue;
+ if (client.tokenIat != null && client.tokenIat * 1000 >= validAfterMs) continue;
+ try { if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'auth_revoked', reason: 'signed_out' })); } catch { /* closing anyway */ }
+ try { ws.close(4001, 'Signed out'); } catch { /* already gone */ }
+ closed++;
+ }
+ return closed;
+ }
+
/**
/**
* Push a profile/identity update to all chat connections belonging to a user.
* Push a profile/identity update to all chat connections belonging to a user.
* Refreshes cached client.user so subsequent messages use the new info.
* Refreshes cached client.user so subsequent messages use the new info.
@@ -37,10 +37,13 @@ const { viaProxy } = require('../net/service-auth');
@@ -37,10 +37,13 @@ const { viaProxy } = require('../net/service-auth');
const deployNotice = require('../chat/deploy-notice');
const deployNotice = require('../chat/deploy-notice');
const prefs = require('../prefs/chat-preferences');
const prefs = require('../prefs/chat-preferences');
const vipBadges = require('../vip/badges');
const vipBadges = require('../vip/badges');
+const revocations = require('../auth/revocations');
+const ctx = require('../live-context');
const CONSUMER = 'chat';
const CONSUMER = 'chat';
const INBOX_TABLE = 'chat_event_inbox';
const INBOX_TABLE = 'chat_event_inbox';
-const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'vip.membership.changed']);
+const TOPICS = Object.freeze(['live.release.deployed', 'network.module.updated', 'network.user.token_valid_after', 'vip.membership.changed']);
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered
const INBOX_KEEP_MS = 35 * 24 * 3600 * 1000; // Events keeps events 30 days: nothing older can be redelivered
@@ -71,6 +74,27 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()
@@ -71,6 +74,27 @@ function createEventsConsumer({ chatServer, secrets = [], now = () => Date.now()
if (p.namespace !== prefs.NAMESPACE) return 'ignored:namespace';
if (p.namespace !== prefs.NAMESPACE) return 'ignored:namespace';
return () => (prefs.handleEvent(event) ? 'invalidated' : 'unchanged');
return () => (prefs.handleEvent(event) ? 'invalidated' : 'unchanged');
}
}
+ if (event.event_type === 'network.user.token_valid_after') {
+ // Signed out everywhere, password changed, banned…: refuse this person's older tokens and
+ // close the sockets they opened (WS-B task 4).
+ if (event.source !== 'network') return 'ignored:source';
+ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
+ const subject = p.subject && p.subject.id;
+ const ms = Date.parse(p.valid_after);
+ if (!SUBJECT_RE.test(String(subject || '')) || !Number.isFinite(ms)) return 'ignored:payload';
+ return () => {
+ const moved = revocations.record(subject, ms, typeof p.reason === 'string' ? p.reason.slice(0, 40) : null, now());
+ if (!moved) return 'unchanged';
+ return {
+ outcome: 'revoked',
+ after: () => {
+ for (const r of db.all('SELECT id FROM ctx_users WHERE subject_id = ?', [subject])) ctx.invalidateUser(r.id);
+ const closed = chatServer ? chatServer.revokeSubject(subject, ms) : 0;
+ if (closed) log.log && log.log(`[Events consumer] ${p.reason || 'revoked'}: closed ${closed} socket(s)`);
+ },
+ };
+ };
+ }
if (event.event_type === 'vip.membership.changed') {
if (event.event_type === 'vip.membership.changed') {
// A membership started, lapsed or was revoked: drop that member's cached badge answers for
// A membership started, lapsed or was revoked: drop that member's cached badge answers for
// that creator now instead of waiting out the cache TTL (VIP convergence).
// that creator now instead of waiting out the cache TTL (VIP convergence).
@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {
@@ -123,8 +123,8 @@ t('boot with the consumer on and a stub Events', async () => {
t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {
t('boot created one subscription per topic, to /internal/events, with the consumer’s secret', async () => {
const results = await h.subscriptions.done;
const results = await h.subscriptions.done;
- assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['vip.membership.changed', 'created']]);
- assert.strictEqual(stubEvents.subs.length, 3);
+ assert.deepStrictEqual(results.map((r) => [r.topic, r.result]), [['live.release.deployed', 'created'], ['network.module.updated', 'created'], ['network.user.token_valid_after', 'created'], ['vip.membership.changed', 'created']]);
+ assert.strictEqual(stubEvents.subs.length, 4);
for (const s of stubEvents.subs) {
for (const s of stubEvents.subs) {
assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);
assert.strictEqual(s.endpoint, `http://127.0.0.1:${h.port}/internal/events`);
assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');
assert.strictEqual(s.secret, SECRET, 'the first CHAT_EVENTS_SECRET is handed to Events');
@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operato
@@ -139,17 +139,17 @@ t('a second boot changes nothing, and never re-enables a subscription an operato
const endpoint = subs.endpointFor(config, h.port);
const endpoint = subs.endpointFor(config, h.port);
const posts = stubEvents.posts;
const posts = stubEvents.posts;
let r = await subs.ensure({ config, endpoint });
let r = await subs.ensure({ config, endpoint });
- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);
+ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists']);
assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');
assert.strictEqual(stubEvents.posts, posts, 'nothing was created again');
r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });
r = await subs.ensure({ config, endpoint, action: 'disable', topics: ['live.release.deployed'] });
assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);
assert.deepStrictEqual(r.map((x) => [x.result, x.enabled]), [['disabled', false]]);
r = await subs.ensure({ config, endpoint });
r = await subs.ensure({ config, endpoint });
- assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['vip.membership.changed', 'exists', true]]);
+ assert.deepStrictEqual(r.map((x) => [x.topic, x.result, x.enabled]), [['live.release.deployed', 'exists', false], ['network.module.updated', 'exists', true], ['network.user.token_valid_after', 'exists', true], ['vip.membership.changed', 'exists', true]]);
assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');
assert.strictEqual(stubEvents.subs[0].enabled, false, 'still disabled');
r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });
r = await subs.ensure({ config, endpoint, action: 'enable', topics: ['live.release.deployed'] });
assert.strictEqual(stubEvents.subs[0].enabled, true);
assert.strictEqual(stubEvents.subs[0].enabled, true);
r = await subs.ensure({ config, endpoint, action: 'list' });
r = await subs.ensure({ config, endpoint, action: 'list' });
- assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists']);
+ assert.deepStrictEqual(r.map((x) => x.result), ['exists', 'exists', 'exists', 'exists']);
});
});
t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {
t('signature v2 only: bad signature, stale or future timestamp, v1-only, unsigned → 401', async () => {
new file mode 100644
new file mode 100644
@@ -0,0 +1,86 @@
@@ -0,0 +1,86 @@
+'use strict';
+/**
+ * Revocation propagation in Chat (WS-B task 4, Contracts 0.39.0 network.user.token_valid_after):
+ * a sign-out everywhere closes that person's chat socket within 5 seconds, their older token is refused
+ * on REST and on a new socket, a token issued after the cutoff still works, other people and bots are
+ * untouched, a stale (older) event changes nothing and a redelivery is a no-op.
+ */
+const assert = require('assert');
+const crypto = require('crypto');
+const { ids } = require('openvibe-contracts');
+const { boot, suite, b64url } = require('./helpers');
+
+const t = suite('revocation');
+let h, alice, bob, bot;
+const now = () => Math.floor(Date.now() / 1000);
+const jwt = (claims) => {
+ const head = `${b64url(JSON.stringify({ alg: 'RS256', typ: 'JWT' }))}.${b64url(JSON.stringify(claims))}`;
+ return `${head}.${b64url(crypto.sign('RSA-SHA256', Buffer.from(head), h.keys.privateKey))}`;
+};
+const tokenFor = (u, iatOffset = -60) => jwt({ iss: h.ISS, sub: String(u.id + 7000), subject_id: u.subject_id, username: u.username, role: u.role, iat: now() + iatOffset, exp: now() + 3600 });
+const event = (subject, validAfter, reason = 'signed_out_everywhere') => ({
+ event_id: ids.newId('event'), event_type: 'network.user.token_valid_after', version: 1, source: 'network',
+ actor: { type: 'user', id: subject }, timestamp: new Date().toISOString(), visibility: 'internal',
+ subject: { type: 'user', id: subject }, payload: { subject: { type: 'user', id: subject }, valid_after: new Date(validAfter).toISOString(), reason },
+});
+async function signedIn(token) {
+ const ws = await h.ws({ bearer: token });
+ ws.sendJson({ type: 'join', token });
+ const auth = await ws.next((m) => m.type === 'auth');
+ return { ws, auth };
+}
+const closed = (ws) => new Promise((resolve) => { if (ws.readyState === 3) return resolve({ code: ws._closeCode }); ws.once('close', (code) => resolve({ code })); });
+
+t('boot', async () => {
+ h = await boot();
+ alice = h.addUser('alice', { subject: ids.newId('user') });
+ bob = h.addUser('bob', { subject: ids.newId('user') });
+ bot = h.addUser('helperbot', { subject: alice.subject_id, apiScopes: ['chat'] });
+ for (const u of [alice, bob]) h.ctx.upsertUser(h.live.users.get(u.id));
+ assert.ok(require('../server/events/consumer').TOPICS.includes('network.user.token_valid_after'), 'Chat subscribes to it');
+});
+
+t('a sign-out everywhere closes the person\'s socket within 5 seconds; others stay', async () => {
+ const a = await signedIn(tokenFor(alice));
+ const b = await signedIn(tokenFor(bob));
+ assert.strictEqual(a.auth.core_username, 'alice');
+ const started = Date.now();
+ const out = h.eventsConsumer.apply(event(alice.subject_id, Date.now() - 5000));
+ assert.strictEqual(out.outcome, 'revoked');
+ const c = await closed(a.ws);
+ assert.strictEqual(c.code, 4001);
+ assert.ok(Date.now() - started < 5000, `closed in ${Date.now() - started} ms`);
+ assert.ok(a.ws.all.some((m) => m.type === 'auth_revoked'), 'told why');
+ assert.strictEqual(b.ws.readyState, 1, 'bob is untouched');
+ b.ws.close();
+});
+
+t('the old token is refused afterwards; a newer one works; a redelivery and an older cutoff change nothing', async () => {
+ const old = tokenFor(alice, -60);
+ assert.strictEqual((await h.http('GET', '/api/chat/search?q=x', { token: old })).status, 401);
+ const again = await signedIn(old);
+ assert.notStrictEqual(again.auth.core_username, 'alice', 'reconnects as a guest');
+ again.ws.close();
+ const fresh = await signedIn(tokenFor(alice, 0));
+ assert.strictEqual(fresh.auth.core_username, 'alice', 'a token issued after the cutoff works');
+ const ev = event(alice.subject_id, Date.now() - 120000, 'password_changed');
+ assert.strictEqual(h.eventsConsumer.apply(ev).outcome, 'unchanged', 'an older cutoff never moves it back');
+ assert.strictEqual(h.eventsConsumer.apply(ev).duplicate, true);
+ assert.strictEqual(fresh.ws.readyState, 1);
+ fresh.ws.close();
+});
+
+t('bots on an API token and events from anyone but Network are left alone', async () => {
+ const b = await signedIn(bot.token);
+ assert.strictEqual(b.auth.core_username, 'helperbot');
+ assert.strictEqual(h.eventsConsumer.apply(event(alice.subject_id, Date.now() + 1000)).outcome, 'revoked');
+ await new Promise((r) => setTimeout(r, 200));
+ assert.strictEqual(b.ws.readyState, 1, 'API-token socket stays');
+ b.ws.close();
+ const forged = { ...event(bob.subject_id, Date.now()), source: 'live' };
+ assert.strictEqual(h.eventsConsumer.apply(forged).outcome, 'ignored:source');
+ const bad = event(bob.subject_id, Date.now()); bad.payload.valid_after = 'soon';
+ assert.strictEqual(h.eventsConsumer.apply(bad).outcome, 'ignored:payload');
+});
+
+t.run(async () => { if (h && h.close) await h.close(); });