From a973e43421c9829a7450a7fe06d5f0e81c1b98c1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 13:13:11 -0800 Subject: [PATCH] Validate inbound WebSocket push payloads before dispatch - Decode `server.welcome`, terminal, and orchestration push events with Effect schemas - Drop malformed payloads with warning logs instead of forwarding invalid data - Add wsNativeApi tests for valid forwarding, caching, and malformed-payload handling --- apps/web/src/wsNativeApi.test.ts | 185 ++++++++++++++++++++++++++++++- apps/web/src/wsNativeApi.ts | 37 +++++-- 2 files changed, 212 insertions(+), 10 deletions(-) diff --git a/apps/web/src/wsNativeApi.test.ts b/apps/web/src/wsNativeApi.test.ts index 1197faa6ea191044ee00eafdc7d1cbc60c4b207e..b34fc0be4d0206a09f200b79d860eb2124b49730 100644 --- a/apps/web/src/wsNativeApi.test.ts +++ b/apps/web/src/wsNativeApi.test.ts @@ -1,8 +1,28 @@ -import { CommandId, ORCHESTRATION_WS_METHODS, ProjectId, ThreadId } from "@t3tools/contracts"; -import { describe, expect, it, vi } from "vitest"; +import { + CommandId, + ORCHESTRATION_WS_CHANNELS, + ORCHESTRATION_WS_METHODS, + ProjectId, + ThreadId, + WS_CHANNELS, +} from "@t3tools/contracts"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const requestMock = vi.fn<(...args: Array) => Promise>(); -const subscribeMock = vi.fn<(...args: Array) => () => void>(() => () => undefined); +const channelListeners = new Map void>>(); +const subscribeMock = vi.fn<(channel: string, listener: (data: unknown) => void) => () => void>( + (channel, listener) => { + const listeners = channelListeners.get(channel) ?? new Set<(data: unknown) => void>(); + listeners.add(listener); + channelListeners.set(channel, listeners); + return () => { + listeners.delete(listener); + if (listeners.size === 0) { + channelListeners.delete(channel); + } + }; + }, +); vi.mock("./wsTransport", () => { return { @@ -13,7 +33,166 @@ vi.mock("./wsTransport", () => { }; }); +function emitPush(channel: string, data: unknown): void { + const listeners = channelListeners.get(channel); + if (!listeners) return; + for (const listener of listeners) { + listener(data); + } +} + +beforeEach(() => { + vi.resetModules(); + requestMock.mockReset(); + subscribeMock.mockClear(); + channelListeners.clear(); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + describe("wsNativeApi", () => { + it("delivers and caches valid server.welcome payloads", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { createWsNativeApi, onServerWelcome } = await import("./wsNativeApi"); + + createWsNativeApi(); + const listener = vi.fn(); + onServerWelcome(listener); + + const payload = { cwd: "/tmp/workspace", projectName: "t3-code" }; + emitPush(WS_CHANNELS.serverWelcome, payload); + + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith(payload); + + const lateListener = vi.fn(); + onServerWelcome(lateListener); + + expect(lateListener).toHaveBeenCalledTimes(1); + expect(lateListener).toHaveBeenCalledWith(payload); + expect(warnSpy).not.toHaveBeenCalled(); + }); + + it("ignores invalid server.welcome payloads and keeps subscription active", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { createWsNativeApi, onServerWelcome } = await import("./wsNativeApi"); + + createWsNativeApi(); + const listener = vi.fn(); + onServerWelcome(listener); + + emitPush(WS_CHANNELS.serverWelcome, { cwd: 42, projectName: "t3-code" }); + emitPush(WS_CHANNELS.serverWelcome, { cwd: "/tmp/workspace", projectName: "t3-code" }); + + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith({ cwd: "/tmp/workspace", projectName: "t3-code" }); + expect(warnSpy).toHaveBeenCalledTimes(1); + expect(warnSpy).toHaveBeenCalledWith("Dropped inbound WebSocket push payload", { + reason: "decode-failed", + raw: { cwd: 42, projectName: "t3-code" }, + issue: expect.stringContaining("SchemaError"), + }); + }); + + it("forwards valid terminal and orchestration events", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { createWsNativeApi } = await import("./wsNativeApi"); + + const api = createWsNativeApi(); + const onTerminalEvent = vi.fn(); + const onDomainEvent = vi.fn(); + + api.terminal.onEvent(onTerminalEvent); + api.orchestration.onDomainEvent(onDomainEvent); + + const terminalEvent = { + threadId: "thread-1", + terminalId: "terminal-1", + createdAt: "2026-02-24T00:00:00.000Z", + type: "output", + data: "hello", + } as const; + emitPush(WS_CHANNELS.terminalEvent, terminalEvent); + + const orchestrationEvent = { + sequence: 1, + eventId: "event-1", + aggregateKind: "project", + aggregateId: "project-1", + occurredAt: "2026-02-24T00:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.created", + payload: { + projectId: "project-1", + title: "Project", + workspaceRoot: "/tmp/workspace", + defaultModel: null, + scripts: [], + createdAt: "2026-02-24T00:00:00.000Z", + updatedAt: "2026-02-24T00:00:00.000Z", + }, + } as const; + emitPush(ORCHESTRATION_WS_CHANNELS.domainEvent, orchestrationEvent); + + expect(onTerminalEvent).toHaveBeenCalledTimes(1); + expect(onTerminalEvent).toHaveBeenCalledWith(terminalEvent); + expect(onDomainEvent).toHaveBeenCalledTimes(1); + expect(onDomainEvent).toHaveBeenCalledWith(orchestrationEvent); + expect(warnSpy).not.toHaveBeenCalled(); + }); + + it("drops malformed terminal and orchestration push payloads", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + const { createWsNativeApi } = await import("./wsNativeApi"); + + const api = createWsNativeApi(); + const onTerminalEvent = vi.fn(); + const onDomainEvent = vi.fn(); + + api.terminal.onEvent(onTerminalEvent); + api.orchestration.onDomainEvent(onDomainEvent); + + emitPush(WS_CHANNELS.terminalEvent, { + threadId: "thread-1", + terminalId: "", + createdAt: "2026-02-24T00:00:00.000Z", + type: "output", + data: "hello", + }); + emitPush(ORCHESTRATION_WS_CHANNELS.domainEvent, { + sequence: -1, + type: "project.created", + }); + + expect(onTerminalEvent).not.toHaveBeenCalled(); + expect(onDomainEvent).not.toHaveBeenCalled(); + expect(warnSpy).toHaveBeenCalledTimes(2); + expect(warnSpy).toHaveBeenNthCalledWith(1, "Dropped inbound WebSocket push payload", { + reason: "decode-failed", + raw: { + threadId: "thread-1", + terminalId: "", + createdAt: "2026-02-24T00:00:00.000Z", + type: "output", + data: "hello", + }, + issue: expect.stringContaining("SchemaError"), + }); + expect(warnSpy).toHaveBeenNthCalledWith(2, "Dropped inbound WebSocket push payload", { + reason: "decode-failed", + raw: { + sequence: -1, + type: "project.created", + }, + issue: expect.stringContaining("SchemaError"), + }); + }); + it("wraps orchestration dispatch commands in the command envelope", async () => { requestMock.mockResolvedValue(undefined); const { createWsNativeApi } = await import("./wsNativeApi"); diff --git a/apps/web/src/wsNativeApi.ts b/apps/web/src/wsNativeApi.ts index 88ab2f3089b37139eaddee2dc94805aab2656525..f156e22adce60f0a8ed6f467bfe1095354c7e9b1 100644 --- a/apps/web/src/wsNativeApi.ts +++ b/apps/web/src/wsNativeApi.ts @@ -1,11 +1,14 @@ import { + OrchestrationEvent, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, type NativeApi, + TerminalEvent, WS_CHANNELS, WS_METHODS, - type WsWelcomePayload, + WsWelcomePayload, } from "@t3tools/contracts"; +import { Cause, Schema } from "effect"; import { showContextMenuFallback } from "./contextMenuFallback"; import { WsTransport } from "./wsTransport"; @@ -14,6 +17,22 @@ let instance: { api: NativeApi; transport: WsTransport } | null = null; const welcomeListeners = new Set<(payload: WsWelcomePayload) => void>(); let lastWelcome: WsWelcomePayload | null = null; +const decodeAndWarnOnFailure = ( + schema: Schema.Schema & { readonly DecodingServices: never }, + raw: unknown, +): T | null => { + const decoded = Schema.decodeUnknownExit(schema)(raw); + if (decoded._tag === "Failure") { + console.warn("Dropped inbound WebSocket push payload", { + reason: "decode-failed", + raw, + issue: Cause.pretty(decoded.cause), + }); + return null; + } + return decoded.value; +}; + /** * Subscribe to the server welcome message. If a welcome was already received * before this call, the listener fires synchronously with the cached payload. @@ -44,7 +63,8 @@ export function createWsNativeApi(): NativeApi { // Listen for server welcome and forward to registered listeners. // Also cache it so late subscribers (React effects) get it immediately. transport.subscribe(WS_CHANNELS.serverWelcome, (data) => { - const payload = data as WsWelcomePayload; + const payload = decodeAndWarnOnFailure(WsWelcomePayload, data); + if (!payload) return; lastWelcome = payload; for (const listener of welcomeListeners) { try { @@ -76,7 +96,10 @@ export function createWsNativeApi(): NativeApi { restart: (input) => transport.request(WS_METHODS.terminalRestart, input), close: (input) => transport.request(WS_METHODS.terminalClose, input), onEvent: (callback) => - transport.subscribe(WS_CHANNELS.terminalEvent, callback as (data: unknown) => void), + transport.subscribe(WS_CHANNELS.terminalEvent, (data) => { + const payload = decodeAndWarnOnFailure(TerminalEvent, data); + if (payload) callback(payload); + }), }, projects: { searchEntries: (input) => transport.request(WS_METHODS.projectsSearchEntries, input), @@ -135,10 +158,10 @@ export function createWsNativeApi(): NativeApi { replayEvents: (fromSequenceExclusive) => transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }), onDomainEvent: (callback) => - transport.subscribe( - ORCHESTRATION_WS_CHANNELS.domainEvent, - callback as (data: unknown) => void, - ), + transport.subscribe(ORCHESTRATION_WS_CHANNELS.domainEvent, (data) => { + const payload = decodeAndWarnOnFailure(OrchestrationEvent, data); + if (payload) callback(payload); + }), }, }; -- 2.54.0