fix(webapp,core): scope stored packet downloads to runs

Adds a run-scoped route for downloading stored realtime payloads and
outputs, and moves the SDK onto it. Public tokens can no longer create
runs that point at existing stored packets.

Mono-RevId: a39d74877a8b7823333f684e006d510c48e71b0b

Author: carderne

.changeset/tidy-packets-scope.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Authorize stored realtime payload and output downloads against their owning run. Large packet hydration continues to work for run-, task-, tag-, and batch-scoped public tokens without permitting arbitrary packet reads.

apps/webapp/app/routes/api.v1.packets.$.ts

@@ -2,6 +2,7 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { generatePresignedUrl, jsonPacketPresignFailure } from "~/v3/objectStore.server";
 
@@ -51,6 +52,15 @@ export const loader = createLoaderApiRoute(
   async ({ params, authentication }) => {
     const filename = params["*"];
 
+    // Public tokens should use /api/v1/runs/:runId/packets/:field, which authorizes
+    // against the owning run. Access here is kept for clients on older SDKs.
+    if (authentication.type === "PUBLIC_JWT") {
+      logger.info("Legacy packet download with public token", {
+        environmentId: authentication.environment.id,
+        projectId: authentication.environment.projectId,
+      });
+    }
+
     const signed = await generatePresignedUrl(
       authentication.environment.project.externalRef,
       authentication.environment.slug,

apps/webapp/app/routes/api.v1.runs.$runId.packets.$field.ts

@@ -0,0 +1,80 @@
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { $replica } from "~/db.server";
+import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { generatePresignedUrl, jsonPacketPresignFailure } from "~/v3/objectStore.server";
+import { runStore } from "~/v3/runStore.server";
+
+const ParamsSchema = z.object({
+  runId: z.string(),
+  field: z.enum(["payload", "output"]),
+});
+
+const select = {
+  friendlyId: true,
+  taskIdentifier: true,
+  runTags: true,
+  payload: true,
+  payloadType: true,
+  output: true,
+  outputType: true,
+  batch: { select: { friendlyId: true } },
+} as const;
+
+function storedPacketPath(
+  run: { payload: string; payloadType: string; output: string | null; outputType: string },
+  field: "payload" | "output"
+): string | null {
+  const [data, dataType] =
+    field === "payload" ? [run.payload, run.payloadType] : [run.output, run.outputType];
+  return dataType === "application/store" && data ? data : null;
+}
+
+export const loader = createLoaderApiRoute(
+  {
+    params: ParamsSchema,
+    allowJWT: true,
+    corsStrategy: "all",
+    findResource: async (params, authentication) => {
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = { select };
+
+      // The replica may not have the stored output yet for a run that just completed.
+      let run = await runStore.findRun(where, args, $replica);
+      if (!run || !storedPacketPath(run, params.field)) {
+        run = await runStore.findRunOnPrimary(where, args);
+      }
+      if (!run) return null;
+
+      const packetPath = storedPacketPath(run, params.field);
+      return packetPath ? { ...run, packetPath } : null;
+    },
+    authorization: {
+      action: "read",
+      resource: (run) =>
+        anyResource([
+          { type: "runs", id: run.friendlyId },
+          { type: "tasks", id: run.taskIdentifier },
+          ...run.runTags.map((tag) => ({ type: "tags", id: tag })),
+          ...(run.batch ? [{ type: "batch", id: run.batch.friendlyId }] : []),
+        ]),
+    },
+  },
+  async ({ authentication, resource }) => {
+    const signed = await generatePresignedUrl(
+      authentication.environment.project.externalRef,
+      authentication.environment.slug,
+      resource.packetPath,
+      "GET"
+    );
+
+    if (!signed.success) {
+      return jsonPacketPresignFailure(signed);
+    }
+
+    return json({ presignedUrl: signed.url });
+  }
+);

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -63,6 +63,10 @@ const { action, loader } = createActionApiRoute(
     corsStrategy: "all",
   },
   async ({ body, headers, params, authentication, ability }) => {
+    if (authentication.type !== "PRIVATE" && body.options?.payloadType === "application/store") {
+      return json({ error: "Stored payloads require a private API key" }, { status: 403 });
+    }
+
     const {
       "idempotency-key": idempotencyKey,
       "idempotency-key-ttl": idempotencyKeyTTL,

apps/webapp/app/routes/api.v1.tasks.batch.ts

@@ -41,6 +41,13 @@ const { action, loader } = createActionApiRoute(
     corsStrategy: "all",
   },
   async ({ body, headers, params, authentication }) => {
+    if (
+      authentication.type !== "PRIVATE" &&
+      body.items.some((item) => item.options?.payloadType === "application/store")
+    ) {
+      return json({ error: "Stored payloads require a private API key" }, { status: 403 });
+    }
+
     if (!body.items.length) {
       return json({ error: "Batch cannot be triggered with no items" }, { status: 400 });
     }

apps/webapp/app/routes/api.v2.tasks.batch.ts

@@ -49,6 +49,13 @@ const { action, loader } = createActionApiRoute(
     corsStrategy: "all",
   },
   async ({ body, headers, params, authentication, ability }) => {
+    if (
+      authentication.type !== "PRIVATE" &&
+      body.items.some((item) => item.options?.payloadType === "application/store")
+    ) {
+      return json({ error: "Stored payloads require a private API key" }, { status: 403 });
+    }
+
     if (!body.items.length) {
       return json({ error: "Batch cannot be triggered with no items" }, { status: 400 });
     }

apps/webapp/app/routes/api.v3.batches.$batchId.items.ts

@@ -95,7 +95,8 @@ export async function action({ request, params }: ActionFunctionArgs) {
     const itemsIterator = await authorizedBatchItemStream(
       streamToAsyncIterable(parsedStream),
       authResult.ability,
-      batchId
+      batchId,
+      authResult.subject.type !== "publicJWT"
     );
 
     // Process the stream

apps/webapp/app/utils/batchItemAuthorization.ts

@@ -36,9 +36,10 @@ export function batchPublicAccessScopes(
 export async function authorizedBatchItemStream(
   items: AsyncIterable<unknown>,
   ability: RbacAbility,
-  batchId: string
+  batchId: string,
+  allowStoredPayloads = true
 ): Promise<AsyncIterable<unknown>> {
-  const authorized = authorizeBatchItems(items, ability, batchId);
+  const authorized = authorizeBatchItems(items, ability, batchId, allowStoredPayloads);
 
   // A batch-level write grant is authorization in its own right, so an empty
   // stream stays legal for credentials that own the batch: root keys (via the
@@ -77,11 +78,26 @@ export async function authorizedBatchItemStream(
 export async function* authorizeBatchItems(
   items: AsyncIterable<unknown>,
   ability: RbacAbility,
-  batchId: string
+  batchId: string,
+  allowStoredPayloads = true
 ): AsyncIterable<unknown> {
   const canWriteBatch = ability.can("write", { type: "batch", id: batchId });
 
   for await (const item of items) {
+    const payloadType =
+      typeof item === "object" &&
+      item !== null &&
+      "options" in item &&
+      typeof item.options === "object" &&
+      item.options !== null &&
+      "payloadType" in item.options
+        ? item.options.payloadType
+        : undefined;
+
+    if (!allowStoredPayloads && payloadType === "application/store") {
+      throw new BatchItemAuthorizationError();
+    }
+
     const task =
       typeof item === "object" && item !== null && "task" in item && typeof item.task === "string"
         ? item.task

apps/webapp/test/api-auth.e2e.test.ts

@@ -70,6 +70,120 @@ describe("API bearer auth — baseline behavior", () => {
   });
 });
 
+describe("Packet download authorization", () => {
+  // The auth harness has no object store. Reaching its configuration error proves that
+  // resource lookup and authorization passed and the handler attempted to presign.
+  const objectStoreError = {
+    error: "Failed to generate presigned URL: Object store is not configured for protocol: default",
+  };
+
+  async function seedStoredRuns() {
+    const seed = await seedTestEnvironment(server.prisma);
+    const first = await seedTestRun(server.prisma, {
+      environmentId: seed.environment.id,
+      projectId: seed.project.id,
+    });
+    const second = await seedTestRun(server.prisma, {
+      environmentId: seed.environment.id,
+      projectId: seed.project.id,
+    });
+
+    await Promise.all([
+      server.prisma.taskRun.update({
+        where: { id: first.run.id },
+        data: {
+          payload: `${first.runFriendlyId}/payload.json`,
+          payloadType: "application/store",
+          output: `${first.runFriendlyId}-1/output.json`,
+          outputType: "application/store",
+        },
+      }),
+      server.prisma.taskRun.update({
+        where: { id: second.run.id },
+        data: {
+          payload: `${second.runFriendlyId}/payload.json`,
+          payloadType: "application/store",
+        },
+      }),
+    ]);
+
+    return { ...seed, first, second };
+  }
+
+  it("authorizes run packet downloads against the token's run scope", async () => {
+    const { environment, first, second } = await seedStoredRuns();
+    const firstJwt = await generateTestJWT(environment, {
+      scopes: [`read:runs:${first.runFriendlyId}`],
+    });
+    const secondJwt = await generateTestJWT(environment, {
+      scopes: [`read:runs:${second.runFriendlyId}`],
+    });
+    const asFirst = { headers: { Authorization: `Bearer ${firstJwt}` } };
+    const asSecond = { headers: { Authorization: `Bearer ${secondJwt}` } };
+
+    const ownPayload = await server.webapp.fetch(
+      `/api/v1/runs/${first.runFriendlyId}/packets/payload`,
+      asFirst
+    );
+    const ownOutput = await server.webapp.fetch(
+      `/api/v1/runs/${first.runFriendlyId}/packets/output`,
+      asFirst
+    );
+    const otherRun = await server.webapp.fetch(
+      `/api/v1/runs/${second.runFriendlyId}/packets/payload`,
+      asFirst
+    );
+    const inlineOutput = await server.webapp.fetch(
+      `/api/v1/runs/${second.runFriendlyId}/packets/output`,
+      asSecond
+    );
+
+    expect(ownPayload.status).toBe(500);
+    expect(await ownPayload.json()).toEqual(objectStoreError);
+    expect(ownOutput.status).toBe(500);
+    expect(await ownOutput.json()).toEqual(objectStoreError);
+    expect(otherRun.status).toBe(403);
+    expect(inlineOutput.status).toBe(404);
+  });
+
+  it("rejects stored-payload pointers from public JWTs on trigger", async () => {
+    const { environment } = await seedTestEnvironment(server.prisma);
+    const jwt = await generateTestJWT(environment, { scopes: ["write:tasks:test-task"] });
+
+    const res = await server.webapp.fetch("/api/v1/tasks/test-task/trigger", {
+      method: "POST",
+      headers: { Authorization: `Bearer ${jwt}`, "Content-Type": "application/json" },
+      body: JSON.stringify({
+        payload: "run_victim/payload.json",
+        options: { payloadType: "application/store" },
+      }),
+    });
+
+    expect(res.status).toBe(403);
+  });
+
+  it("rejects stored-payload pointers from public JWTs on batch trigger", async () => {
+    const { environment } = await seedTestEnvironment(server.prisma);
+    const jwt = await generateTestJWT(environment, { scopes: ["write:tasks:test-task"] });
+
+    const res = await server.webapp.fetch("/api/v2/tasks/batch", {
+      method: "POST",
+      headers: { Authorization: `Bearer ${jwt}`, "Content-Type": "application/json" },
+      body: JSON.stringify({
+        items: [
+          {
+            task: "test-task",
+            payload: "run_victim/payload.json",
+            options: { payloadType: "application/store" },
+          },
+        ],
+      }),
+    });
+
+    expect(res.status).toBe(403);
+  });
+});
+
 describe("JWT bearer auth — baseline behavior", () => {
   it("valid JWT on JWT-enabled route: auth passes", async () => {
     const { environment } = await seedTestEnvironment(server.prisma);

apps/webapp/test/streamBatchItemsAuthorization.test.ts

@@ -45,6 +45,21 @@ describe("streaming batch item authorization", () => {
     ).resolves.toHaveLength(2);
   });
 
+  it("rejects stored packet pointers from delegated batch credentials", async () => {
+    const ability = withActionAliases(buildJwtAbility(["write:batch:batch_123"]));
+    async function* storedPayloadItem() {
+      yield {
+        task: "task-a",
+        payload: "run_victim/payload.json",
+        options: { payloadType: "application/store" },
+      };
+    }
+
+    await expect(
+      collect(authorizeBatchItems(storedPayloadItem(), ability, "batch_123", false))
+    ).rejects.toThrow();
+  });
+
   it("does not delegate batch-wide writes from selected-task credentials", () => {
     const ability = withActionAliases(buildJwtAbility(["batchTrigger:tasks:task-a"]));