mirror of
https://github.com/openclaw/openclaw.git
synced 2026-05-06 13:00:44 +00:00
fix(zalo): add SSRF guard on outbound photo URLs [AI-assisted] (#69593)
* fix: address issue * fix: address review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address build failures * fix: address PR review feedback * fix: address review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address PR review feedback * fix: address review feedback * fix: address build feedback
This commit is contained in:
Pavan Kumar Gondhi
committed by
GitHub
GitHub
parent
4407df6c03
commit
a65eb1b864
@@ -1,5 +1,13 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { deleteWebhook, getWebhookInfo, sendChatAction, type ZaloFetch } from "./api.js";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const resolvePinnedHostnameWithPolicyMock = vi.fn();
|
||||
|
||||
vi.mock("openclaw/plugin-sdk/ssrf-runtime", () => ({
|
||||
resolvePinnedHostnameWithPolicy: (...args: unknown[]) =>
|
||||
resolvePinnedHostnameWithPolicyMock(...args),
|
||||
}));
|
||||
|
||||
import { deleteWebhook, getWebhookInfo, sendChatAction, sendPhoto, type ZaloFetch } from "./api.js";
|
||||
|
||||
function createOkFetcher() {
|
||||
return vi.fn<ZaloFetch>(async () => new Response(JSON.stringify({ ok: true, result: {} })));
|
||||
@@ -15,6 +23,15 @@ async function expectPostJsonRequest(run: (token: string, fetcher: ZaloFetch) =>
|
||||
}
|
||||
|
||||
describe("Zalo API request methods", () => {
|
||||
beforeEach(() => {
|
||||
resolvePinnedHostnameWithPolicyMock.mockReset();
|
||||
resolvePinnedHostnameWithPolicyMock.mockResolvedValue({
|
||||
hostname: "example.com",
|
||||
addresses: ["93.184.216.34"],
|
||||
lookup: vi.fn(),
|
||||
});
|
||||
});
|
||||
|
||||
it("uses POST for getWebhookInfo", async () => {
|
||||
await expectPostJsonRequest(getWebhookInfo);
|
||||
});
|
||||
@@ -55,4 +72,78 @@ describe("Zalo API request methods", () => {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("validates outbound photo URLs against the SSRF guard before posting", async () => {
|
||||
const fetcher = createOkFetcher();
|
||||
|
||||
await sendPhoto(
|
||||
"test-token",
|
||||
{
|
||||
chat_id: "chat-123",
|
||||
photo: "https://example.com/image.png",
|
||||
},
|
||||
fetcher,
|
||||
);
|
||||
|
||||
expect(resolvePinnedHostnameWithPolicyMock).toHaveBeenCalledWith("example.com", {
|
||||
policy: {},
|
||||
});
|
||||
expect(fetcher).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("blocks private-network photo URLs before they reach the Zalo API", async () => {
|
||||
const fetcher = createOkFetcher();
|
||||
resolvePinnedHostnameWithPolicyMock.mockRejectedValueOnce(
|
||||
new Error("Blocked hostname or private/internal/special-use IP address"),
|
||||
);
|
||||
|
||||
await expect(
|
||||
sendPhoto(
|
||||
"test-token",
|
||||
{
|
||||
chat_id: "chat-123",
|
||||
photo: "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
|
||||
},
|
||||
fetcher,
|
||||
),
|
||||
).rejects.toThrow("Blocked hostname or private/internal/special-use IP address");
|
||||
|
||||
expect(fetcher).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects non-http photo URLs", async () => {
|
||||
const fetcher = createOkFetcher();
|
||||
|
||||
await expect(
|
||||
sendPhoto(
|
||||
"test-token",
|
||||
{
|
||||
chat_id: "chat-123",
|
||||
photo: "file:///etc/passwd",
|
||||
},
|
||||
fetcher,
|
||||
),
|
||||
).rejects.toThrow("Zalo photo URL must use HTTP or HTTPS");
|
||||
|
||||
expect(resolvePinnedHostnameWithPolicyMock).not.toHaveBeenCalled();
|
||||
expect(fetcher).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects non-URL strings", async () => {
|
||||
const fetcher = createOkFetcher();
|
||||
|
||||
await expect(
|
||||
sendPhoto(
|
||||
"test-token",
|
||||
{
|
||||
chat_id: "chat-123",
|
||||
photo: "not a url",
|
||||
},
|
||||
fetcher,
|
||||
),
|
||||
).rejects.toThrow("Zalo photo URL must be an absolute HTTP or HTTPS URL");
|
||||
|
||||
expect(resolvePinnedHostnameWithPolicyMock).not.toHaveBeenCalled();
|
||||
expect(fetcher).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -3,7 +3,10 @@
|
||||
* @see https://bot.zaloplatforms.com/docs
|
||||
*/
|
||||
|
||||
import { resolvePinnedHostnameWithPolicy, type SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime";
|
||||
|
||||
const ZALO_API_BASE = "https://bot-api.zaloplatforms.com";
|
||||
const ZALO_MEDIA_SSRF_POLICY: SsrFPolicy = {};
|
||||
|
||||
export type ZaloFetch = (input: string, init?: RequestInit) => Promise<Response>;
|
||||
|
||||
@@ -172,7 +175,28 @@ export async function sendPhoto(
|
||||
params: ZaloSendPhotoParams,
|
||||
fetcher?: ZaloFetch,
|
||||
): Promise<ZaloApiResponse<ZaloMessage>> {
|
||||
return callZaloApi<ZaloMessage>("sendPhoto", token, params, { fetch: fetcher });
|
||||
const photoUrl = params.photo.trim();
|
||||
let parsedPhotoUrl: URL;
|
||||
try {
|
||||
parsedPhotoUrl = new URL(photoUrl);
|
||||
} catch {
|
||||
throw new Error("Zalo photo URL must be an absolute HTTP or HTTPS URL");
|
||||
}
|
||||
|
||||
if (parsedPhotoUrl.protocol !== "http:" && parsedPhotoUrl.protocol !== "https:") {
|
||||
throw new Error("Zalo photo URL must use HTTP or HTTPS");
|
||||
}
|
||||
|
||||
await resolvePinnedHostnameWithPolicy(parsedPhotoUrl.hostname, {
|
||||
policy: ZALO_MEDIA_SSRF_POLICY,
|
||||
});
|
||||
|
||||
return callZaloApi<ZaloMessage>(
|
||||
"sendPhoto",
|
||||
token,
|
||||
{ ...params, photo: parsedPhotoUrl.href },
|
||||
{ fetch: fetcher },
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -148,14 +148,14 @@ describe("monitorZaloProvider lifecycle", () => {
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(setWebhookMock).toHaveBeenCalledTimes(1));
|
||||
expect(registry.httpRoutes).toHaveLength(1);
|
||||
expect(registry.httpRoutes).toHaveLength(2);
|
||||
|
||||
abort.abort();
|
||||
|
||||
await vi.waitFor(() => expect(deleteWebhookMock).toHaveBeenCalledTimes(1));
|
||||
expect(deleteWebhookMock).toHaveBeenCalledWith("test-token", undefined, 5000);
|
||||
expect(settled).toBe(false);
|
||||
expect(registry.httpRoutes).toHaveLength(1);
|
||||
expect(registry.httpRoutes).toHaveLength(2);
|
||||
|
||||
resolveDeleteWebhook?.();
|
||||
await monitoredRun;
|
||||
|
||||
242
extensions/zalo/src/monitor.polling.media-reply.test.ts
Normal file
242
extensions/zalo/src/monitor.polling.media-reply.test.ts
Normal file
@@ -0,0 +1,242 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { createEmptyPluginRegistry } from "../../../src/plugins/registry-empty.js";
|
||||
import { setActivePluginRegistry } from "../../../src/plugins/runtime.js";
|
||||
import { createRuntimeEnv } from "../../../test/helpers/plugins/runtime-env.js";
|
||||
import type { PluginRuntime } from "../runtime-api.js";
|
||||
import {
|
||||
createLifecycleMonitorSetup,
|
||||
createTextUpdate,
|
||||
} from "../test-support/lifecycle-test-support.js";
|
||||
import {
|
||||
getUpdatesMock,
|
||||
loadLifecycleMonitorModule,
|
||||
resetLifecycleTestState,
|
||||
sendPhotoMock,
|
||||
setLifecycleRuntimeCore,
|
||||
} from "../test-support/monitor-mocks-test-support.js";
|
||||
|
||||
const prepareHostedZaloMediaUrlMock = vi.fn();
|
||||
|
||||
vi.mock("./outbound-media.js", async () => {
|
||||
const actual = await vi.importActual<typeof import("./outbound-media.js")>("./outbound-media.js");
|
||||
return {
|
||||
...actual,
|
||||
prepareHostedZaloMediaUrl: (...args: unknown[]) => prepareHostedZaloMediaUrlMock(...args),
|
||||
};
|
||||
});
|
||||
|
||||
describe("Zalo polling media replies", () => {
|
||||
const finalizeInboundContextMock = vi.fn((ctx: Record<string, unknown>) => ctx);
|
||||
const recordInboundSessionMock = vi.fn(async () => undefined);
|
||||
const resolveAgentRouteMock = vi.fn(() => ({
|
||||
agentId: "main",
|
||||
channel: "zalo",
|
||||
accountId: "acct-zalo-polling-media",
|
||||
sessionKey: "agent:main:zalo:direct:dm-chat-1",
|
||||
mainSessionKey: "agent:main:main",
|
||||
matchedBy: "default",
|
||||
}));
|
||||
const dispatchReplyWithBufferedBlockDispatcherMock = vi.fn();
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetLifecycleTestState();
|
||||
prepareHostedZaloMediaUrlMock.mockReset();
|
||||
prepareHostedZaloMediaUrlMock.mockResolvedValue(
|
||||
"https://example.com/hooks/zalo/media/abc123abc123abc123abc123?token=secret",
|
||||
);
|
||||
dispatchReplyWithBufferedBlockDispatcherMock.mockReset();
|
||||
dispatchReplyWithBufferedBlockDispatcherMock.mockImplementation(
|
||||
async (params: {
|
||||
dispatcherOptions: {
|
||||
deliver: (payload: { text: string; mediaUrl: string }) => Promise<void>;
|
||||
};
|
||||
}) => {
|
||||
await params.dispatcherOptions.deliver({
|
||||
text: "caption text",
|
||||
mediaUrl: "https://example.com/reply-image.png",
|
||||
});
|
||||
},
|
||||
);
|
||||
setLifecycleRuntimeCore({
|
||||
routing: {
|
||||
resolveAgentRoute:
|
||||
resolveAgentRouteMock as unknown as PluginRuntime["channel"]["routing"]["resolveAgentRoute"],
|
||||
},
|
||||
reply: {
|
||||
finalizeInboundContext:
|
||||
finalizeInboundContextMock as unknown as PluginRuntime["channel"]["reply"]["finalizeInboundContext"],
|
||||
dispatchReplyWithBufferedBlockDispatcher:
|
||||
dispatchReplyWithBufferedBlockDispatcherMock as unknown as PluginRuntime["channel"]["reply"]["dispatchReplyWithBufferedBlockDispatcher"],
|
||||
},
|
||||
session: {
|
||||
recordInboundSession:
|
||||
recordInboundSessionMock as unknown as PluginRuntime["channel"]["session"]["recordInboundSession"],
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await resetLifecycleTestState();
|
||||
});
|
||||
|
||||
it("hosts and sends media replies while polling when a webhook URL is configured", async () => {
|
||||
const registry = createEmptyPluginRegistry();
|
||||
setActivePluginRegistry(registry);
|
||||
getUpdatesMock
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
result: createTextUpdate({
|
||||
messageId: "polling-media-1",
|
||||
userId: "user-1",
|
||||
userName: "User One",
|
||||
chatId: "dm-chat-1",
|
||||
text: "send media",
|
||||
}),
|
||||
})
|
||||
.mockImplementation(() => new Promise(() => {}));
|
||||
|
||||
const { monitorZaloProvider } = await loadLifecycleMonitorModule();
|
||||
const abort = new AbortController();
|
||||
const runtime = createRuntimeEnv();
|
||||
const { account, config } = createLifecycleMonitorSetup({
|
||||
accountId: "acct-zalo-polling-media",
|
||||
dmPolicy: "open",
|
||||
webhookUrl: "https://example.com/hooks/zalo",
|
||||
});
|
||||
const run = monitorZaloProvider({
|
||||
token: "zalo-token",
|
||||
account,
|
||||
config,
|
||||
runtime,
|
||||
abortSignal: abort.signal,
|
||||
});
|
||||
|
||||
try {
|
||||
await vi.waitFor(() => expect(sendPhotoMock).toHaveBeenCalledTimes(1));
|
||||
|
||||
expect(registry.httpRoutes).toHaveLength(1);
|
||||
expect(prepareHostedZaloMediaUrlMock).toHaveBeenCalledWith({
|
||||
mediaUrl: "https://example.com/reply-image.png",
|
||||
webhookUrl: "https://example.com/hooks/zalo",
|
||||
webhookPath: "/hooks/zalo",
|
||||
maxBytes: 5 * 1024 * 1024,
|
||||
proxyUrl: undefined,
|
||||
});
|
||||
expect(sendPhotoMock).toHaveBeenCalledWith(
|
||||
"zalo-token",
|
||||
{
|
||||
chat_id: "dm-chat-1",
|
||||
photo: "https://example.com/hooks/zalo/media/abc123abc123abc123abc123?token=secret",
|
||||
caption: "caption text",
|
||||
},
|
||||
undefined,
|
||||
);
|
||||
} finally {
|
||||
abort.abort();
|
||||
await run;
|
||||
}
|
||||
|
||||
expect(registry.httpRoutes).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("sends media replies directly when webhook hosting is not configured", async () => {
|
||||
const registry = createEmptyPluginRegistry();
|
||||
setActivePluginRegistry(registry);
|
||||
getUpdatesMock
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
result: createTextUpdate({
|
||||
messageId: "polling-media-2",
|
||||
userId: "user-2",
|
||||
userName: "User Two",
|
||||
chatId: "dm-chat-2",
|
||||
text: "send media directly",
|
||||
}),
|
||||
})
|
||||
.mockImplementation(() => new Promise(() => {}));
|
||||
|
||||
const { monitorZaloProvider } = await loadLifecycleMonitorModule();
|
||||
const abort = new AbortController();
|
||||
const runtime = createRuntimeEnv();
|
||||
const { account, config } = createLifecycleMonitorSetup({
|
||||
accountId: "acct-zalo-polling-direct-media",
|
||||
dmPolicy: "open",
|
||||
webhookUrl: "",
|
||||
});
|
||||
const run = monitorZaloProvider({
|
||||
token: "zalo-token",
|
||||
account,
|
||||
config,
|
||||
runtime,
|
||||
abortSignal: abort.signal,
|
||||
});
|
||||
|
||||
try {
|
||||
await vi.waitFor(() => expect(sendPhotoMock).toHaveBeenCalledTimes(1));
|
||||
|
||||
expect(prepareHostedZaloMediaUrlMock).not.toHaveBeenCalled();
|
||||
expect(sendPhotoMock).toHaveBeenCalledWith(
|
||||
"zalo-token",
|
||||
{
|
||||
chat_id: "dm-chat-2",
|
||||
photo: "https://example.com/reply-image.png",
|
||||
caption: "caption text",
|
||||
},
|
||||
undefined,
|
||||
);
|
||||
} finally {
|
||||
abort.abort();
|
||||
await run;
|
||||
}
|
||||
});
|
||||
|
||||
it("re-registers the hosted media route after the active registry swaps", async () => {
|
||||
const firstRegistry = createEmptyPluginRegistry();
|
||||
setActivePluginRegistry(firstRegistry);
|
||||
getUpdatesMock.mockImplementation(() => new Promise(() => {}));
|
||||
|
||||
const { monitorZaloProvider } = await loadLifecycleMonitorModule();
|
||||
const firstAbort = new AbortController();
|
||||
const firstRuntime = createRuntimeEnv();
|
||||
const { account, config } = createLifecycleMonitorSetup({
|
||||
accountId: "acct-zalo-polling-media",
|
||||
dmPolicy: "open",
|
||||
webhookUrl: "https://example.com/hooks/zalo",
|
||||
});
|
||||
const firstRun = monitorZaloProvider({
|
||||
token: "zalo-token",
|
||||
account,
|
||||
config,
|
||||
runtime: firstRuntime,
|
||||
abortSignal: firstAbort.signal,
|
||||
});
|
||||
|
||||
const secondRegistry = createEmptyPluginRegistry();
|
||||
const secondAbort = new AbortController();
|
||||
const secondRuntime = createRuntimeEnv();
|
||||
let secondRun: Promise<void> | undefined;
|
||||
|
||||
try {
|
||||
await vi.waitFor(() => expect(firstRegistry.httpRoutes).toHaveLength(1));
|
||||
|
||||
setActivePluginRegistry(secondRegistry);
|
||||
secondRun = monitorZaloProvider({
|
||||
token: "zalo-token",
|
||||
account,
|
||||
config,
|
||||
runtime: secondRuntime,
|
||||
abortSignal: secondAbort.signal,
|
||||
});
|
||||
|
||||
await vi.waitFor(() => expect(secondRegistry.httpRoutes).toHaveLength(1));
|
||||
} finally {
|
||||
firstAbort.abort();
|
||||
secondAbort.abort();
|
||||
await firstRun;
|
||||
await secondRun;
|
||||
}
|
||||
|
||||
expect(firstRegistry.httpRoutes).toHaveLength(0);
|
||||
expect(secondRegistry.httpRoutes).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
@@ -26,8 +26,9 @@ import {
|
||||
createChannelPairingController,
|
||||
createChannelReplyPipeline,
|
||||
deliverTextOrMediaReply,
|
||||
resolveWebhookPath,
|
||||
logTypingFailure,
|
||||
registerPluginHttpRoute,
|
||||
resolveWebhookPath,
|
||||
resolveDefaultGroupPolicy,
|
||||
resolveDirectDmAuthorizationOutcome,
|
||||
resolveInboundRouteEnvelopeBuilderWithRuntime,
|
||||
@@ -38,6 +39,11 @@ import {
|
||||
import { getZaloRuntime } from "./runtime.js";
|
||||
export type { ZaloRuntimeEnv } from "./monitor.types.js";
|
||||
import type { ZaloRuntimeEnv } from "./monitor.types.js";
|
||||
import {
|
||||
prepareHostedZaloMediaUrl,
|
||||
resolveHostedZaloMediaRoutePrefix,
|
||||
tryHandleHostedZaloMediaRequest,
|
||||
} from "./outbound-media.js";
|
||||
|
||||
export type ZaloMonitorOptions = {
|
||||
token: string;
|
||||
@@ -67,25 +73,90 @@ type ZaloProcessingContext = {
|
||||
config: OpenClawConfig;
|
||||
runtime: ZaloRuntimeEnv;
|
||||
core: ZaloCoreRuntime;
|
||||
mediaMaxMb: number;
|
||||
canHostMedia: boolean;
|
||||
webhookUrl?: string;
|
||||
webhookPath?: string;
|
||||
statusSink?: ZaloStatusSink;
|
||||
fetcher?: ZaloFetch;
|
||||
};
|
||||
type ZaloPollingLoopParams = ZaloProcessingContext & {
|
||||
abortSignal: AbortSignal;
|
||||
isStopped: () => boolean;
|
||||
mediaMaxMb: number;
|
||||
};
|
||||
type ZaloUpdateProcessingParams = ZaloProcessingContext & {
|
||||
update: ZaloUpdate;
|
||||
mediaMaxMb: number;
|
||||
};
|
||||
|
||||
let zaloWebhookModulePromise: Promise<ZaloWebhookModule> | undefined;
|
||||
const hostedMediaRouteRefs = new Map<string, { count: number; unregisters: Array<() => void> }>();
|
||||
|
||||
function loadZaloWebhookModule(): Promise<ZaloWebhookModule> {
|
||||
zaloWebhookModulePromise ??= import("./monitor.webhook.js");
|
||||
return zaloWebhookModulePromise;
|
||||
}
|
||||
|
||||
function registerSharedHostedMediaRoute(params: {
|
||||
path: string;
|
||||
accountId: string;
|
||||
log?: (message: string) => void;
|
||||
}): () => void {
|
||||
const unregister = registerPluginHttpRoute({
|
||||
auth: "plugin",
|
||||
match: "prefix",
|
||||
path: params.path,
|
||||
replaceExisting: true,
|
||||
pluginId: "zalo",
|
||||
source: "zalo-hosted-media",
|
||||
accountId: params.accountId,
|
||||
log: params.log,
|
||||
handler: async (req, res) => {
|
||||
const handled = await tryHandleHostedZaloMediaRequest(req, res);
|
||||
if (!handled && !res.headersSent) {
|
||||
res.statusCode = 404;
|
||||
res.setHeader("Content-Type", "text/plain; charset=utf-8");
|
||||
res.end("Not Found");
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
const existing = hostedMediaRouteRefs.get(params.path);
|
||||
if (existing) {
|
||||
existing.count += 1;
|
||||
existing.unregisters.push(unregister);
|
||||
return () => {
|
||||
const current = hostedMediaRouteRefs.get(params.path);
|
||||
if (!current) {
|
||||
return;
|
||||
}
|
||||
if (current.count > 1) {
|
||||
current.count -= 1;
|
||||
return;
|
||||
}
|
||||
hostedMediaRouteRefs.delete(params.path);
|
||||
for (const unregisterHandle of current.unregisters) {
|
||||
unregisterHandle();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
hostedMediaRouteRefs.set(params.path, { count: 1, unregisters: [unregister] });
|
||||
return () => {
|
||||
const current = hostedMediaRouteRefs.get(params.path);
|
||||
if (!current) {
|
||||
return;
|
||||
}
|
||||
if (current.count > 1) {
|
||||
current.count -= 1;
|
||||
return;
|
||||
}
|
||||
hostedMediaRouteRefs.delete(params.path);
|
||||
for (const unregisterHandle of current.unregisters) {
|
||||
unregisterHandle();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
type ZaloMessagePipelineParams = ZaloProcessingContext & {
|
||||
message: ZaloMessage;
|
||||
text?: string;
|
||||
@@ -95,7 +166,6 @@ type ZaloMessagePipelineParams = ZaloProcessingContext & {
|
||||
};
|
||||
type ZaloImageMessageParams = ZaloProcessingContext & {
|
||||
message: ZaloMessage;
|
||||
mediaMaxMb: number;
|
||||
};
|
||||
type ZaloMessageAuthorizationResult = {
|
||||
chatId: string;
|
||||
@@ -148,6 +218,9 @@ export async function handleZaloWebhookRequest(
|
||||
runtime: target.runtime,
|
||||
core: target.core as ZaloCoreRuntime,
|
||||
mediaMaxMb: target.mediaMaxMb,
|
||||
canHostMedia: target.canHostMedia,
|
||||
webhookUrl: target.webhookUrl,
|
||||
webhookPath: target.webhookPath,
|
||||
statusSink: target.statusSink,
|
||||
fetcher: target.fetcher,
|
||||
});
|
||||
@@ -161,9 +234,12 @@ function startPollingLoop(params: ZaloPollingLoopParams) {
|
||||
config,
|
||||
runtime,
|
||||
core,
|
||||
mediaMaxMb,
|
||||
canHostMedia,
|
||||
webhookUrl,
|
||||
webhookPath,
|
||||
abortSignal,
|
||||
isStopped,
|
||||
mediaMaxMb,
|
||||
statusSink,
|
||||
fetcher,
|
||||
} = params;
|
||||
@@ -175,6 +251,9 @@ function startPollingLoop(params: ZaloPollingLoopParams) {
|
||||
runtime,
|
||||
core,
|
||||
mediaMaxMb,
|
||||
canHostMedia,
|
||||
webhookUrl,
|
||||
webhookPath,
|
||||
statusSink,
|
||||
fetcher,
|
||||
};
|
||||
@@ -188,6 +267,9 @@ function startPollingLoop(params: ZaloPollingLoopParams) {
|
||||
|
||||
try {
|
||||
const response = await getUpdates(token, { timeout: pollTimeout }, fetcher);
|
||||
if (isStopped() || abortSignal.aborted) {
|
||||
return undefined;
|
||||
}
|
||||
if (response.ok && response.result) {
|
||||
statusSink?.({ lastInboundAt: Date.now() });
|
||||
await processUpdate({
|
||||
@@ -215,7 +297,19 @@ function startPollingLoop(params: ZaloPollingLoopParams) {
|
||||
async function processUpdate(params: ZaloUpdateProcessingParams): Promise<void> {
|
||||
const { update, token, account, config, runtime, core, mediaMaxMb, statusSink, fetcher } = params;
|
||||
const { event_name, message } = update;
|
||||
const sharedContext = { token, account, config, runtime, core, statusSink, fetcher };
|
||||
const sharedContext = {
|
||||
token,
|
||||
account,
|
||||
config,
|
||||
runtime,
|
||||
core,
|
||||
mediaMaxMb,
|
||||
canHostMedia: params.canHostMedia,
|
||||
webhookUrl: params.webhookUrl,
|
||||
webhookPath: params.webhookPath,
|
||||
statusSink,
|
||||
fetcher,
|
||||
};
|
||||
if (!message) {
|
||||
return undefined;
|
||||
}
|
||||
@@ -566,6 +660,11 @@ async function processMessageWithPipeline(params: ZaloMessagePipelineParams): Pr
|
||||
runtime,
|
||||
core,
|
||||
config,
|
||||
webhookUrl: params.webhookUrl,
|
||||
webhookPath: params.webhookPath,
|
||||
proxyUrl: account.config.proxy,
|
||||
mediaMaxBytes: params.mediaMaxMb * 1024 * 1024,
|
||||
canHostMedia: params.canHostMedia,
|
||||
accountId: account.accountId,
|
||||
statusSink,
|
||||
fetcher,
|
||||
@@ -589,12 +688,32 @@ async function deliverZaloReply(params: {
|
||||
runtime: ZaloRuntimeEnv;
|
||||
core: ZaloCoreRuntime;
|
||||
config: OpenClawConfig;
|
||||
webhookUrl?: string;
|
||||
webhookPath?: string;
|
||||
proxyUrl?: string;
|
||||
mediaMaxBytes: number;
|
||||
canHostMedia: boolean;
|
||||
accountId?: string;
|
||||
statusSink?: ZaloStatusSink;
|
||||
fetcher?: ZaloFetch;
|
||||
tableMode?: MarkdownTableMode;
|
||||
}): Promise<void> {
|
||||
const { payload, token, chatId, runtime, core, config, accountId, statusSink, fetcher } = params;
|
||||
const {
|
||||
payload,
|
||||
token,
|
||||
chatId,
|
||||
runtime,
|
||||
core,
|
||||
config,
|
||||
webhookUrl,
|
||||
webhookPath,
|
||||
proxyUrl,
|
||||
mediaMaxBytes,
|
||||
canHostMedia,
|
||||
accountId,
|
||||
statusSink,
|
||||
fetcher,
|
||||
} = params;
|
||||
const tableMode = params.tableMode ?? "code";
|
||||
const reply = resolveSendableOutboundReplyParts(payload, {
|
||||
text: core.channel.text.convertMarkdownTables(payload.text ?? "", tableMode),
|
||||
@@ -614,7 +733,17 @@ async function deliverZaloReply(params: {
|
||||
}
|
||||
},
|
||||
sendMedia: async ({ mediaUrl, caption }) => {
|
||||
await sendPhoto(token, { chat_id: chatId, photo: mediaUrl, caption }, fetcher);
|
||||
const sendableMediaUrl =
|
||||
canHostMedia && webhookUrl && webhookPath
|
||||
? await prepareHostedZaloMediaUrl({
|
||||
mediaUrl,
|
||||
webhookUrl,
|
||||
webhookPath,
|
||||
maxBytes: mediaMaxBytes,
|
||||
proxyUrl,
|
||||
})
|
||||
: mediaUrl;
|
||||
await sendPhoto(token, { chat_id: chatId, photo: sendableMediaUrl, caption }, fetcher);
|
||||
statusSink?.({ lastOutboundAt: Date.now() });
|
||||
},
|
||||
onMediaError: (error) => {
|
||||
@@ -644,6 +773,23 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise<
|
||||
const effectiveMediaMaxMb = account.config.mediaMaxMb ?? DEFAULT_MEDIA_MAX_MB;
|
||||
const fetcher = fetcherOverride ?? resolveZaloProxyFetch(account.config.proxy);
|
||||
const mode = useWebhook ? "webhook" : "polling";
|
||||
const effectiveWebhookUrl = normalizeWebhookUrl(webhookUrl ?? account.config.webhookUrl);
|
||||
const effectiveWebhookPath =
|
||||
effectiveWebhookUrl || webhookPath?.trim() || account.config.webhookPath?.trim()
|
||||
? (resolveWebhookPath({
|
||||
webhookPath: webhookPath ?? account.config.webhookPath,
|
||||
webhookUrl: effectiveWebhookUrl,
|
||||
defaultPath: null,
|
||||
}) ?? undefined)
|
||||
: undefined;
|
||||
const canHostMedia = Boolean(effectiveWebhookUrl && effectiveWebhookPath);
|
||||
const hostedMediaRoutePath =
|
||||
canHostMedia && effectiveWebhookUrl
|
||||
? resolveHostedZaloMediaRoutePrefix({
|
||||
webhookUrl: effectiveWebhookUrl,
|
||||
webhookPath: effectiveWebhookPath,
|
||||
})
|
||||
: undefined;
|
||||
|
||||
let stopped = false;
|
||||
const stopHandlers: Array<() => void> = [];
|
||||
@@ -658,33 +804,49 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise<
|
||||
handler();
|
||||
}
|
||||
};
|
||||
const stopOnAbort = () => {
|
||||
if (!useWebhook) {
|
||||
stop();
|
||||
}
|
||||
};
|
||||
|
||||
abortSignal.addEventListener("abort", stopOnAbort, { once: true });
|
||||
|
||||
runtime.log?.(
|
||||
`[${account.accountId}] Zalo provider init mode=${mode} mediaMaxMb=${String(effectiveMediaMaxMb)}`,
|
||||
);
|
||||
|
||||
try {
|
||||
if (hostedMediaRoutePath) {
|
||||
const unregisterHostedMediaRoute = registerSharedHostedMediaRoute({
|
||||
path: hostedMediaRoutePath,
|
||||
accountId: account.accountId,
|
||||
log: runtime.log,
|
||||
});
|
||||
stopHandlers.push(unregisterHostedMediaRoute);
|
||||
}
|
||||
|
||||
if (useWebhook) {
|
||||
const { registerZaloWebhookTarget } = await loadZaloWebhookModule();
|
||||
if (!webhookUrl || !webhookSecret) {
|
||||
if (!effectiveWebhookUrl || !webhookSecret) {
|
||||
throw new Error("Zalo webhookUrl and webhookSecret are required for webhook mode");
|
||||
}
|
||||
if (!webhookUrl.startsWith("https://")) {
|
||||
if (!effectiveWebhookUrl.startsWith("https://")) {
|
||||
throw new Error("Zalo webhook URL must use HTTPS");
|
||||
}
|
||||
if (webhookSecret.length < 8 || webhookSecret.length > 256) {
|
||||
throw new Error("Zalo webhook secret must be 8-256 characters");
|
||||
}
|
||||
|
||||
const path = resolveWebhookPath({ webhookPath, webhookUrl, defaultPath: null });
|
||||
const path = effectiveWebhookPath;
|
||||
if (!path) {
|
||||
throw new Error("Zalo webhookPath could not be derived");
|
||||
}
|
||||
|
||||
runtime.log?.(
|
||||
`[${account.accountId}] Zalo configuring webhook path=${path} target=${describeWebhookTarget(webhookUrl)}`,
|
||||
`[${account.accountId}] Zalo configuring webhook path=${path} target=${describeWebhookTarget(effectiveWebhookUrl)}`,
|
||||
);
|
||||
await setWebhook(token, { url: webhookUrl, secret_token: webhookSecret }, fetcher);
|
||||
await setWebhook(token, { url: effectiveWebhookUrl, secret_token: webhookSecret }, fetcher);
|
||||
let webhookCleanupPromise: Promise<void> | undefined;
|
||||
cleanupWebhook = async () => {
|
||||
if (!webhookCleanupPromise) {
|
||||
@@ -714,9 +876,12 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise<
|
||||
runtime,
|
||||
core,
|
||||
path,
|
||||
webhookUrl: effectiveWebhookUrl,
|
||||
webhookPath: path,
|
||||
secret: webhookSecret,
|
||||
statusSink: (patch) => statusSink?.(patch),
|
||||
mediaMaxMb: effectiveMediaMaxMb,
|
||||
canHostMedia,
|
||||
fetcher,
|
||||
},
|
||||
{
|
||||
@@ -780,6 +945,9 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise<
|
||||
config,
|
||||
runtime,
|
||||
core,
|
||||
canHostMedia,
|
||||
webhookUrl: effectiveWebhookUrl,
|
||||
webhookPath: effectiveWebhookPath,
|
||||
abortSignal,
|
||||
isStopped: () => stopped,
|
||||
mediaMaxMb: effectiveMediaMaxMb,
|
||||
@@ -794,6 +962,7 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise<
|
||||
);
|
||||
throw err;
|
||||
} finally {
|
||||
abortSignal.removeEventListener("abort", stopOnAbort);
|
||||
await cleanupWebhook?.();
|
||||
stop();
|
||||
runtime.log?.(`[${account.accountId}] Zalo provider stopped mode=${mode}`);
|
||||
@@ -803,4 +972,5 @@ export async function monitorZaloProvider(options: ZaloMonitorOptions): Promise<
|
||||
export const __testing = {
|
||||
evaluateZaloGroupAccess,
|
||||
resolveZaloRuntimeGroupPolicy,
|
||||
clearHostedMediaRouteRefsForTest: () => hostedMediaRouteRefs.clear(),
|
||||
};
|
||||
|
||||
@@ -62,7 +62,10 @@ function registerTarget(params: {
|
||||
core: params.core ?? ({} as PluginRuntime),
|
||||
secret: params.secret ?? "secret",
|
||||
path: params.path,
|
||||
webhookUrl: `https://example.com${params.path}`,
|
||||
webhookPath: params.path,
|
||||
mediaMaxMb: 5,
|
||||
canHostMedia: true,
|
||||
statusSink: params.statusSink,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -31,7 +31,10 @@ export type ZaloWebhookTarget = {
|
||||
core: unknown;
|
||||
secret: string;
|
||||
path: string;
|
||||
webhookUrl: string;
|
||||
webhookPath: string;
|
||||
mediaMaxMb: number;
|
||||
canHostMedia: boolean;
|
||||
statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
|
||||
fetcher?: ZaloFetch;
|
||||
};
|
||||
|
||||
182
extensions/zalo/src/outbound-media.test.ts
Normal file
182
extensions/zalo/src/outbound-media.test.ts
Normal file
@@ -0,0 +1,182 @@
|
||||
import { stat } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const loadOutboundMediaFromUrlMock = vi.fn();
|
||||
|
||||
vi.mock("openclaw/plugin-sdk/outbound-media", () => ({
|
||||
loadOutboundMediaFromUrl: (...args: unknown[]) => loadOutboundMediaFromUrlMock(...args),
|
||||
}));
|
||||
|
||||
import {
|
||||
clearHostedZaloMediaForTest,
|
||||
prepareHostedZaloMediaUrl,
|
||||
resolveHostedZaloMediaRoutePrefix,
|
||||
tryHandleHostedZaloMediaRequest,
|
||||
} from "./outbound-media.js";
|
||||
|
||||
function createMockResponse() {
|
||||
const headers = new Map<string, string>();
|
||||
return {
|
||||
headers,
|
||||
res: {
|
||||
statusCode: 200,
|
||||
setHeader(name: string, value: string) {
|
||||
headers.set(name, value);
|
||||
},
|
||||
end: vi.fn(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("zalo outbound hosted media", () => {
|
||||
beforeEach(() => {
|
||||
clearHostedZaloMediaForTest();
|
||||
loadOutboundMediaFromUrlMock.mockReset();
|
||||
loadOutboundMediaFromUrlMock.mockResolvedValue({
|
||||
buffer: Buffer.from("image-bytes"),
|
||||
contentType: "image/png",
|
||||
fileName: "photo.png",
|
||||
});
|
||||
});
|
||||
|
||||
it("loads outbound media under OpenClaw control and returns a hosted URL", async () => {
|
||||
const hostedUrl = await prepareHostedZaloMediaUrl({
|
||||
mediaUrl: "https://example.com/photo.png",
|
||||
webhookUrl: "https://gateway.example.com/zalo-webhook",
|
||||
maxBytes: 1024,
|
||||
});
|
||||
|
||||
expect(loadOutboundMediaFromUrlMock).toHaveBeenCalledWith("https://example.com/photo.png", {
|
||||
maxBytes: 1024,
|
||||
});
|
||||
expect(hostedUrl).toMatch(
|
||||
/^https:\/\/gateway\.example\.com\/zalo-webhook\/media\/[a-f0-9]+\?token=[a-f0-9]+$/,
|
||||
);
|
||||
});
|
||||
|
||||
it("passes proxy-aware fetch options into hosted media downloads", async () => {
|
||||
await prepareHostedZaloMediaUrl({
|
||||
mediaUrl: "https://example.com/photo.png",
|
||||
webhookUrl: "https://gateway.example.com/zalo-webhook",
|
||||
maxBytes: 1024,
|
||||
proxyUrl: "http://proxy.example:8080",
|
||||
});
|
||||
|
||||
expect(loadOutboundMediaFromUrlMock).toHaveBeenCalledWith("https://example.com/photo.png", {
|
||||
maxBytes: 1024,
|
||||
proxyUrl: "http://proxy.example:8080",
|
||||
});
|
||||
});
|
||||
|
||||
it("creates hosted media storage with private filesystem permissions", async () => {
|
||||
const hostedUrl = await prepareHostedZaloMediaUrl({
|
||||
mediaUrl: "https://example.com/photo.png",
|
||||
webhookUrl: "https://gateway.example.com/zalo-webhook",
|
||||
maxBytes: 1024,
|
||||
});
|
||||
|
||||
if (process.platform === "win32") {
|
||||
expect(hostedUrl).toContain("/zalo-webhook/media/");
|
||||
return;
|
||||
}
|
||||
|
||||
const { pathname } = new URL(hostedUrl);
|
||||
const id = pathname.split("/").pop();
|
||||
expect(id).toBeTruthy();
|
||||
|
||||
const storageDir = join(tmpdir(), "openclaw-zalo-outbound-media");
|
||||
const [dirStats, metadataStats, bufferStats] = await Promise.all([
|
||||
stat(storageDir),
|
||||
stat(join(storageDir, `${id}.json`)),
|
||||
stat(join(storageDir, `${id}.bin`)),
|
||||
]);
|
||||
|
||||
expect(dirStats.mode & 0o777).toBe(0o700);
|
||||
expect(metadataStats.mode & 0o777).toBe(0o600);
|
||||
expect(bufferStats.mode & 0o777).toBe(0o600);
|
||||
});
|
||||
|
||||
it("preserves the root webhook path when deriving the hosted media route", () => {
|
||||
expect(
|
||||
resolveHostedZaloMediaRoutePrefix({
|
||||
webhookUrl: "https://gateway.example.com/",
|
||||
}),
|
||||
).toBe("/media");
|
||||
});
|
||||
|
||||
it("serves hosted media once when the route token matches", async () => {
|
||||
const hostedUrl = await prepareHostedZaloMediaUrl({
|
||||
mediaUrl: "https://example.com/photo.png",
|
||||
webhookUrl: "https://gateway.example.com/zalo-webhook",
|
||||
maxBytes: 1024,
|
||||
});
|
||||
const { pathname, search } = new URL(hostedUrl);
|
||||
const response = createMockResponse();
|
||||
|
||||
const handled = await tryHandleHostedZaloMediaRequest(
|
||||
{
|
||||
method: "GET",
|
||||
url: `${pathname}${search}`,
|
||||
} as never,
|
||||
response.res as never,
|
||||
);
|
||||
|
||||
expect(handled).toBe(true);
|
||||
expect(response.res.statusCode).toBe(200);
|
||||
expect(response.headers.get("Content-Type")).toBe("image/png");
|
||||
expect(response.res.end).toHaveBeenCalledWith(Buffer.from("image-bytes"));
|
||||
|
||||
const secondResponse = createMockResponse();
|
||||
const handledAgain = await tryHandleHostedZaloMediaRequest(
|
||||
{
|
||||
method: "GET",
|
||||
url: `${pathname}${search}`,
|
||||
} as never,
|
||||
secondResponse.res as never,
|
||||
);
|
||||
|
||||
expect(handledAgain).toBe(true);
|
||||
expect(secondResponse.res.statusCode).toBe(404);
|
||||
});
|
||||
|
||||
it("rejects hosted media requests with the wrong token", async () => {
|
||||
const hostedUrl = await prepareHostedZaloMediaUrl({
|
||||
mediaUrl: "https://example.com/photo.png",
|
||||
webhookUrl: "https://gateway.example.com/custom/zalo",
|
||||
webhookPath: "/custom/zalo-hook",
|
||||
maxBytes: 1024,
|
||||
});
|
||||
const pathname = new URL(hostedUrl).pathname;
|
||||
const response = createMockResponse();
|
||||
|
||||
const handled = await tryHandleHostedZaloMediaRequest(
|
||||
{
|
||||
method: "GET",
|
||||
url: `${pathname}?token=wrong`,
|
||||
} as never,
|
||||
response.res as never,
|
||||
);
|
||||
|
||||
expect(handled).toBe(true);
|
||||
expect(response.res.statusCode).toBe(401);
|
||||
expect(response.res.end).toHaveBeenCalledWith("Unauthorized");
|
||||
});
|
||||
|
||||
it("rejects malformed hosted media ids before touching disk", async () => {
|
||||
const response = createMockResponse();
|
||||
|
||||
const handled = await tryHandleHostedZaloMediaRequest(
|
||||
{
|
||||
method: "GET",
|
||||
url: "/zalo-webhook/media/not-a-valid-hex-id?token=wrong",
|
||||
} as never,
|
||||
response.res as never,
|
||||
);
|
||||
|
||||
expect(handled).toBe(true);
|
||||
expect(response.res.statusCode).toBe(404);
|
||||
expect(response.res.end).toHaveBeenCalledWith("Not Found");
|
||||
});
|
||||
});
|
||||
238
extensions/zalo/src/outbound-media.ts
Normal file
238
extensions/zalo/src/outbound-media.ts
Normal file
@@ -0,0 +1,238 @@
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { rmSync } from "node:fs";
|
||||
import { chmod, mkdir, readdir, readFile, stat, unlink, writeFile } from "node:fs/promises";
|
||||
import type { IncomingMessage, ServerResponse } from "node:http";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { loadOutboundMediaFromUrl } from "openclaw/plugin-sdk/outbound-media";
|
||||
import { resolveWebhookPath } from "./runtime-api.js";
|
||||
|
||||
const ZALO_OUTBOUND_MEDIA_TTL_MS = 2 * 60_000;
|
||||
const ZALO_OUTBOUND_MEDIA_SEGMENT = "media";
|
||||
const ZALO_OUTBOUND_MEDIA_PREFIX = `/${ZALO_OUTBOUND_MEDIA_SEGMENT}/`;
|
||||
const ZALO_OUTBOUND_MEDIA_DIR = join(tmpdir(), "openclaw-zalo-outbound-media");
|
||||
const ZALO_OUTBOUND_MEDIA_ID_RE = /^[a-f0-9]{24}$/;
|
||||
|
||||
type HostedZaloMediaMetadata = {
|
||||
routePath: string;
|
||||
token: string;
|
||||
contentType?: string;
|
||||
expiresAt: number;
|
||||
};
|
||||
|
||||
function resolveHostedZaloMediaMetadataPath(id: string): string {
|
||||
return join(ZALO_OUTBOUND_MEDIA_DIR, `${id}.json`);
|
||||
}
|
||||
|
||||
function resolveHostedZaloMediaBufferPath(id: string): string {
|
||||
return join(ZALO_OUTBOUND_MEDIA_DIR, `${id}.bin`);
|
||||
}
|
||||
|
||||
function createHostedZaloMediaId(): string {
|
||||
return randomBytes(12).toString("hex");
|
||||
}
|
||||
|
||||
function createHostedZaloMediaToken(): string {
|
||||
return randomBytes(24).toString("hex");
|
||||
}
|
||||
|
||||
async function ensureHostedZaloMediaDir(): Promise<void> {
|
||||
await mkdir(ZALO_OUTBOUND_MEDIA_DIR, { recursive: true, mode: 0o700 });
|
||||
await chmod(ZALO_OUTBOUND_MEDIA_DIR, 0o700).catch(() => undefined);
|
||||
}
|
||||
|
||||
async function deleteHostedZaloMediaEntry(id: string): Promise<void> {
|
||||
await Promise.all([
|
||||
unlink(resolveHostedZaloMediaMetadataPath(id)).catch(() => undefined),
|
||||
unlink(resolveHostedZaloMediaBufferPath(id)).catch(() => undefined),
|
||||
]);
|
||||
}
|
||||
|
||||
async function cleanupExpiredHostedZaloMedia(nowMs = Date.now()): Promise<void> {
|
||||
let fileNames: string[];
|
||||
try {
|
||||
fileNames = await readdir(ZALO_OUTBOUND_MEDIA_DIR);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
await Promise.all(
|
||||
fileNames
|
||||
.filter((fileName) => fileName.endsWith(".json"))
|
||||
.map(async (fileName) => {
|
||||
const id = fileName.slice(0, -5);
|
||||
try {
|
||||
const metadataRaw = await readFile(resolveHostedZaloMediaMetadataPath(id), "utf8");
|
||||
const metadata = JSON.parse(metadataRaw) as HostedZaloMediaMetadata;
|
||||
if (metadata.expiresAt <= nowMs) {
|
||||
await deleteHostedZaloMediaEntry(id);
|
||||
}
|
||||
} catch {
|
||||
await deleteHostedZaloMediaEntry(id);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async function readHostedZaloMediaEntry(id: string): Promise<{
|
||||
metadata: HostedZaloMediaMetadata;
|
||||
buffer: Buffer;
|
||||
} | null> {
|
||||
try {
|
||||
const [metadataRaw, buffer] = await Promise.all([
|
||||
readFile(resolveHostedZaloMediaMetadataPath(id), "utf8"),
|
||||
readFile(resolveHostedZaloMediaBufferPath(id)),
|
||||
]);
|
||||
return {
|
||||
metadata: JSON.parse(metadataRaw) as HostedZaloMediaMetadata,
|
||||
buffer,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function resolveHostedZaloMediaRoutePrefix(params: {
|
||||
webhookUrl: string;
|
||||
webhookPath?: string;
|
||||
}): string {
|
||||
const webhookRoutePath = resolveWebhookPath({
|
||||
webhookPath: params.webhookPath,
|
||||
webhookUrl: params.webhookUrl,
|
||||
defaultPath: null,
|
||||
});
|
||||
if (!webhookRoutePath) {
|
||||
throw new Error("Zalo webhookPath could not be derived for outbound media hosting");
|
||||
}
|
||||
return webhookRoutePath === "/"
|
||||
? `/${ZALO_OUTBOUND_MEDIA_SEGMENT}`
|
||||
: `${webhookRoutePath}/${ZALO_OUTBOUND_MEDIA_SEGMENT}`;
|
||||
}
|
||||
|
||||
function resolveHostedZaloMediaRoutePath(params: {
|
||||
webhookUrl: string;
|
||||
webhookPath?: string;
|
||||
}): string {
|
||||
return `${resolveHostedZaloMediaRoutePrefix(params)}/`;
|
||||
}
|
||||
|
||||
export async function prepareHostedZaloMediaUrl(params: {
|
||||
mediaUrl: string;
|
||||
webhookUrl: string;
|
||||
webhookPath?: string;
|
||||
maxBytes: number;
|
||||
proxyUrl?: string;
|
||||
}): Promise<string> {
|
||||
await ensureHostedZaloMediaDir();
|
||||
await cleanupExpiredHostedZaloMedia();
|
||||
|
||||
const media = await loadOutboundMediaFromUrl(params.mediaUrl, {
|
||||
maxBytes: params.maxBytes,
|
||||
...(params.proxyUrl ? { proxyUrl: params.proxyUrl } : {}),
|
||||
});
|
||||
|
||||
const routePath = resolveHostedZaloMediaRoutePath({
|
||||
webhookUrl: params.webhookUrl,
|
||||
webhookPath: params.webhookPath,
|
||||
});
|
||||
const id = createHostedZaloMediaId();
|
||||
const token = createHostedZaloMediaToken();
|
||||
const publicBaseUrl = new URL(params.webhookUrl).origin;
|
||||
|
||||
await writeFile(resolveHostedZaloMediaBufferPath(id), media.buffer, { mode: 0o600 });
|
||||
try {
|
||||
await writeFile(
|
||||
resolveHostedZaloMediaMetadataPath(id),
|
||||
JSON.stringify({
|
||||
routePath,
|
||||
token,
|
||||
contentType: media.contentType,
|
||||
expiresAt: Date.now() + ZALO_OUTBOUND_MEDIA_TTL_MS,
|
||||
} satisfies HostedZaloMediaMetadata),
|
||||
{ encoding: "utf8", mode: 0o600 },
|
||||
);
|
||||
} catch (error) {
|
||||
await deleteHostedZaloMediaEntry(id);
|
||||
throw error;
|
||||
}
|
||||
|
||||
return `${publicBaseUrl}${routePath}${id}?token=${token}`;
|
||||
}
|
||||
|
||||
export async function tryHandleHostedZaloMediaRequest(
|
||||
req: IncomingMessage,
|
||||
res: ServerResponse,
|
||||
): Promise<boolean> {
|
||||
await cleanupExpiredHostedZaloMedia();
|
||||
|
||||
const method = req.method ?? "GET";
|
||||
if (method !== "GET" && method !== "HEAD") {
|
||||
return false;
|
||||
}
|
||||
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(req.url ?? "/", "http://localhost");
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
|
||||
const mediaPath = url.pathname;
|
||||
const prefixIndex = mediaPath.lastIndexOf(ZALO_OUTBOUND_MEDIA_PREFIX);
|
||||
if (prefixIndex < 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const routePath = mediaPath.slice(0, prefixIndex + ZALO_OUTBOUND_MEDIA_PREFIX.length);
|
||||
const id = mediaPath.slice(prefixIndex + ZALO_OUTBOUND_MEDIA_PREFIX.length);
|
||||
if (!id || !ZALO_OUTBOUND_MEDIA_ID_RE.test(id)) {
|
||||
res.statusCode = 404;
|
||||
res.end("Not Found");
|
||||
return true;
|
||||
}
|
||||
|
||||
const entry = await readHostedZaloMediaEntry(id);
|
||||
if (!entry || entry.metadata.routePath !== routePath) {
|
||||
res.statusCode = 404;
|
||||
res.end("Not Found");
|
||||
return true;
|
||||
}
|
||||
|
||||
if (entry.metadata.expiresAt <= Date.now()) {
|
||||
await deleteHostedZaloMediaEntry(id);
|
||||
res.statusCode = 410;
|
||||
res.end("Expired");
|
||||
return true;
|
||||
}
|
||||
|
||||
if (url.searchParams.get("token") !== entry.metadata.token) {
|
||||
res.statusCode = 401;
|
||||
res.end("Unauthorized");
|
||||
return true;
|
||||
}
|
||||
|
||||
if (entry.metadata.contentType) {
|
||||
res.setHeader("Content-Type", entry.metadata.contentType);
|
||||
}
|
||||
res.setHeader("Cache-Control", "no-store");
|
||||
res.setHeader("X-Content-Type-Options", "nosniff");
|
||||
const bufferStats = await stat(resolveHostedZaloMediaBufferPath(id)).catch(() => null);
|
||||
if (bufferStats) {
|
||||
res.setHeader("Content-Length", String(bufferStats.size));
|
||||
}
|
||||
|
||||
if (method === "HEAD") {
|
||||
res.statusCode = 200;
|
||||
res.end();
|
||||
return true;
|
||||
}
|
||||
|
||||
res.statusCode = 200;
|
||||
res.end(entry.buffer);
|
||||
await deleteHostedZaloMediaEntry(id);
|
||||
return true;
|
||||
}
|
||||
|
||||
export function clearHostedZaloMediaForTest(): void {
|
||||
rmSync(ZALO_OUTBOUND_MEDIA_DIR, { recursive: true, force: true });
|
||||
}
|
||||
@@ -76,6 +76,7 @@ export {
|
||||
createFixedWindowRateLimiter,
|
||||
createWebhookAnomalyTracker,
|
||||
readJsonWebhookBodyOrReject,
|
||||
registerPluginHttpRoute,
|
||||
registerWebhookTarget,
|
||||
registerWebhookTargetWithPluginRoute,
|
||||
resolveWebhookPath,
|
||||
|
||||
@@ -88,4 +88,33 @@ describe("zalo send", () => {
|
||||
expect(sendMessageMock).not.toHaveBeenCalled();
|
||||
expect(sendPhotoMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("sends cfg-backed media directly without hosted-media rewrites", async () => {
|
||||
sendPhotoMock.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
result: { message_id: "z-photo-2" },
|
||||
});
|
||||
|
||||
const result = await sendPhotoZalo("dm-chat-5", "https://example.com/photo.jpg", {
|
||||
cfg: {
|
||||
channels: {
|
||||
zalo: {
|
||||
botToken: "zalo-token",
|
||||
webhookUrl: "https://gateway.example.com/zalo-webhook",
|
||||
},
|
||||
},
|
||||
} as never,
|
||||
});
|
||||
|
||||
expect(sendPhotoMock).toHaveBeenCalledWith(
|
||||
"zalo-token",
|
||||
{
|
||||
chat_id: "dm-chat-5",
|
||||
photo: "https://example.com/photo.jpg",
|
||||
caption: undefined,
|
||||
},
|
||||
undefined,
|
||||
);
|
||||
expect(result).toEqual({ ok: true, messageId: "z-photo-2" });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -139,14 +139,15 @@ export async function sendPhotoZalo(
|
||||
}
|
||||
|
||||
return await runZaloSend("Failed to send photo", () =>
|
||||
sendPhoto(
|
||||
context.token,
|
||||
{
|
||||
chat_id: context.chatId,
|
||||
photo: photoUrl.trim(),
|
||||
caption: options.caption?.slice(0, 2000),
|
||||
},
|
||||
context.fetcher,
|
||||
),
|
||||
(async () =>
|
||||
sendPhoto(
|
||||
context.token,
|
||||
{
|
||||
chat_id: context.chatId,
|
||||
photo: photoUrl.trim(),
|
||||
caption: options.caption?.slice(0, 2000),
|
||||
},
|
||||
context.fetcher,
|
||||
))(),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -20,6 +20,8 @@ const runtimeModuleId = new URL("../src/runtime.js", import.meta.url).pathname;
|
||||
|
||||
type UnknownMock = Mock<(...args: unknown[]) => unknown>;
|
||||
type AsyncUnknownMock = Mock<(...args: unknown[]) => Promise<unknown>>;
|
||||
const loadedMonitorModules = new Set<MonitorModule>();
|
||||
|
||||
type ZaloLifecycleMocks = {
|
||||
setWebhookMock: AsyncUnknownMock;
|
||||
deleteWebhookMock: AsyncUnknownMock;
|
||||
@@ -87,7 +89,11 @@ async function importMonitorModule(params: {
|
||||
vi.doUnmock(apiModuleId);
|
||||
vi.doUnmock(runtimeModuleId);
|
||||
}
|
||||
return (await import(`${monitorModuleUrl}?t=${params.cacheBust}-${Date.now()}`)) as MonitorModule;
|
||||
const module = (await import(
|
||||
`${monitorModuleUrl}?t=${params.cacheBust}-${Date.now()}`
|
||||
)) as MonitorModule;
|
||||
loadedMonitorModules.add(module);
|
||||
return module;
|
||||
}
|
||||
|
||||
async function importSecretInputModule(cacheBust: string): Promise<SecretInputModule> {
|
||||
@@ -103,6 +109,13 @@ async function importWebhookModule(cacheBust: string): Promise<WebhookModule> {
|
||||
export async function resetLifecycleTestState() {
|
||||
vi.clearAllMocks();
|
||||
(await importWebhookModule("reset-webhook")).clearZaloWebhookSecurityStateForTest();
|
||||
for (const module of loadedMonitorModules) {
|
||||
module.__testing.clearHostedMediaRouteRefsForTest();
|
||||
}
|
||||
(
|
||||
await importMonitorModule({ cacheBust: "reset-monitor", mocked: false })
|
||||
).__testing.clearHostedMediaRouteRefsForTest();
|
||||
loadedMonitorModules.clear();
|
||||
setActivePluginRegistry(createEmptyPluginRegistry());
|
||||
}
|
||||
|
||||
@@ -152,12 +165,16 @@ export async function startWebhookLifecycleMonitor(params: {
|
||||
});
|
||||
|
||||
await vi.waitFor(() => {
|
||||
if (setWebhookMock.mock.calls.length !== 1 || registry.httpRoutes.length !== 1) {
|
||||
const webhookRoute = registry.httpRoutes.find((route) => route.source === "zalo-webhook");
|
||||
const hostedMediaRoute = registry.httpRoutes.find(
|
||||
(route) => route.source === "zalo-hosted-media",
|
||||
);
|
||||
if (setWebhookMock.mock.calls.length !== 1 || !webhookRoute || !hostedMediaRoute) {
|
||||
throw new Error("waiting for webhook registration");
|
||||
}
|
||||
});
|
||||
|
||||
const route = registry.httpRoutes[0];
|
||||
const route = registry.httpRoutes.find((entry) => entry.source === "zalo-webhook");
|
||||
if (!route) {
|
||||
throw new Error("missing plugin HTTP route");
|
||||
}
|
||||
|
||||
@@ -46,6 +46,7 @@ type FetchMediaOptions = {
|
||||
readIdleTimeoutMs?: number;
|
||||
ssrfPolicy?: SsrFPolicy;
|
||||
lookupFn?: LookupFn;
|
||||
dispatcherPolicy?: PinnedDispatcherPolicy;
|
||||
dispatcherAttempts?: FetchDispatcherAttempt[];
|
||||
shouldRetryFetchError?: (error: unknown) => boolean;
|
||||
/**
|
||||
@@ -113,6 +114,7 @@ export async function fetchRemoteMedia(options: FetchMediaOptions): Promise<Fetc
|
||||
readIdleTimeoutMs,
|
||||
ssrfPolicy,
|
||||
lookupFn,
|
||||
dispatcherPolicy,
|
||||
dispatcherAttempts,
|
||||
shouldRetryFetchError,
|
||||
trustExplicitProxyDns,
|
||||
@@ -125,7 +127,7 @@ export async function fetchRemoteMedia(options: FetchMediaOptions): Promise<Fetc
|
||||
const attempts =
|
||||
dispatcherAttempts && dispatcherAttempts.length > 0
|
||||
? dispatcherAttempts
|
||||
: [{ dispatcherPolicy: undefined, lookupFn }];
|
||||
: [{ dispatcherPolicy, lookupFn }];
|
||||
const runGuardedFetch = async (attempt: FetchDispatcherAttempt) =>
|
||||
await fetchWithSsrFGuard(
|
||||
(trustExplicitProxyDns && attempt.dispatcherPolicy?.mode === "explicit-proxy"
|
||||
|
||||
@@ -12,6 +12,10 @@ export type OutboundMediaLoadParams = {
|
||||
mediaAccess?: OutboundMediaAccess;
|
||||
mediaLocalRoots?: readonly string[] | "any";
|
||||
mediaReadFile?: OutboundMediaReadFile;
|
||||
proxyUrl?: string;
|
||||
fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
|
||||
requestInit?: RequestInit;
|
||||
trustExplicitProxyDns?: boolean;
|
||||
optimizeImages?: boolean;
|
||||
/** Agent workspace directory for resolving relative MEDIA: paths. */
|
||||
workspaceDir?: string;
|
||||
@@ -21,6 +25,10 @@ export type OutboundMediaLoadOptions = {
|
||||
maxBytes?: number;
|
||||
localRoots?: readonly string[] | "any";
|
||||
readFile?: (filePath: string) => Promise<Buffer>;
|
||||
proxyUrl?: string;
|
||||
fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
|
||||
requestInit?: RequestInit;
|
||||
trustExplicitProxyDns?: boolean;
|
||||
hostReadCapability?: boolean;
|
||||
optimizeImages?: boolean;
|
||||
/** Agent workspace directory for resolving relative MEDIA: paths. */
|
||||
@@ -81,6 +89,12 @@ export function buildOutboundMediaLoadOptions(
|
||||
...(params.maxBytes !== undefined ? { maxBytes: params.maxBytes } : {}),
|
||||
localRoots,
|
||||
readFile,
|
||||
...(params.fetchImpl ? { fetchImpl: params.fetchImpl } : {}),
|
||||
...(params.proxyUrl ? { proxyUrl: params.proxyUrl } : {}),
|
||||
...(params.requestInit ? { requestInit: params.requestInit } : {}),
|
||||
...(params.trustExplicitProxyDns !== undefined
|
||||
? { trustExplicitProxyDns: params.trustExplicitProxyDns }
|
||||
: {}),
|
||||
hostReadCapability: true,
|
||||
...(params.optimizeImages !== undefined ? { optimizeImages: params.optimizeImages } : {}),
|
||||
...(workspaceDir ? { workspaceDir } : {}),
|
||||
@@ -89,6 +103,12 @@ export function buildOutboundMediaLoadOptions(
|
||||
return {
|
||||
...(params.maxBytes !== undefined ? { maxBytes: params.maxBytes } : {}),
|
||||
...(localRoots ? { localRoots } : {}),
|
||||
...(params.proxyUrl ? { proxyUrl: params.proxyUrl } : {}),
|
||||
...(params.fetchImpl ? { fetchImpl: params.fetchImpl } : {}),
|
||||
...(params.requestInit ? { requestInit: params.requestInit } : {}),
|
||||
...(params.trustExplicitProxyDns !== undefined
|
||||
? { trustExplicitProxyDns: params.trustExplicitProxyDns }
|
||||
: {}),
|
||||
...(params.optimizeImages !== undefined ? { optimizeImages: params.optimizeImages } : {}),
|
||||
...(workspaceDir ? { workspaceDir } : {}),
|
||||
};
|
||||
|
||||
@@ -3,7 +3,7 @@ import { resolveCanvasHttpPathToLocalPath } from "../gateway/canvas-documents.js
|
||||
import { logVerbose, shouldLogVerbose } from "../globals.js";
|
||||
import { SafeOpenError, readLocalFileSafely } from "../infra/fs-safe.js";
|
||||
import { assertNoWindowsNetworkPath, safeFileURLToPath } from "../infra/local-file-access.js";
|
||||
import type { SsrFPolicy } from "../infra/net/ssrf.js";
|
||||
import type { PinnedDispatcherPolicy, SsrFPolicy } from "../infra/net/ssrf.js";
|
||||
import { resolveUserPath } from "../utils.js";
|
||||
import { maxBytesForKind, type MediaKind } from "./constants.js";
|
||||
import { fetchRemoteMedia } from "./fetch.js";
|
||||
@@ -42,6 +42,10 @@ type WebMediaOptions = {
|
||||
maxBytes?: number;
|
||||
optimizeImages?: boolean;
|
||||
ssrfPolicy?: SsrFPolicy;
|
||||
proxyUrl?: string;
|
||||
fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
|
||||
requestInit?: RequestInit;
|
||||
trustExplicitProxyDns?: boolean;
|
||||
workspaceDir?: string;
|
||||
/** Allowed root directories for local path reads. "any" is deprecated; prefer sandboxValidated + readFile. */
|
||||
localRoots?: readonly string[] | "any";
|
||||
@@ -340,6 +344,10 @@ async function loadWebMediaInternal(
|
||||
maxBytes,
|
||||
optimizeImages = true,
|
||||
ssrfPolicy,
|
||||
proxyUrl,
|
||||
fetchImpl,
|
||||
requestInit,
|
||||
trustExplicitProxyDns,
|
||||
workspaceDir,
|
||||
localRoots,
|
||||
sandboxValidated = false,
|
||||
@@ -436,7 +444,22 @@ async function loadWebMediaInternal(
|
||||
: optimizeImages
|
||||
? Math.max(maxBytes, defaultFetchCap)
|
||||
: maxBytes;
|
||||
const fetched = await fetchRemoteMedia({ url: mediaUrl, maxBytes: fetchCap, ssrfPolicy });
|
||||
const dispatcherPolicy: PinnedDispatcherPolicy | undefined = proxyUrl
|
||||
? {
|
||||
mode: "explicit-proxy",
|
||||
proxyUrl,
|
||||
allowPrivateProxy: true,
|
||||
}
|
||||
: undefined;
|
||||
const fetched = await fetchRemoteMedia({
|
||||
url: mediaUrl,
|
||||
fetchImpl,
|
||||
requestInit,
|
||||
maxBytes: fetchCap,
|
||||
ssrfPolicy,
|
||||
dispatcherPolicy,
|
||||
trustExplicitProxyDns,
|
||||
});
|
||||
const { buffer, contentType, fileName } = fetched;
|
||||
const kind = kindFromMime(contentType);
|
||||
return await clampAndFinalize({ buffer, contentType, kind, fileName });
|
||||
|
||||
@@ -6,6 +6,10 @@ export type OutboundMediaLoadOptions = {
|
||||
mediaAccess?: OutboundMediaAccess;
|
||||
mediaLocalRoots?: readonly string[] | "any";
|
||||
mediaReadFile?: (filePath: string) => Promise<Buffer>;
|
||||
proxyUrl?: string;
|
||||
fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>;
|
||||
requestInit?: RequestInit;
|
||||
trustExplicitProxyDns?: boolean;
|
||||
};
|
||||
|
||||
/** Load outbound media from a remote URL or approved local path using the shared web-media policy. */
|
||||
@@ -20,6 +24,10 @@ export async function loadOutboundMediaFromUrl(
|
||||
mediaAccess: options.mediaAccess,
|
||||
mediaLocalRoots: options.mediaLocalRoots,
|
||||
mediaReadFile: options.mediaReadFile,
|
||||
proxyUrl: options.proxyUrl,
|
||||
fetchImpl: options.fetchImpl,
|
||||
requestInit: options.requestInit,
|
||||
trustExplicitProxyDns: options.trustExplicitProxyDns,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user