fix(webapp): harden Electric sync routes against SQLi and add org scoping (#28)

* fix(webapp): harden Electric sync routes against SQL injection and cross-tenant leak

sync.traces.\$traceId.ts and sync.traces.runs.\$traceId.ts interpolated
params.traceId raw into the Electric `where` clause, and that clause
had no organizationId predicate. An authenticated user can craft a
traceparent header so the runtime SDK persists a malicious traceId
into their own org's row, request /sync/traces/<crafted-id>, pass the
membership check on that very row, and then watch Electric stream
rows from any tenant whose traceId matches the crafted SQL.

Closes the vector at four layers:

- OtelTraceIdSchema validates the canonical 32-lowercase-hex format
  at the Zod layer; non-conforming input 404s with no error echo.
- buildElectricTraceWhereClause emits
  `"traceId"='…' AND "organizationId"='…'` so cross-tenant collision
  doesn't matter even if input validation is ever bypassed. Both
  inputs are re-checked at the function boundary.
- RESERVED_ELECTRIC_SHAPE_PARAMS strips `where`/`table`/`columns` from
  the incoming searchParams before forwarding to Electric so callers
  can't override the server-set values.
- validateRealtimeTags + a matching Zod allowlist in realtime.v1.runs.ts
  reject tag values that could break out of the
  `"runTags" @> ARRAY['<tag>',…]` SQL string and steer the
  jumpHash-derived shard key.

Closes TRI-9903, TRI-9904, TRI-9905, TRI-9906 (4 P0). Also closes
TRI-9913 (Shape param allowlist) and TRI-9985 (realtime tag
interpolation).

19 unit tests cover the regex, the where-clause builder, the reserved-
param set, and the tag allowlist.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(webapp): regression tests for Electric sync SQLi + scoping hardening

Centralizes the hardening in a pure electricShape.server module: OtelTraceIdSchema
(32-hex), buildElectricTraceWhereClause (org-scoped, validated), and realtime tag
sanitization (reject unsafe chars, escape quotes), plus reserved-param stripping.
Pure test (23 cases). Verified RED with the trace-id validation and tag sanitizer
neutered (12 injection tests flip), GREEN with them. Bundles the streamBatchItems
timeout bump.

* fix(webapp): scope TaskRun trace sync by projectId, not nullable organizationId

TaskRun.organizationId is nullable and the table is far too large to backfill,
so AND-scoping the trace-runs Electric shape by organizationId silently dropped
legacy NULL-org runs from the trace view. Scope by the non-null projectId
instead: buildElectricTraceWhereClause now takes a typed tenant column (org for
the TaskEvent shape, project for TaskRun). Tenant-safe — membership is verified
against the project's org and a trace's runs share one project, and the column
is a fixed union, never user input.

* docs: reword server-changes note to comply with release-note conventions

* chore(webapp): remove electricShape test and shorten comments

* fix(webapp): preserve project scoping after primary trace lookup

* widen test

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Chris Arderne <chris@trigger.dev>

Author: 3 people

.server-changes/tighten-realtime-and-trace-sync.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Live run and trace subscriptions now validate their identifiers more strictly and only return data from your own organization.

apps/webapp/app/routes/realtime.v1.runs.ts

@@ -2,13 +2,28 @@ import { z } from "zod";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { resolveRealtimeStreamClient } from "~/services/realtime/resolveRealtimeStreamClient.server";
 import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { UNSAFE_REALTIME_TAG_CHARS } from "~/v3/electricShape.server";
 
 const SearchParamsSchema = z.object({
   tags: z
     .string()
     .optional()
     .transform((value) => {
       return value ? value.split(",") : undefined;
+    })
+    .superRefine((tags, ctx) => {
+      if (!tags) return;
+      for (const tag of tags) {
+        // Mirror the runtime sanitiser's reject list so the API returns 400
+        // instead of a 500. Single quotes are allowed — escaped downstream.
+        if (UNSAFE_REALTIME_TAG_CHARS.test(tag) || tag.length === 0) {
+          ctx.addIssue({
+            code: z.ZodIssueCode.custom,
+            message: `Invalid tag: ${JSON.stringify(tag)}`,
+          });
+          return;
+        }
+      }
     }),
   createdAt: z.string().optional(),
 });

apps/webapp/app/routes/sync.traces.$traceId.ts

@@ -1,15 +1,33 @@
 import type { LoaderFunctionArgs } from "@remix-run/node";
+import { z } from "zod";
 import { $replica } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { getUserId } from "~/services/session.server";
 import { longPollingFetch } from "~/utils/longPollingFetch";
+import {
+  OtelTraceIdSchema,
+  RESERVED_ELECTRIC_SHAPE_PARAMS,
+  buildElectricTraceWhereClause,
+} from "~/v3/electricShape.server";
+
+const Params = z.object({
+  traceId: OtelTraceIdSchema,
+});
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   try {
     const userId = await getUserId(request);
 
-    logger.log(`/sync/traces/${params.traceId}`, { userId });
+    const parsedParams = Params.safeParse(params);
+    if (!parsedParams.success) {
+      // Treat a malformed traceId as not-found rather than 400 to avoid
+      // signalling the validator.
+      return new Response("Not found", { status: 404 });
+    }
+    const { traceId } = parsedParams.data;
+
+    logger.log(`/sync/traces/${traceId}`, { userId });
 
     if (!userId) {
       return new Response("No user found in cookie", { status: 401 });
@@ -20,7 +38,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
         organizationId: true,
       },
       where: {
-        traceId: params.traceId,
+        traceId,
       },
     });
 
@@ -41,11 +59,19 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
 
     const url = new URL(request.url);
     const originUrl = new URL(`${env.ELECTRIC_ORIGIN}/v1/shape/public."TaskEvent"`);
+    // Strip params we set ourselves so the caller can't override them.
     url.searchParams.forEach((value, key) => {
+      if (RESERVED_ELECTRIC_SHAPE_PARAMS.has(key)) return;
       originUrl.searchParams.set(key, value);
     });
 
-    originUrl.searchParams.set("where", `"traceId"='${params.traceId}'`);
+    originUrl.searchParams.set(
+      "where",
+      buildElectricTraceWhereClause({
+        traceId,
+        scope: { column: "organizationId", id: trace.organizationId },
+      })
+    );
 
     const finalUrl = originUrl.toString();
 

apps/webapp/app/routes/sync.traces.runs.$traceId.ts

@@ -5,17 +5,27 @@ import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { getUserId } from "~/services/session.server";
 import { longPollingFetch } from "~/utils/longPollingFetch";
-import { runStore } from "~/v3/runStore.server";
+import {
+  OtelTraceIdSchema,
+  RESERVED_ELECTRIC_SHAPE_PARAMS,
+  buildElectricTraceWhereClause,
+} from "~/v3/electricShape.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { runStore } from "~/v3/runStore.server";
 
 const Params = z.object({
-  traceId: z.string(),
+  traceId: OtelTraceIdSchema,
 });
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   try {
     const userId = await getUserId(request);
-    const { traceId } = Params.parse(params);
+
+    const parsedParams = Params.safeParse(params);
+    if (!parsedParams.success) {
+      return new Response("Not found", { status: 404 });
+    }
+    const { traceId } = parsedParams.data;
 
     logger.log(`/sync/runs/${traceId}`, { userId });
 
@@ -29,6 +39,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
       },
       {
         select: {
+          projectId: true,
           runtimeEnvironmentId: true,
         },
       },
@@ -40,7 +51,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
       // primary before 404ing so a live run's realtime trace feed isn't spuriously not-found.
       run = await runStore.findRunOnPrimary(
         { traceId },
-        { select: { runtimeEnvironmentId: true } }
+        { select: { projectId: true, runtimeEnvironmentId: true } }
       );
     }
 
@@ -67,11 +78,22 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
 
     const url = new URL(request.url);
     const originUrl = new URL(`${env.ELECTRIC_ORIGIN}/v1/shape/public."TaskRun"`);
+    // Strip params we set ourselves so the caller can't override them.
     url.searchParams.forEach((value, key) => {
+      if (RESERVED_ELECTRIC_SHAPE_PARAMS.has(key)) return;
       originUrl.searchParams.set(key, value);
     });
 
-    originUrl.searchParams.set("where", `"traceId"='${traceId}'`);
+    originUrl.searchParams.set(
+      "where",
+      // Scope by non-null projectId, not the nullable organizationId (legacy
+      // rows would vanish). Tenant-safe: membership was verified against this
+      // project's org and a trace's runs all live in one project.
+      buildElectricTraceWhereClause({
+        traceId,
+        scope: { column: "projectId", id: run.projectId },
+      })
+    );
 
     const finalUrl = originUrl.toString();
 

apps/webapp/app/services/realtimeClient.server.ts

@@ -15,6 +15,7 @@ import { RedisCacheStore } from "./unkey/redisCacheStore.server";
 import { env } from "~/env.server";
 import type { API_VERSIONS } from "~/api/versions";
 import { CURRENT_API_VERSION } from "~/api/versions";
+import { sanitizeRealtimeTagsForSql } from "~/v3/electricShape.server";
 
 export interface CachedLimitProvider {
   getCachedLimit: (organizationId: string, defaultValue: number) => Promise<number | undefined>;
@@ -171,7 +172,10 @@ export class RealtimeClient {
     const whereClauses: string[] = [`"runtimeEnvironmentId"='${environment.id}'`];
 
     if (params.tags) {
-      whereClauses.push(`"runTags" @> ARRAY[${params.tags.map((t) => `'${t}'`).join(",")}]`);
+      // Reject unsafe chars and escape single quotes so tag values can't
+      // break out of the Electric SQL string literal.
+      const safeTags = sanitizeRealtimeTagsForSql(params.tags);
+      whereClauses.push(`"runTags" @> ARRAY[${safeTags.map((t) => `'${t}'`).join(",")}]`);
     }
 
     const createdAtFilter = await this.#calculateCreatedAtFilter(url, params.createdAt);

apps/webapp/app/v3/electricShape.server.ts

@@ -0,0 +1,69 @@
+import { z } from "zod";
+
+/**
+ * OTel trace IDs are 32 lowercase hex chars. The traceparent parser only
+ * checks the dash-delimited format, so crafted ids can be persisted and later
+ * interpolated into shape `where` clauses. Validate here to close the SQLi vector.
+ */
+export const OtelTraceIdSchema = z
+  .string()
+  .regex(/^[0-9a-f]{32}$/, "traceId must be 32 lowercase hex characters");
+
+/** Params the sync routes set themselves; stripped from incoming requests. */
+export const RESERVED_ELECTRIC_SHAPE_PARAMS = new Set(["where", "table", "columns"]);
+
+const CUID_LIKE = /^[a-z][a-z0-9_]*$/i;
+
+/**
+ * Tenant column a trace shape is scoped by. TaskEvent scopes by non-null
+ * organizationId; TaskRun scopes by non-null projectId (its organizationId is
+ * nullable). The column is from this fixed union, never user input, so it's
+ * safe to interpolate.
+ */
+export type TraceScope =
+  | { column: "organizationId"; id: string }
+  | { column: "projectId"; id: string };
+
+/**
+ * Build the Electric Shape `where` clause for the trace sync routes. Both ids
+ * are re-validated as defense-in-depth so a missed call site can't bypass scope.
+ */
+export function buildElectricTraceWhereClause(args: {
+  traceId: string;
+  scope: TraceScope;
+}): string {
+  const { traceId, scope } = args;
+  if (!OtelTraceIdSchema.safeParse(traceId).success) {
+    throw new Error("buildElectricTraceWhereClause: unsafe traceId");
+  }
+  if (!CUID_LIKE.test(scope.id)) {
+    throw new Error("buildElectricTraceWhereClause: unsafe scope id");
+  }
+  return `"traceId"='${traceId}' AND "${scope.column}"='${scope.id}'`;
+}
+
+/**
+ * Characters rejected in realtime tag values — the single source of truth
+ * shared by the apiBuilder Zod refine (`realtime.v1.runs.ts`) and the runtime
+ * sanitiser. Rejects control chars/DEL, backslash, and double-quote. Single
+ * quotes are allowed and escaped (`'` → `''`) in `sanitizeRealtimeTagForSql`.
+ */
+export const UNSAFE_REALTIME_TAG_CHARS = /[\x00-\x1f\x7f\\"]/;
+
+/**
+ * Sanitise a tag value for interpolation into an Electric Shape `where` clause:
+ * reject unsafe chars, escape single quotes per SQL standard.
+ */
+export function sanitizeRealtimeTagForSql(tag: string): string {
+  if (typeof tag !== "string" || tag.length === 0) {
+    throw new Error("Invalid realtime tag: empty");
+  }
+  if (UNSAFE_REALTIME_TAG_CHARS.test(tag)) {
+    throw new Error(`Invalid realtime tag: ${JSON.stringify(tag)} — contains unsafe character`);
+  }
+  return tag.replace(/'/g, "''");
+}
+
+export function sanitizeRealtimeTagsForSql(tags: string[]): string[] {
+  return tags.map(sanitizeRealtimeTagForSql);
+}

apps/webapp/test/spanTraceRoutes.replicaLag.test.ts

@@ -475,7 +475,7 @@ describe("run-trace/span-detail route loaders under a lagging replica", () => {
       const seed = await seedTenant(prisma14, suffix);
       const runId = `run_${CUID_25}`;
       const friendlyId = `run_${suffix}`;
-      const traceId = `trace_${suffix}`;
+      const traceId = "a".repeat(32);
       const userId = `user_${suffix}`;
 
       // The dashboard user, joined to the org so the route's real orgMember check passes.
@@ -538,7 +538,8 @@ describe("run-trace/span-detail route loaders under a lagging replica", () => {
       holder.resolvedEnv = { organizationId: seed.organization.id };
       holder.replicaMarker = { orgMember: prisma14.orgMember };
 
-      const res = (await syncTraceRunsLoader(syncRequest("trace_does_not_exist"))) as Response;
+      const res = (await syncTraceRunsLoader(syncRequest("b".repeat(32)))) as Response;
+      expect(lagged.legacyReplica.wasHit("taskRun")).toBe(true);
       expect(res.status).toBe(404);
     }
   );