fix(webapp): let scoped credentials trigger with SDK-uploaded stored payloads

Triggering a task with a payload over the offload threshold from inside a running task failed with `Stored payloads require an unrestricted private API key` on SDK 4.8.2 and later. Run tokens and restricted API keys can now submit offloaded payloads. Unrestricted keys are unchanged.

Mono-RevId: 4d4b916ebb9890bf5db4aa842eeb1d45d22fd591

Author: matt-aitken

.server-changes/scoped-credentials-offloaded-trigger-payloads.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Triggering a task with a payload over the offload threshold from inside a running task works again. Run-scoped tokens and restricted API keys upload offloaded payloads into their own namespace and may only trigger with payloads they uploaded, while pointers at other runs' stored payloads or outputs are still rejected.

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

@@ -7,6 +7,10 @@ import { canWriteParentRun } from "~/utils/parentRunAuthorization.server";
 import { clientSafeErrorMessage } from "~/utils/prismaErrors";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
+import {
+  STORED_PAYLOAD_FORBIDDEN_ERROR,
+  storedPayloadPointerPolicy,
+} from "~/services/packetApi.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { BatchTriggerV3Service } from "~/v3/services/batchTriggerV3.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
@@ -29,15 +33,15 @@ const { action } = createActionApiRoute(
       resource: (params) => ({ type: "tasks", id: params.taskId }),
     },
   },
-  async ({ params, headers, body, authentication, restrictedApiKey, ability }) => {
+  async ({ params, headers, body, authentication, subject, ability }) => {
+    const allowStoredPayload = storedPayloadPointerPolicy(subject);
     if (
-      (authentication.type !== "PRIVATE" || restrictedApiKey) &&
-      body.items.some((item) => item.options?.payloadType === "application/store")
+      body.items.some(
+        (item) =>
+          item.options?.payloadType === "application/store" && !allowStoredPayload(item.payload)
+      )
     ) {
-      return json(
-        { error: "Stored payloads require an unrestricted private API key" },
-        { status: 403 }
-      );
+      return json({ error: STORED_PAYLOAD_FORBIDDEN_ERROR }, { status: 403 });
     }
 
     const { taskId } = params;

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

@@ -14,6 +14,10 @@ import { getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server";
 import { determineRealtimeStreamsVersion } from "~/services/realtime/v1StreamsGlobal.server";
+import {
+  STORED_PAYLOAD_FORBIDDEN_ERROR,
+  storedPayloadPointerPolicy,
+} from "~/services/packetApi.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import {
@@ -62,15 +66,10 @@ const { action, loader } = createActionApiRoute(
     },
     corsStrategy: "all",
   },
-  async ({ body, headers, params, authentication, restrictedApiKey, ability }) => {
-    if (
-      (authentication.type !== "PRIVATE" || restrictedApiKey) &&
-      body.options?.payloadType === "application/store"
-    ) {
-      return json(
-        { error: "Stored payloads require an unrestricted private API key" },
-        { status: 403 }
-      );
+  async ({ body, headers, params, authentication, subject, ability }) => {
+    const allowStoredPayload = storedPayloadPointerPolicy(subject);
+    if (body.options?.payloadType === "application/store" && !allowStoredPayload(body.payload)) {
+      return json({ error: STORED_PAYLOAD_FORBIDDEN_ERROR }, { status: 403 });
     }
 
     const {

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

@@ -3,6 +3,10 @@ import { BatchTriggerTaskV2RequestBody } from "@trigger.dev/core/v3";
 import { env } from "~/env.server";
 import { getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import {
+  STORED_PAYLOAD_FORBIDDEN_ERROR,
+  storedPayloadPointerPolicy,
+} from "~/services/packetApi.server";
 import { createActionApiRoute, everyResource } from "~/services/routeBuilders/apiBuilder.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
@@ -40,15 +44,15 @@ const { action, loader } = createActionApiRoute(
     },
     corsStrategy: "all",
   },
-  async ({ body, headers, params, authentication, restrictedApiKey }) => {
+  async ({ body, headers, params, authentication, subject }) => {
+    const allowStoredPayload = storedPayloadPointerPolicy(subject);
     if (
-      (authentication.type !== "PRIVATE" || restrictedApiKey) &&
-      body.items.some((item) => item.options?.payloadType === "application/store")
+      body.items.some(
+        (item) =>
+          item.options?.payloadType === "application/store" && !allowStoredPayload(item.payload)
+      )
     ) {
-      return json(
-        { error: "Stored payloads require an unrestricted private API key" },
-        { status: 403 }
-      );
+      return json({ error: STORED_PAYLOAD_FORBIDDEN_ERROR }, { status: 403 });
     }
 
     if (!body.items.length) {

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

@@ -8,6 +8,10 @@ import { RunEngineBatchTriggerService } from "~/runEngine/services/batchTrigger.
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import {
+  STORED_PAYLOAD_FORBIDDEN_ERROR,
+  storedPayloadPointerPolicy,
+} from "~/services/packetApi.server";
 import { createActionApiRoute, everyResource } from "~/services/routeBuilders/apiBuilder.server";
 import {
   handleRequestIdempotency,
@@ -48,15 +52,15 @@ const { action, loader } = createActionApiRoute(
     },
     corsStrategy: "all",
   },
-  async ({ body, headers, params, authentication, restrictedApiKey, ability }) => {
+  async ({ body, headers, params, authentication, subject, ability }) => {
+    const allowStoredPayload = storedPayloadPointerPolicy(subject);
     if (
-      (authentication.type !== "PRIVATE" || restrictedApiKey) &&
-      body.items.some((item) => item.options?.payloadType === "application/store")
+      body.items.some(
+        (item) =>
+          item.options?.payloadType === "application/store" && !allowStoredPayload(item.payload)
+      )
     ) {
-      return json(
-        { error: "Stored payloads require an unrestricted private API key" },
-        { status: 403 }
-      );
+      return json({ error: STORED_PAYLOAD_FORBIDDEN_ERROR }, { status: 403 });
     }
 
     if (!body.items.length) {

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

@@ -13,6 +13,7 @@ import {
   BatchItemAuthorizationError,
 } from "~/utils/batchItemAuthorization";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { storedPayloadPointerPolicy } from "~/services/packetApi.server";
 
 const ParamsSchema = z.object({
   batchId: z.string(),
@@ -72,8 +73,6 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: authResult.error }, { status: authResult.status });
   }
 
-  const restrictedApiKey = authResult.subject.type === "apiKey" && authResult.subject.restricted;
-
   // Get the request body stream
   const body = request.body;
   if (!body) {
@@ -100,7 +99,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
       streamToAsyncIterable(parsedStream),
       authResult.ability,
       batchId,
-      authResult.subject.type !== "publicJWT" && !restrictedApiKey
+      storedPayloadPointerPolicy(authResult.subject)
     );
 
     // Process the stream

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

@@ -1,8 +1,41 @@
 import { json } from "@remix-run/server-runtime";
+import type { RbacSubject } from "@trigger.dev/plugins";
 import { z } from "zod";
 import type { ApiAuthenticationResultSuccess } from "~/services/apiAuth.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { generatePresignedUrl, jsonPacketPresignFailure } from "~/v3/objectStore.server";
+import {
+  formatStorageUri,
+  generatePresignedUrl,
+  isPacketPathUnder,
+  jsonPacketPresignFailure,
+  parseStorageUri,
+} from "~/v3/objectStore.server";
+
+export const STORED_PAYLOAD_FORBIDDEN_ERROR =
+  "Stored payloads require an unrestricted private API key or a payload this credential uploaded";
+
+/**
+ * Object-store prefix a scoped credential's packet uploads (offloaded trigger payloads and task
+ * outputs alike) land under, so a pointer proves who uploaded it. Unrestricted credentials upload
+ * wherever they ask and get undefined.
+ */
+function scopedPacketNamespace(subject: RbacSubject): string | undefined {
+  if (subject.type === "runToken") return `uploads/run/${subject.runId}`;
+  if (subject.type === "apiKey" && subject.restricted) return `uploads/key/${subject.apiKeyId}`;
+  return undefined;
+}
+
+/**
+ * Which stored-payload pointers a credential may submit when triggering. A scoped credential
+ * (run token, restricted key) may only reference its own uploads, so it cannot create a run
+ * that reads another run's payload or output, nor replay a pointer it learned elsewhere.
+ */
+export function storedPayloadPointerPolicy(subject: RbacSubject): (pointer: unknown) => boolean {
+  if (subject.type === "publicJWT") return () => false;
+  const namespace = scopedPacketNamespace(subject);
+  if (!namespace) return () => true;
+  return (pointer) => isPacketPathUnder(pointer, namespace);
+}
 
 export const PacketParamsSchema = z.object({
   "*": z.string(),
@@ -32,12 +65,22 @@ export function createPacketUploadRoute({ forceNoPrefix = false } = {}) {
         resource: () => ({ type: "packets" }),
       },
     },
-    async ({ params, authentication }) => {
+    async ({ params, authentication, subject }) => {
       assertPacketNamespaceAccess(authentication);
+      const { protocol, path } = parseStorageUri(params["*"]);
+      const namespace = scopedPacketNamespace(subject);
+      // The legacy route returns no storagePath, so a relocated upload could never be referenced.
+      if (forceNoPrefix && namespace) {
+        return json(
+          { error: "Scoped credentials must upload packets via /api/v2/packets; upgrade the SDK" },
+          { status: 400 }
+        );
+      }
+      const filename = formatStorageUri(namespace ? `${namespace}/${path}` : path, protocol);
       const signed = await generatePresignedUrl(
         authentication.environment.project.externalRef,
         authentication.environment.slug,
-        params["*"],
+        filename,
         "PUT",
         { forceNoPrefix }
       );

apps/webapp/app/services/routeBuilders/apiBuilder.server.ts

@@ -11,6 +11,7 @@ import { logger } from "../logger.server";
 import { rbac } from "../rbac.server";
 import { authenticateBearerWithTelemetry } from "~/services/authTelemetry.server";
 import type { RbacAbility, RbacResource } from "@trigger.dev/rbac";
+import type { RbacSubject } from "@trigger.dev/plugins";
 import { isUserActorToken } from "@trigger.dev/rbac";
 import {
   resolveAndRecheckUserActorClaims,
@@ -114,6 +115,7 @@ async function authenticateRequestForApiBuilder(
       authentication: ApiAuthenticationResultSuccess;
       ability: RbacAbility;
       restrictedApiKey: boolean;
+      subject: RbacSubject;
     }
 > {
   const result = await authenticateBearerWithTelemetry(request, { allowJWT });
@@ -151,6 +153,7 @@ async function authenticateRequestForApiBuilder(
     authentication,
     ability: result.ability,
     restrictedApiKey,
+    subject: result.subject,
   };
 }
 
@@ -1213,6 +1216,7 @@ type ApiKeyActionHandlerFunction<
     : undefined;
   authentication: ApiAuthenticationResultSuccess;
   restrictedApiKey: boolean;
+  subject: RbacSubject;
   ability: RbacAbility;
   request: Request;
   resource?: TResource;
@@ -1451,6 +1455,7 @@ export function createActionApiRoute<
             body: parsedBody,
             authentication: authenticationResult,
             restrictedApiKey: authResult.restrictedApiKey,
+            subject: authResult.subject,
             ability,
             request,
             resource,

apps/webapp/app/utils/batchItemAuthorization.ts

@@ -37,9 +37,9 @@ export async function authorizedBatchItemStream(
   items: AsyncIterable<unknown>,
   ability: RbacAbility,
   batchId: string,
-  allowStoredPayloads = true
+  allowStoredPayload: (pointer: unknown) => boolean = () => true
 ): Promise<AsyncIterable<unknown>> {
-  const authorized = authorizeBatchItems(items, ability, batchId, allowStoredPayloads);
+  const authorized = authorizeBatchItems(items, ability, batchId, allowStoredPayload);
 
   // 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
@@ -79,7 +79,7 @@ export async function* authorizeBatchItems(
   items: AsyncIterable<unknown>,
   ability: RbacAbility,
   batchId: string,
-  allowStoredPayloads = true
+  allowStoredPayload: (pointer: unknown) => boolean = () => true
 ): AsyncIterable<unknown> {
   const canWriteBatch = ability.can("write", { type: "batch", id: batchId });
 
@@ -94,7 +94,10 @@ export async function* authorizeBatchItems(
         ? item.options.payloadType
         : undefined;
 
-    if (!allowStoredPayloads && payloadType === "application/store") {
+    const payload =
+      typeof item === "object" && item !== null && "payload" in item ? item.payload : undefined;
+
+    if (payloadType === "application/store" && !allowStoredPayload(payload)) {
       throw new BatchItemAuthorizationError();
     }
 

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

@@ -57,6 +57,16 @@ export function formatStorageUri(path: string, protocol?: string): string {
 
 export const INVALID_PACKET_STORAGE_PATH = "Invalid packet storage path";
 
+/** True when a stored-payload pointer, once normalised, sits under `prefix/`. Fails closed. */
+export function isPacketPathUnder(pointer: unknown, prefix: string): boolean {
+  if (typeof pointer !== "string") return false;
+  try {
+    return resolveSafePacketRelativePath(parseStorageUri(pointer).path).startsWith(`${prefix}/`);
+  } catch {
+    return false;
+  }
+}
+
 export type PacketPresignFailure = {
   success: false;
   error: string;