fix(core): Fix binary data helpers (like prepareBinaryData) with task runner (#12259)

packages/@n8n/task-runner/package.json

@@ -38,6 +38,7 @@
     "@sentry/node": "catalog:",
     "acorn": "8.14.0",
     "acorn-walk": "8.3.4",
+    "lodash.set": "4.3.2",
     "n8n-core": "workspace:*",
     "n8n-workflow": "workspace:*",
     "nanoid": "catalog:",
@@ -45,6 +46,7 @@
     "ws": "^8.18.0"
   },
   "devDependencies": {
+    "@types/lodash.set": "4.3.9",
     "luxon": "catalog:"
   }
 }

packages/@n8n/task-runner/src/js-task-runner/__tests__/js-task-runner.test.ts

@@ -1,5 +1,6 @@
 import { mock } from 'jest-mock-extended';
 import { DateTime } from 'luxon';
+import type { IBinaryData } from 'n8n-workflow';
 import { setGlobalState, type CodeExecutionMode, type IDataObject } from 'n8n-workflow';
 import fs from 'node:fs';
 import { builtinModules } from 'node:module';
@@ -8,10 +9,15 @@ import type { BaseRunnerConfig } from '@/config/base-runner-config';
 import type { JsRunnerConfig } from '@/config/js-runner-config';
 import { MainConfig } from '@/config/main-config';
 import { ExecutionError } from '@/js-task-runner/errors/execution-error';
+import { UnsupportedFunctionError } from '@/js-task-runner/errors/unsupported-function.error';
 import { ValidationError } from '@/js-task-runner/errors/validation-error';
 import type { JSExecSettings } from '@/js-task-runner/js-task-runner';
 import { JsTaskRunner } from '@/js-task-runner/js-task-runner';
-import type { DataRequestResponse, InputDataChunkDefinition } from '@/runner-types';
+import {
+	UNSUPPORTED_HELPER_FUNCTIONS,
+	type DataRequestResponse,
+	type InputDataChunkDefinition,
+} from '@/runner-types';
 import type { Task } from '@/task-runner';
 
 import {
@@ -567,6 +573,120 @@ describe('JsTaskRunner', () => {
 			);
 		});
 
+		describe('helpers', () => {
+			const binaryDataFile: IBinaryData = {
+				data: 'data',
+				fileName: 'file.txt',
+				mimeType: 'text/plain',
+			};
+
+			const groups = [
+				{
+					method: 'helpers.assertBinaryData',
+					invocation: "helpers.assertBinaryData(0, 'binaryFile')",
+					expectedParams: [0, 'binaryFile'],
+				},
+				{
+					method: 'helpers.getBinaryDataBuffer',
+					invocation: "helpers.getBinaryDataBuffer(0, 'binaryFile')",
+					expectedParams: [0, 'binaryFile'],
+				},
+				{
+					method: 'helpers.prepareBinaryData',
+					invocation: "helpers.prepareBinaryData(Buffer.from('123'), 'file.txt', 'text/plain')",
+					expectedParams: [Buffer.from('123'), 'file.txt', 'text/plain'],
+				},
+				{
+					method: 'helpers.setBinaryDataBuffer',
+					invocation:
+						"helpers.setBinaryDataBuffer({ data: '123', mimeType: 'text/plain' }, Buffer.from('321'))",
+					expectedParams: [{ data: '123', mimeType: 'text/plain' }, Buffer.from('321')],
+				},
+				{
+					method: 'helpers.binaryToString',
+					invocation: "helpers.binaryToString(Buffer.from('123'), 'utf8')",
+					expectedParams: [Buffer.from('123'), 'utf8'],
+				},
+				{
+					method: 'helpers.httpRequest',
+					invocation: "helpers.httpRequest({ method: 'GET', url: 'http://localhost' })",
+					expectedParams: [{ method: 'GET', url: 'http://localhost' }],
+				},
+			];
+
+			for (const group of groups) {
+				it(`${group.method} for runOnceForAllItems`, async () => {
+					// Arrange
+					const rpcCallSpy = jest
+						.spyOn(defaultTaskRunner, 'makeRpcCall')
+						.mockResolvedValue(undefined);
+
+					// Act
+					await execTaskWithParams({
+						task: newTaskWithSettings({
+							code: `await ${group.invocation}; return []`,
+							nodeMode: 'runOnceForAllItems',
+						}),
+						taskData: newDataRequestResponse(
+							[{ json: {}, binary: { binaryFile: binaryDataFile } }],
+							{},
+						),
+					});
+
+					expect(rpcCallSpy).toHaveBeenCalledWith('1', group.method, group.expectedParams);
+				});
+
+				it(`${group.method} for runOnceForEachItem`, async () => {
+					// Arrange
+					const rpcCallSpy = jest
+						.spyOn(defaultTaskRunner, 'makeRpcCall')
+						.mockResolvedValue(undefined);
+
+					// Act
+					await execTaskWithParams({
+						task: newTaskWithSettings({
+							code: `await ${group.invocation}; return {}`,
+							nodeMode: 'runOnceForEachItem',
+						}),
+						taskData: newDataRequestResponse(
+							[{ json: {}, binary: { binaryFile: binaryDataFile } }],
+							{},
+						),
+					});
+
+					expect(rpcCallSpy).toHaveBeenCalledWith('1', group.method, group.expectedParams);
+				});
+			}
+
+			describe('unsupported methods', () => {
+				for (const unsupportedFunction of UNSUPPORTED_HELPER_FUNCTIONS) {
+					it(`should throw an error if ${unsupportedFunction} is used in runOnceForAllItems`, async () => {
+						// Act
+
+						await expect(
+							async () =>
+								await executeForAllItems({
+									code: `${unsupportedFunction}()`,
+									inputItems,
+								}),
+						).rejects.toThrow(UnsupportedFunctionError);
+					});
+
+					it(`should throw an error if ${unsupportedFunction} is used in runOnceForEachItem`, async () => {
+						// Act
+
+						await expect(
+							async () =>
+								await executeForEachItem({
+									code: `${unsupportedFunction}()`,
+									inputItems,
+								}),
+						).rejects.toThrow(UnsupportedFunctionError);
+					});
+				}
+			});
+		});
+
 		it('should allow access to Node.js Buffers', async () => {
 			const outcomeAll = await execTaskWithParams({
 				task: newTaskWithSettings({

packages/@n8n/task-runner/src/js-task-runner/errors/unsupported-function.error.ts

@@ -0,0 +1,13 @@
+import { ApplicationError } from 'n8n-workflow';
+
+/**
+ * Error that indicates that a specific function is not available in the
+ * Code Node.
+ */
+export class UnsupportedFunctionError extends ApplicationError {
+	constructor(functionName: string) {
+		super(`The function "${functionName}" is not supported in the Code Node`, {
+			level: 'info',
+		});
+	}
+}

packages/@n8n/task-runner/src/js-task-runner/js-task-runner.ts

@@ -1,3 +1,4 @@
+import set from 'lodash.set';
 import { getAdditionalKeys } from 'n8n-core';
 import { WorkflowDataProxy, Workflow, ObservableObject } from 'n8n-workflow';
 import type {
@@ -19,11 +20,14 @@ import * as a from 'node:assert';
 import { runInNewContext, type Context } from 'node:vm';
 
 import type { MainConfig } from '@/config/main-config';
-import type {
-	DataRequestResponse,
-	InputDataChunkDefinition,
-	PartialAdditionalData,
-	TaskResultData,
+import { UnsupportedFunctionError } from '@/js-task-runner/errors/unsupported-function.error';
+import {
+	EXPOSED_RPC_METHODS,
+	UNSUPPORTED_HELPER_FUNCTIONS,
+	type DataRequestResponse,
+	type InputDataChunkDefinition,
+	type PartialAdditionalData,
+	type TaskResultData,
 } from '@/runner-types';
 import { type Task, TaskRunner } from '@/task-runner';
 
@@ -38,6 +42,10 @@ import { createRequireResolver } from './require-resolver';
 import { validateRunForAllItemsOutput, validateRunForEachItemOutput } from './result-validation';
 import { DataRequestResponseReconstruct } from '../data-request/data-request-response-reconstruct';
 
+export interface RPCCallObject {
+	[name: string]: ((...args: unknown[]) => Promise<unknown>) | RPCCallObject;
+}
+
 export interface JSExecSettings {
 	code: string;
 	nodeMode: CodeExecutionMode;
@@ -439,4 +447,24 @@ export class JsTaskRunner extends TaskRunner {
 			this.nodeTypes.addNodeTypeDescriptions(nodeTypes);
 		}
 	}
+
+	private buildRpcCallObject(taskId: string) {
+		const rpcObject: RPCCallObject = {};
+
+		for (const rpcMethod of EXPOSED_RPC_METHODS) {
+			set(
+				rpcObject,
+				rpcMethod.split('.'),
+				async (...args: unknown[]) => await this.makeRpcCall(taskId, rpcMethod, args),
+			);
+		}
+
+		for (const rpcMethod of UNSUPPORTED_HELPER_FUNCTIONS) {
+			set(rpcObject, rpcMethod.split('.'), () => {
+				throw new UnsupportedFunctionError(rpcMethod);
+			});
+		}
+
+		return rpcObject;
+	}
 }

packages/@n8n/task-runner/src/message-types.ts

@@ -2,7 +2,7 @@ import type { INodeTypeBaseDescription } from 'n8n-workflow';
 
 import type {
 	NeededNodeType,
-	RPC_ALLOW_LIST,
+	AVAILABLE_RPC_METHODS,
 	TaskDataRequestParams,
 	TaskResultData,
 } from './runner-types';
@@ -105,7 +105,7 @@ export namespace BrokerMessage {
 			type: 'broker:rpc';
 			callId: string;
 			taskId: string;
-			name: (typeof RPC_ALLOW_LIST)[number];
+			name: (typeof AVAILABLE_RPC_METHODS)[number];
 			params: unknown[];
 		}
 
@@ -239,7 +239,7 @@ export namespace RunnerMessage {
 			type: 'runner:rpc';
 			callId: string;
 			taskId: string;
-			name: (typeof RPC_ALLOW_LIST)[number];
+			name: (typeof AVAILABLE_RPC_METHODS)[number];
 			params: unknown[];
 		}
 

packages/@n8n/task-runner/src/runner-types.ts

@@ -100,31 +100,73 @@ export interface PartialAdditionalData {
 	variables: IDataObject;
 }
 
-export const RPC_ALLOW_LIST = [
+/** RPC methods that are exposed directly to the Code Node */
+export const EXPOSED_RPC_METHODS = [
+	// assertBinaryData(itemIndex: number, propertyName: string): Promise<IBinaryData>
+	'helpers.assertBinaryData',
+
+	// getBinaryDataBuffer(itemIndex: number, propertyName: string): Promise<Buffer>
+	'helpers.getBinaryDataBuffer',
+
+	// prepareBinaryData(binaryData: Buffer, fileName?: string, mimeType?: string): Promise<IBinaryData>
+	'helpers.prepareBinaryData',
+
+	// setBinaryDataBuffer(metadata: IBinaryData, buffer: Buffer): Promise<IBinaryData>
+	'helpers.setBinaryDataBuffer',
+
+	// binaryToString(body: Buffer, encoding?: string): string
+	'helpers.binaryToString',
+
+	// httpRequest(opts: IHttpRequestOptions): Promise<IN8nHttpFullResponse | IN8nHttpResponse>
+	'helpers.httpRequest',
+];
+
+/** Helpers that exist but that we are not exposing to the Code Node */
+export const UNSUPPORTED_HELPER_FUNCTIONS = [
+	// These rely on checking the credentials from the current node type (Code Node)
+	// and hence they can't even work (Code Node doesn't have credentials)
 	'helpers.httpRequestWithAuthentication',
 	'helpers.requestWithAuthenticationPaginated',
-	// "helpers.normalizeItems"
-	// "helpers.constructExecutionMetaData"
-	// "helpers.assertBinaryData"
-	'helpers.getBinaryDataBuffer',
-	// "helpers.copyInputItems"
-	// "helpers.returnJsonArray"
-	'helpers.getSSHClient',
-	'helpers.createReadStream',
-	// "helpers.getStoragePath"
-	'helpers.writeContentToFile',
-	'helpers.prepareBinaryData',
-	'helpers.setBinaryDataBuffer',
+
+	// This has been removed
 	'helpers.copyBinaryFile',
-	'helpers.binaryToBuffer',
-	// "helpers.binaryToString"
-	// "helpers.getBinaryPath"
+
+	// We can't support streams over RPC without implementing it ourselves
+	'helpers.createReadStream',
 	'helpers.getBinaryStream',
+
+	// Makes no sense to support this, as it returns either a stream or a buffer
+	// and we can't support streams over RPC
+	'helpers.binaryToBuffer',
+
+	// These are pretty low-level, so we shouldn't expose them
+	// (require binary data id, which we don't expose)
 	'helpers.getBinaryMetadata',
+	'helpers.getStoragePath',
+	'helpers.getBinaryPath',
+
+	// We shouldn't allow arbitrary FS writes
+	'helpers.writeContentToFile',
+
+	// Not something we need to expose. Can be done in the node itself
+	// copyInputItems(items: INodeExecutionData[], properties: string[]): IDataObject[]
+	'helpers.copyInputItems',
+
+	// Code Node does these automatically already
+	'helpers.returnJsonArray',
+	'helpers.normalizeItems',
+
+	// The client is instantiated and lives on the n8n instance, so we can't
+	// expose it over RPC without implementing object marshalling
+	'helpers.getSSHClient',
+
+	// Doesn't make sense to expose
 	'helpers.createDeferredPromise',
-	'helpers.httpRequest',
-	'logNodeOutput',
-] as const;
+	'helpers.constructExecutionMetaData',
+];
+
+/** List of all RPC methods that task runner supports */
+export const AVAILABLE_RPC_METHODS = [...EXPOSED_RPC_METHODS, 'logNodeOutput'] as const;
 
 /** Node types needed for the runner to execute a task. */
 export type NeededNodeType = { name: string; version: number };

packages/@n8n/task-runner/src/task-runner.ts

@@ -1,3 +1,4 @@
+import { isSerializedBuffer, toBuffer } from 'n8n-core';
 import { ApplicationError, ensureError, randomInt } from 'n8n-workflow';
 import { nanoid } from 'nanoid';
 import { EventEmitter } from 'node:events';
@@ -6,7 +7,7 @@ import { type MessageEvent, WebSocket } from 'ws';
 import type { BaseRunnerConfig } from '@/config/base-runner-config';
 import type { BrokerMessage, RunnerMessage } from '@/message-types';
 import { TaskRunnerNodeTypes } from '@/node-types';
-import { RPC_ALLOW_LIST, type TaskResultData } from '@/runner-types';
+import type { TaskResultData } from '@/runner-types';
 
 import { TaskCancelledError } from './js-task-runner/errors/task-cancelled-error';
 
@@ -42,10 +43,6 @@ interface RPCCall {
 	reject: (error: unknown) => void;
 }
 
-export interface RPCCallObject {
-	[name: string]: ((...args: unknown[]) => Promise<unknown>) | RPCCallObject;
-}
-
 const OFFER_VALID_TIME_MS = 5000;
 const OFFER_VALID_EXTRA_MS = 100;
 
@@ -464,7 +461,9 @@ export abstract class TaskRunner extends EventEmitter {
 		});
 
 		try {
-			return await dataPromise;
+			const returnValue = await dataPromise;
+
+			return isSerializedBuffer(returnValue) ? toBuffer(returnValue) : returnValue;
 		} finally {
 			this.rpcCalls.delete(callId);
 		}
@@ -486,24 +485,6 @@ export abstract class TaskRunner extends EventEmitter {
 		}
 	}
 
-	buildRpcCallObject(taskId: string) {
-		const rpcObject: RPCCallObject = {};
-		for (const r of RPC_ALLOW_LIST) {
-			const splitPath = r.split('.');
-			let obj = rpcObject;
-
-			splitPath.forEach((s, index) => {
-				if (index !== splitPath.length - 1) {
-					obj[s] = {};
-					obj = obj[s];
-					return;
-				}
-				obj[s] = async (...args: unknown[]) => await this.makeRpcCall(taskId, r, args);
-			});
-		}
-		return rpcObject;
-	}
-
 	/** Close the connection gracefully and wait until has been closed */
 	async stop() {
 		this.clearIdleTimer();