fix(types): validate script JSON boundaries

.github/workflows/ci.yml

@@ -117,6 +117,9 @@ jobs:
       - name: Test user merge CLI
         run: npm run test:user-merge-cli
 
+      - name: Test Cloudflare logs CLI
+        run: npm run test:cf-logs-cli
+
       - name: Check Prettier formatting
         run: npm run format:check
 

package.json

@@ -19,6 +19,7 @@
     "test:lint-sql-portability": "node --test scripts/lint-sql-portability.test.mjs",
     "test:rbac-bootstrap-owner": "node --experimental-transform-types --test scripts/bootstrap-workspace-owner.test.ts",
     "test:user-merge-cli": "node --experimental-transform-types --test scripts/merge-split-users.test.ts",
+    "test:cf-logs-cli": "node --experimental-strip-types --test scripts/cf-logs.test.ts",
     "test:coverage": "npm run test:coverage --workspaces --if-present",
     "test:integration": "npm run test:integration --workspaces --if-present",
     "typecheck": "npm run build -w @open-inspect/shared && npm run typecheck --workspaces --if-present",

scripts/cf-logs.test.ts

@@ -0,0 +1,112 @@
+import assert from "node:assert/strict";
+import { spawnSync } from "node:child_process";
+import { describe, it } from "node:test";
+import { fileURLToPath } from "node:url";
+
+function runLogs(response: unknown, ...args: string[]) {
+  // Exercise the real CLI without credentials or any network requests.
+  const preload = `globalThis.fetch = async () => Response.json(${JSON.stringify(response)});`;
+  return spawnSync(
+    process.execPath,
+    [
+      "--experimental-strip-types",
+      "--import",
+      `data:text/javascript,${encodeURIComponent(preload)}`,
+      fileURLToPath(new URL("./cf-logs.ts", import.meta.url)),
+      "--all",
+      ...args,
+    ],
+    {
+      encoding: "utf8",
+      env: {
+        ...process.env,
+        CLOUDFLARE_API_TOKEN: "test-token",
+        CLOUDFLARE_ACCOUNT_ID: "test-account",
+        NO_COLOR: "1",
+      },
+    }
+  );
+}
+
+function telemetry(events: unknown[]) {
+  return { success: true, result: { events: { events } } };
+}
+
+describe("Cloudflare logs JSON boundary", () => {
+  const event = {
+    timestamp: 1_700_000_000_000,
+    $metadata: { level: "info", service: "worker", id: "event-id" },
+    $workers: { scriptName: "worker", executionModel: "durableObject", outcome: "ok" },
+    source: {
+      level: "info",
+      component: "test",
+      msg: "hello",
+      session_id: "test-session",
+      extra: { n: 1 },
+    },
+    dataset: "worker-logs",
+    links: [{ id: "linked-event" }],
+  };
+
+  it("retains unconsumed fields in raw JSON and formats valid logs", () => {
+    const raw = runLogs(telemetry([event]), "--json");
+    assert.equal(raw.status, 0, raw.stderr);
+    assert.deepEqual(JSON.parse(raw.stdout), [event]);
+    const formatted = runLogs(telemetry([event]));
+    assert.equal(formatted.status, 0, formatted.stderr);
+    assert.match(formatted.stdout, /INFO.*worker.*test.*hello/);
+    assert.match(formatted.stderr, /Sessions:.*test-session/);
+  });
+
+  for (const response of [
+    telemetry([]),
+    { success: true },
+    { success: true, result: {} },
+    { success: true, result: { events: {} } },
+  ]) {
+    it(`preserves an empty optional events result: ${JSON.stringify(response)}`, () => {
+      const child = runLogs(response);
+      assert.equal(child.status, 0, child.stderr);
+      assert.match(child.stderr, /No logs found/);
+    });
+  }
+
+  for (const event of [
+    null,
+    [],
+    42,
+    { timestamp: "1700000000000" },
+    { timestamp: 1e100 },
+    { source: { level: 42 } },
+    { source: [] },
+    { $metadata: { level: false } },
+    { $workers: { scriptName: 42 } },
+  ]) {
+    it(`rejects malformed events without silently dropping them: ${JSON.stringify(event)}`, () => {
+      const child = runLogs(telemetry([event]));
+      assert.equal(child.status, 1);
+      assert.match(child.stderr, /malformed telemetry events/);
+      assert.equal(child.stdout, "");
+    });
+  }
+
+  for (const response of [
+    null,
+    { success: "true" },
+    { success: true, result: [] },
+    { success: true, result: { events: [] } },
+    { success: true, result: { events: { events: {} } } },
+  ]) {
+    it(`rejects malformed response envelopes: ${JSON.stringify(response)}`, () => {
+      const child = runLogs(response);
+      assert.equal(child.status, 1);
+      assert.match(child.stderr, /not a telemetry query result|malformed telemetry events/);
+    });
+  }
+
+  it("reports API errors with their original diagnostics", () => {
+    const child = runLogs({ success: false, errors: [{ message: "query failed" }] });
+    assert.equal(child.status, 1);
+    assert.match(child.stderr, /API error:.*query failed/);
+  });
+});

scripts/cf-logs.ts

@@ -40,32 +40,15 @@ interface TelemetryQuery {
 }
 
 interface WorkersMetadata {
-  id?: string;
-  requestId?: string;
-  traceId?: string;
-  trigger?: string;
   service?: string;
   level?: string;
   message?: string;
-  account?: string;
-  type?: string;
-  fingerprint?: string;
-  origin?: string;
-  messageTemplate?: string;
 }
 
 interface WorkersInfo {
-  truncated?: boolean;
-  event?: Record<string, unknown>;
   outcome?: string;
   scriptName?: string;
-  eventType?: string;
   executionModel?: string;
-  scriptVersion?: { id: string };
-  durableObjectId?: string;
-  requestId?: string;
-  cpuTimeMs?: number;
-  wallTimeMs?: number;
 }
 
 /** Application-level fields emitted by our logger via console.log(JSON.stringify(...)). */
@@ -75,10 +58,7 @@ interface LogSource {
   component?: string;
   msg?: string;
   message?: string;
-  ts?: number;
   session_id?: string;
-  request_id?: string;
-  trace_id?: string;
   [key: string]: unknown;
 }
 
@@ -95,9 +75,74 @@ interface LogEvent {
   $metadata?: WorkersMetadata;
   $workers?: WorkersInfo;
   source?: LogSource;
-  dataset?: string;
   timestamp?: number;
-  links?: unknown[];
+}
+
+interface TelemetryQueryResponse {
+  success: boolean;
+  errors?: unknown[];
+  result?: { events?: { events?: LogEvent[] } };
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function hasOptionalStrings(value: unknown, fields: string[]): boolean {
+  return (
+    value === undefined ||
+    (isRecord(value) &&
+      fields.every((field) => value[field] === undefined || typeof value[field] === "string"))
+  );
+}
+
+/** Validate the fields we consume, retaining all other fields for --json output. */
+function isLogEvent(value: unknown): value is LogEvent {
+  return (
+    isRecord(value) &&
+    (value.timestamp === undefined ||
+      (typeof value.timestamp === "number" &&
+        Number.isFinite(new Date(value.timestamp).getTime()))) &&
+    hasOptionalStrings(value.$metadata, ["service", "level", "message"]) &&
+    hasOptionalStrings(value.$workers, ["outcome", "scriptName", "executionModel"]) &&
+    hasOptionalStrings(value.source, [
+      "level",
+      "service",
+      "component",
+      "msg",
+      "message",
+      "session_id",
+    ])
+  );
+}
+
+function parseTelemetryQueryResponse(value: unknown): TelemetryQueryResponse {
+  if (!isRecord(value) || typeof value.success !== "boolean") {
+    throw new Error("API response was not a telemetry query result");
+  }
+
+  // An API failure may omit its result; preserve the API's error diagnostics.
+  if (!value.success) {
+    return { success: false, errors: Array.isArray(value.errors) ? value.errors : undefined };
+  }
+  const result = value.result;
+  if (result !== undefined && !isRecord(result)) {
+    throw new Error("API response contained malformed telemetry events");
+  }
+  const eventsEnvelope = result?.events;
+  if (eventsEnvelope !== undefined && !isRecord(eventsEnvelope)) {
+    throw new Error("API response contained malformed telemetry events");
+  }
+  const events = eventsEnvelope?.events;
+  if (events !== undefined && (!Array.isArray(events) || !events.every(isLogEvent))) {
+    throw new Error("API response contained malformed telemetry events");
+  }
+
+  return {
+    success: value.success,
+    errors: Array.isArray(value.errors) ? value.errors : undefined,
+    result: events ? { events: { events } } : undefined,
+  };
 }
 
 // ---------------------------------------------------------------------------
@@ -391,11 +436,7 @@ async function fetchLogs(): Promise<LogEvent[]> {
     throw new Error(`API request failed: ${response.status} ${response.statusText}\n${text}`);
   }
 
-  const data = (await response.json()) as {
-    success: boolean;
-    errors?: unknown[];
-    result?: { events?: { events?: LogEvent[] } };
-  };
+  const data = parseTelemetryQueryResponse(await response.json());
 
   if (!data.success) {
     throw new Error(`API error: ${JSON.stringify(data.errors)}`);

scripts/merge-split-users.test.ts

@@ -7,6 +7,67 @@ function result(results: Record<string, unknown>[], changes = 0): string {
 }
 
 describe("Wrangler user-merge database adapter", () => {
+  for (const malformed of [
+    null,
+    {},
+    [null],
+    [{ success: "true", results: [] }],
+    [{ success: true }],
+    [{ success: true, results: {} }],
+    [{ success: true, results: [{ count: 1 }, null] }],
+    [{ success: true, results: [42] }],
+    [{ success: true, results: [[]] }],
+    [{ success: true, results: [], meta: null }],
+    [{ success: true, results: [], meta: { changes: "1" } }],
+    [{ success: true, results: [], meta: { changes: -1 } }],
+    [{ success: true, results: [], meta: { changes: 1.5 } }],
+  ]) {
+    it(`rejects malformed query data: ${JSON.stringify(malformed)}`, async () => {
+      const database = new WranglerD1Database("workspace", true, false, () => ({
+        status: 0,
+        stderr: "",
+        stdout: JSON.stringify(malformed),
+      }));
+      await assert.rejects(
+        database.prepare("SELECT count(*) AS count FROM users").first(),
+        /malformed/
+      );
+    });
+  }
+
+  it("does not report an omitted verification result as an empty table", async () => {
+    const database = new WranglerD1Database("workspace", true, false, () => ({
+      status: 0,
+      stderr: "",
+      stdout: "[]",
+    }));
+    await assert.rejects(
+      database.prepare("SELECT count(*) AS count FROM users").first(),
+      /returned 0 results/
+    );
+  });
+
+  it("preserves failure diagnostics even when an error has no rows", async () => {
+    const database = new WranglerD1Database("workspace", true, false, () => ({
+      status: 0,
+      stderr: "",
+      stdout: JSON.stringify([{ success: false, error: "query failed" }]),
+    }));
+    await assert.rejects(database.prepare("SELECT 1").first(), /Statement failed:.*query failed/);
+  });
+
+  it("preserves empty rows and optional metadata on successful queries", async () => {
+    const database = new WranglerD1Database("workspace", true, false, () => ({
+      status: 0,
+      stderr: "",
+      stdout: JSON.stringify([{ success: true, results: [] }]),
+    }));
+    assert.deepEqual(await database.prepare("SELECT 1 WHERE 0").all(), {
+      results: [],
+      meta: { changes: 0 },
+    });
+  });
+
   it("uses the result-bearing command batch and preserves positional results", async () => {
     let invokedArgs: string[] = [];
     const runner: WranglerRunner = (args) => {