// Line tests cover monitor.lifecycle plugin behavior. import crypto from "node:crypto"; import { EventEmitter } from "node:events"; import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import { createMockIncomingRequest } from "openclaw/plugin-sdk/test-env"; import { WEBHOOK_IN_FLIGHT_DEFAULTS } from "openclaw/plugin-sdk/webhook-request-guards"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; type LineNodeWebhookHandler = (req: IncomingMessage, res: ServerResponse) => Promise; type LineHandleWebhook = (...args: unknown[]) => Promise; const { createLineBotMock, createLineNodeWebhookHandlerMock, registerWebhookTargetWithPluginRouteMock, runDetachedWebhookWorkMock, unregisterHttpMock, } = vi.hoisted(() => ({ createLineBotMock: vi.fn(() => ({ account: { accountId: "default" }, handleWebhook: vi.fn(), stop: vi.fn(), })), createLineNodeWebhookHandlerMock: vi.fn<() => LineNodeWebhookHandler>(() => vi.fn(async () => {}), ), registerWebhookTargetWithPluginRouteMock: vi.fn(), runDetachedWebhookWorkMock: vi.fn(), unregisterHttpMock: vi.fn(), })); let monitorLineProvider: typeof import("./monitor.js").monitorLineProvider; let innerLineWebhookHandlerMock: ReturnType>; type RegisteredRoute = { accountId?: string; auth?: string; handler?: LineNodeWebhookHandler; path?: string; pluginId?: string; replaceExisting?: boolean; source?: string; throwOnFailure?: boolean; }; type RegisteredTarget = { accountId?: string; path: string; }; type WebhookRegistration = { route: RegisteredRoute; target: RegisteredTarget; }; function requireWebhookRegistration(): WebhookRegistration { const registration = registerWebhookTargetWithPluginRouteMock.mock.calls[0]?.[0] as | WebhookRegistration | undefined; if (!registration) { throw new Error("expected registered LINE webhook target"); } return registration; } function requireRegisteredRoute(): { handler: LineNodeWebhookHandler } { const route = requireWebhookRegistration().route; if (!route.handler) { throw new Error("expected registered LINE webhook route"); } return { handler: route.handler }; } vi.mock("./bot.js", () => ({ createLineBot: createLineBotMock, })); vi.mock("openclaw/plugin-sdk/reply-runtime", () => ({ chunkMarkdownText: vi.fn(), dispatchReplyWithBufferedBlockDispatcher: vi.fn(), })); vi.mock("openclaw/plugin-sdk/runtime-env", async () => { const actual = await vi.importActual( "openclaw/plugin-sdk/runtime-env", ); return { ...actual, danger: (value: unknown) => String(value), logVerbose: vi.fn(), }; }); vi.mock("openclaw/plugin-sdk/webhook-ingress", async () => { const actual = await vi.importActual( "openclaw/plugin-sdk/webhook-ingress", ); return { ...actual, normalizePluginHttpPath: (path: string | undefined, fallback: string) => path ?? fallback, registerWebhookTargetWithPluginRoute: registerWebhookTargetWithPluginRouteMock, }; }); vi.mock("openclaw/plugin-sdk/webhook-request-guards", async () => { const actual = await vi.importActual( "openclaw/plugin-sdk/webhook-request-guards", ); runDetachedWebhookWorkMock.mockImplementation(actual.runDetachedWebhookWork); return { ...actual, runDetachedWebhookWork: runDetachedWebhookWorkMock, }; }); vi.mock("./webhook-node.js", async () => { const actual = await vi.importActual("./webhook-node.js"); return { ...actual, createLineNodeWebhookHandler: createLineNodeWebhookHandlerMock, }; }); vi.mock("./auto-reply-delivery.js", () => ({ deliverLineAutoReply: vi.fn(), })); vi.mock("./markdown-to-line.js", () => ({ processLineMessage: vi.fn(), })); vi.mock("./send.js", () => ({ createFlexMessage: vi.fn(), createImageMessage: vi.fn(), createLocationMessage: vi.fn(), createQuickReplyItems: vi.fn(), getUserDisplayName: vi.fn(), pushMessagesLine: vi.fn(), replyMessageLine: vi.fn(), showLoadingAnimation: vi.fn(), })); vi.mock("./template-messages.js", () => ({ buildTemplateMessageFromPayload: vi.fn(), })); describe("monitorLineProvider lifecycle", () => { beforeAll(async () => { ({ monitorLineProvider } = await import("./monitor.js")); }); afterAll(() => { vi.doUnmock("./bot.js"); vi.doUnmock("openclaw/plugin-sdk/reply-runtime"); vi.doUnmock("openclaw/plugin-sdk/runtime-env"); vi.doUnmock("openclaw/plugin-sdk/webhook-ingress"); vi.doUnmock("openclaw/plugin-sdk/webhook-request-guards"); vi.doUnmock("./webhook-node.js"); vi.doUnmock("./auto-reply-delivery.js"); vi.doUnmock("./markdown-to-line.js"); vi.doUnmock("./send.js"); vi.doUnmock("./template-messages.js"); vi.resetModules(); }); beforeEach(() => { createLineBotMock.mockReset(); createLineBotMock.mockImplementation(() => ({ account: { accountId: "default" }, handleWebhook: vi.fn(), stop: vi.fn(async () => undefined), })); // Clear call history only; the implementation was wired to the actual // helper once in the module mock factory. runDetachedWebhookWorkMock.mockClear(); innerLineWebhookHandlerMock = vi.fn(async () => {}); createLineNodeWebhookHandlerMock .mockReset() .mockImplementation(() => innerLineWebhookHandlerMock); unregisterHttpMock.mockReset(); registerWebhookTargetWithPluginRouteMock.mockReset().mockImplementation((params) => { const withLeadingSlash = params.target.path.startsWith("/") ? params.target.path : `/${params.target.path}`; const key = withLeadingSlash .replace(/\/{2,}/g, "/") .replace(/\/+$/, "") .toLowerCase() || "/"; const normalizedTarget = { ...params.target, path: key }; const existing = params.targetsByPath.get(key) ?? []; params.targetsByPath.set(key, [...existing, normalizedTarget]); return { target: normalizedTarget, unregister: () => { unregisterHttpMock(); const updated = (params.targetsByPath.get(key) ?? []).filter( (entry: unknown) => entry !== normalizedTarget, ); if (updated.length > 0) { params.targetsByPath.set(key, updated); } else { params.targetsByPath.delete(key); } }, }; }); }); const createRouteResponse = () => { const resObj = { statusCode: 0, headersSent: false, setHeader: vi.fn(), end: vi.fn(() => { resObj.headersSent = true; }), }; return resObj as unknown as ServerResponse & { end: ReturnType }; }; it("waits for abort before resolving", async () => { const abort = new AbortController(); const statusSink = vi.fn(); let resolved = false; const task = monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, abortSignal: abort.signal, statusSink, }).then((monitor) => { resolved = true; return monitor; }); expect(registerWebhookTargetWithPluginRouteMock).toHaveBeenCalledTimes(1); expect(requireWebhookRegistration().route.auth).toBe("plugin"); await vi.waitFor(() => expect(statusSink).toHaveBeenCalledWith( expect.objectContaining({ lifecycle: "ready", connected: true }), ), ); expect(resolved).toBe(false); abort.abort(); await task; expect(unregisterHttpMock).toHaveBeenCalledTimes(1); expect(statusSink).toHaveBeenLastCalledWith( expect.objectContaining({ lifecycle: "stopped", running: false }), ); }); it("registers an account target without replacing existing route ownership", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret accountId: "work", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const registration = requireWebhookRegistration(); expect(registration.target.accountId).toBe("work"); expect(registration.target.path).toBe("/line/webhook"); expect(registration.route.accountId).toBe("work"); expect(registration.route.auth).toBe("plugin"); expect(registration.route.pluginId).toBe("line"); expect(registration.route.source).toBe("line-webhook"); expect(registration.route.throwOnFailure).toBe(true); expect(registration.route).not.toHaveProperty("path"); expect(registration.route).not.toHaveProperty("replaceExisting"); await monitor.stop(); }); it("stops immediately when signal is already aborted", async () => { const abort = new AbortController(); abort.abort(); await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, abortSignal: abort.signal, }); expect(unregisterHttpMock).toHaveBeenCalledTimes(1); }); it("returns immediately without abort signal and stop is idempotent", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); expect(unregisterHttpMock).not.toHaveBeenCalled(); await monitor.stop(); await monitor.stop(); expect(unregisterHttpMock).toHaveBeenCalledTimes(1); }); it("registers the configured defaultAccount when accountId is omitted", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: { channels: { line: { defaultAccount: "work", accounts: { work: { channelAccessToken: "work-token", channelSecret: "work-secret", }, }, }, }, } as OpenClawConfig, runtime: {} as RuntimeEnv, }); const registration = requireWebhookRegistration(); expect(registration.target.accountId).toBe("work"); expect(registration.route.accountId).toBe("work"); await monitor.stop(); }); it("does not register a webhook when bot startup fails", async () => { createLineBotMock.mockImplementation(() => { throw new Error("line bot startup failed"); }); await expect( monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }), ).rejects.toThrow("line bot startup failed"); expect(registerWebhookTargetWithPluginRouteMock).not.toHaveBeenCalled(); }); it("stops the bot and rejects startup when the webhook route cannot bind", async () => { const statusSink = vi.fn(); registerWebhookTargetWithPluginRouteMock.mockImplementationOnce(() => { throw new Error("LINE route conflict"); }); await expect( monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, statusSink, }), ).rejects.toThrow("LINE route conflict"); const bot = createLineBotMock.mock.results[0]?.value as | { stop: ReturnType } | undefined; expect(bot?.stop).toHaveBeenCalledOnce(); expect(statusSink).not.toHaveBeenCalledWith(expect.objectContaining({ lifecycle: "ready" })); }); it("dispatches shared-path webhook posts to the account matching the signature", async () => { const firstMonitor = await monitorLineProvider({ channelAccessToken: "first-token", channelSecret: "first-secret", // pragma: allowlist secret accountId: "first", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const secondMonitor = await monitorLineProvider({ channelAccessToken: "second-token", channelSecret: "second-secret", // pragma: allowlist secret accountId: "second", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const route = requireRegisteredRoute(); const payload = JSON.stringify({ events: [{ type: "message" }] }); const signature = crypto.createHmac("SHA256", "second-secret").update(payload).digest("base64"); const req = Object.assign(createMockIncomingRequest([payload]), { method: "POST", headers: { "x-line-signature": signature }, }) as unknown as IncomingMessage; const res = createRouteResponse(); await route.handler(req, res); const firstBot = createLineBotMock.mock.results[0]?.value as { handleWebhook: ReturnType; }; const secondBot = createLineBotMock.mock.results[1]?.value as { handleWebhook: ReturnType; }; expect(res.statusCode).toBe(200); expect(firstBot.handleWebhook).not.toHaveBeenCalled(); expect(secondBot.handleWebhook).toHaveBeenCalledTimes(1); await firstMonitor.stop(); await secondMonitor.stop(); }); it("marks only durably admitted signed HTTP webhooks and redacts failures", async () => { const runtimeError = vi.fn<(...args: unknown[]) => void>(); const runtime: RuntimeEnv = { log: vi.fn<(...args: unknown[]) => void>(), error: runtimeError, exit: vi.fn<(code: number) => void>(), }; const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret accountId: "default", config: {} as OpenClawConfig, runtime, }); const route = requireRegisteredRoute(); const server = createServer((req, res) => { void route.handler(req, res); }); try { await new Promise((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", () => { server.removeListener("error", reject); resolve(); }); }); const address = server.address(); if (!address || typeof address === "string") { throw new Error("expected LINE webhook test server to have a TCP address"); } const webhookUrl = `http://127.0.0.1:${address.port}/line/webhook`; const payload = JSON.stringify({ events: [{ type: "message" }] }); const rejected = await fetch(webhookUrl, { method: "POST", headers: { "content-type": "application/json", "x-line-signature": "invalid-signature", }, body: payload, }); expect(rejected.status).toBe(401); expect(rejected.headers.get("x-openclaw-delivery-accepted")).toBeNull(); expect(await rejected.json()).toEqual({ error: "Invalid signature" }); const bot = createLineBotMock.mock.results[0]?.value; if (!bot) { throw new Error("expected registered LINE webhook bot"); } expect(bot.handleWebhook).not.toHaveBeenCalled(); const verificationPayload = JSON.stringify({ events: [] }); const verificationSignature = crypto .createHmac("SHA256", "secret") .update(verificationPayload) .digest("base64"); const verification = await fetch(webhookUrl, { method: "POST", headers: { "content-type": "application/json", "x-line-signature": verificationSignature, }, body: verificationPayload, }); expect(verification.status).toBe(200); expect(verification.headers.get("x-openclaw-delivery-accepted")).toBeNull(); expect(await verification.json()).toEqual({ status: "ok" }); expect(bot.handleWebhook).not.toHaveBeenCalled(); let releaseAdmission: (() => void) | undefined; bot.handleWebhook.mockImplementationOnce( () => new Promise((resolve) => { releaseAdmission = resolve; }), ); const signature = crypto.createHmac("SHA256", "secret").update(payload).digest("base64"); let acceptedResponseReceived = false; const acceptedRequest = fetch(webhookUrl, { method: "POST", headers: { "content-type": "application/json", "x-line-signature": signature, }, body: payload, }).then((response) => { acceptedResponseReceived = true; return response; }); await vi.waitFor(() => { expect(bot.handleWebhook).toHaveBeenCalledTimes(1); }); expect(acceptedResponseReceived).toBe(false); if (!releaseAdmission) { throw new Error("expected pending LINE durable admission"); } releaseAdmission(); const accepted = await acceptedRequest; expect(accepted.status).toBe(200); expect(accepted.headers.get("x-openclaw-delivery-accepted")).toBe("durable"); expect(await accepted.json()).toEqual({ status: "ok" }); const bearerToken = "test_line_access_token_1234567890"; bot.handleWebhook.mockRejectedValueOnce({ code: "LINE_ADMISSION_REJECTED", retryAfterMs: 250, authorization: `Bearer ${bearerToken}`, }); const response = await fetch(webhookUrl, { method: "POST", headers: { "content-type": "application/json", "x-line-signature": signature, }, body: payload, }); expect(response.status).toBe(500); expect(response.headers.get("x-openclaw-delivery-accepted")).toBeNull(); expect(await response.json()).toEqual({ error: "Internal server error" }); expect(bot.handleWebhook).toHaveBeenCalledTimes(2); expect(runtimeError).toHaveBeenCalledTimes(1); const message = String(runtimeError.mock.calls[0]?.[0]); expect(message).toContain("line webhook error:"); expect(message).toContain("LINE_ADMISSION_REJECTED"); expect(message).toContain("retryAfterMs"); expect(message).not.toContain("[object Object]"); expect(message).not.toContain(bearerToken); } finally { try { if (server.listening) { await new Promise((resolve, reject) => { server.close((error) => { if (error) { reject(error); return; } resolve(); }); }); } } finally { await monitor.stop(); } } }); it("dispatches a signed POST through the canonical configured webhook route", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret webhookPath: "/Line//Webhook/", accountId: "default", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const registration = requireWebhookRegistration(); expect(registration.target.path).toBe("/Line//Webhook"); const route = requireRegisteredRoute(); const payload = JSON.stringify({ events: [{ type: "message" }] }); const signature = crypto.createHmac("SHA256", "secret").update(payload).digest("base64"); const req = Object.assign(createMockIncomingRequest([payload]), { method: "POST", headers: { "x-line-signature": signature }, }) as unknown as IncomingMessage; const res = createRouteResponse(); await route.handler(req, res); const bot = createLineBotMock.mock.results[0]?.value as { handleWebhook: ReturnType; }; expect(res.statusCode).toBe(200); expect(bot.handleWebhook).toHaveBeenCalledTimes(1); await monitor.stop(); }); it("durably admits matched events before acknowledging", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret accountId: "default", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const route = requireRegisteredRoute(); const payload = JSON.stringify({ events: [{ type: "message" }] }); const signature = crypto.createHmac("SHA256", "secret").update(payload).digest("base64"); const req = Object.assign(createMockIncomingRequest([payload]), { method: "POST", headers: { "x-line-signature": signature }, }) as unknown as IncomingMessage; const res = createRouteResponse(); await route.handler(req, res); const bot = createLineBotMock.mock.results[0]?.value as { handleWebhook: ReturnType; }; expect(res.statusCode).toBe(200); expect(runDetachedWebhookWorkMock).not.toHaveBeenCalled(); expect(bot.handleWebhook).toHaveBeenCalledTimes(1); await monitor.stop(); }); it("waits for shared-path durable admission before acknowledging", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret accountId: "default", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); let releaseWebhook: (() => void) | undefined; const bot = createLineBotMock.mock.results[0]?.value as { handleWebhook: ReturnType>; }; bot.handleWebhook.mockImplementation( () => new Promise((resolve) => { releaseWebhook = resolve; }), ); const route = requireRegisteredRoute(); const payload = JSON.stringify({ events: [{ type: "message" }] }); const signature = crypto.createHmac("SHA256", "secret").update(payload).digest("base64"); const req = Object.assign(createMockIncomingRequest([payload]), { method: "POST", headers: { "x-line-signature": signature }, }) as unknown as IncomingMessage; const res = createRouteResponse(); const request = route.handler(req, res); await vi.waitFor(() => { expect(bot.handleWebhook).toHaveBeenCalledTimes(1); }); expect(res.headersSent).toBe(false); expect(bot.handleWebhook).toHaveBeenCalledTimes(1); if (!releaseWebhook) { throw new Error("expected pending LINE webhook handler"); } releaseWebhook(); await request; expect(res.statusCode).toBe(200); expect(res.headersSent).toBe(true); await monitor.stop(); }); it("rejects ambiguous shared-path webhook signatures", async () => { const firstMonitor = await monitorLineProvider({ channelAccessToken: "first-token", channelSecret: "shared-secret", // pragma: allowlist secret accountId: "first", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const secondMonitor = await monitorLineProvider({ channelAccessToken: "second-token", channelSecret: "shared-secret", // pragma: allowlist secret accountId: "second", config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const route = requireRegisteredRoute(); const payload = JSON.stringify({ events: [{ type: "message" }] }); const signature = crypto.createHmac("SHA256", "shared-secret").update(payload).digest("base64"); const req = Object.assign(createMockIncomingRequest([payload]), { method: "POST", headers: { "x-line-signature": signature }, }) as unknown as IncomingMessage; const res = createRouteResponse(); await route.handler(req, res); const firstBot = createLineBotMock.mock.results[0]?.value as { handleWebhook: ReturnType; }; const secondBot = createLineBotMock.mock.results[1]?.value as { handleWebhook: ReturnType; }; expect(res.statusCode).toBe(401); expect(res.end).toHaveBeenCalledWith(JSON.stringify({ error: "Ambiguous webhook target" })); expect(firstBot.handleWebhook).not.toHaveBeenCalled(); expect(secondBot.handleWebhook).not.toHaveBeenCalled(); await firstMonitor.stop(); await secondMonitor.stop(); }); it("rejects webhook requests above the shared in-flight limit before body handling", async () => { const limit = WEBHOOK_IN_FLIGHT_DEFAULTS.maxInFlightPerKey; const heldRequests: Array void }> = []; const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const route = requireRegisteredRoute(); const createHeldPostRequest = () => { const req = Object.assign(new EventEmitter(), { destroyed: false, destroy(this: EventEmitter & { destroyed: boolean }) { this.destroyed = true; this.emit("close"); }, }); heldRequests.push(req); return Object.assign(req, { method: "POST", headers: { "x-line-signature": "pending" }, }) as unknown as IncomingMessage; }; const createSignedPostRequest = () => { const payload = JSON.stringify({ events: [{ type: "message" }] }); const signature = crypto.createHmac("SHA256", "secret").update(payload).digest("base64"); const req = createMockIncomingRequest([payload]); return Object.assign(req, { method: "POST", headers: { "x-line-signature": signature }, }) as unknown as IncomingMessage; }; const firstRequests = Array.from({ length: limit }, () => route.handler(createHeldPostRequest(), createRouteResponse()), ); await new Promise((resolve) => { setImmediate(resolve); }); const overflowResponse = createRouteResponse(); await route.handler(createSignedPostRequest(), overflowResponse); const bot = createLineBotMock.mock.results[0]?.value as { handleWebhook: ReturnType; }; expect(bot.handleWebhook).not.toHaveBeenCalled(); expect(overflowResponse.statusCode).toBe(429); expect(overflowResponse.end).toHaveBeenCalledWith("Too Many Requests"); heldRequests.splice(0).forEach((req) => req.destroy()); await Promise.allSettled(firstRequests); await monitor.stop(); }); });