fix: 终审修复——SSE 鉴权 401/503 分类、touch.workflow 默认拒、/subscribe 非数组 422

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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 {

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 {
+    } catch (err) {
-      // TODO(N2-final): JWKS/infra 失败应 503 而非 401
+      // 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 });
   });
 

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<Response> =>
+      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);