openvibers/openvibe.events Commits
Events: the Network key and token checks come from openvibe-sdk/auth (#51)
Changes
@@@ -22,13 +22,13 @@ npm test # every test/*.test.js, temp databases, no network neededNode 22 in production (`fnm exec --using=22.22.1 npm test`). Production: `/opt/openvibe.events`, env `/etc/openvibe/events.env`, unit [deploy/systemd/openvibe-events.service](deploy/systemd/openvibe-events.service), store PostgreSQL (`ov_events` on the host's data role), nginx [deploy/nginx/openvibe.events.conf](deploy/nginx/openvibe.events.conf) (the public vhost, on the API and product origin: it exposes the home and its discovery files, `/realtime/stream`, health, and the token-guarded `/api/v1/events`, `/api/v1/subscriptions` and `/api/v1/checkpoints` for developer apps; `/api/v1/deliveries` stays host-local; services on the host call `127.0.0.1:4300`).--`GET /api/health` is liveness. `GET /api/ready` (openvibe-shared/ready) is 200 only when every required check passes — `db` (a real query), `network_jwks` (the Network signing key has loaded; it retries every 30 s while Network boots) and `delivery_worker` (when `EVENTS_WORKER` is on) — and 503 otherwise, with `failed: [...]`. The optional `dlq` check fails once more than `EVENTS_DLQ_DEGRADED_AT` (default 100) deliveries are dead: the service stays ready (200) and reports `status: "degraded"`, `degraded: ["dlq"]`. Each check carries `status`, `required`, `latency_ms`, `checked_at` and, for `dlq`, `detail: { depth, threshold }`; `latest_seq`, `deliveries`, `worker` and `realtime_connections` are still in the body. **Shape change (Track O):** `checks` used to be booleans (`{ db, worker, key }`); they are now objects keyed `db`, `network_jwks`, `delivery_worker`, `dlq`.++`GET /api/health` is liveness. `GET /api/ready` (openvibe-shared/ready) is 200 only when every required check passes — `db` (a real query), `network_jwks` (the Network signing key has loaded: openvibe-sdk/auth's JWKS client, retried every 30 s while Network boots, refreshed every 15 minutes, refetched at once for a token naming an unknown kid, and the last good keys kept through a Network outage) and `delivery_worker` (when `EVENTS_WORKER` is on) — and 503 otherwise, with `failed: [...]`. The optional `dlq` check fails once more than `EVENTS_DLQ_DEGRADED_AT` (default 100) deliveries are dead: the service stays ready (200) and reports `status: "degraded"`, `degraded: ["dlq"]`. Each check carries `status`, `required`, `latency_ms`, `checked_at` and, for `dlq`, `detail: { depth, threshold }`; `latest_seq`, `deliveries`, `worker` and `realtime_connections` are still in the body. **Shape change (Track O):** `checks` used to be booleans (`{ db, worker, key }`); they are now objects keyed `db`, `network_jwks`, `delivery_worker`, `dlq`.`GET /metrics` (Prometheus text, direct loopback callers only — a request carrying `X-Forwarded-For` gets 404, and nginx refuses the path): HTTP golden signals by route template (`http_requests_total`, `http_request_duration_seconds`, `http_requests_in_flight`; SSE sessions excluded), process metrics, `release_info`, and `events_deliveries{status}`, `events_dlq_depth`, `events_latest_seq`, `events_realtime_connections`, `events_delivery_latency_seconds{priority}` (acceptance → successful delivery, retries included), `events_delivery_attempts_total{outcome}`. `GET /release.json` is the deployed release (ADR-016, `release.mount` from openvibe-shared 1.5.0, with the manifest 1.1.0 fields under openvibe-contracts 0.32.0); its `POST /release-metrics` would count open tabs' update reports as `release_client_updates_total`, but Events has no pages and nginx does not proxy that path. `GET /limits.json` lists the developer limits enforced here (per project and environment), read from the running config; [openvibe.services/docs/limits](https://openvibe.services/docs/limits) renders it.## Auth--Services call with an OpenVibe.Network client-credentials token (`POST /oauth/token`, `audience=openvibe.events`), verified offline against the Network JWKS. Each route checks one capability:++Services call with an OpenVibe.Network client-credentials token (`POST /oauth/token`, `audience=openvibe.events`), verified offline against the Network JWKS (openvibe-sdk/auth `verifyServiceToken` with the pinned openvibe-contracts rules; people's tokens with `verifyUserToken`, which refuses service principals and typed tokens). Each route checks one capability:| Capability | Routes ||---|---|@@@ -3,8 +3,10 @@* Authentication and capability checks.** - 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; or, on++ * openvibe.events), verified with openvibe-sdk/auth verifyServiceToken, which applies the pinned++ * openvibe-contracts serviceAuth rules with the key the token's kid names.++ * - Browsers: Network user JWTs (RS256, same key; openvibe-sdk/auth verifyUserToken, which refuses++ * service principals and typed tokens), 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@@@ -16,14 +18,20 @@* (appOrService guard). Only there is a sandbox token (env=sandbox) accepted; every other route* refuses it with token.sandbox_refused.*++ * The key: Network's JWKS through openvibe-sdk/auth's process-wide client (the last good keys through a++ * Network outage, a rotation honoured on an unknown kid), or a pinned PEM (OV_NETWORK_PUBLIC_KEY).++ ** The service capabilities are in openvibe-contracts since v0.7; events.app.* arrive in v0.28.* capabilities.check() answers capability.unknown for an id the installed release does not know;* until then hasCap() decides with the same grant rule (exact id, or a `family.*` grant).*/const crypto = require('crypto');--const { serviceAuth, capabilities, http } = require('openvibe-contracts');++const contracts = require('openvibe-contracts');++const sdkAuth = require('openvibe-sdk/auth');const apps = require('./apps');++const { capabilities, http } = contracts;++/** The capability ids this service enforces. */const CAPS = Object.freeze({publish: 'events.event.publish',@@@ -41,62 +49,61 @@ const CAPS = Object.freeze({// ── Network public key ─────────────────────────────────────++const JWKS_RETRY_MS = 30 * 1000;++const JWKS_REFRESH_MS = 15 * 60 * 1000;++/**-- * Loads the Network signing key from GET /api/.well-known/jwks ({ keys: [jwk] } or the older-- * { public_key: PEM }), retrying every 30 s until it loads and refreshing every 6 h after that.-- * A PEM given in config (OV_NETWORK_PUBLIC_KEY) is used as is and never fetched.++ * Network's signing keys. With a PEM in config (OV_NETWORK_PUBLIC_KEY) that one key, never fetched. Otherwise++ * <networkInternalUrl>/api/.well-known/jwks through openvibe-sdk/auth's process-wide JWKS client (the verifiers read++ * the same client): start() retries every 30 s until the first load (Network may still be booting), then refreshes++ * every 15 minutes; a token naming an unknown kid refetches at once (a rotation).++ *++ * keys.verifyOptions what the SDK verifiers take: { publicKey } or { jwks: url, fetch, log }++ * keys.keysFor(kid) [{ kid, key }] to verify a realtime ticket with; throws when none has loaded*/--function createKeyStore({ urls = [], pem = null, fetchImpl = globalThis.fetch, log = console } = {}) {-- let key = pem ? toPem(pem) : null;++function createKeys({ url = null, pem = null, fetchImpl = globalThis.fetch, log = console } = {}) {++ const pinned = pem ? crypto.createPublicKey(pem) : null;++ const jwksUrl = pinned || !url ? null : `${url.replace(/\/+$/, '')}/api/.well-known/jwks`;++ const client = jwksUrl ? sdkAuth.jwksClient(jwksUrl, { fetch: fetchImpl, log }) : null;let retryTimer = null;-- let refreshTimer = null;-- function toPem(value) {-- return crypto.createPublicKey(value).export({ type: 'spki', format: 'pem' });-- }++ const loaded = () => Boolean(pinned) || Boolean(client && client.status().ready);-- async function fetchOnce() {-- for (const base of urls) {-- if (!base) continue;-- const url = `${base}/api/.well-known/jwks`;-- try {-- const res = await fetchImpl(url, { signal: AbortSignal.timeout(8000) });-- if (!res.ok) throw new Error(`HTTP ${res.status}`);-- const body = await res.json();-- const jwk = (body.keys || []).find(k => k.kty === 'RSA');-- if (jwk) key = toPem({ key: jwk, format: 'jwk' });-- else if (typeof body.public_key === 'string' && body.public_key.includes('BEGIN')) key = toPem(body.public_key);-- else throw new Error('no RSA key in response');-- log.log(`[auth] Network public key loaded from ${base}`);-- return key;-- } catch (err) {-- log.warn(`[auth] key fetch from ${url} failed: ${err.message}`);-- }-- }-- return null;++ /** One fetch now; the keys, or null (the client logged why). */++ async function refresh() {++ if (!client) return null;++ try { return await client.refresh(); } catch { return null; }}async function start() {-- if (pem) return Promise.resolve(key);++ if (!client) return pinned;++ client.start({ intervalMs: JWKS_REFRESH_MS });const attempt = async () => {-- const k = await fetchOnce();-- if (!k && !key) {-- retryTimer = setTimeout(attempt, 30 * 1000);++ retryTimer = null;++ const got = await refresh();++ if (!loaded()) {++ retryTimer = setTimeout(attempt, JWKS_RETRY_MS);retryTimer.unref?.();}-- return k;++ return got;};-- refreshTimer = setInterval(() => { fetchOnce().catch(() => {}); }, 6 * 60 * 60 * 1000);-- refreshTimer.unref?.();return await attempt();}function stop() {clearTimeout(retryTimer);-- clearInterval(refreshTimer);++ retryTimer = null;++ if (client) client.stop();++ }++++ async function keysFor(kid) {++ if (pinned) return [{ kid: null, key: pinned }];++ if (!client) return [];++ return await client.keysForKid(kid || null);}-- return { get: () => key, loaded: () => Boolean(key), start, stop, fetchOnce };++ const verifyOptions = pinned ? { publicKey: pinned } : { jwks: jwksUrl, fetch: fetchImpl, log };++ return { url: jwksUrl, loaded, start, stop, refresh, keysFor, verifyOptions };}// ── Capabilities ───────────────────────────────────────────@@@ -119,31 +126,6 @@ function allows(claims, id) {const b64json = (s) => JSON.parse(Buffer.from(String(s), 'base64url').toString('utf8'));--/** Network user JWT (RS256). Returns claims or null. Service tokens are never accepted as users. */--function verifyUserJwt(token, { publicKey, issuer, audiences, now = Date.now() }) {-- if (!publicKey || typeof token !== 'string') return null;-- const parts = token.split('.');-- if (parts.length !== 3) return null;-- try {-- const header = b64json(parts[0]);-- if (header.alg !== 'RS256') return null;-- const ok = crypto.verify('RSA-SHA256', Buffer.from(`${parts[0]}.${parts[1]}`), publicKey, Buffer.from(parts[2], 'base64url'));-- if (!ok) return null;-- const claims = b64json(parts[1]);-- const t = Math.floor(now / 1000);-- if (typeof claims.exp !== 'number' || claims.exp + 30 < t) return null;-- if (issuer && claims.iss !== issuer) return null;-- 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}$/;@@@ -151,13 +133,14 @@ 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).++ * { ok: false, code, reason }. Checks the signature (against `keys`, [{ kid, key }]: the one the ticket's kid++ * names, else each), 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). Events is the one service that accepts these, so the check stays here rather than in the SDK.*/--function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {++function verifyRealtimeTicket(token, { keys = [], 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 (!keys.length) 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');@@@ -169,8 +152,12 @@ function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {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; }++ const byKid = header.kid ? keys.filter((k) => k.kid === header.kid) : [];++ const input = Buffer.from(`${parts[0]}.${parts[1]}`);++ const sig = Buffer.from(parts[2], 'base64url');++ const signed = (byKid.length ? byKid : keys).some(({ key }) => {++ try { return crypto.verify('RSA-SHA256', input, key, sig); } catch { return 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)');@@@ -185,6 +172,11 @@ function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {return { ok: true, claims };}++/** The kid a JWT's header names, or null (never throws: an unreadable token simply has none). */++function headerKid(token) {++ try { return JSON.parse(Buffer.from(String(token).split('.')[0], 'base64url').toString('utf8')).kid || null; } catch { return null; }++}++function bearer(req) {const h = String(req.headers.authorization || '');return h.startsWith('Bearer ') ? h.slice(7).trim() : null;@@@ -219,17 +211,18 @@ function createAuth({ config, keys, store = null }) {}// Token lifetimes are judged on wall-clock time, never on the worker's (injectable) clock.-- function verifyService(token, { acceptSandbox = false } = {}) {-- const publicKey = keys.get();-- if (!publicKey) return { ok: false, code: 'token.unavailable', reason: 'signing key not loaded yet' };-- const r = serviceAuth.verifyServiceToken(token, { publicKey, issuer: config.issuer, audience: config.audience, acceptSandbox });-- // Belt and braces for a contracts release that does not know `env` yet.-- if (r.ok && r.claims.env === 'sandbox' && !acceptSandbox) return { ok: false, code: 'token.sandbox_refused', reason: 'sandbox tokens are not accepted here' };-- return r;++ // → { ok: true, claims } | { ok: false, code, reason }; token.unavailable when no key has loaded (a 503).++ async function verifyService(token, { acceptSandbox = false } = {}) {++ return await sdkAuth.verifyServiceToken(token, { ...keys.verifyOptions, issuer: config.issuer, audience: config.audience, contracts, acceptSandbox });}-- function verifyUser(token) {-- return verifyUserJwt(token, { publicKey: keys.get(), issuer: config.issuer, audiences: config.userAudiences });++ /** A person's session token → its claims, or null (service principals and typed tokens never are one). */++ async function verifyUser(token) {++ try {++ return await sdkAuth.verifyUserToken(token, { ...keys.verifyOptions, issuer: config.issuer, audience: config.userAudiences });++ } catch {++ return null;++ }}/**@@@ -238,10 +231,13 @@ function createAuth({ config, keys, store = null }) {*/function requireCap(id, { requireService = false } = {}) {return function capGuard(req, res, next) {++ capCheck(req, res, next).catch(next);++ };++ async function capCheck(req, res, next) {const ctx = req.ov;const token = bearer(req);if (!token) return http.sendProblem(res, 401, 'token.missing', { detail: 'a service token is required', ctx });-- const r = verifyService(token);++ const r = await verifyService(token);if (!r.ok) return http.sendProblem(res, r.code === 'token.unavailable' ? 503 : 401, r.code, { detail: r.reason, ctx });// Developer apps only ever act through events.app.* (appOrService); a first-party// capability in an app token is never honoured.@@@ -252,7 +248,7 @@ function createAuth({ config, keys, store = null }) {if (requireService && !service) return http.sendProblem(res, 403, 'capability.denied', { detail: 'only service principals may do this', ctx });req.principal = { kind: 'service', sub: r.claims.sub, service, cap: r.claims.cap, ns: r.claims.ns || [], jti: r.claims.jti };return next();-- };++ }}/**@@@ -263,11 +259,14 @@ function createAuth({ config, keys, store = null }) {*/function appOrService(serviceCap, appCap, { requireService = false } = {}) {const serviceGuard = requireCap(serviceCap, { requireService });-- return async function appOrServiceGuard(req, res, next) {++ return function appOrServiceGuard(req, res, next) {++ appCheck(req, res, next).catch(next);++ };++ async function appCheck(req, res, next) {const ctx = req.ov;const token = bearer(req);if (!token || !config.apps.enabled) return serviceGuard(req, res, next);-- const r = verifyService(token, { acceptSandbox: true });++ const r = await verifyService(token, { acceptSandbox: true });if (!r.ok || !/^app:/.test(String(r.claims.sub))) return serviceGuard(req, res, next);const principal = apps.appPrincipal(r.claims);if (principal.error) return http.sendProblem(res, 401, 'token.invalid_claims', { detail: principal.error, ctx });@@@ -278,7 +277,7 @@ function createAuth({ config, keys, store = null }) {if (!c.allowed) return http.sendProblem(res, 403, c.code, { detail: c.reason, ctx });req.principal = { ...principal, cap: r.claims.cap, jti: r.claims.jti, iat: r.claims.iat };return next();-- };++ }}/**@@@ -288,17 +287,20 @@ function createAuth({ config, keys, store = null }) {* 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) {++ async 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 });++ const ticket = Array.isArray(q.ticket) ? null : q.ticket;++ let ticketKeys = [];++ try { ticketKeys = await keys.keysFor(headerKid(ticket)); } catch { ticketKeys = []; }++ const r = verifyRealtimeTicket(ticket, { keys: ticketKeys, 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);++ const svc = await verifyService(token);if (svc.ok) {// Only a first-party service principal is a service viewer: a developer app (or any// non-svc principal) never is, whatever capabilities its token carries (as in requireCap).@@@ -308,12 +310,12 @@ function createAuth({ config, keys, store = null }) {return { kind: 'service', sub: svc.claims.sub };}if (svc.code === 'token.unavailable') return { error: { status: 503, code: svc.code, detail: svc.reason } };-- const user = verifyUser(token);++ const user = await verifyUser(token);if (user) return userViewer(user);return { error: { status: 401, code: 'token.invalid', detail: 'token does not verify' } };}const fromCookie = cookie(req, 'ov_token');-- const user = fromCookie ? verifyUser(fromCookie) : null;++ const user = fromCookie ? await verifyUser(fromCookie) : null;if (user) return userViewer(user);return { kind: 'anonymous' };}@@@ -326,4 +328,4 @@ function createAuth({ config, keys, store = null }) {return { verifyService, verifyUser, requireCap, appOrService, realtimeViewer };}--module.exports = { CAPS, createKeyStore, createAuth, hasCap, allows, verifyUserJwt, verifyRealtimeTicket, serviceSlug, bearer, cookie };++module.exports = { CAPS, createKeys, createAuth, hasCap, allows, verifyRealtimeTicket, serviceSlug, bearer, cookie };@@@ -12,7 +12,7 @@ const { openDb, createStore } = require('./store');const { createPolicyLibrary } = require('./fabric/policy');const { createCarriers } = require('./fabric/carriers');const { createPlanner } = require('./fabric/planner');--const { createKeyStore, createAuth } = require('./auth');++const { createKeys, createAuth } = require('./auth');const { createWorker } = require('./worker');const { createRealtime } = require('./realtime');const { createApp } = require('./app');@@@ -38,7 +38,7 @@ async function start({const carriers = createCarriers({ config, clock, log, valkey });const planner = createPlanner({ registry: carriers, db, clock, log });const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner });-- const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });++ const keys = createKeys({ url: config.networkInternalUrl, pem: config.networkPublicKey, fetchImpl, log });const auth = createAuth({ config, keys, store });const metrics = createMetrics({ store });// appPost / dnsLookup: developer-app delivery and the subscribe-time DNS check (server/egress.js);@@@ -119,7 +119,7 @@ function createRealtime({ store, auth, config, clock = { now: () => Date.now() }const bad = wanted.find(t => !topics.isValidPattern(t));if (bad) return http.sendProblem(res, 400, 'realtime.bad_topic', { detail: `invalid topic pattern ${bad}`, ctx });-- const viewer = auth.realtimeViewer(req);++ const viewer = await auth.realtimeViewer(req);if (viewer.error) return http.sendProblem(res, viewer.error.status, viewer.error.code, { detail: viewer.error.detail, ctx });if (viewer.kind === 'anonymous' && !opts.allowAnonymous) {return http.sendProblem(res, 401, 'token.missing', { detail: 'sign in to open a realtime stream', ctx });@@@ -149,14 +149,16 @@ t('stop', async () => { await h.stop(); });t('key loader: JWKS {keys:[jwk]} and legacy {public_key}; ready is 503 until it loads', async () => {const jwk = crypto.createPublicKey(publicKey).export({ format: 'jwk' });let shape = 'none';-- const net = nodeHttp.createServer((req, res) => {-- if (req.url !== '/api/.well-known/jwks' || shape === 'none') { res.statusCode = 503; return res.end(); }-- res.setHeader('Content-Type', 'application/json');-- res.end(JSON.stringify(shape === 'jwks' ? { keys: [{ ...jwk, alg: 'RS256', kid: 'k1' }] } : { public_key: publicKey, algorithm: 'RS256' }));-- });-- await new Promise(r => net.listen(0, '127.0.0.1', r));-- const url = `http://127.0.0.1:${net.address().port}`;++ // One Network per shape: openvibe-sdk/auth keeps one JWKS client per URL for the process, so a second boot on++ // the same URL would start with the first one's keys (as a restarted verifier in one process should).for (const s of ['jwks', 'pem']) {++ const net = nodeHttp.createServer((req, res) => {++ if (req.url !== '/api/.well-known/jwks' || shape === 'none') { res.statusCode = 503; return res.end(); }++ res.setHeader('Content-Type', 'application/json');++ res.end(JSON.stringify(shape === 'jwks' ? { keys: [{ ...jwk, alg: 'RS256', kid: 'k1' }] } : { public_key: publicKey, algorithm: 'RS256' }));++ });++ await new Promise(r => net.listen(0, '127.0.0.1', r));++ const url = `http://127.0.0.1:${net.address().port}`;shape = 'none';const x = await boot({ env: { OV_NETWORK_PUBLIC_KEY: '', OV_NETWORK_INTERNAL_URL: url, OV_NETWORK_URL: url, OV_NETWORK_ISSUER: 'https://openvibe.network' }, worker: 'on' });await x.keyLoaded;@@@ -167,14 +169,62 @@ t('key loader: JWKS {keys:[jwk]} and legacy {public_key}; ready is 503 until itr = await request(x.base, 'POST', '/api/v1/events', { token: live, body: envelope() });assert.strictEqual(r.status, 503, 'no key yet: service unavailable, not unauthorized');shape = s;-- await x.keys.fetchOnce();++ await x.keys.refresh();r = await request(x.base, 'GET', '/api/ready');assert.strictEqual(r.status, 200, s);r = await request(x.base, 'POST', '/api/v1/events', { token: live, body: envelope() });assert.strictEqual(r.status, 201, s);await x.stop();++ net.close();++ }++ await sleep(10);++});++++t('key rotation: a token naming a new kid verifies after one refetch; an unpublished key never does', async () => {++ const pair = () => crypto.generateKeyPairSync('rsa', { modulusLength: 2048 });++ const k1 = pair(); const k2 = pair(); const rogue = pair();++ const jwkOf = (k, kid) => ({ ...k.publicKey.export({ format: 'jwk' }), alg: 'RS256', use: 'sig', kid });++ // Network's tokens carry the kid of the key that signed them.++ const sign = (k, kid) => {++ const now = Math.floor(Date.now() / 1000);++ const enc = (o) => Buffer.from(JSON.stringify(o)).toString('base64url');++ const input = `${enc({ alg: 'RS256', typ: 'JWT', kid })}.${enc({++ iss: 'https://openvibe.network', sub: 'svc:live', actor_type: 'service', aud: ['openvibe.events'], cap: ['events.event.publish'], ns: [],++ iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(8).toString('hex')}`,++ })}`;++ return `${input}.${crypto.sign('RSA-SHA256', Buffer.from(input), k.privateKey).toString('base64url')}`;++ };++ let published = [jwkOf(k1, 'k1')];++ let fetches = 0;++ const net = nodeHttp.createServer((req, res) => {++ if (req.url !== '/api/.well-known/jwks') { res.statusCode = 404; return res.end(); }++ fetches++;++ res.setHeader('Content-Type', 'application/json');++ res.end(JSON.stringify({ keys: published }));++ });++ await new Promise(r => net.listen(0, '127.0.0.1', r));++ const url = `http://127.0.0.1:${net.address().port}`;++ const x = await boot({ env: { OV_NETWORK_PUBLIC_KEY: '', OV_NETWORK_INTERNAL_URL: url, OV_NETWORK_URL: url, OV_NETWORK_ISSUER: 'https://openvibe.network' } });++ await x.keyLoaded;++ try {++ let r = await request(x.base, 'POST', '/api/v1/events', { token: sign(k1, 'k1'), body: envelope() });++ assert.strictEqual(r.status, 201, JSON.stringify(r.body));++ const before = fetches;++ published = [jwkOf(k2, 'k2'), jwkOf(k1, 'k1')];++ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(k2, 'k2'), body: envelope() });++ assert.strictEqual(r.status, 201, `the rotated key verifies: ${JSON.stringify(r.body)}`);++ assert.strictEqual(fetches, before + 1, 'one refetch for the unknown kid');++ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(k1, 'k1'), body: envelope() });++ assert.strictEqual(r.status, 201, 'the old key still verifies while Network publishes it');++ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(rogue, 'k9'), body: envelope() });++ assert.strictEqual(r.status, 401, 'a key Network never published');++ assert.strictEqual(r.body.code, 'token.bad_signature');++ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(rogue, 'k1'), body: envelope() });++ assert.strictEqual(r.status, 401, 'a published kid with another key behind it');++ } finally {++ await x.stop();++ net.close();}-- net.close();await sleep(10);});@@@ -215,7 +215,6 @@ const PATTERNS = {'net.connect': /\b(?:net|tls)\.(?:connect|createConnection)\s*\(/g,};const REVIEWED = {-- 'server/auth.js fetch': [1, 'the Network signing key: OV_NETWORK_INTERNAL_URL / OV_NETWORK_URL /api/.well-known/jwks (operator config)'],'server/worker.js fetch': [1, 'first-party deliveries: endpoints on the host allow-list (server/endpoints.js), checked again at delivery, redirect: manual'],'server/worker.js appPost': [1, 'developer-app deliveries: the guarded poster (server/egress.js), covered above'],'server/egress.js http.request': [1, 'the guarded poster\'s https.request (default requestImpl): its own lookup pins the checked address'],
@@ -22,13 +22,13 @@ npm test # every test/*.test.js, temp databases, no network needed
@@ -22,13 +22,13 @@ npm test # every test/*.test.js, temp databases, no network needed
Node 22 in production (`fnm exec --using=22.22.1 npm test`). Production: `/opt/openvibe.events`, env `/etc/openvibe/events.env`, unit [deploy/systemd/openvibe-events.service](deploy/systemd/openvibe-events.service), store PostgreSQL (`ov_events` on the host's data role), nginx [deploy/nginx/openvibe.events.conf](deploy/nginx/openvibe.events.conf) (the public vhost, on the API and product origin: it exposes the home and its discovery files, `/realtime/stream`, health, and the token-guarded `/api/v1/events`, `/api/v1/subscriptions` and `/api/v1/checkpoints` for developer apps; `/api/v1/deliveries` stays host-local; services on the host call `127.0.0.1:4300`).
Node 22 in production (`fnm exec --using=22.22.1 npm test`). Production: `/opt/openvibe.events`, env `/etc/openvibe/events.env`, unit [deploy/systemd/openvibe-events.service](deploy/systemd/openvibe-events.service), store PostgreSQL (`ov_events` on the host's data role), nginx [deploy/nginx/openvibe.events.conf](deploy/nginx/openvibe.events.conf) (the public vhost, on the API and product origin: it exposes the home and its discovery files, `/realtime/stream`, health, and the token-guarded `/api/v1/events`, `/api/v1/subscriptions` and `/api/v1/checkpoints` for developer apps; `/api/v1/deliveries` stays host-local; services on the host call `127.0.0.1:4300`).
-`GET /api/health` is liveness. `GET /api/ready` (openvibe-shared/ready) is 200 only when every required check passes — `db` (a real query), `network_jwks` (the Network signing key has loaded; it retries every 30 s while Network boots) and `delivery_worker` (when `EVENTS_WORKER` is on) — and 503 otherwise, with `failed: [...]`. The optional `dlq` check fails once more than `EVENTS_DLQ_DEGRADED_AT` (default 100) deliveries are dead: the service stays ready (200) and reports `status: "degraded"`, `degraded: ["dlq"]`. Each check carries `status`, `required`, `latency_ms`, `checked_at` and, for `dlq`, `detail: { depth, threshold }`; `latest_seq`, `deliveries`, `worker` and `realtime_connections` are still in the body. **Shape change (Track O):** `checks` used to be booleans (`{ db, worker, key }`); they are now objects keyed `db`, `network_jwks`, `delivery_worker`, `dlq`.
+`GET /api/health` is liveness. `GET /api/ready` (openvibe-shared/ready) is 200 only when every required check passes — `db` (a real query), `network_jwks` (the Network signing key has loaded: openvibe-sdk/auth's JWKS client, retried every 30 s while Network boots, refreshed every 15 minutes, refetched at once for a token naming an unknown kid, and the last good keys kept through a Network outage) and `delivery_worker` (when `EVENTS_WORKER` is on) — and 503 otherwise, with `failed: [...]`. The optional `dlq` check fails once more than `EVENTS_DLQ_DEGRADED_AT` (default 100) deliveries are dead: the service stays ready (200) and reports `status: "degraded"`, `degraded: ["dlq"]`. Each check carries `status`, `required`, `latency_ms`, `checked_at` and, for `dlq`, `detail: { depth, threshold }`; `latest_seq`, `deliveries`, `worker` and `realtime_connections` are still in the body. **Shape change (Track O):** `checks` used to be booleans (`{ db, worker, key }`); they are now objects keyed `db`, `network_jwks`, `delivery_worker`, `dlq`.
`GET /metrics` (Prometheus text, direct loopback callers only — a request carrying `X-Forwarded-For` gets 404, and nginx refuses the path): HTTP golden signals by route template (`http_requests_total`, `http_request_duration_seconds`, `http_requests_in_flight`; SSE sessions excluded), process metrics, `release_info`, and `events_deliveries{status}`, `events_dlq_depth`, `events_latest_seq`, `events_realtime_connections`, `events_delivery_latency_seconds{priority}` (acceptance → successful delivery, retries included), `events_delivery_attempts_total{outcome}`. `GET /release.json` is the deployed release (ADR-016, `release.mount` from openvibe-shared 1.5.0, with the manifest 1.1.0 fields under openvibe-contracts 0.32.0); its `POST /release-metrics` would count open tabs' update reports as `release_client_updates_total`, but Events has no pages and nginx does not proxy that path. `GET /limits.json` lists the developer limits enforced here (per project and environment), read from the running config; [openvibe.services/docs/limits](https://openvibe.services/docs/limits) renders it.
`GET /metrics` (Prometheus text, direct loopback callers only — a request carrying `X-Forwarded-For` gets 404, and nginx refuses the path): HTTP golden signals by route template (`http_requests_total`, `http_request_duration_seconds`, `http_requests_in_flight`; SSE sessions excluded), process metrics, `release_info`, and `events_deliveries{status}`, `events_dlq_depth`, `events_latest_seq`, `events_realtime_connections`, `events_delivery_latency_seconds{priority}` (acceptance → successful delivery, retries included), `events_delivery_attempts_total{outcome}`. `GET /release.json` is the deployed release (ADR-016, `release.mount` from openvibe-shared 1.5.0, with the manifest 1.1.0 fields under openvibe-contracts 0.32.0); its `POST /release-metrics` would count open tabs' update reports as `release_client_updates_total`, but Events has no pages and nginx does not proxy that path. `GET /limits.json` lists the developer limits enforced here (per project and environment), read from the running config; [openvibe.services/docs/limits](https://openvibe.services/docs/limits) renders it.
## Auth
## Auth
-Services call with an OpenVibe.Network client-credentials token (`POST /oauth/token`, `audience=openvibe.events`), verified offline against the Network JWKS. Each route checks one capability:
+Services call with an OpenVibe.Network client-credentials token (`POST /oauth/token`, `audience=openvibe.events`), verified offline against the Network JWKS (openvibe-sdk/auth `verifyServiceToken` with the pinned openvibe-contracts rules; people's tokens with `verifyUserToken`, which refuses service principals and typed tokens). Each route checks one capability:
| Capability | Routes |
| Capability | Routes |
|---|---|
|---|---|
@@ -3,8 +3,10 @@
@@ -3,8 +3,10 @@
* Authentication and capability checks.
* Authentication and capability checks.
*
*
* - 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.
- * - Browsers: Network user JWTs (RS256, same key), cookie `ov_token` or Bearer; or, on
+ * openvibe.events), verified with openvibe-sdk/auth verifyServiceToken, which applies the pinned
+ * openvibe-contracts serviceAuth rules with the key the token's kid names.
+ * - Browsers: Network user JWTs (RS256, same key; openvibe-sdk/auth verifyUserToken, which refuses
+ * service principals and typed tokens), cookie `ov_token` or Bearer; or, on
* /realtime/stream only, a realtime ticket (?ticket=): a two-minute, single-use RS256 JWT Network
* /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
* 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
* page on any OpenVibe site can open an EventSource without a third-party cookie. A ticket is
@@ -16,14 +18,20 @@
@@ -16,14 +18,20 @@
* (appOrService guard). Only there is a sandbox token (env=sandbox) accepted; every other route
* (appOrService guard). Only there is a sandbox token (env=sandbox) accepted; every other route
* refuses it with token.sandbox_refused.
* refuses it with token.sandbox_refused.
*
*
+ * The key: Network's JWKS through openvibe-sdk/auth's process-wide client (the last good keys through a
+ * Network outage, a rotation honoured on an unknown kid), or a pinned PEM (OV_NETWORK_PUBLIC_KEY).
+ *
* The service capabilities are in openvibe-contracts since v0.7; events.app.* arrive in v0.28.
* The service capabilities are in openvibe-contracts since v0.7; events.app.* arrive in v0.28.
* capabilities.check() answers capability.unknown for an id the installed release does not know;
* capabilities.check() answers capability.unknown for an id the installed release does not know;
* until then hasCap() decides with the same grant rule (exact id, or a `family.*` grant).
* until then hasCap() decides with the same grant rule (exact id, or a `family.*` grant).
*/
*/
const crypto = require('crypto');
const crypto = require('crypto');
-const { serviceAuth, capabilities, http } = require('openvibe-contracts');
+const contracts = require('openvibe-contracts');
+const sdkAuth = require('openvibe-sdk/auth');
const apps = require('./apps');
const apps = require('./apps');
+const { capabilities, http } = contracts;
+
/** The capability ids this service enforces. */
/** The capability ids this service enforces. */
const CAPS = Object.freeze({
const CAPS = Object.freeze({
publish: 'events.event.publish',
publish: 'events.event.publish',
@@ -41,62 +49,61 @@ const CAPS = Object.freeze({
@@ -41,62 +49,61 @@ const CAPS = Object.freeze({
// ── Network public key ─────────────────────────────────────
// ── Network public key ─────────────────────────────────────
+const JWKS_RETRY_MS = 30 * 1000;
+const JWKS_REFRESH_MS = 15 * 60 * 1000;
+
/**
/**
- * Loads the Network signing key from GET /api/.well-known/jwks ({ keys: [jwk] } or the older
- * { public_key: PEM }), retrying every 30 s until it loads and refreshing every 6 h after that.
- * A PEM given in config (OV_NETWORK_PUBLIC_KEY) is used as is and never fetched.
+ * Network's signing keys. With a PEM in config (OV_NETWORK_PUBLIC_KEY) that one key, never fetched. Otherwise
+ * <networkInternalUrl>/api/.well-known/jwks through openvibe-sdk/auth's process-wide JWKS client (the verifiers read
+ * the same client): start() retries every 30 s until the first load (Network may still be booting), then refreshes
+ * every 15 minutes; a token naming an unknown kid refetches at once (a rotation).
+ *
+ * keys.verifyOptions what the SDK verifiers take: { publicKey } or { jwks: url, fetch, log }
+ * keys.keysFor(kid) [{ kid, key }] to verify a realtime ticket with; throws when none has loaded
*/
*/
-function createKeyStore({ urls = [], pem = null, fetchImpl = globalThis.fetch, log = console } = {}) {
- let key = pem ? toPem(pem) : null;
+function createKeys({ url = null, pem = null, fetchImpl = globalThis.fetch, log = console } = {}) {
+ const pinned = pem ? crypto.createPublicKey(pem) : null;
+ const jwksUrl = pinned || !url ? null : `${url.replace(/\/+$/, '')}/api/.well-known/jwks`;
+ const client = jwksUrl ? sdkAuth.jwksClient(jwksUrl, { fetch: fetchImpl, log }) : null;
let retryTimer = null;
let retryTimer = null;
- let refreshTimer = null;
- function toPem(value) {
- return crypto.createPublicKey(value).export({ type: 'spki', format: 'pem' });
- }
+ const loaded = () => Boolean(pinned) || Boolean(client && client.status().ready);
- async function fetchOnce() {
- for (const base of urls) {
- if (!base) continue;
- const url = `${base}/api/.well-known/jwks`;
- try {
- const res = await fetchImpl(url, { signal: AbortSignal.timeout(8000) });
- if (!res.ok) throw new Error(`HTTP ${res.status}`);
- const body = await res.json();
- const jwk = (body.keys || []).find(k => k.kty === 'RSA');
- if (jwk) key = toPem({ key: jwk, format: 'jwk' });
- else if (typeof body.public_key === 'string' && body.public_key.includes('BEGIN')) key = toPem(body.public_key);
- else throw new Error('no RSA key in response');
- log.log(`[auth] Network public key loaded from ${base}`);
- return key;
- } catch (err) {
- log.warn(`[auth] key fetch from ${url} failed: ${err.message}`);
- }
- }
- return null;
+ /** One fetch now; the keys, or null (the client logged why). */
+ async function refresh() {
+ if (!client) return null;
+ try { return await client.refresh(); } catch { return null; }
}
}
async function start() {
async function start() {
- if (pem) return Promise.resolve(key);
+ if (!client) return pinned;
+ client.start({ intervalMs: JWKS_REFRESH_MS });
const attempt = async () => {
const attempt = async () => {
- const k = await fetchOnce();
- if (!k && !key) {
- retryTimer = setTimeout(attempt, 30 * 1000);
+ retryTimer = null;
+ const got = await refresh();
+ if (!loaded()) {
+ retryTimer = setTimeout(attempt, JWKS_RETRY_MS);
retryTimer.unref?.();
retryTimer.unref?.();
}
}
- return k;
+ return got;
};
};
- refreshTimer = setInterval(() => { fetchOnce().catch(() => {}); }, 6 * 60 * 60 * 1000);
- refreshTimer.unref?.();
return await attempt();
return await attempt();
}
}
function stop() {
function stop() {
clearTimeout(retryTimer);
clearTimeout(retryTimer);
- clearInterval(refreshTimer);
+ retryTimer = null;
+ if (client) client.stop();
+ }
+
+ async function keysFor(kid) {
+ if (pinned) return [{ kid: null, key: pinned }];
+ if (!client) return [];
+ return await client.keysForKid(kid || null);
}
}
- return { get: () => key, loaded: () => Boolean(key), start, stop, fetchOnce };
+ const verifyOptions = pinned ? { publicKey: pinned } : { jwks: jwksUrl, fetch: fetchImpl, log };
+ return { url: jwksUrl, loaded, start, stop, refresh, keysFor, verifyOptions };
}
}
// ── Capabilities ───────────────────────────────────────────
// ── Capabilities ───────────────────────────────────────────
@@ -119,31 +126,6 @@ function allows(claims, id) {
@@ -119,31 +126,6 @@ function allows(claims, id) {
const b64json = (s) => JSON.parse(Buffer.from(String(s), 'base64url').toString('utf8'));
const b64json = (s) => JSON.parse(Buffer.from(String(s), 'base64url').toString('utf8'));
-/** Network user JWT (RS256). Returns claims or null. Service tokens are never accepted as users. */
-function verifyUserJwt(token, { publicKey, issuer, audiences, now = Date.now() }) {
- if (!publicKey || typeof token !== 'string') return null;
- const parts = token.split('.');
- if (parts.length !== 3) return null;
- try {
- const header = b64json(parts[0]);
- if (header.alg !== 'RS256') return null;
- const ok = crypto.verify('RSA-SHA256', Buffer.from(`${parts[0]}.${parts[1]}`), publicKey, Buffer.from(parts[2], 'base64url'));
- if (!ok) return null;
- const claims = b64json(parts[1]);
- const t = Math.floor(now / 1000);
- if (typeof claims.exp !== 'number' || claims.exp + 30 < t) return null;
- if (issuer && claims.iss !== issuer) return null;
- 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_LIFETIME_MAX_S = 300;
const TICKET_SKEW_S = 30;
const TICKET_SKEW_S = 30;
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
@@ -151,13 +133,14 @@ const TICKET_JTI_RE = /^rtk_[0-9a-f]{24}$/;
@@ -151,13 +133,14 @@ const TICKET_JTI_RE = /^rtk_[0-9a-f]{24}$/;
/**
/**
* A realtime ticket (identity.realtime-ticket-claims@1). Returns { ok: true, claims } or
* 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).
+ * { ok: false, code, reason }. Checks the signature (against `keys`, [{ kid, key }]: the one the ticket's kid
+ * names, else each), 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). Events is the one service that accepts these, so the check stays here rather than in the SDK.
*/
*/
-function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {
+function verifyRealtimeTicket(token, { keys = [], issuer, now = Date.now() }) {
const bad = (reason) => ({ ok: false, code: 'ticket.invalid', reason });
const bad = (reason) => ({ ok: false, code: 'ticket.invalid', reason });
- if (!publicKey) return { ok: false, code: 'token.unavailable', reason: 'signing key not loaded yet' };
+ if (!keys.length) return { ok: false, code: 'token.unavailable', reason: 'signing key not loaded yet' };
if (typeof token !== 'string' || token.length > 4096) return bad('not a ticket');
if (typeof token !== 'string' || token.length > 4096) return bad('not a ticket');
const parts = token.split('.');
const parts = token.split('.');
if (parts.length !== 3) return bad('not a ticket');
if (parts.length !== 3) return bad('not a ticket');
@@ -169,8 +152,12 @@ function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {
@@ -169,8 +152,12 @@ function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {
return bad('not a ticket');
return bad('not a ticket');
}
}
if (!header || header.alg !== 'RS256' || !claims || typeof claims !== 'object') 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; }
+ const byKid = header.kid ? keys.filter((k) => k.kid === header.kid) : [];
+ const input = Buffer.from(`${parts[0]}.${parts[1]}`);
+ const sig = Buffer.from(parts[2], 'base64url');
+ const signed = (byKid.length ? byKid : keys).some(({ key }) => {
+ try { return crypto.verify('RSA-SHA256', input, key, sig); } catch { return false; }
+ });
if (!signed) return bad('signature does not verify');
if (!signed) return bad('signature does not verify');
if (claims.iss !== `${issuer}/realtime`) return bad('not a realtime ticket (issuer)');
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)');
if (claims.typ !== 'realtime' || claims.purpose !== 'realtime') return bad('not a realtime ticket (purpose)');
@@ -185,6 +172,11 @@ function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {
@@ -185,6 +172,11 @@ function verifyRealtimeTicket(token, { publicKey, issuer, now = Date.now() }) {
return { ok: true, claims };
return { ok: true, claims };
}
}
+/** The kid a JWT's header names, or null (never throws: an unreadable token simply has none). */
+function headerKid(token) {
+ try { return JSON.parse(Buffer.from(String(token).split('.')[0], 'base64url').toString('utf8')).kid || null; } catch { return null; }
+}
+
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;
@@ -219,17 +211,18 @@ function createAuth({ config, keys, store = null }) {
@@ -219,17 +211,18 @@ function createAuth({ config, keys, store = null }) {
}
}
// 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 } = {}) {
- const publicKey = keys.get();
- if (!publicKey) return { ok: false, code: 'token.unavailable', reason: 'signing key not loaded yet' };
- const r = serviceAuth.verifyServiceToken(token, { publicKey, issuer: config.issuer, audience: config.audience, acceptSandbox });
- // Belt and braces for a contracts release that does not know `env` yet.
- if (r.ok && r.claims.env === 'sandbox' && !acceptSandbox) return { ok: false, code: 'token.sandbox_refused', reason: 'sandbox tokens are not accepted here' };
- return r;
+ // → { ok: true, claims } | { ok: false, code, reason }; token.unavailable when no key has loaded (a 503).
+ async function verifyService(token, { acceptSandbox = false } = {}) {
+ return await sdkAuth.verifyServiceToken(token, { ...keys.verifyOptions, issuer: config.issuer, audience: config.audience, contracts, acceptSandbox });
}
}
- function verifyUser(token) {
- return verifyUserJwt(token, { publicKey: keys.get(), issuer: config.issuer, audiences: config.userAudiences });
+ /** A person's session token → its claims, or null (service principals and typed tokens never are one). */
+ async function verifyUser(token) {
+ try {
+ return await sdkAuth.verifyUserToken(token, { ...keys.verifyOptions, issuer: config.issuer, audience: config.userAudiences });
+ } catch {
+ return null;
+ }
}
}
/**
/**
@@ -238,10 +231,13 @@ function createAuth({ config, keys, store = null }) {
@@ -238,10 +231,13 @@ function createAuth({ config, keys, store = null }) {
*/
*/
function requireCap(id, { requireService = false } = {}) {
function requireCap(id, { requireService = false } = {}) {
return function capGuard(req, res, next) {
return function capGuard(req, res, next) {
+ capCheck(req, res, next).catch(next);
+ };
+ async function capCheck(req, res, next) {
const ctx = req.ov;
const ctx = req.ov;
const token = bearer(req);
const token = bearer(req);
if (!token) return http.sendProblem(res, 401, 'token.missing', { detail: 'a service token is required', ctx });
if (!token) return http.sendProblem(res, 401, 'token.missing', { detail: 'a service token is required', ctx });
- const r = verifyService(token);
+ const r = await verifyService(token);
if (!r.ok) return http.sendProblem(res, r.code === 'token.unavailable' ? 503 : 401, r.code, { detail: r.reason, ctx });
if (!r.ok) return http.sendProblem(res, r.code === 'token.unavailable' ? 503 : 401, r.code, { detail: r.reason, ctx });
// Developer apps only ever act through events.app.* (appOrService); a first-party
// Developer apps only ever act through events.app.* (appOrService); a first-party
// capability in an app token is never honoured.
// capability in an app token is never honoured.
@@ -252,7 +248,7 @@ function createAuth({ config, keys, store = null }) {
@@ -252,7 +248,7 @@ function createAuth({ config, keys, store = null }) {
if (requireService && !service) return http.sendProblem(res, 403, 'capability.denied', { detail: 'only service principals may do this', ctx });
if (requireService && !service) return http.sendProblem(res, 403, 'capability.denied', { detail: 'only service principals may do this', ctx });
req.principal = { kind: 'service', sub: r.claims.sub, service, cap: r.claims.cap, ns: r.claims.ns || [], jti: r.claims.jti };
req.principal = { kind: 'service', sub: r.claims.sub, service, cap: r.claims.cap, ns: r.claims.ns || [], jti: r.claims.jti };
return next();
return next();
- };
+ }
}
}
/**
/**
@@ -263,11 +259,14 @@ function createAuth({ config, keys, store = null }) {
@@ -263,11 +259,14 @@ function createAuth({ config, keys, store = null }) {
*/
*/
function appOrService(serviceCap, appCap, { requireService = false } = {}) {
function appOrService(serviceCap, appCap, { requireService = false } = {}) {
const serviceGuard = requireCap(serviceCap, { requireService });
const serviceGuard = requireCap(serviceCap, { requireService });
- return async function appOrServiceGuard(req, res, next) {
+ return function appOrServiceGuard(req, res, next) {
+ appCheck(req, res, next).catch(next);
+ };
+ async function appCheck(req, res, next) {
const ctx = req.ov;
const ctx = req.ov;
const token = bearer(req);
const token = bearer(req);
if (!token || !config.apps.enabled) return serviceGuard(req, res, next);
if (!token || !config.apps.enabled) return serviceGuard(req, res, next);
- const r = verifyService(token, { acceptSandbox: true });
+ const r = await verifyService(token, { acceptSandbox: true });
if (!r.ok || !/^app:/.test(String(r.claims.sub))) return serviceGuard(req, res, next);
if (!r.ok || !/^app:/.test(String(r.claims.sub))) return serviceGuard(req, res, next);
const principal = apps.appPrincipal(r.claims);
const principal = apps.appPrincipal(r.claims);
if (principal.error) return http.sendProblem(res, 401, 'token.invalid_claims', { detail: principal.error, ctx });
if (principal.error) return http.sendProblem(res, 401, 'token.invalid_claims', { detail: principal.error, ctx });
@@ -278,7 +277,7 @@ function createAuth({ config, keys, store = null }) {
@@ -278,7 +277,7 @@ function createAuth({ config, keys, store = null }) {
if (!c.allowed) return http.sendProblem(res, 403, c.code, { detail: c.reason, ctx });
if (!c.allowed) return http.sendProblem(res, 403, c.code, { detail: c.reason, ctx });
req.principal = { ...principal, cap: r.claims.cap, jti: r.claims.jti, iat: r.claims.iat };
req.principal = { ...principal, cap: r.claims.cap, jti: r.claims.jti, iat: r.claims.iat };
return next();
return next();
- };
+ }
}
}
/**
/**
@@ -288,17 +287,20 @@ function createAuth({ config, keys, store = null }) {
@@ -288,17 +287,20 @@ function createAuth({ config, keys, store = null }) {
* then nothing else is looked at. A Bearer token must verify; an unverifiable cookie (expired
* 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.
* session) degrades to anonymous so the browser still gets public events. The ticket is never logged.
*/
*/
- function realtimeViewer(req) {
+ async function realtimeViewer(req) {
const q = req.query || {};
const q = req.query || {};
if (q.ticket !== undefined) {
if (q.ticket !== undefined) {
- const r = verifyRealtimeTicket(Array.isArray(q.ticket) ? null : q.ticket, { publicKey: keys.get(), issuer: config.issuer });
+ const ticket = Array.isArray(q.ticket) ? null : q.ticket;
+ let ticketKeys = [];
+ try { ticketKeys = await keys.keysFor(headerKid(ticket)); } catch { ticketKeys = []; }
+ const r = verifyRealtimeTicket(ticket, { keys: ticketKeys, issuer: config.issuer });
if (!r.ok) return { error: { status: r.code === 'token.unavailable' ? 503 : 401, code: r.code, detail: r.reason } };
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' } };
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' };
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 = await verifyService(token);
if (svc.ok) {
if (svc.ok) {
// Only a first-party service principal is a service viewer: a developer app (or any
// Only a first-party service principal is a service viewer: a developer app (or any
// non-svc principal) never is, whatever capabilities its token carries (as in requireCap).
// non-svc principal) never is, whatever capabilities its token carries (as in requireCap).
@@ -308,12 +310,12 @@ function createAuth({ config, keys, store = null }) {
@@ -308,12 +310,12 @@ function createAuth({ config, keys, store = null }) {
return { kind: 'service', sub: svc.claims.sub };
return { kind: 'service', sub: svc.claims.sub };
}
}
if (svc.code === 'token.unavailable') return { error: { status: 503, code: svc.code, detail: svc.reason } };
if (svc.code === 'token.unavailable') return { error: { status: 503, code: svc.code, detail: svc.reason } };
- const user = verifyUser(token);
+ const user = await verifyUser(token);
if (user) return userViewer(user);
if (user) return userViewer(user);
return { error: { status: 401, code: 'token.invalid', detail: 'token does not verify' } };
return { error: { status: 401, code: 'token.invalid', detail: 'token does not verify' } };
}
}
const fromCookie = cookie(req, 'ov_token');
const fromCookie = cookie(req, 'ov_token');
- const user = fromCookie ? verifyUser(fromCookie) : null;
+ const user = fromCookie ? await verifyUser(fromCookie) : null;
if (user) return userViewer(user);
if (user) return userViewer(user);
return { kind: 'anonymous' };
return { kind: 'anonymous' };
}
}
@@ -326,4 +328,4 @@ function createAuth({ config, keys, store = null }) {
@@ -326,4 +328,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, verifyRealtimeTicket, serviceSlug, bearer, cookie };
+module.exports = { CAPS, createKeys, createAuth, hasCap, allows, verifyRealtimeTicket, serviceSlug, bearer, cookie };
@@ -12,7 +12,7 @@ const { openDb, createStore } = require('./store');
@@ -12,7 +12,7 @@ const { openDb, createStore } = require('./store');
const { createPolicyLibrary } = require('./fabric/policy');
const { createPolicyLibrary } = require('./fabric/policy');
const { createCarriers } = require('./fabric/carriers');
const { createCarriers } = require('./fabric/carriers');
const { createPlanner } = require('./fabric/planner');
const { createPlanner } = require('./fabric/planner');
-const { createKeyStore, createAuth } = require('./auth');
+const { createKeys, createAuth } = require('./auth');
const { createWorker } = require('./worker');
const { createWorker } = require('./worker');
const { createRealtime } = require('./realtime');
const { createRealtime } = require('./realtime');
const { createApp } = require('./app');
const { createApp } = require('./app');
@@ -38,7 +38,7 @@ async function start({
@@ -38,7 +38,7 @@ async function start({
const carriers = createCarriers({ config, clock, log, valkey });
const carriers = createCarriers({ config, clock, log, valkey });
const planner = createPlanner({ registry: carriers, db, clock, log });
const planner = createPlanner({ registry: carriers, db, clock, log });
const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner });
const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner });
- const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
+ const keys = createKeys({ url: config.networkInternalUrl, pem: config.networkPublicKey, fetchImpl, log });
const auth = createAuth({ config, keys, store });
const auth = createAuth({ config, keys, store });
const metrics = createMetrics({ store });
const metrics = createMetrics({ store });
// appPost / dnsLookup: developer-app delivery and the subscribe-time DNS check (server/egress.js);
// appPost / dnsLookup: developer-app delivery and the subscribe-time DNS check (server/egress.js);
@@ -119,7 +119,7 @@ function createRealtime({ store, auth, config, clock = { now: () => Date.now() }
@@ -119,7 +119,7 @@ function createRealtime({ store, auth, config, clock = { now: () => Date.now() }
const bad = wanted.find(t => !topics.isValidPattern(t));
const bad = wanted.find(t => !topics.isValidPattern(t));
if (bad) return http.sendProblem(res, 400, 'realtime.bad_topic', { detail: `invalid topic pattern ${bad}`, ctx });
if (bad) return http.sendProblem(res, 400, 'realtime.bad_topic', { detail: `invalid topic pattern ${bad}`, ctx });
- const viewer = auth.realtimeViewer(req);
+ const viewer = await auth.realtimeViewer(req);
if (viewer.error) return http.sendProblem(res, viewer.error.status, viewer.error.code, { detail: viewer.error.detail, ctx });
if (viewer.error) return http.sendProblem(res, viewer.error.status, viewer.error.code, { detail: viewer.error.detail, ctx });
if (viewer.kind === 'anonymous' && !opts.allowAnonymous) {
if (viewer.kind === 'anonymous' && !opts.allowAnonymous) {
return http.sendProblem(res, 401, 'token.missing', { detail: 'sign in to open a realtime stream', ctx });
return http.sendProblem(res, 401, 'token.missing', { detail: 'sign in to open a realtime stream', ctx });
@@ -149,14 +149,16 @@ t('stop', async () => { await h.stop(); });
@@ -149,14 +149,16 @@ t('stop', async () => { await h.stop(); });
t('key loader: JWKS {keys:[jwk]} and legacy {public_key}; ready is 503 until it loads', async () => {
t('key loader: JWKS {keys:[jwk]} and legacy {public_key}; ready is 503 until it loads', async () => {
const jwk = crypto.createPublicKey(publicKey).export({ format: 'jwk' });
const jwk = crypto.createPublicKey(publicKey).export({ format: 'jwk' });
let shape = 'none';
let shape = 'none';
- const net = nodeHttp.createServer((req, res) => {
- if (req.url !== '/api/.well-known/jwks' || shape === 'none') { res.statusCode = 503; return res.end(); }
- res.setHeader('Content-Type', 'application/json');
- res.end(JSON.stringify(shape === 'jwks' ? { keys: [{ ...jwk, alg: 'RS256', kid: 'k1' }] } : { public_key: publicKey, algorithm: 'RS256' }));
- });
- await new Promise(r => net.listen(0, '127.0.0.1', r));
- const url = `http://127.0.0.1:${net.address().port}`;
+ // One Network per shape: openvibe-sdk/auth keeps one JWKS client per URL for the process, so a second boot on
+ // the same URL would start with the first one's keys (as a restarted verifier in one process should).
for (const s of ['jwks', 'pem']) {
for (const s of ['jwks', 'pem']) {
+ const net = nodeHttp.createServer((req, res) => {
+ if (req.url !== '/api/.well-known/jwks' || shape === 'none') { res.statusCode = 503; return res.end(); }
+ res.setHeader('Content-Type', 'application/json');
+ res.end(JSON.stringify(shape === 'jwks' ? { keys: [{ ...jwk, alg: 'RS256', kid: 'k1' }] } : { public_key: publicKey, algorithm: 'RS256' }));
+ });
+ await new Promise(r => net.listen(0, '127.0.0.1', r));
+ const url = `http://127.0.0.1:${net.address().port}`;
shape = 'none';
shape = 'none';
const x = await boot({ env: { OV_NETWORK_PUBLIC_KEY: '', OV_NETWORK_INTERNAL_URL: url, OV_NETWORK_URL: url, OV_NETWORK_ISSUER: 'https://openvibe.network' }, worker: 'on' });
const x = await boot({ env: { OV_NETWORK_PUBLIC_KEY: '', OV_NETWORK_INTERNAL_URL: url, OV_NETWORK_URL: url, OV_NETWORK_ISSUER: 'https://openvibe.network' }, worker: 'on' });
await x.keyLoaded;
await x.keyLoaded;
@@ -167,14 +169,62 @@ t('key loader: JWKS {keys:[jwk]} and legacy {public_key}; ready is 503 until it
@@ -167,14 +169,62 @@ t('key loader: JWKS {keys:[jwk]} and legacy {public_key}; ready is 503 until it
r = await request(x.base, 'POST', '/api/v1/events', { token: live, body: envelope() });
r = await request(x.base, 'POST', '/api/v1/events', { token: live, body: envelope() });
assert.strictEqual(r.status, 503, 'no key yet: service unavailable, not unauthorized');
assert.strictEqual(r.status, 503, 'no key yet: service unavailable, not unauthorized');
shape = s;
shape = s;
- await x.keys.fetchOnce();
+ await x.keys.refresh();
r = await request(x.base, 'GET', '/api/ready');
r = await request(x.base, 'GET', '/api/ready');
assert.strictEqual(r.status, 200, s);
assert.strictEqual(r.status, 200, s);
r = await request(x.base, 'POST', '/api/v1/events', { token: live, body: envelope() });
r = await request(x.base, 'POST', '/api/v1/events', { token: live, body: envelope() });
assert.strictEqual(r.status, 201, s);
assert.strictEqual(r.status, 201, s);
await x.stop();
await x.stop();
+ net.close();
+ }
+ await sleep(10);
+});
+
+t('key rotation: a token naming a new kid verifies after one refetch; an unpublished key never does', async () => {
+ const pair = () => crypto.generateKeyPairSync('rsa', { modulusLength: 2048 });
+ const k1 = pair(); const k2 = pair(); const rogue = pair();
+ const jwkOf = (k, kid) => ({ ...k.publicKey.export({ format: 'jwk' }), alg: 'RS256', use: 'sig', kid });
+ // Network's tokens carry the kid of the key that signed them.
+ const sign = (k, kid) => {
+ const now = Math.floor(Date.now() / 1000);
+ const enc = (o) => Buffer.from(JSON.stringify(o)).toString('base64url');
+ const input = `${enc({ alg: 'RS256', typ: 'JWT', kid })}.${enc({
+ iss: 'https://openvibe.network', sub: 'svc:live', actor_type: 'service', aud: ['openvibe.events'], cap: ['events.event.publish'], ns: [],
+ iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(8).toString('hex')}`,
+ })}`;
+ return `${input}.${crypto.sign('RSA-SHA256', Buffer.from(input), k.privateKey).toString('base64url')}`;
+ };
+ let published = [jwkOf(k1, 'k1')];
+ let fetches = 0;
+ const net = nodeHttp.createServer((req, res) => {
+ if (req.url !== '/api/.well-known/jwks') { res.statusCode = 404; return res.end(); }
+ fetches++;
+ res.setHeader('Content-Type', 'application/json');
+ res.end(JSON.stringify({ keys: published }));
+ });
+ await new Promise(r => net.listen(0, '127.0.0.1', r));
+ const url = `http://127.0.0.1:${net.address().port}`;
+ const x = await boot({ env: { OV_NETWORK_PUBLIC_KEY: '', OV_NETWORK_INTERNAL_URL: url, OV_NETWORK_URL: url, OV_NETWORK_ISSUER: 'https://openvibe.network' } });
+ await x.keyLoaded;
+ try {
+ let r = await request(x.base, 'POST', '/api/v1/events', { token: sign(k1, 'k1'), body: envelope() });
+ assert.strictEqual(r.status, 201, JSON.stringify(r.body));
+ const before = fetches;
+ published = [jwkOf(k2, 'k2'), jwkOf(k1, 'k1')];
+ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(k2, 'k2'), body: envelope() });
+ assert.strictEqual(r.status, 201, `the rotated key verifies: ${JSON.stringify(r.body)}`);
+ assert.strictEqual(fetches, before + 1, 'one refetch for the unknown kid');
+ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(k1, 'k1'), body: envelope() });
+ assert.strictEqual(r.status, 201, 'the old key still verifies while Network publishes it');
+ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(rogue, 'k9'), body: envelope() });
+ assert.strictEqual(r.status, 401, 'a key Network never published');
+ assert.strictEqual(r.body.code, 'token.bad_signature');
+ r = await request(x.base, 'POST', '/api/v1/events', { token: sign(rogue, 'k1'), body: envelope() });
+ assert.strictEqual(r.status, 401, 'a published kid with another key behind it');
+ } finally {
+ await x.stop();
+ net.close();
}
}
- net.close();
await sleep(10);
await sleep(10);
});
});
@@ -215,7 +215,6 @@ const PATTERNS = {
@@ -215,7 +215,6 @@ const PATTERNS = {
'net.connect': /\b(?:net|tls)\.(?:connect|createConnection)\s*\(/g,
'net.connect': /\b(?:net|tls)\.(?:connect|createConnection)\s*\(/g,
};
};
const REVIEWED = {
const REVIEWED = {
- 'server/auth.js fetch': [1, 'the Network signing key: OV_NETWORK_INTERNAL_URL / OV_NETWORK_URL /api/.well-known/jwks (operator config)'],
'server/worker.js fetch': [1, 'first-party deliveries: endpoints on the host allow-list (server/endpoints.js), checked again at delivery, redirect: manual'],
'server/worker.js fetch': [1, 'first-party deliveries: endpoints on the host allow-list (server/endpoints.js), checked again at delivery, redirect: manual'],
'server/worker.js appPost': [1, 'developer-app deliveries: the guarded poster (server/egress.js), covered above'],
'server/worker.js appPost': [1, 'developer-app deliveries: the guarded poster (server/egress.js), covered above'],
'server/egress.js http.request': [1, 'the guarded poster\'s https.request (default requestImpl): its own lookup pins the checked address'],
'server/egress.js http.request': [1, 'the guarded poster\'s https.request (default requestImpl): its own lookup pins the checked address'],