From 511a90356dea555048d7b56c84d284e9d1692ee1 Mon Sep 17 00:00:00 2001 From: ayush-that Date: Fri, 17 Jul 2026 21:06:52 +0530 Subject: [PATCH 1/2] allow channel-token ws subscriptions within their own namespace --- apps/gateway/src/ws-handler.ts | 15 ++- apps/gateway/test/ws-subscribe-scope.test.ts | 122 +++++++++++++++++++ 2 files changed, 135 insertions(+), 2 deletions(-) create mode 100644 apps/gateway/test/ws-subscribe-scope.test.ts diff --git a/apps/gateway/src/ws-handler.ts b/apps/gateway/src/ws-handler.ts index 07c71abe..6a35c434 100644 --- a/apps/gateway/src/ws-handler.ts +++ b/apps/gateway/src/ws-handler.ts @@ -279,8 +279,19 @@ const handleRequest = async ( sendError(ws, id, 'INVALID_PARAMS', 'Missing sessionId.'); return; } - // Verify caller is a participant before subscribing to events - await requireSessionAccess(conn, runtime, callerUserId, sessionId); + // Channel-token connections may subscribe to sessions inside their + // own namespace (mirrors the SSE events route, which enforces the + // namespace instead of participant membership — a channel bridge has + // no single channelUserId across its conversations). User JWTs keep + // the participant check. + const channelScoped = + conn.auth?.mode === 'channel' && + !!conn.auth.channelNamespace && + sessionId.startsWith(`${conn.auth.channelNamespace}:`); + if (!channelScoped) { + // Verify caller is a participant before subscribing to events + await requireSessionAccess(conn, runtime, callerUserId, sessionId); + } conn.subscriptions.get(sessionId)?.(); const afterEventId = typeof p.lastEventId === 'number' ? p.lastEventId : 0; const unsubscribe = runtime.events.subscribeFrom( diff --git a/apps/gateway/test/ws-subscribe-scope.test.ts b/apps/gateway/test/ws-subscribe-scope.test.ts new file mode 100644 index 00000000..d282d951 --- /dev/null +++ b/apps/gateway/test/ws-subscribe-scope.test.ts @@ -0,0 +1,122 @@ +import assert from 'node:assert/strict'; +import { createServer, type Server } from 'node:http'; +import { once } from 'node:events'; +import { test } from 'node:test'; + +import { WebSocket } from 'ws'; + +import { ChannelRegistry, createJwtConfig, type AuthResolverOptions } from '../src/auth.js'; +import { attachGatewayWs } from '../src/ws-handler.js'; +import type { AgentInstanceManager } from '../src/agent-instance.js'; + +const CHANNEL_KEY = 'test-channel-key'; + +const makeRuntimeStub = () => { + const subscribed: string[] = []; + return { + subscribed, + runtime: { + events: { + subscribeFrom: (sessionId: string) => { + subscribed.push(sessionId); + return () => {}; + }, + }, + // Channel connections carry no channelUserId, so the caller never + // resolves to a userId; mirror that here. + resolveCallerUserId: async () => undefined, + verifySessionAccess: async () => { + throw new Error('verifySessionAccess should not be reached in these tests'); + }, + } as never, + }; +}; + +const makeInstancesStub = (runtime: never): AgentInstanceManager => + ({ + getOrHydrate: async () => runtime, + wsConnect: () => {}, + wsDisconnect: () => {}, + touch: () => {}, + }) as unknown as AgentInstanceManager; + +const makeAuthOptions = (): AuthResolverOptions => { + const channels = new ChannelRegistry(); + channels.register({ + channelId: 'chan-1', + apiKey: CHANNEL_KEY, + namespace: 'amiko', + agentId: 'agent-a', + }); + return { + userProviders: [], + channels, + jwt: createJwtConfig('test-secret'), + }; +}; + +const listen = async (server: Server): Promise => { + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const address = server.address(); + assert.ok(address && typeof address === 'object'); + return address.port; +}; + +interface WsResponse { + kind: 'response'; + id: string; + result?: { subscribed?: boolean }; + error?: { code: string; message: string }; +} + +const subscribeOnce = (port: number, sessionId: string): Promise => + new Promise((resolve, reject) => { + const ws = new WebSocket(`ws://127.0.0.1:${port}/api/agents/agent-a/ws`, { + headers: { authorization: `Bearer ${CHANNEL_KEY}` }, + }); + ws.on('open', () => { + ws.send( + JSON.stringify({ + kind: 'request', + id: 'req-1', + method: 'session.subscribe', + params: { sessionId }, + }), + ); + }); + ws.on('message', (data) => { + ws.close(); + resolve(JSON.parse(String(data)) as WsResponse); + }); + ws.on('error', reject); + setTimeout(() => reject(new Error('ws subscribe timed out')), 5000).unref(); + }); + +test('channel token subscribes to a session in its own namespace', async () => { + const { runtime, subscribed } = makeRuntimeStub(); + const server = createServer(); + attachGatewayWs(server, { instances: makeInstancesStub(runtime), auth: makeAuthOptions() }); + const port = await listen(server); + try { + const response = await subscribeOnce(port, 'amiko:conv-1'); + assert.equal(response.result?.subscribed, true); + assert.deepEqual(subscribed, ['amiko:conv-1']); + } finally { + server.close(); + } +}); + +test('channel token cannot subscribe outside its namespace', async () => { + const { runtime, subscribed } = makeRuntimeStub(); + const server = createServer(); + attachGatewayWs(server, { instances: makeInstancesStub(runtime), auth: makeAuthOptions() }); + const port = await listen(server); + try { + const response = await subscribeOnce(port, 'telegram:conv-9'); + assert.equal(response.error?.code, 'SESSION_NOT_FOUND'); + assert.deepEqual(subscribed, []); + } finally { + server.close(); + } +}); From ad4fab060329f7114de55ea9ee14cd5c9ff82027 Mon Sep 17 00:00:00 2001 From: ayush-that Date: Fri, 17 Jul 2026 21:21:33 +0530 Subject: [PATCH 2/2] await server shutdown in ws subscribe scope tests --- apps/gateway/test/ws-subscribe-scope.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/gateway/test/ws-subscribe-scope.test.ts b/apps/gateway/test/ws-subscribe-scope.test.ts index d282d951..71a9a08d 100644 --- a/apps/gateway/test/ws-subscribe-scope.test.ts +++ b/apps/gateway/test/ws-subscribe-scope.test.ts @@ -104,6 +104,7 @@ test('channel token subscribes to a session in its own namespace', async () => { assert.deepEqual(subscribed, ['amiko:conv-1']); } finally { server.close(); + await once(server, 'close'); } }); @@ -118,5 +119,6 @@ test('channel token cannot subscribe outside its namespace', async () => { assert.deepEqual(subscribed, []); } finally { server.close(); + await once(server, 'close'); } });