fix(api): derive secure sse subscriptions

packages/api/src/__tests__/sse-subscription-policy.test.ts

@@ -0,0 +1,165 @@
+import { PermissionKey, SSE_EVENT_TYPES, SystemRole } from "@capakraken/shared";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  cancelPendingEvents,
+  eventBus,
+  permissionAudience,
+  resourceAudience,
+  roleAudience,
+  type SseEvent,
+  userAudience,
+} from "../sse/event-bus.js";
+import { deriveUserSseSubscription } from "../sse/subscription-policy.js";
+
+vi.mock("ioredis", () => {
+  const RedisMock = vi.fn().mockImplementation(() => ({
+    on: vi.fn(),
+    subscribe: vi.fn().mockResolvedValue(undefined),
+    publish: vi.fn().mockImplementation(() => {
+      throw new Error("Redis unavailable (test)");
+    }),
+  }));
+  return { Redis: RedisMock };
+});
+
+describe("sse subscription policy", () => {
+  let consoleWarnSpy: ReturnType<typeof vi.spyOn>;
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+    consoleWarnSpy = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+  });
+
+  afterEach(() => {
+    cancelPendingEvents();
+    consoleWarnSpy.mockRestore();
+    vi.useRealTimers();
+  });
+
+  it("derives canonical user, role, resource, and permission audiences server-side", () => {
+    const subscription = deriveUserSseSubscription({
+      userId: "user_1",
+      systemRole: SystemRole.USER,
+      resourceId: "res_1",
+      permissionOverrides: {
+        granted: [PermissionKey.VIEW_PLANNING, PermissionKey.MANAGE_ALLOCATIONS],
+      },
+    });
+
+    expect(subscription.includeUnscoped).toBe(false);
+    expect(subscription.permissions).toEqual(new Set([
+      PermissionKey.VIEW_PLANNING,
+      PermissionKey.MANAGE_ALLOCATIONS,
+    ]));
+    expect(subscription.audiences).toEqual([
+      permissionAudience(PermissionKey.MANAGE_ALLOCATIONS),
+      permissionAudience(PermissionKey.VIEW_PLANNING),
+      resourceAudience("res_1"),
+      roleAudience(SystemRole.USER),
+      userAudience("user_1"),
+    ]);
+  });
+
+  it("does not deliver planning events to a standard user without the matching audience", () => {
+    const received: SseEvent[] = [];
+    const unsubscribe = eventBus.subscribe(
+      (event) => {
+        received.push(event);
+      },
+      deriveUserSseSubscription({
+        userId: "user_1",
+        systemRole: SystemRole.USER,
+        resourceId: "res_self",
+      }),
+    );
+
+    eventBus.emit(
+      SSE_EVENT_TYPES.ALLOCATION_UPDATED,
+      { id: "allocation_1", resourceId: "res_other" },
+      [permissionAudience(PermissionKey.MANAGE_ALLOCATIONS), resourceAudience("res_other")],
+    );
+
+    vi.advanceTimersByTime(50);
+
+    expect(received).toHaveLength(0);
+    unsubscribe();
+  });
+
+  it("does not leak personal notifications to a manager subscribed with elevated permissions", () => {
+    const received: SseEvent[] = [];
+    const unsubscribe = eventBus.subscribe(
+      (event) => {
+        received.push(event);
+      },
+      deriveUserSseSubscription({
+        userId: "manager_1",
+        systemRole: SystemRole.MANAGER,
+      }),
+    );
+
+    eventBus.emit(
+      SSE_EVENT_TYPES.NOTIFICATION_CREATED,
+      { notificationId: "notification_1", userId: "user_2" },
+      [userAudience("user_2")],
+    );
+
+    vi.advanceTimersByTime(50);
+
+    expect(received).toHaveLength(0);
+    unsubscribe();
+  });
+
+  it("delivers a multi-audience planning event only to matching manager and affected resource subscribers", () => {
+    const managerReceived: SseEvent[] = [];
+    const affectedUserReceived: SseEvent[] = [];
+    const unrelatedUserReceived: SseEvent[] = [];
+
+    const unsubscribeManager = eventBus.subscribe(
+      (event) => {
+        managerReceived.push(event);
+      },
+      deriveUserSseSubscription({
+        userId: "manager_1",
+        systemRole: SystemRole.MANAGER,
+      }),
+    );
+
+    const unsubscribeAffectedUser = eventBus.subscribe(
+      (event) => {
+        affectedUserReceived.push(event);
+      },
+      deriveUserSseSubscription({
+        userId: "user_1",
+        systemRole: SystemRole.USER,
+        resourceId: "res_1",
+      }),
+    );
+
+    const unsubscribeUnrelatedUser = eventBus.subscribe(
+      (event) => {
+        unrelatedUserReceived.push(event);
+      },
+      deriveUserSseSubscription({
+        userId: "user_2",
+        systemRole: SystemRole.USER,
+        resourceId: "res_2",
+      }),
+    );
+
+    eventBus.emit(
+      SSE_EVENT_TYPES.ALLOCATION_UPDATED,
+      { id: "allocation_1", resourceId: "res_1" },
+      [permissionAudience(PermissionKey.MANAGE_ALLOCATIONS), resourceAudience("res_1")],
+    );
+
+    vi.advanceTimersByTime(50);
+
+    expect(managerReceived).toHaveLength(1);
+    expect(affectedUserReceived).toHaveLength(1);
+    expect(unrelatedUserReceived).toHaveLength(0);
+
+    unsubscribeManager();
+    unsubscribeAffectedUser();
+    unsubscribeUnrelatedUser();
+  });
+});

packages/api/src/sse/event-bus.ts

@@ -1,7 +1,11 @@
 import { Redis } from "ioredis";
 import { PermissionKey, SSE_EVENT_TYPES, SystemRole, type SseEventType } from "@capakraken/shared";
 
-export type SseAudience = string;
+export type UserSseAudience = `user:${string}`;
+export type RoleSseAudience = `role:${string}`;
+export type PermissionSseAudience = `permission:${string}`;
+export type ResourceSseAudience = `resource:${string}`;
+export type SseAudience = UserSseAudience | RoleSseAudience | PermissionSseAudience | ResourceSseAudience;
 
 export interface SseEvent {
   type: SseEventType;
@@ -42,8 +46,8 @@ interface BufferEntry {
 
 const debounceBuffer = new Map<string, BufferEntry>();
 
-function normalizeAudiences(audiences?: Iterable<SseAudience>): SseAudience[] {
-  return [...new Set(Array.from(audiences ?? [], (audience) => audience.trim()).filter(Boolean))].sort();
+export function canonicalizeSseAudiences(audiences?: Iterable<SseAudience>): SseAudience[] {
+  return [...new Set(Array.from(audiences ?? [], (audience) => audience.trim() as SseAudience).filter(Boolean))].sort();
 }
 
 function getBufferKey(type: SseEventType, audience: readonly SseAudience[]): string {
@@ -91,7 +95,7 @@ function extractScopedResourceIds(payload: Record<string, unknown>): string[] {
 }
 
 function buildPlanningAudiences(payload: Record<string, unknown>): SseAudience[] {
-  return normalizeAudiences([
+  return canonicalizeSseAudiences([
     permissionAudience(PermissionKey.MANAGE_ALLOCATIONS),
     ...extractScopedResourceIds(payload).map((resourceId) => resourceAudience(resourceId)),
   ]);
@@ -181,7 +185,7 @@ function setupSubscriber(): void {
           type: parsed.type,
           payload: parsed.payload,
           timestamp: parsed.timestamp,
-          audience: normalizeAudiences(parsed.audience),
+          audience: canonicalizeSseAudiences(parsed.audience),
         });
       } catch {
         // ignore parse errors
@@ -200,7 +204,7 @@ class EventBus {
   subscribe(fn: Subscriber, options: SseSubscriptionOptions = {}): () => void {
     const subscription: Subscription = {
       fn,
-      audiences: new Set(normalizeAudiences(options.audiences)),
+      audiences: new Set(canonicalizeSseAudiences(options.audiences)),
       includeUnscoped: options.includeUnscoped ?? true,
     };
     subscribers.add(subscription);
@@ -210,7 +214,7 @@ class EventBus {
   publish(event: SseEvent): void {
     const normalizedEvent: SseEvent = {
       ...event,
-      audience: normalizeAudiences(event.audience),
+      audience: canonicalizeSseAudiences(event.audience),
     };
 
     // Broadcast via Redis (all instances receive via subscriber.on("message"))
@@ -237,7 +241,7 @@ class EventBus {
       type,
       payload,
       timestamp: new Date().toISOString(),
-      audience: normalizeAudiences(audience),
+      audience: canonicalizeSseAudiences(audience),
     });
   }
 
@@ -249,7 +253,7 @@ class EventBus {
 // Local delivery with debounce: buffer events of the same type and audience
 // within a 50ms window and then deliver a single (possibly aggregated) event.
 function publishLocal(event: SseEvent): void {
-  const audience = normalizeAudiences(event.audience);
+  const audience = canonicalizeSseAudiences(event.audience);
   const key = getBufferKey(event.type, audience);
   const existing = debounceBuffer.get(key);
 

packages/api/src/sse/index.ts

@@ -0,0 +1,2 @@
+export * from "./event-bus.js";
+export * from "./subscription-policy.js";

packages/api/src/sse/subscription-policy.ts

@@ -0,0 +1,45 @@
+import { resolvePermissions, type PermissionKey, type PermissionOverrides, SystemRole } from "@capakraken/shared";
+import {
+  canonicalizeSseAudiences,
+  permissionAudience,
+  resourceAudience,
+  roleAudience,
+  type SseAudience,
+  type SseSubscriptionOptions,
+  userAudience,
+} from "./event-bus.js";
+
+export interface SseSubscriberIdentity {
+  userId: string;
+  systemRole: SystemRole;
+  permissionOverrides?: PermissionOverrides | null;
+  resourceId?: string | null;
+}
+
+export interface DerivedSseSubscription extends SseSubscriptionOptions {
+  audiences: SseAudience[];
+  permissions: Set<PermissionKey>;
+  includeUnscoped: false;
+}
+
+export function deriveUserSseSubscription(
+  identity: SseSubscriberIdentity,
+  roleDefaults?: Record<string, PermissionKey[]>,
+): DerivedSseSubscription {
+  const permissions = resolvePermissions(
+    identity.systemRole,
+    identity.permissionOverrides ?? null,
+    roleDefaults,
+  );
+
+  return {
+    audiences: canonicalizeSseAudiences([
+      userAudience(identity.userId),
+      roleAudience(identity.systemRole),
+      ...(identity.resourceId ? [resourceAudience(identity.resourceId)] : []),
+      ...Array.from(permissions, (permission) => permissionAudience(permission)),
+    ]),
+    permissions,
+    includeUnscoped: false,
+  };
+}