#!/usr/bin/env python3 """Check an n8n workflow export or a Make scenario blueprint for delivery risks. Usage: python3 scripts/check_ipaas_export.py EXPORT.json [--tier 1|2|3] [--ordered yes|no] [--json] n8n (workflow Download, or get_workflow_details from the n8n MCP server; the wrappers {"workflow": ..., "triggerInfo": ...} and {"data": ...} are unwrapped, and the published nodes are read when activeVersion.sameAsDraft is false): IX01 FAIL retryOnFail on a create or POST/PUT/PATCH/DELETE node with no idempotency evidence; WARN on an app node whose operation is not in the export (n8n leaves parameters at their default out of the export, so it may be a create) IX02 FAIL On Error "Continue" (continueRegularOutput) whose output reaches a write; WARN elsewhere IX03 WARN legacy continueOnFail IX04 FAIL no error workflow on a tier 1 or 2 flow (WARN on tier 3) IX05 FAIL literal secret in a header or query parameter (name reported, value never printed) IX06 WARN webhook without authentication IX07 WARN webhook answers on receipt while failed runs are not saved, or answers after the last node while the path calls external systems IX08 WARN retryOnFail on an HTTP node: fixed waits that ignore Retry-After IX09 INFO workflow active IX10 WARN Code or Function node: review by hand A node with no parameters key (sanitized MCP output) skips IX01, IX05, IX06 and IX07 and is listed under "Not checked"; those four rules then print no PASS. Make (Export Blueprint; a Make API blueprint response is unwrapped): IX05 FAIL literal secret in a mapper header or query string (name reported, value never printed) IX21 FAIL Store incomplete executions off on tier 1 or 2 (WARN on tier 3) IX22 FAIL data loss allowed when incomplete-execution storage is full IX23 WARN parallel runs when --ordered yes IX24 FAIL create module with a Retry (Break) handler and no idempotency evidence; WARN when only automatic retry of incomplete executions covers it IX25 FAIL Skip (Ignore) handler on a write module on tier 1 (WARN otherwise) IX26 WARN Resume handler on a write module IX27 INFO confidential: Make keeps no payloads to resolve errors with IX28 WARN instant webhook trigger with maxErrors above 1 IX29 WARN webhook response that is not the last module on its route IX30 WARN write module with no error handler on a tier 1 or 2 flow Idempotency evidence: a header named Idempotency-Key, an upsert operation, or a parameter named like externalId, external_id, idProperty or unique_id. Writes are found by name and method, which the output lists under "Not checked". Exit 1 on any FAIL, 2 on bad input or an unsupported export, 0 otherwise. Changes nothing. """ import argparse import json import os import re import sys from collections import deque EVIDENCE = re.compile(r"idempotency|external_?id|idproperty|unique_?id", re.I) SECRET = re.compile(r"authorization|api[-_]?key|token|secret", re.I) CREATE_WORDS = ("create", "add", "insert", "send", "post") WRITE_WORDS = CREATE_WORDS + ("update", "upsert", "delete", "remove", "upload", "archive", "publish", "replace", "append") WRITE_METHODS = {"POST", "PUT", "PATCH", "DELETE"} N8N_HTTP = "n8n-nodes-base.httprequest" N8N_WEBHOOK = "n8n-nodes-base.webhook" N8N_CODE = {"n8n-nodes-base.code", "n8n-nodes-base.function", "n8n-nodes-base.functionitem"} N8N_LOCAL = {"set", "code", "function", "functionitem", "if", "switch", "merge", "noop", "respondtowebhook", "filter", "splitinbatches", "itemlists", "datetime", "wait", "stopanderror", "aggregate", "splitout", "sort", "limit", "removeduplicates", "renamekeys", "html", "xml", "markdown", "crypto", "comparedatasets", "summarize", "stickynote", "manualtrigger", "errortrigger", "scheduletrigger", "webhook"} MAKE_LOCAL_PREFIXES = ("builtin:", "gateway:", "util:", "json:", "regexp:", "placeholder:") UNSUPPORTED = ("Unsupported export: this script reads n8n workflow JSON and Make blueprints only; review Zapier, " "Workato, Boomi and other tools with the manual checklist in references/method-selection.md.") def names_in(value): """Every dict key and every 'name' value found anywhere inside a parameters block.""" found = [] if isinstance(value, dict): for k, v in value.items(): found.append(str(k)) if k == "name" and isinstance(v, str): found.append(v) found += names_in(v) elif isinstance(value, list): for v in value: found += names_in(v) return found def name_value_pairs(params, keys): """name/value entries under header or query parameter blocks.""" pairs = [] def walk(value, inside): if isinstance(value, dict): if inside and "name" in value and "value" in value: pairs.append((str(value["name"]), value["value"])) for k, v in value.items(): walk(v, inside or k.lower() in keys) elif isinstance(value, list): for v in value: walk(v, inside) walk(params, False) return pairs class Report: def __init__(self): self.items = [] def add(self, level, rule, target, message): self.items.append({"level": level, "rule": rule, "target": target, "message": message}) def unwrap_n8n(data): """Return the workflow from a Download, a {"data": ...} response or get_workflow_details output.""" for key in ("data", "workflow"): inner = data.get(key) if isinstance(data, dict) else None if isinstance(inner, dict) and "nodes" in inner: return inner return data def check_n8n(wf, tier, report, not_checked): nodes = [n for n in wf.get("nodes", []) if isinstance(n, dict)] by_name = {n.get("name"): n for n in nodes} links = wf.get("connections") or {} def ntype(node): return str(node.get("type", "")).lower() def params(node): return node.get("parameters") or {} def op(node): return str(params(node).get("operation", "")).lower() def method(node): return str(params(node).get("method") or params(node).get("requestMethod") or "GET").upper() def is_upsert(node): return "upsert" in op(node) or "orupdate" in op(node) def is_create(node): if ntype(node) == N8N_HTTP: return method(node) in WRITE_METHODS return not is_upsert(node) and op(node).startswith(CREATE_WORDS) def is_write(node): return is_create(node) or is_upsert(node) or op(node).startswith(WRITE_WORDS) def external(node): short = ntype(node).rsplit(".", 1)[-1] return ntype(node) == N8N_HTTP or short not in N8N_LOCAL def downstream(name): seen, queue = set(), deque([name]) while queue: for outputs in (links.get(queue.popleft()) or {}).values(): for branch in outputs or []: for link in branch or []: nxt = link.get("node") if nxt in by_name and nxt not in seen: seen.add(nxt) queue.append(nxt) return [by_name[n] for n in seen] def evidence(node): return is_upsert(node) or any(EVIDENCE.search(n) for n in names_in(params(node))) settings = wf.get("settings") blind = [str(n.get("name")) for n in nodes if "parameters" not in n] for node in nodes: label = f'node "{node.get("name")}"' seen = "parameters" in node # without them a retrying POST reads as a GET and would pass IX01 if seen and node.get("retryOnFail") and is_create(node) and not evidence(node): what = f"{method(node)} request" if ntype(node) == N8N_HTTP else f"{op(node)} operation" report.add("FAIL", "IX01", label, f"retries a {what} up to {node.get('maxTries', 3)} times with no idempotency evidence " "(Idempotency-Key header, upsert or external ID); n8n retries on any error, including a timeout after " "the destination accepted the request, so each retry can create a duplicate") elif seen and node.get("retryOnFail") and ntype(node) != N8N_HTTP and external(node) and not op(node) and not evidence(node): report.add("WARN", "IX01", label, "retryOnFail on an app node whose operation is not in the export (n8n leaves " "parameters at their default out of the export); open the node and check whether it creates") if node.get("onError") == "continueRegularOutput": writes = [n.get("name") for n in downstream(node.get("name")) if is_write(n)] if writes: report.add("FAIL", "IX02", label, f"On Error is Continue, which passes the last valid data on, and the output reaches " f"write node(s) {', '.join(sorted(writes))}; use Continue (using error output) routed to an exception path") else: report.add("WARN", "IX02", label, "On Error is Continue, which passes the last valid data on; prefer Continue " "(using error output) so failures are visible") if node.get("continueOnFail"): report.add("WARN", "IX03", label, "legacy continueOnFail hides failures; set On Error explicitly") for pname, value in name_value_pairs(params(node), {"headerparameters", "queryparameters"}): if SECRET.search(pname) and isinstance(value, str) and value and not value.startswith("="): report.add("FAIL", "IX05", label, f"parameter {pname} holds a literal value (not printed); move it into an n8n " "credential and rotate it, because the export carries it") if seen and ntype(node) == N8N_WEBHOOK: auth = str(params(node).get("authentication") or "none") if auth == "none": report.add("WARN", "IX06", label, "webhook has no authentication; anyone with the URL can start the workflow") mode = str(params(node).get("responseMode") or "onReceived") no_errors_saved = isinstance(settings, dict) and settings.get("saveDataErrorExecution") == "none" calls = [n.get("name") for n in downstream(node.get("name")) if external(n)] if mode == "onReceived" and no_errors_saved: report.add("WARN", "IX07", label, "answers on receipt while failed executions are not saved " "(saveDataErrorExecution none): a failed run loses the event") elif mode == "lastNode" and calls: extra = "; failed executions are not saved either" if no_errors_saved else "" report.add("WARN", "IX07", label, f"answers only after the last node while calling {', '.join(sorted(calls))}; " f"slow calls push the reply past the sender's timeout and the sender retries{extra}") if ntype(node) == N8N_HTTP and node.get("retryOnFail"): wait = node.get("waitBetweenTries", 1000) louder = f"; {wait} ms between tries is under 1,000 ms and hammers a limited API" if isinstance(wait, (int, float)) and wait < 1000 else "" report.add("WARN", "IX08", label, f"retryOnFail uses fixed waits (at most 5 tries, at most 5,000 ms apart) on every " f"error and ignores Retry-After; it is not backoff{louder}") if ntype(node) in N8N_CODE: report.add("WARN", "IX10", label, "Code or Function node: review its writes, retries and error handling by hand") if not isinstance(settings, dict): not_checked.append("IX04 and IX07 saved-error setting: the export has no settings block (copied nodes?)") elif not settings.get("errorWorkflow"): level = "WARN" if tier == "3" else "FAIL" report.add(level, "IX04", "workflow", f"no error workflow is set (Settings, Error workflow); on a tier {tier} flow a " "failed run alerts nobody") if wf.get("active") is True: report.add("INFO", "IX09", "workflow", "workflow is active: these findings affect production runs now") if blind: not_checked.append(f"node parameters missing from this export (sanitized MCP output?): IX01, IX05, IX06 and IX07 not " f"checked on {', '.join(sorted(blind))}, and IX02 cannot see writes there; use the workflow Download") rules = ["IX02", "IX03", "IX08", "IX10"] + ([] if blind else ["IX01", "IX05", "IX06", "IX07"]) return rules + (["IX04"] if isinstance(settings, dict) else []), len(nodes) def unwrap_make(obj): for path in (("response", "blueprint"), ("blueprint",), ()): cur = obj for key in path: cur = cur.get(key) if isinstance(cur, dict) else None if isinstance(cur, dict) and isinstance(cur.get("flow"), list): return cur, cur["flow"], (cur.get("metadata") or {}).get("scenario"), (cur.get("metadata") or {}).get("instant") if isinstance(obj, dict) and isinstance(obj.get("modules"), list): settings = obj.get("config") if isinstance(obj.get("config"), dict) else (obj.get("metadata") or {}).get("scenario") return obj, obj["modules"], settings, (obj.get("metadata") or {}).get("instant") return None def check_make(bp, flow, settings, instant, tier, ordered, report, not_checked): routes, modules = [], [] def walk(route): routes.append(route) for m in route: if isinstance(m, dict) and "module" in m: modules.append(m) for r in m.get("routes") or []: walk((r or {}).get("flow") or []) walk(flow) def mid(m): return str(m.get("module", "")) def label(m): custom = ((m.get("metadata") or {}).get("designer") or {}).get("name") return f"module {m.get('id')} ({mid(m)}{', ' + custom if custom else ''})" def method(m): for block in (m.get("mapper"), m.get("parameters")): if isinstance(block, dict) and block.get("method"): return str(block["method"]).upper() return "GET" def http_like(m): return mid(m).lower().startswith("http:") or "apicall" in mid(m).lower() def func(m): return mid(m).split(":", 1)[-1].lower() def is_create(m): if mid(m).lower().startswith(MAKE_LOCAL_PREFIXES): return False if http_like(m): return method(m) == "POST" return "upsert" not in func(m) and any(w in func(m) for w in CREATE_WORDS) def is_write(m): if mid(m).lower().startswith(MAKE_LOCAL_PREFIXES): return False if http_like(m): return method(m) in WRITE_METHODS return any(w in func(m) for w in WRITE_WORDS) def evidence(m): return "upsert" in func(m) or any(EVIDENCE.search(n) for n in names_in([m.get("mapper"), m.get("parameters")])) def directives(m): return {mid(h) for h in (m.get("onerror") or []) if isinstance(h, dict)} rules = ["IX05", "IX24", "IX25", "IX26", "IX29"] + (["IX30"] if tier in ("1", "2") else []) if isinstance(settings, dict): rules += ["IX21", "IX22", "IX28"] + (["IX23"] if ordered == "yes" else []) if not settings.get("dlq"): report.add("WARN" if tier == "3" else "FAIL", "IX21", "scenario", "Store incomplete executions is off: a failed run " "keeps no copy to retry or resolve, and nothing is retried automatically") if settings.get("dataloss") is True: report.add("FAIL", "IX22", "scenario", "data loss is allowed: when incomplete-execution storage is full, failed data " "is discarded") if ordered == "yes" and settings.get("sequential") is not True: report.add("WARN", "IX23", "scenario", "runs are not sequential (Process data in order is off); instant webhooks run " "in parallel, so changes to one record can apply out of order") if settings.get("confidential") is True: report.add("INFO", "IX27", "scenario", "data is confidential: Make keeps no payloads, so failed runs are harder to resolve") first = flow[0] if flow and isinstance(flow[0], dict) else {} if (instant is True or mid(first) == "gateway:CustomWebHook") and (settings.get("maxErrors") or 0) > 1: report.add("WARN", "IX28", label(first) if first else "scenario", f"instant webhook trigger with maxErrors " f"{settings.get('maxErrors')}: a webhook-triggered scenario deactivates on its first error regardless, so " "plan an alert on deactivation") else: not_checked.append("IX21, IX22, IX23, IX27 and IX28: scenario settings not found (looked for metadata.scenario and config)") dlq_on = isinstance(settings, dict) and bool(settings.get("dlq")) for m in modules: handlers = directives(m) for pname, value in name_value_pairs(m.get("mapper") or {}, {"headers", "qs"}): if SECRET.search(pname) and isinstance(value, str) and value and "{{" not in value: report.add("FAIL", "IX05", label(m), f"parameter {pname} holds a literal value (not printed); move it into a stored " "connection and rotate it, because the blueprint carries it") if is_create(m) and not evidence(m): if "builtin:Break" in handlers: report.add("FAIL", "IX24", label(m), "creates with a Retry (Break) handler and no idempotency evidence: a retry after " "a timeout can create a second record") elif dlq_on: report.add("WARN", "IX24", label(m), "creates with no idempotency evidence; automatic retry of incomplete executions " "re-runs it after connection, rate-limit or timeout errors") if is_write(m) and "builtin:Ignore" in handlers: report.add("FAIL" if tier == "1" else "WARN", "IX25", label(m), "Skip (Ignore) handler on a write: the bundle is dropped " "and the run is marked successful") if is_write(m) and "builtin:Resume" in handlers: report.add("WARN", "IX26", label(m), "Resume handler on a write: the substitute values get written as if real") if tier in ("1", "2") and is_write(m) and not m.get("onerror"): report.add("WARN", "IX30", label(m), f"write with no error handler on a tier {tier} flow") if m.get("onerror") and not any(h.startswith("builtin:") for h in handlers): not_checked.append(f"{label(m)}: error route has no recognised directive; check in the editor how the run ends") for route in routes: for i, m in enumerate(route): if isinstance(m, dict) and mid(m) == "gateway:WebhookRespond" and i < len(route) - 1: nxt = route[i + 1] report.add("WARN", "IX29", label(m), f"webhook response is not the last module on its route ({label(nxt)} runs after " "it): later errors do not deactivate the scenario and send no error notice") return rules, len(modules) def main(argv=None): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("export") p.add_argument("--tier", choices=["1", "2", "3"], default="2", help="flow tier from the inventory (default 2)") p.add_argument("--ordered", choices=["yes", "no"], default="no", help="the flow must apply changes in order") p.add_argument("--json", action="store_true", help="print findings as JSON") args = p.parse_args(argv) try: with open(args.export, encoding="utf-8-sig") as handle: data = json.load(handle) except OSError: print(f"error: file not found or unreadable: {os.path.basename(args.export)}", file=sys.stderr) return 2 except ValueError as exc: print(f"error: not valid JSON: {exc}", file=sys.stderr) return 2 report, not_checked = Report(), [] if isinstance(data, dict) and "triggerInfo" in data and isinstance(data.get("workflow"), dict) and "nodes" not in data["workflow"]: print('error: this n8n MCP output has no node graph (detailLevel "execution"?); ask get_workflow_details for ' 'detailLevel "full", or use the workflow Download', file=sys.stderr) return 2 data = unwrap_n8n(data) if isinstance(data, dict) and isinstance(data.get("nodes"), list) and isinstance(data.get("connections"), dict): kind, name = "n8n workflow", data.get("name", "(unnamed)") version = data.get("activeVersion") published = isinstance(version, dict) and version.get("sameAsDraft") is False and isinstance(version.get("nodes"), list) if published: # top-level nodes are an unpublished draft; the published ones run in production data = dict(data, nodes=version["nodes"], connections=version.get("connections") or {}) not_checked.append("the unpublished draft, which differs from the published version checked here") rules, count = check_n8n(data, args.tier, report, not_checked) state = "active" if data.get("active") is True else "inactive" header = (f'{kind} "{name}" ({os.path.basename(args.export)}): {count} nodes{", published version" if published else ""}, ' f'{state}, tier {args.tier}') not_checked += ["expressions and Code node content", "credentials stored in n8n", "execution history", "secrets outside header and query parameters (URLs, bodies)", "app-node writes whose operation is not named create, add, insert, send, post, update, upsert or delete", "the operation of app nodes left at their default, which n8n omits from the export (IX01 warns when they retry)"] else: found = unwrap_make(data) if isinstance(data, dict) else None if not found or not all(isinstance(m, dict) and "module" in m for m in found[1]): print(f"error: {UNSUPPORTED}", file=sys.stderr) return 2 bp, flow, settings, instant = found kind, name = "Make scenario", bp.get("name", "(unnamed)") rules, count = check_make(bp, flow, settings, instant, args.tier, args.ordered, report, not_checked) header = (f'{kind} "{name}" ({os.path.basename(args.export)}): {count} modules, ' f'{"instant trigger" if instant is True else "trigger type from the first module"}, tier {args.tier}, ordered {args.ordered}') not_checked += ["module writes recognised only by name (create, add, insert, send, post, update, upsert, delete, " "upload) or HTTP method", "modules inside error routes", "connections and stored credentials", "secrets outside mapper headers and query strings (URLs, bodies)", "scenario history and incomplete executions"] order = {r: i for i, r in enumerate(sorted({i["rule"] for i in report.items} | set(rules)))} findings = sorted(report.items, key=lambda f: order[f["rule"]]) passes = [{"level": "PASS", "rule": r, "target": "", "message": "no finding"} for r in sorted(rules) if not any(f["rule"] == r for f in findings)] counts = {lvl: sum(1 for f in findings + passes if f["level"] == lvl) for lvl in ("FAIL", "WARN", "PASS", "INFO")} summary = f"{counts['FAIL']} FAIL, {counts['WARN']} WARN, {counts['PASS']} PASS" if counts["INFO"]: summary += f", {counts['INFO']} INFO" summary += ". Nothing was changed." if args.json: print(json.dumps({"export": header, "findings": findings, "passes": passes, "not_checked": not_checked, "summary": counts}, indent=2)) else: print(header) for f in findings: print(f"{f['level']} {f['rule']} {f['target']}: {f['message']}") for f in passes: print(f"PASS {f['rule']}: no finding") print(summary) print("Not checked by this script: " + "; ".join(not_checked) + ".") return 1 if counts["FAIL"] else 0 if __name__ == "__main__": sys.exit(main())