zalo: gate image downloads before DM auth (#55979)

* zalo: gate image downloads before DM auth

* zalo: clarify pre-download auth sentinel

extensions/zalo/src/monitor.image.polling.test.ts

@@ -8,6 +8,7 @@ import {
   getUpdatesMock,
   getZaloRuntimeMock,
   resetLifecycleTestState,
+  sendMessageMock,
 } from "../../../test/helpers/extensions/zalo-lifecycle.js";
 
 describe("Zalo polling image handling", () => {
@@ -62,4 +63,38 @@ describe("Zalo polling image handling", () => {
     abort.abort();
     await run;
   });
+
+  it("rejects unauthorized DM images before downloading media", async () => {
+    getUpdatesMock
+      .mockResolvedValueOnce({
+        ok: true,
+        result: createImageUpdate(),
+      })
+      .mockImplementation(() => new Promise(() => {}));
+
+    const { monitorZaloProvider } = await import("./monitor.js");
+    const abort = new AbortController();
+    const runtime = createRuntimeEnv();
+    const { account, config } = createLifecycleMonitorSetup({
+      accountId: "default",
+      dmPolicy: "pairing",
+      allowFrom: ["allowed-user"],
+    });
+    const run = monitorZaloProvider({
+      token: "zalo-token", // pragma: allowlist secret
+      account,
+      config,
+      runtime,
+      abortSignal: abort.signal,
+    });
+
+    await vi.waitFor(() => expect(sendMessageMock).toHaveBeenCalledTimes(1));
+    expect(fetchRemoteMediaMock).not.toHaveBeenCalled();
+    expect(saveMediaBufferMock).not.toHaveBeenCalled();
+    expect(finalizeInboundContextMock).not.toHaveBeenCalled();
+    expect(recordInboundSessionMock).not.toHaveBeenCalled();
+
+    abort.abort();
+    await run;
+  });
 });

extensions/zalo/src/monitor.ts

@@ -93,11 +93,20 @@ type ZaloMessagePipelineParams = ZaloProcessingContext & {
   text?: string;
   mediaPath?: string;
   mediaType?: string;
+  authorization?: ZaloMessageAuthorizationResult;
 };
 type ZaloImageMessageParams = ZaloProcessingContext & {
   message: ZaloMessage;
   mediaMaxMb: number;
 };
+type ZaloMessageAuthorizationResult = {
+  chatId: string;
+  commandAuthorized: boolean | undefined;
+  isGroup: boolean;
+  rawBody: string;
+  senderId: string;
+  senderName: string | undefined;
+};
 
 function formatZaloError(error: unknown): string {
   if (error instanceof Error) {
@@ -285,6 +294,16 @@ async function handleTextMessage(
 async function handleImageMessage(params: ZaloImageMessageParams): Promise<void> {
   const { message, mediaMaxMb, account, core, runtime } = params;
   const { photo_url, caption } = message;
+  const authorization = await authorizeZaloMessage({
+    ...params,
+    text: caption,
+    // Use a sentinel so auth sees this as an inbound image before the download happens.
+    mediaPath: photo_url ? "__pending_media__" : undefined,
+    mediaType: undefined,
+  });
+  if (!authorization) {
+    return;
+  }
 
   let mediaPath: string | undefined;
   let mediaType: string | undefined;
@@ -308,32 +327,24 @@ async function handleImageMessage(params: ZaloImageMessageParams): Promise<void>
 
   await processMessageWithPipeline({
     ...params,
+    authorization,
     text: caption,
     mediaPath,
     mediaType,
   });
 }
 
-async function processMessageWithPipeline(params: ZaloMessagePipelineParams): Promise<void> {
-  const {
-    message,
-    token,
-    account,
-    config,
-    runtime,
-    core,
-    text,
-    mediaPath,
-    mediaType,
-    statusSink,
-    fetcher,
-  } = params;
+async function authorizeZaloMessage(
+  params: ZaloMessagePipelineParams,
+): Promise<ZaloMessageAuthorizationResult | undefined> {
+  const { message, account, config, runtime, core, text, mediaPath, token, statusSink, fetcher } =
+    params;
   const pairing = createChannelPairingController({
     core,
     channel: "zalo",
     accountId: account.accountId,
   });
-  const { from, chat, message_id, date } = message;
+  const { from, chat } = message;
 
   const isGroup = chat.chat_type === "GROUP";
   const chatId = chat.id;
@@ -436,6 +447,44 @@ async function processMessageWithPipeline(params: ZaloMessagePipelineParams): Pr
     return;
   }
 
+  return {
+    chatId,
+    commandAuthorized,
+    isGroup,
+    rawBody,
+    senderId,
+    senderName,
+  };
+}
+
+async function processMessageWithPipeline(params: ZaloMessagePipelineParams): Promise<void> {
+  const {
+    message,
+    token,
+    account,
+    config,
+    runtime,
+    core,
+    text,
+    mediaPath,
+    mediaType,
+    statusSink,
+    fetcher,
+    authorization: authorizationOverride,
+  } = params;
+  const { message_id, date } = message;
+  const authorization =
+    authorizationOverride ??
+    (await authorizeZaloMessage({
+      ...params,
+      mediaPath,
+      mediaType,
+    }));
+  if (!authorization) {
+    return;
+  }
+  const { isGroup, chatId, senderId, senderName, rawBody, commandAuthorized } = authorization;
+
   const { route, buildEnvelope } = resolveInboundRouteEnvelopeBuilderWithRuntime({
     cfg: config,
     channel: "zalo",