diff --git a/.changeset/graphql-realtime-identity-2992.md b/.changeset/graphql-realtime-identity-2992.md new file mode 100644 index 0000000000..df55bd4adf --- /dev/null +++ b/.changeset/graphql-realtime-identity-2992.md @@ -0,0 +1,38 @@ +--- +'@objectstack/runtime': patch +'@objectstack/spec': patch +'@objectstack/service-realtime': patch +--- + +fix(security): pre-wiring identity admission for the GraphQL and realtime surfaces (#2992, ADR-0096 D4) + +Two latent execution surfaces — neither reachable by a client today — would +have fallen open the instant a real transport was wired, because both drop or +lack the caller's identity. Per ADR-0096, the identity story is fixed and +pinned in CI *before* wiring, not after an adversarial review: + +- **GraphQL (surface 1 — latent context-drop, now threaded).** + `handleGraphQL` passed only `{ request }` to `kernel.graphql`, dropping the + resolved `ExecutionContext` — the moment a real engine resolved objects + through ObjectQL it would have run context-less (security middleware falls + OPEN on a missing principal = full authority). The entry point now resolves + the caller identity even on the direct dispatcher-plugin route and even when + `requireAuth` is off, and threads it as `options.context`; + `IGraphQLService.execute` documents that implementations MUST forward it to + every data-engine call. Unit-proven; the authz conformance matrix pins the + threading (`graphql-identity-thread` row) so removing it goes STALE and + fails CI. + +- **realtime (surface 2 — no per-recipient authz seam, posture registered).** + Delivery is a pure fan-out (subscriptions carry no principal, + `matchesSubscription` filters only by object+eventTypes, the engine + publishes the full `after` row), safe only while every subscriber is + server-internal. The posture is now registered as an `experimental` matrix + row (`realtime-delivery-authz`) stating the admission requirement + (per-recipient RLS/FLS/tenant re-check on delivery, or id-only payload + + client re-fetch), and transport TRIPWIRE probes turn any newly wired + WebSocket/SSE/subscribe/client transport into an UNCLASSIFIED surface → red + CI until the identity story ships with it. The `service-realtime` README — + which advertised `authorizeChannel`/`broadcastToUser`/presence auth that do + not exist — is rewritten to describe the real, trusted-internal-only + surface, and the contract docs carry the admission requirement at the seam. diff --git a/content/docs/protocol/kernel/realtime-protocol.mdx b/content/docs/protocol/kernel/realtime-protocol.mdx index 1b9d3d6563..aeafc79861 100644 --- a/content/docs/protocol/kernel/realtime-protocol.mdx +++ b/content/docs/protocol/kernel/realtime-protocol.mdx @@ -11,6 +11,8 @@ The **Real-Time Protocol** describes how live data synchronization is intended t **Implementation status (v1):** The shipping realtime service is an **in-memory pub/sub adapter** (`@objectstack/service-realtime`, `InMemoryRealtimeAdapter`) plus a **long-polling** client (`RealtimeAPI` in `@objectstack/client`). The `IRealtimeService` contract reserves an optional `handleUpgrade()` for a WebSocket handshake, but **no WebSocket (`/ws`) or SSE (`/api/v1/stream`) transport is wired up yet** — those sections below document the planned wire protocol, not a deployed endpoint. The in-memory adapter is **single-instance only** (v1 deployment contract); a Redis-backed adapter for multi-node HA is a post-GA fast-follow. Treat the WebSocket/SSE message formats, connection limits, and debug endpoints in this page as a forward-looking design spec until that transport lands. + + **Identity admission (framework#2992, ADR-0096 D4):** today's delivery path is a trusted server-internal fan-out with **no per-recipient authorization** — subscriptions carry no principal and events carry the full record body. Before any client transport ships, delivery must re-check each subscriber's authority (RLS/FLS/tenant) per event — the subscribe-time permission check shown below is *not* sufficient — or switch to id-only payloads with client re-fetch. The authz conformance matrix (`realtime-delivery-authz` row + transport tripwires) enforces this in CI. ## Why Real-Time Matters diff --git a/packages/dogfood/test/authz-conformance.matrix.ts b/packages/dogfood/test/authz-conformance.matrix.ts index d12b0a4dfb..a0897715da 100644 --- a/packages/dogfood/test/authz-conformance.matrix.ts +++ b/packages/dogfood/test/authz-conformance.matrix.ts @@ -75,6 +75,20 @@ export const AUTHZ_CONFORMANCE: AuthzPrimitive[] = [ proof: 'showcase-anonymous-deny-surfaces.dogfood.test.ts', covers: ['data:hono-plugin.ts:POST /data/:object', 'data:hono-plugin.ts:GET /data/:object/:id', 'data:hono-plugin.ts:GET /data/:object'], note: 'These routes delegate straight to ObjectQL and were only shadowed when the REST plugin registered the same paths FIRST — so the posture depended on plugin registration order (a load-order change silently reopened it, no test failing). Gating each route makes the deny decision a property of this entry point too. Handler-level proof in plugin-hono-server/hono-anonymous-deny.test.ts.' }, + + // ── #2992 / ADR-0096 D4 — latent execution surfaces (pre-wiring identity + // admission). Neither surface is reachable by a client today; these rows + // register their identity posture NOW so the ratchet (see the probes + + // transport tripwires in authz-conformance.test.ts) blocks wiring a client + // transport without the identity story — in CI, not in an adversarial + // review after the fact. + { id: 'graphql-identity-thread', summary: 'GraphQL entry point threads the caller identity to the engine (#2992 surface 1, ADR-0096 D1)', state: 'enforced', + enforcement: 'runtime/http-dispatcher.ts handleGraphQL — resolves the caller ExecutionContext (also on the direct dispatcher-plugin route, requireAuth on or off) and threads it as options.context on every kernel.graphql call; spec IGraphQLService.execute documents that implementations MUST forward it to ObjectQL as options.context', + covers: ['graphql:http-dispatcher.ts:kernel.graphql(context-threaded)'], + note: 'Surface posture: user (caller identity), latent — kernel.graphql is never assigned in the monorepo, so every POST /graphql 501s before an engine call; the only IGraphQLService is the plugin-dev stub. The threading exists so the FIRST real engine runs caller-scoped instead of context-less (the security middleware falls OPEN on a missing principal = full authority). Threading unit-proven in runtime/http-dispatcher.requireauth.test.ts (identity threading block); removing it goes STALE here and fails CI.' }, + { id: 'realtime-delivery-authz', summary: 'realtime delivery fan-out has NO per-recipient authorization — trusted server-internal subscribers only (#2992 surface 2)', state: 'experimental', + covers: ['realtime:in-memory-realtime-adapter.ts:publish(trusted-fan-out)'], + note: 'Surface posture: system (trusted-implicit), pre-wiring — no end-user transport exists (handleUpgrade unimplemented, no REST subscribe route, client RealtimeAPI is a placeholder); the only subscribers are server-internal plugins (webhook auto-enqueuer, knowledge sync). Structural defect: Subscription carries no principal, matchesSubscription filters only by object+eventTypes (RealtimeSubscriptionOptions.filter is declared but never read), and the engine publishes the FULL after-row — so any future external subscriber would receive record bodies cross-tenant that its own find would hide. ADMISSION REQUIREMENT before any WebSocket/SSE/subscribe transport ships: per-recipient RLS/FLS/tenant re-check on delivery (subscription carries the subscriber ExecutionContext) OR id-only payload + client re-fetch. The transport tripwire probes in authz-conformance.test.ts turn a wired transport into an UNCLASSIFIED surface → red CI until this row is upgraded with the enforcement site.' }, { id: 'default-profile', summary: 'app-declared default profile (isDefault)', state: 'enforced', enforcement: 'plugin-security/security-plugin.ts fallback resolution', proof: 'showcase-default-profile.dogfood.test.ts' }, diff --git a/packages/dogfood/test/authz-conformance.test.ts b/packages/dogfood/test/authz-conformance.test.ts index d79545b15f..ce9aad5a89 100644 --- a/packages/dogfood/test/authz-conformance.test.ts +++ b/packages/dogfood/test/authz-conformance.test.ts @@ -60,6 +60,60 @@ const PROBES: ReadonlyArray<{ file: string; re: RegExp; key: (m: RegExpExecArray re: /rawApp\.(get|post|put|patch|delete)\(\s*`\$\{prefix\}(\/data[^`]*)`/g, key: (m) => `data:hono-plugin.ts:${m[1].toUpperCase()} ${m[2]}`, }, + + // ── #2992 / ADR-0096 D4 — latent-surface identity pins ───────────────── + // GraphQL identity threading: the ONLY kernel.graphql(...) call site must + // carry `context:` in its options. If a refactor drops the threading the key + // vanishes → the `graphql-identity-thread` row's covers goes STALE → red CI. + { + file: 'packages/runtime/src/http-dispatcher.ts', + re: /kernel\.graphql\([^)]*\bcontext:/g, + key: () => 'graphql:http-dispatcher.ts:kernel.graphql(context-threaded)', + }, + // Realtime delivery fan-out: pins the trusted-internal-only posture of the + // in-memory adapter's publish loop (`realtime-delivery-authz` row). + { + file: 'packages/services/service-realtime/src/in-memory-realtime-adapter.ts', + re: /async\s+publish\s*\(/g, + key: () => 'realtime:in-memory-realtime-adapter.ts:publish(trusted-fan-out)', + }, + + // ── #2992 transport TRIPWIRES — deliberately covered by NO row ────────── + // Delivery today is a pure fan-out with no per-recipient authorization + // (subscriptions carry no principal, payload is the full record), which is + // safe ONLY while every subscriber is server-internal. These patterns match + // nothing today; the moment someone wires an end-user realtime transport + // (WebSocket handshake, SSE, a client transport) a NEW key appears → + // UNCLASSIFIED surface → red CI with this checklist: add per-recipient + // RLS/FLS/tenant re-check on delivery (or switch to id-only payloads), + // THEN register the enforcement site in a matrix row covering the new key. + { + file: 'packages/services/service-realtime/src/in-memory-realtime-adapter.ts', + re: /handleUpgrade\s*\(/g, + key: () => 'realtime:in-memory-realtime-adapter.ts:handleUpgrade(TRANSPORT-WIRED)', + }, + { + file: 'packages/services/service-realtime/src/realtime-service-plugin.ts', + re: /handleUpgrade\s*\(|new\s+WebSocketServer|text\/event-stream/g, + key: () => 'realtime:realtime-service-plugin.ts:transport(TRANSPORT-WIRED)', + }, + { + file: 'packages/runtime/src/http-dispatcher.ts', + re: /async\s+handle(Realtime|Upgrade|Subscribe)\w*\s*\(/g, + key: (m) => `realtime:http-dispatcher.ts:handle${m[1]}(TRANSPORT-WIRED)`, + }, + { + file: 'packages/client/src/realtime-api.ts', + re: /new\s+WebSocket\b|new\s+EventSource\b/g, + key: () => 'realtime:client/realtime-api.ts:transport(TRANSPORT-WIRED)', + }, + // packages/rest/src has ZERO realtime refs today (#2992) — a `/realtime` + // route literal appearing there is a subscribe endpoint. Same tripwire. + { + file: 'packages/rest/src/rest-server.ts', + re: /['"`][^'"`]*\/realtime[^'"`]*['"`]/g, + key: () => 'realtime:rest-server.ts:route(TRANSPORT-WIRED)', + }, ]; /** Statically enumerate the anonymous-deny HTTP entry points from source. */ @@ -132,4 +186,25 @@ describe('#2567 — anonymous-deny surface ratchet bites', () => { const problems = checkLedger(m, opts(() => discoverAnonymousDenySurfaces())); expect(problems.some((p) => /STALE covers/.test(p) && /handleRemovedThing/.test(p))).toBe(true); }); + + // ── #2992 — the latent-surface pins bite too ────────────────────────── + it('(d) wiring a realtime transport (tripwire key appears) → UNCLASSIFIED surface failure (#2992)', () => { + const fake = 'realtime:in-memory-realtime-adapter.ts:handleUpgrade(TRANSPORT-WIRED)'; + const problems = checkLedger( + AUTHZ_CONFORMANCE, + opts(() => new Set([...discoverAnonymousDenySurfaces(), fake])), + ); + expect(problems.some((p) => p.includes('UNCLASSIFIED surface') && p.includes(fake))).toBe(true); + }); + + it('(e) dropping the GraphQL context-thread → STALE covers failure (#2992)', () => { + const threaded = 'graphql:http-dispatcher.ts:kernel.graphql(context-threaded)'; + // Baseline sanity: the threading is discovered from source today. + expect(discoverAnonymousDenySurfaces().has(threaded)).toBe(true); + const problems = checkLedger( + AUTHZ_CONFORMANCE, + opts(() => new Set([...discoverAnonymousDenySurfaces()].filter((k) => k !== threaded))), + ); + expect(problems.some((p) => /STALE covers/.test(p) && p.includes(threaded))).toBe(true); + }); }); diff --git a/packages/runtime/src/http-dispatcher.requireauth.test.ts b/packages/runtime/src/http-dispatcher.requireauth.test.ts index b8e29ab291..7ed14159c8 100644 --- a/packages/runtime/src/http-dispatcher.requireauth.test.ts +++ b/packages/runtime/src/http-dispatcher.requireauth.test.ts @@ -100,6 +100,62 @@ describe('HttpDispatcher requireAuth gate — GraphQL (handleGraphQL)', () => { }); }); +// #2992 / ADR-0096 D1 — the GraphQL entry point must THREAD the caller's +// identity to the engine, not just gate anonymity. kernel.graphql is a stub +// surface today (never assigned in the monorepo), but the moment a real +// engine lands it resolves objects through ObjectQL, whose security +// middleware falls OPEN on a missing principal — so the entry point passes +// the resolved ExecutionContext as `options.context` (the same key the REST +// callData path threads). Dropping it also goes STALE in the authz +// conformance matrix (dogfood/test/authz-conformance.test.ts). +describe('HttpDispatcher identity threading — GraphQL (handleGraphQL, #2992)', () => { + const gqlBody = { query: '{ __typename }', variables: { a: 1 } }; + + const makeGraphQLKernel = (calls: any[]) => + makeKernel({ + graphql: (query: string, variables: any, options: any) => { + calls.push({ query, variables, options }); + return { data: {} }; + }, + }); + + it('threads the resolved ExecutionContext as options.context', async () => { + const calls: any[] = []; + const d = new HttpDispatcher(makeGraphQLKernel(calls), undefined, { requireAuth: true }); + await d.handleGraphQL(gqlBody, authed); + expect(calls).toHaveLength(1); + expect(calls[0].query).toBe(gqlBody.query); + expect(calls[0].variables).toEqual(gqlBody.variables); + expect(calls[0].options.context).toBe(authed.executionContext); + }); + + it('threads a system context unchanged', async () => { + const calls: any[] = []; + const d = new HttpDispatcher(makeGraphQLKernel(calls), undefined, { requireAuth: true }); + await d.handleGraphQL(gqlBody, system); + expect(calls[0].options.context).toBe(system.executionContext); + }); + + it('threads the caller identity even when requireAuth is OFF (an authenticated caller on an open deployment still runs under their own authority)', async () => { + const calls: any[] = []; + const d = new HttpDispatcher(makeGraphQLKernel(calls), undefined, { requireAuth: false }); + await d.handleGraphQL(gqlBody, authed); + expect(calls[0].options.context).toBe(authed.executionContext); + }); + + it('an anonymous caller on an open deployment carries NO authority (explicit guest principal or nothing — never a forged user/system identity)', async () => { + const calls: any[] = []; + const d = new HttpDispatcher(makeGraphQLKernel(calls), undefined, { requireAuth: false }); + // Fresh context object: handleGraphQL caches the resolved identity on it. + await d.handleGraphQL(gqlBody, { request: {}, executionContext: undefined } as any); + const threaded = calls[0].options.context; + // The resolver yields an explicit guest principal (mirroring dispatch()); + // whatever is threaded must carry no user and no system authority. + expect(threaded?.userId).toBeUndefined(); + expect(threaded?.isSystem ?? false).toBe(false); + }); +}); + describe('HttpDispatcher requireAuth gate — metadata catch-all (handleMetadata)', () => { it('401s an anonymous caller when requireAuth is on', async () => { const d = new HttpDispatcher(makeKernel(), undefined, { requireAuth: true }); diff --git a/packages/runtime/src/http-dispatcher.ts b/packages/runtime/src/http-dispatcher.ts index 2b683418c2..6668cfcfe3 100644 --- a/packages/runtime/src/http-dispatcher.ts +++ b/packages/runtime/src/http-dispatcher.ts @@ -1631,9 +1631,13 @@ export class HttpDispatcher { // // The dispatcher-plugin's direct `/graphql` route calls us WITHOUT // resolving identity first (unlike `dispatch()`, which populates - // `context.executionContext`), so resolve it here when absent. + // `context.executionContext`), so resolve it here when absent — + // REGARDLESS of `requireAuth`: the resolved identity is also what we + // thread to the engine below (#2992), and an authenticated caller on a + // `requireAuth: false` deployment must still run under their own + // authority, not context-less. let ec: any = context.executionContext; - if (this.requireAuth && !ec) { + if (!ec) { ec = await this.resolveRequestExecutionContext(context); if (ec) context.executionContext = ec; } @@ -1652,8 +1656,16 @@ export class HttpDispatcher { throw { statusCode: 501, message: 'GraphQL service not available' }; } + // ADR-0096 D1 / #2992 — thread the caller's identity to the engine. + // `kernel.graphql` is still unassigned everywhere (this call 501s + // above), but the moment a real engine lands it resolves objects + // through ObjectQL, whose security middleware falls OPEN on a missing + // principal — so the entry point must already carry the caller as + // `options.context` (the same key the REST `callData` path threads). + // An implementation MUST forward it to every data-engine call. return this.kernel.graphql(body.query, body.variables, { - request: context.request + request: context.request, + context: ec, }); } @@ -1663,8 +1675,10 @@ export class HttpDispatcher { * `context.executionContext`). The dispatcher-plugin's direct `/graphql` * route is the current caller. Mirrors the identity resolution `dispatch()` * performs so an anonymous-deny gate can tell an authenticated caller from - * an anonymous one. Best-effort: returns `undefined` on failure (treated as - * anonymous, i.e. denied under `requireAuth`). + * an anonymous one, and so the caller's identity can be THREADED to the + * engine (#2992 / ADR-0096 D1) instead of dropped. Best-effort: returns + * `undefined` on failure (treated as anonymous, i.e. denied under + * `requireAuth`). */ private async resolveRequestExecutionContext( context: HttpProtocolContext, diff --git a/packages/services/service-realtime/README.md b/packages/services/service-realtime/README.md index 4359795549..5c612b45c8 100644 --- a/packages/services/service-realtime/README.md +++ b/packages/services/service-realtime/README.md @@ -1,428 +1,104 @@ # @objectstack/service-realtime -Realtime Service for ObjectStack — implements `IRealtimeService` with WebSocket and in-memory pub/sub. +In-process pub/sub for ObjectStack — the production `IRealtimeService` implementation, +backed by an in-memory adapter. -## Features +> ⚠️ **Server-internal only — no client transport.** This service delivers events to +> **trusted, in-process subscribers** (today: the webhook auto-enqueuer and knowledge +> sync). There is **no WebSocket/SSE endpoint, no REST subscribe route, and no working +> client transport** — `IRealtimeService.handleUpgrade` is deliberately unimplemented +> platform-wide, and the `@objectstack/client` `RealtimeAPI` is a placeholder. +> See [Security posture](#security-posture-2992--adr-0096-d4) before changing that. -- **WebSocket Support**: Real-time bidirectional communication -- **Pub/Sub Pattern**: Subscribe to channels and receive updates -- **Room-Based Architecture**: Organize connections into rooms -- **Presence Tracking**: Track online users and their status -- **Message Broadcasting**: Send messages to all connections or specific rooms -- **Event Streaming**: Stream database changes and system events -- **Auto-Reconnection**: Client auto-reconnects on connection loss -- **Type-Safe**: Full TypeScript support for events and messages +## What this package actually provides -## Installation +- **`InMemoryRealtimeAdapter`** — a Map-backed pub/sub implementing `IRealtimeService` + (`publish` / `subscribe` / `unsubscribe`), with: + - channel-based routing and per-subscription filtering by **object name** and + **event types** (`RealtimeSubscriptionOptions.object` / `eventTypes`; + `options.filter` is declared in the contract but **not evaluated**); + - a `maxSubscriptions` safety cap (default 50 000; `0` = unbounded, tests only) so a + subscription leak can't grow the map until the pod OOMs; + - handler errors swallowed per-delivery so one bad subscriber can't break the + publish loop. +- **`RealtimeServicePlugin`** — registers the adapter as the kernel `realtime` service, + registers the `sys_presence` system object, and contributes its translations. -```bash -pnpm add @objectstack/service-realtime -``` - -## Basic Usage - -```typescript -import { defineStack } from '@objectstack/spec'; -import { ServiceRealtime } from '@objectstack/service-realtime'; - -const stack = defineStack({ - services: [ - ServiceRealtime.configure({ - port: 3001, - path: '/ws', - }), - ], -}); -``` - -## Configuration - -```typescript -interface RealtimeServiceConfig { - /** WebSocket server port (default: 3001) */ - port?: number; - - /** WebSocket path (default: '/ws') */ - path?: string; - - /** Enable CORS (default: true) */ - cors?: boolean; - - /** Maximum connections per user (default: 10) */ - maxConnectionsPerUser?: number; - - /** Ping interval in ms (default: 30000) */ - pingInterval?: number; -} -``` - -## Service API (Server-Side) - -```typescript -// Get realtime service -const realtime = kernel.getService('realtime'); -``` - -### Broadcasting - -```typescript -// Broadcast to all connected clients -await realtime.broadcast({ - event: 'notification', - data: { message: 'System update in 5 minutes' }, -}); - -// Broadcast to specific room -await realtime.broadcastToRoom('opportunity:123', { - event: 'record_updated', - data: { recordId: '123', field: 'stage', value: 'closed_won' }, -}); - -// Broadcast to specific user -await realtime.broadcastToUser('user:456', { - event: 'mention', - data: { commentId: 'comment:789', mentionedBy: 'user:123' }, -}); -``` - -### Channel Management - -```typescript -// Join a channel (room) -await realtime.join(connectionId, 'opportunity:123'); - -// Leave a channel -await realtime.leave(connectionId, 'opportunity:123'); - -// Get all connections in a channel -const connections = await realtime.getChannelConnections('opportunity:123'); - -// Get all channels for a connection -const channels = await realtime.getConnectionChannels(connectionId); -``` - -### Presence - -```typescript -// Set user presence -await realtime.setPresence('user:456', { - status: 'online', - currentPage: '/opportunity/123', - lastActive: new Date(), -}); - -// Get user presence -const presence = await realtime.getPresence('user:456'); - -// Get all online users -const onlineUsers = await realtime.getOnlineUsers(); - -// Get users in a specific channel -const channelUsers = await realtime.getChannelPresence('opportunity:123'); -``` - -## Client-Side Usage - -### React Hook - -```typescript -import { useRealtime } from '@objectstack/client-react'; - -function OpportunityDetails({ id }: { id: string }) { - const { subscribe, send, isConnected } = useRealtime(); - - useEffect(() => { - // Subscribe to record updates - const unsubscribe = subscribe(`opportunity:${id}`, (event) => { - if (event.type === 'record_updated') { - console.log('Record updated:', event.data); - // Update UI - } - }); - - return unsubscribe; - }, [id]); - - return ( -
- {isConnected ? '🟢 Connected' : '🔴 Disconnected'} -
- ); -} -``` - -### JavaScript Client - -```typescript -import { RealtimeClient } from '@objectstack/client'; - -const client = new RealtimeClient({ - url: 'ws://localhost:3001/ws', - auth: { - token: 'your-auth-token', - }, -}); +**v1 deployment contract (launch-readiness P0-5): single-instance only.** The adapter is +process-local — events published on node A are not delivered to subscribers on node B. +An HA adapter (Redis-backed, over `service-cluster-redis`) is a post-GA fast-follow. -// Connect -await client.connect(); - -// Subscribe to a channel -client.subscribe('opportunity:123', (event) => { - console.log('Received event:', event); -}); - -// Send a message -client.send('typing', { - recordId: '123', - userId: 'user:456', - isTyping: true, -}); - -// Disconnect -await client.disconnect(); -``` - -## Advanced Features - -### Event Streaming - -Stream database changes in real-time: +## Usage (server-side, trusted code only) ```typescript -// Server-side: Stream record changes -realtime.streamRecordChanges('opportunity', { - onInsert: async (record) => { - await realtime.broadcast({ - event: 'record_created', - data: { object: 'opportunity', record }, - }); - }, - onUpdate: async (record, changes) => { - await realtime.broadcastToRoom(`opportunity:${record.id}`, { - event: 'record_updated', - data: { recordId: record.id, changes }, - }); - }, - onDelete: async (recordId) => { - await realtime.broadcast({ - event: 'record_deleted', - data: { object: 'opportunity', recordId }, - }); - }, -}); -``` - -### Private Channels - -```typescript -// Server-side: Authorize private channel access -realtime.authorizeChannel = async (userId, channel) => { - if (channel.startsWith('user:')) { - // Only allow users to join their own private channel - return channel === `user:${userId}`; - } - - if (channel.startsWith('opportunity:')) { - // Check if user has access to the opportunity - const opportunityId = channel.split(':')[1]; - return await hasAccess(userId, 'opportunity', opportunityId); - } +import { ObjectKernel } from '@objectstack/core'; +import { RealtimeServicePlugin } from '@objectstack/service-realtime'; - return false; -}; -``` +const kernel = new ObjectKernel(); +kernel.use(new RealtimeServicePlugin()); +await kernel.bootstrap(); -### Typing Indicators +const realtime = kernel.getService('realtime'); -```typescript -// Client sends typing event -client.send('typing', { - recordId: '123', - userId: 'user:456', - isTyping: true, -}); +const subId = await realtime.subscribe('records', (event) => { + console.log(event.type, event.payload); +}, { object: 'account', eventTypes: ['record.created'] }); -// Server broadcasts to room -realtime.on('typing', async (connectionId, data) => { - await realtime.broadcastToRoom(`opportunity:${data.recordId}`, { - event: 'user_typing', - data: { userId: data.userId, isTyping: data.isTyping }, - }, { exclude: [connectionId] }); // Don't send back to sender +await realtime.publish({ + type: 'record.created', + object: 'account', + payload: { id: 'acc-1', name: 'Acme' }, + timestamp: new Date().toISOString(), }); -// Other clients receive typing notification -client.subscribe('opportunity:123', (event) => { - if (event.type === 'user_typing') { - showTypingIndicator(event.data.userId, event.data.isTyping); - } -}); +await realtime.unsubscribe(subId); ``` -### Live Cursor Tracking +Configuration: ```typescript -// Client sends cursor position -client.send('cursor', { - recordId: '123', - x: 450, - y: 200, -}); - -// Server broadcasts to room -realtime.on('cursor', async (connectionId, data) => { - const user = await getConnectionUser(connectionId); - - await realtime.broadcastToRoom(`opportunity:${data.recordId}`, { - event: 'cursor_moved', - data: { - userId: user.id, - userName: user.name, - x: data.x, - y: data.y, - }, - }, { exclude: [connectionId] }); +new RealtimeServicePlugin({ + adapter: 'memory', // only supported adapter today + memory: { maxSubscriptions: 50_000 }, // 0 = unbounded (tests only) }); ``` -### Collaborative Editing +## Security posture (#2992 / ADR-0096 D4) -```typescript -// Operational Transform (OT) for collaborative editing -client.send('edit', { - documentId: '123', - operation: { - type: 'insert', - position: 42, - text: 'Hello', - }, -}); - -realtime.on('edit', async (connectionId, data) => { - // Apply operation transform - const transformedOp = await applyOT(data.operation); +Delivery is a **pure fan-out with no per-recipient authorization seam**: - // Broadcast to all editors - await realtime.broadcastToRoom(`document:${data.documentId}`, { - event: 'operation', - data: transformedOp, - }, { exclude: [connectionId] }); -}); -``` +- subscriptions carry **no principal** — there is nothing to check a row against; +- `matchesSubscription` filters only by object name + event type; +- the ObjectQL engine publishes `record.created` / `record.updated` events with the + **full record body** (the `after` row) — rows and fields a subscriber's own `find` + would hide under RLS/FLS/tenant scoping. -## Integration with ObjectStack Features +That is safe **only while every subscriber is trusted server-internal code**. Before any +end-user transport ships (WebSocket `handleUpgrade`, SSE, a REST subscribe route, or a +real client transport), the delivery path MUST gain one of: -### Feed Updates +1. **a per-recipient re-check on delivery** — the subscription carries the subscriber's + `ExecutionContext`, and every event is re-authorized (RLS/FLS/tenant) against it + before the handler fires; **or** +2. **id-only payloads** — the client re-fetches the record under its own authority. -```typescript -// When a comment is added -feed.on('comment_added', async (comment) => { - await realtime.broadcastToRoom(`${comment.object}:${comment.recordId}`, { - event: 'feed_update', - data: { type: 'comment', comment }, - }); -}); -``` - -### Workflow Status - -```typescript -// When a flow step completes -automation.on('step_completed', async (execution) => { - await realtime.broadcastToUser(execution.userId, { - event: 'flow_progress', - data: { - flowId: execution.flowId, - step: execution.currentStep, - progress: execution.progress, - }, - }); -}); -``` - -### Analytics Dashboard - -```typescript -// Stream real-time metrics -setInterval(async () => { - const metrics = await analytics.getCurrentMetrics(); - - await realtime.broadcastToRoom('dashboard:sales', { - event: 'metrics_update', - data: metrics, - }); -}, 5000); // Every 5 seconds -``` - -## Connection Events - -```typescript -// Server-side event handlers -realtime.on('connection', async (connectionId, userId) => { - console.log(`User ${userId} connected (${connectionId})`); - - // Set initial presence - await realtime.setPresence(userId, { status: 'online' }); -}); - -realtime.on('disconnection', async (connectionId, userId) => { - console.log(`User ${userId} disconnected`); - - // Update presence - await realtime.setPresence(userId, { - status: 'offline', - lastSeen: new Date(), - }); -}); - -realtime.on('error', async (connectionId, error) => { - console.error(`Connection error:`, error); -}); -``` - -## Best Practices - -1. **Channel Organization**: Use namespaced channels (e.g., `object:recordId`) -2. **Authorization**: Always verify channel access before joining -3. **Message Size**: Keep messages small (< 10KB) -4. **Rate Limiting**: Limit message frequency per connection -5. **Cleanup**: Remove disconnected users from channels -6. **Heartbeat**: Implement ping/pong for connection health -7. **Compression**: Enable WebSocket compression for large messages - -## Performance Considerations - -- **Scaling**: Use Redis adapter for multi-server deployments -- **Connection Pooling**: Limit concurrent connections per user -- **Channel Limits**: Limit channels per connection -- **Message Batching**: Batch frequent updates to reduce traffic -- **Binary Protocol**: Use binary for large data transfers - -## REST API Endpoints - -``` -POST /api/v1/realtime/broadcast # Broadcast to all -POST /api/v1/realtime/broadcast/room/:room # Broadcast to room -POST /api/v1/realtime/broadcast/user/:userId # Broadcast to user -GET /api/v1/realtime/presence # Get online users -GET /api/v1/realtime/presence/:userId # Get user presence -GET /api/v1/realtime/channels/:channel # Get channel connections -``` +This posture is registered in the authz conformance matrix +(`packages/dogfood/test/authz-conformance.matrix.ts`, row `realtime-delivery-authz`), +and transport **tripwire probes** in `authz-conformance.test.ts` fail CI if a transport +is wired without upgrading that row with a real enforcement site. -## Contract Implementation +## Contract Implements `IRealtimeService` from `@objectstack/spec/contracts`: ```typescript interface IRealtimeService { - broadcast(message: Message): Promise; - broadcastToRoom(room: string, message: Message): Promise; - broadcastToUser(userId: string, message: Message): Promise; - join(connectionId: string, channel: string): Promise; - leave(connectionId: string, channel: string): Promise; - setPresence(userId: string, presence: PresenceData): Promise; - getPresence(userId: string): Promise; - getOnlineUsers(): Promise; - on(event: string, handler: EventHandler): void; + publish(event: RealtimeEventPayload): Promise; + subscribe(channel: string, handler: RealtimeEventHandler, options?: RealtimeSubscriptionOptions): Promise; + unsubscribe(subscriptionId: string): Promise; + handleUpgrade?(request: Request): Promise; // deliberately unimplemented — see above + subscribeMetadata?(filter, handler): Promise; // optional convenience — not implemented here + subscribeData?(filter, handler): Promise; // optional convenience — not implemented here } ``` @@ -432,7 +108,6 @@ Apache-2.0. See [LICENSING.md](../../../LICENSING.md). ## See Also -- [WebSocket API Documentation](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API) -- [@objectstack/client](../../client/) -- [@objectstack/client-react](../../client-react/) -- [Realtime Features Guide](/content/docs/protocol/kernel/realtime-protocol.mdx) +- [@objectstack/spec — realtime contract](../../spec/src/contracts/realtime-service.ts) +- framework#2992 — the identity-admission tracking issue for this surface +- [ADR-0096 — Execution-Surface Identity Admission](../../../docs/adr/0096-execution-surface-identity-admission.md) diff --git a/packages/services/service-realtime/src/in-memory-realtime-adapter.ts b/packages/services/service-realtime/src/in-memory-realtime-adapter.ts index 8ad6f00409..02c22cf22c 100644 --- a/packages/services/service-realtime/src/in-memory-realtime-adapter.ts +++ b/packages/services/service-realtime/src/in-memory-realtime-adapter.ts @@ -76,7 +76,15 @@ export class InMemoryRealtimeAdapter implements IRealtimeService { } async publish(event: RealtimeEventPayload): Promise { - // Deliver to all channel subscriptions that match filters + // Deliver to all channel subscriptions that match filters. + // + // ⚠️ TRUSTED fan-out (#2992 / ADR-0096 D4): there is NO per-recipient + // authorization here — `matchesSubscription` filters by object/event type + // only, `Subscription` carries no principal, and the engine publishes the + // full record body. Safe ONLY while every subscriber is server-internal. + // A client transport must not be wired to this path until delivery + // re-checks each recipient's authority (or payloads become id-only); the + // authz conformance matrix (`realtime-delivery-authz`) pins this. for (const sub of this.subscriptions.values()) { if (this.matchesSubscription(event, sub)) { try { diff --git a/packages/spec/src/contracts/graphql-service.ts b/packages/spec/src/contracts/graphql-service.ts index 0306c30bde..1cb8e1dcd5 100644 --- a/packages/spec/src/contracts/graphql-service.ts +++ b/packages/spec/src/contracts/graphql-service.ts @@ -43,8 +43,18 @@ export interface GraphQLResponse { export interface IGraphQLService { /** * Execute a GraphQL query or mutation + * + * ⚠️ Identity admission (ADR-0096 D1, #2992): `context` carries the + * caller's resolved `ExecutionContext`. An implementation that resolves + * objects through the data engine (ObjectQL) MUST forward it on every + * engine call as `options.context` — the security middleware falls OPEN + * on a missing principal, so executing resolvers context-less silently + * grants full authority (no RLS/FLS/CRUD/tenant scoping). The dispatcher's + * `/graphql` entry point threads the caller identity for exactly this + * purpose; dropping it here is a defect, never an authorization. + * * @param request - The GraphQL request - * @param context - Optional execution context (e.g. auth user) + * @param context - The caller's execution context (auth user / principal) * @returns GraphQL response with data and/or errors */ execute(request: GraphQLRequest, context?: Record): Promise; diff --git a/packages/spec/src/contracts/realtime-service.ts b/packages/spec/src/contracts/realtime-service.ts index 3705eb8efd..ea8032ade4 100644 --- a/packages/spec/src/contracts/realtime-service.ts +++ b/packages/spec/src/contracts/realtime-service.ts @@ -40,7 +40,14 @@ export interface RealtimeSubscriptionOptions { object?: string; /** Event types to listen for */ eventTypes?: string[]; - /** Additional filter conditions */ + /** + * Additional filter conditions. + * + * ⚠️ EXPERIMENTAL — declared but NOT evaluated by the in-memory adapter + * (`matchesSubscription` reads only `object` + `eventTypes`). Do not rely + * on it to narrow delivery, and NEVER as an authorization mechanism + * (see the identity-admission note on {@link IRealtimeService}). + */ filter?: Record; } @@ -60,6 +67,31 @@ export interface RealtimeSubscriptionFilter { fields?: string[]; } +/** + * ⚠️ Identity admission — READ BEFORE WIRING A CLIENT TRANSPORT (#2992, + * ADR-0096 D4). + * + * This contract currently serves TRUSTED, SERVER-INTERNAL subscribers only + * (webhook auto-enqueuer, knowledge sync). Delivery is a pure fan-out with + * **no per-recipient authorization seam**: subscriptions carry no principal, + * `matchesSubscription` filters only by object name + event type, and the + * engine publishes the FULL record body (`after` row) — rows and fields a + * subscriber's own `find` would hide under RLS/FLS/tenant scoping. + * + * Before ANY end-user transport ships (`handleUpgrade` WebSocket, SSE, a REST + * subscribe route, or a real client in `@objectstack/client`), the delivery + * path MUST gain one of: + * 1. a per-recipient re-check on delivery — the subscription carries the + * subscriber's `ExecutionContext` and every event is re-authorized + * (RLS/FLS/tenant) against it before the handler fires; or + * 2. id-only payloads — the client re-fetches the record under its own + * authority. + * + * Wiring a transport without this is a fall-open (full-authority broadcast, + * cross-tenant). The authz conformance matrix pins this posture + * (`realtime-delivery-authz` row + transport tripwire probes in + * `dogfood/test/authz-conformance.test.ts`) so CI blocks it, not review. + */ export interface IRealtimeService { /** * Publish an event to all subscribers @@ -84,6 +116,11 @@ export interface IRealtimeService { /** * Handle an incoming HTTP upgrade request (WebSocket handshake) + * + * ⚠️ Deliberately UNIMPLEMENTED platform-wide: implementing this hands + * external clients the unauthorized fan-out described on the interface — + * satisfy the identity-admission requirement above first (#2992). + * * @param request - Standard Request object * @returns Standard Response object */