mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
fix(core): Redact a parent execution if any sub-workflows have redaction enabled (#41129)
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
c28d130c6c
commit
c70b5d49b3
|
|
@ -23,6 +23,7 @@ import type {
|
|||
ITaskData,
|
||||
WorkflowExecuteMode,
|
||||
ExecuteAgentWorkflowContext,
|
||||
IExecutionContext,
|
||||
IRunExecutionData,
|
||||
IWorkflowExecutionDataProcess,
|
||||
} from 'n8n-workflow';
|
||||
|
|
@ -898,6 +899,144 @@ describe('WorkflowExecuteAdditionalData', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('sub-workflow redaction reporting', () => {
|
||||
const childContext = (fields: Partial<IExecutionContext>): IExecutionContext => ({
|
||||
version: 1,
|
||||
establishedAt: 0,
|
||||
source: 'integrated',
|
||||
...fields,
|
||||
});
|
||||
|
||||
// A plain object: `mock()` does not keep `false` values in nested objects.
|
||||
const getMockRunWithContext = (runtimeData: IExecutionContext, failed = false) =>
|
||||
({
|
||||
data: {
|
||||
subWorkflowOutput: undefined,
|
||||
resultData: {
|
||||
runData: {
|
||||
[LAST_NODE_EXECUTED]: [
|
||||
{ startTime: 100, data: { main: [[{ json: { test: 1 } }]] } },
|
||||
],
|
||||
},
|
||||
lastNodeExecuted: LAST_NODE_EXECUTED,
|
||||
...(failed && { error: { message: 'child failed', name: 'NodeOperationError' } }),
|
||||
},
|
||||
executionData: { runtimeData },
|
||||
},
|
||||
finished: !failed,
|
||||
mode: 'integrated',
|
||||
startedAt: new Date(),
|
||||
status: failed ? 'error' : 'success',
|
||||
waitTill: undefined,
|
||||
}) as unknown as IRun;
|
||||
|
||||
const run = async (doNotWaitToFinish = false) =>
|
||||
await executeWorkflow(
|
||||
mock<IExecuteWorkflowInfo>(),
|
||||
mock<IWorkflowExecuteAdditionalData>(),
|
||||
mock<ExecuteWorkflowOptions>({ loadedWorkflowData: undefined, doNotWaitToFinish }),
|
||||
);
|
||||
|
||||
it('reports the channels the sub-workflow redacts', async () => {
|
||||
processRunExecutionData.mockReturnValue(
|
||||
getCancelablePromise(
|
||||
getMockRunWithContext(
|
||||
childContext({ redaction: { version: 2, production: true, manual: false } }),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
const response = await run();
|
||||
|
||||
expect(response.subWorkflowRedaction).toEqual({ production: true, manual: false });
|
||||
});
|
||||
|
||||
it('also reports the channels its own sub-workflows reported to it', async () => {
|
||||
processRunExecutionData.mockReturnValue(
|
||||
getCancelablePromise(
|
||||
getMockRunWithContext(
|
||||
childContext({
|
||||
redaction: { version: 2, production: false, manual: false },
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
}),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
const response = await run();
|
||||
|
||||
expect(response.subWorkflowRedaction).toEqual({ production: true, manual: true });
|
||||
});
|
||||
|
||||
it('reports nothing when the sub-workflow redacts no channel', async () => {
|
||||
processRunExecutionData.mockReturnValue(
|
||||
getCancelablePromise(
|
||||
getMockRunWithContext(
|
||||
childContext({ redaction: { version: 2, production: false, manual: false } }),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
const response = await run();
|
||||
|
||||
expect(response.subWorkflowRedaction).toBeUndefined();
|
||||
});
|
||||
|
||||
it('reports nothing for detached sub-workflows (no embedded output)', async () => {
|
||||
processRunExecutionData.mockReturnValue(
|
||||
getCancelablePromise(
|
||||
getMockRunWithContext(
|
||||
childContext({ redaction: { version: 2, production: true, manual: true } }),
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
const response = await run(true);
|
||||
|
||||
expect(response.subWorkflowRedaction).toBeUndefined();
|
||||
});
|
||||
|
||||
it('attaches the channels to the error of a failed sub-workflow', async () => {
|
||||
processRunExecutionData.mockReturnValue(
|
||||
getCancelablePromise(
|
||||
getMockRunWithContext(
|
||||
childContext({ redaction: { version: 2, production: true, manual: false } }),
|
||||
true,
|
||||
),
|
||||
),
|
||||
);
|
||||
|
||||
await expect(run()).rejects.toMatchObject({
|
||||
message: 'child failed',
|
||||
subWorkflowRedaction: { production: true, manual: false },
|
||||
});
|
||||
});
|
||||
|
||||
it('attaches the channels to the error when the engine throws mid-run', async () => {
|
||||
const executionData = createRunExecutionData();
|
||||
// The engine mutates the run data in place, so the context it established
|
||||
// before the crash is still readable.
|
||||
processRunExecutionData.mockImplementationOnce(async () => {
|
||||
executionData.executionData!.runtimeData = childContext({
|
||||
redaction: { version: 2, production: true, manual: true },
|
||||
});
|
||||
throw new Error('engine crashed');
|
||||
});
|
||||
|
||||
await expect(
|
||||
executeWorkflow(mock<IExecuteWorkflowInfo>(), mock<IWorkflowExecuteAdditionalData>(), {
|
||||
...mock<ExecuteWorkflowOptions>(),
|
||||
loadedWorkflowData: undefined,
|
||||
loadedRunData: { ...mock<IWorkflowExecutionDataProcess>(), executionData },
|
||||
doNotWaitToFinish: false,
|
||||
}),
|
||||
).rejects.toMatchObject({
|
||||
message: 'engine crashed',
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Tests for workflow version selection based on execution mode.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -1387,6 +1387,97 @@ describe('updateParentExecutionWithChildResults', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('sub-workflow redaction', () => {
|
||||
const parentWithContext = (subWorkflowRedaction?: {
|
||||
production: boolean;
|
||||
manual: boolean;
|
||||
}): IExecutionResponse => {
|
||||
const parent = waitingParent();
|
||||
parent.data.executionData!.runtimeData = {
|
||||
version: 1,
|
||||
establishedAt: 0,
|
||||
source: 'manual',
|
||||
redaction: { version: 2, production: false, manual: false },
|
||||
...(subWorkflowRedaction && { subWorkflowRedaction }),
|
||||
};
|
||||
return parent;
|
||||
};
|
||||
|
||||
const childWithRedaction = (
|
||||
redaction: { production: boolean; manual: boolean },
|
||||
error?: ExecutionError,
|
||||
): IRun =>
|
||||
({
|
||||
mode: 'integrated',
|
||||
status: error ? 'error' : 'success',
|
||||
data: {
|
||||
resultData: {
|
||||
lastNodeExecuted: 'Done',
|
||||
error,
|
||||
runData: { Done: [error ? { error } : { data: { main: [[{ json: { out: 1 } }]] } }] },
|
||||
},
|
||||
executionData: {
|
||||
runtimeData: {
|
||||
version: 1,
|
||||
establishedAt: 0,
|
||||
source: 'integrated',
|
||||
redaction: { version: 2, ...redaction },
|
||||
},
|
||||
},
|
||||
},
|
||||
}) as unknown as IRun;
|
||||
|
||||
async function persistedContextAfter(parent: IExecutionResponse, child: IRun) {
|
||||
const executionPersistence = mockInstance(ExecutionPersistence);
|
||||
executionPersistence.findSingleExecution.mockResolvedValue(parent);
|
||||
|
||||
await updateParentExecutionWithChildResults(PARENT_ID, child);
|
||||
|
||||
const [, payload] = executionPersistence.updateExistingExecution.mock.calls[0];
|
||||
return (payload as IExecutionResponse).data.executionData?.runtimeData;
|
||||
}
|
||||
|
||||
// The resume keeps the saved context, so the escalation must be in it before it is saved.
|
||||
it('escalates the parent context with the channels of the child final run', async () => {
|
||||
const runtimeData = await persistedContextAfter(
|
||||
parentWithContext(),
|
||||
childWithRedaction({ production: true, manual: true }),
|
||||
);
|
||||
|
||||
expect(runtimeData?.subWorkflowRedaction).toEqual({ production: true, manual: true });
|
||||
// The start snapshot keeps its meaning.
|
||||
expect(runtimeData?.redaction).toEqual({ version: 2, production: false, manual: false });
|
||||
});
|
||||
|
||||
it('escalates the parent context when the child failed', async () => {
|
||||
const error = { name: 'NodeOperationError', message: 'ERROR' } as unknown as ExecutionError;
|
||||
const runtimeData = await persistedContextAfter(
|
||||
parentWithContext(),
|
||||
childWithRedaction({ production: true, manual: false }, error),
|
||||
);
|
||||
|
||||
expect(runtimeData?.subWorkflowRedaction).toEqual({ production: true, manual: false });
|
||||
});
|
||||
|
||||
it('never lowers a channel that an earlier child reported', async () => {
|
||||
const runtimeData = await persistedContextAfter(
|
||||
parentWithContext({ production: true, manual: true }),
|
||||
childWithRedaction({ production: true, manual: false }),
|
||||
);
|
||||
|
||||
expect(runtimeData?.subWorkflowRedaction).toEqual({ production: true, manual: true });
|
||||
});
|
||||
|
||||
it('leaves the parent context alone when the child redacts no channel', async () => {
|
||||
const runtimeData = await persistedContextAfter(
|
||||
parentWithContext(),
|
||||
childWithRedaction({ production: false, manual: false }),
|
||||
);
|
||||
|
||||
expect(runtimeData).not.toHaveProperty('subWorkflowRedaction');
|
||||
});
|
||||
});
|
||||
|
||||
// In "run once for each item" mode multiple children update the same waiting
|
||||
// parent. Contract: if any child errored before the parent resumed, the parent
|
||||
// node fails — a later successful sibling must not mask the error.
|
||||
|
|
|
|||
|
|
@ -84,6 +84,7 @@ describe('ExecutionRedactionService', () => {
|
|||
withDynamicCredentials?: boolean;
|
||||
usesDynamicCredentials?: boolean;
|
||||
executedByUserId?: string | null;
|
||||
subWorkflowRedaction?: { production: boolean; manual: boolean };
|
||||
nodes?: INode[];
|
||||
} = {},
|
||||
): RedactableExecution => {
|
||||
|
|
@ -97,6 +98,7 @@ describe('ExecutionRedactionService', () => {
|
|||
withDynamicCredentials = false,
|
||||
usesDynamicCredentials = false,
|
||||
executedByUserId = null,
|
||||
subWorkflowRedaction,
|
||||
nodes = [],
|
||||
} = overrides;
|
||||
|
||||
|
|
@ -155,6 +157,18 @@ describe('ExecutionRedactionService', () => {
|
|||
};
|
||||
}
|
||||
|
||||
// Stamp the channels a sub-workflow reported, mirroring what
|
||||
// `BaseExecuteContext.executeWorkflow` does when a sub-workflow returns.
|
||||
if (subWorkflowRedaction) {
|
||||
executionData.runtimeData = {
|
||||
version: 1 as const,
|
||||
establishedAt: Date.now(),
|
||||
source: mode,
|
||||
...executionData.runtimeData,
|
||||
subWorkflowRedaction,
|
||||
};
|
||||
}
|
||||
|
||||
const runData = withDynamicCredentials
|
||||
? {
|
||||
SomeNode: [{ startTime: 0, executionTime: 0, usedDynamicCredentials: true } as ITaskData],
|
||||
|
|
@ -1000,6 +1014,170 @@ describe('ExecutionRedactionService', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('sub-workflow redaction', () => {
|
||||
const NO_CHANNELS = { production: false, manual: false };
|
||||
const BOTH_CHANNELS = { production: true, manual: true };
|
||||
const PRODUCTION_ONLY = { production: true, manual: false };
|
||||
|
||||
it('redacts an execution whose sub-workflow reported a channel that covers its mode', async () => {
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: BOTH_CHANNELS,
|
||||
mode: 'manual',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
expect(fullItemRedactionStrategy.apply).toHaveBeenCalledTimes(1);
|
||||
const [, context] = fullItemRedactionStrategy.apply.mock.calls[0];
|
||||
expect(context.userCanReveal).toBe(false);
|
||||
});
|
||||
|
||||
it('does not redact a manual execution when the sub-workflow reported production only', async () => {
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: PRODUCTION_ONLY,
|
||||
mode: 'manual',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
expect(fullItemRedactionStrategy.apply).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('redacts a production execution when the sub-workflow reported production only', async () => {
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: PRODUCTION_ONLY,
|
||||
mode: 'trigger',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
expect(fullItemRedactionStrategy.apply).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('combines the channels with a V1 snapshot', async () => {
|
||||
const execution = makeExecution({
|
||||
policy: 'non-manual',
|
||||
subWorkflowRedaction: BOTH_CHANNELS,
|
||||
mode: 'manual',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
expect(fullItemRedactionStrategy.apply).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('combines the channels with the workflow-setting fallback', async () => {
|
||||
const execution = makeExecution({
|
||||
withRuntimeData: false,
|
||||
workflowSettingsPolicy: 'none',
|
||||
subWorkflowRedaction: PRODUCTION_ONLY,
|
||||
mode: 'webhook',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
expect(fullItemRedactionStrategy.apply).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('ignores the channels when the data-redaction license is missing', async () => {
|
||||
licenseState.isDataRedactionLicensed.mockReturnValue(false);
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: BOTH_CHANNELS,
|
||||
mode: 'manual',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
expect(fullItemRedactionStrategy.apply).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// Escalated redaction follows the policy rule: `execution:reveal` on the calling
|
||||
// workflow allows reveal.
|
||||
it('lets a user with execution:reveal on the workflow reveal', async () => {
|
||||
workflowFinderService.findWorkflowIdsWithScopeForUser.mockResolvedValue(
|
||||
new Set(['workflow-123']),
|
||||
);
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: BOTH_CHANNELS,
|
||||
mode: 'manual',
|
||||
});
|
||||
|
||||
await expect(
|
||||
service.processExecution(execution, { user: mockUser, redactExecutionData: false }),
|
||||
).resolves.toBe(execution);
|
||||
expect(fullItemRedactionStrategy.apply).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('advertises canReveal to a user with execution:reveal on the workflow', async () => {
|
||||
workflowFinderService.findWorkflowIdsWithScopeForUser.mockResolvedValue(
|
||||
new Set(['workflow-123']),
|
||||
);
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: BOTH_CHANNELS,
|
||||
mode: 'manual',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
const [, context] = fullItemRedactionStrategy.apply.mock.calls[0];
|
||||
expect(context.userCanReveal).toBe(true);
|
||||
});
|
||||
|
||||
it('rejects reveal for a user without execution:reveal on the workflow', async () => {
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: BOTH_CHANNELS,
|
||||
mode: 'manual',
|
||||
});
|
||||
|
||||
await expect(
|
||||
service.processExecution(execution, { user: mockUser, redactExecutionData: false }),
|
||||
).rejects.toBeInstanceOf(ScopeForbiddenError);
|
||||
expect(eventService.emit).toHaveBeenCalledWith(
|
||||
'execution-data-reveal-failure',
|
||||
expect.objectContaining({ redactionPolicy: 'all' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps the dynamic-credentials rule ahead of the sub-workflow redaction', async () => {
|
||||
workflowFinderService.findWorkflowIdsWithScopeForUser.mockResolvedValue(
|
||||
new Set(['workflow-123']),
|
||||
);
|
||||
const execution = makeExecution({
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: BOTH_CHANNELS,
|
||||
withDynamicCredentials: true,
|
||||
mode: 'manual',
|
||||
});
|
||||
await service.processExecution(execution, { user: mockUser });
|
||||
|
||||
const [, context] = fullItemRedactionStrategy.apply.mock.calls[0];
|
||||
expect(context.enforceDynCredRedaction).toBe(true);
|
||||
// The execution:reveal scope does not bypass the dynamic-credentials rule.
|
||||
expect(context.userCanReveal).toBe(false);
|
||||
});
|
||||
|
||||
it.each([
|
||||
[PRODUCTION_ONLY, 'manual' as const],
|
||||
[PRODUCTION_ONLY, 'trigger' as const],
|
||||
[BOTH_CHANNELS, 'manual' as const],
|
||||
[BOTH_CHANNELS, 'webhook' as const],
|
||||
])('channels %j mode %s: console gate matches data pipeline', async (reported, mode) => {
|
||||
const redaction = { version: 2 as const, ...NO_CHANNELS };
|
||||
const execution = makeExecution({
|
||||
mode,
|
||||
channels: NO_CHANNELS,
|
||||
subWorkflowRedaction: reported,
|
||||
});
|
||||
|
||||
await service.processExecutions([execution], { user: mockUser });
|
||||
const dataPipelineRedacts = fullItemRedactionStrategy.apply.mock.calls.length > 0;
|
||||
|
||||
expect(shouldRedactConsoleOutput(redaction, undefined, mode, reported)).toBe(
|
||||
dataPipelineRedacts,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe('console-gate equivalence', () => {
|
||||
// The console gate (shouldRedactConsoleOutput) and the execution-data
|
||||
// pipeline resolve the same snapshot independently; this pins them to
|
||||
|
|
|
|||
|
|
@ -9,6 +9,8 @@ import type { User } from '@n8n/db';
|
|||
import { Service } from '@n8n/di';
|
||||
import {
|
||||
channelsToPolicy,
|
||||
mergeRedactionChannels,
|
||||
policyToChannels,
|
||||
runDataUsedCredentialIds,
|
||||
runDataUsedDynamicCredentials,
|
||||
WorkflowExecuteMode,
|
||||
|
|
@ -415,18 +417,32 @@ export class ExecutionRedactionService implements ExecutionRedaction {
|
|||
}
|
||||
|
||||
/**
|
||||
* Resolves the effective redaction policy for an execution.
|
||||
*
|
||||
* Prefers the snapshot captured in `runtimeData.redaction` at execution time. That snapshot
|
||||
* is versioned: V2 stores per-channel booleans (reconstructed into the policy enum), while
|
||||
* V1 (older executions) stores the policy enum directly. Falls back to `workflowData.settings`
|
||||
* for executions captured before runtime snapshots existed, and defaults to 'none'.
|
||||
* Resolves the effective redaction policy for an execution: its own policy (see
|
||||
* {@link resolveBasePolicy}) combined, strictest per channel, with the channels its
|
||||
* sub-workflows reported (`runtimeData.subWorkflowRedaction`). The execution keeps a
|
||||
* copy of each sub-workflow's output, so it redacts at least what they redact.
|
||||
* Returns 'none' when the data-redaction license is not active, so that user-configured
|
||||
* policies are not applied without the license.
|
||||
*/
|
||||
private resolvePolicy(execution: RedactableExecution): WorkflowSettings.RedactionPolicy {
|
||||
if (!this.licenseState.isDataRedactionLicensed()) return 'none';
|
||||
|
||||
const policy = this.resolveBasePolicy(execution);
|
||||
const reported = execution.data.executionData?.runtimeData?.subWorkflowRedaction;
|
||||
if (!reported?.production && !reported?.manual) return policy;
|
||||
|
||||
return channelsToPolicy(mergeRedactionChannels(policyToChannels(policy), reported));
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves the execution's own redaction policy.
|
||||
*
|
||||
* Prefers the snapshot captured in `runtimeData.redaction` at execution time. That snapshot
|
||||
* is versioned: V2 stores per-channel booleans (reconstructed into the policy enum), while
|
||||
* V1 (older executions) stores the policy enum directly. Falls back to `workflowData.settings`
|
||||
* for executions captured before runtime snapshots existed, and defaults to 'none'.
|
||||
*/
|
||||
private resolveBasePolicy(execution: RedactableExecution): WorkflowSettings.RedactionPolicy {
|
||||
const redaction = execution.data.executionData?.runtimeData?.redaction;
|
||||
|
||||
if (redaction?.version === 2) {
|
||||
|
|
|
|||
|
|
@ -54,6 +54,8 @@ import {
|
|||
getSubWorkflowOutputPolicy,
|
||||
attachDynamicCredentialsUsage,
|
||||
summarizeDynamicCredentialsUsage,
|
||||
attachSubWorkflowRedaction,
|
||||
summarizeSubWorkflowRedaction,
|
||||
} from 'n8n-workflow';
|
||||
|
||||
import type { PrepareWorkflowAgentForEval } from './modules/agents/agent-runtime-instrumentation';
|
||||
|
|
@ -755,18 +757,21 @@ async function startExecution(
|
|||
// The engine mutates `runData.executionData` in place, so a mid-run crash may leave
|
||||
// credential flags on already-executed tasks — ride them on the error like the
|
||||
// regular failure path below.
|
||||
throw attachDynamicCredentialsUsage(
|
||||
objectToError(
|
||||
{
|
||||
...executionError,
|
||||
executionId,
|
||||
workflowId: workflowData.id,
|
||||
stack: executionError?.stack,
|
||||
message: executionError?.message,
|
||||
},
|
||||
workflow,
|
||||
throw attachSubWorkflowRedaction(
|
||||
attachDynamicCredentialsUsage(
|
||||
objectToError(
|
||||
{
|
||||
...executionError,
|
||||
executionId,
|
||||
workflowId: workflowData.id,
|
||||
stack: executionError?.stack,
|
||||
message: executionError?.message,
|
||||
},
|
||||
workflow,
|
||||
),
|
||||
summarizeDynamicCredentialsUsage(runData.executionData, workflowCredentialIds),
|
||||
),
|
||||
summarizeDynamicCredentialsUsage(runData.executionData, workflowCredentialIds),
|
||||
summarizeSubWorkflowRedaction(runData.executionData, { includeOwn: true }),
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -785,6 +790,8 @@ async function startExecution(
|
|||
waitTill: data.waitTill,
|
||||
// Report private-credential usage to the caller (detached runs return earlier, skipping this).
|
||||
...summarizeDynamicCredentialsUsage(data.data, workflowCredentialIds),
|
||||
// The caller keeps a copy of the output above, so it must redact what this run redacts.
|
||||
subWorkflowRedaction: summarizeSubWorkflowRedaction(data.data, { includeOwn: true }),
|
||||
};
|
||||
}
|
||||
activeExecutions.finalizeExecution(executionId, data);
|
||||
|
|
@ -794,18 +801,21 @@ async function startExecution(
|
|||
|
||||
// A failed child may still have attempted or resolved private credentials before failing;
|
||||
// ride the usage on the error so a continue-on-fail caller still flags its task
|
||||
// (see `BaseExecuteContext.executeWorkflow`).
|
||||
throw attachDynamicCredentialsUsage(
|
||||
objectToError(
|
||||
{
|
||||
...error,
|
||||
executionId,
|
||||
workflowId: workflowData.id,
|
||||
stack: error?.stack,
|
||||
},
|
||||
workflow,
|
||||
// (see `BaseExecuteContext.executeWorkflow`). The redaction channels ride along the same way.
|
||||
throw attachSubWorkflowRedaction(
|
||||
attachDynamicCredentialsUsage(
|
||||
objectToError(
|
||||
{
|
||||
...error,
|
||||
executionId,
|
||||
workflowId: workflowData.id,
|
||||
stack: error?.stack,
|
||||
},
|
||||
workflow,
|
||||
),
|
||||
summarizeDynamicCredentialsUsage(data.data, workflowCredentialIds),
|
||||
),
|
||||
summarizeDynamicCredentialsUsage(data.data, workflowCredentialIds),
|
||||
summarizeSubWorkflowRedaction(data.data, { includeOwn: true }),
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import type { WorkflowEntity, WorkflowHistory } from '@n8n/db';
|
|||
import { Container } from '@n8n/di';
|
||||
import {
|
||||
collectSubWorkflowOutput,
|
||||
escalateSubWorkflowRedaction,
|
||||
getLastExecutedNodeData,
|
||||
Workflow,
|
||||
UnexpectedError,
|
||||
|
|
@ -17,6 +18,7 @@ import {
|
|||
resolveVariables,
|
||||
safeParseWorkflowStructure,
|
||||
summarizeDynamicCredentialsUsage,
|
||||
summarizeSubWorkflowRedaction,
|
||||
validateWorkflowGroups,
|
||||
type IDataObject,
|
||||
type INodeCredentialsDetails,
|
||||
|
|
@ -554,6 +556,13 @@ export async function updateParentExecutionWithChildResults(
|
|||
}
|
||||
}
|
||||
|
||||
// The parent keeps a copy of the child's output (or error), so it redacts at least what the
|
||||
// child redacts. The resume restores this context as is, so escalate it before it is saved.
|
||||
escalateSubWorkflowRedaction(
|
||||
parentWithSubWorkflowResults.data.executionData?.runtimeData,
|
||||
summarizeSubWorkflowRedaction(subworkflowResults.data, { includeOwn: true }),
|
||||
);
|
||||
|
||||
if (subworkflowError) {
|
||||
// Record the error on the waiting parent's Execute Workflow node so the node
|
||||
// fails with error on resume instead of appearing as successful. `subExecution`
|
||||
|
|
|
|||
|
|
@ -0,0 +1,270 @@
|
|||
/**
|
||||
* Runs a parent workflow that calls a sub-workflow with a redaction policy, then reads
|
||||
* the executions back through the REST API. The parent keeps a copy of the
|
||||
* sub-workflow's output, so the parent's execution must be redacted as well.
|
||||
*/
|
||||
import { createWorkflow, testDb } from '@n8n/backend-test-utils';
|
||||
import { ExecutionRepository, type IWorkflowDb, type User } from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
import { parse } from 'flatted';
|
||||
import { ExecutionContextHookRegistry } from 'n8n-core';
|
||||
import type { IRunExecutionData, WorkflowSettings } from 'n8n-workflow';
|
||||
import { NodeConnectionTypes } from 'n8n-workflow';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
|
||||
import { createOwner } from './shared/db/users';
|
||||
import type { SuperAgentTest } from './shared/types';
|
||||
import * as utils from './shared/utils/';
|
||||
import { loadNodesFromDist } from './shared/utils/node-types-data';
|
||||
import { createParentWorkflowFixture, createSubWorkflowFixture } from './shared/workflow-fixtures';
|
||||
|
||||
const testServer = utils.setupTestServer({
|
||||
endpointGroups: ['workflows', 'executions'],
|
||||
modules: ['redaction'],
|
||||
enabledFeatures: ['feat:dataRedaction'],
|
||||
});
|
||||
|
||||
let owner: User;
|
||||
let ownerAgent: SuperAgentTest;
|
||||
let executionRepository: ExecutionRepository;
|
||||
|
||||
beforeAll(async () => {
|
||||
// The redaction module registers its context hooks in `init()`, and the registry
|
||||
// only picks up decorated hooks when it initializes.
|
||||
await Container.get(ExecutionContextHookRegistry).init();
|
||||
|
||||
await utils.initNodeTypes(
|
||||
loadNodesFromDist([
|
||||
'n8n-nodes-base.manualTrigger',
|
||||
'n8n-nodes-base.executeWorkflow',
|
||||
'n8n-nodes-base.executeWorkflowTrigger',
|
||||
'n8n-nodes-base.stopAndError',
|
||||
]),
|
||||
);
|
||||
await utils.initBinaryDataService();
|
||||
|
||||
owner = await createOwner();
|
||||
ownerAgent = testServer.authAgentFor(owner);
|
||||
executionRepository = Container.get(ExecutionRepository);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await testDb.truncate(['ExecutionEntity', 'WorkflowEntity', 'SharedWorkflow']);
|
||||
});
|
||||
|
||||
async function createChild(name: string, redactionPolicy?: WorkflowSettings.RedactionPolicy) {
|
||||
return await createWorkflow(
|
||||
{
|
||||
name,
|
||||
...createSubWorkflowFixture(),
|
||||
settings: redactionPolicy ? { redactionPolicy } : {},
|
||||
} as unknown as IWorkflowDb,
|
||||
owner,
|
||||
);
|
||||
}
|
||||
|
||||
/** Manual Trigger → Execute Workflow (first child) → Execute Workflow (second child). */
|
||||
function createTwoCallParentFixture(firstChildId: string, secondChildId: string) {
|
||||
const callNode = (name: string, workflowId: string, x: number) => ({
|
||||
id: uuid(),
|
||||
name,
|
||||
type: 'n8n-nodes-base.executeWorkflow',
|
||||
typeVersion: 1.3,
|
||||
position: [x, 0] as [number, number],
|
||||
parameters: {
|
||||
workflowId: { __rl: true, value: workflowId, mode: 'id' },
|
||||
workflowInputs: {
|
||||
mappingMode: 'defineBelow',
|
||||
value: { test: 'test' },
|
||||
matchingColumns: [],
|
||||
schema: [],
|
||||
attemptToConvertTypes: false,
|
||||
convertFieldsToString: true,
|
||||
},
|
||||
options: {},
|
||||
},
|
||||
});
|
||||
const main = (node: string) => ({
|
||||
main: [[{ node, type: NodeConnectionTypes.Main, index: 0 }]],
|
||||
});
|
||||
|
||||
return {
|
||||
nodes: [
|
||||
{
|
||||
id: uuid(),
|
||||
name: 'Trigger',
|
||||
type: 'n8n-nodes-base.manualTrigger',
|
||||
typeVersion: 1,
|
||||
position: [0, 0] as [number, number],
|
||||
parameters: {},
|
||||
},
|
||||
callNode('Call first', firstChildId, 200),
|
||||
callNode('Call second', secondChildId, 400),
|
||||
],
|
||||
connections: { Trigger: main('Call first'), 'Call first': main('Call second') },
|
||||
pinData: {},
|
||||
};
|
||||
}
|
||||
|
||||
async function runManually(workflowId: string) {
|
||||
const response = await ownerAgent
|
||||
.post(`/workflows/${workflowId}/run`)
|
||||
.send({ triggerToStartFrom: { name: 'Trigger' } })
|
||||
.expect(200);
|
||||
const executionId = response.body.data.executionId as string;
|
||||
await waitForFinalStatus(executionId);
|
||||
return executionId;
|
||||
}
|
||||
|
||||
async function waitForFinalStatus(executionId: string, timeout = 20000) {
|
||||
const start = Date.now();
|
||||
while (Date.now() - start < timeout) {
|
||||
const execution = await executionRepository.findOne({ where: { id: executionId } });
|
||||
if (execution && ['error', 'success', 'crashed'].includes(execution.status)) {
|
||||
return execution;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
}
|
||||
throw new Error(`Execution ${executionId} did not finish within ${timeout}ms`);
|
||||
}
|
||||
|
||||
async function readExecution(executionId: string) {
|
||||
const response = await ownerAgent.get(`/executions/${executionId}`).expect(200);
|
||||
return parse(response.body.data.data as string) as IRunExecutionData;
|
||||
}
|
||||
|
||||
async function findChildExecutionId(workflowId: string) {
|
||||
const [execution] = await executionRepository.find({ where: { workflowId } });
|
||||
return execution.id;
|
||||
}
|
||||
|
||||
test('redacts the calling execution when the sub-workflow redacts its channel', async () => {
|
||||
const child = await createChild('Child with policy', 'all');
|
||||
const parent = await createWorkflow(
|
||||
{ name: 'Parent', ...createParentWorkflowFixture(child.id) } as unknown as IWorkflowDb,
|
||||
owner,
|
||||
);
|
||||
|
||||
const parentExecutionId = await runManually(parent.id);
|
||||
|
||||
const stored = await executionRepository.findSingleExecution(parentExecutionId, {
|
||||
includeData: true,
|
||||
unflattenData: true,
|
||||
});
|
||||
expect(stored?.status).toBe('success');
|
||||
expect(stored?.data.executionData?.runtimeData).toMatchObject({
|
||||
redaction: { version: 2, production: false, manual: false },
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
});
|
||||
|
||||
const data = await readExecution(parentExecutionId);
|
||||
expect(data.redactionInfo).toEqual({
|
||||
isRedacted: true,
|
||||
reason: 'workflow_redaction_policy',
|
||||
canReveal: true,
|
||||
});
|
||||
for (const nodeName of ['Trigger', 'Execute Workflow']) {
|
||||
for (const items of data.resultData.runData[nodeName][0].data!.main) {
|
||||
for (const item of items ?? []) {
|
||||
expect(item.json).toEqual({});
|
||||
expect(item.redaction).toEqual({
|
||||
redacted: true,
|
||||
reason: 'workflow_redaction_policy',
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
test('does not redact a manual calling execution when the sub-workflow redacts production only', async () => {
|
||||
const child = await createChild('Child with production policy', 'non-manual');
|
||||
const parent = await createWorkflow(
|
||||
{ name: 'Parent', ...createParentWorkflowFixture(child.id) } as unknown as IWorkflowDb,
|
||||
owner,
|
||||
);
|
||||
|
||||
const parentExecutionId = await runManually(parent.id);
|
||||
|
||||
const data = await readExecution(parentExecutionId);
|
||||
expect(data.redactionInfo).toBeUndefined();
|
||||
expect(data.resultData.runData['Execute Workflow'][0].data!.main[0]![0].json).toEqual({
|
||||
test: 'test',
|
||||
});
|
||||
});
|
||||
|
||||
test('redacts a later sibling sub-workflow that receives the redacted output', async () => {
|
||||
const redactedChild = await createChild('Child with policy', 'all');
|
||||
const plainChild = await createChild('Child without policy');
|
||||
const parent = await createWorkflow(
|
||||
{
|
||||
name: 'Parent',
|
||||
...createTwoCallParentFixture(redactedChild.id, plainChild.id),
|
||||
} as unknown as IWorkflowDb,
|
||||
owner,
|
||||
);
|
||||
|
||||
const parentExecutionId = await runManually(parent.id);
|
||||
expect((await readExecution(parentExecutionId)).redactionInfo?.reason).toBe(
|
||||
'workflow_redaction_policy',
|
||||
);
|
||||
|
||||
const plainChildData = await readExecution(await findChildExecutionId(plainChild.id));
|
||||
expect(plainChildData.redactionInfo).toMatchObject({ isRedacted: true });
|
||||
// The sibling inherits the escalation with the parent's context. Its own snapshot keeps its own policy.
|
||||
expect(plainChildData.executionData?.runtimeData).toMatchObject({
|
||||
redaction: { production: false, manual: false },
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
});
|
||||
});
|
||||
|
||||
test('redacts a retry that reuses the sub-workflow output without calling the sub-workflow again', async () => {
|
||||
const child = await createChild('Child with policy', 'all');
|
||||
const fixture = createParentWorkflowFixture(child.id);
|
||||
const parent = await createWorkflow(
|
||||
{
|
||||
name: 'Parent',
|
||||
...fixture,
|
||||
nodes: [
|
||||
...fixture.nodes,
|
||||
{
|
||||
id: uuid(),
|
||||
name: 'Stop and Error',
|
||||
type: 'n8n-nodes-base.stopAndError',
|
||||
typeVersion: 1,
|
||||
position: [400, 0] as [number, number],
|
||||
parameters: { errorMessage: 'Fail after the call' },
|
||||
},
|
||||
],
|
||||
connections: {
|
||||
...fixture.connections,
|
||||
'Execute Workflow': {
|
||||
main: [[{ node: 'Stop and Error', type: NodeConnectionTypes.Main, index: 0 }]],
|
||||
},
|
||||
},
|
||||
} as unknown as IWorkflowDb,
|
||||
owner,
|
||||
);
|
||||
|
||||
const parentExecutionId = await runManually(parent.id);
|
||||
expect((await waitForFinalStatus(parentExecutionId)).status).toBe('error');
|
||||
|
||||
await ownerAgent.post(`/executions/${parentExecutionId}/retry`).send({}).expect(200);
|
||||
|
||||
// The retry keeps the stored output of `Execute Workflow` and runs only the failed node, so
|
||||
// the escalation must come from the stored execution context.
|
||||
expect(await executionRepository.count({ where: { workflowId: child.id } })).toBe(1);
|
||||
const retry = await executionRepository.findOneOrFail({
|
||||
where: { retryOf: parentExecutionId },
|
||||
});
|
||||
expect(retry.mode).toBe('retry');
|
||||
|
||||
const data = await readExecution(retry.id);
|
||||
expect(data.executionData?.runtimeData?.subWorkflowRedaction).toEqual({
|
||||
production: true,
|
||||
manual: true,
|
||||
});
|
||||
expect(data.redactionInfo).toMatchObject({ isRedacted: true });
|
||||
for (const item of data.resultData.runData['Execute Workflow'][0].data!.main[0] ?? []) {
|
||||
expect(item.json).toEqual({});
|
||||
}
|
||||
});
|
||||
|
|
@ -835,6 +835,19 @@ describe('ExecuteContext', () => {
|
|||
});
|
||||
});
|
||||
|
||||
it('redacts console messages when a sub-workflow reported the channel', () => {
|
||||
const context = makeContext('manual', {
|
||||
executionData: {
|
||||
runtimeData: {
|
||||
redaction: { version: 2, production: false, manual: false },
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
},
|
||||
},
|
||||
} as unknown as IRunExecutionData);
|
||||
|
||||
expect(context.isConsoleOutputRedacted()).toBe(true);
|
||||
});
|
||||
|
||||
it('fails closed when resolving the policy throws', () => {
|
||||
const throwingRunData = {
|
||||
get executionData(): never {
|
||||
|
|
|
|||
|
|
@ -597,5 +597,69 @@ export const describeCommonTests = (
|
|||
expect(additionalData.dynamicCredentialsResolvedUserId).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('sub-workflow redaction reporting', () => {
|
||||
let executionContext: IExecutionContext;
|
||||
const originalExecutionData = runExecutionData.executionData;
|
||||
|
||||
beforeEach(() => {
|
||||
executionContext = { version: 1, establishedAt: 123, source: 'manual' };
|
||||
runExecutionData.executionData = mock<NonNullable<IRunExecutionData['executionData']>>({
|
||||
runtimeData: executionContext,
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
runExecutionData.executionData = originalExecutionData;
|
||||
});
|
||||
|
||||
it('escalates the execution context with the channels the sub-workflow reports', async () => {
|
||||
additionalData.executeWorkflow.mockResolvedValue({
|
||||
...executeWorkflowData,
|
||||
subWorkflowRedaction: { production: true, manual: false },
|
||||
});
|
||||
|
||||
await context.executeWorkflow(workflowInfo);
|
||||
|
||||
expect(executionContext.subWorkflowRedaction).toEqual({ production: true, manual: false });
|
||||
// The start snapshot keeps its meaning.
|
||||
expect(executionContext.redaction).toBeUndefined();
|
||||
});
|
||||
|
||||
it('combines the report with earlier reports and leaves the earlier object unchanged', async () => {
|
||||
const earlier = { production: true, manual: false };
|
||||
executionContext.subWorkflowRedaction = earlier;
|
||||
additionalData.executeWorkflow.mockResolvedValue({
|
||||
...executeWorkflowData,
|
||||
subWorkflowRedaction: { production: false, manual: true },
|
||||
});
|
||||
|
||||
await context.executeWorkflow(workflowInfo);
|
||||
|
||||
expect(executionContext.subWorkflowRedaction).toEqual({ production: true, manual: true });
|
||||
expect(earlier).toEqual({ production: true, manual: false });
|
||||
});
|
||||
|
||||
it('does not escalate when the sub-workflow reports nothing', async () => {
|
||||
additionalData.executeWorkflow.mockResolvedValue({ ...executeWorkflowData });
|
||||
|
||||
await context.executeWorkflow(workflowInfo);
|
||||
|
||||
expect(executionContext).not.toHaveProperty('subWorkflowRedaction');
|
||||
});
|
||||
|
||||
it('escalates from the channels attached to a failed sub-workflow error and rethrows', async () => {
|
||||
const error = Object.assign(new Error('sub-workflow failed'), {
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
});
|
||||
additionalData.executeWorkflow.mockRejectedValue(error);
|
||||
|
||||
await expect(context.executeWorkflow(workflowInfo)).rejects.toThrow('sub-workflow failed');
|
||||
|
||||
expect(executionContext.subWorkflowRedaction).toEqual({ production: true, manual: true });
|
||||
// The marker is transport-only and must not persist into the node's taskData.error.
|
||||
expect('subWorkflowRedaction' in error).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
};
|
||||
|
|
|
|||
|
|
@ -42,6 +42,8 @@ import {
|
|||
createEnvProviderState,
|
||||
applyDynamicCredentialsUsage,
|
||||
takeAttachedDynamicCredentialsUsage,
|
||||
escalateSubWorkflowRedaction,
|
||||
takeAttachedSubWorkflowRedaction,
|
||||
shouldRedactConsoleOutput,
|
||||
CONSOLE_OUTPUT_REDACTED_MESSAGE,
|
||||
} from 'n8n-workflow';
|
||||
|
|
@ -196,6 +198,11 @@ export class BaseExecuteContext extends NodeExecutionContext {
|
|||
// still inherits the flags.
|
||||
const usage = takeAttachedDynamicCredentialsUsage(error);
|
||||
if (usage) applyDynamicCredentialsUsage(this.additionalData, usage);
|
||||
// A failed sub-workflow may have produced output before it failed.
|
||||
escalateSubWorkflowRedaction(
|
||||
this.runExecutionData.executionData?.runtimeData,
|
||||
takeAttachedSubWorkflowRedaction(error),
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
|
||||
|
|
@ -203,6 +210,12 @@ export class BaseExecuteContext extends NodeExecutionContext {
|
|||
// usage onto this parent node so its task inherits the flag and redaction covers the
|
||||
// embedded output.
|
||||
applyDynamicCredentialsUsage(this.additionalData, result);
|
||||
// This execution keeps a copy of the sub-workflow's output, so it redacts at least
|
||||
// the channels the sub-workflow redacts.
|
||||
escalateSubWorkflowRedaction(
|
||||
this.runExecutionData.executionData?.runtimeData,
|
||||
result.subWorkflowRedaction,
|
||||
);
|
||||
|
||||
// If a sub-workflow execution goes into the waiting state
|
||||
if (result.waitTill) {
|
||||
|
|
@ -380,10 +393,12 @@ export class BaseExecuteContext extends NodeExecutionContext {
|
|||
*/
|
||||
isConsoleOutputRedacted(): boolean {
|
||||
try {
|
||||
const runtimeData = this.runExecutionData.executionData?.runtimeData;
|
||||
return shouldRedactConsoleOutput(
|
||||
this.runExecutionData.executionData?.runtimeData?.redaction,
|
||||
runtimeData?.redaction,
|
||||
this.workflow.settings,
|
||||
this.mode,
|
||||
runtimeData?.subWorkflowRedaction,
|
||||
);
|
||||
} catch {
|
||||
return true;
|
||||
|
|
|
|||
|
|
@ -233,6 +233,16 @@ const ExecutionContextSchemaV1 = z.object({
|
|||
*/
|
||||
redaction: RedactionSettingSchema.optional(),
|
||||
|
||||
/**
|
||||
* Redaction channels that the sub-workflows of this execution reported to it.
|
||||
* A sub-workflow reports the channels it redacts (and what its own
|
||||
* sub-workflows reported), because the caller keeps a copy of its output.
|
||||
* Kept apart from `redaction`, which stays the snapshot taken at start, so
|
||||
* the read path can tell why an execution is redacted. Readers combine the
|
||||
* two fields, strictest per channel. A channel is never lowered.
|
||||
*/
|
||||
subWorkflowRedaction: z.object({ production: z.boolean(), manual: z.boolean() }).optional(),
|
||||
|
||||
/**
|
||||
* The n8n user the execution ran as. Stamped at context establishment by
|
||||
* deriving the user from the identity carrier with the same identifier that
|
||||
|
|
@ -301,6 +311,9 @@ export function toErrorWorkflowContext(
|
|||
parentExecutionId: context.parentExecutionId,
|
||||
}),
|
||||
...(context.redaction !== undefined && { redaction: context.redaction }),
|
||||
...(context.subWorkflowRedaction !== undefined && {
|
||||
subWorkflowRedaction: context.subWorkflowRedaction,
|
||||
}),
|
||||
...(context.executedByUserId !== undefined && {
|
||||
executedByUserId: context.executedByUserId,
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -39,6 +39,7 @@ import type {
|
|||
WorkflowExecuteModeValues as WorkflowExecuteMode,
|
||||
} from './execution-context';
|
||||
import type { ExecutionStatus } from './execution-status';
|
||||
import type { RedactionChannels } from './redaction-channels';
|
||||
import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
|
||||
import type { Result } from '@n8n/utils/result';
|
||||
import type { Workflow } from './workflow';
|
||||
|
|
@ -2406,6 +2407,8 @@ export interface ExecuteWorkflowData extends DynamicCredentialsUsage {
|
|||
/** Terminal node output: items from every run concatenated per output branch, unless the caller sets `returnLastRunOnly`. */
|
||||
data: Array<INodeExecutionData[] | null>;
|
||||
waitTill?: Date | null;
|
||||
/** Redaction channels the sub-execution reports to its caller, so the caller's execution redacts the copy of its output. Absent when no channel is redacted. */
|
||||
subWorkflowRedaction?: RedactionChannels;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import type { IRedactionSetting } from './execution-context';
|
||||
import type { IExecutionContext, IRedactionSetting } from './execution-context';
|
||||
import type { IWorkflowSettings, WorkflowExecuteMode, WorkflowSettings } from './interfaces';
|
||||
import type { IRunExecutionData } from './run-execution-data/run-execution-data';
|
||||
|
||||
/**
|
||||
* Redaction expressed as two independent channels:
|
||||
|
|
@ -58,6 +59,108 @@ export function redactionSettingToChannels(
|
|||
return policyToChannels(redaction.policy);
|
||||
}
|
||||
|
||||
const NO_REDACTION: RedactionChannels = { production: false, manual: false };
|
||||
|
||||
function redactsAnyChannel(channels: RedactionChannels): boolean {
|
||||
return channels.production || channels.manual;
|
||||
}
|
||||
|
||||
/** Combines two sets of channels, strictest per channel. */
|
||||
export function mergeRedactionChannels(
|
||||
a: RedactionChannels,
|
||||
b: RedactionChannels | undefined,
|
||||
): RedactionChannels {
|
||||
if (!b) return a;
|
||||
return { production: a.production || b.production, manual: a.manual || b.manual };
|
||||
}
|
||||
|
||||
/**
|
||||
* The channels an execution redacts: the snapshot taken at start combined with
|
||||
* the channels its sub-workflows reported (see `subWorkflowRedaction`).
|
||||
*/
|
||||
export function effectiveRedactionChannels(
|
||||
context: Pick<IExecutionContext, 'redaction' | 'subWorkflowRedaction'> | undefined,
|
||||
): RedactionChannels {
|
||||
return mergeRedactionChannels(
|
||||
redactionSettingToChannels(context?.redaction),
|
||||
context?.subWorkflowRedaction,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* The channels a sub-execution reports to its caller. The caller keeps a copy
|
||||
* of the sub-execution's output, so it must redact at least what the
|
||||
* sub-execution redacts.
|
||||
*
|
||||
* What the sub-execution received from its own sub-workflows always passes up.
|
||||
* `includeOwn` adds the sub-execution's own effective channels. Returns
|
||||
* `undefined` when no channel is redacted.
|
||||
*/
|
||||
export function summarizeSubWorkflowRedaction(
|
||||
executionData: IRunExecutionData | undefined,
|
||||
options: { includeOwn: boolean },
|
||||
): RedactionChannels | undefined {
|
||||
const context = executionData?.executionData?.runtimeData;
|
||||
const channels = options.includeOwn
|
||||
? effectiveRedactionChannels(context)
|
||||
: mergeRedactionChannels(NO_REDACTION, context?.subWorkflowRedaction);
|
||||
return redactsAnyChannel(channels) ? channels : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Raises the caller's `subWorkflowRedaction` with the channels a sub-execution
|
||||
* reported. A channel is never lowered.
|
||||
*
|
||||
* Assigns a new object and never mutates the existing one: a sub-execution's
|
||||
* context spreads its caller's context, so the nested object is shared when
|
||||
* both run in the same process.
|
||||
*/
|
||||
export function escalateSubWorkflowRedaction(
|
||||
runtimeData: IExecutionContext | undefined,
|
||||
channels: RedactionChannels | undefined,
|
||||
): void {
|
||||
if (!runtimeData || !channels || !redactsAnyChannel(channels)) return;
|
||||
runtimeData.subWorkflowRedaction = mergeRedactionChannels(
|
||||
runtimeData.subWorkflowRedaction ?? NO_REDACTION,
|
||||
channels,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Rides the reported channels on an error that crosses the sub-execution
|
||||
* boundary, so a caller that continues on fail still escalates. Read back via
|
||||
* {@link takeAttachedSubWorkflowRedaction}.
|
||||
*/
|
||||
export function attachSubWorkflowRedaction(
|
||||
error: Error,
|
||||
channels: RedactionChannels | undefined,
|
||||
): Error {
|
||||
if (channels && redactsAnyChannel(channels)) {
|
||||
Object.assign(error, { subWorkflowRedaction: channels });
|
||||
}
|
||||
return error;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads back and strips the channels attached via
|
||||
* {@link attachSubWorkflowRedaction}, and validates their shape. The marker is
|
||||
* transport-only: stripping it keeps it out of the persisted `taskData.error`.
|
||||
*/
|
||||
export function takeAttachedSubWorkflowRedaction(error: unknown): RedactionChannels | undefined {
|
||||
if (typeof error !== 'object' || error === null || !('subWorkflowRedaction' in error)) {
|
||||
return undefined;
|
||||
}
|
||||
const attached = error.subWorkflowRedaction;
|
||||
Reflect.deleteProperty(error, 'subWorkflowRedaction');
|
||||
if (typeof attached !== 'object' || attached === null) return undefined;
|
||||
|
||||
const channels: RedactionChannels = {
|
||||
production: 'production' in attached && attached.production === true,
|
||||
manual: 'manual' in attached && attached.manual === true,
|
||||
};
|
||||
return redactsAnyChannel(channels) ? channels : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Emitted in place of user console output when the run's resolved redaction
|
||||
* policy redacts the channel it executes on. Console output is ephemeral
|
||||
|
|
@ -73,18 +176,28 @@ export const CONSOLE_OUTPUT_REDACTED_MESSAGE =
|
|||
* fallback; manual runs consult the manual channel, every other mode the
|
||||
* production channel. For a V2 snapshot, production is clamped up when
|
||||
* manual is set, matching the fail-strict clamp in {@link channelsToPolicy}.
|
||||
* The channels reported by sub-workflows (`subWorkflowRedaction`) are
|
||||
* combined in, strictest per channel, with the same clamp.
|
||||
*/
|
||||
export function shouldRedactConsoleOutput(
|
||||
redaction: IRedactionSetting | undefined,
|
||||
workflowSettings: IWorkflowSettings | undefined,
|
||||
mode: WorkflowExecuteMode,
|
||||
subWorkflowRedaction?: RedactionChannels,
|
||||
): boolean {
|
||||
const channels: RedactionChannels =
|
||||
const base: RedactionChannels =
|
||||
redaction === undefined
|
||||
? policyToChannels(workflowSettings?.redactionPolicy ?? 'none')
|
||||
: redaction.version === 2
|
||||
? { production: redaction.production || redaction.manual, manual: redaction.manual }
|
||||
: policyToChannels(redaction.policy);
|
||||
const channels = mergeRedactionChannels(
|
||||
base,
|
||||
subWorkflowRedaction && {
|
||||
production: subWorkflowRedaction.production || subWorkflowRedaction.manual,
|
||||
manual: subWorkflowRedaction.manual,
|
||||
},
|
||||
);
|
||||
|
||||
return mode === 'manual' ? channels.manual : channels.production;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -52,6 +52,14 @@ describe('toExecutionContext — redaction snapshot', () => {
|
|||
).toThrow();
|
||||
});
|
||||
|
||||
it('parses the channels reported by sub-workflows', () => {
|
||||
const parsed = toExecutionContext({
|
||||
...baseContext,
|
||||
subWorkflowRedaction: { production: true, manual: false },
|
||||
});
|
||||
expect(parsed.subWorkflowRedaction).toEqual({ production: true, manual: false });
|
||||
});
|
||||
|
||||
it('parses a context without a redaction snapshot', () => {
|
||||
const parsed = toExecutionContext({ ...baseContext });
|
||||
|
||||
|
|
@ -87,6 +95,7 @@ describe('toErrorWorkflowContext', () => {
|
|||
credentials: 'encrypted-identity-carrier',
|
||||
secureArtifacts: 'encrypted-secure-artifacts',
|
||||
redaction: { version: 2, production: true, manual: false },
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
executedByUserId: 'user-1',
|
||||
usesDynamicCredentials: true,
|
||||
};
|
||||
|
|
@ -106,6 +115,7 @@ describe('toErrorWorkflowContext', () => {
|
|||
triggerNode: { name: 'Webhook', type: 'n8n-nodes-base.webhook' },
|
||||
parentExecutionId: 'parent-1',
|
||||
redaction: { version: 2, production: true, manual: false },
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
executedByUserId: 'user-1',
|
||||
usesDynamicCredentials: true,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,10 +1,28 @@
|
|||
import type { IExecutionContext } from '../src/execution-context';
|
||||
import type { WorkflowSettings } from '../src/interfaces';
|
||||
import {
|
||||
attachSubWorkflowRedaction,
|
||||
channelsToPolicy,
|
||||
effectiveRedactionChannels,
|
||||
escalateSubWorkflowRedaction,
|
||||
mergeRedactionChannels,
|
||||
policyToChannels,
|
||||
redactionSettingToChannels,
|
||||
shouldRedactConsoleOutput,
|
||||
summarizeSubWorkflowRedaction,
|
||||
takeAttachedSubWorkflowRedaction,
|
||||
} from '../src/redaction-channels';
|
||||
import type { IRunExecutionData } from '../src/run-execution-data/run-execution-data';
|
||||
|
||||
const context = (fields: Partial<IExecutionContext> = {}): IExecutionContext => ({
|
||||
version: 1,
|
||||
establishedAt: 0,
|
||||
source: 'integrated',
|
||||
...fields,
|
||||
});
|
||||
|
||||
const runWith = (runtimeData: IExecutionContext | undefined): IRunExecutionData =>
|
||||
({ resultData: { runData: {} }, executionData: { runtimeData } }) as unknown as IRunExecutionData;
|
||||
|
||||
describe('redaction-channels', () => {
|
||||
describe('policyToChannels', () => {
|
||||
|
|
@ -92,6 +110,35 @@ describe('redaction-channels', () => {
|
|||
expect(shouldRedactConsoleOutput({ version: 1, policy }, undefined, mode)).toBe(expected);
|
||||
});
|
||||
|
||||
it.each([
|
||||
[{ production: true, manual: false }, 'trigger', true],
|
||||
[{ production: true, manual: false }, 'manual', false],
|
||||
[{ production: true, manual: true }, 'manual', true],
|
||||
// Manual without production is clamped up, like a V2 snapshot.
|
||||
[{ production: false, manual: true }, 'trigger', true],
|
||||
] as const)(
|
||||
'combines the sub-workflow channels %j in mode %s → %s',
|
||||
(subWorkflowRedaction, mode, expected) => {
|
||||
expect(
|
||||
shouldRedactConsoleOutput(
|
||||
{ version: 2, production: false, manual: false },
|
||||
undefined,
|
||||
mode,
|
||||
subWorkflowRedaction,
|
||||
),
|
||||
).toBe(expected);
|
||||
},
|
||||
);
|
||||
|
||||
it('combines the sub-workflow channels with the workflow-setting fallback', () => {
|
||||
expect(
|
||||
shouldRedactConsoleOutput(undefined, { redactionPolicy: 'none' }, 'trigger', {
|
||||
production: true,
|
||||
manual: false,
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it('falls back to the workflow setting when the snapshot is absent', () => {
|
||||
expect(shouldRedactConsoleOutput(undefined, { redactionPolicy: 'all' }, 'trigger')).toBe(
|
||||
true,
|
||||
|
|
@ -104,4 +151,174 @@ describe('redaction-channels', () => {
|
|||
expect(shouldRedactConsoleOutput(undefined, undefined, 'trigger')).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('mergeRedactionChannels', () => {
|
||||
it('combines the channels, strictest per channel', () => {
|
||||
expect(
|
||||
mergeRedactionChannels(
|
||||
{ production: true, manual: false },
|
||||
{ production: false, manual: true },
|
||||
),
|
||||
).toEqual({ production: true, manual: true });
|
||||
});
|
||||
|
||||
it('returns the first set when the second is absent', () => {
|
||||
const a = { production: true, manual: false };
|
||||
expect(mergeRedactionChannels(a, undefined)).toBe(a);
|
||||
});
|
||||
});
|
||||
|
||||
describe('effectiveRedactionChannels', () => {
|
||||
it('returns no redaction for an absent context', () => {
|
||||
expect(effectiveRedactionChannels(undefined)).toEqual({ production: false, manual: false });
|
||||
});
|
||||
|
||||
it('combines the start snapshot with the sub-workflow channels', () => {
|
||||
expect(
|
||||
effectiveRedactionChannels({
|
||||
redaction: { version: 2, production: true, manual: false },
|
||||
subWorkflowRedaction: { production: false, manual: true },
|
||||
}),
|
||||
).toEqual({ production: true, manual: true });
|
||||
});
|
||||
|
||||
it('reads a V1 snapshot', () => {
|
||||
expect(
|
||||
effectiveRedactionChannels({
|
||||
redaction: { version: 1, policy: 'non-manual' },
|
||||
subWorkflowRedaction: { production: false, manual: false },
|
||||
}),
|
||||
).toEqual({ production: true, manual: false });
|
||||
});
|
||||
});
|
||||
|
||||
describe('summarizeSubWorkflowRedaction', () => {
|
||||
it('reports the own snapshot when includeOwn is true', () => {
|
||||
const run = runWith(context({ redaction: { version: 2, production: true, manual: false } }));
|
||||
expect(summarizeSubWorkflowRedaction(run, { includeOwn: true })).toEqual({
|
||||
production: true,
|
||||
manual: false,
|
||||
});
|
||||
});
|
||||
|
||||
it('reports the own snapshot combined with the received channels', () => {
|
||||
const run = runWith(
|
||||
context({
|
||||
redaction: { version: 2, production: true, manual: false },
|
||||
subWorkflowRedaction: { production: true, manual: true },
|
||||
}),
|
||||
);
|
||||
expect(summarizeSubWorkflowRedaction(run, { includeOwn: true })).toEqual({
|
||||
production: true,
|
||||
manual: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('reports only the received channels when includeOwn is false', () => {
|
||||
const run = runWith(
|
||||
context({
|
||||
redaction: { version: 2, production: true, manual: true },
|
||||
subWorkflowRedaction: { production: true, manual: false },
|
||||
}),
|
||||
);
|
||||
expect(summarizeSubWorkflowRedaction(run, { includeOwn: false })).toEqual({
|
||||
production: true,
|
||||
manual: false,
|
||||
});
|
||||
});
|
||||
|
||||
it('returns undefined when no channel is redacted', () => {
|
||||
const run = runWith(context({ redaction: { version: 2, production: false, manual: false } }));
|
||||
expect(summarizeSubWorkflowRedaction(run, { includeOwn: true })).toBeUndefined();
|
||||
expect(summarizeSubWorkflowRedaction(run, { includeOwn: false })).toBeUndefined();
|
||||
});
|
||||
|
||||
it('returns undefined when the run has no context', () => {
|
||||
expect(
|
||||
summarizeSubWorkflowRedaction(runWith(undefined), { includeOwn: true }),
|
||||
).toBeUndefined();
|
||||
expect(summarizeSubWorkflowRedaction(undefined, { includeOwn: true })).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('escalateSubWorkflowRedaction', () => {
|
||||
it('sets the channels on a context that has none', () => {
|
||||
const runtimeData = context();
|
||||
escalateSubWorkflowRedaction(runtimeData, { production: true, manual: false });
|
||||
expect(runtimeData.subWorkflowRedaction).toEqual({ production: true, manual: false });
|
||||
});
|
||||
|
||||
it('never lowers a channel', () => {
|
||||
const runtimeData = context({ subWorkflowRedaction: { production: true, manual: true } });
|
||||
escalateSubWorkflowRedaction(runtimeData, { production: true, manual: false });
|
||||
expect(runtimeData.subWorkflowRedaction).toEqual({ production: true, manual: true });
|
||||
});
|
||||
|
||||
it('assigns a new object and leaves the existing one unchanged', () => {
|
||||
const shared = { production: true, manual: false };
|
||||
const runtimeData = context({ subWorkflowRedaction: shared });
|
||||
escalateSubWorkflowRedaction(runtimeData, { production: false, manual: true });
|
||||
expect(runtimeData.subWorkflowRedaction).toEqual({ production: true, manual: true });
|
||||
expect(runtimeData.subWorkflowRedaction).not.toBe(shared);
|
||||
expect(shared).toEqual({ production: true, manual: false });
|
||||
});
|
||||
|
||||
it('does nothing without channels, or when no channel is redacted', () => {
|
||||
const runtimeData = context();
|
||||
escalateSubWorkflowRedaction(runtimeData, undefined);
|
||||
escalateSubWorkflowRedaction(runtimeData, { production: false, manual: false });
|
||||
expect(runtimeData).not.toHaveProperty('subWorkflowRedaction');
|
||||
});
|
||||
|
||||
it('does nothing without a context', () => {
|
||||
expect(() =>
|
||||
escalateSubWorkflowRedaction(undefined, { production: true, manual: true }),
|
||||
).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe('attachSubWorkflowRedaction / takeAttachedSubWorkflowRedaction', () => {
|
||||
it('round-trips the channels and strips the marker', () => {
|
||||
const error = attachSubWorkflowRedaction(new Error('boom'), {
|
||||
production: true,
|
||||
manual: false,
|
||||
});
|
||||
expect(takeAttachedSubWorkflowRedaction(error)).toEqual({ production: true, manual: false });
|
||||
expect(error).not.toHaveProperty('subWorkflowRedaction');
|
||||
expect(takeAttachedSubWorkflowRedaction(error)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('attaches nothing when no channel is redacted', () => {
|
||||
const error = attachSubWorkflowRedaction(new Error('boom'), {
|
||||
production: false,
|
||||
manual: false,
|
||||
});
|
||||
expect(error).not.toHaveProperty('subWorkflowRedaction');
|
||||
expect(attachSubWorkflowRedaction(new Error('boom'), undefined)).not.toHaveProperty(
|
||||
'subWorkflowRedaction',
|
||||
);
|
||||
});
|
||||
|
||||
it('returns the same error instance', () => {
|
||||
const error = new Error('boom');
|
||||
expect(attachSubWorkflowRedaction(error, { production: true, manual: true })).toBe(error);
|
||||
});
|
||||
|
||||
it.each([null, undefined, 'error', 42])('returns undefined for a non-object %p', (error) => {
|
||||
expect(takeAttachedSubWorkflowRedaction(error)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('ignores and strips a malformed marker', () => {
|
||||
const error = Object.assign(new Error('boom'), { subWorkflowRedaction: 'all' });
|
||||
expect(takeAttachedSubWorkflowRedaction(error)).toBeUndefined();
|
||||
expect(error).not.toHaveProperty('subWorkflowRedaction');
|
||||
});
|
||||
|
||||
it('reads only boolean true channel values', () => {
|
||||
const error = Object.assign(new Error('boom'), {
|
||||
subWorkflowRedaction: { production: 'yes', manual: true },
|
||||
});
|
||||
expect(takeAttachedSubWorkflowRedaction(error)).toEqual({ production: false, manual: true });
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user