fix: security release 2026-08-12 (#4735)

2026-08-20 12:34:33 +01:00
parent 518978bc52
commit 06f99aeb31
30 changed files with 1238 additions and 72 deletions
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Reject alert webhook destinations in reserved benchmarking IP ranges.
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+TRQL queries using the PREWHERE clause are now rejected with a clear error message. Use WHERE instead, which is filtered the same way but keeps your data isolation guarantees intact.
@@ -9,10 +9,15 @@ import { z } from "zod";
 import { $transaction, prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
+import { redirectWithErrorMessage } from "./message.server";
 import { slackSecretLogFields } from "./safeIntegrationLog";
 import { slackAccessResultLogFields } from "./slackOAuthResultLog";
 import { getSecretStore } from "~/services/secrets/secretStore.server";
-import { commitSession, getUserSession } from "~/services/sessionStorage.server";
+import {
+  clearSlackOAuthSessionBinding,
+  consumeSlackOAuthStateForSession,
+  createSlackOAuthStateForSession,
+} from "~/models/slackOAuthState.server";
 import { generateFriendlyId } from "~/v3/friendlyIdentifiers";

 const SlackSecretSchema = z.object({
@@ -27,8 +32,6 @@ const SlackSecretSchema = z.object({

 type SlackSecret = z.infer<typeof SlackSecretSchema>;

-const REDIRECT_AFTER_AUTH_KEY = "redirect-back-after-auth";
-
 export type OrganizationIntegrationForService<TService extends IntegrationService> = Omit<
  AuthenticatableIntegration,
  "service"
@@ -138,22 +141,26 @@ export class OrgIntegrationRepository {

  static async redirectToAuthService(
    service: IntegrationService,
-    state: string,
+    organizationId: string,
+    userId: string,
    request: Request,
    redirectTo: string
  ) {
-    const session = await getUserSession(request);
-    session.set(REDIRECT_AFTER_AUTH_KEY, redirectTo);
-
-    const authUrl = service === "SLACK" ? this.slackAuthorizationUrl(state) : undefined;
-
-    if (!authUrl) {
+    if (service !== "SLACK") {
      throw new Response("Unsupported service", { status: 400 });
    }

+    const { nonce, sessionCookie } = await createSlackOAuthStateForSession(request, {
+      userId,
+      organizationId,
+      service: "slack",
+      redirectTo,
+    });
+
+    const authUrl = this.slackAuthorizationUrl(nonce);
+
    logger.debug("Redirecting to auth service", {
      service,
-      authUrl,
      redirectTo,
    });

@@ -161,35 +168,33 @@ export class OrgIntegrationRepository {
      status: 302,
      headers: {
        location: authUrl,
-        "Set-Cookie": await commitSession(session),
+        "Set-Cookie": sessionCookie,
      },
    });
  }

-  static async redirectAfterAuth(request: Request) {
-    const session = await getUserSession(request);
+  static async redirectAfterAuth(request: Request, redirectTo: string, errorMessage?: string) {
+    const sessionCookie = await clearSlackOAuthSessionBinding(request);

-    logger.debug("Redirecting back after auth", {
-      sessionData: session.data,
-    });
-
-    const redirectTo = session.get(REDIRECT_AFTER_AUTH_KEY);
-
-    if (!redirectTo) {
-      throw new Response("Invalid redirect", { status: 400 });
+    if (errorMessage) {
+      const response = await redirectWithErrorMessage(redirectTo, request, errorMessage);
+      response.headers.append("Set-Cookie", sessionCookie);
+      return response;
    }

-    session.unset(REDIRECT_AFTER_AUTH_KEY);
-
    return new Response(null, {
      status: 302,
      headers: {
        location: redirectTo,
-        "Set-Cookie": await commitSession(session),
+        "Set-Cookie": sessionCookie,
      },
    });
  }

+  static async consumeSlackOAuthState(request: Request, state: string, userId: string) {
+    return consumeSlackOAuthStateForSession(request, state, userId);
+  }
+
  static async createOrgIntegration(serviceName: string, code: string, org: Organization) {
    switch (serviceName) {
      case "slack": {
@@ -0,0 +1,139 @@
+import { randomBytes } from "node:crypto";
+import { z } from "zod";
+import { env } from "~/env.server";
+import { createRedisClient, type RedisClient } from "~/redis.server";
+import { commitSession, getUserSession } from "~/services/sessionStorage.server";
+import { singleton } from "~/utils/singleton";
+
+const STATE_TTL_SECONDS = 10 * 60;
+const CREATE_ATTEMPTS = 2;
+const KEY_PREFIX = "oauth:slack:state:";
+const SLACK_OAUTH_SESSION_BINDING_KEY = "slack-oauth-session-binding";
+
+const SlackOAuthStateSchema = z.object({
+  userId: z.string(),
+  sessionBinding: z.string(),
+  organizationId: z.string(),
+  service: z.literal("slack"),
+  redirectTo: z.string().regex(/^\/(?!\/)/),
+});
+
+export type SlackOAuthState = z.infer<typeof SlackOAuthStateSchema>;
+
+type CreateSlackOAuthState = SlackOAuthState;
+type StartSlackOAuthState = Omit<CreateSlackOAuthState, "sessionBinding">;
+type ConsumeSlackOAuthState = Pick<SlackOAuthState, "userId" | "sessionBinding" | "service">;
+
+const consumeScript = `
+local raw = redis.call("GET", KEYS[1])
+if not raw then return nil end
+local decoded, state = pcall(cjson.decode, raw)
+if not decoded or type(state) ~= "table" then return nil end
+if state.userId ~= ARGV[1] or state.sessionBinding ~= ARGV[2] or state.service ~= ARGV[3] then
+  return nil
+end
+redis.call("DEL", KEYS[1])
+return raw
+`;
+
+export class SlackOAuthStateStore {
+  constructor(private readonly redis: Pick<RedisClient, "set" | "eval">) {}
+
+  async create(state: CreateSlackOAuthState): Promise<string> {
+    const parsedState = SlackOAuthStateSchema.parse(state);
+
+    for (let attempt = 0; attempt < CREATE_ATTEMPTS; attempt++) {
+      const nonce = randomBytes(32).toString("base64url");
+      const created = await this.redis.set(
+        this.#key(nonce),
+        JSON.stringify(parsedState),
+        "EX",
+        STATE_TTL_SECONDS,
+        "NX"
+      );
+      if (created === "OK") return nonce;
+    }
+
+    throw new Error("Failed to create a unique Slack OAuth state");
+  }
+
+  async consume(
+    nonce: string,
+    expected: ConsumeSlackOAuthState
+  ): Promise<SlackOAuthState | undefined> {
+    if (!/^[A-Za-z0-9_-]{43}$/.test(nonce)) return undefined;
+
+    const raw = await this.redis.eval(
+      consumeScript,
+      1,
+      this.#key(nonce),
+      expected.userId,
+      expected.sessionBinding,
+      expected.service
+    );
+    if (typeof raw !== "string") return undefined;
+
+    try {
+      return SlackOAuthStateSchema.safeParse(JSON.parse(raw)).data;
+    } catch {
+      return undefined;
+    }
+  }
+
+  #key(nonce: string): string {
+    return `${KEY_PREFIX}{${nonce}}`;
+  }
+}
+
+export async function createSlackOAuthStateForSession(
+  request: Request,
+  state: StartSlackOAuthState,
+  stateStore: SlackOAuthStateStore = getSlackOAuthStateStore()
+): Promise<{ nonce: string; sessionCookie: string }> {
+  const session = await getUserSession(request);
+  const sessionBinding = randomBytes(32).toString("base64url");
+  const nonce = await stateStore.create({ ...state, sessionBinding });
+  session.set(SLACK_OAUTH_SESSION_BINDING_KEY, sessionBinding);
+
+  return { nonce, sessionCookie: await commitSession(session) };
+}
+
+export async function consumeSlackOAuthStateForSession(
+  request: Request,
+  nonce: string,
+  userId: string,
+  stateStore: SlackOAuthStateStore = getSlackOAuthStateStore()
+): Promise<SlackOAuthState | undefined> {
+  const session = await getUserSession(request);
+  const sessionBinding = session.get(SLACK_OAUTH_SESSION_BINDING_KEY);
+  if (typeof sessionBinding !== "string") return undefined;
+
+  return stateStore.consume(nonce, { userId, sessionBinding, service: "slack" });
+}
+
+export async function clearSlackOAuthSessionBinding(request: Request): Promise<string> {
+  const session = await getUserSession(request);
+  session.unset(SLACK_OAUTH_SESSION_BINDING_KEY);
+  return commitSession(session);
+}
+
+function getSlackOAuthStateStore(): SlackOAuthStateStore {
+  if (!env.CACHE_REDIS_HOST) {
+    throw new Error("Cache Redis is required for Slack OAuth state");
+  }
+
+  return singleton(
+    "slackOAuthStateStore",
+    () =>
+      new SlackOAuthStateStore(
+        createRedisClient("trigger:slack-oauth-state", {
+          host: env.CACHE_REDIS_HOST,
+          port: env.CACHE_REDIS_PORT,
+          username: env.CACHE_REDIS_USERNAME,
+          password: env.CACHE_REDIS_PASSWORD,
+          tlsDisabled: env.CACHE_REDIS_TLS_DISABLED === "true",
+          clusterMode: env.CACHE_REDIS_CLUSTER_MODE_ENABLED === "1",
+        })
+      )
+  );
+}
@@ -28,6 +28,7 @@ import { $replica } from "~/db.server";
 import { env } from "~/env.server";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { inviteMembers } from "~/models/member.server";
+import { checkInviteRateLimit, InviteRateLimitError } from "~/services/inviteRateLimiter.server";
 import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
 import { resolveOrgIdFromSlug } from "~/models/organization.server";
 import { TeamPresenter } from "~/presenters/TeamPresenter.server";
@@ -176,6 +177,23 @@ export const action = dashboardAction(
      }
    }

+    // Every invite emails the address, so cap per-org and per-inviter sends
+    // (same limiter as the invite-create API). With no org scope the
+    // slug didn't resolve and inviteMembers rejects anyway.
+    if (env.LOGIN_RATE_LIMITS_ENABLED && context.organizationId) {
+      try {
+        await checkInviteRateLimit(context.organizationId, userId, submission.value.emails.length);
+      } catch (error) {
+        if (error instanceof InviteRateLimitError) {
+          return json(
+            { errors: { body: "Too many invites sent. Please try again later." } },
+            { status: 429 }
+          );
+        }
+        throw error;
+      }
+    }
+
    // Resolve the RBAC role choice. NO_RBAC_ROLE / undefined / unknown
    // role → don't pass one through; the runtime fallback handles it.
    // Validation: the chosen role must be in the org's assignable set
@@ -47,6 +47,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
  return await OrgIntegrationRepository.redirectToAuthService(
    "SLACK",
    project.organizationId,
+    userId,
    request,
    v3NewProjectAlertPathConnectToSlackPath({ slug: organizationSlug }, project, {
      slug: envParam,
@@ -42,6 +42,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
  return await OrgIntegrationRepository.redirectToAuthService(
    "SLACK",
    project.organizationId,
+    userId,
    request,
    v3ErrorsConnectToSlackPath({ slug: organizationSlug }, project, { slug: envParam })
  );
@@ -3,6 +3,7 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { inviteMembers } from "~/models/member.server";
+import { checkInviteRateLimit, InviteRateLimitError } from "~/services/inviteRateLimiter.server";
 import { logger } from "~/services/logger.server";
 import { resolveOrganizationForApiUser } from "~/services/organizationApiAccess.server";
 import { createActionPATApiRoute } from "~/services/routeBuilders/apiBuilder.server";
@@ -57,6 +58,26 @@ export const action = createActionPATApiRoute(
      return json({ error: "Membership is managed by Directory Sync" }, { status: 403 });
    }

+    // Every invite emails the address, so cap per-org and per-inviter sends.
+    if (env.LOGIN_RATE_LIMITS_ENABLED) {
+      try {
+        await checkInviteRateLimit(organization.id, authentication.userId, body.emails.length);
+      } catch (error) {
+        if (error instanceof InviteRateLimitError) {
+          return json(
+            { error: "Too many invites sent. Please try again later." },
+            {
+              status: 429,
+              headers: {
+                "Retry-After": Math.ceil(error.retryAfter / 1000).toString(),
+              },
+            }
+          );
+        }
+        throw error;
+      }
+    }
+
    const { created, alreadyMembers, alreadyInvited } = await inviteMembers({
      slug: organization.slug,
      emails: body.emails,
@@ -1,6 +1,5 @@
 import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import z from "zod";
-import { redirectBackWithErrorMessage } from "~/models/message.server";
 import { OrgIntegrationRepository } from "~/models/orgIntegration.server";
 import { requireUserId } from "~/services/session.server";
 import { requestUrl } from "~/utils/requestUrl.server";
@@ -45,22 +44,35 @@ export async function loader({ request, params }: LoaderFunctionArgs) {

  const parsedParams = ParamsSchema.safeParse(params);

-  if (!parsedParams.success) {
+  if (!parsedParams.success || parsedParams.data.serviceName !== "slack") {
    throw new Response("Invalid params", { status: 400 });
  }

+  const oauthState = await OrgIntegrationRepository.consumeSlackOAuthState(
+    request,
+    parsedSearchParams.data.state,
+    userId
+  );
+  if (!oauthState) {
+    throw new Response("Invalid state", { status: 400 });
+  }
+
  const service = new CreateOrgIntegrationService();

  const integration = await service.call(
    userId,
-    parsedSearchParams.data.state,
-    parsedParams.data.serviceName,
+    oauthState.organizationId,
+    oauthState.service,
    parsedSearchParams.data.code
  );

  if (integration) {
-    return await OrgIntegrationRepository.redirectAfterAuth(request);
+    return await OrgIntegrationRepository.redirectAfterAuth(request, oauthState.redirectTo);
  }

-  return redirectBackWithErrorMessage(request, "Failed to connect to the service");
+  return await OrgIntegrationRepository.redirectAfterAuth(
+    request,
+    oauthState.redirectTo,
+    "Failed to connect to the service"
+  );
 }
@@ -1,9 +1,10 @@
 import { parseWithZod } from "@conform-to/zod";
 import { json } from "@remix-run/server-runtime";
-import { env } from "process";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { resendInvite } from "~/models/member.server";
+import { env } from "~/env.server";
+import { checkInviteRateLimit, InviteRateLimitError } from "~/services/inviteRateLimiter.server";
 import { redirectWithSuccessMessage } from "~/models/message.server";
 import { scheduleEmail } from "~/services/scheduleEmail.server";
 import { ssoController } from "~/services/sso.server";
@@ -50,6 +51,23 @@ export const action = dashboardAction(
      }
    }

+    // Every resend emails the invitee, so apply the same per-org /
+    // per-inviter cap as the invite-create API. With no org scope (the
+    // inviteId lookup above found nothing), resendInvite rejects anyway.
+    if (env.LOGIN_RATE_LIMITS_ENABLED && context.organizationId) {
+      try {
+        await checkInviteRateLimit(context.organizationId, user.id, 1);
+      } catch (error) {
+        if (error instanceof InviteRateLimitError) {
+          return json(
+            { errors: { body: "Too many invites sent. Please try again later." } },
+            { status: 429 }
+          );
+        }
+        throw error;
+      }
+    }
+
    try {
      const invite = await resendInvite({
        inviteId: submission.value.inviteId,
@@ -29,6 +29,7 @@ import {
  checkMagicLinkEmailDailyRateLimit,
  MagicLinkRateLimitError,
  checkMagicLinkIpRateLimit,
+  canonicalizeEmailForRateLimit,
 } from "~/services/magicLinkRateLimiter.server";
 import { ssoRedirectForEmail } from "~/services/ssoAutoDiscovery.server";
 import { logger, tryCatch } from "@trigger.dev/core/v3";
@@ -167,11 +168,16 @@ export async function action({ request }: ActionFunctionArgs) {
        const xff = request.headers.get("x-forwarded-for");
        const clientIp = extractClientIp(xff);

+        // Key the buckets on the canonical address so `+tag` aliases (and
+        // Gmail dot variants) of one inbox share it. Delivery still uses the
+        // raw submitted address below.
+        const rateLimitKey = canonicalizeEmailForRateLimit(email);
+
        const [error] = await tryCatch(
          Promise.all([
            clientIp ? checkMagicLinkIpRateLimit(clientIp) : Promise.resolve(),
-            checkMagicLinkEmailRateLimit(email),
-            checkMagicLinkEmailDailyRateLimit(email),
+            checkMagicLinkEmailRateLimit(rateLimitKey),
+            checkMagicLinkEmailDailyRateLimit(rateLimitKey),
          ])
        );

@@ -9,6 +9,7 @@ import {
 import { MultiFactorAuthenticationService } from "~/services/mfa/multiFactorAuthentication.server";
 import { requireUserId } from "~/services/session.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { MfaRateLimitError } from "~/services/mfa/mfaRateLimiterGlobal.server";
 import { useMfaSetup } from "./useMfaSetup";
 import { MfaToggle } from "./MfaToggle";
 import { MfaSetupDialog } from "./MfaSetupDialog";
@@ -137,6 +138,14 @@ export async function action({ request }: ActionFunctionArgs) {
      return redirectWithErrorMessage("/account/security", request, error.message);
    }

+    if (error instanceof MfaRateLimitError) {
+      return redirectWithErrorMessage(
+        "/account/security",
+        request,
+        "Too many attempts. Please try again later."
+      );
+    }
+
    // Re-throw unexpected errors
    throw error;
  }
@@ -42,6 +42,9 @@ const FORWARDED_HEADERS = [

 // The only turn metadata a browser may set: everything else the agent reads is injected
 // server-side. A whitelist — a new clientData field is server-owned until listed here on purpose.
+// `repoSnapshot` is the dangerous one to smuggle past this: its `tarballUrl` is fetched and
+// extracted on the agent worker, so a client-supplied one is SSRF from inside the worker
+// network plus an attacker-controlled untar.
 const CLIENT_METADATA_KEYS = ["currentPage", "pageContext"] as const;

 export function pickAgentClientMetadata(
@@ -600,8 +600,25 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
      } catch {
        /* invalid JSON — start without metadata */
      }
+      const runtimeEnv = await findEnvironmentBySlug(project.id, envParam, userId);
+      if (!runtimeEnv) return json({ error: "Environment not found" }, { status: 404 });
+
      try {
-        const { publicAccessToken } = await startDashboardAgentSession({ chatId, clientData });
+        // Whitelisted like `create` and the `in` proxy: this object lands in the resumed
+        // run's `basePayload.metadata` verbatim, so without the pick a client could inject
+        // any server-owned field into the agent's first turn (a `repoSnapshot.tarballUrl`
+        // is fetched and extracted on the worker).
+        const { publicAccessToken } = await startDashboardAgentSession({
+          chatId,
+          clientData: {
+            ...pickAgentClientMetadata(clientData),
+            organizationId: project.organizationId,
+            userId,
+            projectId: project.id,
+            environmentId: runtimeEnv.id,
+            ...dashboardAgentEnvironmentAddress(runtimeEnv),
+          },
+        });
        return json({ publicAccessToken });
      } catch (error) {
        logger.error("Failed to start dashboard agent session", { chatId, error });
@@ -79,27 +79,41 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
      throw new Error("Project not found");
    }

+    const environment = await findEnvironmentBySlug(project.id, envParam, userId);
+    if (!environment) {
+      return redirectWithErrorMessage(
+        submission.value.failureRedirect,
+        request,
+        "No waitpoint found"
+      );
+    }
+
    const waitpointId = WaitpointId.toId(waitpointFriendlyId);

    let waitpoint = await runStore.findWaitpoint({
      select: {
-        projectId: true,
-        environmentId: true,
+        id: true,
      },
      where: {
        id: waitpointId,
+        projectId: project.id,
+        environmentId: environment.id,
      },
    });
    if (!waitpoint) {
      // Read-your-writes: a just-minted token may not have replicated. Re-read the owning primary
      // before the auth guard / "No waitpoint found" (mirrors the token complete/callback routes).
      waitpoint = await runStore.findWaitpointOnPrimary({
-        select: { projectId: true, environmentId: true },
-        where: { id: waitpointId },
+        select: { id: true },
+        where: {
+          id: waitpointId,
+          projectId: project.id,
+          environmentId: environment.id,
+        },
      });
    }

-    if (waitpoint?.projectId !== project.id) {
+    if (!waitpoint) {
      return redirectWithErrorMessage(
        submission.value.failureRedirect,
        request,
@@ -157,23 +171,6 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
            );
          }

-          const environment = await findEnvironmentBySlug(project.id, envParam, userId);
-          if (!environment) {
-            return redirectWithErrorMessage(
-              submission.value.failureRedirect,
-              request,
-              "Environment not found"
-            );
-          }
-
-          if (environment.id !== waitpoint.environmentId) {
-            return redirectWithErrorMessage(
-              submission.value.failureRedirect,
-              request,
-              "No waitpoint found"
-            );
-          }
-
          const data = submission.value.payload ? JSON.parse(submission.value.payload) : {};
          const stringifiedData = await stringifyIO(data);
          const finalData = await processWaitpointCompletionPacket(
@@ -0,0 +1,98 @@
+import { Ratelimit } from "@upstash/ratelimit";
+import { env } from "~/env.server";
+import { createRedisRateLimitClient, RateLimiter } from "~/services/rateLimiter.server";
+import { singleton } from "~/utils/singleton";
+
+/**
+ * Rate limiting for organization invite sends (API create + dashboard
+ * resend). Each sent invite is an email to an arbitrary address, so a
+ * member with manage:members could otherwise mass-mail Trigger.dev-branded
+ * invites. Limits are per-organization (the blast radius is the org's
+ * brand) and per-inviter (spreads a burst across collaborators sharing an
+ * org).
+ *
+ * Sized against the 50-emails-per-request body cap: the per-minute windows
+ * pass a few bulk imports but stop a scripted loop; the daily org cap
+ * (500 = 10 bulk imports) leaves headroom for onboarding batches while
+ * bounding a day of abuse.
+ */
+export class InviteRateLimitError extends Error {
+  public readonly retryAfter: number;
+
+  constructor(retryAfter: number) {
+    super("Invite rate limit exceeded.");
+    this.retryAfter = retryAfter;
+  }
+}
+
+function getRedisClient() {
+  return createRedisRateLimitClient({
+    port: env.RATE_LIMIT_REDIS_PORT,
+    host: env.RATE_LIMIT_REDIS_HOST,
+    username: env.RATE_LIMIT_REDIS_USERNAME,
+    password: env.RATE_LIMIT_REDIS_PASSWORD,
+    tlsDisabled: env.RATE_LIMIT_REDIS_TLS_DISABLED === "true",
+    clusterMode: env.RATE_LIMIT_REDIS_CLUSTER_MODE_ENABLED === "1",
+  });
+}
+
+const inviteOrgPerMinuteRateLimiter = singleton(
+  "inviteOrgPerMinuteRateLimiter",
+  () =>
+    new RateLimiter({
+      redisClient: getRedisClient(),
+      keyPrefix: "invites:org",
+      limiter: Ratelimit.slidingWindow(100, "1 m"), // 100 invite emails / min / org
+      logSuccess: false,
+      logFailure: true,
+    })
+);
+
+const inviteOrgDailyRateLimiter = singleton(
+  "inviteOrgDailyRateLimiter",
+  () =>
+    new RateLimiter({
+      redisClient: getRedisClient(),
+      keyPrefix: "invites:org:daily",
+      limiter: Ratelimit.slidingWindow(500, "1 d"), // 500 invite emails / day / org
+      logSuccess: false,
+      logFailure: true,
+    })
+);
+
+const inviteInviterRateLimiter = singleton(
+  "inviteInviterRateLimiter",
+  () =>
+    new RateLimiter({
+      redisClient: getRedisClient(),
+      keyPrefix: "invites:inviter",
+      limiter: Ratelimit.slidingWindow(60, "1 m"), // 60 invite emails / min / inviter
+      logSuccess: false,
+      logFailure: true,
+    })
+);
+
+/**
+ * Check whether `count` invite emails can be sent on behalf of
+ * `organizationId` by `inviterId`. All windows are charged `count` so a
+ * single 50-email request counts as 50 sends, not 1 request.
+ * @throws {InviteRateLimitError} If any limit is exceeded
+ */
+export async function checkInviteRateLimit(
+  organizationId: string,
+  inviterId: string,
+  count: number
+): Promise<void> {
+  const results = await Promise.all([
+    inviteOrgPerMinuteRateLimiter.limit(organizationId, count),
+    inviteOrgDailyRateLimiter.limit(organizationId, count),
+    inviteInviterRateLimiter.limit(inviterId, count),
+  ]);
+
+  for (const result of results) {
+    if (!result.success) {
+      const retryAfter = Math.max(0, new Date(result.reset).getTime() - Date.now());
+      throw new InviteRateLimitError(retryAfter);
+    }
+  }
+}
@@ -68,6 +68,33 @@ function initializeMagicLinkIpRateLimiter() {
  });
 }

+/**
+ * Canonicalize an email address for rate-limit keying so address variants
+ * that resolve to the same inbox share one bucket: lowercase, strip any
+ * `+tag` subaddress, and (for Gmail domains, which ignore dots in the local
+ * part) remove `.` from the local part. Use only for the limiter key —
+ * delivery must always use the raw submitted address.
+ */
+export function canonicalizeEmailForRateLimit(email: string): string {
+  const atIndex = email.lastIndexOf("@");
+  if (atIndex < 1) {
+    return email.toLowerCase();
+  }
+
+  const localPart = email.slice(0, atIndex).toLowerCase();
+  const domain = email.slice(atIndex + 1).toLowerCase();
+
+  const withoutTag = localPart.split("+")[0];
+
+  // Gmail (and Google Workspace on gmail.com/googlemail.com) ignores dots in
+  // the local part; other providers treat them as distinct addresses.
+  if (domain === "gmail.com" || domain === "googlemail.com") {
+    return `${withoutTag.replaceAll(".", "")}@gmail.com`;
+  }
+
+  return `${withoutTag}@${domain}`;
+}
+
 export async function checkMagicLinkEmailRateLimit(identifier: string): Promise<void> {
  const result = await magicLinkEmailRateLimiter.limit(identifier);

@@ -8,6 +8,7 @@ import { createOTP } from "@better-auth/utils/otp";
 import { base32 } from "@better-auth/utils/base32";
 import { z } from "zod";
 import { scheduleEmail } from "../scheduleEmail.server";
+import { checkMfaRateLimit } from "./mfaRateLimiterGlobal.server";

 const generateRandomString = createRandomStringGenerator("A-Z", "0-9");

@@ -48,6 +49,11 @@ export class MultiFactorAuthenticationService {
      };
    }

+    // Rate limit before checking the code: disabling MFA asks for the second
+    // factor, so unlimited attempts here would let a hijacked session
+    // brute-force it. Same limiter as the login verify path.
+    await checkMfaRateLimit(userId);
+
    // validate the TOTP code
    const secretStore = getSecretStore(user.mfaSecretReference.provider);
    const secretResult = await secretStore.getSecret(SecretSchema, user.mfaSecretReference.key);
@@ -171,6 +177,11 @@ export class MultiFactorAuthenticationService {
      throw new ServiceValidationError("User has not enabled MFA");
    }

+    // Rate limit enrollment confirmation too: it verifies a TOTP code against
+    // a real secret, so it needs the same brute-force protection as the
+    // disable and login verify paths.
+    await checkMfaRateLimit(userId);
+
    const secret = secretResult.secret;

    const otp = createOTP(secret, {
@@ -3,9 +3,11 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 const mocks = vi.hoisted(() => ({
  fetch: vi.fn(),
  findEnvironmentBySlug: vi.fn<(...args: any[]) => Promise<any>>(),
+  startSession: vi.fn<(...args: any[]) => Promise<any>>(),
+  chatExists: vi.fn<(...args: any[]) => Promise<any>>(),
 }));

-vi.mock("~/db.server", () => ({ $replica: {} }));
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
 vi.mock("~/env.server", () => ({ env: { SESSION_SECRET: "test-session-secret" } }));
 vi.mock("~/services/session.server", () => ({
  requireUser: async () => ({ id: "usr_real", admin: false, isImpersonating: false }),
@@ -25,14 +27,32 @@ vi.mock("~/models/runtimeEnvironment.server", () => ({
 }));
 vi.mock("~/services/dashboardAgent.server", () => ({
  dashboardAgentApiOrigin: () => "https://api.trigger.dev",
+  isDashboardAgentConfigured: () => true,
+  mintDashboardAgentToken: async () => "pat_public",
  mintDashboardAgentUserActorToken: async () => "tr_uat_real",
  resolveDashboardAgentRepoSnapshot: async () => null,
+  startDashboardAgentSession: mocks.startSession,
+}));
+vi.mock("~/services/dashboardAgentHeadStart.server", () => ({
+  startDashboardAgentHeadStart: vi.fn(),
+}));
+vi.mock("~/services/dashboardAgentDb.server", () => ({ dashboardAgentDb: {} }));
+vi.mock("~/services/resolveTriggerUri.server", () => ({ resolveTriggerUri: () => null }));
+// The chat route reaches the ClickHouse factory through the watch services, and the factory
+// builds its client at import time from an env var no test sets.
+vi.mock("~/services/clickhouse/clickhouseFactoryInstance.server", () => ({
+  clickhouseFactory: { getClickhouseForOrganization: async () => ({}) },
+}));
+vi.mock("@internal/dashboard-agent-db", async (importOriginal) => ({
+  ...((await importOriginal()) as Record<string, unknown>),
+  chatExists: mocks.chatExists,
 }));
 vi.mock("~/services/logger.server", () => ({
  logger: { debug: vi.fn(), error: vi.fn(), warn: vi.fn(), info: vi.fn() },
 }));

 import { action } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboard-agent.in.$";
+import { action as chatAction } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboard-agent";

 async function appendTurn(metadata: Record<string, unknown>): Promise<Record<string, unknown>> {
  const request = new Request(
@@ -160,3 +180,100 @@ describe("dashboard agent `in` proxy — client metadata", () => {
    expect(metadata).not.toHaveProperty("somethingNew");
  });
 });
+
+// `intent=start` resumes an owned chat; the client-supplied clientData is folded into the
+// resumed run's payload metadata verbatim, so it goes through the same whitelist as the
+// `in` proxy. Otherwise a client could inject `repoSnapshot.tarballUrl` and the agent
+// worker would fetch and extract it.
+describe("dashboard agent `start` intent — client metadata", () => {
+  beforeEach(() => {
+    mocks.chatExists.mockReset().mockResolvedValue(true);
+    mocks.startSession.mockReset().mockResolvedValue({ publicAccessToken: "pat_public" });
+    mocks.findEnvironmentBySlug.mockReset().mockResolvedValue({
+      id: "env_real",
+      type: "DEVELOPMENT",
+      branchName: null,
+    });
+  });
+
+  async function startChat(clientData: Record<string, unknown>) {
+    const form = new URLSearchParams({
+      intent: "start",
+      chatId: "chat_real",
+      clientData: JSON.stringify(clientData),
+    });
+
+    const response = await chatAction({
+      request: new Request(
+        "https://app.trigger.dev/resources/orgs/acme/projects/api/env/dev/dashboard-agent",
+        {
+          method: "POST",
+          headers: { "content-type": "application/x-www-form-urlencoded" },
+          body: form.toString(),
+        }
+      ),
+      params: { organizationSlug: "acme", projectParam: "api", envParam: "dev" },
+      context: {},
+    } as any);
+
+    expect(response.status).toBe(200);
+    expect(mocks.startSession).toHaveBeenCalledTimes(1);
+    return mocks.startSession.mock.calls[0][0].clientData as Record<string, unknown>;
+  }
+
+  it("keeps the whitelisted page context", async () => {
+    const clientData = await startChat({ currentPage: "/runs", pageContext: { kind: "runs" } });
+
+    expect(clientData).toMatchObject({ currentPage: "/runs", pageContext: { kind: "runs" } });
+  });
+
+  it("drops every server-owned field a client sends", async () => {
+    const clientData = await startChat({
+      currentPage: "/runs",
+      organizationId: "org_evil",
+      userId: "usr_evil",
+      projectId: "proj_evil",
+      projectRef: "proj_ref_evil",
+      environmentId: "env_evil",
+      environmentName: "prod",
+      environmentBranch: "evil-branch",
+      apiOrigin: "https://evil.example.com",
+      userActorToken: "tr_uat_evil",
+      repoSnapshot: { tarballUrl: "https://evil.example.com/x.tar.gz" },
+      somethingNew: "smuggled",
+    });
+
+    expect(clientData).toMatchObject({
+      currentPage: "/runs",
+      organizationId: "org_real",
+      userId: "usr_real",
+      projectId: "proj_real",
+      environmentId: "env_real",
+      environmentName: "dev",
+    });
+    expect(clientData.projectRef).toBeUndefined();
+    expect(clientData.environmentBranch).toBeUndefined();
+    expect(clientData.apiOrigin).toBeUndefined();
+    expect(clientData.userActorToken).toBeUndefined();
+    expect(clientData.repoSnapshot).toBeUndefined();
+    expect(clientData).not.toHaveProperty("somethingNew");
+  });
+
+  it("re-injects the server-owned identity the resumed run boots with", async () => {
+    const clientData = await startChat({
+      currentPage: "/runs",
+      organizationId: "org_evil",
+      userId: "usr_evil",
+      projectId: "proj_evil",
+      environmentId: "env_evil",
+    });
+
+    expect(clientData).toMatchObject({
+      currentPage: "/runs",
+      organizationId: "org_real",
+      userId: "usr_real",
+      projectId: "proj_real",
+      environmentId: "env_real",
+    });
+  });
+});
@@ -63,10 +63,11 @@ vi.mock("~/services/logger.server", () => ({ logger: mocks.logger }));

 import { action } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.dashboard-agent";

-function createChatRequest() {
+function createChatRequest(clientData?: Record<string, unknown>) {
  const form = new URLSearchParams({
    intent: "create",
    message: JSON.stringify({ id: "m1", role: "user", parts: [{ type: "text", text: "hi" }] }),
+    ...(clientData ? { clientData: JSON.stringify(clientData) } : {}),
  });

  return action({
@@ -129,6 +130,44 @@ describe("dashboard agent chat creation — nothing fallible after the row exist
    });
    expect(mocks.softDeleteChat).not.toHaveBeenCalled();
  });
+
+  // The head-start metadata is built from the client's clientData: a smuggled
+  // `repoSnapshot.tarballUrl` would be fetched and extracted on the agent worker, so only
+  // the whitelisted page context may survive the merge.
+  it("strips server-owned fields from the clientData before head-starting", async () => {
+    const response = await createChatRequest({
+      currentPage: "/runs",
+      pageContext: { kind: "runs" },
+      organizationId: "org_evil",
+      userId: "usr_evil",
+      projectId: "proj_evil",
+      environmentId: "env_evil",
+      userActorToken: "tr_uat_evil",
+      apiOrigin: "https://evil.example.com",
+      repoSnapshot: { tarballUrl: "https://evil.example.com/x.tar.gz" },
+      somethingNew: "smuggled",
+    });
+
+    expect(response.status).toBe(200);
+    const metadata = mocks.headStart.mock.calls[0][0].metadata;
+    expect(metadata.currentPage).toBe("/runs");
+    expect(metadata.pageContext).toEqual({ kind: "runs" });
+    expect(metadata.organizationId).toBe("org_real");
+    expect(metadata.userId).toBe("usr_real");
+    expect(metadata.projectId).toBe("proj_real");
+    expect(metadata.environmentId).toBe("env_real");
+    expect(metadata.userActorToken).toBe("tr_uat_real");
+    expect(metadata.apiOrigin).toBe("https://api.trigger.dev");
+    expect(metadata.repoSnapshot).toBeUndefined();
+    expect(metadata).not.toHaveProperty("somethingNew");
+
+    // The chat row's stored context is whitelisted too. createChat(db, params) takes
+    // the db as arg 0, so the params object (carrying metadata) is arg 1.
+    const chatMetadata = mocks.createChat.mock.calls[0][1].metadata;
+    expect(chatMetadata).toEqual({
+      context: { currentPage: "/runs", pageContext: { kind: "runs" } },
+    });
+  });
 });

 // A failed start means no handover was dispatched and no message was sent, so any session it
@@ -0,0 +1,87 @@
+import { describe, expect, it, beforeEach, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+  requireUserId: vi.fn(),
+  consumeState: vi.fn(),
+  redirectAfterAuth: vi.fn(),
+  createIntegration: vi.fn(),
+}));
+
+vi.mock("~/services/session.server", () => ({ requireUserId: mocks.requireUserId }));
+vi.mock("~/models/orgIntegration.server", () => ({
+  OrgIntegrationRepository: {
+    consumeSlackOAuthState: mocks.consumeState,
+    redirectAfterAuth: mocks.redirectAfterAuth,
+  },
+}));
+vi.mock("~/v3/services/createOrgIntegration.server", () => ({
+  CreateOrgIntegrationService: class {
+    call = mocks.createIntegration;
+  },
+}));
+vi.mock("~/utils/requestUrl.server", () => ({
+  requestUrl: (request: Request) => new URL(request.url),
+}));
+const { loader } = await import("../app/routes/integrations.$serviceName.callback.js");
+
+const request = () =>
+  new Request("https://example.com/integrations/slack/callback?code=code_123&state=state_123");
+const args = () => ({ request: request(), params: { serviceName: "slack" } }) as any;
+
+beforeEach(() => {
+  vi.resetAllMocks();
+  mocks.requireUserId.mockResolvedValue("user_123");
+});
+
+describe("Slack OAuth callback", () => {
+  it("uses the consumed state scope before exchanging the authorization code", async () => {
+    mocks.consumeState.mockResolvedValue({
+      organizationId: "org_123",
+      service: "slack",
+      redirectTo: "/orgs/acme/projects/app/env/prod/alerts/new/connect-to-slack",
+    });
+    const response = new Response(null, { status: 302 });
+    mocks.createIntegration.mockResolvedValue({ id: "integration_123" });
+    mocks.redirectAfterAuth.mockResolvedValue(response);
+
+    await expect(loader(args())).resolves.toBe(response);
+    expect(mocks.consumeState).toHaveBeenCalledWith(expect.any(Request), "state_123", "user_123");
+    expect(mocks.createIntegration).toHaveBeenCalledWith(
+      "user_123",
+      "org_123",
+      "slack",
+      "code_123"
+    );
+    expect(mocks.redirectAfterAuth).toHaveBeenCalledWith(
+      expect.any(Request),
+      "/orgs/acme/projects/app/env/prod/alerts/new/connect-to-slack"
+    );
+  });
+
+  it("rejects invalid state before the authorization code exchange or integration writes", async () => {
+    mocks.consumeState.mockResolvedValue(undefined);
+
+    await expect(loader(args())).rejects.toMatchObject({ status: 400 });
+    expect(mocks.createIntegration).not.toHaveBeenCalled();
+    expect(mocks.redirectAfterAuth).not.toHaveBeenCalled();
+  });
+
+  it("clears the session binding and returns to the stored path when integration fails", async () => {
+    const redirectTo = "/orgs/acme/projects/app/env/prod/alerts/new/connect-to-slack";
+    mocks.consumeState.mockResolvedValue({
+      organizationId: "org_123",
+      service: "slack",
+      redirectTo,
+    });
+    mocks.createIntegration.mockResolvedValue(undefined);
+    const response = new Response(null, { status: 302 });
+    mocks.redirectAfterAuth.mockResolvedValue(response);
+
+    await expect(loader(args())).resolves.toBe(response);
+    expect(mocks.redirectAfterAuth).toHaveBeenCalledWith(
+      expect.any(Request),
+      redirectTo,
+      "Failed to connect to the service"
+    );
+  });
+});
@@ -0,0 +1,205 @@
+import { redisTest } from "@internal/testcontainers";
+import Redis from "ioredis";
+import { describe, expect, vi } from "vitest";
+import {
+  clearSlackOAuthSessionBinding,
+  consumeSlackOAuthStateForSession,
+  createSlackOAuthStateForSession,
+  SlackOAuthStateStore,
+} from "../app/models/slackOAuthState.server.js";
+
+vi.setConfig({ testTimeout: 30_000 });
+
+const state = {
+  userId: "user_123",
+  sessionBinding: "session_123",
+  organizationId: "org_123",
+  service: "slack" as const,
+  redirectTo: "/orgs/acme/projects/app/env/prod/alerts/new/connect-to-slack",
+};
+
+const expectedState = {
+  userId: state.userId,
+  sessionBinding: state.sessionBinding,
+  service: state.service,
+};
+
+const keyForNonce = (nonce: string) => `oauth:slack:state:{${nonce}}`;
+
+describe("SlackOAuthStateStore", () => {
+  redisTest(
+    "creates a cryptographically random, expiring state value",
+    async ({ redisOptions }) => {
+      const redis = new Redis(redisOptions);
+
+      try {
+        const store = new SlackOAuthStateStore(redis);
+        const nonce = await store.create(state);
+
+        expect(nonce).toMatch(/^[A-Za-z0-9_-]{43}$/);
+        await expect(redis.get(keyForNonce(nonce))).resolves.toBe(JSON.stringify(state));
+        const ttl = await redis.ttl(keyForNonce(nonce));
+        expect(ttl).toBeGreaterThan(0);
+        expect(ttl).toBeLessThanOrEqual(600);
+      } finally {
+        redis.disconnect();
+      }
+    }
+  );
+
+  redisTest(
+    "atomically accepts one matching callback and rejects concurrent replays",
+    async ({ redisOptions }) => {
+      const redis = new Redis(redisOptions);
+
+      try {
+        const store = new SlackOAuthStateStore(redis);
+        const nonce = await store.create(state);
+        const results = await Promise.all([
+          store.consume(nonce, expectedState),
+          store.consume(nonce, expectedState),
+        ]);
+
+        expect(results.filter((result) => result !== undefined)).toEqual([state]);
+        expect(results.filter((result) => result === undefined)).toHaveLength(1);
+      } finally {
+        redis.disconnect();
+      }
+    }
+  );
+
+  redisTest("rejects mismatched state without consuming it", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+    const mismatches = [
+      { userId: "user_456", sessionBinding: state.sessionBinding, service: "slack" },
+      { userId: state.userId, sessionBinding: "session_456", service: "slack" },
+      { userId: state.userId, sessionBinding: state.sessionBinding, service: "vercel" },
+    ];
+
+    try {
+      const store = new SlackOAuthStateStore(redis);
+
+      for (const mismatch of mismatches) {
+        const nonce = await store.create(state);
+        await expect(store.consume(nonce, mismatch as any)).resolves.toBeUndefined();
+        await expect(store.consume(nonce, expectedState)).resolves.toEqual(state);
+      }
+    } finally {
+      redis.disconnect();
+    }
+  });
+
+  redisTest("rejects missing state and malformed nonces", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+
+    try {
+      const store = new SlackOAuthStateStore(redis);
+      const nonce = await store.create(state);
+      await redis.del(keyForNonce(nonce));
+
+      await expect(store.consume(nonce, expectedState)).resolves.toBeUndefined();
+      await expect(store.consume("malformed-state", expectedState)).resolves.toBeUndefined();
+    } finally {
+      redis.disconnect();
+    }
+  });
+
+  redisTest("rejects corrupt stored state", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+
+    try {
+      const store = new SlackOAuthStateStore(redis);
+      const nonce = "a".repeat(43);
+      await redis.set(keyForNonce(nonce), "{not-json", "EX", 600);
+
+      await expect(store.consume(nonce, expectedState)).resolves.toBeUndefined();
+    } finally {
+      redis.disconnect();
+    }
+  });
+
+  redisTest("rejects protocol-relative return paths", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+
+    try {
+      const store = new SlackOAuthStateStore(redis);
+      await expect(store.create({ ...state, redirectTo: "//example.com" })).rejects.toThrow();
+    } finally {
+      redis.disconnect();
+    }
+  });
+
+  redisTest("binds callbacks to the initiating browser session", async ({ redisOptions }) => {
+    const redis = new Redis(redisOptions);
+
+    try {
+      const store = new SlackOAuthStateStore(redis);
+      const stateWithoutBinding = {
+        userId: state.userId,
+        organizationId: state.organizationId,
+        service: state.service,
+        redirectTo: state.redirectTo,
+      };
+      const first = await createSlackOAuthStateForSession(
+        new Request("https://example.com/connect"),
+        stateWithoutBinding,
+        store
+      );
+      const second = await createSlackOAuthStateForSession(
+        new Request("https://example.com/connect"),
+        stateWithoutBinding,
+        store
+      );
+      const firstCookie = first.sessionCookie.split(";", 1)[0];
+      const secondCookie = second.sessionCookie.split(";", 1)[0];
+      const requestWithCookie = (cookie?: string) =>
+        new Request("https://example.com/integrations/slack/callback", {
+          headers: cookie ? { Cookie: cookie } : undefined,
+        });
+
+      await expect(
+        consumeSlackOAuthStateForSession(requestWithCookie(), first.nonce, state.userId, store)
+      ).resolves.toBeUndefined();
+      await expect(
+        consumeSlackOAuthStateForSession(
+          requestWithCookie(secondCookie),
+          first.nonce,
+          state.userId,
+          store
+        )
+      ).resolves.toBeUndefined();
+
+      const consumedState = await consumeSlackOAuthStateForSession(
+        requestWithCookie(firstCookie),
+        first.nonce,
+        state.userId,
+        store
+      );
+      expect(consumedState).toMatchObject({
+        userId: state.userId,
+        organizationId: state.organizationId,
+        service: state.service,
+        redirectTo: state.redirectTo,
+        sessionBinding: expect.any(String),
+      });
+
+      const clearedCookie = (
+        await clearSlackOAuthSessionBinding(requestWithCookie(firstCookie))
+      ).split(";", 1)[0];
+      const staleNonce = await store.create({
+        ...state,
+        sessionBinding: consumedState!.sessionBinding,
+      });
+      await expect(
+        consumeSlackOAuthStateForSession(
+          requestWithCookie(clearedCookie),
+          staleNonce,
+          state.userId,
+          store
+        )
+      ).resolves.toBeUndefined();
+    } finally {
+      redis.disconnect();
+    }
+  });
+});
@@ -0,0 +1,183 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const auth = vi.hoisted(() => ({
+  userId: "user_attacker",
+  projectId: "proj_shared",
+  environmentId: "env_attacker",
+}));
+
+const waitpointHolder = vi.hoisted(() => ({
+  waitpoint: undefined as { id: string; projectId: string; environmentId: string } | undefined,
+  replicaCalls: [] as Array<{ where: Record<string, string> }>,
+  primaryCalls: [] as Array<{ where: Record<string, string> }>,
+}));
+
+const engineHolder = vi.hoisted(() => ({ calls: [] as Array<Record<string, unknown>> }));
+
+function matchesWaitpoint(where: Record<string, string>) {
+  const waitpoint = waitpointHolder.waitpoint;
+  if (!waitpoint || where.id !== waitpoint.id) return undefined;
+  if (where.projectId !== undefined && where.projectId !== waitpoint.projectId) return undefined;
+  if (where.environmentId !== undefined && where.environmentId !== waitpoint.environmentId) {
+    return undefined;
+  }
+  return waitpoint;
+}
+
+vi.mock("~/v3/runStore.server", () => ({
+  runStore: {
+    findWaitpoint: async ({ where }: { where: Record<string, string> }) => {
+      waitpointHolder.replicaCalls.push({ where });
+      return matchesWaitpoint(where);
+    },
+    findWaitpointOnPrimary: async ({ where }: { where: Record<string, string> }) => {
+      waitpointHolder.primaryCalls.push({ where });
+      return matchesWaitpoint(where);
+    },
+  },
+}));
+
+vi.mock("~/db.server", () => ({
+  $replica: {
+    project: {
+      findUnique: async () => ({ id: auth.projectId }),
+    },
+  },
+}));
+
+vi.mock("~/models/runtimeEnvironment.server", () => ({
+  findEnvironmentBySlug: async () => ({ id: auth.environmentId }),
+}));
+
+vi.mock("~/v3/runEngine.server", () => ({
+  engine: {
+    completeWaitpoint: async (args: Record<string, unknown>) => {
+      engineHolder.calls.push(args);
+      return { id: args.id };
+    },
+  },
+}));
+
+vi.mock("~/services/session.server", () => ({
+  requireUserId: async () => auth.userId,
+}));
+
+vi.mock("~/env.server", () => ({
+  env: { TASK_PAYLOAD_MAXIMUM_SIZE: 3_000_000 },
+}));
+
+vi.mock("~/services/logger.server", () => ({
+  logger: { error: () => {}, info: () => {}, debug: () => {}, warn: () => {} },
+}));
+
+vi.mock("~/models/message.server", () => ({
+  redirectWithErrorMessage: (redirect: string, _request: Request, message: string) =>
+    new Response(null, {
+      status: 302,
+      headers: { location: redirect, "x-outcome": "error", "x-message": message },
+    }),
+  redirectWithSuccessMessage: (redirect: string, _request: Request, message: string) =>
+    new Response(null, {
+      status: 302,
+      headers: { location: redirect, "x-outcome": "success", "x-message": message },
+    }),
+}));
+
+vi.mock("~/runEngine/concerns/waitpointCompletionPacket.server", () => ({
+  processWaitpointCompletionPacket: async () => ({ data: undefined, dataType: "application/json" }),
+}));
+
+import { WaitpointId } from "@trigger.dev/core/v3/isomorphic";
+import { action } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route";
+
+function completeRequest(type: "DATETIME" | "MANUAL", isTimeout = false) {
+  const body = new URLSearchParams({
+    type,
+    successRedirect: "/success",
+    failureRedirect: "/failure",
+  });
+  if (type === "MANUAL") body.set("payload", "{}");
+  if (isTimeout) body.set("isTimeout", "1");
+
+  return new Request("http://localhost/complete", {
+    method: "POST",
+    headers: { "content-type": "application/x-www-form-urlencoded" },
+    body,
+  });
+}
+
+async function complete(friendlyId: string, type: "DATETIME" | "MANUAL", isTimeout = false) {
+  return (await action({
+    request: completeRequest(type, isTimeout),
+    params: {
+      organizationSlug: "org-slug",
+      projectParam: "project-slug",
+      envParam: "dev",
+      waitpointFriendlyId: friendlyId,
+    },
+    context: {} as never,
+  })) as Response;
+}
+
+beforeEach(() => {
+  waitpointHolder.waitpoint = undefined;
+  waitpointHolder.replicaCalls = [];
+  waitpointHolder.primaryCalls = [];
+  engineHolder.calls = [];
+});
+
+describe("dashboard waitpoint completion authorization", () => {
+  it.each([
+    { type: "DATETIME" as const, isTimeout: false },
+    { type: "MANUAL" as const, isTimeout: true },
+  ])(
+    "rejects $type completion for another development environment",
+    async ({ type, isTimeout }) => {
+      const { id, friendlyId } = WaitpointId.generate();
+      waitpointHolder.waitpoint = {
+        id,
+        projectId: auth.projectId,
+        environmentId: "env_victim",
+      };
+
+      const response = await complete(friendlyId, type, isTimeout);
+
+      expect(response.headers.get("x-outcome")).toBe("error");
+      expect(response.headers.get("x-message")).toBe("No waitpoint found");
+      expect(engineHolder.calls).toHaveLength(0);
+      expect(waitpointHolder.replicaCalls[0]?.where).toEqual({
+        id,
+        projectId: auth.projectId,
+        environmentId: auth.environmentId,
+      });
+      expect(waitpointHolder.primaryCalls[0]?.where).toEqual({
+        id,
+        projectId: auth.projectId,
+        environmentId: auth.environmentId,
+      });
+    }
+  );
+
+  it.each([
+    { type: "DATETIME" as const, isTimeout: false, message: "Waitpoint skipped" },
+    { type: "MANUAL" as const, isTimeout: true, message: "Waitpoint timed out" },
+  ])(
+    "allows $type completion in the authorized environment",
+    async ({ type, isTimeout, message }) => {
+      const { id, friendlyId } = WaitpointId.generate();
+      waitpointHolder.waitpoint = {
+        id,
+        projectId: auth.projectId,
+        environmentId: auth.environmentId,
+      };
+
+      const response = await complete(friendlyId, type, isTimeout);
+
+      expect(response.headers.get("x-outcome")).toBe("success");
+      expect(response.headers.get("x-message")).toBe(message);
+      expect(engineHolder.calls).toHaveLength(1);
+      expect(engineHolder.calls[0]?.id).toBe(id);
+      expect(waitpointHolder.primaryCalls).toHaveLength(0);
+    }
+  );
+});
@@ -1,7 +1,7 @@
 // Property: under split replica lag the dashboard "complete waitpoint" route action still completes a
-// just-minted token. It resolves the waitpoint by id via findWaitpoint (owning REPLICA), and on a null
-// re-reads via findWaitpointOnPrimary before the projectId guard, so a token invisible on the lagging
-// replica passes the guard and completion proceeds instead of failing with "No waitpoint found".
+// just-minted token. It resolves the waitpoint by project and environment via findWaitpoint (owning
+// REPLICA), and on a null re-reads via findWaitpointOnPrimary before returning "No waitpoint found".
+// A token invisible on the lagging replica still completes after the primary fallback.
 // Drives the REAL exported action; only peripheral collaborators are mocked. The seam — runStore over a
 // split topology whose owning replica is frozen — is a REAL RoutingRunStore over real testcontainer
 // Postgres.
@@ -64,8 +64,8 @@ vi.mock("~/models/message.server", () => ({
    }),
 }));

-// MANUAL-branch collaborators — the token completion path resolves the env then completes. Return an
-// env whose id matches the seeded waitpoint's environmentId so the env guard passes.
+// The completion path resolves the environment before loading the waitpoint. Return an environment
+// whose id matches the seeded waitpoint so the scoped lookup can authorize it.
 const envHolder = vi.hoisted(() => ({ id: undefined as string | undefined }));
 vi.mock("~/models/runtimeEnvironment.server", () => ({
  findEnvironmentBySlug: async () => (envHolder.id ? { id: envHolder.id } : null),
@@ -147,8 +147,8 @@ const params = (friendlyId: string) => ({

 describe("complete-waitpoint dashboard route reads-your-writes under split replica lag", () => {
  // LEGACY-resident (cuid) token minted on the control-plane writer; its replica lags. The action's
-  // findWaitpoint(id) misses, and the findWaitpointOnPrimary fallback must resolve it so the just-
-  // minted token passes the projectId guard and completes — NOT "No waitpoint found".
+  // environment-scoped findWaitpoint misses, and the findWaitpointOnPrimary fallback must resolve it
+  // so the just-minted token completes instead of returning "No waitpoint found".
  heteroRunOpsPostgresTest(
    "MANUAL token invisible on the lagging owning replica completes via the primary fallback",
    async ({ prisma14, prisma17 }) => {
@@ -200,8 +200,8 @@ describe("complete-waitpoint dashboard route reads-your-writes under split repli
    }
  );

-  // Same seam via the DATETIME "skip" branch (also gated by the shared projectId guard). Kept as a
-  // second, mock-light assertion of the fallback so the guard doesn't hinge on MANUAL-branch helpers.
+  // Same seam via the DATETIME "skip" branch. Kept as a second, mock-light assertion of the fallback
+  // so the authorization behavior doesn't hinge on MANUAL-branch helpers.
  heteroRunOpsPostgresTest(
    "DATETIME skip on a lag-invisible token resolves via the primary fallback",
    async ({ prisma14, prisma17 }) => {
@@ -271,6 +271,18 @@ export const dashboardAgentToolsKey = locals.create<ToolSet>("dashboard-agent.to
 // within a recycle.
 type DashboardAgentMode = "assistant" | "code";

+// The snapshot is fetched and extracted on the agent worker, so its URL must be one the
+// server would have minted: plain https. The host check lives in repo-tools' fetch.
+// Guarded at the schema edge too because old workers (the deployed version is pinned)
+// can still replay metadata that carries a snapshot.
+const repoSnapshotTarballUrlSchema = z.string().refine((value) => {
+  try {
+    return new URL(value).protocol === "https:";
+  } catch {
+    return false;
+  }
+}, "tarballUrl must be an https URL");
+
 // A turn is in `code` mode when the project has a connected repo. Drives both the
 // tool set and the prompt.
 export function modeFor(clientData: { repoSnapshot?: unknown } | undefined): DashboardAgentMode {
@@ -312,7 +324,7 @@ export const clientDataSchema = z.object({
  // short-lived archive pointer the code-mode source tools read from.
  repoSnapshot: z
    .object({
-      tarballUrl: z.string(),
+      tarballUrl: repoSnapshotTarballUrlSchema,
      owner: z.string(),
      repo: z.string(),
      sha: z.string(),
@@ -1126,6 +1126,36 @@ describe("clientDataSchema", () => {
    });
    expect(parsed.success).toBe(false);
  });
+
+  // The snapshot URL is fetched and extracted on the worker, so the schema refuses
+  // anything but plain https (the host allowlist is enforced at the fetch site).
+  it("rejects a repoSnapshot whose tarballUrl is not an https URL", () => {
+    for (const tarballUrl of [
+      "http://codeload.github.com/acme/demo/tar.gz/abc",
+      "ftp://example.com/x.tar.gz",
+      "file:///etc/passwd",
+      "not a url",
+    ]) {
+      const parsed = clientDataSchema.safeParse({
+        userId: "user_1",
+        organizationId: "org_1",
+        repoSnapshot: { tarballUrl, owner: "acme", repo: "demo", sha: "c".repeat(40) },
+      });
+      expect(parsed.success).toBe(false);
+    }
+
+    const ok = clientDataSchema.safeParse({
+      userId: "user_1",
+      organizationId: "org_1",
+      repoSnapshot: {
+        tarballUrl: "https://codeload.github.com/acme/demo/tar.gz/abc",
+        owner: "acme",
+        repo: "demo",
+        sha: "c".repeat(40),
+      },
+    });
+    expect(ok.success).toBe(true);
+  });
 });

 describe("buildDashboardAgentTools", () => {
@@ -194,6 +194,27 @@ describe("repo-tools", () => {
    expect(res.error).toMatch(/Couldn't resolve the source/);
  });

+  // The snapshot fetch runs on an internal worker, so only GitHub's archive host is
+  // allowed; anything else must fail before a request leaves the worker.
+  it("refuses to fetch a snapshot whose tarballUrl is not an allowed host", async () => {
+    for (const tarballUrl of [
+      "http://codeload.github.com/acme/attacker/tar.gz/abc",
+      "https://attacker.example.com/x.tar.gz",
+      "https://github.com.evil.example/x.tar.gz",
+      "not a url",
+    ]) {
+      const bad: RepoSnapshot = {
+        tarballUrl,
+        owner: "acme",
+        repo: "attacker",
+        sha: "b".repeat(40),
+      };
+      const res: any = await call(buildRepoTools(bad).read_file, { path: "README.md" });
+      expect(res.error).toMatch(/Couldn't load the repository/);
+      expect(res.error).toMatch(/not a valid URL|not allowed/);
+    }
+  });
+
  it.runIf(hasRg)("search_code finds a match (and does not hang on stdin)", async () => {
    const res: any = await call(tools.search_code, { query: "const LIMIT" });
    expect(res.error).toBeUndefined();
@@ -39,6 +39,24 @@ export type RepoSnapshot = {
 };

 const MAX_ARCHIVE_BYTES = 100 * 1024 * 1024; // 100MB ceiling on the download
+
+// The snapshot fetch is a network primitive on an internal worker, so the URL must be one
+// the webapp would have minted: https to GitHub's own archive hosts (the signed redirect
+// target of the `tarball` API is codeload). A validation failure is thrown before any
+// request leaves the worker.
+const ALLOWED_TARBALL_HOSTS = new Set(["codeload.github.com"]);
+
+function assertAllowedTarballUrl(tarballUrl: string): void {
+  let url: URL;
+  try {
+    url = new URL(tarballUrl);
+  } catch {
+    throw new Error("repo snapshot URL is not a valid URL");
+  }
+  if (url.protocol !== "https:" || !ALLOWED_TARBALL_HOSTS.has(url.hostname)) {
+    throw new Error(`repo snapshot URL host is not allowed (${url.hostname || "unparseable"})`);
+  }
+}
 // A tool result the model has to pay for on every later turn of the conversation:
 // 48KB is ~12k tokens, where the old 256KB ceiling was ~65k.
 export const MAX_READ_BYTES = 48 * 1024;
@@ -101,7 +119,13 @@ async function ensureWorkspace(snapshot: RepoSnapshot): Promise<string> {
    const timer = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS);
    let tarPath: string | undefined;
    try {
-      const res = await fetch(snapshot.tarballUrl, { signal: controller.signal });
+      assertAllowedTarballUrl(snapshot.tarballUrl);
+      // No redirects: the server resolves the one hop to a signed codeload URL itself, so
+      // following one here would just reopen the host check to a redirect target.
+      const res = await fetch(snapshot.tarballUrl, {
+        signal: controller.signal,
+        redirect: "error",
+      });
      if (!res.ok) throw new Error(`archive download failed (status ${res.status})`);
      const length = Number(res.headers.get("content-length") ?? 0);
      if (length > MAX_ARCHIVE_BYTES) throw new Error(`archive too large (${length} bytes)`);
@@ -113,6 +137,29 @@ async function ensureWorkspace(snapshot: RepoSnapshot): Promise<string> {
      tarPath = join(scratch, "repo.tar.gz");
      await writeFile(tarPath, bytes);

+      // List the archive's top-level entries before extracting: anything that resolves
+      // outside the root after `--strip-components=1` (`..` segments, absolute paths)
+      // rejects the archive. Both bsdtar and GNU tar refuse `..` members outright; this
+      // rejects rather than relying on the extractor, and the read tools' realpath checks
+      // keep any surviving symlink member from pointing a read outside the root.
+      const { stdout: listing } = await execFileAsync("tar", ["-tzf", tarPath], {
+        // The listing is a line per member; a 100MB archive can't hold more members
+        // than this fits at any plausible path length.
+        maxBuffer: 64 * 1024 * 1024,
+      });
+      const roots = new Set<string>();
+      for (const entry of listing.split("\n")) {
+        const name = entry.replace(/\/+$/, "");
+        if (!name) continue;
+        if (isAbsolute(name) || name.split("/").includes("..")) {
+          throw new Error(`archive entry escapes the workspace (${name})`);
+        }
+        roots.add(name.split("/")[0]);
+        if (roots.size > 1) {
+          throw new Error("archive has more than one top-level entry; refusing to extract");
+        }
+      }
+
      await mkdir(workdir, { recursive: true });
      await execFileAsync("tar", ["-xzf", tarPath, "-C", workdir, "--strip-components=1"]);
      await writeFile(join(workdir, ".ready"), snapshot.sha);
@@ -429,8 +429,15 @@ export class ClickHousePrinter {
      windowClause = windowDefs.join(", ");
    }

+    // PREWHERE runs before enforced tenant conditions in WHERE, so it cannot be
+    // exposed to customer-authored queries.
+    if (node.prewhere) {
+      throw new QueryError("PREWHERE is not supported. Use WHERE instead.", {
+        node: node.prewhere,
+      });
+    }
+
    // Process other clauses
-    const prewhere = node.prewhere ? this.visit(node.prewhere) : null;
    const whereStr = where ? this.visit(where) : null;

    // Process GROUP BY with context flags:
@@ -502,7 +509,6 @@ export class ClickHousePrinter {
      `SELECT${space}${node.distinct ? "DISTINCT " : ""}${columns.join(comma)}`,
      joinedTables.length > 0 ? `FROM${space}${joinedTables.join(space)}` : null,
      arrayJoin || null,
-      prewhere ? `PREWHERE${space}${prewhere}` : null,
      whereStr ? `WHERE${space}${whereStr}` : null,
      groupBy && groupBy.length > 0 ? `GROUP BY${space}${groupBy.join(comma)}` : null,
      having ? `HAVING${space}${having}` : null,
@@ -191,6 +191,30 @@ describe("Cross-Tenant Security", () => {
    });
  });

+  describe("PREWHERE", () => {
+    it.each([
+      ["top-level query", "SELECT count(*) FROM task_runs PREWHERE toUInt8(task_identifier) = 1"],
+      [
+        "case and whitespace variant",
+        "SELECT count(*) FROM task_runs\nPrEwHeRe\n  toUInt8(task_identifier) = 1",
+      ],
+      [
+        "nested query",
+        `SELECT id FROM task_runs WHERE id IN (
+          SELECT run_id FROM task_events PREWHERE event_type = 'completed'
+        )`,
+      ],
+      [
+        "set query",
+        `SELECT id FROM task_runs
+         UNION ALL
+         SELECT id FROM task_runs PREWHERE status = 'completed'`,
+      ],
+    ])("should reject PREWHERE in a %s", (_, query) => {
+      expect(() => compile(query)).toThrowError("PREWHERE is not supported. Use WHERE instead.");
+    });
+  });
+
  describe("Table allowlisting", () => {
    it("should reject queries to unknown tables", () => {
      expect(() => {