diff --git a/src/modules/sse/hub.ts b/src/modules/sse/hub.ts index ea14ca6..ff8d840 100644 --- a/src/modules/sse/hub.ts +++ b/src/modules/sse/hub.ts @@ -27,6 +27,16 @@ export type SubscribeResult = const SUBSCRIBABLE = /^(touch\.task\.|touch\.workflow\.|stream\.task\.)[^.]+$/; +/** 主题形状唯一判据。导出是为了让路由能在**动任何状态之前**预检,见 badShapeMessage */ +export function isSubscribable(topic: string): boolean { + return SUBSCRIBABLE.test(topic); +} + +/** 形状错的文案也只有这一份——路由预检与 hub.subscribe 报的必须是同一句 */ +export function badShapeMessage(topic: string): string { + return `主题 ${topic} 不可订阅——只接受 touch.task/workflow.* 与 stream.task.*`; +} + export class SseHub { private conns = new Map(); private ring: SseEvent[] = []; @@ -67,6 +77,16 @@ export class SseHub { } } + /** + * F2:连接归属。订阅面的授权判据是**连接的** staffSub,不是调用方的—— + * 路由必须先确认调用方就是连接主人,否则任何拿到别人 connectionId 的登录用户 + * 都能改人家的订阅集,还能拿 results 当权限探测器。 + * 连接不存在返回 null(交回路由走原本的 no-conn 路径,不额外泄漏存在性)。 + */ + ownerOf(connId: string): string | null { + return this.conns.get(connId)?.staffSub ?? null; + } + unregister(connId: string): void { const conn = this.conns.get(connId); this.conns.delete(connId); @@ -77,13 +97,7 @@ export class SseHub { // D3:形状校验排在连接查找**之前**。请求体里写了个根本不可订阅的主题,是这个请求 // 自身不合法(422),跟连接在不在无关;反过来先查连接的话,一个失效 connectionId // 会把形状错整个遮掉,HTTP 面只能回 200。 - if (!SUBSCRIBABLE.test(topic)) { - return { - ok: false, - reason: 'bad-shape', - message: `主题 ${topic} 不可订阅——只接受 touch.task/workflow.* 与 stream.task.*`, - }; - } + if (!isSubscribable(topic)) return { ok: false, reason: 'bad-shape', message: badShapeMessage(topic) }; const conn = this.conns.get(connId); if (!conn) return { ok: false, reason: 'no-conn', message: `连接 ${connId} 不存在——先建立 /api/stream` }; const taskId = topic.split('.').pop() ?? ''; diff --git a/src/modules/sse/routes.ts b/src/modules/sse/routes.ts index 2260dc5..d745eca 100644 --- a/src/modules/sse/routes.ts +++ b/src/modules/sse/routes.ts @@ -2,6 +2,7 @@ import { Hono } from 'hono'; import { randomUUID } from 'node:crypto'; import { streamSSE } from 'hono/streaming'; import { requireAuth, toAuthUser, verifyAccessToken } from '../../shared/keycloak-auth'; +import { badShapeMessage, isSubscribable } from './hub'; import type { Connection, SseHub, TaskAccessPort } from './hub'; export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono { @@ -60,18 +61,38 @@ export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono { r.post('/subscribe', requireAuth(), async (c) => { const body = (await c.req.json()) as { connectionId?: string; add?: string[]; remove?: string[] }; - if (!body.connectionId) return c.json({ error: 'validation_failed', message: '缺 connectionId(hello 事件里给了)' }, 422); + const connectionId = body.connectionId; + if (!connectionId) return c.json({ error: 'validation_failed', message: '缺 connectionId(hello 事件里给了)' }, 422); + const add = body.add ?? []; + + // F1:形状预检必须**先于**任何变更。此前是边订边查,`add:['touch.task.t1','bogus']` + // 会先把 t1 真订上(还白跑一次 task-access 往返)再整单 422——客户端以为全失败, + // 服务端却留了半截订阅,两边模型就此错开。422 是"请求写错了",应当零副作用。 + const badShape = add.find((t) => !isSubscribable(t)); + if (badShape !== undefined) { + return c.json({ error: 'validation_failed', message: badShapeMessage(badShape) }, 422); + } + + // F2:只能操作自己名下的连接。hub.subscribe 是拿**连接的** staffSub 去授权的, + // 不校验归属的话,任何登录用户只要知道别人的 connectionId 就能改人家的订阅集 + // (remove 更是连 subscribe 都不经过,此前完全没有任何校验)。 + const user = c.get('user'); + const owner = hub.ownerOf(connectionId); + if (owner !== null && owner !== user.id && !user.roles.includes('admin')) { + return c.json({ error: 'forbidden', message: '不能操作他人连接的订阅' }, 403); + } + const results: Record = {}; - for (const topic of body.add ?? []) { - const res = await hub.subscribe(body.connectionId, topic, access); - // D3:只有"形状不合法"是请求本身写错了 → 整单 422;连接不存在/被拒是运行时状态, - // 逐条落进 results 回 200(客户端据此决定重连还是申请权限)。判别式而非文案匹配。 + for (const topic of add) { + const res = await hub.subscribe(connectionId, topic, access); + // 形状已由上面的预检挡掉,这里只是不变量兜底:真漏进来也绝不降级成 200。 + // 连接不存在/被拒是运行时状态,逐条落 results 回 200(客户端据此决定重连还是申请权限)。 if (!res.ok && res.reason === 'bad-shape') { return c.json({ error: 'validation_failed', message: res.message }, 422); } results[topic] = res.ok ? 'ok' : `denied: ${res.message}`; } - for (const topic of body.remove ?? []) hub.unsubscribe(body.connectionId, topic); + for (const topic of body.remove ?? []) hub.unsubscribe(connectionId, topic); return c.json({ results }); }); diff --git a/src/modules/sse/task-access.ts b/src/modules/sse/task-access.ts index 7ccec58..ea6344c 100644 --- a/src/modules/sse/task-access.ts +++ b/src/modules/sse/task-access.ts @@ -11,8 +11,17 @@ export function createHttpTaskAccess(token: TokenProvider): TaskAccessPort { return { async allowed(taskId, staffSub) { const url = `${config.taskServiceUrl}/task/${encodeURIComponent(taskId)}/access?staff=${encodeURIComponent(staffSub)}`; - const res = await fetch(url, { headers: { authorization: `Bearer ${await token()}` } }); - if (res.status !== 200) return false; + const res = await fetch(url, { + headers: { authorization: `Bearer ${await token()}` }, + // F3:/subscribe 的 add 循环是串行的——任务服务吊着不答就会把整个订阅请求钉死 + signal: AbortSignal.timeout(3000), + }); + if (res.status !== 200) { + // F3:失败关闭是对的,但"人人被拒"如果不出声就无从诊断——把下游状态码打出来。 + // 注意只在非 200 时告警,不把网络异常吞成 false(下游宕机不该读作"没权限")。 + console.warn('[task-access] 非200', taskId, res.status); + return false; + } const body = (await res.json()) as { allowed?: boolean }; return body.allowed === true; }, diff --git a/test/integration/sse.test.ts b/test/integration/sse.test.ts index cad56d2..b7d94ab 100644 --- a/test/integration/sse.test.ts +++ b/test/integration/sse.test.ts @@ -1,6 +1,9 @@ -import { describe, expect, it, beforeAll } from 'vitest'; +import { describe, expect, it, beforeAll, vi } from 'vitest'; +import { Hono } from 'hono'; import { SseHub } from '../../src/modules/sse/hub'; import type { Connection, SseEvent } from '../../src/modules/sse/hub'; +import { sseRoutes } from '../../src/modules/sse/routes'; +import { createHttpTaskAccess } from '../../src/modules/sse/task-access'; import { buildApp } from '../../src/app/app'; import { makeToken, setupTestJwks } from '../support/token'; @@ -88,6 +91,89 @@ describe('SSE HTTP 面', () => { expect(bad.status).toBe(422); }); + it('F1:形状预检先于变更——整单 422 时一条订阅都不留', async () => { + const hub = new SseHub(); + const a = conn('live-1', 'test-sub-1'); // token 的默认 sub,过得了 F2 的归属校验 + hub.register(a.c); + // 直接挂 sseRoutes 并注入"一律放行"的授权口:走 buildApp 的话内建 access 会在 + // await token() 处就抛(测试环境没有 KEYCLOAK_CLIENT_SECRET),t1 到不了订阅集, + // 这条用例也就证明不了"预检之前不会发生变更"这件事。 + let accessCalls = 0; + const access = { + allowed: async (): Promise => { + accessCalls += 1; + return true; + }, + }; + const app = new Hono().route('/api/stream', sseRoutes(hub, access)); + const token = await makeToken({ roles: ['user'], aud: 'api-gateway' }); + const res = await app.request('/api/stream/subscribe', { + method: 'POST', + headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, + body: JSON.stringify({ connectionId: 'live-1', add: ['touch.task.t1', 'bogus'] }), + }); + expect(res.status).toBe(422); + // 关键:合法的 t1 排在非法项**前面**,边订边查的写法会把它真订上 + expect(a.c.topics.has('touch.task.t1')).toBe(false); + expect(a.c.topics.size).toBe(0); + expect(accessCalls).toBe(0); // 且一次授权往返都不该白跑 + }); + + it('F2:连接归属校验——他人 connectionId 一律 403(add 与 remove 同守)', async () => { + const hub = new SseHub(); + const app = buildApp(hub); + const owner = conn('own-1', 'test-sub-1'); + owner.c.topics.add('touch.task.t1'); + hub.register(owner.c); + const post = async (tok: string, payload: unknown): Promise => + app.request('/api/stream/subscribe', { + method: 'POST', + headers: { authorization: `Bearer ${tok}`, 'content-type': 'application/json' }, + body: JSON.stringify(payload), + }); + + // 闯入者:合法 token,但连接不是他的 —— remove 此前完全不过任何校验 + const intruder = await makeToken({ sub: 'sub-intruder', roles: ['user'], aud: 'api-gateway' }); + const denied = await post(intruder, { connectionId: 'own-1', remove: ['touch.task.t1'] }); + expect(denied.status).toBe(403); + expect(await denied.json()).toMatchObject({ error: 'forbidden' }); + expect(owner.c.topics.has('touch.task.t1')).toBe(true); // 没被别人退订掉 + + // 主人自己来则照常放行——校验不是一刀切拒绝 + const self = await makeToken({ roles: ['user'], aud: 'api-gateway' }); // sub=test-sub-1 + const ok = await post(self, { connectionId: 'own-1', remove: ['touch.task.t1'] }); + expect(ok.status).toBe(200); + expect(owner.c.topics.has('touch.task.t1')).toBe(false); + }); + + it('F3:task-access 非 200 → false(失败关闭)且 id 都转义进 URL', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + const calls: string[] = []; + try { + vi.stubGlobal('fetch', (url: string | URL) => { + calls.push(String(url)); + return Promise.resolve(new Response('down', { status: 503 })); + }); + const access = createHttpTaskAccess(async () => 'stub-token'); + expect(await access.allowed('t 1', 'sub/a')).toBe(false); + expect(calls[0]).toContain('/task/t%201/access?staff=sub%2Fa'); + expect(warn).toHaveBeenCalled(); // 不出声的"人人被拒"无从诊断 + + vi.stubGlobal('fetch', () => + Promise.resolve( + new Response(JSON.stringify({ allowed: true }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + ), + ); + expect(await access.allowed('t1', 's1')).toBe(true); + } finally { + vi.unstubAllGlobals(); + warn.mockRestore(); + } + }); + it('D11/D12:hello 先于回放;hub.shutdown() 能把在途长连收尾', async () => { const hub = new SseHub(8); const app = buildApp(hub);