diff --git a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts index a632dd74652..e61b4f9dd68 100644 --- a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts @@ -202,6 +202,13 @@ export class WorkflowPublicationOutboxRepository extends Repository { const outboxTableName = this.getTableName('workflow_publication_outbox'); @@ -211,6 +218,9 @@ export class WorkflowPublicationOutboxRepository extends Repository (y IS NULL) OR x <> y` is the portable spelling of // `activeVersionId IS DISTINCT FROM publishedVersionId`, which sqlite // lacks; both-null (never published, no mapping) compares as equal. + // + // The failed-record match relies on plain `=`: an unpublished workflow has a + // null `activeVersionId`, so nothing matches and its skew stays detectable. const rows: Array<{ workflowId: string }> = await this.query( `SELECT w."id" AS "workflowId" FROM ${workflowTableName} w @@ -223,6 +233,16 @@ export class WorkflowPublicationOutboxRepository extends Repository { const workflowRepository = mock(); const workflowsConfig = mock({ useWorkflowPublicationService: false }); + // Shared by every construction below; clears by default, like an instance with + // no policy backend. + const policyEnforcementService = mock(); + const ownershipService = mock(); + beforeEach(() => { vi.clearAllMocks(); + // `clearAllMocks` keeps implementations, so restore the clearing defaults. + ownershipService.getWorkflowProjectCached.mockResolvedValue(mock({ id: 'project-1' })); + policyEnforcementService.hasChecksFor.mockReturnValue(true); + policyEnforcementService.enforceWorkflowPublish.mockResolvedValue(mock()); activeWorkflowManager = new ActiveWorkflowManager( mockLogger(), mock(), @@ -79,6 +90,8 @@ describe('ActiveWorkflowManager', () => { mock(), mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + policyEnforcementService, + ownershipService, ); }); @@ -181,6 +194,223 @@ describe('ActiveWorkflowManager', () => { }); }); + describe('policy enforcement', () => { + // `chunk` coerces an unset `activationBatchSize` to 0 and yields no batches, so + // the startup loop would silently never run. + const policyWorkflowsConfig = mock({ + useWorkflowPublicationService: false, + activationBatchSize: 1, + }); + const activationErrorsService = mock(); + const activeWorkflowTriggers = mock(); + const errorReporter = mock(); + + const VERSION_NODES = [{ id: 'node-1', name: 'Active trigger' } as INode]; + const DRAFT_NODES = [{ id: 'node-2', name: 'Draft node' } as INode]; + + const makeManager = () => + new ActiveWorkflowManager( + mockLogger(), + errorReporter, + activeWorkflowTriggers, + mock(), + nodeTypes, + mock(), + workflowRepository, + activationErrorsService, + mock(), + mock(), + instanceSettings, + mock(), + policyWorkflowsConfig, + mock(), + mock(), + mock(), + mock(), // scheduleTriggerJobRegistrar + mock(), // pollTriggerJobRegistrar + policyEnforcementService, + ownershipService, + ); + + const makeWorkflow = (overrides: Partial = {}) => + mock({ + id: 'wf-1', + name: 'My workflow', + active: true, + isArchived: false, + activeVersionId: 'v1', + // Differs from the published version on purpose: only the latter runs. + nodes: DRAFT_NODES, + activeVersion: mock({ + versionId: 'v1', + nodes: VERSION_NODES, + connections: {}, + }), + ...overrides, + }); + + beforeEach(() => { + Object.assign(instanceSettings, { isLeader: true, isFollower: false }); + activeWorkflowManager = makeManager(); + }); + + test('enforces with the published version nodes, not the draft', async () => { + workflowRepository.findById.mockResolvedValue(makeWorkflow()); + + // Registration fails here (no real node types); the check runs before it. + await activeWorkflowManager.add('wf-1', 'activate').catch(() => {}); + + expect(policyEnforcementService.enforceWorkflowPublish).toHaveBeenCalledExactlyOnceWith({ + workflow: { id: 'wf-1', name: 'My workflow', nodes: VERSION_NODES }, + projectId: 'project-1', + }); + }); + + test('registers nothing and records an activation error when policy blocks', async () => { + workflowRepository.findById.mockResolvedValue(makeWorkflow()); + const addWebhooksSpy = vi.spyOn(activeWorkflowManager, 'addWebhooks'); + const addNonWebhookTriggersSpy = vi.spyOn(activeWorkflowManager, 'addNonWebhookTriggers'); + policyEnforcementService.enforceWorkflowPublish.mockRejectedValue( + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]), + ); + + await expect(activeWorkflowManager.add('wf-1', 'activate')).rejects.toBeInstanceOf( + PolicyViolationError, + ); + + expect(addWebhooksSpy).not.toHaveBeenCalled(); + expect(addNonWebhookTriggersSpy).not.toHaveBeenCalled(); + expect(activationErrorsService.register).toHaveBeenCalledWith('wf-1', 'Blocked by policy'); + }); + + test('does not enforce for a workflow that is no longer active', async () => { + workflowRepository.findById.mockResolvedValue( + makeWorkflow({ active: false, activeVersionId: null, activeVersion: null }), + ); + + await activeWorkflowManager.add('wf-1', 'init'); + + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + // An unevaluated project rule is not a passed one, so the lookup is unguarded. + test('propagates a failed ownership lookup instead of policing a null scope', async () => { + workflowRepository.findById.mockResolvedValue(makeWorkflow()); + ownershipService.getWorkflowProjectCached.mockRejectedValue(new Error('no owner row')); + + await expect(activeWorkflowManager.add('wf-1', 'activate')).rejects.toThrow('no owner row'); + + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + // A feature that is merely absent must not cost a lookup on every activation. + test('does not resolve ownership when no check is registered', async () => { + policyEnforcementService.hasChecksFor.mockReturnValue(false); + workflowRepository.findById.mockResolvedValue(makeWorkflow()); + + await activeWorkflowManager.add('wf-1', 'activate').catch(() => {}); + + expect(ownershipService.getWorkflowProjectCached).not.toHaveBeenCalled(); + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + test('does not enforce for an archived workflow', async () => { + workflowRepository.findById.mockResolvedValue(makeWorkflow({ isArchived: true })); + + await activeWorkflowManager.add('wf-1', 'activate'); + + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + // The forward is unconditional on multi-main, not leader-specific; the leader + // enforces when it handles the pubsub command with `shouldPublish: false`. + test('does not enforce when a multi-main instance forwards the activation', async () => { + Object.assign(instanceSettings, { isMultiMain: true }); + workflowRepository.findById.mockResolvedValue(makeWorkflow()); + + try { + await activeWorkflowManager.add('wf-1', 'activate'); + + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + } finally { + Object.assign(instanceSettings, { isMultiMain: false }); + } + }); + + // Already queued for an unrelated transient failure: if policy then blocks it, + // the retry must drop it rather than reschedule forever. + test('drops a queued retry when policy blocks on the retry attempt', async () => { + vi.useFakeTimers(); + try { + const dbWorkflow = makeWorkflow(); + workflowRepository.findById.mockResolvedValue(dbWorkflow); + const manager = activeWorkflowManager as unknown as { + addQueuedWorkflowActivation: ( + mode: WorkflowActivateMode, + workflow: WorkflowEntity, + ) => void; + queuedActivations: Record; + }; + + manager.addQueuedWorkflowActivation('update', dbWorkflow); + expect(manager.queuedActivations['wf-1']).toBeDefined(); + + policyEnforcementService.enforceWorkflowPublish.mockRejectedValue( + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]), + ); + + await vi.runOnlyPendingTimersAsync(); + + expect(manager.queuedActivations['wf-1']).toBeUndefined(); + // Nor a fault report: the refusal is expected and permanent. + expect(errorReporter.error).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + // A refusal, not a failure: the owner's error automation must not fire. + test('does not run the error workflow when policy blocks at startup', async () => { + workflowRepository.getAllActiveIds.mockResolvedValue(['wf-1']); + workflowRepository.findById.mockResolvedValue(makeWorkflow()); + policyEnforcementService.enforceWorkflowPublish.mockRejectedValue( + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]), + ); + const errorWorkflowSpy = vi.spyOn(activeWorkflowManager, 'executeErrorWorkflow'); + + await activeWorkflowManager.addActiveWorkflows('init'); + + expect(errorWorkflowSpy).not.toHaveBeenCalled(); + // Nor a fault report, or every restart alerts on an expected refusal. + expect(errorReporter.error).not.toHaveBeenCalled(); + }); + + // A block is permanent, so the indefinite retry must not pick it up. + test('does not queue a startup retry when policy blocks', async () => { + workflowRepository.getAllActiveIds.mockResolvedValue(['wf-1']); + workflowRepository.findById.mockResolvedValue(makeWorkflow()); + policyEnforcementService.enforceWorkflowPublish.mockRejectedValue( + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]), + ); + const queueSpy = vi.spyOn( + activeWorkflowManager as unknown as { addQueuedWorkflowActivation: () => void }, + 'addQueuedWorkflowActivation', + ); + + await activeWorkflowManager.addActiveWorkflows('init'); + + expect(queueSpy).not.toHaveBeenCalled(); + }); + }); + describe('handleAddWebhooksAndNonWebhookTriggers', () => { const push = mock(); const publisher = mock(); @@ -205,6 +435,33 @@ describe('ActiveWorkflowManager', () => { mock(), mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + policyEnforcementService, + ownershipService, + ); + }); + + // The leader's generic failure path clears `activeVersionId`. A refusal must not + // unpublish a workflow — that decision is not this path's to make. + test('does not unpublish the workflow when policy blocks the forwarded activation', async () => { + const violation = new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]); + vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(violation); + const clearWebhooksSpy = vi.spyOn(activeWorkflowManager, 'clearWebhooks'); + const removeTriggersSpy = vi.spyOn(activeWorkflowManager, 'removeNonWebhookTriggers'); + + await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({ + workflowId: 'wf-1', + activeVersionId: 'v1', + activationMode: 'init', + }); + + expect(workflowRepository.update).not.toHaveBeenCalled(); + expect(clearWebhooksSpy).not.toHaveBeenCalled(); + expect(removeTriggersSpy).not.toHaveBeenCalled(); + // The user is still told why it did not start. + expect(push.broadcast).toHaveBeenCalledWith( + expect.objectContaining({ type: 'workflowFailedToActivate' }), ); }); @@ -463,6 +720,8 @@ describe('ActiveWorkflowManager', () => { mock(), // eventBus mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + policyEnforcementService, + ownershipService, ); }); @@ -914,6 +1173,8 @@ describe('ActiveWorkflowManager', () => { mock(), scheduleTriggerJobRegistrar, pollTriggerJobRegistrar, + policyEnforcementService, + ownershipService, ); }); @@ -1100,6 +1361,8 @@ describe('ActiveWorkflowManager', () => { mock(), scheduleTriggerJobRegistrar, pollTriggerJobRegistrar, + policyEnforcementService, + ownershipService, ); beforeEach(() => vi.clearAllMocks()); @@ -1177,6 +1440,8 @@ describe('ActiveWorkflowManager', () => { mock(), scheduleTriggerJobRegistrar, mock(), // pollTriggerJobRegistrar + policyEnforcementService, + ownershipService, ); const makeWorkflow = () => { diff --git a/packages/cli/src/active-workflow-manager.ts b/packages/cli/src/active-workflow-manager.ts index 1d1b5dbe798..b01541c8e36 100644 --- a/packages/cli/src/active-workflow-manager.ts +++ b/packages/cli/src/active-workflow-manager.ts @@ -44,6 +44,8 @@ import { ActivationErrorsService } from '@/activation-errors.service'; import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; import { ExternalHooks } from '@/external-hooks'; import { NodeTypes } from '@/node-types'; +import { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; +import { isPolicyRefusal } from '@/policy/policy-violation.error'; import { Push } from '@/push'; import { Publisher } from '@/scaling/pubsub/publisher.service'; import { PubSubCommandMap } from '@/scaling/pubsub/pubsub.event-map'; @@ -51,6 +53,7 @@ import type { ScheduleTriggerCollectionSession } from '@/scheduling/schedule-tri import { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar'; import { ScheduleTriggerJobRegistrar } from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar'; import { ActiveWorkflowsService } from '@/services/active-workflows.service'; +import { OwnershipService } from '@/services/ownership.service'; import * as WebhookHelpers from '@/webhooks/webhook-helpers'; import { WebhookService } from '@/webhooks/webhook.service'; import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data'; @@ -89,6 +92,8 @@ export class ActiveWorkflowManager { private readonly eventBus: MessageEventBus, private readonly scheduleTriggerJobRegistrar: ScheduleTriggerJobRegistrar, private readonly pollTriggerJobRegistrar: PollTriggerJobRegistrar, + private readonly policyEnforcementService: PolicyEnforcementService, + private readonly ownershipService: OwnershipService, ) { this.logger = this.logger.scoped(['workflow-activation']); } @@ -446,6 +451,16 @@ export class ActiveWorkflowManager { }); } } catch (error) { + // An expected refusal, permanent until policy or workflow changes: no fault + // report, no error workflow and no retry, or every restart alerts. + if (isPolicyRefusal(error)) { + this.logger.warn(`Publication of ${formatWorkflow(dbWorkflow)} blocked by policy`, { + workflowId: dbWorkflow.id, + }); + + return; + } + this.errorReporter.error(error); this.logger.error( `Issue on initial workflow activation try of ${formatWorkflow(dbWorkflow)} (startup)`, @@ -583,6 +598,17 @@ export class ActiveWorkflowManager { dbWorkflow.nodes = nodes; dbWorkflow.connections = connections; + // Trigger and poller nodes run code at registration, so this gates startup + // and leadership change too, not just the activate button. + if (this.policyEnforcementService.hasChecksFor('workflowPublish')) { + const project = await this.ownershipService.getWorkflowProjectCached(dbWorkflow.id); + + await this.policyEnforcementService.enforceWorkflowPublish({ + workflow: { id: dbWorkflow.id, name: dbWorkflow.name, nodes }, + projectId: project.id, + }); + } + workflow = new Workflow({ id: dbWorkflow.id, name: dbWorkflow.name, @@ -741,44 +767,48 @@ export class ActiveWorkflowManager { return; } - const dbWorkflow = await this.workflowRepository.findById(workflowId); + // A policy refusal happens before anything is registered, so there is + // nothing to tear down — and unpublishing is not this path's call to make. + if (!isPolicyRefusal(error)) { + const dbWorkflow = await this.workflowRepository.findById(workflowId); - // Activation may have failed partway with triggers already registered, - // in memory and as durable jobs. Tear them down before the - // deactivation below so the active version is still resolvable, or - // they keep firing a workflow marked inactive. Each teardown is caught - // on its own so a webhook failure never skips the durable-job cleanup. - try { - await this.clearWebhooks(workflowId); - } catch (cleanupError) { - this.logger.error(`Failed to remove webhooks of workflow "${workflowId}"`, { - workflowId, - error: ensureError(cleanupError), - }); - } - - try { - await this.removeNonWebhookTriggers(workflowId); - } catch (cleanupError) { - this.logger.error(`Failed to remove triggers of workflow "${workflowId}"`, { - workflowId, - error: ensureError(cleanupError), - }); - } - - await this.workflowRepository.update(workflowId, { active: false, activeVersionId: null }); - - if (dbWorkflow && (activationMode === 'init' || activationMode === 'leadershipChange')) { - void this.eventBus.sendAuditEvent({ - eventName: 'n8n.audit.workflow.deactivated', - payload: { + // Activation may have failed partway with triggers already registered, + // in memory and as durable jobs. Tear them down before the + // deactivation below so the active version is still resolvable, or + // they keep firing a workflow marked inactive. Each teardown is caught + // on its own so a webhook failure never skips the durable-job cleanup. + try { + await this.clearWebhooks(workflowId); + } catch (cleanupError) { + this.logger.error(`Failed to remove webhooks of workflow "${workflowId}"`, { workflowId, - workflowName: dbWorkflow.name, - deactivatedVersionId: dbWorkflow.activeVersionId ?? null, - activationMode, - reason: error.name, - }, - }); + error: ensureError(cleanupError), + }); + } + + try { + await this.removeNonWebhookTriggers(workflowId); + } catch (cleanupError) { + this.logger.error(`Failed to remove triggers of workflow "${workflowId}"`, { + workflowId, + error: ensureError(cleanupError), + }); + } + + await this.workflowRepository.update(workflowId, { active: false, activeVersionId: null }); + + if (dbWorkflow && (activationMode === 'init' || activationMode === 'leadershipChange')) { + void this.eventBus.sendAuditEvent({ + eventName: 'n8n.audit.workflow.deactivated', + payload: { + workflowId, + workflowName: dbWorkflow.name, + deactivatedVersionId: dbWorkflow.activeVersionId ?? null, + activationMode, + reason: error.name, + }, + }); + } } this.push.broadcast({ @@ -833,6 +863,7 @@ export class ActiveWorkflowManager { ) { const workflowId = workflowData.id; const workflowName = workflowData.name; + const own: { activation?: QueuedActivation } = {}; const retryFunction = async () => { this.logger.info(`Try to activate workflow "${workflowName}" (${workflowId})`, { @@ -842,7 +873,23 @@ export class ActiveWorkflowManager { try { await this.add(workflowId, activationMode, workflowData, { shouldPublish: false }); } catch (error) { + // An expected refusal, permanent until policy or workflow changes: no fault + // report, and leave the queue rather than retrying forever. + if (isPolicyRefusal(error)) { + this.logger.warn(`Publication of workflow "${workflowId}" blocked by policy`, { + workflowId, + }); + + // Only our own entry: a newer failure may have replaced it since. + if (this.queuedActivations[workflowId] === own.activation) { + this.removeQueuedWorkflowActivation(workflowId); + } + + return; + } + this.errorReporter.error(error); + const queuedActivation = this.queuedActivations[workflowId]; if (!queuedActivation) { return; @@ -878,12 +925,13 @@ export class ActiveWorkflowManager { // multiple run in parallel this.removeQueuedWorkflowActivation(workflowId); - this.queuedActivations[workflowId] = { + own.activation = { activationMode, lastTimeout: WORKFLOW_REACTIVATE_INITIAL_TIMEOUT, timeout: setTimeout(retryFunction, WORKFLOW_REACTIVATE_INITIAL_TIMEOUT), workflowData, }; + this.queuedActivations[workflowId] = own.activation; } /** diff --git a/packages/cli/src/execution-lifecycle/__tests__/execute-error-workflow.test.ts b/packages/cli/src/execution-lifecycle/__tests__/execute-error-workflow.test.ts index 8cb6f1ab7d9..49d101b3197 100644 --- a/packages/cli/src/execution-lifecycle/__tests__/execute-error-workflow.test.ts +++ b/packages/cli/src/execution-lifecycle/__tests__/execute-error-workflow.test.ts @@ -7,6 +7,7 @@ import type { INode, IRun, IWorkflowBase } from 'n8n-workflow'; import { createRunExecutionData, NodeOperationError } from 'n8n-workflow'; import { mock } from 'vitest-mock-extended'; +import { PolicyViolationError } from '@/policy/policy-violation.error'; import { OwnershipService } from '@/services/ownership.service'; import { UrlService } from '@/services/url.service'; import { WorkflowExecutionService } from '@/workflows/workflow-execution.service'; @@ -126,4 +127,85 @@ describe('executeErrorWorkflow', () => { }); }); }); + + describe('policy violations', () => { + const makeRunData = (error: Error): IRun => ({ + data: createRunExecutionData({ + resultData: { error: error as IRun['data']['resultData']['error'], runData: {} }, + }), + mode: 'trigger', + startedAt: new Date(), + storedAt: 'db', + status: 'error', + }); + + const violation = () => + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]); + + // A refusal repeats on every attempt, so it must not reach user automation. + it('does not run a configured error workflow', () => { + const workflowData = mock({ + id: 'workflow-123', + settings: { errorWorkflow: 'error-workflow-456' }, + nodes: [], + }); + + executeErrorWorkflow(workflowData, makeRunData(violation()), 'trigger'); + + expect(workflowExecutionService.executeErrorWorkflow).not.toHaveBeenCalled(); + }); + + // The worse case: this would execute the graph policy just refused. + it('does not run the workflow through its own Error Trigger', () => { + const workflowData = mock({ + id: 'workflow-123', + settings: {}, + nodes: [mock({ type: 'n8n-nodes-base.errorTrigger' })], + }); + + executeErrorWorkflow(workflowData, makeRunData(violation()), 'internal'); + + expect(workflowExecutionService.executeErrorWorkflow).not.toHaveBeenCalled(); + }); + + // What `WorkflowRunner.processError` hands to the lifecycle hook: the error + // spread into a plain object, so `instanceof` no longer holds. + it('does not run the error workflow for a flattened refusal', () => { + const error = violation(); + const workflowData = mock({ + id: 'workflow-123', + settings: { errorWorkflow: 'error-workflow-456' }, + nodes: [], + }); + + executeErrorWorkflow( + workflowData, + makeRunData({ ...error, message: error.message, stack: error.stack } as Error), + 'trigger', + 'execution-1', + ); + + expect(workflowExecutionService.executeErrorWorkflow).not.toHaveBeenCalled(); + }); + + it('still runs the error workflow for an ordinary failure', async () => { + const workflowData = mock({ + id: 'workflow-123', + settings: { errorWorkflow: 'error-workflow-456' }, + nodes: [], + }); + ownershipService.getWorkflowProjectCached.mockResolvedValue({ id: 'project-1' } as never); + + executeErrorWorkflow( + workflowData, + makeRunData(new NodeOperationError(mockNode, 'Test error')), + 'trigger', + ); + await new Promise(process.nextTick); + + expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalled(); + }); + }); }); diff --git a/packages/cli/src/execution-lifecycle/execute-error-workflow.ts b/packages/cli/src/execution-lifecycle/execute-error-workflow.ts index de2fc4c3bc7..3ebfc040d42 100644 --- a/packages/cli/src/execution-lifecycle/execute-error-workflow.ts +++ b/packages/cli/src/execution-lifecycle/execute-error-workflow.ts @@ -5,6 +5,7 @@ import { ErrorReporter } from 'n8n-core'; import type { IRun, IWorkflowBase, WorkflowExecuteMode } from 'n8n-workflow'; import type { IWorkflowErrorData } from '@/interfaces'; +import { isPolicyRefusal } from '@/policy/policy-violation.error'; import { OwnershipService } from '@/services/ownership.service'; import { UrlService } from '@/services/url.service'; @@ -55,6 +56,11 @@ export function executeErrorWorkflow( } if (fullRunData.data.resultData.error !== undefined) { + // An administrative refusal, not a workflow failure — and where a workflow is + // its own error handler, this would run the graph policy just refused. + // Not `instanceof`: a failed execution arrives here with its error flattened. + if (isPolicyRefusal(fullRunData.data.resultData.error)) return; + let workflowErrorData: IWorkflowErrorData; const workflowId = workflowData.id; diff --git a/packages/cli/src/policy/__tests__/policy-violation.error.test.ts b/packages/cli/src/policy/__tests__/policy-violation.error.test.ts index 930e1f2e618..1d81fd0ea68 100644 --- a/packages/cli/src/policy/__tests__/policy-violation.error.test.ts +++ b/packages/cli/src/policy/__tests__/policy-violation.error.test.ts @@ -4,7 +4,11 @@ import { UserError } from 'n8n-workflow'; import { classifyHttpError, HttpErrorKind } from '@/errors/http-error-classifier'; import { serializeInternalRestError } from '@/errors/http-error-serializers'; -import { PolicyViolationError, type NonEmptyViolations } from '../policy-violation.error'; +import { + isPolicyRefusal, + PolicyViolationError, + type NonEmptyViolations, +} from '../policy-violation.error'; const violation = (overrides: Partial = {}): PolicyViolation => ({ kind: 'node-type-unavailable', @@ -88,4 +92,34 @@ describe('PolicyViolationError', () => { expect(body.meta).toEqual({ violations: [violation()] }); }); }); + + describe('isPolicyRefusal', () => { + const violation = () => + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked' }, + ]); + + it('recognises a live error', () => { + expect(isPolicyRefusal(violation())).toBe(true); + }); + + // `WorkflowRunner.processError` spreads a failed execution's error into a plain + // object, dropping the prototype, so `instanceof` no longer holds. + it('recognises one flattened by execution failure serialization', () => { + const error = violation(); + const flattened = { ...error, message: error.message, stack: error.stack }; + + expect(flattened instanceof PolicyViolationError).toBe(false); + expect(isPolicyRefusal(flattened)).toBe(true); + }); + + it.each([ + ['an ordinary error', new Error('boom')], + ['an unrelated object', { violations: [] }], + ['null', null], + ['undefined', undefined], + ])('does not recognise %s', (_label, value) => { + expect(isPolicyRefusal(value)).toBe(false); + }); + }); }); diff --git a/packages/cli/src/policy/policy-violation.error.ts b/packages/cli/src/policy/policy-violation.error.ts index 057d44bb863..018197b055c 100644 --- a/packages/cli/src/policy/policy-violation.error.ts +++ b/packages/cli/src/policy/policy-violation.error.ts @@ -31,6 +31,13 @@ export class PolicyViolationError extends UserError { readonly meta: { violations: PolicyViolation[] }; + /** + * An own property, so it survives the flattening that execution failure + * serialization does to errors — where `instanceof` no longer holds. Read it + * through {@link isPolicyRefusal}, never directly. + */ + readonly isPolicyRefusal = true; + /** * @param violations All of them, not just the first — a user fixing a workflow deserves the * whole list. @@ -45,3 +52,23 @@ export class PolicyViolationError extends UserError { this.meta = { violations: this.violations }; } } + +/** + * Whether an error is a policy refusal, including one flattened into a plain + * object by execution failure serialization, where `instanceof` no longer holds. + * + * The default way to ask. Use `instanceof` only where the typed instance itself + * is needed, e.g. to read {@link PolicyViolationError.violations}. + */ +export function isPolicyRefusal(error: unknown): boolean { + if (error instanceof PolicyViolationError) return true; + + // `hasOwn`, not `in`: the marker is an own property by contract, and an inherited + // one would let an unrelated error suppress cleanup and retries. + return ( + typeof error === 'object' && + error !== null && + Object.hasOwn(error, 'isPolicyRefusal') && + (error as { isPolicyRefusal?: unknown }).isPolicyRefusal === true + ); +} diff --git a/packages/cli/src/workflows/__tests__/workflow.service.test.ts b/packages/cli/src/workflows/__tests__/workflow.service.test.ts index 89f811cb374..3da17362fee 100644 --- a/packages/cli/src/workflows/__tests__/workflow.service.test.ts +++ b/packages/cli/src/workflows/__tests__/workflow.service.test.ts @@ -35,6 +35,7 @@ import type { RedactionEnforcementService } from '@/modules/redaction/redaction- import type { PolicyCleared } from '@n8n/decorators'; import { userHasScopes } from '@/permissions.ee/check-access'; import type { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; +import { PolicyViolationError } from '@/policy/policy-violation.error'; import type { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar'; import type { ScheduleTriggerJobRegistrar } from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar'; import type { OwnershipService } from '@/services/ownership.service'; @@ -1062,6 +1063,8 @@ describe('WorkflowService', () => { let pollTriggerJobRegistrarMock: MockProxy; let workflowPublishGuardMock: MockProxy; let workflowMutationHooksMock: MockProxy; + let policyEnforcementServiceMock: MockProxy; + let ownershipServiceMock: MockProxy; const WORKFLOW_ID = 'workflow-1'; const PREVIOUS_VERSION_ID = 'v1'; @@ -1117,6 +1120,14 @@ describe('WorkflowService', () => { pollTriggerJobRegistrarMock = mock(); workflowPublishGuardMock = mock(); workflowMutationHooksMock = mock(); + // Stands in for the dummy always-allow check: with no backend registered the + // real service clears every publish. + policyEnforcementServiceMock = mock(); + policyEnforcementServiceMock.hasChecksFor.mockReturnValue(true); + ownershipServiceMock = mock(); + ownershipServiceMock.getWorkflowProjectCached.mockResolvedValue( + mock({ id: 'project-1' }), + ); workflowRepositoryMock.create.mockImplementation( (data) => Object.assign(new WorkflowEntity(), data) as WorkflowEntity, @@ -1127,7 +1138,7 @@ describe('WorkflowService', () => { mock(), // sharedWorkflowRepository workflowRepositoryMock, // workflowRepository mock(), // workflowTagMappingRepository - mock(), // ownershipService + ownershipServiceMock, // ownershipService mock(), // tagService workflowHistoryServiceMock, // workflowHistoryService externalHooksMock, // externalHooks @@ -1156,7 +1167,7 @@ describe('WorkflowService', () => { workflowHookContextServiceMock, // workflowHookContextService workflowPublishGuardMock, // workflowPublishGuard workflowMutationHooksMock, // workflowMutationHooks - mock(), // policyEnforcementService + policyEnforcementServiceMock, // policyEnforcementService ); // Bypass validation internals @@ -1622,6 +1633,175 @@ describe('WorkflowService', () => { expect(externalHooksMock.run).not.toHaveBeenCalled(); }); + + describe('policy enforcement', () => { + const arrangeSuccessfulActivation = (workflow: WorkflowEntity) => { + workflowFinderServiceMock.findWorkflowForUser.mockResolvedValue(workflow); + workflowRepositoryMock.findOne.mockResolvedValue(workflow); + externalHooksMock.run.mockResolvedValue(undefined); + vi.spyOn( + workflowService as unknown as { _addToActiveWorkflowManager: () => Promise }, + '_addToActiveWorkflowManager', + ).mockResolvedValue(undefined); + }; + + test('enforces the publish with the version being activated and the owning project', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + const versionToActivate = makeVersionToActivate(); + arrangeSuccessfulActivation(workflow); + workflowHistoryServiceMock.getVersion.mockResolvedValue(versionToActivate); + + await workflowService.activateWorkflow(mock(), WORKFLOW_ID, { + versionId: TARGET_VERSION_ID, + }); + + expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledExactlyOnceWith( + { + workflow: { + id: WORKFLOW_ID, + name: workflow.name, + nodes: versionToActivate.nodes, + }, + projectId: 'project-1', + }, + ); + }); + + // The candidate shares its node array with the version row, so an in-place + // mutation would otherwise be policed even though it is never persisted. + test('polices the version even when the hook mutates the nodes in place', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + const versionToActivate = makeVersionToActivate(); + const originalNodes = [...versionToActivate.nodes]; + arrangeSuccessfulActivation(workflow); + workflowHistoryServiceMock.getVersion.mockResolvedValue(versionToActivate); + + externalHooksMock.run.mockImplementation(async (_name, args) => { + const [candidate] = args as unknown as [WorkflowEntity]; + candidate.nodes.push({ name: 'Injected by hook' } as INode); + }); + + await workflowService.activateWorkflow(mock(), WORKFLOW_ID, { + versionId: TARGET_VERSION_ID, + }); + + expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledWith( + expect.objectContaining({ + workflow: expect.objectContaining({ nodes: originalNodes }), + }), + ); + }); + + // Only the version row gets registered, so a hook that rewrites the candidate + // changes a graph that never runs. + test('polices the version being published, not a hook-mutated candidate', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + const versionToActivate = makeVersionToActivate(); + arrangeSuccessfulActivation(workflow); + workflowHistoryServiceMock.getVersion.mockResolvedValue(versionToActivate); + + externalHooksMock.run.mockImplementation(async (_name, args) => { + const [candidate] = args as unknown as [WorkflowEntity]; + candidate.nodes = [{ name: 'Rewritten by hook' } as INode]; + }); + + await workflowService.activateWorkflow(mock(), WORKFLOW_ID, { + versionId: TARGET_VERSION_ID, + }); + + expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledWith( + expect.objectContaining({ + workflow: expect.objectContaining({ nodes: versionToActivate.nodes }), + }), + ); + }); + + // Unlike the review gate: re-applying the current version still re-registers. + test('enforces when re-applying the already-published version', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + arrangeSuccessfulActivation(workflow); + workflowHistoryServiceMock.getVersion.mockResolvedValue(makeActiveVersion()); + + await workflowService.activateWorkflow(mock(), WORKFLOW_ID, { + versionId: PREVIOUS_VERSION_ID, + }); + + expect(workflowPublishGuardMock.assertCanPublish).not.toHaveBeenCalled(); + expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledTimes(1); + }); + + test.each([ + ['current publication path', false], + ['outbox publication path', true], + ] as const)( + 'publishes nothing on the %s when policy blocks the version', + async (_path, useWorkflowPublicationService) => { + globalConfigMock.workflows.useWorkflowPublicationService = useWorkflowPublicationService; + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + arrangeSuccessfulActivation(workflow); + workflowHistoryServiceMock.getVersion.mockResolvedValue(makeVersionToActivate()); + policyEnforcementServiceMock.enforceWorkflowPublish.mockRejectedValue( + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked' }, + ]), + ); + + await expect( + workflowService.activateWorkflow(mock(), WORKFLOW_ID, { + versionId: TARGET_VERSION_ID, + }), + ).rejects.toBeInstanceOf(PolicyViolationError); + + expect(workflow.activeVersionId).toBe(PREVIOUS_VERSION_ID); + expect(workflowRepositoryMock.update).not.toHaveBeenCalled(); + expect(activeWorkflowManagerMock.add).not.toHaveBeenCalled(); + expect(activeWorkflowManagerMock.remove).not.toHaveBeenCalled(); + expect(outboxRepositoryMock.enqueue).not.toHaveBeenCalled(); + expect(workflowPublishHistoryRepositoryMock.addRecord).not.toHaveBeenCalled(); + expect(workflowMutationHooksMock.afterWorkflowPublished).not.toHaveBeenCalled(); + }, + ); + + test('does not enforce while unpublishing', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + workflowFinderServiceMock.findWorkflowForUser.mockResolvedValue(workflow); + + await workflowService.deactivateWorkflow(mock(), WORKFLOW_ID); + + expect(policyEnforcementServiceMock.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + // An unevaluated project rule is not a passed one, so the lookup is unguarded. + test('propagates a failed ownership lookup instead of policing a null scope', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + arrangeSuccessfulActivation(workflow); + workflowHistoryServiceMock.getVersion.mockResolvedValue(makeVersionToActivate()); + ownershipServiceMock.getWorkflowProjectCached.mockRejectedValue(new Error('no owner row')); + + await expect( + workflowService.activateWorkflow(mock(), WORKFLOW_ID, { + versionId: TARGET_VERSION_ID, + }), + ).rejects.toThrow('no owner row'); + + expect(policyEnforcementServiceMock.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + // A feature that is merely absent must not cost a lookup on every publish. + test('does not resolve ownership when no check is registered', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID }); + arrangeSuccessfulActivation(workflow); + workflowHistoryServiceMock.getVersion.mockResolvedValue(makeVersionToActivate()); + policyEnforcementServiceMock.hasChecksFor.mockReturnValue(false); + + await workflowService.activateWorkflow(mock(), WORKFLOW_ID, { + versionId: TARGET_VERSION_ID, + }); + + expect(ownershipServiceMock.getWorkflowProjectCached).not.toHaveBeenCalled(); + expect(policyEnforcementServiceMock.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + }); }); describe('deactivateWorkflowAsSystem()', () => { diff --git a/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts b/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts index ebb23611e4c..5e9d5599074 100644 --- a/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts @@ -10,6 +10,7 @@ import { mock } from 'vitest-mock-extended'; import type { ErrorReporter } from 'n8n-core'; import type { ActivationErrorsService } from '@/activation-errors.service'; +import { PolicyViolationError } from '@/policy/policy-violation.error'; import type { Push } from '@/push'; import type { Publisher } from '@/scaling/pubsub/publisher.service'; import { PublicationStatusReporter } from '@/workflows/publication/publication-status-reporter'; @@ -305,6 +306,23 @@ describe('PublicationStatusReporter', () => { }); }); + // An expected denial: the record must still fail and the UI must still be told, + // but it is not a fault to report. + test('failed by policy marks the record failed without reporting a fault', async () => { + const error = new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]); + + await reporter.report(makeRecord(), { type: 'failed', error }); + + expect(errorReporter.error).not.toHaveBeenCalled(); + expect(outboxRepository.markFailed).toHaveBeenCalledWith(1, error.message, entityManager); + expect(push.broadcast).toHaveBeenCalledWith({ + type: 'workflowFailedToActivate', + data: { workflowId: 'wf-1', errorMessage: error.message }, + }); + }); + test('failed with triggerStatuses writes rows before marking failed', async () => { const error = new Error('partial registration failed'); diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts index a42076b480e..1eef8c00612 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts @@ -15,6 +15,9 @@ import { WebhookPathTakenError } from 'n8n-workflow'; import { TELEMETRY_EVENT } from '@n8n/telemetry'; import type { NodeTypes } from '@/node-types'; +import type { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; +import { PolicyViolationError } from '@/policy/policy-violation.error'; +import type { OwnershipService } from '@/services/ownership.service'; import type { Telemetry } from '@/telemetry'; import { WorkflowPublicationApplier } from '@/workflows/publication/workflow-publication-applier'; import type { WorkflowTriggerActivator } from '@/workflows/triggers/workflow-trigger-activator'; @@ -32,6 +35,9 @@ describe('WorkflowPublicationApplier', () => { const nodeTypes = mock(); const workflowService = mock(); const telemetry = mock(); + // Clears by default, which is what the real service does with no policy backend. + const policyEnforcementService = mock(); + const ownershipService = mock(); const applier = new WorkflowPublicationApplier( logger, @@ -43,6 +49,8 @@ describe('WorkflowPublicationApplier', () => { nodeTypes, workflowService, telemetry, + policyEnforcementService, + ownershipService, ); function makeRecord( @@ -129,6 +137,10 @@ describe('WorkflowPublicationApplier', () => { workflowTriggerActivator.getTriggerKinds.mockImplementation( (nodes) => new Map(nodes.map((node) => [node.id, 'in-memory'])), ); + ownershipService.getWorkflowProjectCached.mockResolvedValue(mock({ id: 'project-1' })); + // `clearAllMocks` keeps implementations, so restore the clearing default. + policyEnforcementService.hasChecksFor.mockReturnValue(true); + policyEnforcementService.enforceWorkflowPublish.mockResolvedValue(mock()); }); test('skips with workflow-not-found when the workflow is gone', async () => { @@ -242,6 +254,127 @@ describe('WorkflowPublicationApplier', () => { }); }); + describe('policy enforcement', () => { + const violation = () => + new PolicyViolationError([ + { kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' }, + ]); + + test('enforces with the version being published and the owning project', async () => { + workflowRepository.findOneBy.mockResolvedValue( + makeWorkflow({ activeVersionId: 'v-1', name: 'My workflow' }), + ); + const versionWithNodes = { + ...makeVersion('v-2'), + nodes: [triggerNode('a')], + } as WorkflowHistory; + workflowHistoryRepository.findOneBy.mockResolvedValue(versionWithNodes); + + await applier.apply(makeRecord(), abort); + + expect(policyEnforcementService.enforceWorkflowPublish).toHaveBeenCalledExactlyOnceWith({ + workflow: { + id: 'wf-1', + name: 'My workflow', + nodes: versionWithNodes.nodes, + }, + projectId: 'project-1', + }); + }); + + // The no-change branch still advances the published version, which running + // triggers re-read on their next fire. + test('enforces even when the trigger diff is empty', async () => { + const trigger = triggerNode('a'); + setTriggerSets([trigger], [{ ...trigger }]); + + const result = await applier.apply(makeRecord(), abort); + + expect(result.type).toBe('completed'); + expect(policyEnforcementService.enforceWorkflowPublish).toHaveBeenCalledTimes(1); + }); + + test('fails the record without advancing or touching triggers when policy blocks', async () => { + policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(violation()); + + const result = await applier.apply(makeRecord(), abort); + + expect(result).toMatchObject({ type: 'failed' }); + expect((result as { error: Error }).error).toBeInstanceOf(PolicyViolationError); + expect(workflowPublishedVersionRepository.setPublishedVersion).not.toHaveBeenCalled(); + expect(workflowTriggerActivator.activate).not.toHaveBeenCalled(); + expect(workflowTriggerActivator.deactivate).not.toHaveBeenCalled(); + }); + + // Left as `activated`, these rows read as drift and get re-enqueued forever. + test('reports every desired trigger as failed so no activated rows survive', async () => { + workflowTriggerActivator.getEnabledTriggerNodes.mockReturnValue([ + triggerNode('a'), + triggerNode('b'), + ]); + policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(violation()); + + const result = await applier.apply(makeRecord(), abort); + + expect(result).toMatchObject({ + type: 'failed', + triggerStatuses: [ + { + nodeId: 'a', + nodeName: 'a', + status: 'failed', + triggerKind: 'in-memory', + errorMessage: 'Blocked by policy', + }, + { + nodeId: 'b', + nodeName: 'b', + status: 'failed', + triggerKind: 'in-memory', + errorMessage: 'Blocked by policy', + }, + ], + }); + }); + + // An unevaluated project rule is not a passed one, so the lookup is unguarded. + test('propagates a failed ownership lookup instead of policing a null scope', async () => { + ownershipService.getWorkflowProjectCached.mockRejectedValue(new Error('no owner row')); + + await expect(applier.apply(makeRecord(), abort)).rejects.toThrow('no owner row'); + + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + // A feature that is merely absent must not cost a lookup on every publication. + test('does not resolve ownership when no check is registered', async () => { + policyEnforcementService.hasChecksFor.mockReturnValue(false); + + const result = await applier.apply(makeRecord(), abort); + + expect(result.type).toBe('completed'); + expect(ownershipService.getWorkflowProjectCached).not.toHaveBeenCalled(); + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + test('does not enforce while unpublishing', async () => { + workflowRepository.findOneBy.mockResolvedValue( + makeWorkflow({ active: true, activeVersionId: null }), + ); + + await applier.apply(makeRecord(), abort); + + expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled(); + }); + + // Only a violation is a verdict; a broken check must not read as "blocked". + test('propagates a non-violation error instead of failing the record', async () => { + policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(new Error('boom')); + + await expect(applier.apply(makeRecord(), abort)).rejects.toThrow('boom'); + }); + }); + test('returns version-missing when the published version history row is gone', async () => { workflowHistoryRepository.findOneBy.mockResolvedValue(null); diff --git a/packages/cli/src/workflows/publication/publication-status-reporter.ts b/packages/cli/src/workflows/publication/publication-status-reporter.ts index 9c18f5c6122..a4a1659334c 100644 --- a/packages/cli/src/workflows/publication/publication-status-reporter.ts +++ b/packages/cli/src/workflows/publication/publication-status-reporter.ts @@ -12,6 +12,7 @@ import { ErrorReporter } from 'n8n-core'; import { OperationalError } from 'n8n-workflow'; import { ActivationErrorsService } from '@/activation-errors.service'; +import { isPolicyRefusal } from '@/policy/policy-violation.error'; import { Push } from '@/push'; import { Publisher } from '@/scaling/pubsub/publisher.service'; import type { @@ -103,7 +104,11 @@ export class PublicationStatusReporter { } await this.outboxRepository.markFailed(record.id, result.error.message, trx); }); - this.errorReporter.error(result.error, { shouldBeLogged: true }); + // An expected denial, already logged as a warning by the applier — the + // terminal state and the UI push stand, the fault report does not. + if (!isPolicyRefusal(result.error)) { + this.errorReporter.error(result.error, { shouldBeLogged: true }); + } this.pushFailedToActivate(record.workflowId, result.error.message); return; } diff --git a/packages/cli/src/workflows/publication/workflow-publication-applier.ts b/packages/cli/src/workflows/publication/workflow-publication-applier.ts index 8359a9455f2..1887b22cd84 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-applier.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-applier.ts @@ -15,6 +15,9 @@ import { ensureError } from '@n8n/utils/errors/ensure-error'; import type { INode, WorkflowActivateMode } from 'n8n-workflow'; import { NodeTypes } from '@/node-types'; +import { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; +import { PolicyViolationError } from '@/policy/policy-violation.error'; +import { OwnershipService } from '@/services/ownership.service'; import { Telemetry } from '@/telemetry'; import { healNodeIds } from '@/workflows/publication/heal-node-ids'; import type { @@ -69,6 +72,8 @@ export class WorkflowPublicationApplier { private readonly nodeTypes: NodeTypes, private readonly workflowService: WorkflowService, private readonly telemetry: Telemetry, + private readonly policyEnforcementService: PolicyEnforcementService, + private readonly ownershipService: OwnershipService, ) { this.logger = this.logger.scoped('workflow-publication'); } @@ -132,6 +137,9 @@ export class WorkflowPublicationApplier { const healSkip = await this.healBrokenNodeIds(workflow, newVersion); if (healSkip !== null) return healSkip; + const blocked = await this.enforcePublishPolicy(workflow, newVersion); + if (blocked !== null) return blocked; + const oldTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(oldVersion); const desiredTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(newVersion); const triggerKinds = this.workflowTriggerActivator.getTriggerKinds(desiredTriggerNodes); @@ -238,6 +246,60 @@ export class WorkflowPublicationApplier { }); } + /** + * Blocks a version that policy objects to, or `null` to carry on. + * + * Runs before the trigger diff: the no-change branch still advances the + * published version, which running triggers re-read on their next fire. + */ + private async enforcePublishPolicy( + workflow: WorkflowEntity, + newVersion: WorkflowHistory, + ): Promise { + if (!this.policyEnforcementService.hasChecksFor('workflowPublish')) return null; + + // Unguarded, as in `PolicyLifecycleHandler`: an unevaluated project rule is not + // a passed one, so a failed lookup fails the publication. + const project = await this.ownershipService.getWorkflowProjectCached(workflow.id); + + try { + await this.policyEnforcementService.enforceWorkflowPublish({ + workflow: { id: workflow.id, name: workflow.name, nodes: newVersion.nodes }, + projectId: project.id, + }); + + return null; + } catch (e) { + // `instanceof`, not `isPolicyRefusal`: the violations and the `Error` shape + // below both need the typed instance. + if (!(e instanceof PolicyViolationError)) throw e; + + this.logger.warn('Workflow publication blocked by policy', { + workflowId: workflow.id, + versionId: newVersion.versionId, + violations: e.violations.map((violation) => violation.kind), + }); + + // Report every trigger as failed: `activated` rows left behind read as drift + // to the reconciler, which would then re-enqueue this forever. + const desiredTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(newVersion); + const triggerKinds = this.workflowTriggerActivator.getTriggerKinds(desiredTriggerNodes); + + return { + type: 'failed', + error: e, + triggerStatuses: this.buildTriggerStatuses(desiredTriggerNodes, triggerKinds, { + activated: [], + failures: desiredTriggerNodes.map((node) => ({ + nodeId: node.id, + nodeName: node.name, + error: e, + })), + }), + }; + } + } + /** * A first publication (no old version) reports `activate`, so the n8n Trigger's * "Workflow Published" event fires; otherwise the mode follows the record's reason. diff --git a/packages/cli/src/workflows/workflow.service.ts b/packages/cli/src/workflows/workflow.service.ts index e00cbab43b4..f8f41fc7089 100644 --- a/packages/cli/src/workflows/workflow.service.ts +++ b/packages/cli/src/workflows/workflow.service.ts @@ -862,6 +862,10 @@ export class WorkflowService { this._validateTriggerNodeIds(workflowId, versionToActivate); } + // The candidate below shares this array with the version row, and the hook may + // mutate it in place, so snapshot what will actually be registered. + const nodesToPublish = structuredClone(versionToActivate.nodes); + // Run hook before destructive state changes so a rejection leaves // the previous active version running instead of deactivating it. const candidateWorkflow = this.workflowRepository.create({ @@ -886,6 +890,23 @@ export class WorkflowService { }); } + // Polices what gets registered — the version row, not the hook's candidate. + // Enforced on a same-version republish too. + if (this.policyEnforcementService.hasChecksFor('workflowPublish')) { + // Unguarded, as in `PolicyLifecycleHandler`: an unevaluated project rule is + // not a passed one, so a failed lookup fails the publish. + const project = await this.ownershipService.getWorkflowProjectCached(workflowId); + + await this.policyEnforcementService.enforceWorkflowPublish({ + workflow: { + id: workflowId, + name: workflow.name, + nodes: nodesToPublish, + }, + projectId: project.id, + }); + } + // re-applying the already-published version (e.g. a settings-only update) // publishes no new version, so the review gate must not block it. // diff --git a/packages/cli/test/integration/active-workflow-manager.test.ts b/packages/cli/test/integration/active-workflow-manager.test.ts index c2bdc6318e5..450bc664825 100644 --- a/packages/cli/test/integration/active-workflow-manager.test.ts +++ b/packages/cli/test/integration/active-workflow-manager.test.ts @@ -25,6 +25,7 @@ import { ActiveWorkflowManager } from '@/active-workflow-manager'; import { ExecutionService } from '@/executions/execution.service'; import { ExternalHooks } from '@/external-hooks'; import { NodeTypes } from '@/node-types'; +import { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; import { Push } from '@/push'; import { OwnershipService } from '@/services/ownership.service'; import * as WebhookHelpers from '@/webhooks/webhook-helpers'; @@ -136,6 +137,21 @@ describe('init()', () => { expect(validateWorkflowHasTriggerLikeNodeSpy).toHaveBeenCalledTimes(2); }); + + // Startup reactivation is a publish in its own right. Spied rather than registered + // via `setImplementation`, which is single-shot and would leak; `restoreMocks` undoes + // a spy between tests. + it('should enforce the publish policy for each workflow it reactivates', async () => { + const policyEnforcementService = Container.get(PolicyEnforcementService); + vi.spyOn(policyEnforcementService, 'hasChecksFor').mockReturnValue(true); + const enforceSpy = vi.spyOn(policyEnforcementService, 'enforceWorkflowPublish'); + await Promise.all([createActiveWorkflow(), createActiveWorkflow()]); + + await activeWorkflowManager.init(); + + expect(enforceSpy).toHaveBeenCalledTimes(2); + expect(activeWorkflowManager.allActiveInMemory()).toHaveLength(2); + }); }); describe('add()', () => { diff --git a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts b/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts index fbc2c4be782..0694934e8aa 100644 --- a/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts +++ b/packages/cli/test/integration/workflows/workflow-publication-reconciler.test.ts @@ -20,6 +20,7 @@ import type { INode, INodeTypeData } from 'n8n-workflow'; import { ActiveExecutions } from '@/active-executions'; import { ActiveWorkflowManager } from '@/active-workflow-manager'; +import { EventService } from '@/events/event.service'; import { ExecutionService } from '@/executions/execution.service'; import { ExternalHooks } from '@/external-hooks'; import { Push } from '@/push'; @@ -330,6 +331,81 @@ describe('WorkflowPublicationReconciler (integration)', () => { expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); }); + // A publication failing before the mapping advances (e.g. a policy block) leaves + // the skew permanently, so re-enqueueing would loop on every pass. + test('leaves a workflow skewed by a terminally failed publication alone', async () => { + const owner = await createOwner(); + + const trigger = scheduleNode('skew-failed'); + const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); + await setActiveVersion(workflow.id, workflow.versionId); + + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); + + // A newer version is made active, but its publication fails before the + // mapping advances — so activeVersionId and publishedVersionId diverge. + const newVersionId = 'version-2-failed-publish'; + await createWorkflowHistory(workflow, owner, undefined, { + versionId: newVersionId, + nodes: [trigger], + }); + await setActiveVersion(workflow.id, newVersionId); + await outboxRepository.enqueue(workflow.id, newVersionId, 'publish'); + const failing = (await outboxRepository.claimNextPendingRecord())!; + await outboxRepository.markFailed(failing.id, 'Blocked by policy'); + + expect(await publishedVersionRepository.getPublishedVersionId(workflow.id)).toBe( + workflow.versionId, + ); + + await reconciler.reconcile('reconcile'); + + // No third record: the skew is real but permanent, so the reconciler must + // not keep retrying it. + expect(await outboxRepository.countBy({ workflowId: workflow.id })).toBe(2); + expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); + }); + + // The failed-record exclusion above is scoped to the active version so it never + // reaches an unpublish, where the mapping is all that is left to heal. + test('removes a mapping left behind by a failed unpublish, despite the failed record', async () => { + const owner = await createOwner(); + + const trigger = scheduleNode('unpublish-failed'); + const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner); + await setActiveVersion(workflow.id, workflow.versionId); + + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); + await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal); + + // An unpublish another main tore down and reported (triggers gone, status + // rows cleared) whose mapping removal never landed, so its record is + // terminal `failed`. With no in-memory trigger and no status rows, the + // registry detections are blind — only the version comparison can see it. + await Container.get(WorkflowRepository).update(workflow.id, { activeVersionId: null }); + await activeWorkflowTriggers.remove(workflow.id); + await triggerStatusRepository.delete({ workflowId: workflow.id }); + await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish'); + const failing = (await outboxRepository.claimNextPendingRecord())!; + await outboxRepository.markFailed(failing.id, 'Removing the published version failed'); + + expect(await outboxRepository.findVersionSkewedWorkflowIds()).toContain(workflow.id); + + const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); + await reconciler.reconcile('reconcile'); + + // Attribution: the skew detector drove the repair, not another pass. + expect(emitSpy).toHaveBeenCalledWith( + 'workflow-publication-reconciliation', + expect.objectContaining({ versionSkewCount: 1, deficientCount: 0, surplusCount: 0 }), + ); + expect(await publishedVersionRepository.getPublishedVersionId(workflow.id)).toBeNull(); + expect(await outboxRepository.claimNextPendingRecord()).toBeNull(); + + emitSpy.mockRestore(); + }); + test('removes a published-version mapping left behind by a missed unpublish', async () => { const owner = await createOwner(); diff --git a/packages/cli/test/integration/workflows/workflow.service.test.ts b/packages/cli/test/integration/workflows/workflow.service.test.ts index ec10a3e48d9..dcddda8094d 100644 --- a/packages/cli/test/integration/workflows/workflow.service.test.ts +++ b/packages/cli/test/integration/workflows/workflow.service.test.ts @@ -114,8 +114,8 @@ beforeAll(async () => { Container.get(WorkflowHookContextService), // workflowHookContextService workflowPublishGuard, mock(), // workflowMutationHooks - // Real service on purpose: with no policy backend registered it clears every save, - // so these tests also prove save behavior is unchanged when the module is off. + // Real service on purpose: with no backend registered it clears every save and + // publish, so these tests also prove behavior is unchanged with the module off. Container.get(PolicyEnforcementService), // policyEnforcementService ); }); @@ -342,6 +342,29 @@ describe('update()', () => { }); describe('activateWorkflow()', () => { + // The rest of this suite runs with no checks registered, proving activation is + // unchanged when the module is off. Spied rather than registered via + // `setImplementation`, which is single-shot and would leak into those tests. + test('should enforce the publish policy with the version being activated', async () => { + const owner = await createOwner(); + const workflow = await createWorkflowWithHistory({}, owner); + const policyEnforcementService = Container.get(PolicyEnforcementService); + vi.spyOn(policyEnforcementService, 'hasChecksFor').mockReturnValue(true); + const enforceSpy = vi.spyOn(policyEnforcementService, 'enforceWorkflowPublish'); + + const updatedWorkflow = await workflowService.activateWorkflow(owner, workflow.id); + + expect(enforceSpy).toHaveBeenCalledExactlyOnceWith({ + workflow: { + id: workflow.id, + name: workflow.name, + nodes: expect.any(Array), + }, + projectId: expect.any(String), + }); + expect(updatedWorkflow.activeVersionId).toBe(workflow.versionId); + }); + test('should activate current workflow version if no version provided', async () => { const owner = await createOwner(); const workflow = await createWorkflowWithHistory({}, owner);