feat(workflows): route webhook node through the delivery pipeline (full Option 1)
Replaces the one-shot guarded POST with the maintainer's intended end-state: the webhook node now references a CONFIGURED webhook subscription (Settings → Webhooks) and enqueues a real webhook_deliveries row via webhookService.enqueueForWebhook. Delivery then rides the existing worker pipeline, inheriting — not reimplementing — per-delivery SSRF re-validation (validateExternalUrl / GHSA-wmjx-pc37-272r), HMAC signing with the subscription's secret, retry/backoff, and the deliveries audit log. - webhookService.enqueueForWebhook(webhookId, eventType, data): enqueue for one active subscription, bypassing fire()'s event-type matching. No schema change. - webhook action: config.webhookId; unset/missing/inactive → observable skip; dry-run does not enqueue. event_type = workflow.<trigger>. - Editor: webhook node config is now a subscription dropdown (was a raw URL), fed by the admin webhooks list, with a hint pointing to Settings → Webhooks. - EN/DE strings; test asserts enqueue + dry-run no-op + inactive skip.
This commit is contained in:
Luca
@@ -415,22 +415,39 @@ describe('workflow engine', () => {
|
||||
expect(await _internal.resolveTemplateKey('zzznotype', 'promo_')).toBe('promo_default');
|
||||
});
|
||||
|
||||
test('webhook action: dry-run, missing url, and SSRF-guarded private/metadata url', async () => {
|
||||
test('webhook action enqueues a delivery for a configured subscription (full pipeline)', async () => {
|
||||
const webhook = engine.registry.getAction('webhook');
|
||||
expect(typeof webhook).toBe('function'); // registered — no longer a silent no-op
|
||||
const ctx = (config, vars = {}) => ({
|
||||
run: { id: 1, workflow_id: 1, version: 1, trigger_event: 't', entity_type: 'x', entity_id: 1 },
|
||||
run: { id: 1, workflow_id: 1, version: 1, trigger_event: 'invoice.sent', entity_type: 'invoice', entity_id: 5 },
|
||||
node: { config }, vars, db, logger: { warn() {} },
|
||||
});
|
||||
// Dry run never calls out.
|
||||
expect(await webhook(ctx({ url: 'https://example.com/hook' }, { __dryRun: true })))
|
||||
.toMatchObject({ dryRun: true, would: 'webhook' });
|
||||
// No URL → observable skip, not a crash.
|
||||
// No webhook selected → observable skip, not a crash.
|
||||
expect(await webhook(ctx({}))).toMatchObject({ skipped: true });
|
||||
// SSRF: cloud-metadata / private target rejected before any request.
|
||||
const r = await webhook(ctx({ url: 'http://169.254.169.254/latest/meta-data' }));
|
||||
expect(r.skipped).toBe(true);
|
||||
expect(r.reason).toMatch(/rejected/i);
|
||||
|
||||
// A configured, active webhook subscription.
|
||||
const [adminId] = await db('admin_users').insert({ username: 'wfhook', email: 'wf@x.test', password_hash: 'x' });
|
||||
const [whId] = await db('webhooks').insert({
|
||||
name: 'Flow hook', url: 'https://example.com/hook', secret: 'whsec_test',
|
||||
events: JSON.stringify([]), active: true, created_by: adminId,
|
||||
});
|
||||
|
||||
// Dry run does not enqueue.
|
||||
expect(await webhook(ctx({ webhookId: whId }, { __dryRun: true }))).toMatchObject({ dryRun: true, would: 'webhook' });
|
||||
expect(await db('webhook_deliveries').where({ webhook_id: whId }).count('id as c').first()).toMatchObject({ c: 0 });
|
||||
|
||||
// Real run → a pending delivery is enqueued for the worker (which does the
|
||||
// signing + SSRF re-validation + retries).
|
||||
const res = await webhook(ctx({ webhookId: whId }));
|
||||
expect(res.webhook_enqueued).toBe(whId);
|
||||
const del = await db('webhook_deliveries').where({ webhook_id: whId }).first();
|
||||
expect(del).toBeTruthy();
|
||||
expect(del.status).toBe('pending');
|
||||
expect(del.event_type).toBe('workflow.invoice.sent');
|
||||
|
||||
// Inactive / missing subscription → skip.
|
||||
await db('webhooks').where({ id: whId }).update({ active: false });
|
||||
expect((await webhook(ctx({ webhookId: whId }))).skipped).toBe(true);
|
||||
});
|
||||
|
||||
test('isBuiltinFlowActive reflects the built-in ENABLED state (enabled-based mutex)', async () => {
|
||||
|
||||
@@ -210,6 +210,38 @@ function parseJsonField(value, fallback) {
|
||||
try { return JSON.parse(value) ?? fallback; } catch { return fallback; }
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a delivery for ONE specific active webhook subscription, bypassing the
|
||||
* event-type subscription matching that `fire` does. Used by the workflow engine
|
||||
* `webhook` action: the flow author picks a configured webhook (which carries
|
||||
* the URL + signing secret + create-time URL validation), and the delivery then
|
||||
* rides the SAME worker pipeline as every other webhook — per-delivery SSRF
|
||||
* re-validation, HMAC signing, retries/backoff and the audit log, all for free.
|
||||
* Never throws. Returns { enqueued, reason?, deliveryId? }.
|
||||
*/
|
||||
async function enqueueForWebhook(webhookId, eventType, data) {
|
||||
try {
|
||||
const w = await db('webhooks').where({ id: webhookId, active: true }).first();
|
||||
if (!w) return { enqueued: false, reason: 'webhook not found or inactive' };
|
||||
const now = new Date();
|
||||
const deliveryUuid = crypto.randomUUID();
|
||||
const envelope = { id: deliveryUuid, type: eventType, created_at: now.toISOString(), data };
|
||||
await db('webhook_deliveries').insert({
|
||||
webhook_id: w.id,
|
||||
event_type: String(eventType).slice(0, 64),
|
||||
payload: JSON.stringify(envelope),
|
||||
attempt_count: 0,
|
||||
status: 'pending',
|
||||
next_retry_at: now,
|
||||
created_at: now,
|
||||
});
|
||||
return { enqueued: true, webhookId: w.id, deliveryId: deliveryUuid };
|
||||
} catch (err) {
|
||||
logger.error(`[webhookService.enqueueForWebhook] failed for #${webhookId}: ${err.message}`);
|
||||
return { enqueued: false, reason: err.message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Canonical event sub-object for outbound webhooks (#341). Always returns
|
||||
* the full key set so receivers don't have to handle "field missing vs
|
||||
@@ -235,6 +267,7 @@ function buildEventSubject(input = {}) {
|
||||
|
||||
module.exports = {
|
||||
fire,
|
||||
enqueueForWebhook,
|
||||
generateSecret,
|
||||
signPayload,
|
||||
verifySignature,
|
||||
|
||||
@@ -179,25 +179,21 @@ registry.registerAction('notify_pre_event', async (ctx) => {
|
||||
return res;
|
||||
});
|
||||
|
||||
// Call an external webhook (the `webhook` node type + the "Call a webhook"
|
||||
// action both resolve here). POSTs the run context to config.url. SSRF-guarded
|
||||
// via validateExternalUrl — the same NAT64/private-range protection the webhook
|
||||
// delivery worker uses — unless WEBHOOK_ALLOW_PRIVATE_URLS=true (local-dev
|
||||
// opt-out). No redirects. Best-effort: a rejected URL / network error records an
|
||||
// observable skipped step rather than throwing the run.
|
||||
// Call a webhook (the `webhook` node type + the "Call a webhook" action both
|
||||
// resolve here). The flow author picks a CONFIGURED webhook subscription
|
||||
// (config.webhookId, managed in Settings → Webhooks); this enqueues a real
|
||||
// delivery for it, so it rides the same worker pipeline as every other webhook:
|
||||
// per-delivery SSRF re-validation (validateExternalUrl / GHSA-wmjx-pc37-272r),
|
||||
// HMAC signing with the subscription's secret, retries/backoff, and the audit
|
||||
// log — all inherited, nothing reimplemented. Best-effort: an unset / missing /
|
||||
// inactive webhook records an observable skipped step.
|
||||
registry.registerAction('webhook', async (ctx) => {
|
||||
const url = ctx.node.config?.url;
|
||||
if (!url) return { skipped: true, reason: 'no webhook url configured' };
|
||||
if (ctx.vars?.__dryRun) return { dryRun: true, would: 'webhook', url };
|
||||
const webhookId = ctx.node.config?.webhookId ? Number(ctx.node.config.webhookId) : null;
|
||||
if (!webhookId) return { skipped: true, reason: 'no webhook selected (pick one in Settings → Webhooks)' };
|
||||
if (ctx.vars?.__dryRun) return { dryRun: true, would: 'webhook', webhookId };
|
||||
|
||||
if (process.env.WEBHOOK_ALLOW_PRIVATE_URLS !== 'true') {
|
||||
const { validateExternalUrl } = require('../../utils/networkValidation');
|
||||
const check = validateExternalUrl(url);
|
||||
if (!check.valid) return { skipped: true, reason: `url rejected: ${check.error}` };
|
||||
}
|
||||
|
||||
const axios = require('axios');
|
||||
const body = {
|
||||
const eventType = `workflow.${ctx.run.trigger_event || 'webhook'}`;
|
||||
const res = await require('../webhookService').enqueueForWebhook(webhookId, eventType, {
|
||||
workflow: { id: ctx.run.workflow_id, version: ctx.run.version },
|
||||
run: {
|
||||
id: ctx.run.id,
|
||||
@@ -206,18 +202,10 @@ registry.registerAction('webhook', async (ctx) => {
|
||||
entity_id: ctx.run.entity_id,
|
||||
},
|
||||
vars: ctx.vars || {},
|
||||
};
|
||||
try {
|
||||
const res = await axios.post(url, body, {
|
||||
headers: { 'Content-Type': 'application/json', 'User-Agent': 'PicPeak-Workflows/1.0', 'X-PicPeak-Event': ctx.run.trigger_event || '' },
|
||||
timeout: parseInt(process.env.WEBHOOK_HTTP_TIMEOUT_MS || '10000', 10),
|
||||
maxRedirects: 0, // no redirects — SSRF + receivers should give a final URL
|
||||
validateStatus: () => true,
|
||||
});
|
||||
return { webhook_posted: url, status: res.status };
|
||||
} catch (err) {
|
||||
return { skipped: true, reason: `webhook request failed: ${err.message}` };
|
||||
}
|
||||
});
|
||||
return res.enqueued
|
||||
? { webhook_enqueued: res.webhookId, deliveryId: res.deliveryId }
|
||||
: { skipped: true, reason: res.reason };
|
||||
});
|
||||
|
||||
// Create/prepare-document actions — registered so flows referencing them are
|
||||
|
||||
@@ -266,6 +266,10 @@
|
||||
"recipientAdmin": "Admin (sofort gesendet)",
|
||||
"emailTemplate": "E-Mail-Vorlagenschlüssel",
|
||||
"webhookUrl": "Webhook-URL",
|
||||
"webhookTarget": "Webhook",
|
||||
"webhookNone": "— Konfigurierten Webhook wählen —",
|
||||
"webhookInactive": "(inaktiv)",
|
||||
"webhookHint": "Wird über die Webhook-Pipeline zugestellt (Signatur, Wiederholungen, SSRF-Prüfungen). Endpunkte unter Einstellungen → Webhooks verwalten.",
|
||||
"condition": "Bedingung",
|
||||
"exprField": "Feld",
|
||||
"exprOp": "Operator",
|
||||
|
||||
@@ -266,6 +266,10 @@
|
||||
"recipientAdmin": "Admin (sent immediately)",
|
||||
"emailTemplate": "Email template key",
|
||||
"webhookUrl": "Webhook URL",
|
||||
"webhookTarget": "Webhook",
|
||||
"webhookNone": "— Select a configured webhook —",
|
||||
"webhookInactive": "(inactive)",
|
||||
"webhookHint": "Delivered via the webhook pipeline (signing, retries, SSRF checks). Manage endpoints in Settings → Webhooks.",
|
||||
"condition": "Condition",
|
||||
"exprField": "Field",
|
||||
"exprOp": "Operator",
|
||||
|
||||
@@ -9,10 +9,13 @@ import { useTranslation } from 'react-i18next';
|
||||
|
||||
type Cfg = Record<string, any>;
|
||||
|
||||
interface WebhookOption { id: number; name: string; active: boolean }
|
||||
|
||||
interface Props {
|
||||
nodeType: string;
|
||||
config: Cfg;
|
||||
onChange: (next: Cfg) => void;
|
||||
webhooks?: WebhookOption[];
|
||||
}
|
||||
|
||||
const field = 'w-full px-2 py-1.5 rounded border border-neutral-300 dark:border-neutral-600 bg-white dark:bg-neutral-900 text-neutral-900 dark:text-neutral-100 text-sm';
|
||||
@@ -52,7 +55,7 @@ const Row: React.FC<{ label: string; children: React.ReactNode }> = ({ label, ch
|
||||
<div><label className={lbl}>{label}</label>{children}</div>
|
||||
);
|
||||
|
||||
export const NodeConfigPanel: React.FC<Props> = ({ nodeType, config, onChange }) => {
|
||||
export const NodeConfigPanel: React.FC<Props> = ({ nodeType, config, onChange, webhooks = [] }) => {
|
||||
const { t } = useTranslation();
|
||||
const [showJson, setShowJson] = useState(false);
|
||||
const [jsonText, setJsonText] = useState(JSON.stringify(config || {}, null, 2));
|
||||
@@ -109,8 +112,16 @@ export const NodeConfigPanel: React.FC<Props> = ({ nodeType, config, onChange })
|
||||
)}
|
||||
|
||||
{(nodeType === 'action' || nodeType === 'webhook') && (config.action === 'webhook' || nodeType === 'webhook') && (
|
||||
<Row label={t('workflows.editor.webhookUrl', 'Webhook URL')}>
|
||||
<input className={field} value={config.url || ''} onChange={(e) => set({ url: e.target.value })} placeholder="https://…" />
|
||||
<Row label={t('workflows.editor.webhookTarget', 'Webhook')}>
|
||||
<select className={field} value={config.webhookId ?? ''} onChange={(e) => set({ webhookId: e.target.value ? Number(e.target.value) : undefined })}>
|
||||
<option value="">{t('workflows.editor.webhookNone', '— Select a configured webhook —')}</option>
|
||||
{webhooks.map((w) => (
|
||||
<option key={w.id} value={w.id}>{w.name}{w.active ? '' : ` ${t('workflows.editor.webhookInactive', '(inactive)')}`}</option>
|
||||
))}
|
||||
</select>
|
||||
<p className="text-xs text-neutral-500 dark:text-neutral-400 mt-1">
|
||||
{t('workflows.editor.webhookHint', 'Delivered via the webhook pipeline (signing, retries, SSRF checks). Manage endpoints in Settings → Webhooks.')}
|
||||
</p>
|
||||
</Row>
|
||||
)}
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ import '@xyflow/react/dist/style.css';
|
||||
import dagre from '@dagrejs/dagre';
|
||||
import { ArrowLeft, Save, Trash2, Wand2, Code } from 'lucide-react';
|
||||
import { Button, Loading } from '../../../components/common';
|
||||
import { api } from '../../../config/api';
|
||||
import { useAdminDarkMode } from '../../../contexts/AdminDarkModeContext';
|
||||
import { workflowsService, type WorkflowNodeType } from '../../../services/workflows.service';
|
||||
import { NodeConfigPanel } from './NodeConfigPanel';
|
||||
@@ -127,6 +128,13 @@ export const WorkflowEditorPage: React.FC = () => {
|
||||
enabled: Number.isFinite(workflowId),
|
||||
});
|
||||
|
||||
// Configured webhook subscriptions — the webhook node references one of these
|
||||
// (the delivery then rides the webhook worker pipeline).
|
||||
const { data: webhooks = [] } = useQuery({
|
||||
queryKey: ['admin-webhooks'],
|
||||
queryFn: async () => (await api.get<Array<{ id: number; name: string; active: boolean }>>('/admin/webhooks')).data,
|
||||
});
|
||||
|
||||
const [nodes, setNodes, onNodesChange] = useNodesState<Node>([]);
|
||||
const [edges, setEdges, onEdgesChange] = useEdgesState<Edge>([]);
|
||||
const [name, setName] = useState('');
|
||||
@@ -366,6 +374,7 @@ export const WorkflowEditorPage: React.FC = () => {
|
||||
nodeType={(selectedNode.data as any).nodeType}
|
||||
config={(selectedNode.data as any).config || {}}
|
||||
onChange={(cfg) => updateNodeConfig(selectedNode.id, cfg)}
|
||||
webhooks={webhooks}
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
|
||||
Reference in New Issue
Block a user