From 9a0209428163c34496824b87f9545719b4c909bc Mon Sep 17 00:00:00 2001 From: lovefyf Date: Thu, 13 Aug 2026 14:26:48 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=BB=88=E5=AE=A1=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E2=80=94=E2=80=94SSE=20=E9=89=B4=E6=9D=83=20401/503=20?= =?UTF-8?q?=E5=88=86=E7=B1=BB=E3=80=81touch.workflow=20=E9=BB=98=E8=AE=A4?= =?UTF-8?q?=E6=8B=92=E3=80=81/subscribe=20=E9=9D=9E=E6=95=B0=E7=BB=84=2042?= =?UTF-8?q?2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 --- src/modules/sse/hub.ts | 10 ++++++-- src/modules/sse/routes.ts | 25 +++++++++++++++----- test/integration/sse.test.ts | 46 +++++++++++++++++++++++++++++++++++- 3 files changed, 72 insertions(+), 9 deletions(-) diff --git a/src/modules/sse/hub.ts b/src/modules/sse/hub.ts index ff8d840..b56d136 100644 --- a/src/modules/sse/hub.ts +++ b/src/modules/sse/hub.ts @@ -25,7 +25,13 @@ export type SubscribeResult = | { ok: true } | { ok: false; reason: 'bad-shape' | 'no-conn' | 'denied'; message: string }; -const SUBSCRIBABLE = /^(touch\.task\.|touch\.workflow\.|stream\.task\.)[^.]+$/; +/** + * I6：可订阅集合＝**有授权规则**的集合，默认拒。此前这里收着 touch.workflow.，而 + * subscribe 的鉴权分支只覆盖 touch.task./stream.task.——形状过了就直接进订阅集， + * 于是任何登录用户订 touch.workflow.<任意 id> 都能收到别人流程的推送。工作流的 + * 可见性规则（发起人/参与节点/admin）要到 N3 才有，规则到位之前不放进来。 + */ +const SUBSCRIBABLE = /^(touch\.task\.|stream\.task\.)[^.]+$/; /** 主题形状唯一判据。导出是为了让路由能在**动任何状态之前**预检，见 badShapeMessage */ export function isSubscribable(topic: string): boolean { @@ -34,7 +40,7 @@ export function isSubscribable(topic: string): boolean { /** 形状错的文案也只有这一份——路由预检与 hub.subscribe 报的必须是同一句 */ export function badShapeMessage(topic: string): string { - return `主题 ${topic} 不可订阅——只接受 touch.task/workflow.* 与 stream.task.*`; + return `主题 ${topic} 不可订阅——只接受 touch.task.* 与 stream.task.*`; } export class SseHub { diff --git a/src/modules/sse/routes.ts b/src/modules/sse/routes.ts index d745eca..0e1563e 100644 --- a/src/modules/sse/routes.ts +++ b/src/modules/sse/routes.ts @@ -1,7 +1,7 @@ import { Hono } from 'hono'; import { randomUUID } from 'node:crypto'; import { streamSSE } from 'hono/streaming'; -import { requireAuth, toAuthUser, verifyAccessToken } from '../../shared/keycloak-auth'; +import { classifyAuthError, requireAuth, toAuthUser, verifyAccessToken } from '../../shared/keycloak-auth'; import { badShapeMessage, isSubscribable } from './hub'; import type { Connection, SseHub, TaskAccessPort } from './hub'; @@ -15,8 +15,12 @@ export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono { let user; try { user = toAuthUser(await verifyAccessToken(token)); - } catch { - // TODO(N2-final): JWKS/infra 失败应 503 而非 401 + } catch (err) { + // I4：这里不能自己另判一套。SSE 是长连，客户端拿到 401 会清 token 去重登录， + // 而 JWKS 拉不动的时候重登录打的还是同一个 Keycloak——分类判据只此一份（requireAuth 同源）。 + if (classifyAuthError(err) === 503) { + return c.json({ error: 'auth_unavailable', message: '鉴权服务暂不可达——稍后重试' }, 503); + } return c.json({ error: 'invalid_token', message: 'token 校验失败' }, 401); } const lastRaw = c.req.header('last-event-id'); @@ -60,10 +64,19 @@ 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[] }; + const body = (await c.req.json()) as { connectionId?: string; add?: unknown; remove?: unknown }; const connectionId = body.connectionId; if (!connectionId) return c.json({ error: 'validation_failed', message: '缺 connectionId（hello 事件里给了）' }, 422); - const add = body.add ?? []; + // I7：类型断言不是校验。`{"add":"touch.task.t1"}` 过去被 as string[] 一路带到 .find， + // 字符串上没有 find 就是 TypeError → 500：请求写错了却报成服务端故障，且 500 在网关 + // 这一层还会被当成可重试。形状不对一律 422，且必须排在任何使用之前。 + if (body.add !== undefined && !Array.isArray(body.add)) { + return c.json({ error: 'validation_failed', message: 'add 必须是主题数组' }, 422); + } + if (body.remove !== undefined && !Array.isArray(body.remove)) { + return c.json({ error: 'validation_failed', message: 'remove 必须是主题数组' }, 422); + } + const add = (body.add ?? []) as string[]; // F1：形状预检必须**先于**任何变更。此前是边订边查，`add:['touch.task.t1','bogus']` // 会先把 t1 真订上（还白跑一次 task-access 往返）再整单 422——客户端以为全失败， @@ -92,7 +105,7 @@ export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono { } results[topic] = res.ok ? 'ok' : `denied: ${res.message}`; } - for (const topic of body.remove ?? []) hub.unsubscribe(connectionId, topic); + for (const topic of (body.remove ?? []) as string[]) hub.unsubscribe(connectionId, topic); return c.json({ results }); }); diff --git a/test/integration/sse.test.ts b/test/integration/sse.test.ts index b7d94ab..c1355ce 100644 --- a/test/integration/sse.test.ts +++ b/test/integration/sse.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it, beforeAll, vi } from 'vitest'; import { Hono } from 'hono'; -import { SseHub } from '../../src/modules/sse/hub'; +import { SseHub, isSubscribable } 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'; @@ -62,6 +62,24 @@ describe('SseHub 路由', () => { }); }); + it('I6：touch.workflow.* 默认拒——没有授权规则的主题不进可订阅集', async () => { + // 形状白名单曾经收着 touch.workflow.，而 subscribe 的鉴权分支只覆盖 touch.task./stream.task.： + // 形状一过就直接进订阅集，任何登录用户订 touch.workflow.<任意 id> 都能收别人流程的推送。 + // N3 给工作流补上可见性规则之后再放回来。 + expect(isSubscribable('touch.workflow.w1')).toBe(false); + expect(isSubscribable('touch.task.t1')).toBe(true); + expect(isSubscribable('stream.task.t1')).toBe(true); + const hub = new SseHub(); + const a = conn('a', 'sub-a'); + hub.register(a.c); + // 放行一切的 access 口：证明拒绝来自形状白名单本身，而不是碰巧被授权挡住了 + const res = await hub.subscribe('a', 'touch.workflow.w1', { allowed: async () => true }); + expect(res).toMatchObject({ ok: false, reason: 'bad-shape' }); + expect(a.c.topics.size).toBe(0); + hub.deliver('touch.workflow.w1', '{}'); + expect(a.got).toEqual([]); + }); + it('D12：shutdown 逐条调 close 把手并摘除连接', () => { const hub = new SseHub(); let closed = 0; @@ -119,6 +137,32 @@ describe('SSE HTTP 面', () => { expect(accessCalls).toBe(0); // 且一次授权往返都不该白跑 }); + it('I7：add/remove 不是数组 → 422（不是 500）', async () => { + // `as string[]` 只是断言。字符串上没有 .find，此前直接 TypeError 冒到 500—— + // 请求写错却报成服务端故障，而 500 在网关这一层还会被客户端/重试层当成可重试。 + const hub = new SseHub(); + const a = conn('live-7', 'test-sub-1'); + hub.register(a.c); + const app = new Hono().route('/api/stream', sseRoutes(hub, { allowed: async () => true })); + const token = await makeToken({ roles: ['user'], aud: 'api-gateway' }); + const post = async (payload: unknown): Promise => + app.request('/api/stream/subscribe', { + method: 'POST', + headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, + body: JSON.stringify(payload), + }); + + const badAdd = await post({ connectionId: 'live-7', add: 'touch.task.t1' }); + expect(badAdd.status).toBe(422); + expect(await badAdd.json()).toMatchObject({ error: 'validation_failed' }); + const badRemove = await post({ connectionId: 'live-7', remove: { '0': 'touch.task.t1' } }); + expect(badRemove.status).toBe(422); + expect(a.c.topics.size).toBe(0); // 422 零副作用 + // 缺省（undefined）仍然合法：只订不退、只退不订都是正常用法 + expect((await post({ connectionId: 'live-7', add: ['touch.task.t1'] })).status).toBe(200); + expect(a.c.topics.has('touch.task.t1')).toBe(true); + }); + it('F2：连接归属校验——他人 connectionId 一律 403（add 与 remove 同守）', async () => { const hub = new SseHub(); const app = buildApp(hub);