fix: SSE 订阅——形状预检先于变更、连接归属校验、task-access 超时与告警

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<string, Connection>();
   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)) {
+    if (!isSubscribable(topic)) return { ok: false, reason: 'bad-shape', message: badShapeMessage(topic) };
-      return {
-        ok: false,
-        reason: 'bad-shape',
-        message: `主题 ${topic} 不可订阅——只接受 touch.task/workflow.* 与 stream.task.*`,
-      };
-    }
     const conn = this.conns.get(connId);
     if (!conn) return { ok: false, reason: 'no-conn', message: `连接 ${connId} 不存在——先建立 /api/stream` };
     const taskId = topic.split('.').pop() ?? '';

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<string, string> = {};
-    for (const topic of body.add ?? []) {
+    for (const topic of add) {
-      const res = await hub.subscribe(body.connectionId, topic, access);
+      const res = await hub.subscribe(connectionId, topic, access);
-      // D3：只有"形状不合法"是请求本身写错了 → 整单 422；连接不存在／被拒是运行时状态，
+      // 形状已由上面的预检挡掉，这里只是不变量兜底：真漏进来也绝不降级成 200。
-      // 逐条落进 results 回 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 });
   });
 

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()}` } });
+      const res = await fetch(url, {
-      if (res.status !== 200) return false;
+        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;
     },

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