diff --git a/src/app/app.ts b/src/app/app.ts index 4c6cadc..7c84e09 100644 --- a/src/app/app.ts +++ b/src/app/app.ts @@ -7,6 +7,7 @@ import { uploadRoutes } from '../modules/upload/routes'; import { SseHub } from '../modules/sse/hub'; import { sseRoutes } from '../modules/sse/routes'; import { createHttpTaskAccess } from '../modules/sse/task-access'; +import { createHttpWorkflowAccess } from '../modules/sse/workflow-access'; /** * hub 由外面传：server.ts 要拿同一个实例去 attachNats／shutdown。缺省自建一个， @@ -23,9 +24,12 @@ export function buildApp(hub: SseHub = new SseHub()): Hono { app.route('/health', healthRoute()); // 公开：不过 requireAuth // 惰性构造：createKeycloakTokenProvider 只在闭包被调用时读 KEYCLOAK_CLIENT_SECRET， // 建 app 本身不要求环境齐备，否则任何 buildApp() 的测试都会当场炸。 - const access = createHttpTaskAccess(createKeycloakTokenProvider()); + // 两口共用同一个 token provider：换来的是同一个服务账号 token，缓存也只此一份 + const serviceToken = createKeycloakTokenProvider(); + const access = createHttpTaskAccess(serviceToken); + const workflowAccess = createHttpWorkflowAccess(serviceToken); // 顺序要紧：/api/stream 与 /api/upload 必须先于 /api 代理挂载，否则被代理的 catch-all 吞掉 - app.route('/api/stream', sseRoutes(hub, access)); + app.route('/api/stream', sseRoutes(hub, access, workflowAccess)); app.route('/api/upload', uploadRoutes()); app.route('/api', proxyRoutes()); return app; diff --git a/src/modules/sse/hub.ts b/src/modules/sse/hub.ts index b56d136..fefa676 100644 --- a/src/modules/sse/hub.ts +++ b/src/modules/sse/hub.ts @@ -15,6 +15,11 @@ export interface Connection { close?(): void; } export interface TaskAccessPort { allowed(taskId: string, staffSub: string): Promise } +/** + * 工作流可见性口。与 TaskAccessPort 分开而不是复用同一口：判据不同源——任务问的是单张单的 + * 归属，工作流问的是**实例定义里**的节点归属（参与人/审批人），任务服务上也是两条端点。 + */ +export interface WorkflowAccessPort { allowed(workflowId: string, staffSub: string): Promise } /** * 订阅失败带判别式：路由要按失败**种类**分流 HTTP 码（形状错＝422 整单驳回， @@ -26,12 +31,16 @@ export type SubscribeResult = | { ok: false; reason: 'bad-shape' | 'no-conn' | 'denied'; message: string }; /** - * I6：可订阅集合＝**有授权规则**的集合，默认拒。此前这里收着 touch.workflow.，而 - * subscribe 的鉴权分支只覆盖 touch.task./stream.task.——形状过了就直接进订阅集， - * 于是任何登录用户订 touch.workflow.<任意 id> 都能收到别人流程的推送。工作流的 - * 可见性规则（发起人/参与节点/admin）要到 N3 才有，规则到位之前不放进来。 + * 可订阅集合＝**有授权规则**的集合，默认拒。 + * + * touch.workflow. 在 I6 被摘出去过：那时形状一过就直接进订阅集，而 subscribe 的鉴权分支只 + * 覆盖 touch.task./stream.task.，于是任何登录用户订 touch.workflow.<任意 id> 都能收别人流程 + * 的推送——摘掉它是因为**当时没有规则**，不是因为工作流不该推。 + * N3 规则到位，故放回来：可见性＝该实例定义里的参与人（节点 ownerSub）或审批人 + * （approverSub），经 workflowAccess 口向任务服务问；admin 旁路。判据落在下面 subscribe 的 + * 鉴权分支里——这两处必须同一次改动，只放形状白名单就是把 I6 原样退回去。 */ -const SUBSCRIBABLE = /^(touch\.task\.|stream\.task\.)[^.]+$/; +const SUBSCRIBABLE = /^(touch\.task\.|touch\.workflow\.|stream\.task\.)[^.]+$/; /** 主题形状唯一判据。导出是为了让路由能在**动任何状态之前**预检，见 badShapeMessage */ export function isSubscribable(topic: string): boolean { @@ -40,7 +49,7 @@ export function isSubscribable(topic: string): boolean { /** 形状错的文案也只有这一份——路由预检与 hub.subscribe 报的必须是同一句 */ export function badShapeMessage(topic: string): string { - return `主题 ${topic} 不可订阅——只接受 touch.task.* 与 stream.task.*`; + return `主题 ${topic} 不可订阅——只接受 touch.task.*、touch.workflow.* 与 stream.task.*`; } export class SseHub { @@ -99,17 +108,33 @@ export class SseHub { if (conn) for (const t of conn.topics) this.maybeDropStreamSub(t); } - async subscribe(connId: string, topic: string, access: TaskAccessPort): Promise { + async subscribe( + connId: string, + topic: string, + access: TaskAccessPort, + workflowAccess: WorkflowAccessPort, + ): Promise { // D3：形状校验排在连接查找**之前**。请求体里写了个根本不可订阅的主题，是这个请求 // 自身不合法（422），跟连接在不在无关；反过来先查连接的话，一个失效 connectionId // 会把形状错整个遮掉，HTTP 面只能回 200。 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() ?? ''; + const subjectId = topic.split('.').pop() ?? ''; if (!conn.isAdmin && (topic.startsWith('stream.task.') || topic.startsWith('touch.task.'))) { - if (!(await access.allowed(taskId, conn.staffSub))) { - return { ok: false, reason: 'denied', message: `无权订阅任务 ${taskId}——需要负责人/审批人/admin` }; + if (!(await access.allowed(subjectId, conn.staffSub))) { + return { ok: false, reason: 'denied', message: `无权订阅任务 ${subjectId}——需要负责人/审批人/admin` }; + } + } + // 工作流另走一口：判据是实例**定义里**的节点归属，与单张单的归属不是一回事。 + // admin 同样旁路——既是规则（admin 全可见），也免掉一次白跑的跨服务往返。 + if (!conn.isAdmin && topic.startsWith('touch.workflow.')) { + if (!(await workflowAccess.allowed(subjectId, conn.staffSub))) { + return { + ok: false, + reason: 'denied', + message: `无权订阅工作流 ${subjectId}——需要该流程的参与人/审批人或 admin`, + }; } } conn.topics.add(topic); diff --git a/src/modules/sse/routes.ts b/src/modules/sse/routes.ts index 0e1563e..896ad3a 100644 --- a/src/modules/sse/routes.ts +++ b/src/modules/sse/routes.ts @@ -3,9 +3,9 @@ import { randomUUID } from 'node:crypto'; import { streamSSE } from 'hono/streaming'; import { classifyAuthError, requireAuth, toAuthUser, verifyAccessToken } from '../../shared/keycloak-auth'; import { badShapeMessage, isSubscribable } from './hub'; -import type { Connection, SseHub, TaskAccessPort } from './hub'; +import type { Connection, SseHub, TaskAccessPort, WorkflowAccessPort } from './hub'; -export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono { +export function sseRoutes(hub: SseHub, access: TaskAccessPort, workflowAccess: WorkflowAccessPort): Hono { const r = new Hono(); r.get('/', async (c) => { @@ -97,7 +97,7 @@ export function sseRoutes(hub: SseHub, access: TaskAccessPort): Hono { const results: Record = {}; for (const topic of add) { - const res = await hub.subscribe(connectionId, topic, access); + const res = await hub.subscribe(connectionId, topic, access, workflowAccess); // 形状已由上面的预检挡掉，这里只是不变量兜底：真漏进来也绝不降级成 200。 // 连接不存在／被拒是运行时状态，逐条落 results 回 200（客户端据此决定重连还是申请权限）。 if (!res.ok && res.reason === 'bad-shape') { diff --git a/src/modules/sse/workflow-access.ts b/src/modules/sse/workflow-access.ts new file mode 100644 index 0000000..47f4eb1 --- /dev/null +++ b/src/modules/sse/workflow-access.ts @@ -0,0 +1,30 @@ +import { config } from '../../shared/config'; +import type { TokenProvider } from '../../shared/service-token'; +import type { WorkflowAccessPort } from './hub'; + +/** + * 订阅工作流主题前问任务服务一句：`GET {TASK_SERVICE_URL}/workflow/:id/access?staff=`。 + * 走服务账号 token（该端点在任务侧是 requireService(['api-gateway'])——只认网关，代理没有 + * 理由问流程可见性），不是转发用户 token。 + * 非 200 一律当"无权"——推送面失败关闭：宁可少订一条，不可把别人流程的推进扇给不该看的人。 + */ +export function createHttpWorkflowAccess(token: TokenProvider): WorkflowAccessPort { + return { + async allowed(workflowId, staffSub) { + const url = `${config.taskServiceUrl}/workflow/${encodeURIComponent(workflowId)}/access?staff=${encodeURIComponent(staffSub)}`; + const res = await fetch(url, { + headers: { authorization: `Bearer ${await token()}` }, + // F3 同理：/subscribe 的 add 循环是串行的——任务服务吊着不答就会把整个订阅请求钉死 + signal: AbortSignal.timeout(3000), + }); + if (res.status !== 200) { + // 失败关闭是对的，但"人人被拒"如果不出声就无从诊断——把下游状态码打出来。 + // 注意只在非 200 时告警，不把网络异常吞成 false（下游宕机不该读作"没权限"）。 + console.warn('[workflow-access] 非200', workflowId, 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 c1355ce..3547846 100644 --- a/test/integration/sse.test.ts +++ b/test/integration/sse.test.ts @@ -4,6 +4,7 @@ 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'; +import { createHttpWorkflowAccess } from '../../src/modules/sse/workflow-access'; import { buildApp } from '../../src/app/app'; import { makeToken, setupTestJwks } from '../support/token'; @@ -12,6 +13,10 @@ function conn(id: string, staffSub: string, topics: string[] = []): { c: Connect return { c: { id, staffSub, isAdmin: false, topics: new Set(topics), send: (e) => got.push(e) }, got }; } +/** 一律放行/一律拒绝的假口：两口分开传，才证明得了"这一拒来自哪一口" */ +const ALLOW = { allowed: async (): Promise => true }; +const DENY = { allowed: async (): Promise => false }; + describe('SseHub 路由', () => { it('staff 精确投递、system 广播、task 按订阅集、回放窗口', () => { const hub = new SseHub(8); @@ -36,48 +41,85 @@ describe('SseHub 路由', () => { const hub = new SseHub(); const a = conn('a', 'sub-a'); hub.register(a.c); - const denyAccess = { allowed: async () => false }; - const res = await hub.subscribe('a', 'stream.task.t9', denyAccess); + const res = await hub.subscribe('a', 'stream.task.t9', DENY, ALLOW); expect(res.ok).toBe(false); expect(a.c.topics.has('stream.task.t9')).toBe(false); }); it('D3：形状校验排在连接查找之前，失败带判别式 reason 供路由分流', async () => { const hub = new SseHub(); - const denyAccess = { allowed: async () => false }; // 连接不存在 + 形状也不对 → 必须报形状（否则 HTTP 面永远回不出 422） - expect(await hub.subscribe('nope', 'touch.staff.hack', denyAccess)).toMatchObject({ + expect(await hub.subscribe('nope', 'touch.staff.hack', DENY, DENY)).toMatchObject({ ok: false, reason: 'bad-shape', }); - expect(await hub.subscribe('nope', 'touch.task.t1', denyAccess)).toMatchObject({ + expect(await hub.subscribe('nope', 'touch.task.t1', DENY, DENY)).toMatchObject({ ok: false, reason: 'no-conn', }); const a = conn('a', 'sub-a'); hub.register(a.c); - expect(await hub.subscribe('a', 'stream.task.t9', denyAccess)).toMatchObject({ + expect(await hub.subscribe('a', 'stream.task.t9', DENY, DENY)).toMatchObject({ ok: false, reason: 'denied', }); }); - it('I6：touch.workflow.* 默认拒——没有授权规则的主题不进可订阅集', async () => { - // 形状白名单曾经收着 touch.workflow.，而 subscribe 的鉴权分支只覆盖 touch.task./stream.task.： - // 形状一过就直接进订阅集，任何登录用户订 touch.workflow.<任意 id> 都能收别人流程的推送。 - // N3 给工作流补上可见性规则之后再放回来。 - expect(isSubscribable('touch.workflow.w1')).toBe(false); + it('N3（原 I6 反转）：touch.workflow.* 放回可订阅集——规则到位了才收得进来', async () => { + // I6 当初把它摘出去，是因为**当时没有规则**：形状一过就直接进订阅集，而鉴权分支只覆盖 + // touch.task./stream.task.，任何登录用户订 touch.workflow.<任意 id> 都能收别人流程的推送。 + // N3 补上归属口（参与人/审批人，admin 旁路）之后放回来——放形状与补鉴权是同一次改动， + // 下面两条用例就是"规则确实生效"的那一半，缺了它这条断言等于把 I6 原样退回去。 + expect(isSubscribable('touch.workflow.w1')).toBe(true); expect(isSubscribable('touch.task.t1')).toBe(true); expect(isSubscribable('stream.task.t1')).toBe(true); + // 放回的只是这一节：多一节的主题、以及别的域照旧不可订阅 + expect(isSubscribable('touch.workflow.w1.extra')).toBe(false); + expect(isSubscribable('touch.workflow.')).toBe(false); + expect(isSubscribable('touch.staff.hack')).toBe(false); + }); + + it('N3：工作流按归属口放行——拒则不入集不投递，准则入集且收得到', async () => { 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' }); + // 任务口一律放行：这样"被拒"只可能来自**工作流**那一口，而不是碰巧被任务授权挡住 + const denied = await hub.subscribe('a', 'touch.workflow.w1', ALLOW, DENY); + expect(denied).toMatchObject({ ok: false, reason: 'denied' }); + expect(denied.ok === false && denied.message).toContain('参与人/审批人或 admin'); expect(a.c.topics.size).toBe(0); hub.deliver('touch.workflow.w1', '{}'); - expect(a.got).toEqual([]); + expect(a.got).toEqual([]); // 拒了就一条都不该扇过去 + + const asked: string[][] = []; + const ok = await hub.subscribe('a', 'touch.workflow.w1', ALLOW, { + allowed: async (id: string, sub: string): Promise => { + asked.push([id, sub]); + return true; + }, + }); + expect(ok).toEqual({ ok: true }); + expect(asked).toEqual([['w1', 'sub-a']]); // 问的是主题末节＋**连接的** staffSub，不是调用方的 + hub.deliver('touch.workflow.w1', '{"n":1}'); + expect(a.got.map((e) => e.topic)).toEqual(['touch.workflow.w1']); + }); + + it('N3：admin 订工作流不问归属口——admin 全可见，白跑一次跨服务往返没意义', async () => { + const hub = new SseHub(); + const a = conn('adm', 'sub-adm'); + hub.register({ ...a.c, isAdmin: true }); // topics/send 与 a.c 共用，断言照旧看 a + let calls = 0; + const counting = { + allowed: async (): Promise => { + calls += 1; + return false; // 真被问到就会拒——订上了即证明没问 + }, + }; + expect(await hub.subscribe('adm', 'touch.workflow.w9', counting, counting)).toEqual({ ok: true }); + expect(calls).toBe(0); + expect(a.c.topics.has('touch.workflow.w9')).toBe(true); + hub.deliver('touch.workflow.w9', '{}'); + expect(a.got.map((e) => e.topic)).toEqual(['touch.workflow.w9']); }); it('D12：shutdown 逐条调 close 把手并摘除连接', () => { @@ -123,7 +165,7 @@ describe('SSE HTTP 面', () => { return true; }, }; - const app = new Hono().route('/api/stream', sseRoutes(hub, access)); + const app = new Hono().route('/api/stream', sseRoutes(hub, access, access)); const token = await makeToken({ roles: ['user'], aud: 'api-gateway' }); const res = await app.request('/api/stream/subscribe', { method: 'POST', @@ -143,7 +185,7 @@ describe('SSE HTTP 面', () => { 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 app = new Hono().route('/api/stream', sseRoutes(hub, ALLOW, ALLOW)); const token = await makeToken({ roles: ['user'], aud: 'api-gateway' }); const post = async (payload: unknown): Promise => app.request('/api/stream/subscribe', { @@ -218,6 +260,35 @@ describe('SSE HTTP 面', () => { } }); + it('workflow-access 与 task-access 同形：非 200 → false（失败关闭），打的是 /workflow/:id/access', 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 wf = createHttpWorkflowAccess(async () => 'stub-token'); + expect(await wf.allowed('w 1', 'sub/a')).toBe(false); + // 路径与任务口分开：任务侧是两条端点，判据也不同源 + expect(calls[0]).toContain('/workflow/w%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 wf.allowed('w1', 's1')).toBe(true); + } finally { + vi.unstubAllGlobals(); + warn.mockRestore(); + } + }); + it('D11/D12：hello 先于回放；hub.shutdown() 能把在途长连收尾', async () => { const hub = new SseHub(8); const app = buildApp(hub);